package provision

// ReconcileSidecars turns bucket SIDECAR records into registry routes, for
// THIS agent's own region, pool set and inbound posture. Spec:
// docs/superpowers/specs/2026-08-17-sidecar-members-design.md §4, §8.
//
// It is Reconcile's twin, one record family over, and it is deliberately a
// separate pass rather than a widening of that one: the two families answer
// different questions about a box (can it SERVE vs may it RUN), and a single
// loop would have had to carry a Kind branch through every phase — including
// the removal phase, where a missed branch deletes production rows.
//
// Which makes the collision this file exists to close worth stating plainly.
// Reconcile's removal phase is scoped by Provisioned, region, pool and
// absence-from-records. A sidecar row is Provisioned too (that flag is what
// grants a pass removal authority over a row), and it can never appear in the
// INSTANCE record set — so, unguarded, that pass deletes every sidecar row on
// every tick: minted by this pass, destroyed by that one, forever, with no
// error anywhere and nothing in the logs that reads as a failure. Each pass's
// removal authority is therefore scoped to its OWN Kind, asserted from both
// sides (TestInstanceReconcileNeverRemovesSidecarRows,
// TestSidecarReconcileRemovesOnlyItsOwnKind).
//
// ONE EXCEPTION, and it is deliberate: domains.Reconcile drops EVERY route of
// a trashed app (internal/domains/reconcile.go, the trashed-apps loop) with no
// Kind filter at all, sidecar rows included. So removal authority over a
// sidecar row is held by TWO passes, not one. That is not a leak — it is the
// only path by which a trashed app's sidecars stop running, it carries its own
// production-flag guard, and it is precisely what makes THIS pass's
// decline-but-keep response to a trashed app correct rather than a way for
// rows to outlive their app. Anyone tightening the rule stated above should
// know where its exception lives before assuming a sidecar row can only be
// deleted here.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"time"

	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// SidecarKind is the route Kind every sidecar row carries — the whole basis
// of each pass's removal authority, and the value the edge refuses outright.
// A literal in two passes' removal conditions is a typo away from a fleet
// deletion, so both read this constant.
const SidecarKind = "sidecar"

// sidecarCoSpawnSuffix names the LEGACY, file-staged co-spawn: a sidecar
// attached to a host by dropping <DataDir>/<host>-sidecar.env
// (internal/activator/backend_exec.go). It is legacy but LIVE — Eleven runs
// on it today — so a host that already has one is refused a native sidecar
// row. One host, one mechanism.
const sidecarCoSpawnSuffix = "-sidecar.env"

// sidecarArgsJSON renders a manifest's argv as the wire text registry.Route's
// Args column carries: "" for nil/empty (no argv), else the JSON array. Same
// shape and same reason as seedJSON — json.Marshal of a nil slice is "null",
// which is neither empty nor a usable argv.
func sidecarArgsJSON(m SidecarManifest) string {
	if len(m.Args) == 0 {
		return ""
	}
	b, err := json.Marshal(m.Args) // []string always marshals; err is unreachable
	if err != nil {
		return ""
	}
	return string(b)
}

// sidecarPlacement resolves the placement policy a record is placed under
// (design §8): MOST RESTRICTIVE WINS. Either voice asking for no-inbound gets
// no-inbound, and only when both are silent (or both say "any") does the
// workload place anywhere.
//
// The record can therefore still NARROW an app-wide "any" — one bot that must
// sit somewhere with no inbound path, under an app with no such rule — which
// is the direction that reads as a requirement. What it can no longer do is
// WIDEN: a record stating "any" under a no-inbound manifest used to place that
// bot on inbound boxes, turning app-wide policy into a suggestion. The console
// door already refuses to write such a record (it never accepts a placement
// field at all, exactly so the manifest stays authoritative); a direct bucket
// write, an older writer, or anything else holding the prefix's keys does not
// go through that door, and both doors hold the invariant rather than trusting
// each other.
func sidecarPlacement(rec SidecarRecord, m SidecarManifest) string {
	if rec.Placement == PlacementNoInbound || m.PlacementOrDefault() == PlacementNoInbound {
		return PlacementNoInbound
	}
	if rec.Placement != "" {
		return rec.Placement
	}
	return m.PlacementOrDefault()
}

