package main

// The activator (issue #550 Part 4 v2, Phase 2): instances hibernate. When the
// router is given an S3 bucket (LCHAT_HIBERNATE_BUCKET) it stops relying on
// systemd to keep instance processes alive and instead runs them as its own
// children, on demand — "units are for the router, instances are its
// children". A request for a host with no live process claims the per-host S3
// single-writer lease, restores the DB from S3 if the local copy is missing,
// spawns the instance, and proxies; after an idle window the instance is
// checkpointed back to S3, the lease released, and the process stopped. State
// parks in S3 as ciphertext (an instance DB is opaque blobs plus routing
// metadata the server already sees — the E2EE rule makes the bucket
// threat-model-free); processes exist only while someone's talking.
//
// Scope (this phase): ONE box. No gossip/HRW placement, no peer proxying —
// membership of one, hash picks self, leases uncontested except against a
// previous incarnation of ourselves.
//
// Bot sidecar co-spawn (a former non-goal, now in): a host with a sidecar
// staged (/data/<host>-bot.env — the same file lchat-bot@.service execs
// against) gets its `bot run` child spawned right after the instance turns
// healthy, and SIGTERMed first at sleep — the sidecar exists exactly while its
// instance is awake, so its lifecycle solves itself. Its own connections flow
// back through this router (the box pins its hosted names to loopback), so the
// sidecar marks them (bot.SidecarHeader) and the activator excludes them from
// the idle clock — otherwise the pair could never sleep, and an orphaned
// sidecar's reconnect loop would wake the instance right back (see serveHTTP).
// Member-bot charters (docs/member-bots.md) run IN the sidecar, and their
// schedules live in blobs the server can't read — the former honest gap. It's
// closed by charter-wake registration: the sidecar reports its earliest
// charter run to the instance (POST /api/charter-next-due), the instance folds
// it into GET /api/next-due (combineNextDue, miniapps.go), and the due-time
// index below wakes for it like any scheduled delivery — no activator changes.
// The hardened dedicated-sidecar tier (bot keys off-box) stays a non-goal here.
//
// Opt-in is PER HOST (routes.hibernate, registry.go): with the bucket env set,
// only flagged instances are activator-managed; the rest stay on their systemd
// units and are proxied straight. That's what makes a production canary safe —
// the env ships fleet-wide as a no-op, then `lchat hibernate <host> --on`
// (control.go) migrates exactly one instance: its unit stops, the flag flips,
// and the first request wakes it here.
//
// The due-time index (a former non-goal, now in): scheduled future-messages/
// reminders fire on time even while their instance sleeps. At sleep the stop
// path asks the instance for its earliest pending delivery (GET /api/next-due
// over the socket, admin-gated) and records it in the registry (routes.next_due
// — persisted so it survives router restarts); the sweeper wakes any sleeping
// host whose time has arrived, with no client request. Once up, the instance's
// own scheduler (miniapps.go runScheduler, a 5s tick) delivers, and the normal
// idle window puts it back to sleep — which re-records the NEXT due time, so
// the cycle maintains itself. Worst-case lateness ≈ one activator sweep
// interval (≤30s) + spawn + one scheduler tick (≤5s).

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/benbjohnson/litestream"

	"lchat/internal/activator"
	"lchat/internal/bot"
)

// hibernateConfig is the router-process opt-in, read from env. Nil (bucket
// unset) keeps today's behavior byte-identical: instances run under systemd,
// the router only proxies.
type hibernateConfig struct {
	Bucket string        // LCHAT_HIBERNATE_BUCKET (required to enable)
	Region string        // LCHAT_HIBERNATE_REGION (optional; AWS default chain resolves "")
	Prefix string        // LCHAT_HIBERNATE_PREFIX (optional key prefix inside the bucket)
	Idle   time.Duration // LCHAT_HIBERNATE_IDLE (default 15m; seconds-granular, e.g. "90s")
}

func hibernateConfigFromEnv() (*hibernateConfig, error) {
	bucket := strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_BUCKET"))
	if bucket == "" {
		return nil, nil
	}
	cfg := &hibernateConfig{
		Bucket: bucket,
		Region: strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_REGION")),
		Prefix: strings.Trim(strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_PREFIX")), "/"),
		Idle:   15 * time.Minute,
	}
	if v := strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_IDLE")); v != "" {
		d, err := time.ParseDuration(v)
		if err != nil || d <= 0 {
			return nil, fmt.Errorf("LCHAT_HIBERNATE_IDLE %q: want a positive duration like 15m or 90s", v)
		}
		cfg.Idle = d
	}
	return cfg, nil
}

// hibernatePrefix is the per-instance S3 key prefix: <LCHAT_HIBERNATE_PREFIX>/<host>/
// (the lease's lock.json is colocated under it by the activator package).
func (c *hibernateConfig) hibernatePrefix(host string) string {
	return path.Join(c.Prefix, host)
}

// meshPrefix is where the mesh's membership heartbeats live, beside the
// per-instance prefixes — the bucket is the mesh's only shared substrate.
// It can't collide with hibernatePrefix: instance prefixes are public FQDNs
// (they contain a dot), and the bare label "mesh" never is one.
func (c *hibernateConfig) meshPrefix() string {
	return path.Join(c.Prefix, "mesh/members")
}

// --- backend seam ------------------------------------------------------------
//
// The state machine's side effects (S3 lease, restore, process spawn) sit
// behind these interfaces so the machine itself is unit-testable with fakes —
// wake coalescing, idle→sleep, and lease-lost stand-down need no AWS.

type actBackend interface {
	// acquire claims the per-host single-writer lease. Contention surfaces as
	// errLeaseHeld so the router can answer 503 + Retry-After.
	acquire(ctx context.Context, host string) (actLease, error)
	// restore materializes the local DB from S3 when it's absent (a no-op when
	// the warm-cache copy is still on disk).
	restore(ctx context.Context, host, dbPath string) error
	// start spawns the instance process and waits until its socket accepts. It
	// also returns the build it execed (the live/<host> symlink target read at
	// spawn), so the sweep can later spot a repoint under a still-running child
	// and converge it. Empty when unknown — the sweep treats that as "don't
	// cycle" (fail-safe).
	start(ctx context.Context, rt Route) (actProc, string, error)
	// nextDue asks a running instance for its earliest pending scheduled
	// delivery (unix seconds; 0 = nothing scheduled). Queried at sleep time,
	// while the socket still answers, to feed the due-time index.
	nextDue(ctx context.Context, rt Route) (int64, error)
	// hasSidecar reports whether host has a bot sidecar staged on this box.
	hasSidecar(host string) bool
	// startSidecar spawns the host's bot sidecar. Fire-and-forget: a sidecar
	// is a pure outbound client with no socket, so there's nothing to
	// health-check — its own connect/backoff loop recovers from a slow start.
	startSidecar(ctx context.Context, rt Route) (actProc, error)
}

