package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"slices"
	"sort"
	"strings"
	"time"

	"github.com/novox/mesh-controller/internal/catalogue"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
)

// No build reaches a machine without a gate (novox/hq ADR 0236).
//
// **A send carries the machine's whole declaration.** A plan sending one module to its first machine
// sends every other module whose build moved there too; a cascade, a healer's resend and a whole-mesh
// push do the same. Under the old default (`record`) builds were registered and sent nowhere, so on the
// day the default became `roll` every machine was behind on dozens of builds no gate had seen — and the
// next send of anything would have restarted all of them at once, on every machine.
//
// So **a build moves on a machine only through a send that judges it there**, or a person's:
//
//   - a gated send — a plan's first machine, a release plan's machine — carries every move waiting on that
//     machine, and its gate judges each of them; a pass is each build's verdict, a failure puts back what
//     failed;
//   - a module a send exists for may move anywhere it is sent: a policy of *together*, a rollback, a build
//     whose gate already passed;
//   - a send naming a machine by a person (`push <node>`, the bus step) carries what it carries;
//   - every other send — a plan's "rest", a cascade, a healer's, the bus's user list carried — is refused,
//     or leaves the machine, while a move there waits for a gate. A rebuild that made the same artifacts
//     from the same manifest is no move.
//
// **The release plan** is what walks the waiting moves through: whenever moves wait for a gate that no
// started plan is walking, the mesh opens one — every such machine, one at a time, the control node last,
// each sent and judged before the next. A release plan that fails stops the next one opening on its own:
// a person releases it again (`upgrade release-backlog --why`), after the condition says why.

// sendScope is what a send may carry that no gate has seen.
type sendScope struct {
	// judged are the machines whose every move this send's gate judges.
	judged map[string]bool
	// modules are those a send exists for, which may move wherever it goes.
	modules map[string]bool
	// person is a person's act: it carries what it carries.
	person bool
}

type sendScopeKey struct{}

func withScope(ctx context.Context, s sendScope) context.Context {
	return context.WithValue(ctx, sendScopeKey{}, s)
}

func scopeOf(ctx context.Context) sendScope {
	s, _ := ctx.Value(sendScopeKey{}).(sendScope)
	return s
}

// errUngated is a send refused because it would carry a build no gate has seen.
var errUngated = errors.New("a build waits there for its gate")

// errWalkedElsewhere is a gated send refused because a plan that has started walks a move there.
var errWalkedElsewhere = errors.New("a plan already walking a build there sends it")

// moveFacts is what tells a move from a rebuild, and a gated build from one no gate has seen.
type moveFacts struct {
	current map[string]inventory.CurrentBuild
	fps     map[string]map[string]string
	// srcs is, per module, per commit, its build's source fingerprint (novox/hq issue 280).
	srcs   map[string]map[string]string
	passed map[string]map[string]bool
	plans  []inventory.Plan
}

func readMoveFacts(ctx context.Context, inv *inventory.Inventory) (moveFacts, error) {
	var f moveFacts
	var err error
	if f.current, err = inv.CurrentBuilds(ctx); err != nil {
		return f, err
	}
	if f.fps, err = inv.Fingerprints(ctx); err != nil {
		return f, err
	}
	if f.srcs, err = inv.SourceFingerprints(ctx); err != nil {
		return f, err
	}
	if f.passed, err = inv.PassedCommits(ctx); err != nil {
		return f, err
	}
	f.plans, err = inv.OpenPlans(ctx)
	return f, err
}

// identical is whether two builds of a module put the same thing on a machine: the same commit,
// builds made from the same source (novox/hq issue 280) — the module's tree, the contexts it read, its
// bases and toolchains, whatever digests an image rebuild made of them — or builds that made the same
// artifacts from the same manifest.
func (f moveFacts) identical(module, a, b string) bool {
	if a == b || sameCommit(a, b) {
		return true
	}
	if sa := f.srcs[module][a]; sa != "" && sa == f.srcs[module][b] {
		return true
	}
	fa := f.fps[module][a]
	return fa != "" && fa == f.fps[module][b]
}

// gated is whether a build of a module from this commit — or one identical to it — passed a gate.
func (f moveFacts) gated(module, commit string) bool {
	for c := range f.passed[module] {
		if f.identical(module, c, commit) {
			return true
		}
	}
	return false
}

// unchangedOnEvery is whether a plan's build of a module was made from the source of the build every
// machine running it was last sent (novox/hq issue 280): no move on any of them. False when no machine
// runs it, when what one was sent is not known, or when the build stands for itself.
func unchangedOnEvery(ctx context.Context, inv *inventory.Inventory, module, build string, running []string) (bool, error) {
	if build == "" || len(running) == 0 {
		return false, nil
	}
	same, err := inv.SameSourceCommits(ctx, module, build)
	if err != nil || len(same) == 0 {
		return false, err
	}
	for _, n := range running {
		sent, known, err := inv.SentBuilds(ctx, n)
		if err != nil {
			return false, err
		}
		was, carried := sent[module]
		if !known || !carried || !inRun(same, was) {
			return false, nil
		}
	}
	return true, nil
}

// inRun is whether a commit is one of a run's, however either is abbreviated.
func inRun(run map[string]bool, commit string) bool {
	for c := range run {
		if sameCommit(c, commit) {
			return true
		}
	}
	return false
}

