package provision

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

// --- fixtures -----------------------------------------------------------------

// failStore is the read-failure seam these tests need and no other store
// shape provides: a Store whose Get (and List) refuse a chosen prefix, so a
// pass can be put in exactly the state the fail-safe exists for — a record
// that IS there and could not be read — rather than the state that looks the
// same from the outside and means the opposite (a record the bucket retired).
type failStore struct {
	release.Store
	fail []string
}

func (s *failStore) failReadsFor(prefix string) { s.fail = append(s.fail, prefix) }

func (s *failStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	for _, p := range s.fail {
		if strings.HasPrefix(key, p) {
			return nil, "", fmt.Errorf("failStore: reads disabled under %s", p)
		}
	}
	return s.Store.Get(ctx, key)
}

func (s *failStore) List(ctx context.Context, prefix string) ([]string, error) {
	for _, p := range s.fail {
		if strings.HasPrefix(prefix, p) {
			return nil, fmt.Errorf("failStore: reads disabled under %s", p)
		}
	}
	return s.Store.List(ctx, prefix)
}

type sidecarFixture struct {
	store *failStore
	reg   *registry.Registry
	dir   string
	sqid  string
}

// sidecarSetup writes a valid account manifest and hands back a fixture whose
// data dir and registry are empty — the sidecar family's equivalent of
// setup() above, minus the instances manifest (a sidecar app opts in through
// its OWN manifest, sidecars.json).
func sidecarSetup(t *testing.T) *sidecarFixture {
	t.Helper()
	s := &failStore{Store: dirStore(t)}
	sqid := writeAccountManifest(t, s, acctID, "Eleven")
	return &sidecarFixture{store: s, reg: regWith(t), dir: t.TempDir(), sqid: sqid}
}

func (f *sidecarFixture) host(name, app string) string { return SidecarHost(name, app, f.sqid) }

func (f *sidecarFixture) putManifest(t *testing.T, app string, m SidecarManifest) {
	t.Helper()
	if err := PutSidecarManifest(ctx, f.store, f.sqid, app, m); err != nil {
		t.Fatal(err)
	}
}

func (f *sidecarFixture) putRecord(t *testing.T, rec SidecarRecord) {
	t.Helper()
	if rec.Account == "" {
		rec.Account = f.sqid
	}
	if rec.Host == "" {
		rec.Host = SidecarHost(rec.Name, rec.App, rec.Account)
	}
	if err := CreateSidecarRecord(ctx, f.store, rec); err != nil {
		t.Fatal(err)
	}
}

func (f *sidecarFixture) mustUpsert(t *testing.T, rt registry.Route) {
	t.Helper()
	if err := f.reg.Upsert(rt); err != nil {
		t.Fatal(err)
	}
}

func (f *sidecarFixture) reconcile(t *testing.T, cfg ReconcileConfig) {
	t.Helper()
	cfg.DataDir = f.dir
	if err := ReconcileSidecars(ctx, f.store, f.reg, cfg); err != nil {
		t.Fatalf("ReconcileSidecars: %v", err)
	}
}

// noRoute asserts a host has no row. registry.Get answers a missing host with
// (nil, nil), so a bare `err == nil` check would pass whether the row is
// there or not.
func noRoute(t *testing.T, reg *registry.Registry, host, msg string) {
	t.Helper()
	rt, err := reg.Get(host)
	if err != nil {
		t.Fatal(err)
	}
	if rt != nil {
		t.Fatal(msg)
	}
}

func hasRoute(t *testing.T, reg *registry.Registry, host, msg string) *registry.Route {
	t.Helper()
	rt, err := reg.Get(host)
	if err != nil {
		t.Fatal(err)
	}
	if rt == nil {
		t.Fatal(msg)
	}
	return rt
}

// --- the removal-authority collision -------------------------------------------