// actDueStore persists the due-time index — the registry in production
// (routes.next_due), so a sleeping host's timer survives router restarts
// instead of dying with the process.
type actDueStore interface {
	setNextDue(host string, due int64) error
	dueInstances(now int64) ([]Route, error)
}

// actLease is a held-but-not-yet-replicating lease (wake in progress).
type actLease interface {
	// replicate trades the lease for running replication of the (now present)
	// local DB; the lease handle is spent on success. syncInterval is the
	// route's replication-class tick (replicationSyncInterval; 0 = stream).
	replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (actRep, error)
	// release gives the lease up — the bail-out when a wake fails partway.
	release(ctx context.Context) error
}

// actRep is running replication coupled to the lease.
type actRep interface {
	// Err reports lost leadership (a lease renewal failed).
	Err() error
	// Close does the final checkpoint/sync and releases the lease.
	Close(ctx context.Context) error
}

// actProc is a running instance child process.
type actProc interface {
	// stop terminates gracefully (SIGTERM, bounded wait, then SIGKILL).
	stop(ctx context.Context) error
	// exited reports whether the process has died on its own.
	exited() bool
}

// --- state machine -----------------------------------------------------------

// Per-host lifecycle. An absent entry is the fourth state: asleep.
//
//	asleep --request--------------------> waking   (one wake; concurrent
//	waking --wake ok--------------------> awake     requests coalesce on it)
//	waking --wake failed----------------> asleep   (error served; next request retries)
//	awake  --idle timeout / lease lost /
//	          child died----------------> stopping (final checkpoint, release lease)
//	stopping --stop done----------------> asleep   (local DB left as a warm cache)
//
// A request that lands on waking waits (bounded) for the wake; one that lands
// on stopping waits for the stop, then re-enters as a fresh wake.
type actState int

const (
	actWaking actState = iota
	actAwake
	actStopping
)

type actEntry struct {
	host  string
	route Route
	state actState

	ready   chan struct{} // waking: closed when the wake resolves; then check err
	err     error         // waking: the wake's outcome
	stopped chan struct{} // stopping: closed when the stop completes

	// Activity tracking: an instance is idle only when nothing is in flight
	// AND the idle window has passed since the last request finished. In-flight
	// counting (not just timestamps) is what keeps an instance with one open
	// WebSocket awake — the proxy's ServeHTTP blocks for the tunnel's life.
	// (Sidecar-marked requests register in neither — see beginSidecarRequest.)
	inflight   int
	lastActive time.Time

	rep  actRep
	proc actProc

	// Convergence: the `live/<host>` symlink target this child was spawned
	// against. A running process never re-reads the symlink, so once a ship
	// (or `deploy --version`) repoints it, an awake child keeps serving the old
	// build until it's cycled. spawnVersion lets the sweep notice the drift;
	// staleSince is when it first did, so a brief transient (a symlink moved
	// then rolled straight back) doesn't trigger a needless cycle. Both empty/
	// zero until the wake resolves.
	spawnVersion string
	staleSince   time.Time

	// Bot sidecar co-spawn: the sidecar exists exactly while its instance is
	// awake. All sidecar fields are written under a.mu in the awake state only
	// (wake's attach, the sweeper's restart); a stop flips the state under the
	// same lock first, so stop() owns them uncontended from then on.
	sidecar        actProc       // the co-spawned `bot run` child; nil = none running
	sidecarWanted  bool          // the host had a sidecar staged at wake time
	sidecarStarted time.Time     // when the current sidecar spawned (backoff reset)
	sidecarBackoff time.Duration // last restart delay (doubles, capped)
	sidecarRetryAt time.Time     // earliest next restart attempt
}

// Sidecar restart pacing — Restart=always semantics without a supervisor: the
// sweeper redoes a dead sidecar with exponential backoff from the systemd
// unit's RestartSec, capped, and a run that survived sidecarHealthyRun earns a
// fresh backoff (so one crash a day never escalates to the cap).
const (
	sidecarRestartBase = 2 * time.Second
	sidecarRestartMax  = time.Minute
	sidecarHealthyRun  = time.Minute
)

// sidecarStopTimeout bounds the sidecar's stop at sleep time. `bot run` has no
// drain handler, so the SIGTERM is effectively immediate; the SIGKILL fallback
// behind this timeout is insurance.
const sidecarStopTimeout = 5 * time.Second

// Build convergence: how long an awake child may serve a symlink-stale binary
// before the sweep gracefully cycles it, and how many to cycle per tick.
//   - convergeGrace comfortably exceeds one sweep interval (≤30s) and the
//     sidecar's ~32s reconnect, so a genuinely-persistent drift is required —
//     a symlink that moved then rolled straight back never trips it, and a
//     briefly-busy space gets a chance to idle-hibernate (which converges for
//     free) first.
//   - one per tick staggers a fleet-wide ship across sweeps instead of blipping
//     every space's connections at once.
//
// Convergence is a latency/observability fix, not urgent (the supply-chain bake
// window already holds releases 72h), so a leisurely, gentle cadence is right.
const (
	convergeGrace             = 90 * time.Second
	maxConvergeCyclesPerSweep = 1
)

// convergeDecision is the pure state machine behind build convergence: given the
// symlink target read this tick (cur/readErr), the build the child execed
// (spawnVer), and when the drift was first seen (staleSince), it says whether to
// cycle now and returns the staleSince to carry forward. Split out so the grace/
// reset logic is unit-testable without a live activator. Three cases:
//   - read error: unknown — neither cycle nor clear the grace; retry next tick.
//   - converged / rolled-back / nothing to compare (cur=="" or spawnVer=="" or
//     cur==spawnVer): clear the grace clock, don't cycle.
//   - stale (symlink moved since spawn): start the grace clock on first sight,
//     cycle once the drift has persisted past grace.
func convergeDecision(cur string, readErr error, spawnVer string, staleSince, now time.Time, grace time.Duration) (cycle bool, nextStale time.Time) {
	switch {
	case readErr != nil:
		return false, staleSince
	case cur == "" || spawnVer == "" || cur == spawnVer:
		return false, time.Time{}
	default:
		if staleSince.IsZero() {
			staleSince = now
		}
		return now.Sub(staleSince) >= grace, staleSince
	}
}

// wakeWaitBound caps how long a request queues on someone else's wake or stop
// before giving up with 503 — a wedged wake must not pile up connections
// forever. A var so tests can shrink it.
var wakeWaitBound = 30 * time.Second

// errLeaseHeld: the per-host lease is held elsewhere. On one box this means a
// previous incarnation hasn't released (crashed router; lease expires within
// ~LeaseTTL) — the router answers 503 with Retry-After.
type errLeaseHeld struct {
	holder     string
	retryAfter time.Duration
}