// moves is what a machine's next send would move that no gate has seen: modules whose policy rolls out
// (a recorded one is a person's push), that the machine was sent before, moving to a build not identical
// to the one it runs and that has passed no gate. A machine whose last send's builds are not known names
// none: the cascade holds it whole (ADR 0221).
//
// With all, a move to a build that passed a gate elsewhere counts too: what a gated send carries.
func (f moveFacts) moves(node string, modules []string, sent map[string]string, known, all bool) []inventory.CarriedMove {
	if !known {
		return nil
	}
	var out []inventory.CarriedMove
	for _, m := range modules {
		now := f.current[m]
		was, carried := sent[m]
		if !now.RollOut || !carried || f.identical(m, was, now.Commit) || (!all && f.gated(m, now.Commit)) {
			continue
		}
		out = append(out, inventory.CarriedMove{Module: m, Node: node, From: was, To: now.Commit})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Module < out[j].Module })
	return out
}

// walkedBy is the open plan that has started walking a module's build — sent it to a first machine, not
// yet passed — and whose walk this move would cross; empty when none does. A move of the same build to a
// machine that plan already sent it is not a crossing: the build is there. A move of **another** build of
// the module to that machine is (novox/hq issue 352): on 2026-10-09 a merge's plan sent the control node a
// newer controller while a release's gate was judging the controller there, the release's gate read the
// machine's report against the newer send, failed three builds and put them back under the new plan's
// feet. A release keeps no module records: its open gate's carried moves are its walk.
func (f moveFacts) walkedBy(module, node, to string) string {
	for _, p := range f.plans {
		if !p.Open() {
			continue
		}
		if p.Release != nil {
			g := p.Release.Gate
			if g == nil || g.Verdict != "" {
				continue
			}
			for _, c := range g.Carried {
				if c.Module == module && !(slices.Contains(g.Machines, node) && sameCommit(c.To, to)) {
					return p.ID
				}
			}
			continue
		}
		s, holds := p.Modules[module]
		if !holds || s == nil || s.FirstAt == nil || s.SentAt != nil {
			continue
		}
		if slices.Contains(s.First, node) && sameCommit(s.Commit, to) {
			continue
		}
		if s.Gate != nil && s.Gate.Verdict == inventory.GatePassed {
			continue
		}
		return p.ID
	}
	return ""
}

// machineMoves is the moves no gate has seen on one machine, read from what it would be sent now.
var machineMoves = func(ctx context.Context, open *stores, f moveFacts, node string, all bool) ([]inventory.CarriedMove, error) {
	plan, _, err := planFor(ctx, open, node)
	if err != nil {
		return nil, nil // it cannot be worked out: the send says why
	}
	modules := make([]string, 0, len(plan.Modules))
	for _, m := range plan.Modules {
		modules = append(modules, m.Module)
	}
	sent, known, err := open.inventory.SentBuilds(ctx, node)
	if err != nil {
		return nil, err
	}
	return f.moves(node, modules, sent, known, all), nil
}

// ungatedIn refuses a send whose machines would move a build no gate has seen, outside its scope: the
// machines named, and why; the machine holding the bus, added only for its user list, is left instead.
func ungatedIn(ctx context.Context, open *stores, names []string, addedHolder string) ([]string, error) {
	scope := scopeOf(ctx)
	if scope.person {
		return names, nil
	}
	f, err := readMoveFacts(ctx, open.inventory)
	if err != nil {
		return nil, err
	}
	var kept, refused []string
	for _, n := range names {
		moves, err := machineMoves(ctx, open, f, n, false)
		if err != nil {
			return nil, err
		}
		// The grants step composes every module at the build its machine was last sent (novox/hq issue 490):
		// a move it keeps is not made. Read, not assumed, so a move it could not keep is still refused here.
		var keeps map[string]string
		if everyBuildKept(ctx) {
			if keeps, err = recordedKept(ctx, open, n); err != nil {
				return nil, err
			}
		}
		var waiting []string
		for _, mv := range moves {
			if was, kept := keeps[mv.Module]; kept && sameCommit(was, mv.From) {
				continue
			}
			if !scope.judged[n] && !scope.modules[mv.Module] {
				waiting = append(waiting, fmt.Sprintf("%s %s → %s", mv.Module, short(mv.From), short(mv.To)))
			}
		}
		switch {
		case len(waiting) == 0:
			kept = append(kept, n)
		case n == addedHolder:
			fmt.Printf("%s holds the bus and its user list changed, and it is not sent now: %s wait there for a gate "+
				"— the bus may refuse what was newly granted until it is\n", n, strings.Join(waiting, ", "))
		default:
			refused = append(refused, fmt.Sprintf("%s (%s)", n, strings.Join(waiting, ", ")))
		}
	}
	if len(refused) > 0 {
		return nil, fmt.Errorf("%w: %s — a walk sends them, one machine at a time, each judged (novox/hq ADR "+
			"0236); `upgrade backlog` lists them", errUngated, strings.Join(refused, "; "))
	}
	return kept, nil
}