// TestInstanceReconcileNeverRemovesSidecarRows is the guard on the OTHER
// pass. A sidecar row is Provisioned (that flag is what grants a pass removal
// authority) and can never appear in the instance pass's record set, so
// without a Kind guard the instance pass deletes every sidecar row on every
// tick — minted by one pass, destroyed by the other, forever, with no error
// anywhere and nothing in the logs that reads as a failure.
func TestInstanceReconcileNeverRemovesSidecarRows(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	f.mustUpsert(t, registry.Route{Host: host, Kind: "sidecar", Slug: "bot",
		App: "eleven", Account: f.sqid, Provisioned: true, Status: "ready"})

	// No instance records at all: the pass's removal phase runs in full.
	if err := Reconcile(ctx, f.store, f.reg, ReconcileConfig{DataDir: f.dir}); err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	hasRoute(t, f.reg, host, "the instance pass deleted a sidecar row — each pass's removal authority is scoped to its OWN kind, or the two passes destroy each other's work forever")
}

// TestSidecarReconcileRemovesOnlyItsOwnKind is the same guard from the other
// side: this pass retires its own kind and must never reach an instance row,
// which can never appear in ITS record set either.
func TestSidecarReconcileRemovesOnlyItsOwnKind(t *testing.T) {
	f := sidecarSetup(t)
	gone := f.host("gone", "eleven")
	f.mustUpsert(t, registry.Route{Host: gone, Kind: "sidecar", Slug: "gone",
		App: "eleven", Account: f.sqid, Provisioned: true})
	f.mustUpsert(t, registry.Route{Host: "app.example.com", Kind: "instance",
		App: "eleven", Account: f.sqid, Provisioned: true})
	// A sidecar-kind row nothing provisioned: Provisioned is the flag that
	// GRANTS removal authority, so a row without it is not this pass's to
	// retire however well it matches on every other column.
	hand := f.host("hand", "eleven")
	f.mustUpsert(t, registry.Route{Host: hand, Kind: "sidecar", Slug: "hand",
		App: "eleven", Account: f.sqid})

	f.reconcile(t, ReconcileConfig{})

	noRoute(t, f.reg, gone, "a provisioned sidecar with no record must be removed")
	hasRoute(t, f.reg, "app.example.com", "the sidecar pass must never touch an instance row")
	hasRoute(t, f.reg, hand, "a sidecar row that was never provisioned is not this pass's to remove")
}

// --- mint / converge / remove --------------------------------------------------

func TestSidecarReconcileMints(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{
		Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}, Channel: "stable"})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})

	f.reconcile(t, ReconcileConfig{})

	rt := hasRoute(t, f.reg, f.host("bot", "eleven"), "no sidecar route minted")
	if rt.Kind != "sidecar" {
		t.Fatalf("Kind = %q", rt.Kind)
	}
	if rt.Socket != "" || rt.Addr != "" {
		t.Fatalf("a sidecar row must carry NO socket and NO addr, got socket=%q addr=%q", rt.Socket, rt.Addr)
	}
	if rt.Slug != "bot" || rt.App != "eleven" || rt.Account != f.sqid || rt.Channel != "stable" {
		t.Fatalf("route identity = %+v", rt)
	}
	if rt.DBPath != DefaultDBPath(f.dir, rt.Host) || !rt.Provisioned || rt.Args != `["sidecar","run"]` {
		t.Fatalf("route = %+v", rt)
	}
	if !rt.Hibernate {
		t.Fatal("Hibernate defaults true (HibernateOrDefault), and the row is what the activator reads")
	}
	if rt.IdleSeconds != 300 || rt.MaxRunSeconds != 3600 {
		t.Fatalf("ceilings = %d/%d, want the manifest defaults 300/3600 materialized onto the row — the sweeper must never need a bucket read",
			rt.IdleSeconds, rt.MaxRunSeconds)
	}
}

func TestSidecarReconcileConvergesArgsFromTheManifest(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	f.reconcile(t, ReconcileConfig{})

	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"worker", "run"}})
	f.reconcile(t, ReconcileConfig{})

	rt := hasRoute(t, f.reg, f.host("bot", "eleven"), "route gone")
	if rt.Args != `["worker","run"]` {
		t.Fatalf("Args = %q — truth lives in the bucket and is re-stamped every pass, exactly as seed is", rt.Args)
	}
}

func TestSidecarReconcileConvergesTheCeilings(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven",
		Args: []string{"sidecar", "run"}, IdleSeconds: 300, MaxRunSeconds: 3600})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	f.reconcile(t, ReconcileConfig{})

	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven",
		Args: []string{"sidecar", "run"}, IdleSeconds: 60, MaxRunSeconds: 120})
	f.reconcile(t, ReconcileConfig{})

	rt := hasRoute(t, f.reg, f.host("bot", "eleven"), "route gone")
	if rt.IdleSeconds != 60 || rt.MaxRunSeconds != 120 {
		t.Fatalf("ceilings = %d/%d, want 60/120 — the sweeper reads the row, so the row must follow the manifest", rt.IdleSeconds, rt.MaxRunSeconds)
	}
}