func (e *errLeaseHeld) Error() string {
	return fmt.Sprintf("instance lease held by %q (retry in %s)", e.holder, e.retryAfter)
}

var errWakeTimeout = errors.New("timed out waiting for instance wake")

type activatorSvc struct {
	cfg     hibernateConfig
	backend actBackend
	due     actDueStore
	now     func() time.Time             // a seam for the unit tests' clock
	liveVer func(string) (string, error) // live/<host> symlink target; a seam for tests

	// mesh, when non-nil, is the multi-box placement layer (mesh.go): requests
	// for hosts whose HRW owner is another live box are proxied there, and the
	// due-time sweeper only wakes hosts this box owns. Nil (LCHAT_MESH_SELF
	// unset) is a single box — hashing would pick self everywhere, so no mesh
	// codepath is touched at all.
	mesh *meshSvc

	mu      sync.Mutex
	entries map[string]*actEntry

	sweepStop chan struct{}
	sweepDone chan struct{}
}

func newActivator(cfg hibernateConfig, backend actBackend, due actDueStore) *activatorSvc {
	return &activatorSvc{
		cfg:     cfg,
		backend: backend,
		due:     due,
		now:     time.Now,
		liveVer: liveVersion,
		entries: make(map[string]*actEntry),
	}
}

// beginRequest ensures the host's instance is awake and registers one in-flight
// request against it. The returned release must be called when the request
// finishes (it stamps last-activity, which starts the idle clock).
func (a *activatorSvc) beginRequest(ctx context.Context, rt Route) (release func(), err error) {
	deadline := time.NewTimer(wakeWaitBound)
	defer deadline.Stop()
	for {
		a.mu.Lock()
		e := a.entries[rt.Host]
		if e == nil {
			// Asleep → kick off ONE wake; this request and any latecomers all
			// coalesce on e.ready rather than racing a second lease claim
			// (singleflight). The wake runs detached so even its initiator is
			// held to the same bounded wait as everyone queued behind it.
			e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
			a.entries[rt.Host] = e
			a.mu.Unlock()
			go a.wake(e)
			continue
		}

		switch e.state {
		case actAwake:
			// Stand down synchronously-detected failure: a dead child or a
			// lost lease found at request time turns into a stop, and the
			// request waits for it then re-wakes — recovery is the routine
			// wake path.
			if repErr := e.rep.Err(); repErr != nil || e.proc.exited() {
				if repErr != nil {
					log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
				} else {
					log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
				}
				a.beginStopLocked(e)
				a.mu.Unlock()
				continue
			}
			e.inflight++
			a.mu.Unlock()
			return a.releaseFunc(e), nil

		case actWaking:
			ready := e.ready
			a.mu.Unlock()
			select {
			case <-ready:
				// A failed wake reports to every coalesced waiter (retrying
				// here would turn one bad wake into a storm); a successful
				// one loops to register against the now-awake entry.
				if e.err != nil {
					return nil, e.err
				}
				continue
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-deadline.C:
				return nil, errWakeTimeout
			}

		case actStopping:
			stopped := e.stopped
			a.mu.Unlock()
			select {
			case <-stopped:
				continue // entry is gone; loop wakes it fresh
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-deadline.C:
				return nil, errWakeTimeout
			}
		}
	}
}

// errSidecarAsleep: a sidecar-marked request arrived for a host with no awake
// instance. Sidecar traffic never wakes anything — the activator itself spawns
// the sidecar as part of a wake, so a marked request for a sleeping host is
// either a stray reconnect racing a stop or an outsider spoofing the marker;
// both get a 503 and change nothing.
var errSidecarAsleep = errors.New("instance asleep; sidecar requests do not wake it")

// beginSidecarRequest is beginRequest for sidecar-marked traffic: pass it
// through to an awake instance WITHOUT registering activity — no in-flight
// count, no lastActive stamp (the sidecar's own long-lived WebSocket must not
// hold its instance awake; that non-count is the whole reason the pair can
// sleep) — and refuse, rather than wake, when the instance is asleep or
// stopping.
func (a *activatorSvc) beginSidecarRequest(ctx context.Context, rt Route) error {
	deadline := time.NewTimer(wakeWaitBound)
	defer deadline.Stop()
	for {
		a.mu.Lock()
		e := a.entries[rt.Host]
		if e == nil {
			a.mu.Unlock()
			return errSidecarAsleep
		}
		switch e.state {
		case actAwake:
			// Same dead-child / lost-lease detection as a member request, but
			// the recovery stops at the stop: the loop then finds a stopping
			// (or absent) entry and turns this request away instead of
			// chaining into a re-wake.
			if repErr := e.rep.Err(); repErr != nil || e.proc.exited() {
				if repErr != nil {
					log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
				} else {
					log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
				}
				a.beginStopLocked(e)
				a.mu.Unlock()
				continue
			}
			a.mu.Unlock()
			return nil

		case actWaking:
			// Joining a wake someone else started is fine — the co-spawned
			// sidecar boots moments before its entry flips awake, so its first
			// dial can land in that sliver; joining can't initiate anything.
			// A failed wake just reads as asleep (the member path already
			// logged it).
			ready := e.ready
			a.mu.Unlock()
			select {
			case <-ready:
				if e.err != nil {
					return errSidecarAsleep
				}
				continue
			case <-ctx.Done():
				return ctx.Err()
			case <-deadline.C:
				return errWakeTimeout
			}

		default: // actStopping — the stop exists to shed exactly this connection
			a.mu.Unlock()
			return errSidecarAsleep
		}
	}
}

// hasLiveEntry reports whether host has an entry in any state — the mesh's
// "already living here" check (waking and stopping both count: a waking host
// is committed to this box, and a stopping one still holds the lease, so
// proxying it away would only trade a short wait for a loud lease race).
func (a *activatorSvc) hasLiveEntry(host string) bool {
	a.mu.Lock()
	defer a.mu.Unlock()
	_, ok := a.entries[host]
	return ok
}

// awakeCount is this box's own count of genuinely serving hibernating
// children — actAwake only, not actWaking/actStopping transients — for the
// Fleet status page's per-box awake/asleep split (meshstats.go). Unlike
// deploy/status-publish.sh's own `ss`-probing of the LOCAL box (an external
// process with no other view into the router), this is the activator's
// authoritative in-memory state — no socket probing needed.
func (a *activatorSvc) awakeCount() int {
	a.mu.Lock()
	defer a.mu.Unlock()
	n := 0
	for _, e := range a.entries {
		if e.state == actAwake {
			n++
		}
	}
	return n
}