// gatedSend sends one machine everything waiting there, under a gate that judges it all: what moved is
// answered, with the build each moved to, for the gate to judge and to put back. Refused while a plan that
// has started walks one of those builds elsewhere: that plan sends it here once its gate passed.
//
// owns are the moves the send exists for — every module of a plan's tier whose first machine this is, in
// one send (novox/hq issue 281): a send carries the machine's whole declaration (ADR 0221), so a send per
// module was the same declaration sent again and again, each one setting aside the one before.
//
// And the grants step first, when the send changes what the bus's user list must say (novox/hq issue 490):
// answered for the gate to keep, nil when there was none.
func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.CarriedMove) ([]inventory.CarriedMove, []string, *inventory.GrantsStep, error) {
	inv := open.inventory
	f, err := readMoveFacts(ctx, inv)
	if err != nil {
		return nil, nil, nil, err
	}
	moves, err := machineMoves(ctx, open, f, node, true)
	if err != nil {
		return nil, nil, nil, err
	}
	own := func(module string) bool {
		return slices.ContainsFunc(owns, func(o inventory.CarriedMove) bool { return o.Module == module })
	}
	for _, mv := range moves {
		if own(mv.Module) {
			continue
		}
		if id := f.walkedBy(mv.Module, node, mv.To); id != "" {
			return nil, nil, nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
				mv.Module, short(mv.To), node, id)
		}
	}
	// **And the plan's own modules wait too** (novox/hq issue 352): a walk of the same module by another
	// plan, or a release, on this machine is not crossed with a newer build; this send waits for its gate.
	for _, o := range owns {
		if id := f.walkedBy(o.Module, node, o.To); id != "" {
			return nil, nil, nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
				o.Module, short(o.To), node, id)
		}
	}
	for _, o := range owns {
		i := slices.IndexFunc(moves, func(mv inventory.CarriedMove) bool { return mv.Module == o.Module })
		switch {
		case i < 0:
			moves = append(moves, o)
		case moves[i].Build == "":
			moves[i].Build = o.Build
		}
	}
	if len(moves) == 0 && len(owns) == 0 {
		return nil, nil, nil, nil
	}
	for i := range moves {
		if moves[i].Build == "" {
			if moves[i].Build, err = inv.BuildOf(ctx, moves[i].Module, moves[i].To); err != nil {
				return nil, nil, nil, err
			}
		}
	}
	sort.Slice(moves, func(i, j int) bool { return moves[i].Module < moves[j].Module })
	sayRecreations(ctx, open, moves)
	sent, grants, err := sendJudged(ctx, open, node)
	if err != nil {
		return nil, nil, nil, err
	}
	return moves, sent, grants, nil
}

// passCarried keeps a pass as the verdict of every build the gate judged beside its own module.
func passCarried(ctx context.Context, open *stores, p *inventory.Plan, g *inventory.PlanGate, except string) {
	seen := map[string]bool{}
	for _, c := range g.Carried {
		if c.Module == except || c.Build == "" || seen[c.Build] {
			continue
		}
		seen[c.Build] = true
		if err := open.inventory.RecordGate(ctx, inventory.GateVerdict{Build: c.Build, Module: c.Module, Commit: c.To,
			Previous: c.From, Plan: p.ID, Machines: []string{c.Node}, Verdict: inventory.GatePassed, Why: whyFor(g, c.Module),
			Component: coreComponent(c.Module), JudgingFrom: g.Since}); err != nil {
			fmt.Printf("%s: %s passed its gate on %s, and the verdict could not be kept: %v\n", p.ID, c.Module, c.Node, err)
		}
	}
}

// failCarried puts back every build the gate carried that it found wanting, on the machines that were
// sent it, once each.
func failCarried(ctx context.Context, open *stores, p *inventory.Plan, g *inventory.PlanGate, except string) {
	var notes []string
	if p.Note != "" {
		notes = append(notes, p.Note)
	}
	done := map[string]bool{except: true}
	for _, m := range g.Returned {
		done[m] = true // put back at once when it broke
	}
	// **What passed on its own keeps its pass** (novox/hq ADR 0254, issue 318): healthy for the passes the
	// gate asks while another module of the send failed, it is neither put back nor left unjudged — a build
	// left without a verdict is one more move every other walk would wait for.
	if kept := keepPassing(ctx, open, p, g, except); len(kept) > 0 {
		notes = append(notes, "passed on their own and kept: "+strings.Join(kept, ", "))
		for _, m := range kept {
			done[m] = true
		}
	}
	for _, c := range g.Carried {
		if done[c.Module] || (len(g.Failing) > 0 && !slices.Contains(g.Failing, c.Module)) {
			continue
		}
		done[c.Module] = true
		machines, err := sentTheBuild(ctx, open, c.Module, c.To)
		if err != nil || len(machines) == 0 {
			machines = []string{c.Node}
		}
		state := &inventory.PlanModule{Build: c.Build, Previous: c.From, Commit: c.To}
		// A module of the plan's tier sent in the same send (issue 281) keeps its own record of it.
		if s := p.Modules[c.Module]; s != nil && except != "" && s.GatedBy == except && s.Build == c.Build {
			state = s
		}
		p.Note = ""
		gateFailed(ctx, open, p, c.Module, state, machines, g.Why)
		notes = append(notes, p.Note)
	}
	p.State = inventory.PlanFailed
	p.Note = strings.Join(notes, "; ")
}

