// The box half of `carlos restart` (issue #113): when an app's
// restart-generation object (release.RestartGenKey) MOVES, cycle that app's
// live processes on this box. Pull-based like everything else here — no push
// channel, no box access, no version or config change involved.
//
// IT RIDES THE SERIAL, and that is the load-bearing design decision. The
// console door bumps control/serial after touching the object, exactly as
// every other console write does, and the agent's existing 2s serial poll
// (fastTick) already turns a bump into the full convergence pass set. The
// restart check is one more leg of THAT — so a box with nothing happening
// pays its one serial GET per tick and NOT ONE READ MORE. An earlier revision
// polled every app's restart generation on its own 3s timer; measured, that
// was ~575k needless GETs per box per day to notice an event that arrives a
// handful of times a month. A verb that costs the fleet a permanent poll storm
// to exist is not a verb worth having.
//
// Direct-bucket touches that never bump the serial (a script writing the
// object straight into the bucket) converge on the hourly root pass instead —
// the same floor every other direct-bucket write already relies on.
//
// The work is split across two processes, for the reason config fast
// convergence is split the same way:
//
//   - the AGENT (carlos-edge.service, User=carlos, ProtectSystem=strict) owns
//     the EXEC-backed instances outright, so its half is the etag comparison,
//     a bounce of its own children, and one marker file for the rest.
//     systemctl cannot reach a child of another process, so a root-only
//     dispatch would leave every warm exec instance — the commonest shape on
//     the box, and the one a wedged instance usually is — serving on,
//     unrestarted.
//   - the ROOT pass (carlos adopt, out of carlos-restart.service via
//     carlos-restart.path, or the hourly timer) is the only one that can
//     restart a systemd unit — the carlos-app@ tenants and the bespoke units
//     addr-backed routes name (issue #110's consoles). Its half is the
//     comparison against a persisted, per-app high-water mark and the
//     dispatch.
//
// SEED-DON'T-REPLAY governs both halves, PER APP. An app this box has not
// recorded yet is recorded and NOT restarted, whether its object exists or
// not; only movement away from a value this box has already written counts.
// The absent case gets its own recorded sentinel precisely so that an app's
// first ever request (absent -> present) is still movement rather than a first
// sight — that distinction is the difference between the verb working and the
// verb cycling a box's whole route table the day a marks file goes missing.
package host

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

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

// restartEligible widens release.Adoptable BY NAME, for this feature alone:
// a route release.Adoptable would cycle, plus a sidecar. release.Adoptable
// itself is deliberately NOT changed — it is the most reused predicate in
// the repo (adopt membership, capability convergence, socket and backing
// advisories, status pages), and every one of those is correct to exclude
// a sidecar. Widening it centrally would silently enrol sidecars in
// capability drop-ins and adopt bookkeeping that make no sense for them.
// `carlos restart -app X` genuinely wants sidecars too (design §9.3), so it
// opts in here, explicitly, rather than through Adoptable itself.
func restartEligible(rt registry.Route) bool {
	return release.Adoptable(rt) || rt.IsSidecar()
}

// restartCandidates filters routes to the ones the restart feature may
// cycle: restartEligible, applied. Exported to this package's tests as the
// single predicate restartApps, bounceExecRoutes' membership check and
// cycleWorkers all trace back to, so the three cannot silently diverge on
// who counts.
func restartCandidates(routes []registry.Route) []registry.Route {
	var out []registry.Route
	for _, rt := range routes {
		if restartEligible(rt) {
			out = append(out, rt)
		}
	}
	return out
}

// restartApps lists the distinct (account, app) pairs with at least one
// restartable route on this box — the apps whose restart generation this box
// reads when a serial bump (or an hourly pass) gives it a reason to.
//
// The membership predicate is restartCandidates (release.Adoptable plus a
// sidecar — see its doc comment): a route neither would cycle is not one
// whose process this feature has any business reading a restart signal
// for, and a second copy of that judgement here is exactly the drift that
// would let them disagree. It admits addr-backed routes that name their
// own unit (issue #110's consoles) — which is half of why #113 exists —
// because Adoptable filters on lifecycle and app, never on backing.
func restartApps(routes []registry.Route) [][2]string {
	seen := map[[2]string]bool{}
	var out [][2]string
	for _, rt := range restartCandidates(routes) {
		k := [2]string{release.RouteAccount(rt), rt.App}
		if seen[k] {
			continue
		}
		seen[k] = true
		out = append(out, k)
	}
	return out
}