func (a *activatorSvc) releaseFunc(e *actEntry) func() {
	var once sync.Once
	return func() {
		once.Do(func() {
			a.mu.Lock()
			e.inflight--
			e.lastActive = a.now()
			a.mu.Unlock()
		})
	}
}

// wake drives asleep→awake: lease, restore-if-missing, spawn, replicate,
// co-spawn the sidecar. It runs on the first requester's goroutine but under
// its own generous timeout, detached from that request's context — queued
// requests still need the wake to finish even if the requester who started it
// hangs up.
func (a *activatorSvc) wake(e *actEntry) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	w, err := a.doWake(ctx, e.route)

	a.mu.Lock()
	if err != nil {
		e.err = err
		delete(a.entries, e.host) // next request retries from scratch
	} else {
		e.state = actAwake
		e.rep, e.proc = w.rep, w.proc
		e.sidecar, e.sidecarWanted = w.sidecar, w.sidecarWanted
		e.sidecarStarted = a.now()
		e.sidecarBackoff = 0
		e.sidecarRetryAt = time.Time{} // a wake-time spawn failure retries on the next sweep
		e.lastActive = a.now()
		// The build this child actually execed, captured at Start() inside the
		// backend (not re-read here — a ship during the wake window would make a
		// re-read name the new target while the process runs the old binary, and
		// the sweep would never converge it). Empty = unknown → the sweep won't
		// cycle. Fresh wake, so it can't be stale yet.
		e.spawnVersion = w.spawnVersion
		e.staleSince = time.Time{}
	}
	a.mu.Unlock()
	close(e.ready)
}

// wokenInstance is a successful wake's yield. sidecar is nil when the host has
// none staged — or when the spawn failed, which is deliberately not an error:
// members > bots, so a space whose bots won't start still serves (the sweeper
// retries via sidecarWanted).
type wokenInstance struct {
	rep           actRep
	proc          actProc
	sidecar       actProc
	sidecarWanted bool
	spawnVersion  string // live/<host> target read at exec — for build convergence
}

func (a *activatorSvc) doWake(ctx context.Context, rt Route) (*wokenInstance, error) {
	host := rt.Host
	dbPath := rt.DBPath
	if dbPath == "" {
		dbPath = filepath.Join(instanceDataDir, host+".db")
	}

	// 1. The lease first: everything after this line is done as the prefix's
	// single writer. Contention is not retried here — on one box it means a
	// previous incarnation hasn't released; tell the client to come back after
	// the TTL instead of stalling its request.
	lease, err := a.backend.acquire(ctx, host)
	if err != nil {
		var held *errLeaseHeld
		if errors.As(err, &held) {
			log.Printf("activator: LEASE HELD for %s by %q — a previous incarnation hasn't released; refusing to wake (retry after %s)",
				host, held.holder, held.retryAfter)
		}
		return nil, err
	}

	// 2. Restore the DB from S3 when the local copy is gone. When it's still
	// on disk it is the warm cache: on one box, with the lease held, the local
	// copy can only be what we ourselves checkpointed at the last sleep.
	if err := a.backend.restore(ctx, host, dbPath); err != nil {
		_ = lease.release(ctx)
		return nil, fmt.Errorf("restore %s: %w", host, err)
	}

	// 3. Spawn before replication starts: a fresh instance creates its own DB
	// on boot, so waiting for the socket guarantees the file exists for
	// litestream to open. Writes that land in the gap are picked up by the
	// replica's first sync.
	proc, spawnVer, err := a.backend.start(ctx, rt)
	if err != nil {
		_ = lease.release(ctx)
		return nil, fmt.Errorf("start %s: %w", host, err)
	}

	// 4. Trade the held lease for running replication, at the route's
	// replication-class sync interval (the per-plan durability ladder).
	rep, err := lease.replicate(ctx, dbPath, replicationSyncInterval(rt.Replication))
	if err != nil {
		stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
		defer cancel()
		_ = proc.stop(stopCtx)
		_ = lease.release(ctx)
		return nil, fmt.Errorf("replicate %s: %w", host, err)
	}

	// 5. Co-spawn the bot sidecar, last — only once the instance is healthy
	// and replicating, so the sidecar always boots against a live instance
	// (its marked requests pass through the router while we're awake). A
	// spawn failure is loud but never fails the wake: members come before
	// bots, and the sweeper retries with backoff.
	w := &wokenInstance{rep: rep, proc: proc, spawnVersion: spawnVer}
	if a.backend.hasSidecar(host) {
		w.sidecarWanted = true
		if sc, err := a.backend.startSidecar(ctx, rt); err != nil {
			log.Printf("activator: SIDECAR for %s failed to start: %v — serving without bots (the sweep retries)", host, err)
		} else {
			w.sidecar = sc
		}
	}
	return w, nil
}

// actStopTimeout bounds a graceful instance stop (the instance's own SIGTERM
// drain is 10s; leave headroom before the SIGKILL).
const actStopTimeout = 20 * time.Second

// beginStopLocked flips an awake entry to stopping and kicks the stop off; the
// caller holds a.mu. In-flight requests are not waited for — a stop is only
// initiated when the instance is idle, dead, or has lost its lease, and in the
// latter two cases open requests are already doomed.
func (a *activatorSvc) beginStopLocked(e *actEntry) {
	e.state = actStopping
	e.stopped = make(chan struct{})
	go a.stop(e)
}