// keepPassing keeps a pass as the verdict of every build the gate carried that passed on its own when the
// send failed (ADR 0254), except the plan's own module; answers the modules kept.
func keepPassing(ctx context.Context, open *stores, p *inventory.Plan, g *inventory.PlanGate, except string) []string {
	var kept []string
	seen := map[string]bool{}
	for _, c := range g.Carried {
		if c.Module == except || c.Build == "" || seen[c.Build] || !slices.Contains(g.Passing, c.Module) {
			continue
		}
		seen[c.Build] = true
		why := passedAloneWhy(g, c.Module)
		if err := open.inventory.RecordGate(ctx, inventory.GateVerdict{Build: c.Build, Module: c.Module, Commit: c.To,
			Previous: c.From, Plan: p.ID, Machines: []string{c.Node}, Verdict: inventory.GatePassed, Why: why,
			Component: coreComponent(c.Module), JudgingFrom: g.Since}); err != nil {
			fmt.Printf("%s: %s passed its gate on %s on its own, and the verdict could not be kept: %v\n", p.ID, c.Module,
				c.Node, err)
			continue
		}
		if !slices.Contains(kept, c.Module) {
			kept = append(kept, c.Module)
		}
	}
	return kept
}

// putBackBroken puts back, at once, every module a gate found broken and has not put back yet (issue 318
// review): a witness that reverted a core component, or a machine that refused or failed its send, is not
// left registered at the build that broke for as long as the modules beside it take to be judged — a resend
// meanwhile would declare it again. The send goes on judging the others to their own verdicts; its own
// verdict is still failed. lead and leadState are the module a plan's gate is kept on, when it is a plan's.
// Answers whether anything was put back.
func putBackBroken(ctx context.Context, open *stores, p *inventory.Plan, g *inventory.PlanGate, lead string,
	leadState *inventory.PlanModule) bool {
	var todo []string
	for _, m := range g.Broken {
		if !slices.Contains(g.Returned, m) {
			todo = append(todo, m)
		}
	}
	if len(todo) == 0 || g.Verdict != "" {
		return false
	}
	stateWas, noteWas := p.State, p.Note
	batched, back := batchingRollbacks(ctx)
	var said []string
	for _, m := range todo {
		machines := g.Machines
		var state *inventory.PlanModule
		if m == lead && leadState != nil {
			judged := *leadState
			judged.Gate = nil // its own record of the failure; the send's gate goes on judging
			state = &judged
		} else {
			i := slices.IndexFunc(g.Carried, func(c inventory.CarriedMove) bool { return c.Module == m })
			if i < 0 {
				continue
			}
			c := g.Carried[i]
			state = &inventory.PlanModule{Build: c.Build, Previous: c.From, Commit: c.To}
			if sent, err := sentTheBuild(ctx, open, m, c.To); err == nil && len(sent) > 0 {
				machines = sent
			} else {
				machines = []string{c.Node}
			}
		}
		// Counted as put back only once there is something to put back (a carried move or the plan's own).
		g.Returned = append(g.Returned, m)
		p.Note = ""
		gateFailed(batched, open, p, m, state, machines, g.BrokenWhy)
		if m == lead && leadState != nil {
			leadState.Why = "put back at once, its send still judged: " + g.BrokenWhy
		}
		said = append(said, m)
	}
	sendRollbacks(ctx, open, p, back)
	p.State = stateWas
	p.Note = noteWas
	if len(said) > 0 {
		p.Note = fmt.Sprintf("put back at once: %s (%s); the rest of the send judged to their own verdicts",
			strings.Join(said, ", "), g.BrokenWhy)
		fmt.Printf("%s: %s\n", p.ID, p.Note)
	}
	return true
}

// passedAloneWhy is the verdict of a module that passed on its own while its send failed, with its own wait.
func passedAloneWhy(g *inventory.PlanGate, module string) string {
	why := fmt.Sprintf("healthy %d times on its own while the send failed: %s", g.Healthy[module], g.Why)
	if w := g.Waits[module]; w != "" {
		why += waitsPrefix + w
	}
	return why
}

// sentTheBuild is every machine running a module that was last sent this build of it.
func sentTheBuild(ctx context.Context, open *stores, module, commit string) ([]string, error) {
	running, err := open.inventory.Running(ctx, module)
	if err != nil {
		return nil, err
	}
	var out []string
	for _, n := range running {
		sent, known, err := open.inventory.SentBuilds(ctx, n)
		if err != nil {
			return nil, err
		}
		if known && sameCommit(sent[module], commit) {
			out = append(out, n)
		}
	}
	return out, nil
}

// releaseRepository is what a release plan says it is for, where a merge's says its repository.
const releaseRepository = "the builds waiting for a gate"

// releaseHeard is the machines a release plan may send: heard within their heartbeat's bound. A machine
// away is left, not judged against a bound it cannot meet. A variable so a test says who is heard.
var releaseHeard = func(ctx context.Context, open *stores) (map[string]bool, error) {
	if d := doctorFrom; d != nil && d.watchdogs != nil {
		return heardMachines(d), nil
	}
	reports, err := open.inventory.LastReports(ctx)
	if err != nil {
		return nil, err
	}
	out := map[string]bool{}
	for _, r := range reports {
		if r.At != nil && time.Since(*r.At) < 15*time.Minute {
			out[r.Node] = true
		}
	}
	return out, nil
}