// TestSidecarReconcileConvergesChannelEnvironmentAndHibernate pins the rest
// of the owned columns: the record owns channel and environment (the #130
// discipline), and the manifest owns hibernation.
func TestSidecarReconcileConvergesChannelEnvironmentAndHibernate(t *testing.T) {
	f := sidecarSetup(t)
	off := false
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven",
		Args: []string{"sidecar", "run"}, Hibernate: &off})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven",
		Channel: "canary/eleven", Environment: "relay"})
	f.reconcile(t, ReconcileConfig{})

	rt := hasRoute(t, f.reg, f.host("bot", "eleven"), "route gone")
	if rt.Channel != "canary/eleven" || rt.Environment != "relay" || rt.Hibernate {
		t.Fatalf("minted route = channel %q env %q hibernate %v", rt.Channel, rt.Environment, rt.Hibernate)
	}

	// Drift the row the way `carlos route` would, then converge it back.
	if err := f.reg.SetChannel(rt.Host, "stable"); err != nil {
		t.Fatal(err)
	}
	if err := f.reg.SetEnvironment(rt.Host, "other"); err != nil {
		t.Fatal(err)
	}
	if err := f.reg.SetHibernate(rt.Host, true); err != nil {
		t.Fatal(err)
	}
	f.reconcile(t, ReconcileConfig{})

	rt = hasRoute(t, f.reg, f.host("bot", "eleven"), "route gone")
	if rt.Channel != "canary/eleven" || rt.Environment != "relay" || rt.Hibernate {
		t.Fatalf("after converge = channel %q env %q hibernate %v — the record and manifest own these columns",
			rt.Channel, rt.Environment, rt.Hibernate)
	}
}

// --- the fail-safe --------------------------------------------------------------

// TestSidecarReconcileSkipsRemovalWhenAReadFailed pins the fail-safe: a
// transient read failure must never mass-retire a tenant's fleet. Minting
// still proceeds, because minting is additive.
func TestSidecarReconcileSkipsRemovalWhenAReadFailed(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	keep := f.host("keep", "eleven")
	f.mustUpsert(t, registry.Route{Host: keep, Kind: "sidecar", Slug: "keep",
		App: "eleven", Account: f.sqid, Provisioned: true})

	// The record IS there; this pass simply cannot read it.
	f.store.failReadsFor(SidecarsPrefix(f.sqid, "eleven"))
	f.reconcile(t, ReconcileConfig{})

	hasRoute(t, f.reg, keep, "a read failure skipped nothing — one bad list must not retire a fleet")
}

// TestSidecarReconcileProtectsAMalformedRecord is the narrower guard: only
// the host whose body could not be trusted is held back, and it is held back
// rather than read as "the bucket retired it" — a non-additive record
// encoding change reaching an older binary would otherwise delete every
// sidecar in scope in one tick.
func TestSidecarReconcileProtectsAMalformedRecord(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	f.mustUpsert(t, registry.Route{Host: host, Kind: "sidecar", Slug: "bot",
		App: "eleven", Account: f.sqid, Provisioned: true})
	if _, err := f.store.Put(ctx, SidecarRecordKey(f.sqid, "eleven", "bot"), []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}

	f.reconcile(t, ReconcileConfig{})

	hasRoute(t, f.reg, host, "a record that cannot be PARSED is not a record that was retired")
}