// placeable reports whether this box satisfies a record's placement — the
// BOX half of design §8's two independent halves. It reads as a SCOPE test,
// beside region and pool, and is used as one: a record this box may not run
// is not this box's business at all, so it never enters the record set and
// the removal phase takes any row it left behind. That is how a placement
// change takes effect — the same way a region change does.
func placeable(rec SidecarRecord, m SidecarManifest, cfg ReconcileConfig) bool {
	return sidecarPlacement(rec, m) != PlacementNoInbound || cfg.Inbound == InboundNone
}

// sidecarRouteFor is the one answer to "what route does this sidecar record
// describe" — the mint builds from it and the converge branch compares
// against it, so a shape this pass would mint and a shape it calls converged
// can never drift apart (routeFor's rule, same reason).
//
// Socket and Addr are deliberately left EMPTY, and that is the whole
// reachability story: a sidecar binds nothing and nothing dials it. The edge
// refuses Kind == "sidecar" outright (design §6.2) rather than relying on
// this omission, but the row must not carry an address for any future arm to
// find either.
//
// DBPath, by contrast, IS derived: a sidecar gets a database like any other
// tenant, and a hibernating route whose row carried no path would be swept up
// by ops.FleetDBs' dataDir glob and replicated box-wide as if it never slept
// (routeFor's comment records what that costs).
//
// The ceilings are materialized onto the row from the manifest precisely so
// the activator's sweeper never needs a bucket read to decide a worker has
// outrun its limits.
func sidecarRouteFor(rec SidecarRecord, m SidecarManifest, dataDir string) registry.Route {
	return registry.Route{
		Host: rec.Host, Slug: rec.Name, Kind: SidecarKind,
		App: rec.App, Account: rec.Account, Pool: rec.Pool, Region: rec.Region,
		// The RECORD's channel, never the manifest's: a record that reached
		// this pass without one was refused above (a route with no channel
		// resolves no release, so it has no build to exec), so there is
		// nothing here for a manifest-side fallback to rescue. The manifest's
		// Channel is the console door's default at CREATE time.
		Channel: rec.Channel, Environment: rec.Environment,
		DBPath:        DefaultDBPath(dataDir, rec.Host),
		Hibernate:     m.HibernateOrDefault(),
		Args:          sidecarArgsJSON(m),
		IdleSeconds:   int(m.IdleOrDefault() / time.Second),
		MaxRunSeconds: int(m.MaxRunOrDefault() / time.Second),
		Provisioned:   true,
		Status:        "ready",
	}
}

// coSpawnStaged reports whether host already has a legacy file-staged
// co-spawn in dataDir. A read error other than "not there" answers TRUE: not
// minting is the reversible direction, and a host whose staging state cannot
// be established is exactly the host a second mechanism must not be attached
// to.
func coSpawnStaged(dataDir, host string) bool {
	_, err := os.Stat(filepath.Join(dataDir, host+sidecarCoSpawnSuffix))
	return err == nil || !errors.Is(err, os.ErrNotExist)
}