// backlogNow is what the newest look found waiting, for `upgrade backlog` and the gate's probe.
var backlogNow struct {
	held    string
	waiting map[string][]inventory.CarriedMove
}

// waitingMoves is every machine's moves no gate has seen and no started plan walks.
func waitingMoves(ctx context.Context, open *stores, all bool) (map[string][]inventory.CarriedMove, error) {
	f, err := readMoveFacts(ctx, open.inventory)
	if err != nil {
		return nil, err
	}
	nodes, err := open.inventory.Nodes(ctx)
	if err != nil {
		return nil, err
	}
	out := map[string][]inventory.CarriedMove{}
	for _, n := range nodes {
		moves, err := machineMoves(ctx, open, f, n.Name, all)
		if err != nil {
			return nil, err
		}
		for _, mv := range moves {
			if f.walkedBy(mv.Module, n.Name, mv.To) == "" {
				out[n.Name] = append(out[n.Name], mv)
			}
		}
	}
	return out, nil
}

// releaseBacklog opens a release plan when builds wait for a gate and none is open; not after a release
// plan failed, until a person releases one (by). Called with the plans held.
func releaseBacklog(ctx context.Context, open *stores, by string) (*inventory.Plan, error) {
	inv := open.inventory
	plans, err := inv.OpenPlans(ctx)
	if err != nil {
		return nil, err
	}
	for _, p := range plans {
		if p.Release != nil {
			if by != "" {
				return nil, fmt.Errorf("%s is already releasing what waits; `plans %s` says where it is", p.ID, p.ID)
			}
			return nil, nil
		}
	}
	waiting, err := waitingMoves(ctx, open, false)
	if err != nil {
		return nil, err
	}
	backlogNow.waiting, backlogNow.held = waiting, ""
	if len(waiting) == 0 {
		return nil, nil
	}
	// Opened by what no gate has seen; it walks every machine where anything of the release waits — a
	// build that passed on the first machine still goes to the next one by this plan, judged there too.
	if waiting, err = waitingMoves(ctx, open, true); err != nil {
		return nil, err
	}
	if by == "" {
		recent, err := inv.RecentPlans(ctx, 50)
		if err != nil {
			return nil, err
		}
		for _, p := range recent {
			if p.Release == nil {
				continue
			}
			if p.State == inventory.PlanFailed {
				backlogNow.held = fmt.Sprintf("%s failed (%s); what waits is released again by a person", p.ID, p.Note)
				return nil, nil
			}
			break
		}
	}
	heard, err := releaseHeard(ctx, open)
	if err != nil {
		return nil, err
	}
	controllers, err := inv.Running(ctx, catalogue.ControllerSeatName)
	if err != nil {
		return nil, err
	}
	var order, last []string
	modules := map[string]bool{}
	for node, moves := range waiting {
		if !heard[node] {
			continue
		}
		for _, mv := range moves {
			modules[mv.Module] = true
		}
		if slices.Contains(controllers, node) {
			last = append(last, node)
		} else {
			order = append(order, node)
		}
	}
	if len(order)+len(last) == 0 {
		return nil, nil
	}
	sort.Strings(order)
	sort.Strings(last)
	order = append(order, last...)
	names := make([]string, 0, len(modules))
	for m := range modules {
		names = append(names, m)
	}
	sort.Strings(names)
	now := time.Now().UTC()
	p := inventory.Plan{ID: fmt.Sprintf("release-%d", now.UnixNano()), Repository: releaseRepository,
		Created: now, State: inventory.PlanRolling, Tiers: [][]string{names}, Modules: map[string]*inventory.PlanModule{},
		Release: &inventory.PlanRelease{Order: order, By: by},
		Note: fmt.Sprintf("%d build(s) wait for a gate on %s; one machine at a time, each judged", len(names),
			strings.Join(order, ", "))}
	if err := inv.SavePlan(ctx, &p); err != nil {
		return nil, err
	}
	fmt.Printf("%s: %s\n", p.ID, p.Note)
	return &p, nil
}