// TestSidecarReconcileProtectsAChannellessRecord: a route with no channel
// resolves no release, so it has no build to exec — 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, and
// malformed is exactly what protect is for: not minted, and not read as "the
// bucket retired it" either.
func TestSidecarReconcileProtectsAChannellessRecord(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven",
		Args: []string{"sidecar", "run"}, Channel: "stable"})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven"}) // no channel, no row yet
	// A channel-less record whose row already exists: protected, not retired.
	held := f.host("held", "eleven")
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "held", App: "eleven"})
	f.mustUpsert(t, registry.Route{Host: held, Kind: "sidecar", Slug: "held",
		App: "eleven", Account: f.sqid, Channel: "stable", Provisioned: true})
	// A row with no record at all, to prove the removal phase still ran.
	old := f.host("old", "eleven")
	f.mustUpsert(t, registry.Route{Host: old, Kind: "sidecar", Slug: "old",
		App: "eleven", Account: f.sqid, Provisioned: true})

	f.reconcile(t, ReconcileConfig{})

	noRoute(t, f.reg, host, "a channel-less record must not mint a row that can never resolve a release — the manifest's channel is the console door's default at CREATE time, not a runtime rescue")
	hasRoute(t, f.reg, held, "a record that cannot be USED is not a record that was retired")
	noRoute(t, f.reg, old, "protect is narrow: an unusable record holds back ITS OWN host only, and the pass's removal phase still runs")
}

// --- decline-but-keep -------------------------------------------------------------

// TestSidecarReconcileKeepsRowsWhenTheManifestIsAbsent pins the invariant that
// stands between "an app's sidecars.json was deleted" and "every sidecar row
// of that app is retired". An absent manifest is a confirmed answer, not a
// blip, so it does NOT arm the pass-wide fail-safe — which is exactly why the
// record must stay in the record set on its own account. Declining to mint is
// the whole of the response.
//
// The hazard is one line away: manifestFor is consulted in the record scan
// (placement needs it), so a natural-looking `if !mKnown { continue }` there
// would convert a missing manifest into a fleet evacuation.
func TestSidecarReconcileKeepsRowsWhenTheManifestIsAbsent(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	// No sidecars.json for this app at all — the feature is not enabled.
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "new", App: "eleven", Channel: "stable"})
	f.mustUpsert(t, registry.Route{Host: host, Kind: "sidecar", Slug: "bot", App: "eleven",
		Account: f.sqid, Channel: "stable", Provisioned: true, Args: `["sidecar","run"]`})

	f.reconcile(t, ReconcileConfig{})

	rt := hasRoute(t, f.reg, host, "an absent sidecars manifest must not RETIRE a running sidecar — it is the app's opt-in door, not a statement that this row is gone")
	if rt.Args != `["sidecar","run"]` {
		t.Fatalf("Args = %q — with no manifest there is nothing authoritative to converge from, so the row must be left exactly as it was", rt.Args)
	}
	noRoute(t, f.reg, f.host("new", "eleven"), "with no manifest there is no argv to run, so nothing may be minted either")
}

// TestSidecarReconcileKeepsRowsOfATrashedApp: a trashed app mints nothing
// here, and its rows are not this pass's to retire either. domains.Reconcile
// drops every route of a trashed app on its own authority (with its own
// production-flag guard) — this pass merely declines to create, or the two
// loops would fight forever, one re-raising what the other removes.
func TestSidecarReconcileKeepsRowsOfATrashedApp(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "new", App: "eleven", Channel: "stable"})
	f.mustUpsert(t, registry.Route{Host: host, Kind: "sidecar", Slug: "bot", App: "eleven",
		Account: f.sqid, Channel: "stable", Provisioned: true})
	if err := release.TrashApp(ctx, f.store, f.sqid, "eleven", "someone@example.com", time.Now()); err != nil {
		t.Fatal(err)
	}

	f.reconcile(t, ReconcileConfig{})

	hasRoute(t, f.reg, host, "dropping a trashed app's routes is domains.Reconcile's call, with its production-flag guard — this pass only declines to create")
	noRoute(t, f.reg, f.host("new", "eleven"), "a trashed app provisions nothing")
}

// --- placement -------------------------------------------------------------------