// restartGenPass is the agent's comparison, split out for tests: read the
// restart-generation etag of every app this box serves and report WHICH of
// them moved. Same semantics — and the same reasons — as genPollPass:
//
//   - First sight of an app records silently and asks for nothing. A box
//     coming up must not restart everything it serves because it has never
//     seen these etags before; that is the seed-don't-replay rule, and it is
//     the whole safety property of this feature. restartSeed runs one such
//     silent pass at arming, so the FIRST request after a boot is acted on
//     rather than swallowed as the baseline.
//   - An unchanged etag does nothing.
//   - Absent that stays absent does nothing, FOREVER: an app nobody has ever
//     restarted this way has no object and never will until someone asks.
//     That is the ordinary state of almost every app, not a fault.
//   - Any movement asks: the etag changed, or absent -> present (an app's
//     first ever restart request). present -> absent should not happen
//     (restart generations are only created and overwritten, never deleted),
//     but it counts as movement too — one extra root pass is cheap, and
//     pretending we did not see it would be a lie about the only thing this
//     function reports.
//   - The moved apps come back as a LIST, but the root request is one marker
//     touch however long that list is: the root pass re-reads every app's etag
//     itself and dispatches per app, so asking once per moved app would only
//     ask the same pass to run several times. The list is for the caller's own
//     half — which of ITS exec children to bounce.
//   - A read error that is not "not found" leaves that app's cache entry
//     exactly as it was and keeps evaluating the rest — the errors come back
//     for the caller to aggregate into ONE log line per pass. Clearing the
//     entry would turn a recovered blip into a bogus first sight or a bogus
//     restart.
//   - Apps that vanish from the route set lose their cache entries: a re-added
//     app is a first sight again, which is right, because a route that has
//     just been (re-)minted is starting its process fresh anyway.
//
// The cache is mutated in place and lives for the life of the agent process
// (fastConvergeLoop owns it, restartSeed fills it), which is the whole point:
// there is nowhere else to remember what "changed" means between two serial
// moves that may be days apart.
func restartGenPass(ctx context.Context, routes []registry.Route, store release.Store, cache map[[2]string]string) (moved [][2]string, readErrs []error) {
	apps := restartApps(routes)
	live := make(map[[2]string]bool, len(apps))
	for _, k := range apps {
		live[k] = true
		_, etag, err := store.Get(ctx, release.RestartGenKey(k[0], k[1]))
		switch {
		case errors.Is(err, os.ErrNotExist):
			etag = genAbsent
		case err != nil:
			readErrs = append(readErrs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
			continue // cache entry untouched: this app is unknown this tick, not changed
		}
		prev, seen := cache[k]
		cache[k] = etag
		if seen && prev != etag {
			moved = append(moved, k)
		}
	}
	for k := range cache {
		if !live[k] {
			delete(cache, k)
		}
	}
	return moved, readErrs
}

// bounceExecRoutes cycles the agent's OWN children for the apps whose restart
// generation just moved, and returns how many it actually stopped. This is the
// half no root pass can perform: an exec instance is a child of THIS process,
// and systemctl has no name for it.
//
// The route filter is the complement of what rootRestartRoute's systemctl arms
// can reach — no named unit, no unit backing — so the two halves never cycle
// the same process twice.
//
// Coverage is NOT total, and the gap is worth naming: a route that is
// activator-backed (Backing "", an exec child) but ALSO carries a Unit stamp
// is skipped here (it names a unit) while the root half try-restarts that
// named unit, which is not the process the activator is running. Such a route
// is a misconfiguration — `carlos route -unit` is for addr-backed tenants
// whose process an operator owns — but nothing in this file repairs it, and
// its exec child is cycled by NEITHER half until somebody clears the stamp.
//
// HIBERNATE-flagged routes are included, and that is not a contradiction of
// "never wake a sleeper": Bounce provably cannot start anything (it is a no-op
// unless the instance is AWAKE), so a sleeping instance is untouched and picks
// its next start up naturally, while a WARM hibernator — which is what a
// wedged instance usually is — gets exactly the cycle that was asked for. This
// is release.RestartHibernating's argument, in the one process that can act on
// it.
//
// A nil bounce (no activator armed on this box) is a working no-op: there are
// no children to cycle, and the unit half still runs.
func bounceExecRoutes(routes []registry.Route, moved [][2]string, bounce func(host string) bool) int {
	if bounce == nil || len(moved) == 0 {
		return 0
	}
	want := make(map[[2]string]bool, len(moved))
	for _, k := range moved {
		want[k] = true
	}
	n := 0
	for _, rt := range routes {
		if !release.Adoptable(rt) || !execBounceable(rt) {
			continue
		}
		if !want[[2]string{release.RouteAccount(rt), rt.App}] {
			continue
		}
		if bounce(rt.Host) {
			n++
			log.Printf("agent: restart requested: bounced exec instance %s", rt.Host)
		}
	}
	return n
}

// dispatchBoundedWakes wakes every route in toWake, bounded by
// workerBootConcurrency wakes running at once, entirely inside ONE
// goroutine this function launches — so dispatchBoundedWakes ITSELF
// returns to its caller immediately, no matter how many routes are queued
// behind it.
//
// This is fix round 2's own correction of round 1's own fix (sidecar-
// members Task 13): the round-1 shape acquired the semaphore in the
// CALLER's loop, one route at a time — which dispatches each wake off the
// blocking path individually, but still leaves the loop ITSELF, and
// therefore the caller, blocked on the sem acquisition once
// workerBootConcurrency wakes are already in flight and a
// (workerBootConcurrency+1)th route needs dispatching. For an app shaped
// like Eleven's — a sidecar PER ENROLLED MEMBER, so routinely more than 8
// sidecars for one app on one box — that reintroduces exactly the bug this
// idiom exists to fix: a restart or wake burst on such an app would stall
// the box's single per-tick convergence goroutine for up to wakeWaitBound
// (30s) per sidecar past the 8th. Moving the WHOLE bounded fan-out inside
// one dispatched goroutine fixes it: the bound still governs how many
// wakes run concurrently, and now ONLY that — never when the caller gets
// to continue.
//
// Shared by cycleWorkers (this file) and sidecarwake.go's
// dispatchSidecarWakes, the two callers of `carlos restart` and the
// sidecar wake feature respectively that both need "wake N sidecars
// without blocking the caller, bounded" — one idiom, not two copies that
// would each need this fix applied separately.
//
// wake is called once per route in toWake; report, if non-nil, is called
// once per wake with its result (nil error on success) so each caller can
// log in its own voice. A nil wake or an empty toWake is a working no-op.
func dispatchBoundedWakes(toWake []registry.Route, wake func(rt registry.Route) error, report func(rt registry.Route, err error)) {
	if wake == nil || len(toWake) == 0 {
		return
	}
	go func() {
		sem := make(chan struct{}, workerBootConcurrency)
		for _, rt := range toWake {
			rt := rt
			sem <- struct{}{}
			go func() {
				defer func() { <-sem }()
				err := wake(rt)
				if report != nil {
					report(rt, err)
				}
			}()
		}
	}()
}

// cycleWorkers is bounceExecRoutes' sidecar counterpart, and deliberately a
// separate function rather than a widened bounceExecRoutes: a sidecar is
// activator-managed exactly like a socket-exec instance, but Bounce alone
// cannot cycle one the way it cycles an instance. Nothing dials a worker
// the way a request dials an instance — an instance's next request supplies
// the wake half of "stop now, respawn with fresh config"; a worker stopped
// here has no other wake source at all (an always-on one is stranded until
// the agent restarts, since the boot pass that starts it runs once, at
// arming, never again; a hibernating one stays silent until its schedule
// next comes due, which "carlos restart" did not ask for). Stop and wake
// are therefore ONE operation for a sidecar — see
// activator.Service.Bounce's own doc comment, which states the same
// contract from the activator's side.
//
// The BOUNCE half is synchronous, in-process and effectively instant — it
// is counted in the return value, exactly like bounceExecRoutes' own bounce
// call, and it stays in THIS function's own loop: only the eligible,
// actually-stopped routes are collected into toWake. The WAKE half is NOT
// instant: WakeWorker can block up to wakeWaitBound (30s, activator.go) per
// sidecar waiting out the settle window, and this function's only caller
// (restartRequestLeg) runs synchronously on the box's single per-tick
// convergence goroutine (fastTick) — waiting on it, even indirectly via a
// per-route semaphore acquisition in this function's own loop, would delay
// every other app's serial-driven convergence behind it (see
// dispatchBoundedWakes' own doc comment for the exact shape of that bug,
// caught and fixed in review round 2). The whole bounded fan-out is
// therefore handed to dispatchBoundedWakes, which returns immediately
// regardless of how many routes it was given. Safe because WakeWorker is
// idempotent and `carlos restart`'s contract is "cycle the processes", not
// "return once they are up" — a sidecar that comes up a moment after this
// tick has still been restarted. A wake that errors is logged and does not
// stop, delay, or fail the others.
//
// wake is called ONLY for a host bounce actually stopped (true): a sidecar
// that was already asleep is left asleep, the same "never wake a sleeper"
// rule bounceExecRoutes' own HIBERNATE handling already gives an instance.
// A nil bounce or wake (no activator armed) is a working no-op.
func cycleWorkers(routes []registry.Route, moved [][2]string, bounce func(host string) bool, wake func(rt registry.Route) error) int {
	if bounce == nil || wake == nil || len(moved) == 0 {
		return 0
	}
	want := make(map[[2]string]bool, len(moved))
	for _, k := range moved {
		want[k] = true
	}
	n := 0
	var toWake []registry.Route
	for _, rt := range routes {
		if !rt.IsSidecar() || !want[[2]string{release.RouteAccount(rt), rt.App}] {
			continue
		}
		if !bounce(rt.Host) {
			continue
		}
		n++
		log.Printf("agent: restart requested: bounced sidecar %s", rt.Host)
		toWake = append(toWake, rt)
	}
	dispatchBoundedWakes(toWake, wake, func(rt registry.Route, err error) {
		if err != nil {
			log.Printf("agent: restart requested: re-wake sidecar %s: %v", rt.Host, err)
		}
	})
	return n
}

// restartRequestMarker is the file the agent's restart leg touches to ask a
// root pass to run now: carlos-restart.path watches it (PathChanged) and starts
// carlos-restart.service (`carlos adopt -restart-only`), which owns the half
// this unprivileged process cannot do — restarting units — and only that half.
//
// A sibling of configSyncMarker rather than a reuse of it, because the two
// relays land on different services: config's target is `carlos adopt
// -config-only`, which renders /etc/carlos/apps and restarts only the routes
// whose env file actually changed. A restart request changes no file, so that
// pass would find nothing to do and cycle nothing at all.
//
// It lives in the data dir, deliberately outside the *.config.env namespace
// reapConfigEnv sweeps — the reaper matches that suffix alone and must go on
// ignoring this.
const restartRequestMarker = "restart-requested"

// requestRootRestart touches the marker. The content is a timestamp only so an
// operator reading /data can see when the last request went out; systemd
// watches the path, not the bytes. Errors are the caller's to log and swallow:
// a request that cannot be written costs latency, not correctness, because
// carlos-adopt.timer's hourly pass runs the same comparison underneath it.
func requestRootRestart(dd string) error {
	return safefile.WriteFile(filepath.Join(dd, restartRequestMarker),
		[]byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644)
}

// restartMarksFile is the ROOT pass's memory: the last restart generation it
// has already acted on, per app. It lives in the data dir beside the bake
// clock, in the same JSON + temp-file-rename shape and for the same
// cross-uid reason (see restartMarks.save).
const restartMarksFile = "restart-marks.json"

func restartMarksPath(dd string) string { return filepath.Join(dd, restartMarksFile) }

// restartMarks is that memory. Apps is keyed "<account>|<app>" — "|" is
// outside every character class a sqid or an app name allows, the same
// separator argument internal/release's bake clock makes for its own keys.
//
// A value is either the etag this box last acted on OR restartAbsent, the
// recorded "this app has no restart generation and I have looked". On disk:
//
//	{
//	  "schema": 1,
//	  "apps": {
//	    "bab|keymail": "\"9f86d0…\"",
//	    "bab|titogo":  "\u0000absent"
//	  }
//	}
//
// The absent sentinel is what makes per-app seed-don't-replay possible. Without
// it "no entry" has to mean both "never looked" and "looked, nothing there",
// and one of those two has to be wrong: read as movement, a box gaining a new
// route for a previously-restarted app cycles it for a request served weeks
// ago (and a box whose registry lands after its marks file cycles EVERY route
// at once); read as a first sight, an app's first ever request is silently
// dropped. Recording the absence tells them apart — a later object sighting is
// movement from a value this box wrote, which is exactly what it is.
//
// Entries are never pruned when a route leaves the box. A vanished app's mark
// costs one map entry; DROPPING it would make the app's return look like a
// brand-new object and cycle it once for a request that was already served.
type restartMarks struct {
	Schema int               `json:"schema"` // 1
	Apps   map[string]string `json:"apps"`
	// Scopes is the same memory for NARROWED requests (discussion #16),
	// keyed identically: the id of the last request in
	// release.RestartScopeKey's log this box has acted on, or
	// release.RestartScopeNone for "looked, nothing there".
	//
	// Additive in both directions, because this file is read and written by
	// two binaries that roll at different times. A binary that predates the
	// field ignores it on read (encoding/json drops unknown keys) and DROPS
	// it on write — which costs a downgraded box exactly one re-baseline,
	// after which it cycles nothing it was not asked to.
	//
	// The other direction is free of REPLAYS but not of drops, and the
	// difference matters: a marks file with no `scopes` key is a box that has
	// never looked at the log, which is a first sight, which is a seed — safe,
	// and exactly wrong for a narrowed request that lands before that seed is
	// written. Since EVERY box that ran the app-wide feature has such a file,
	// that window is closed from the agent's side rather than left to the
	// hourly floor; see seedRestartMarks' needRootPass.
	Scopes map[string]string `json:"scopes,omitempty"`
}

// restartAbsent is the marks value for an app this box has SEEN and found to
// have no restart-generation object at all. It is deliberately the SAME value
// the in-memory sweep uses (genAbsent), so the agent's memory and the root
// pass's memory cannot drift apart on what "absent" is, and it is safe for the
// same reason: no store produces an etag containing a NUL byte (S3's are
// hex/quoted, DirStore's is hex). JSON renders it "\u0000absent".
const restartAbsent = genAbsent

func restartMarkKey(acct, app string) string { return acct + "|" + app }

// lockRestartMarks takes an exclusive flock for the whole of one restartPass.
// Three root passes can run this — carlos-restart.service (the relay),
// carlos-stage.service and carlos-adopt.service — and systemd will happily
// start them at the same second. Unserialized, two passes read the same marks,
// both dispatch (one request, two try-restarts, on a verb whose entire job is
// stopping processes) and then both write, so whichever finishes last silently
// reverts the other's advance and the NEXT pass restarts everything a third
// time.
//
// The lock lives in a sibling <marks>.lock rather than on the marks file
// itself, because the marks file is replaced by rename: a flock is held on an
// inode, and the winner's rename would leave the loser holding a lock on a file
// that is no longer the marks file.
//
// Opening it is ALSO the writability test the dispatch is gated on (see
// restartPass): it is created in the same directory, with the same permissions
// the marks save needs, so a full or read-only /data fails here — before
// anything is cycled — instead of after, which is the difference between one
// logged refusal and an hourly restart of every app on the box forever.
//
// The lock itself is taken through lockFileEx, which is per-OS: linux and
// darwin flock, Windows refuses (see flock_windows.go — refusing dispatches
// nothing, which is the safe answer on a platform that never runs a pass).
func lockRestartMarks(path string) (unlock func(), err error) {
	return lockExclusive(path + ".lock")
}

// newRestartMarks is an empty memory: both maps non-nil, so every caller can
// write to them without a nil check and a marks file that predates one of them
// cannot leave a nil map behind.
func newRestartMarks() *restartMarks {
	return &restartMarks{Schema: 1, Apps: map[string]string{}, Scopes: map[string]string{}}
}

// loadRestartMarks reads the marks file. `known` is the seed-don't-replay
// switch and the only subtle thing here: it is FALSE when the file is missing,
// unreadable, or does not parse — three different accidents with one correct
// response, which is to record what the bucket currently says and restart
// NOTHING this pass.
//
// That direction is the opposite of the bake clock's ("an unreadable clock
// fails toward baking"), and deliberately so: baking is the cautious act there,
// while restarting is the DISRUPTIVE act here. The cost of failing this way is
// that a box whose marks file was just wiped can drop ONE restart request —
// logged, and re-runnable by the person who typed it. The cost of failing the
// other way is a box that cycles every app it serves the moment its state file
// goes missing.
func loadRestartMarks(path string) (marks *restartMarks, known bool) {
	m := newRestartMarks()
	raw, err := safefile.ReadFile(path)
	if err != nil {
		if !errors.Is(err, os.ErrNotExist) {
			log.Printf("adopt: restart marks %s unreadable (%v) — re-baselining, no restarts this pass", path, err)
		}
		return m, false
	}
	if err := json.Unmarshal(raw, m); err != nil {
		log.Printf("adopt: restart marks %s do not parse (%v) — re-baselining, no restarts this pass", path, err)
		return newRestartMarks(), false
	}
	if m.Apps == nil {
		m.Apps = map[string]string{}
	}
	// A file written before narrowed restarts existed carries no `scopes` key
	// at all; that is a first sight of every app's log, not an empty memory of
	// one, and the seed rules below turn it into exactly that. Which is why
	// the agent asks for a pass that performs that seed as it arms, instead of
	// leaving a request typed in the meantime to meet it — scopeBaselineIncomplete.
	if m.Scopes == nil {
		m.Scopes = map[string]string{}
	}
	m.Schema = 1
	return m, true
}

// save persists the marks via temp-file-in-the-same-dir + rename, the
// co-written-file pattern release.FirstSeen.Save documents at length: /data is
// written by BOTH uids on this box (carlos for the agent's seed below, root
// for this pass), and an in-place O_TRUNC write needs write permission on the
// existing file, so whichever uid created it would lock the other out forever.
// rename(2) only needs write permission on the directory, which both have. The
// temp name is minted fresh per attempt so a crash between write and rename
// can never leave cross-uid residue for the next save to trip over, and the
// 0600 CreateTemp mints is widened to 0644 BEFORE the rename so the other uid
// can still read it.
func (m *restartMarks) save(path string) error {
	raw, err := json.MarshalIndent(m, "", "  ")
	if err != nil {
		return err
	}
	raw = append(raw, '\n')
	return safefile.WriteFile(path, raw, 0o644)
}

// seedRestartMarks writes the baseline the ROOT pass compares against, ONCE,
// from the agent's arming sweep — create-only (O_EXCL), so it can never
// overwrite a mark the root pass has already advanced.
//
// It exists to close a real hole in the handover between the two halves. The
// agent seeds its own baseline as it arms; the root pass would otherwise not
// seed until it next ran, and a restart requested inside that window would
// arrive at a pass with no baseline — which, by the rule above, records and
// restarts nothing. The request would be silently dropped, having told its
// caller "requested". Creating the file here shrinks that window to the moment
// the agent starts, where the box has just restarted everything anyway.
//
// An EMPTY cache writes nothing at all. A sweep that saw no adoptable routes
// is not a baseline — most often it is a box whose registry has not landed yet
// (a fresh machine, or one restoring /data from litestream) — and a file
// claiming "this box knows about no apps" is precisely the file that makes the
// root pass treat every route it later gains as movement, and cycle the lot in
// one pass.
//
// Absent objects ARE recorded, with restartAbsent: "looked, nothing there" is
// the fact that makes an app's first ever request movement rather than a first
// sight.
//
// needRootPass is the SECOND half of the handover, and it exists because
// create-only cannot repair a baseline that is merely INCOMPLETE. Every box
// that ran the app-wide feature already has a marks file, written before
// narrowed restarts existed and so carrying no `scopes` key at all. This
// function will not touch it — that is the whole create-only rule — so the
// narrowed mark stays missing until a ROOT pass records it, and until then
// restartScopeLeg reads every app as a first sight and cycles nothing for a
// request the CLI has already answered "restart requested". Exec-shaped routes
// survive that (the agent bounces its own children from mem.scope), but
// unit-shaped ones — Tito's relay and router, the shape these flags exist for
// — are dropped silently, for as long as it takes the next root pass to run:
// worst case the hourly timer.
//
// So when the baseline on disk is missing a narrowed mark for an app this box
// serves, say so, and let restartSeed ask for ONE root pass. That pass is the
// same pass carlos-adopt.timer runs anyway, under the same lock, against the
// same marks file and the same seed rules — this changes WHEN it runs, never
// what it decides — so it can neither replay a request (it dispatches on the
// per-app marks, which nothing here writes) nor drop one (it strictly shrinks
// the window in which a narrowed request meets a pass with no narrowed
// baseline, from an hour to the moment the agent arms, which is the same floor
// seedRestartMarks already buys the app-wide half).
func seedRestartMarks(path string, mem *restartMemory) (needRootPass bool, err error) {
	if mem == nil || len(mem.gen) == 0 {
		return false, nil // nothing was seen; nothing may be claimed
	}
	if _, err := os.Stat(path); err == nil {
		// The root pass (or an earlier agent) already has a baseline — but it
		// may predate narrowing, and a half-baseline is what silently eats a
		// narrowed request.
		return scopeBaselineIncomplete(path, mem), nil
	} else if !errors.Is(err, os.ErrNotExist) {
		return false, err
	}
	m := newRestartMarks()
	for k, etag := range mem.gen {
		m.Apps[restartMarkKey(k[0], k[1])] = etag
	}
	// Both memories or neither: restartSeed only gets here when every app's
	// BOTH objects read cleanly, so a baseline that recorded one and left the
	// other for the root pass to establish would be a baseline with a known
	// hole in it — which is the one shape restartPass refuses to write.
	for k, id := range mem.scope {
		m.Scopes[restartMarkKey(k[0], k[1])] = id
	}
	raw, err := json.MarshalIndent(m, "", "  ")
	if err != nil {
		return false, err
	}
	raw = append(raw, '\n')
	f, err := safefile.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644)
	if err != nil {
		if errors.Is(err, os.ErrExist) {
			// Lost the race to the root pass; its mark is the better one — but
			// a pass running an OLD binary writes no scopes key, so the same
			// question has to be asked of what actually landed.
			return scopeBaselineIncomplete(path, mem), nil
		}
		return false, err
	}
	_, writeErr := f.Write(raw)
	if err := f.Close(); err != nil && writeErr == nil {
		writeErr = err
	}
	// The file this call just wrote carries both memories, so there is nothing
	// left for a root pass to record.
	return false, writeErr
}