// advanceRelease takes one step of a release plan: the next machine sent everything waiting there under a
// gate, or the machine being judged judged once more; the plan done when every machine is.
func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool, error) {
	r := p.Release
	now := time.Now().UTC()
	if r.Gate == nil {
		heard, err := releaseHeard(ctx, open)
		if err != nil {
			return false, err
		}
		for r.Next < len(r.Order) {
			node := r.Order[r.Next]
			if !heard[node] {
				r.Skipped = append(r.Skipped, node)
				r.Next++
				continue
			}
			moves, sent, grants, err := gatedSend(ctx, open, node, nil)
			if errors.Is(err, errWalkedElsewhere) {
				note := fmt.Sprintf("waiting before %s: %v", node, err)
				changed := p.Note != note
				p.Note = note
				return changed, nil
			}
			if err != nil {
				return false, fmt.Errorf("sending %s what waits there: %w", node, err)
			}
			if len(moves) == 0 {
				r.Done = append(r.Done, node)
				r.Next++
				continue
			}
			r.Gate = &inventory.PlanGate{Machines: sent, Since: &now, Carried: moves, Sent: sentNow(ctx, open.inventory, sent),
				Grants: grants}
			p.Note = fmt.Sprintf("sent %s %d build(s) that waited for a gate; judging them there", node, len(moves))
			if said := grantsSaid(grants); said != "" {
				p.Note += "; " + said
			}
			if said := recreationsSaid(moves); said != "" {
				p.Note += "; " + said
			}
			fmt.Printf("%s: %s\n", p.ID, p.Note)
			return true, nil
		}
		p.State, p.Tier = inventory.PlanDone, len(p.Tiers)
		p.Note = fmt.Sprintf("released on %s", orNone(strings.Join(r.Done, ", ")))
		if len(r.Skipped) > 0 {
			p.Note += "; not heard from, left as they were: " + strings.Join(r.Skipped, ", ")
		}
		return true, nil
	}
	g := r.Gate
	verdict, err := judgeMoves(ctx, open, g, nil, now)
	if err != nil {
		return false, err
	}
	switch verdict {
	case "":
		if putBackBroken(ctx, open, p, g, "", nil) {
			return true, nil
		}
		note := fmt.Sprintf("judging %s: %s", strings.Join(g.Machines, ", "), gateLine(g))
		changed := p.Note != note
		p.Note = note
		return changed, nil
	case inventory.GatePassed:
		passCarried(ctx, open, p, g, "")
		r.Done = append(r.Done, firstOf(g.Machines))
		r.Next++
		r.Gate = nil
		return true, nil
	case inventory.GateSuperseded:
		// Another send moved a judged module on the judged machine (novox/hq issue 352): no verdict on what
		// was carried, nothing put back, and this release ends; the builds still waiting are released again
		// by the next pass, judged afresh.
		p.State = inventory.PlanSuperseded
		p.Note = fmt.Sprintf("superseded on %s: %s — nothing judged, nothing put back; what still waits is released again",
			strings.Join(g.Machines, ", "), g.Why)
		fmt.Printf("%s: %s\n", p.ID, p.Note)
		return true, nil
	}
	p.Note = ""
	batched, back := batchingRollbacks(ctx)
	failCarried(batched, open, p, g, "")
	sendRollbacks(ctx, open, p, back)
	p.Note = fmt.Sprintf("failed its gate on %s: %s — %s", strings.Join(g.Machines, ", "), g.Why, p.Note)
	fmt.Printf("%s: %s\n", p.ID, p.Note)
	return true, nil
}

// backlogObservation is what the gate's probe says of a release held after a failure.
func backlogObservation() []conditions.Observation {
	if backlogNow.held == "" || len(backlogNow.waiting) == 0 {
		return nil
	}
	n := 0
	var machines []string
	seen := map[string]bool{}
	var modules []string
	for node, moves := range backlogNow.waiting {
		n += len(moves)
		machines = append(machines, node)
		for _, mv := range moves {
			if !seen[mv.Module] {
				seen[mv.Module] = true
				modules = append(modules, mv.Module)
			}
		}
	}
	sort.Strings(machines)
	sort.Strings(modules)
	o := conditions.Observation{Scope: conditions.ScopeMesh, ID: "release", Token: "held", Kind: "release-held",
		Severity: conditions.Warning, Resolver: conditions.ResolverOperator, Also: machines,
		Summary: fmt.Sprintf("%d build move(s) on %s wait for a gate and are not released: %s — `upgrade backlog` lists "+
			"them, `upgrade release-backlog --why …` releases them", n, strings.Join(machines, ", "), backlogNow.held)}
	w := releaseHeldWords(modules, machines)
	o.Headline, o.Explanation, o.Resolved, o.Needs = w.Headline, w.Explanation, w.Resolved, w.Needs
	return []conditions.Observation{o}
}

// backlogCommand is `upgrade backlog`, read-only, and `upgrade release-backlog --why`.
func backlogCommand(ctx context.Context, sub string, args []string) error {
	set := flag.NewFlagSet("upgrade "+sub, flag.ContinueOnError)
	why := addHandActFlags(set)
	if rest, err := parseAround(set, args); err != nil {
		return err
	} else if len(rest) > 0 {
		return errors.New("upgrade backlog | upgrade release-backlog --why <text>")
	}
	if sub == "release-backlog" {
		if err := why.require("upgrade release-backlog"); err != nil {
			return err
		}
	}
	open, err := openStores(ctx)
	if err != nil {
		return err
	}
	defer open.Close()
	if sub == "release-backlog" {
		release, err := open.inventory.HoldPlans(ctx, true)
		if err != nil {
			return err
		}
		defer release()
		why.record(ctx, "upgrade release-backlog", nil)
		p, err := releaseBacklog(ctx, open, link.Caller())
		if err != nil {
			return err
		}
		if p == nil {
			fmt.Println("nothing waits for a gate on any machine heard from: nothing to release")
			return nil
		}
		fmt.Printf("%s releases it; `plans %s` says where it is\n", p.ID, p.ID)
		return nil
	}
	waiting, err := waitingMoves(ctx, open, false)
	if err != nil {
		return err
	}
	if len(waiting) == 0 {
		fmt.Println("no build waits for a gate on any machine")
		return nil
	}
	nodes := make([]string, 0, len(waiting))
	for n := range waiting {
		nodes = append(nodes, n)
	}
	sort.Strings(nodes)
	for _, n := range nodes {
		fmt.Printf("%s: %d build(s) wait for a gate\n", n, len(waiting[n]))
		for _, mv := range waiting[n] {
			fmt.Printf("  %-28s %s → %s\n", mv.Module, short(mv.From), short(mv.To))
		}
	}
	return nil
}