// stop drives awake→asleep: stop the child (its SIGTERM drain is the final
// write), checkpoint what's on disk to S3 and release the lease, then forget
// the entry. The local DB stays in place — it's the warm cache for the next
// wake; S3 is the durable truth under the lease.
func (a *activatorSvc) stop(e *actEntry) {
	// Co-stop, sidecar FIRST — before the next-due probe and the instance's
	// own SIGTERM. Two reasons: the sidecar must never outlive its instance
	// (an orphan's reconnect loop would be a standing wake source — the
	// ping-pong this feature exists to prevent — and its member-bot charters
	// would keep firing into a host we're about to stop), and with the bots
	// gone first, the schedule the next-due probe reads can't shift under it.
	if e.sidecar != nil {
		scCtx, scCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
		if err := e.sidecar.stop(scCtx); err != nil {
			log.Printf("activator: stop sidecar %s: %v", e.host, err)
		}
		scCancel()
	}

	// The due-time index: ask the instance for its earliest pending scheduled
	// delivery BEFORE stopping it — the socket dies with the process. (When the
	// stop was triggered by a dead child, the query just fails fast.) Any
	// failure records 0, i.e. "sleeps until a request": a missed probe must
	// degrade to pre-index behavior, never block or crash the sleep. A due time
	// already in the past is recorded as-is and the sweeper wakes the host on
	// its next tick — there is deliberately no minimum-sleep guard beyond that,
	// because thrash is already impossible: an awake instance only sleeps after
	// a full idle window, so wake/sleep cycles can never be spaced closer than
	// cfg.Idle.
	dueCtx, cancelDue := context.WithTimeout(context.Background(), 5*time.Second)
	due, err := a.backend.nextDue(dueCtx, e.route)
	cancelDue()
	if err != nil {
		log.Printf("activator: next-due probe %s: %v — recording none (it sleeps until a request)", e.host, err)
		due = 0
	}
	if err := a.due.setNextDue(e.host, due); err != nil {
		log.Printf("activator: record next-due %s: %v", e.host, err)
	} else if due > 0 {
		log.Printf("activator: %s next scheduled delivery at %s — will wake for it", e.host, time.Unix(due, 0).UTC().Format(time.RFC3339))
	}

	// Separate budgets: a SIGTERM drain that runs long must not eat the final
	// checkpoint's time — the S3 sync is the part that makes sleep durable.
	procCtx, cancelProc := context.WithTimeout(context.Background(), actStopTimeout)
	err = e.proc.stop(procCtx)
	cancelProc()
	if err != nil {
		log.Printf("activator: stop %s: %v", e.host, err)
	}
	// Close = final sync + lease release. After a lost lease the release part
	// fails harmlessly (someone else owns the lock); the sync still preserves
	// a local checkpoint.
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	if err := e.rep.Close(ctx); err != nil {
		log.Printf("activator: final checkpoint %s: %v", e.host, err)
	}
	a.mu.Lock()
	delete(a.entries, e.host)
	a.mu.Unlock()
	close(e.stopped)
	log.Printf("activator: %s hibernated", e.host)
}

// start launches the background sweeper that puts idle instances to sleep and
// stands down lost leases/dead children between requests.
func (a *activatorSvc) start() {
	a.sweepStop = make(chan struct{})
	a.sweepDone = make(chan struct{})
	// Sweep well inside the idle window so a sleep happens promptly after it
	// elapses; clamp so second-granular test idles still sweep sanely.
	interval := a.cfg.Idle / 4
	if interval < 100*time.Millisecond {
		interval = 100 * time.Millisecond
	}
	if interval > 30*time.Second {
		interval = 30 * time.Second
	}
	go func() {
		defer close(a.sweepDone)
		t := time.NewTicker(interval)
		defer t.Stop()
		for {
			select {
			case <-a.sweepStop:
				return
			case <-t.C:
				a.sweepOnce()
			}
		}
	}()
}

// sweepOnce examines every awake instance once: idle ones hibernate; a lost
// lease or dead child stands down immediately regardless of activity. It then
// checks the due-time index and wakes any sleeping host whose scheduled
// delivery has come due.
func (a *activatorSvc) sweepOnce() {
	now := a.now()
	// The due query runs before the lock — it's a registry read, and nothing
	// under a.mu should wait on SQLite.
	dueRoutes, err := a.due.dueInstances(now.Unix())
	if err != nil {
		log.Printf("activator: due-time sweep: %v", err)
	}
	a.mu.Lock()
	cycled := 0 // build-convergence cycles started this tick (rate-limited)
	for _, e := range a.entries {
		if e.state != actAwake {
			continue
		}
		if err := e.rep.Err(); err != nil {
			log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, err)
			a.beginStopLocked(e)
			continue
		}
		if e.proc.exited() {
			log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
			a.beginStopLocked(e)
			continue
		}
		if e.inflight == 0 && now.Sub(e.lastActive) >= a.cfg.Idle {
			log.Printf("activator: %s idle %s — hibernating", e.host, now.Sub(e.lastActive).Truncate(time.Second))
			a.beginStopLocked(e)
			continue
		}
		// Converge onto a shipped build. A running child never re-reads
		// live/<host>, so once a ship or `deploy --version` repoints the symlink
		// this awake child is serving a stale binary. Idle spaces converge for
		// free on their next wake; this cycle is for the always-awake ones —
		// notably every bot-sidecar host, which never sleeps — that otherwise
		// serve the old build indefinitely. Gracefully cycle: beginStopLocked
		// drains via SIGTERM, takes the final checkpoint, releases the lease, and
		// leaves the host asleep; a member request (or the due-time index for
		// scheduled charters) then wakes it onto the new binary — the same clean
		// cycle `systemctl restart` gives a unit-managed host. Unlike idle
		// hibernation this can fire with a request in flight; the stop still
		// drains HTTP gracefully and reconnecting clients coalesce onto the fresh
		// wake, so a long upload past the ~10s drain is cut exactly as a restart
		// would cut it. A pinned canary never trips this: the updater leaves its
		// symlink alone, so spawnVersion still matches. See
		// docs/awake-instance-convergence.md.
		cur, readErr := a.liveVer(e.host)
		cycle, next := convergeDecision(cur, readErr, e.spawnVersion, e.staleSince, now, convergeGrace)
		e.staleSince = next
		if cycle && cycled < maxConvergeCyclesPerSweep {
			log.Printf("activator: %s serving %s but live symlink is now %s — cycling onto the new build", e.host, e.spawnVersion, cur)
			a.beginStopLocked(e)
			cycled++
			continue
		}
		// Sidecar supervision — Restart=always, sweep-driven, no supervisor
		// goroutine: a sidecar that died (or failed to spawn at wake) while
		// its instance stays awake is respawned with doubling, capped backoff;
		// a run past sidecarHealthyRun resets it. Spawned under a.mu
		// deliberately: an exec is quick, and the lock is what makes a restart
		// impossible to race with a stop (sidecar fields only change in the
		// awake state, and every stop flips the state under this same lock
		// first).
		//
		// A bot env staged MID-FLIGHT (attach-bot on an awake child — the AI
		// opt-in, #858) is ADOPTED here rather than waiting for the next
		// wake, so opting in takes effect in seconds. Safe against
		// double-running: attach-bot never enables the systemd unit for a
		// hibernating host, and only hibernating hosts are activator
		// children — the staged sidecar has exactly one possible runner.
		if !e.sidecarWanted && a.backend.hasSidecar(e.host) {
			e.sidecarWanted = true
			log.Printf("activator: sidecar staged for awake %s — adopting (AI opt-in)", e.host)
		}
		if e.sidecarWanted && (e.sidecar == nil || e.sidecar.exited()) && !now.Before(e.sidecarRetryAt) {
			if e.sidecar != nil && now.Sub(e.sidecarStarted) >= sidecarHealthyRun {
				e.sidecarBackoff = 0
			}
			if e.sidecarBackoff == 0 {
				e.sidecarBackoff = sidecarRestartBase
			} else {
				e.sidecarBackoff = min(e.sidecarBackoff*2, sidecarRestartMax)
			}
			e.sidecarRetryAt = now.Add(e.sidecarBackoff)
			sc, err := a.backend.startSidecar(context.Background(), e.route)
			if err != nil {
				log.Printf("activator: SIDECAR restart for %s failed: %v (next attempt in %s)", e.host, err, e.sidecarBackoff)
				e.sidecar = nil
				continue
			}
			log.Printf("activator: sidecar for %s was down — restarted (backoff now %s)", e.host, e.sidecarBackoff)
			e.sidecar = sc
			e.sidecarStarted = now
		}
	}
	// Due-time wakes: a request-less wake through the same singleflight state
	// machine, so a client request arriving mid-wake just coalesces. Only hosts
	// with NO live entry are woken — an awake host is already serving its
	// schedule; a waking one is on its way; a stopping one will re-record its
	// next_due when the stop completes and be picked up on a later tick. The
	// registry's stale past value while the host is up is harmless for the same
	// reason. A failed due wake (e.g. a lease still held by a crashed
	// incarnation) removes the entry, and the index — only rewritten at the
	// next successful sleep — naturally retries it every tick until it works.
	for _, rt := range dueRoutes {
		if _, live := a.entries[rt.Host]; live {
			continue
		}
		// Mesh: only the HRW owner due-wakes a host — otherwise every box
		// sweeping a due entry would race a wake per tick (the lease would let
		// exactly one win, but loudly, forever). If the owner box is dead its
		// heartbeat goes stale and the next tick's membership makes someone
		// else the owner — eventual, bounded by staleAfter, and a scheduled
		// delivery is late by at most that. (Today each box records next_due
		// in its own registry, so the recorder and the owner are usually the
		// same box; a host whose ownership moved while asleep waits for the
		// registry-sync work — or a member request — to fire. Known gap,
		// deliberately deferred with the shared-registry question.)
		if a.mesh != nil && !a.mesh.owns(rt.Host) {
			continue
		}
		log.Printf("activator: %s has a scheduled delivery due — waking it", rt.Host)
		e := &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
		a.entries[rt.Host] = e
		go a.wake(e)
	}
	a.mu.Unlock()
}