// ReconcileSidecars materializes bucket sidecar records into registry routes
// for this box's region, pools and inbound posture: mint on appearance,
// converge the manifest-owned columns, remove on record deletion — removal
// only for Kind "sidecar" routes this pass (or an earlier one running the
// same scope) minted, marked by Provisioned.
//
// Per-item failures log and skip. A failed account-manifest or record read
// additionally disables the removal phase for the WHOLE pass: a transient
// read failure must never mass-retire a tenant's fleet. Minting is additive
// and always safe, so it proceeds regardless.
//
// SCOPE — region, pool and PLACEMENT (design §8) — decides what enters the
// record set at all, and so decides removal: a record this box may not run is
// not this box's business, and any row it left behind is retired by the
// ordinary removal phase. That is how a placement change takes effect, exactly
// as a region change does through the instance pass; a member who asks that a
// bot only ever run where there is no inbound path must not find it still
// running on an inbound box after flipping the manifest. Evacuation is
// governed by the same fail-safe as every other removal: if any read failed
// this pass, nothing is evacuated.
//
// Two states instead decline to MINT while leaving the record in the record
// set — so removal never fires for them — because neither is a statement
// about where the workload belongs:
//
//   - the app is in the trash (domains.Reconcile drops its routes on its own
//     authority; minting here would re-raise them every tick),
//   - the app's sidecars manifest is absent or unreadable (no argv to run,
//     and no ceilings to stamp).
func ReconcileSidecars(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig) error {
	if cfg.Now == nil {
		cfg.Now = time.Now
	}

	// scan is the record-collection phase over st: the placeable in-scope
	// records by host, the hosts whose record is there but cannot be used
	// (protect), the per-app sidecars manifest lookup the pass goes on using,
	// and whether anything failed to read (readFailed, which disables removal
	// for the whole pass; a manifest read that fails later in the pass sets
	// it too). The pass scans s; a pass over a record view snapshot that has
	// a removal to decide scans the bucket too (freshStore), so both judge
	// every record by exactly these rules, placement against the record and
	// the app's sidecars manifest included (S3 change-index design §4.4).
	type sidecarScan struct {
		recs        map[string]SidecarRecord
		protect     map[string]bool
		readFailed  bool
		manifestFor func(acct, app string) (SidecarManifest, bool)
	}
	scan := func(st release.Store) (*sidecarScan, error) {
		sc := &sidecarScan{}
		known, readFailed, err := knownAccounts(ctx, st)
		if err != nil {
			return nil, err
		}
		sc.readFailed = readFailed

		// --- manifest cache, per (acct, app), lazily ------------------------------
		// known=false covers BOTH "no manifest" and "the read failed", and in both
		// cases this pass declines to mint or converge: every manifest-owned
		// column (argv, the ceilings, hibernation) would otherwise be stamped from
		// a zero value — an empty argv over a working one, on a blip. This is
		// seedFor's seedKnown discipline, applied to a family where the manifest
		// carries the RUN instruction rather than a seed.
		//
		// Declared BEFORE the record scan because placement is a scope axis and
		// placement can be stated app-wide: deciding whether a record is in scope
		// needs the manifest in hand. An unknown manifest resolves placement from
		// the record alone (sidecarPlacement over a zero manifest) — and a manifest
		// whose read FAILED has already disabled the removal phase, so an
		// unknowable placement can never evacuate anything.
		manCache := map[acctApp]SidecarManifest{}
		manKnown := map[acctApp]bool{}
		manSeen := map[acctApp]bool{}
		manifestFor := func(acct, app string) (SidecarManifest, bool) {
			k := acctApp{acct, app}
			if manSeen[k] {
				return manCache[k], manKnown[k]
			}
			manSeen[k] = true
			m, err := GetSidecarManifest(ctx, st, acct, app)
			if err != nil {
				if errors.Is(err, os.ErrNotExist) {
					// Absence IS the answer (design §3.1): sidecars are not
					// enabled for this app. Not a read failure — removal stays
					// armed for the pass — but the records themselves stay in
					// recs, so nothing of theirs is retired on it.
					log.Printf("provision: %s/%s has sidecar records but no sidecars manifest — not minting", acct, app)
					return SidecarManifest{}, false
				}
				log.Printf("provision: sidecar manifest %s/%s: read failed: %v", acct, app, err)
				sc.readFailed = true
				return SidecarManifest{}, false
			}
			manCache[k], manKnown[k] = *m, true
			return manCache[k], manKnown[k]
		}

		// --- collect in-scope records --------------------------------------------
		keys, partial, err := listAppSubtree(ctx, st, known, "sidecars/")
		if err != nil {
			return nil, fmt.Errorf("provision: list apps: %w", err)
		}
		if partial {
			sc.readFailed = true // unseen records must not read as retired
		}
		recs := map[string]SidecarRecord{} // host -> record, in-scope only
		// protect holds hosts whose record KEY parsed but whose body could not be
		// trusted — unparseable JSON, a body disagreeing with its own path, or a
		// shape this binary refuses. Such a host must NEVER read as "the bucket
		// retired it": a non-additive record encoding change reaching an older
		// binary would make EVERY record fail exactly this way, and one tick would
		// then delete every sidecar in scope. Narrower than readFailed (which
		// vetoes removal for the whole pass): only the unusable host is held back.
		protect := map[string]bool{}
		var recordKeys []string
		for _, key := range keys {
			if _, _, _, ok := ParseSidecarKey(key, known); ok {
				recordKeys = append(recordKeys, key)
			}
		}
		for _, part := range chunks(recordKeys) {
			bodies := getEach(ctx, st, part)
			for _, key := range part {
				acct, app, name, _ := ParseSidecarKey(key, known)
				// Derived from the KEY, never from the body: the body is what is
				// under suspicion in every branch below.
				host := SidecarHost(name, app, acct)
				raw, err := bodies[key].body, bodies[key].err
				if err != nil {
					log.Printf("provision: skipping sidecar record %s: read failed: %v", key, err)
					// Its absence from recs below must not read as "the bucket retired
					// it" — a transient Get failure disables removal for the whole
					// pass, same as an unreadable account manifest.
					sc.readFailed = true
					continue
				}
				var rec SidecarRecord
				if err := json.Unmarshal(raw, &rec); err != nil {
					log.Printf("provision: skipping sidecar record %s: malformed: %v", key, err)
					protect[host] = true
					continue
				}
				if rec.Name != name || rec.Account != acct || rec.App != app {
					log.Printf("provision: skipping sidecar record %s: disagrees with its own path (name=%q account=%q app=%q)",
						key, rec.Name, rec.Account, rec.App)
					protect[host] = true
					continue
				}
				// ValidateSidecarShape is the console create door's policy check — but
				// a door is not a guarantee about what is already in the bucket:
				// anything with bucket write access, and any record written by a
				// different (older, or future) writer, arrives here unexamined. Both
				// doors hold the invariant, neither trusts the other. Protected like a
				// corrupted body, and for the same reason: a record that cannot be
				// used is not a record that was retired.
				if err := ValidateSidecarShape(rec); err != nil {
					log.Printf("provision: skipping sidecar record %s: invalid shape: %v", key, err)
					protect[host] = true
					continue
				}
				// The host is DERIVED from the identity, never taken on trust: it
				// becomes a database path and a registry primary key, and a record
				// free to state its own host could name any row on the box.
				if rec.Host != host {
					log.Printf("provision: skipping sidecar record %s: host %q is not %q", key, rec.Host, host)
					protect[host] = true
					continue
				}
				// ValidateSidecarShape deliberately does not inspect Channel (it
				// validates shape fields, and Channel is not one), but a channel-less
				// record must not become — or converge — a route pinned to no channel:
				// such a row resolves no release, so it has no build to exec, and the
				// worker boot pass would try to run it against a symlink that does not
				// exist. The console stamps a channel at create, so a record without
				// one is malformed. Protected like a corrupted body: the record is
				// there, it just cannot be used.
				if rec.Channel == "" {
					log.Printf("provision: skipping sidecar record %s: empty channel", key)
					protect[host] = true
					continue
				}
				if rec.Region != cfg.Region || !poolServed(cfg.Pools, rec.Pool) {
					continue // another agent's jurisdiction, not this pass's business
				}
				// Placement is the third scope axis (design §8), and it is applied
				// HERE, beside region and pool, precisely so it behaves like them: a
				// record this box may not run never enters recs, and the removal phase
				// below therefore evacuates any row it left behind. "none" is a
				// property of the BOX's network; "no-inbound" is a requirement of the
				// WORKLOAD — the point of the two halves being independent is that a
				// member gets a mechanical guarantee rather than an operator's promise,
				// which a row left running on an inbound box would not be.
				if m, _ := manifestFor(acct, app); !placeable(rec, m, cfg) {
					log.Printf("provision: sidecar %s is not placeable here: it asks for %q placement and this box accepts inbound work (CARLOS_INBOUND=%q)",
						rec.Host, sidecarPlacement(rec, m), cfg.Inbound)
					continue
				}
				// Two records claiming one host is a collision this pass cannot
				// arbitrate silently: whichever this listing reaches FIRST wins, and
				// the loser is skipped loudly rather than overwriting the winner in
				// the map with no trace of the conflict.
				if prev, dup := recs[rec.Host]; dup {
					log.Printf("provision: duplicate sidecar host %s: keeping %s, skipping %s", rec.Host,
						SidecarRecordKey(prev.Account, prev.App, prev.Name), key)
					continue
				}
				recs[rec.Host] = rec
			}
		}
		sc.recs, sc.protect, sc.manifestFor = recs, protect, manifestFor
		return sc, nil
	}
	sc, err := scan(s)
	if err != nil {
		return err
	}
	recs, protect, manifestFor := sc.recs, sc.protect, sc.manifestFor

	trashed := trashGate(ctx, s, "sidecar")

	// --- mint / converge -------------------------------------------------------
	routes, err := reg.List()
	if err != nil {
		return fmt.Errorf("provision: list routes: %w", err)
	}
	byHost := map[string]registry.Route{}
	for _, rt := range routes {
		byHost[rt.Host] = rt
	}

	for _, rec := range recs {
		if trashed(rec.Account, rec.App) {
			continue
		}
		m, mKnown := manifestFor(rec.Account, rec.App)
		if !mKnown {
			continue
		}
		existing, ok := byHost[rec.Host]

		// Refusals that need no claim are decided BEFORE cfg.claimInstall: the
		// hook activates the host guard and writes inventory, effects that a
		// record this pass will not act on must never cause. Only work that
		// is going to touch the registry reaches it.
		var prior *registry.Route
		nr := sidecarRouteFor(rec, m, cfg.DataDir)
		if ok {
			prior = &existing
		} else {
			// One host, one mechanism: a host already carrying the legacy
			// file-staged co-spawn must not ALSO get a native sidecar row, or
			// the same workload runs twice under two owners that know nothing
			// of each other.
			//
			// This is a MINT-TIME check and nothing more. A -sidecar.env
			// staged AFTER this row exists is not detected, and that host then
			// carries both mechanisms with no complaint from this pass. It is
			// left that way deliberately: the reverse — a converge-time
			// refusal — would mean a file dropped by hand could stop an
			// already-provisioned sidecar converging, which hands anything
			// that can write the data dir a way to freeze a member's row. The
			// direction that matters is the one guarded: the file came first,
			// so the file keeps the host.
			if coSpawnStaged(cfg.DataDir, rec.Host) {
				log.Printf("provision: REFUSING to mint sidecar %s: %s%s is staged on this box — "+
					"the legacy file co-spawn and a native sidecar row are two mechanisms for one host; "+
					"remove the file to migrate this host onto sidecar records",
					rec.Host, rec.Host, sidecarCoSpawnSuffix)
				continue
			}
			// Same admission rule as the instance pass (amadan #213): a
			// sidecar that would start without its sealed config is not
			// minted here. See ReconcileConfig.ConfigGate.
			if cfg.ConfigGate != nil {
				if err := cfg.ConfigGate(ctx, nr); err != nil {
					continue
				}
			}
		}
		// A row at this host that this pass does not own is left alone before
		// any claim work, so the hook never activates the guard or writes
		// inventory for it. (Flag off, the same decision is made inside the
		// write below.)
		if ok && cfg.ClaimEnabled && !(existing.Kind == SidecarKind && existing.Provisioned &&
			existing.App == rec.App && existing.Account == rec.Account) {
			log.Printf("provision: %s exists but is not a sidecar row this pass owns (kind %q provisioned %v app %q acct %q) — leaving it alone",
				rec.Host, existing.Kind, existing.Provisioned, existing.App, existing.Account)
			continue
		}
		selected := ClaimRecord{Host: rec.Host, Key: SidecarRecordKey(rec.Account, rec.App, rec.Name), Family: "sidecar", Account: rec.Account, App: rec.App, CreatedAt: rec.CreatedAt, Value: rec, Route: nr, DryRun: cfg.DryRun}
		if err := cfg.claimInstall(ctx, selected, prior, func(owner registry.ClaimOwnership) error {
			switch {
			case !ok:
				nr.ClaimOwnership = owner
				if cfg.DryRun {
					log.Printf("provision: dry-run: would mint sidecar route %s (app %q acct %q channel %q args %s)",
						rec.Host, rec.App, rec.Account, nr.Channel, nr.Args)
					return nil
				}
				if err := reg.Upsert(nr); err != nil {
					log.Printf("provision: mint sidecar %s: %v", rec.Host, err)
					return nil
				}
				log.Printf("provision: minted sidecar route %s (app %q acct %q args %s)", rec.Host, rec.App, rec.Account, nr.Args)
			case existing.Kind == SidecarKind && existing.Provisioned &&
				existing.App == rec.App && existing.Account == rec.Account:
				convergeSidecar(reg, existing, rec, m, cfg)
			default:
				// A row exists at this host that this pass does not own. There is
				// deliberately NO adoption path here, unlike the instance pass:
				// the .sidecar.internal suffix is reserved and refused at every
				// door a name could enter through, so no hand-added row to
				// backfill can legitimately exist — and adoption is an authority
				// TRANSFER (it is what grants the removal phase power over a row),
				// never a shape coincidence.
				log.Printf("provision: %s exists but is not a sidecar row this pass owns (kind %q provisioned %v app %q acct %q) — leaving it alone",
					rec.Host, existing.Kind, existing.Provisioned, existing.App, existing.Account)
			}
			return nil
		}); err != nil {
			log.Printf("provision: claim install %s: %v", rec.Host, err)
		}
	}

	// --- removal, only when nothing failed to read this pass --------------------
	// Four conditions, each load-bearing: this pass's OWN Kind (the guard at
	// the top of this file — the instance pass carries its mirror), Provisioned
	// (the flag, not the app and not the channel, is what grants removal
	// authority; a hand-added row never has it), scope (this box's own
	// region/pool jurisdiction), and absence from the record set this pass just
	// read (the bucket's own decision to retire the sidecar, not a guess).
	//
	// retire runs that phase over one set of records and protected hosts.
	// With stopAtFirst it removes nothing and reports whether any row would
	// go: a record view snapshot's absence is not the bucket's, so a pass
	// over a snapshot that would remove anything decides every removal again
	// from the bucket's own walk, below (S3 change-index design §4.4).
	retire := func(recs map[string]SidecarRecord, protect map[string]bool, stopAtFirst bool) (wouldRemove bool) {
		for _, rt := range routes {
			if rt.Kind != SidecarKind || !rt.Provisioned {
				continue
			}
			if rt.Region != cfg.Region || !poolServed(cfg.Pools, rt.Pool) {
				continue
			}
			if _, ok := recs[rt.Host]; ok {
				continue
			}
			if protect[rt.Host] {
				log.Printf("provision: not removing sidecar %s: its record exists but could not be read cleanly this pass", rt.Host)
				continue
			}
			if stopAtFirst {
				return true
			}
			if cfg.DryRun {
				log.Printf("provision: dry-run: would remove retired sidecar route %s (record gone)", rt.Host)
				continue
			}
			if err := cfg.claimRetire(ctx, reg, rt); err != nil {
				log.Printf("provision: remove retired sidecar route %s: %v", rt.Host, err)
				continue
			}
			log.Printf("provision: removed retired sidecar route %s (record gone)", rt.Host)
		}
		return false
	}
	if sc.readFailed || !retire(recs, protect, onSnapshot(s)) {
		return nil
	}
	// A snapshot pass with a removal to make: the record walk again, over
	// the bucket, and the removal phase over what it read, by the walk's own
	// rules (unusable records protected before scope, hosts matched across
	// every key, placement from the bucket's record and sidecars manifest
	// against this box's inbound). Anything it cannot read keeps every row
	// this pass, as in the walk.
	lsc, err := scan(freshStore(s))
	if err != nil || lsc.readFailed {
		log.Printf("provision: not removing any sidecar route this pass: the record view lacks a sidecar this box would run, and the bucket's own walk could not be read cleanly%s", walkErrSuffix(err))
		return nil
	}
	retire(lsc.recs, lsc.protect, false)
	return nil
}