// TestSidecarReconcilePlacementGate pins §8's second half: a no-inbound
// workload lands only on a box that declares it accepts no inbound work. Both
// spellings are honoured — the record's own placement and the app manifest's.
func TestSidecarReconcilePlacementGate(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable", Placement: PlacementNoInbound})
	// A second app whose MANIFEST asks for the placement, record silent.
	f.putManifest(t, "relay", SidecarManifest{Schema: 1, App: "relay",
		Args: []string{"relay"}, Placement: PlacementNoInbound})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "hub", App: "relay", Channel: "stable"})

	f.reconcile(t, ReconcileConfig{}) // an ordinary box: accepts inbound work
	noRoute(t, f.reg, f.host("bot", "eleven"), "a no-inbound workload must not land on a box that accepts inbound work")
	noRoute(t, f.reg, f.host("hub", "relay"), "a manifest-stated no-inbound placement must gate exactly as a record-stated one does")

	f.reconcile(t, ReconcileConfig{Inbound: InboundNone})
	hasRoute(t, f.reg, f.host("bot", "eleven"), "it must land on a box that declares CARLOS_INBOUND=none")
	hasRoute(t, f.reg, f.host("hub", "relay"), "it must land on a box that declares CARLOS_INBOUND=none")
}

// TestSidecarReconcilePlacementIsMostRestrictive: a record may NARROW the
// app's placement but never widen it. "any" on a record under a no-inbound
// manifest is the widening direction — the console door refuses to write it
// at all, and this pass must refuse to honour one that reached the bucket by
// any other route, or app-wide policy is only a suggestion.
func TestSidecarReconcilePlacementIsMostRestrictive(t *testing.T) {
	f := sidecarSetup(t)
	// App-wide: everything this app runs sits somewhere with no inbound path.
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven",
		Args: []string{"sidecar", "run"}, Placement: PlacementNoInbound})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable", Placement: PlacementAny})

	f.reconcile(t, ReconcileConfig{}) // an ordinary box: accepts inbound work
	noRoute(t, f.reg, f.host("bot", "eleven"),
		`a record's "any" must not override the app's no-inbound placement — that is the widening direction the console door already refuses`)

	f.reconcile(t, ReconcileConfig{Inbound: InboundNone})
	hasRoute(t, f.reg, f.host("bot", "eleven"), "it still lands where the app's own policy says it may")
}

// The narrowing direction is untouched: one bot that must run with no inbound
// path, under an app that states nothing, is honoured.
func TestSidecarReconcileRecordMayNarrowAnUnplacedApp(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable", Placement: PlacementNoInbound})

	f.reconcile(t, ReconcileConfig{})
	noRoute(t, f.reg, f.host("bot", "eleven"), "a record asking for no-inbound must not land on an inbound box")
	f.reconcile(t, ReconcileConfig{Inbound: InboundNone})
	hasRoute(t, f.reg, f.host("bot", "eleven"), "a record may narrow an app that states no placement of its own")
}

// TestSidecarReconcilePlacementChangeEvacuatesTheRow: placement is a SCOPE
// axis, beside region and pool, so flipping a manifest to no-inbound takes the
// row off a box that accepts inbound work — the same way repointing a
// record's region moves an instance. Decline-but-keep would break the only
// guarantee the axis exists to give: a member who asks that a bot only ever
// run where there is no inbound path would find it still running on an
// inbound box.
func TestSidecarReconcilePlacementChangeEvacuatesTheRow(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	f.reconcile(t, ReconcileConfig{})
	hasRoute(t, f.reg, host, "an unplaced record mints here")

	// The app's owner now requires no inbound path anywhere it runs.
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven",
		Args: []string{"sidecar", "run"}, Placement: PlacementNoInbound})

	f.reconcile(t, ReconcileConfig{}) // this box accepts inbound work
	noRoute(t, f.reg, host, "flipping placement to no-inbound must EVACUATE the row from an inbound box, not leave the bot running there")

	f.reconcile(t, ReconcileConfig{Inbound: InboundNone}) // and land where it belongs
	hasRoute(t, f.reg, host, "the same record must mint on a box that declares CARLOS_INBOUND=none")
}

// TestSidecarReconcilePlacementEvacuationObeysTheFailSafe: evacuation is a
// removal, so it is governed by the pass-wide fail-safe like every other
// removal. A read that failed anywhere this pass means this box cannot tell a
// placement change from a blip, and a transient blip must never evacuate a
// tenant's fleet.
func TestSidecarReconcilePlacementEvacuationObeysTheFailSafe(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven",
		Args: []string{"sidecar", "run"}, Placement: PlacementNoInbound})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	f.mustUpsert(t, registry.Route{Host: host, Kind: "sidecar", Slug: "bot",
		App: "eleven", Account: f.sqid, Provisioned: true})
	// An UNRELATED app's records cannot be read this pass.
	f.putManifest(t, "relay", SidecarManifest{Schema: 1, App: "relay", Args: []string{"relay"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "hub", App: "relay", Channel: "stable"})
	f.store.failReadsFor(SidecarsPrefix(f.sqid, "relay"))

	f.reconcile(t, ReconcileConfig{}) // placement does not match, but a read failed

	hasRoute(t, f.reg, host, "a pass that could not read everything must evacuate nothing — the fail-safe governs placement removals too")
}