// retireExcept stops (final checkpoint + lease release) every live entry whose
// host is not in keep — the reload-time reconcile behind `lchat hibernate
// <host> --off` (and, as a bonus, route removal while awake). The flip clears
// the route's flag and SIGHUPs the router; the map swap stops routing requests
// through the activator, and this puts the child down and gives the S3 lease
// back so the host's systemd unit can bind the socket. It reuses the ordinary
// stop path — the narrowest hook that exists — so the release still records
// next_due and checkpoints to S3 like any sleep (the flip-off's caller waits
// for the socket to go quiet, not for the checkpoint; see hibernateOff).
// In-flight requests on a retired child are cut off as a deploy restart would
// cut them — the flip is a deliberate operator action.
func (a *activatorSvc) retireExcept(keep map[string]bool) {
	a.mu.Lock()
	var waking []*actEntry
	for _, e := range a.entries {
		if keep[e.host] {
			continue
		}
		switch e.state {
		case actAwake:
			log.Printf("activator: %s is no longer hibernate-managed — releasing", e.host)
			a.beginStopLocked(e)
		case actWaking:
			// A wake can't be stopped mid-flight; finish the retire once it
			// resolves (below, off-lock).
			waking = append(waking, e)
		case actStopping:
			// Already on its way out.
		}
	}
	a.mu.Unlock()
	for _, e := range waking {
		go func(e *actEntry) {
			<-e.ready
			a.mu.Lock()
			defer a.mu.Unlock()
			// Only if THIS entry is still the live one and its wake succeeded —
			// a failed wake already removed itself, and a fresh entry for the
			// same host means the route was re-flagged in the meantime.
			if a.entries[e.host] == e && e.state == actAwake {
				log.Printf("activator: %s is no longer hibernate-managed — releasing (wake finished first)", e.host)
				a.beginStopLocked(e)
			}
		}(e)
	}
}

// shutdown checkpoints and stops every child — the router going down takes its
// instances with it (they're its children; leaving them orphaned would strand
// the leases until TTL expiry and leave the sockets bound against the next
// incarnation's spawns). Blocks until all stops complete or ctx expires.
func (a *activatorSvc) shutdown(ctx context.Context) {
	// Drain step 1, order-pinned: leave the mesh membership FIRST — the
	// heartbeat is deleted before any checkpoint or lease release below, so
	// peers stop proxying here immediately and their wakes succeed the moment
	// each lease releases. Deploys, box failure drills, and scale-down all
	// ride this same path.
	if a.mesh != nil {
		a.mesh.stop(ctx)
	}
	if a.sweepStop != nil {
		close(a.sweepStop)
		<-a.sweepDone
	}
	a.mu.Lock()
	var waits []chan struct{}
	for _, e := range a.entries {
		switch e.state {
		case actAwake:
			a.beginStopLocked(e)
			waits = append(waits, e.stopped)
		case actStopping:
			waits = append(waits, e.stopped)
		case actWaking:
			// A mid-flight wake resolves on its own; its instance will be
			// orphaned only if it wins the race with process exit — acceptable
			// for the rare shutdown-during-wake, and the lease TTL still
			// bounds the damage.
		}
	}
	a.mu.Unlock()
	for _, w := range waits {
		select {
		case <-w:
		case <-ctx.Done():
			return
		}
	}
}