// convergeSidecar applies the columns the record and the manifest are
// authoritative for onto a row this pass owns. Narrow UPDATEs throughout,
// never a re-Upsert: a read-modify-write racing the activator's own writes
// (a worker's reported version, its next-due stamp) would clobber them.
//
// The split of authority is the same one the design states: the RECORD owns
// channel and environment (what this sidecar runs and which config it gets —
// the #130 discipline), the MANIFEST owns argv, the ceilings and hibernation
// (app-wide policy). Both are re-stamped every pass, exactly as an instance's
// seed is, because truth lives in the bucket and a box-local edit to one of
// these columns is drift, not a decision.
//
// Every write logs on success as well as on failure: these are the writes
// that REVERT an operator's own `carlos route` edit, and a row that moved with
// nothing saying so is the failure class behind #130 seen from the outside.
func convergeSidecar(reg *registry.Registry, existing registry.Route, rec SidecarRecord, m SidecarManifest, cfg ReconcileConfig) {
	want := sidecarRouteFor(rec, m, cfg.DataDir)

	if want.Channel != "" && existing.Channel != want.Channel {
		if cfg.DryRun {
			log.Printf("provision: dry-run: would converge sidecar %s channel %s -> %s", rec.Host, existing.Channel, want.Channel)
		} else if err := reg.SetChannel(rec.Host, want.Channel); err != nil {
			log.Printf("provision: converge channel on sidecar %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged sidecar %s channel %s -> %s", rec.Host, existing.Channel, want.Channel)
		}
	}
	if existing.Environment != want.Environment {
		if cfg.DryRun {
			log.Printf("provision: dry-run: would converge sidecar %s environment %q -> %q", rec.Host, existing.Environment, want.Environment)
		} else if err := reg.SetEnvironment(rec.Host, want.Environment); err != nil {
			log.Printf("provision: converge environment on sidecar %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged sidecar %s environment %q -> %q", rec.Host, existing.Environment, want.Environment)
		}
	}
	if existing.Hibernate != want.Hibernate {
		if cfg.DryRun {
			log.Printf("provision: dry-run: would converge sidecar %s hibernate %v -> %v", rec.Host, existing.Hibernate, want.Hibernate)
		} else if err := reg.SetHibernate(rec.Host, want.Hibernate); err != nil {
			log.Printf("provision: converge hibernate on sidecar %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged sidecar %s hibernate %v -> %v", rec.Host, existing.Hibernate, want.Hibernate)
		}
	}
	if existing.Args != want.Args {
		if cfg.DryRun {
			log.Printf("provision: dry-run: would converge sidecar %s args %s -> %s", rec.Host, existing.Args, want.Args)
		} else if err := reg.SetArgs(rec.Host, want.Args); err != nil {
			log.Printf("provision: converge args on sidecar %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged sidecar %s args %s -> %s", rec.Host, existing.Args, want.Args)
		}
	}
	// One UPDATE for both ceilings, because they are one policy: the idle
	// window and the run ceiling together decide when the sweeper stops a
	// worker, and a row carrying the new idle under the old maximum (or the
	// reverse) is a combination the manifest never stated.
	if existing.IdleSeconds != want.IdleSeconds || existing.MaxRunSeconds != want.MaxRunSeconds {
		if cfg.DryRun {
			log.Printf("provision: dry-run: would converge sidecar %s ceilings %d/%d -> %d/%d", rec.Host,
				existing.IdleSeconds, existing.MaxRunSeconds, want.IdleSeconds, want.MaxRunSeconds)
		} else if err := reg.SetWorkerLimits(rec.Host, want.IdleSeconds, want.MaxRunSeconds); err != nil {
			log.Printf("provision: converge ceilings on sidecar %s: %v", rec.Host, err)
		} else {
			log.Printf("provision: converged sidecar %s ceilings %d/%d -> %d/%d", rec.Host,
				existing.IdleSeconds, existing.MaxRunSeconds, want.IdleSeconds, want.MaxRunSeconds)
		}
	}
	// The columns nothing converges in place: the database path (moving it
	// under a running process would point it at a second copy of its own
	// data) and the two reachability columns a sidecar row must never carry.
	// Reported, never rewritten — `carlos remove` purges a route's capability
	// grants, so it is a last resort, not a routine fix.
	if existing.DBPath != want.DBPath || existing.Socket != "" || existing.Addr != "" {
		log.Printf("provision: sidecar %s: row disagrees with the shape this record describes "+
			"(db %q/%q socket %q addr %q) — not converged in place; a sidecar binds nothing, "+
			"so a socket or addr on this row is a row that was not minted by this pass",
			rec.Host, existing.DBPath, want.DBPath, existing.Socket, existing.Addr)
	}
}