// scopeBaselineIncomplete reports whether the marks file on disk leaves the
// ROOT pass with no narrowed-restart position for an app this box serves.
//
// Two shapes count, and they are the same fault:
//
//   - the file does not load (missing, unreadable, unparseable), so the next
//     pass re-baselines everything — which is right, and worth doing NOW
//     rather than at the hourly floor, since a re-baselining pass dispatches
//     nothing;
//   - the file loads but has no `scopes` entry for some app in this sweep,
//     which is exactly what a marks file written before narrowing existed
//     looks like.
//
// It reads and never writes. The agent cannot take restartPass's lock — that
// lock file is created by root and doubles as the ROOT pass's writability
// test — and a lock-free read-modify-write here would race a running pass and
// revert the advance it had just made, which is the one accident the lock
// exists to prevent. Reporting the gap and letting the pass that owns the file
// fill it in is the whole fix.
func scopeBaselineIncomplete(path string, mem *restartMemory) bool {
	marks, known := loadRestartMarks(path)
	if !known {
		return true
	}
	for k := range mem.scope {
		if _, seen := marks.Scopes[restartMarkKey(k[0], k[1])]; !seen {
			return true
		}
	}
	return false
}

// restartPass is the ROOT half: for every app on this box whose restart
// generation has moved since the mark this box recorded, cycle that app's
// routes. Called from RunAdopt — on the marker-triggered `-restart-only` run
// (seconds after the request) and on the ordinary pass (the hourly floor
// underneath the relay, which can drop an event that lands while the service
// is mid-run).
//
// The rules, each a deliberate choice:
//
//   - ZERO adoptable routes ends the pass before anything is read or written.
//     A box whose registry has not landed yet (fresh, or restoring /data) must
//     not be able to record "I know about no apps" — that file is what turns
//     every route the box later gains into movement, and cycles the whole
//     table in one pass.
//   - An app this box has NOT recorded yet is recorded and NOT cycled,
//     whether its object exists or not. Per-app seed-don't-replay: the marks
//     file is this box's memory of what it has acted on, and a request it
//     never saw is not one it owes anybody.
//   - An app with NO restart generation records restartAbsent — the sentinel
//     that makes its FIRST ever request (absent -> present) movement rather
//     than another first sight. Absent is legal forever; unrecorded absence is
//     what was not.
//   - An unknown baseline (see loadRestartMarks) records every app and
//     dispatches nothing — and if any app could not be READ, records nothing
//     at all: a baseline missing the apps a transient 503 hid is a baseline
//     that turns those apps into movement on the next pass. A dropped request
//     is recoverable (run the verb again); a restart nobody asked for is not.
//   - HIBERNATING routes are skipped outright and not marked as a failure.
//     Restarting a process that is not running is meaningless, waking it to
//     restart it is worse (it would defeat hibernation for no gain), and it
//     will exec fresh on its next wake regardless — which is the restart. A
//     WARM hibernating exec instance is not left serving, though: the agent's
//     own bounce (bounceExecRoutes) reaches it, in the one process that can.
//   - A dispatch FAILURE still advances the mark, and logs. A restart request
//     is a point-in-time event, not a desired state: a named unit that exits
//     non-zero (a wrong unit name, a service that will not come back) would
//     otherwise be retried on every pass, forever, for a request somebody made
//     once. The journal line is the report; `carlos restart` again is the
//     retry.
//   - The mark is written only AFTER the dispatch, so a pass that dies
//     mid-cycle retries rather than swallowing the request.
//   - The whole pass holds an exclusive lock (lockRestartMarks), because three
//     root units can run it at once; failing to TAKE that lock dispatches
//     nothing, since the lock file's creation is also the proof that the marks
//     can be saved afterwards. A pass that cycles processes and then cannot
//     record what it did would cycle them again every hour, forever.
//
// Holds, bake windows and the release ladder take no part in any of this:
// nothing about which version runs changes here, so there is nothing for them
// to gate.
func restartPass(ctx context.Context, routes []registry.Route, store release.Store, marksPath string, restart func(host string) error) {
	apps := restartApps(routes)
	if len(apps) == 0 {
		return // nothing to converge, and nothing this box may claim to know
	}
	unlock, err := lockRestartMarks(marksPath)
	if err != nil {
		// No lock means either a concurrent pass we cannot serialize against or
		// a /data we cannot write. Both make dispatching the wrong move: the
		// first double-restarts, the second restarts hourly and forever because
		// the mark can never advance.
		log.Printf("adopt: restart pass: cannot lock %s.lock (%v) — nothing cycled this pass", marksPath, err)
		return
	}
	defer unlock()

	marks, known := loadRestartMarks(marksPath)
	changed := false
	readFailed := false
	// What this pass has already cycled. A route restarted by an app-wide
	// request must not be restarted a second time by a narrowed one that
	// landed in the same window: one pass, one cycle per host.
	already := map[string]bool{}
	for _, k := range apps {
		_, etag, err := store.Get(ctx, release.RestartGenKey(k[0], k[1]))
		switch {
		case errors.Is(err, os.ErrNotExist):
			etag = restartAbsent
		case err != nil:
			log.Printf("adopt: restart generation %s/%s: %v (skipped this pass)", k[0], k[1], err)
			readFailed = true
			continue
		}
		key := restartMarkKey(k[0], k[1])
		prev, seen := marks.Apps[key]
		if !known || !seen {
			// SEED. Either this box's whole memory is untrustworthy, or this is
			// the first time it has looked at this app — a route it has just
			// gained, whose process is starting fresh anyway. Record, cycle
			// nothing.
			if !seen || prev != etag {
				marks.Apps[key] = etag
				changed = true
			}
			continue
		}
		if prev == etag {
			continue
		}
		dispatched := 0
		for _, rt := range routes {
			if !release.Adoptable(rt) || rt.App != k[1] || release.RouteAccount(rt) != k[0] {
				continue
			}
			if rt.Hibernate {
				log.Printf("adopt: restart requested for %s/%s: %s is hibernating — left asleep (it execs fresh on its next wake)",
					k[0], k[1], rt.Host)
				continue
			}
			if err := restart(rt.Host); err != nil {
				log.Printf("adopt: restart requested for %s/%s: %s did not cycle: %v", k[0], k[1], rt.Host, err)
				continue
			}
			already[rt.Host] = true
			dispatched++
		}
		// One line per APP, not per route, and it says DISPATCHED rather than
		// cycled: an exec-backed route reaching rootRestartRoute lands on a
		// try-restart of a carlos-app@ unit that does not exist — a deliberate
		// no-op (the agent bounced that child itself) which this pass cannot
		// tell apart from a real one. Claiming "cycled" for it would be the
		// kind of lie issue #110 was made of.
		log.Printf("adopt: restart requested for %s/%s: dispatched to %d route(s)", k[0], k[1], dispatched)
		marks.Apps[key] = etag
		changed = true
	}
	// The narrowed half (discussion #16), under the SAME lock, against the
	// same marks file and the same seed rules — a second pass with a second
	// lock would be two memories of one box's restarts, free to disagree.
	scopeChanged, scopeReadFailed := restartScopeLeg(ctx, routes, store, apps, marks, known, restart, already)
	changed = changed || scopeChanged
	readFailed = readFailed || scopeReadFailed
	if !known && readFailed {
		// A baseline that omits the apps a blip hid is worse than no baseline:
		// this pass would write "known", and the next one would read those
		// missing apps as movement and cycle them. Leave the file as it is and
		// re-baseline on a pass that can read everything.
		log.Printf("adopt: restart pass: baselining deferred — some apps could not be read this pass")
		return
	}
	// A first, unknown baseline is written even when nothing changed: that
	// write is what makes the NEXT pass a known one, and until it lands every
	// pass would re-baseline and drop whatever it was asked to do.
	if changed || !known {
		if err := marks.save(marksPath); err != nil {
			log.Printf("adopt: save restart marks %s: %v (the next pass may repeat a restart)", marksPath, err)
		}
	}
}