// serveHTTP answers one proxied request for a hibernating instance: wake (or
// join a wake), proxy, stamp activity. Lease contention and wake timeouts are
// 503s — "sleeping spaces are slow to wake", never a 5xx cascade of retries.
func (a *activatorSvc) serveHTTP(w http.ResponseWriter, r *http.Request, rt Route, proxy http.Handler) {
	// Sidecar-marked traffic (bot.SidecarHeader — the Go sidecar stamps every
	// REST call and its WebSocket dial) is activity-invisible: proxied while
	// the instance is awake, refused while it isn't, never a wake. The
	// co-spawned sidecar dials its instance's public URL through this very
	// router (the box pins its hosted names to loopback), so without this its
	// keepalives would read as member activity — the instance could never
	// idle — and after a sleep its reconnect loop would be the wake source:
	// ping-pong forever. Spoofing the marker is harmless by construction: it
	// only DOWNGRADES the sender's own requests (no activity credit; 503 while
	// asleep). It cannot suppress other clients' activity stamps and cannot
	// wake a sleeping host, so an outsider sending it gains nothing.
	if r.Header.Get(bot.SidecarHeader) != "" {
		if err := a.beginSidecarRequest(r.Context(), rt); err != nil {
			if errors.Is(err, context.Canceled) {
				return // the sidecar hung up while queued on a wake
			}
			w.Header().Set("Retry-After", "5")
			http.Error(w, "instance is asleep; sidecar requests do not wake it", http.StatusServiceUnavailable)
			return
		}
		proxy.ServeHTTP(w, r)
		return
	}

	// Mesh placement (mesh.go): a member request for a host with no live local
	// entry and an HRW owner on another live box is proxied there instead of
	// woken here. A forwarded request is never re-proxied (one hop max — see
	// meshForwardedHeader). A host with ANY live local entry stays local: an
	// awake instance finishes its natural life where it is, and placement
	// converges at the next sleep boundary — its state is already in S3, so
	// "migrating" is just waking on the new owner. The check-then-proxy is
	// deliberately not atomic with the entry map: a race sends one request the
	// long way, or wakes a not-the-hint box, and the lease arbitrates exactly
	// as it would for real membership disagreement. (Sidecar-marked traffic
	// above never routes through the mesh — a co-spawned sidecar's instance is
	// by construction local, and a stray marked request for a host living
	// elsewhere is exactly the reconnect-racing-a-stop case the 503 exists for.)
	if a.mesh != nil && r.Header.Get(meshForwardedHeader) == "" && !a.hasLiveEntry(rt.Host) {
		if peer, remote := a.mesh.remoteOwner(rt.Host); remote {
			a.mesh.proxyTo(w, r, peer)
			return
		}
	}

	release, err := a.beginRequest(r.Context(), rt)
	if err != nil {
		var held *errLeaseHeld
		switch {
		case errors.As(err, &held):
			retry := int(held.retryAfter/time.Second) + 1
			w.Header().Set("Retry-After", strconv.Itoa(retry))
			http.Error(w, "instance is waking elsewhere; retry shortly", http.StatusServiceUnavailable)
		case errors.Is(err, errWakeTimeout):
			w.Header().Set("Retry-After", "5")
			http.Error(w, "instance is waking; retry shortly", http.StatusServiceUnavailable)
		case errors.Is(err, context.Canceled):
			// The client hung up while queued; nothing to write.
		default:
			log.Printf("activator: wake %s: %v", rt.Host, err)
			http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
		}
		return
	}
	defer release()
	proxy.ServeHTTP(w, r)
}

// --- the real backend ---------------------------------------------------------

// s3Backend is the production actBackend: leases + restore via the
// internal/activator litestream primitives, processes via exec of the same
// versioned binary the systemd unit would run.
type s3Backend struct {
	cfg hibernateConfig
	// materialize, when non-nil, stages the running binary and points a
	// host's live symlink at it via the control daemon (cloud.go's
	// controldMaterializeFunc) instead of doing it directly — needed because
	// this process runs sandboxed and can't write to /opt/lchat itself. nil
	// falls back to the direct calls (start's else branch), today's
	// long-standing behavior for any box that hasn't opted in.
	materialize func(ctx context.Context, host string) error
}

func (b *s3Backend) acquire(ctx context.Context, host string) (actLease, error) {
	held, err := activator.AcquireLease(ctx, b.cfg.Bucket, b.cfg.hibernatePrefix(host), b.cfg.Region)
	if err != nil {
		var lx *litestream.LeaseExistsError
		if errors.As(err, &lx) {
			retry := time.Until(lx.ExpiresAt)
			if retry < time.Second {
				retry = time.Second
			}
			return nil, &errLeaseHeld{holder: lx.Owner, retryAfter: retry}
		}
		return nil, err
	}
	return &s3Lease{held: held}, nil
}

type s3Lease struct{ held *activator.HeldLease }

func (l *s3Lease) replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (actRep, error) {
	return l.held.Replicate(ctx, dbPath, syncInterval)
}
func (l *s3Lease) release(ctx context.Context) error { return l.held.Release(ctx) }

func (b *s3Backend) restore(ctx context.Context, host, dbPath string) error {
	if _, err := os.Stat(dbPath); err == nil {
		return nil // warm cache — see doWake
	} else if !os.IsNotExist(err) {
		return err
	}
	// The DB is gone, so any leftover litestream sidecar state (the
	// <db>-litestream meta dir, WAL/SHM) describes a database that no longer
	// exists — litestream's own recovery guidance is to delete it. Clear it so
	// the restored copy starts a clean lineage.
	for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} {
		_ = os.Remove(p)
	}
	_ = os.RemoveAll(dbPath + litestream.MetaDirSuffix)

	restored, err := activator.Restore(ctx, dbPath, b.cfg.Bucket, b.cfg.hibernatePrefix(host), b.cfg.Region)
	if err != nil {
		return err
	}
	if restored {
		log.Printf("activator: restored %s from s3://%s/%s", host, b.cfg.Bucket, b.cfg.hibernatePrefix(host))
	} else {
		// Never replicated: a brand-new space boots fresh and creates its own
		// DB — this is the future register-only create path (provisioning as a
		// metadata write) landing on its first wake.
		log.Printf("activator: no replica for %s in s3://%s/%s — booting fresh", host, b.cfg.Bucket, b.cfg.hibernatePrefix(host))
	}
	return nil
}

// nextDue asks the (still-running) instance for its earliest pending scheduled
// delivery over its socket. The endpoint is admin-gated like every instance
// data endpoint, so present the instance's own admin token — read from the same
// env file the spawn path hands the child (the activator holds it either way).
func (b *s3Backend) nextDue(ctx context.Context, rt Route) (int64, error) {
	fileEnv, err := readEnvFile(filepath.Join(instanceDataDir, rt.Host+".env"))
	if err != nil {
		return 0, fmt.Errorf("read env file: %w", err)
	}
	token := fileEnv["LCHAT_ADMIN_TOKEN"]
	if token == "" {
		return 0, errors.New("no LCHAT_ADMIN_TOKEN in env file")
	}
	req, err := http.NewRequestWithContext(ctx, "GET", "http://lchat/api/next-due", nil)
	if err != nil {
		return 0, err
	}
	req.Header.Set("Authorization", "Bearer "+token)
	resp, err := socketClient(rt.Socket).Do(req)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return 0, fmt.Errorf("GET /api/next-due: HTTP %d", resp.StatusCode)
	}
	var body struct {
		Due int64 `json:"due"`
	}
	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<12)).Decode(&body); err != nil {
		return 0, fmt.Errorf("decode /api/next-due: %w", err)
	}
	return body.Due, nil
}

// spawnHealthTimeout bounds how long a spawned instance may take to bind its
// socket and answer. A var so tests can shrink the failure path.
var spawnHealthTimeout = 30 * time.Second