// sayRecreations says, on each move a send carries, what it does to the module's containers (novox/hq
// ADR 0245): the build the machine ran against the one it is sent, container by container. Left unsaid
// for a move whose earlier build is not in the records.
func sayRecreations(ctx context.Context, open *stores, moves []inventory.CarriedMove) {
	if len(moves) == 0 {
		return
	}
	shelf, err := open.inventory.Catalogue(ctx)
	if err != nil {
		return
	}
	for i, mv := range moves {
		to, held := shelf[mv.Module]
		if !held || mv.From == "" {
			continue
		}
		from, found, err := open.inventory.ManifestAt(ctx, mv.Module, mv.From)
		if err != nil || !found {
			continue
		}
		moves[i].Recreates = catalogue.Recreates(from, to).Say(mv.Module)
	}
}

// sayWhatAPushRecreates says, per machine a push is about to send, every module the send moves and what
// the move recreates (novox/hq ADR 0245) — the same words a gated send keeps on its plan. A person's push
// carries every module's new build, whatever its policy and whether or not a gate saw it, so it is the
// send that most needs to say so: on 2026-10-07 a `push` of one machine moved mail to a new build and
// recreated one of its containers with a new image, and the answer said nothing about it. A machine whose
// last send's builds are not known is not said, and neither is a failure to read: the push goes on.
func sayWhatAPushRecreates(ctx context.Context, open *stores, sending []readyNode, w io.Writer) {
	if len(sending) == 0 {
		return
	}
	f, err := readMoveFacts(ctx, open.inventory)
	if err != nil {
		return
	}
	for _, r := range sending {
		sent, known, err := open.inventory.SentBuilds(ctx, r.node)
		if err != nil || !known {
			continue
		}
		moves := pushMoves(f, r.node, r.declared.Builds, sent)
		if len(moves) == 0 {
			continue
		}
		sayRecreations(ctx, open, moves)
		fmt.Fprintf(w, "%s moves %d module(s):\n", r.node, len(moves))
		for _, mv := range moves {
			said := mv.Recreates
			if said == "" {
				said = "recreates none of its containers, or what it ran is not in the records"
			}
			fmt.Fprintf(w, "  %-28s %s → %s: %s\n", mv.Module, short(mv.From), short(mv.To), said)
		}
	}
}

// pushMoves is what a send carrying these builds moves on a machine last sent `sent`: every module it ran
// before at a build not identical to the one it is now sent. A module new to the machine is not a move.
func pushMoves(f moveFacts, node string, carries, sent map[string]string) []inventory.CarriedMove {
	var out []inventory.CarriedMove
	for m, to := range carries {
		was, ran := sent[m]
		if !ran || was == "" || to == "" || f.identical(m, was, to) {
			continue
		}
		out = append(out, inventory.CarriedMove{Module: m, Node: node, From: was, To: to})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Module < out[j].Module })
	return out
}

// recreationsSaid is every recreation a send's moves say, joined: what a plan's note carries.
func recreationsSaid(moves []inventory.CarriedMove) string {
	var said []string
	for _, mv := range moves {
		if mv.Recreates != "" {
			said = append(said, fmt.Sprintf("on %s %s", mv.Node, mv.Recreates))
		}
	}
	return strings.Join(said, "; ")
}

// The grants step (novox/hq issue 490).
//
// The bus's user list — every module's grants, composed mesh-wide from the catalogue — travels only in the
// declaration of the machine holding the bus. A send to any other machine that changed it put that machine
// first (issue 249), unless a build waited there for a gate: then ungatedIn left it out, and said so. On
// 2026-10-11 a merge gave restic a new health tool; its walk sent restic to its first machine and judged it
// there, the machine's node-engine asked the tool and the bus refused it ("this host's grant does not name
// …"), because the machine holding the bus also runs restic, whose new build waited for that very gate. The
// gate could not pass; a person pushed the bus's machine by hand, which carried restic's new build there
// unjudged.
//
// So before a gated send, when the user list composed now is not the one the machine holding the bus was
// last sent, that machine is sent its declaration with **every build kept at the one it runs**: the new
// list, and nothing of any module's new code. Said on the gate (`plans`), and, when it cannot be sent, said
// with why and passed over: the send goes on as it did before, and its gate says what it finds.

// brokerBehindOf is brokerBehind: a variable so a test of the walk needs no store.
var brokerBehindOf = brokerBehind