// --- the co-spawn conflict ---------------------------------------------------------

// TestSidecarReconcileRefusesHostWithAStagedCoSpawnFile: the file-staged
// co-spawn is legacy but LIVE (Eleven runs on it today). One host, one
// mechanism.
func TestSidecarReconcileRefusesHostWithAStagedCoSpawnFile(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	if err := os.WriteFile(filepath.Join(f.dir, host+"-sidecar.env"), []byte("X=1\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})

	f.reconcile(t, ReconcileConfig{})

	noRoute(t, f.reg, host, "a host with a staged legacy co-spawn file must not also get a native sidecar row — one host, one mechanism")
}

// --- scope ---------------------------------------------------------------------------

// TestSidecarReconcileSkipsAnotherBoxesScope: region and pool scope the
// sidecar pass exactly as they scope the instance one — a record for another
// agent's jurisdiction is neither minted nor (for the row) removed here.
func TestSidecarReconcileSkipsAnotherBoxesScope(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable", Region: "ie", Pool: "eleven"})
	// A foreign box's own provisioned sidecar row, visible in this registry.
	foreign := f.host("far", "eleven")
	f.mustUpsert(t, registry.Route{Host: foreign, Kind: "sidecar", Slug: "far", App: "eleven",
		Account: f.sqid, Region: "de", Pool: "other", Provisioned: true})

	f.reconcile(t, ReconcileConfig{}) // this box: no region, no pools

	noRoute(t, f.reg, f.host("bot", "eleven"), "a record for another region/pool is not this pass's business")
	hasRoute(t, f.reg, foreign, "a foreign box's provisioned route is not this pass's to retire")

	f.reconcile(t, ReconcileConfig{Region: "ie", Pools: []string{"eleven"}})
	hasRoute(t, f.reg, f.host("bot", "eleven"), "the in-scope box must mint it")
}

// TestSidecarReconcileDryRunWritesNothing mirrors the instance pass's §4.3
// mitigation: a pass that reports is a pass that writes nothing at all.
func TestSidecarReconcileDryRunWritesNothing(t *testing.T) {
	f := sidecarSetup(t)
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	gone := f.host("gone", "eleven")
	f.mustUpsert(t, registry.Route{Host: gone, Kind: "sidecar", Slug: "gone",
		App: "eleven", Account: f.sqid, Provisioned: true})

	f.reconcile(t, ReconcileConfig{DryRun: true})

	noRoute(t, f.reg, f.host("bot", "eleven"), "dry-run minted a route")
	hasRoute(t, f.reg, gone, "dry-run removed a route")
}

// TestSidecarReconcileLeavesAForeignRowAlone: a row this pass does not own at
// a host a record names is left exactly as alone as before. There is no
// adoption path for sidecars — the suffix is reserved and refused at every
// door, so nothing but this pass can have minted one — and adoption is an
// authority transfer, never a shape coincidence.
func TestSidecarReconcileLeavesAForeignRowAlone(t *testing.T) {
	f := sidecarSetup(t)
	host := f.host("bot", "eleven")
	f.putManifest(t, "eleven", SidecarManifest{Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}})
	f.putRecord(t, SidecarRecord{Schema: 1, Name: "bot", App: "eleven", Channel: "stable"})
	f.mustUpsert(t, registry.Route{Host: host, Kind: "sidecar", Slug: "bot",
		App: "eleven", Account: f.sqid, Args: `["hand","written"]`})

	f.reconcile(t, ReconcileConfig{})

	rt := hasRoute(t, f.reg, host, "route gone")
	if rt.Provisioned {
		t.Fatal("an un-provisioned row must not be adopted by this pass — adoption grants removal authority")
	}
	if rt.Args != `["hand","written"]` {
		t.Fatalf("Args = %q — a row this pass does not own must be left alone", rt.Args)
	}
}