// ensureLiveLink materializes host's live symlink (/opt/lchat/live/<host> →
// a staged version) if this box hasn't already — the future create-is-a-row
// flow, or (found live, 2026-07-13) a mesh-replicated route, which only ever
// inserts a registry row (meshreplicate.go), never a symlink. Split out of
// s3Backend.start for focused testing: exercising this doesn't need a real
// spawned process, just the filesystem.
//
// Prefers routing the write through controld (materialize, non-nil) over
// doing it directly: this process runs sandboxed (ProtectSystem=strict) and
// can't write to /opt/lchat itself — a direct call here only ever worked by
// accident, when some other, unsandboxed path (add-instance) had already
// materialized every host this box would ever wake.
func ensureLiveLink(ctx context.Context, host string, materialize func(ctx context.Context, host string) error) error {
	bin := liveLink(host)
	if _, err := os.Stat(bin); err == nil {
		return nil
	}
	if materialize != nil {
		if err := materialize(ctx, host); err != nil {
			return fmt.Errorf("materialize via controld: %w", err)
		}
		return nil
	}
	ver, err := ensureSelfStaged()
	if err != nil {
		return fmt.Errorf("stage binary: %w", err)
	}
	if err := pointLive(host, ver); err != nil {
		return fmt.Errorf("point live symlink: %w", err)
	}
	return nil
}

func (b *s3Backend) start(ctx context.Context, rt Route) (actProc, string, error) {
	host := rt.Host
	dbPath := rt.DBPath
	if dbPath == "" {
		dbPath = filepath.Join(instanceDataDir, host+".db")
	}

	// Exec exactly what the systemd unit would: the per-instance live symlink
	// (/opt/lchat/live/<host> → a staged version), same --socket/--db flags,
	// same EnvironmentFile. ensureLiveLink materializes it first if this box
	// never has (a mesh-replicated route, or the future create-is-a-row flow).
	bin := liveLink(host)
	if err := ensureLiveLink(ctx, host, b.materialize); err != nil {
		return nil, "", err
	}

	// The env file is optional, like the unit's "-" prefix.
	fileEnv, err := readEnvFile(filepath.Join(instanceDataDir, host+".env"))
	if err != nil && !os.IsNotExist(err) {
		return nil, "", fmt.Errorf("read env file: %w", err)
	}

	cmd := exec.Command(bin, "--socket", rt.Socket, "--db", dbPath)
	cmd.Env = childEnv(fileEnv)
	// The child logs through the router's own stderr/stdout (journald under
	// systemd); there's no per-instance unit to own a log stream anymore.
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	// Record the build we're about to exec — read the symlink immediately before
	// Start() so it names the version the kernel will resolve. Reading it here,
	// rather than back in wake() after the health wait, closes a real window: a
	// concurrent ship's pointLive during the up-to-30s wait would otherwise make
	// us record the NEW target while this child runs the OLD binary, and the
	// sweep would then think it's already converged and never cycle it. The
	// residual sub-exec race (a repoint between this read and Start) is in the
	// safe direction — we'd record the older version, so the sweep cycles once
	// too often rather than never. Empty on a read error: the sweep won't cycle.
	spawnVer, _ := liveVersion(host)
	if err := cmd.Start(); err != nil {
		return nil, "", err
	}

	p := &childProc{cmd: cmd, done: make(chan struct{})}
	go func() {
		p.err = cmd.Wait()
		close(p.done)
	}()

	if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil {
		stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
		defer cancel()
		_ = p.stop(stopCtx)
		return nil, "", err
	}
	return p, spawnVer, nil
}

// childEnv approximates a systemd unit's environment from inside the router:
// inherit the process env but strip every LCHAT_* var first, so the router's
// own config (hibernation, registry, wildcard tokens) can never leak into a
// child; then layer the child's EnvironmentFile on top. A nil fileEnv (no env
// file, like the unit's "-" prefix) yields just the cleaned inherit.
func childEnv(fileEnv map[string]string) []string {
	env := make([]string, 0, 32)
	for _, kv := range os.Environ() {
		if strings.HasPrefix(kv, "LCHAT_") {
			continue
		}
		env = append(env, kv)
	}
	for k, v := range fileEnv {
		env = append(env, k+"="+v)
	}
	return env
}

// hasSidecar: a host has a bot sidecar iff its sidecar EnvironmentFile exists
// — the same /data/<host>-bot.env convention add-bot writes and
// lchat-bot@.service reads. Derived from the filesystem on purpose, the way
// the units do it: no registry column to drift from what's actually staged.
func (b *s3Backend) hasSidecar(host string) bool {
	_, err := os.Stat(botEnvPath(host))
	return err == nil
}

// startSidecar mirrors lchat-bot@.service: exec the host's live symlink with
// `bot run` under the sidecar's EnvironmentFile (LCHAT_BASE, session paths,
// invites, optional ANTHROPIC_API_KEY), LCHAT_*-clean like the instance
// spawn. No health wait — the sidecar is a pure outbound client with no
// socket to probe; its own connect/backoff loop absorbs a slow instance. The
// live symlink exists by now: the instance spawn staged it. The ctx is unused
// (an exec doesn't block); it's here for the backend seam's symmetry.
func (b *s3Backend) startSidecar(_ context.Context, rt Route) (actProc, error) {
	fileEnv, err := readEnvFile(botEnvPath(rt.Host))
	if err != nil {
		return nil, fmt.Errorf("read bot env: %w", err)
	}
	cmd := exec.Command(liveLink(rt.Host), "bot", "run")
	cmd.Env = childEnv(fileEnv)
	// Like the instance child: logs through the router's own streams
	// (journald under systemd) — no per-host unit owns a log stream anymore.
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		return nil, err
	}
	p := &childProc{cmd: cmd, done: make(chan struct{})}
	go func() {
		p.err = cmd.Wait()
		close(p.done)
	}()
	return p, nil
}

// waitHealthyCtx is waitHealthy bounded by ctx as well as its own timeout, so
// an aborted wake doesn't keep probing a socket nobody wants anymore.
func waitHealthyCtx(ctx context.Context, socket string, timeout time.Duration) error {
	done := make(chan error, 1)
	go func() { done <- waitHealthy(socket, timeout) }()
	select {
	case err := <-done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// childProc is a spawned instance: SIGTERM starts the instance's own graceful
// drain (main.go's shutdown path); SIGKILL only if it overstays.
type childProc struct {
	cmd  *exec.Cmd
	done chan struct{}
	err  error
}

func (p *childProc) exited() bool {
	select {
	case <-p.done:
		return true
	default:
		return false
	}
}

func (p *childProc) stop(ctx context.Context) error {
	if p.exited() {
		return nil
	}
	if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
		if errors.Is(err, os.ErrProcessDone) { // lost the race with its own exit
			<-p.done
			return nil
		}
		return err
	}
	select {
	case <-p.done:
		return nil
	case <-ctx.Done():
		_ = p.cmd.Process.Kill()
		<-p.done
		return fmt.Errorf("instance did not exit on SIGTERM; killed")
	}
}