// grantsStep sends the machine holding the bus its user list alone, ahead of a gated send to node, when the
// list changed; answers what it did, or nil when there was nothing to send.
func grantsStep(ctx context.Context, open *stores, node string) *inventory.GrantsStep {
	holder, behind, err := brokerBehindOf(ctx, open, []string{node})
	if err != nil {
		fmt.Printf("grants step before %s: whether the bus's user list changed could not be read: %v\n", node, err)
		return &inventory.GrantsStep{Node: "the machine holding the bus", Failed: err.Error()}
	}
	if holder == "" || holder == node || !behind {
		// No bus with a user list, the gate's own machine holds it (its send carries the list first), or the
		// list is the one already sent.
		return nil
	}
	if _, err := sendRollout(keepingEveryBuild(withScope(ctx, sendScope{})), open, []string{holder}); err != nil {
		fmt.Printf("grants step: the bus's user list could not be sent to %s ahead of %s, which is sent without "+
			"it — the bus may refuse what the send newly grants: %v\n", holder, node, err)
		return &inventory.GrantsStep{Node: holder, Failed: err.Error()}
	}
	now := time.Now().UTC()
	fmt.Printf("grants step: %s sent the bus's user list alone, every build there kept, before %s is judged\n",
		holder, node)
	return &inventory.GrantsStep{Node: holder, At: &now}
}

// errNotGrantsOnly is a grants step refused unsent: its declaration would change more than the user list.
var errNotGrantsOnly = errors.New("the grants step would change more than the bus's user list, and is not sent")

// grantsOnlyIn is what a grants step's declaration, composed for sending, would change on the machine holding
// the bus besides its user list, against what it was last sent (novox/hq issue 490); empty when nothing else.
// Judged by sendToEach on the very declaration it then sends — one composition, judged, then sent or refused
// unsent. Kept builds are not the whole of a declaration: a new assignment, a settings change, an assignment
// taken away, a provision's logins are composed as the mesh holds them now, and a step that carried any of
// it would be the unjudged change this step exists to avoid. Compared resource by resource with the summary
// the last send kept (ADR 0217): exact, rather than a list of the cases that can differ. A module there that
// was never sent is a new assignment, refused whatever its resources.
func grantsOnlyIn(ctx context.Context, open *stores, holder string, plan catalogue.Resolution, declared sendable) (string, error) {
	inv := open.inventory
	sent, known, err := inv.SentBuilds(ctx, holder)
	if err != nil {
		return "", err
	}
	if !known {
		return fmt.Sprintf("what %s was last sent is not known", holder), nil
	}
	var other []string
	listID := ""
	for _, m := range plan.Modules {
		if _, was := sent[m.Module]; !was {
			other = append(other, m.Module+" newly assigned")
		}
		if m.BusUsers != "" && m.ClaimsSeat(catalogue.BrokerSeat) {
			listID = m.Module + "." + catalogue.BusUsersID()
		}
	}
	var facts *moveFacts
	for m, was := range sent {
		now, carried := declared.Builds[m]
		if !carried || sameCommit(now, was) {
			continue
		}
		if facts == nil {
			f, err := readMoveFacts(ctx, inv)
			if err != nil {
				return "", err
			}
			facts = &f
		}
		if !facts.identical(m, was, now) {
			other = append(other, fmt.Sprintf("%s %s → %s", m, short(was), short(now)))
		}
	}
	for _, f := range declared.foreseen {
		other = append(other, f+" would be minted")
	}
	body, err := declared.Body()
	if err != nil {
		return "", err
	}
	after, err := summarize(body)
	if err != nil {
		return "", err
	}
	prev, err := inv.SentSummary(ctx, holder)
	if err != nil {
		return "", err
	}
	if len(prev) == 0 {
		return fmt.Sprintf("what %s was last sent is not kept for comparison", holder), nil
	}
	var before []sentResource
	if err := json.Unmarshal(prev, &before); err != nil {
		return "", fmt.Errorf("what %s was last sent is kept in a form this version does not read: %w", holder, err)
	}
	other = append(other, otherThanTheList(diffSent(before, after), listID)...)
	sort.Strings(other)
	if len(other) == 0 {
		return "", nil
	}
	return "its send would change more than the bus's user list: " + strings.Join(other, "; "), nil
}

// otherThanTheList is what a diff against the last send changes besides the user list's content: every
// resource added, removed or changed, but the list's own resource when its content is all that changed —
// its path, mode and every other field must be as last sent (novox/hq issue 490).
func otherThanTheList(d sentDiff, listID string) []string {
	var other []string
	for _, id := range d.Added {
		other = append(other, "+"+id)
	}
	for _, id := range d.Removed {
		other = append(other, "-"+id)
	}
	for _, c := range d.Changed {
		if c.ID == listID && len(c.Fields) == 1 && c.Fields[0] == "content" {
			continue
		}
		other = append(other, fmt.Sprintf("~%s (%s)", c.ID, strings.Join(c.Fields, ", ")))
	}
	return other
}

// sendJudged is a gated send's sends: the grants step when the user list changed, then the machine judged.
func sendJudged(ctx context.Context, open *stores, node string) ([]string, *inventory.GrantsStep, error) {
	grants := grantsStep(ctx, open, node)
	sent, err := sendRollout(withScope(ctx, sendScope{judged: map[string]bool{node: true}}), open, []string{node})
	return sent, grants, err
}

// grantsSaid is a grants step as `plans` and a walk's note say it.
func grantsSaid(g *inventory.GrantsStep) string {
	switch {
	case g == nil:
		return ""
	case g.Failed != "":
		return fmt.Sprintf(grantsStepWord+" to %s FAILED, sent without it: %s", g.Node, g.Failed)
	default:
		return fmt.Sprintf(grantsStepWord+": %s sent the bus's user list first, its builds kept", g.Node)
	}
}