// restartRequestLeg is the agent's half of one serial move: compare each app's
// restart generation against the in-memory cache, bounce this process's own
// exec children for the apps that moved, and touch the marker that asks a root
// pass to cycle everything unit-shaped. Reports whether it asked.
//
// It runs ONLY when the serial moved (fastTick), which is what makes the
// feature free at rest: the console bumps the serial after touching the
// object, so every request is announced, and an idle box reads no restart
// generation at all. There is no loop of its own and there must not be one —
// see this file's package comment for what the 3s version of this cost.
//
// The bounce goes first, and not only for tidiness: it is in-process and
// instant, while the marker write can block on a busy /data, and the children
// bounced here are the ones cycled here or nowhere. That "instant" holds for
// EVERY bounce call this leg makes (bounceExecRoutes' and cycleWorkers'
// alike) — cycleWorkers' own wake calls are the one thing on this path that
// are not instant (WakeWorker can block up to 30s per sidecar), which is
// exactly why cycleWorkers dispatches them off its own return path rather
// than this leg waiting on them: this whole function still returns promptly
// after the bounces, this leg's synchronous work (the marker touch) is
// never delayed by a sidecar's wake, and neither is the next app's turn on
// this box's single per-tick convergence goroutine (fastTick).
//
// wake is svc.WakeWorker in production (wakeFn), nil-safe like bounce: it is
// cycleWorkers' other half, called only for the sidecars this leg actually
// stops (see cycleWorkers' doc comment for why a sidecar needs both calls,
// where an instance needs only bounce, and for why the wake is async).
func restartRequestLeg(ctx context.Context, routes []registry.Route, store release.Store, dd string,
	bounce func(host string) bool, wake func(rt registry.Route) error, mem *restartMemory) bool {
	moved, readErrs := restartGenPass(ctx, routes, store, mem.gen)
	// The narrowed requests are read on the same move and reported the same
	// way. Both, always: an app-wide request and a narrowed one can land in
	// one poll window, and a leg that stopped at the first would leave the
	// other for the hourly floor.
	scoped, scopeErrs := restartScopePass(ctx, routes, store, mem.scope)
	readErrs = append(readErrs, scopeErrs...)
	// `carlos instances retry` grants ride their own log (release.HostRetryKey),
	// read on the same move.
	grants, grantErrs := hostRetryPass(ctx, routes, store, mem)
	readErrs = append(readErrs, grantErrs...)
	if len(readErrs) > 0 {
		// ONE line per pass, never one per app: the count carries the scale,
		// the first error the cause — they are the same cause in practice.
		log.Printf("agent: restart check: %d apps unreadable: %v", len(readErrs), readErrs[0])
	}
	// Each grant is verified, then applied where the hold is (applyRetries):
	// the host-retry log's, and a signed retry a console built before that
	// log left in the narrowed one, with the same checks.
	applyRetries(routes, grants, mem)
	applyRetries(routes, withoutHistory(scoped, mem), mem)
	scoped = withoutRetries(scoped)
	if len(moved) == 0 && len(scoped) == 0 {
		return false
	}
	// CARLOS_SWAP=overlap: each exec restart is the scheduler's attempt,
	// paced and identified by its request (spec §10). While managed starts
	// are refused (refuseManaged), no exec restart is made: the wake after
	// the stop would be refused, and the box's status names why. A box that
	// declined the flag (swapArming) bounces as today, except an instance a
	// failure record from an armed run holds (final review F6): its wake
	// would be refused, so it keeps running.
	m := armedRollout()
	legacy := m == nil && rolloutUnavailable() == ""
	if h := heldHolds(); legacy && h != nil {
		bounce = h.skipRoutes(routes, bounce)
	}
	if m != nil {
		feedRestarts(routes, moved, scoped, mem.gen, m.Restart)
	} else if legacy {
		bounceExecRoutes(routes, moved, bounce)
		cycleWorkers(routes, moved, bounce, wake)
	}
	// The narrowed half reaches no sidecar at all: scopedHosts gates on
	// release.Adoptable, which is Kind=="instance" and stays that way (design
	// §9.3 — widening it centrally would enrol sidecars in adopt bookkeeping
	// and capability drop-ins that make no sense for them). So a narrowed
	// `carlos restart --host <sidecar>` cycles nothing, which is the SAFE
	// direction — a scoped bounce with no wake beside it would stop a worker
	// that has no other wake source — but it is a gap, not a decision this
	// design made: narrowed restarts landed on main after this branch was
	// written. Reaching a sidecar by name wants its own selector, and is a
	// follow-up rather than something to widen here by accident.
	if legacy {
		bounceScopedRoutes(routes, scoped, bounce)
	}
	if err := requestRootRestart(dd); err != nil {
		log.Printf("agent: request root restart pass: %v", err)
		return false
	}
	log.Printf("agent: restart requested for %d app(s) and narrowed for %d on this box, asked for a root restart pass",
		len(moved), len(scoped))
	return true
}

// restartSeed is the ONE sweep this agent makes on its own initiative: at
// arming, read each app's restart generation once, silently, so the cache the
// leg compares against exists before the first serial move — and hand the same
// sightings to seedRestartMarks so the root pass has a baseline too.
//
// Without it the first request after a boot is the sweep that establishes the
// baseline, and is therefore the one request that does nothing: the person who
// typed `carlos restart` would have to type it twice, having been told
// "requested" the first time. One read per app per agent start is the whole
// cost, and it is paid where a box has just restarted everything anyway.
//
// A read error leaves that app out of the cache (its next sighting is a first
// sight — safe) and blocks the marks seed entirely, for the reason
// seedRestartMarks and restartPass both give: a baseline with holes in it is
// the thing that manufactures restarts.
//
// The one thing it may ask for is a root pass, and only when the baseline on
// disk is INCOMPLETE — see seedRestartMarks' needRootPass. That is the case of
// every box already carrying a pre-narrowing marks file, and asking is what
// keeps the first narrowed request after a roll from being answered "restart
// requested" and then cycling nothing unit-shaped. It is one marker touch,
// once, at arming: the pass it starts is the ordinary one, and once that pass
// has recorded the narrowed marks no later agent start asks again.
//
// Retry grants are the exception to "record, act on nothing" (Astra plan
// 9): every grant in the host-retry log, and every legacy retry in the
// narrowed log, is verified and applied here (applyRetries), so one written
// before this agent started is not lost to the baseline. One spent or made
// stale since clears nothing.
func restartSeed(ctx context.Context, reg *registry.Registry, store release.Store, dd string, mem *restartMemory) {
	routes, err := reg.List()
	if err != nil {
		log.Printf("agent: restart baseline: list routes: %v (the first request may need a second try)", err)
		return
	}
	_, readErrs := restartGenPass(ctx, routes, store, mem.gen)
	// A first sight cycles nothing, but it hands over the retry grants in
	// both logs: they are not seed-skipped (Astra plan 9), so a grant
	// written before this agent started, and not yet applied, is applied
	// now. A grant spent or made stale since clears nothing.
	scoped, scopeErrs := restartScopePass(ctx, routes, store, mem.scope)
	readErrs = append(readErrs, scopeErrs...)
	grants, grantErrs := hostRetryPass(ctx, routes, store, mem)
	if len(grantErrs) > 0 {
		// Not a baseline hole: the host-retry log is no restart signal, and
		// an app it could not be read for is read again on the next move.
		log.Printf("agent: retry baseline: %d apps unreadable: %v", len(grantErrs), grantErrs[0])
	}
	applyRetries(routes, grants, mem)
	if mem.retryHost != nil {
		// A box with nothing that could clear a hold reads every legacy
		// retry at its start as history, silently, as it always has: its
		// line would only say, at every start, that nothing is held.
		applyRetries(routes, scoped, mem)
	}
	if len(readErrs) > 0 {
		log.Printf("agent: restart baseline: %d apps unreadable: %v", len(readErrs), readErrs[0])
		return
	}
	needRootPass, err := seedRestartMarks(restartMarksPath(dd), mem)
	if err != nil {
		log.Printf("agent: seed restart marks: %v", err)
		return
	}
	if !needRootPass {
		return
	}
	if err := requestRootRestart(dd); err != nil {
		// Latency, not correctness: carlos-adopt.timer's hourly pass records
		// the same marks, it just leaves the window open for longer.
		log.Printf("agent: ask for a baselining root restart pass: %v", err)
		return
	}
	log.Printf("agent: restart baseline: the root pass has no narrowed-restart mark for at least one app on this box " +
		"(a marks file written before narrowed restarts existed) — asked for one root pass to record it")
}