// Operator CLI verbs: route management (add/remove/instances) and the
// release pipeline (ship/promote/channels/rollback). Route verbs adapted
// from titogo internal/carlos/cli.go — deliberately thin: a route is a row,
// and starting the process is the agent's job.
package main

import (
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/caps"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/host"
	"github.com/carlosframework/platform/internal/ledger"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
	"github.com/carlosframework/platform/internal/version"
)

// stdout is the print seam for the four release-pipeline verbs that now
// speak the console API (ship/promote/channels/rollback): a package-level
// var rather than an os.Stdout literal, so a test can swap it for a buffer
// and assert on it. Every OTHER verb in this file still writes straight to
// os.Stdout/fmt.Printf, unchanged — this seam exists only where Task 5 added
// behaviour worth pinning with an assertion; log.Fatal paths are untouched by
// it on purpose, exactly as they were before.
var stdout io.Writer = os.Stdout

func dataDir() string {
	if d := os.Getenv("CARLOS_DATA_DIR"); d != "" {
		return d
	}
	return "/data"
}

func defaultRegistry() string { return filepath.Join(dataDir(), "registry.db") }

// reservedSlugs comes from configuration, not a hardcoded list (titogo's
// lesson: its reserved list named Tito features).
func reservedSlugs() map[string]bool {
	out := map[string]bool{}
	for _, s := range strings.Split(os.Getenv("CARLOS_RESERVED_SLUGS"), ",") {
		if s = strings.TrimSpace(s); s != "" {
			out[s] = true
		}
	}
	return out
}

// refuseAddNested stops a re-add of a host whose row is on the nested
// layout. Upsert overwrites the socket, database path and hibernate flag
// with this verb's flat defaults, which would leave the row pointing at
// the shared directory over files that live under the instance tree, and
// the next wake would restore a second database there (design §8.2,
// downgrade rules). A host with a layout journal is mid-move, row or no
// row, and a re-add would rewrite what recovery is about to check, so it
// is refused the same way. A host with no row, or a flat one, goes ahead.
func refuseAddNested(host string, existing *registry.Route, dataDir string) error {
	if (existing != nil && provision.NestedPaths(dataDir, *existing)) || provision.LayoutJournalExists(dataDir, host) {
		return fmt.Errorf("add %s: its paths are nested. revert the layout first; re-adding it would point it back at the shared directory", host)
	}
	return nil
}

// addMintsNested reports whether `carlos add` on this box mints the
// nested defaults: the box's layout intent names nested, or, with no
// intent recorded, the box is already nested by default, which this verb
// can only tell from the registry (nestedByEvidence: an activator-managed
// row already at the nested paths), because the agent's readiness gate is
// not reachable from an operator's shell. With no deploy store in the
// environment (a box before its bootstrap, a dev shell) the registry's
// evidence is all there is. A recorded flat intent is the opt-out and
// wins over the evidence. An intent that cannot be read is an error, not
// a guess: a flat row on a nested box is the thing this exists to
// prevent, and the operator can pass --socket and --db to say so
// explicitly.
func addMintsNested(ctx context.Context, store release.Store, boxLabel func() string, routes []registry.Route, dd string) (bool, error) {
	if store != nil {
		in, found, err := release.LoadLayoutIntent(ctx, store, boxLabel())
		if err != nil {
			return false, fmt.Errorf("could not read the box's layout intent, so the row's paths cannot be chosen: %v. Retry, or pass --socket and --db", err)
		}
		if found {
			return in.Mode == release.LayoutModeNested, nil
		}
	}
	for _, rt := range routes {
		if activator.ActivatorManaged(rt) && provision.NestedPaths(dd, rt) {
			return true, nil
		}
	}
	return false, nil
}

// boxLabelForAdd is the label this box's status publishes under, which is
// what a layout intent's scope names. A function value so runAdd's `host`
// string does not shadow the package.
var boxLabelForAdd = func() string { return host.BoxLabel() }

func runAdd(args []string) {
	fs := flag.NewFlagSet("add", flag.ExitOnError)
	reg := fs.String("registry", defaultRegistry(), "registry database")
	app := appFlag(fs, "app whose releases this route runs")
	acct := fs.String("account", "ops", "account (sqid or name) the app belongs to")
	acctID := fs.String("account-id", "", "account sqid, matched as a sqid ONLY — supersedes --account")
	slug := fs.String("slug", "", "root-domain short name (optional)")
	kind := fs.String("kind", "instance", "instance | service | static")
	socket := fs.String("socket", "", "unix socket upstream")
	addr := fs.String("addr", "", "TCP upstream host:port (whole-app tenants)")
	channel := aliasedString(fs, "channel", "stable", "release channel to follow")
	hibernate := fs.Bool("hibernate", false, "hibernate when idle")
	statusNamed := fs.Bool("status-named", false, "opt this app into being named on the public status page")
	backing := fs.String("backing", "", `process backing: "" (agent exec child) | "unit" (carlos-app@<host> systemd unit)`)
	unit := fs.String("unit", "", "systemd unit that owns this route's process when it is neither an exec child nor a carlos-app@ tenant (e.g. console.service)")
	dbPath := fs.String("db", "", "instance database path")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	if fs.NArg() != 1 {
		log.Fatal("usage: carlos add [flags] <host>")
	}
	host := fs.Arg(0)
	if err := addRail(host, *kind); err != nil {
		log.Fatalf("add %s: %v", host, err)
	}
	if err := appRail(*app); err != nil {
		log.Fatalf("add %s: %v", host, err)
	}
	if reservedSlugs()[*slug] {
		log.Fatalf("slug %q is reserved", *slug)
	}
	if *backing != "" && *backing != registry.BackingUnit {
		log.Fatalf("backing %q: want \"\" or %q", *backing, registry.BackingUnit)
	}
	// Checked here, before anything is opened or written, for the same reason
	// --backing is: registry.SetUnit would refuse it too, but a route created
	// with a typo'd unit is a route whose adoption silently restarts nothing
	// — the failure this column exists to end.
	if !registry.ValidUnitName(*unit) {
		log.Fatalf("add %s: unit %q: want a systemd service unit name (e.g. console.service)", host, *unit)
	}
	// The same combination `carlos route --unit` refuses (unitRail), refused at
	// the OTHER door onto these two columns — and this is the door that could
	// mint the row from nothing. A route that is both unit-backed and
	// unit-named is two answers to one question, and the dispatch resolves it
	// by honouring the name: the carlos-app@<host> unit would then never be
	// cycled by adoption again, which is issue #110 recreated for an ordinary
	// tenant. Same wording as unitRail's, so an operator who meets it at one
	// door recognises it at the other.
	if *unit != "" && *backing == registry.BackingUnit {
		log.Fatalf("add %s: --backing unit with --unit %s: a unit-backed route's process is the carlos-app@%s unit the platform derives and already restarts, "+
			"so a second unit name here would be two answers to one question — drop one of the two flags", host, *unit, host)
	}
	if *socket == "" && *addr == "" && *kind == "instance" {
		if *backing == registry.BackingUnit {
			// The carlos-app@<host>.socket unit's path — enable that unit
			// alongside this route (or use --addr for a TCP-only app). Not a
			// provision.DefaultSocket concern: that helper names the
			// exec-child convention, and a unit tenant's socket directory is
			// a different (systemd-owned) tree entirely.
			*socket = filepath.Join("/run/carlos/apps", host+".sock")
		} else {
			*socket = provision.DefaultSocket(host)
		}
	}
	// No DB default for unit tenants: their data lives in the unit's
	// StateDirectory (/var/lib/carlos-app/<host>), under a filename the APP
	// chooses (no platform-wide convention the way dataDir/<host>.db is for
	// exec children) — so there is no honest path to default to. --db stays
	// an explicit operator choice there, and ops.FleetDBs (the box-wide
	// litestream unit's config generator) only learns a unit tenant's
	// database path from this flag; leave it unset and that tenant's data
	// is replicated by nobody.
	if *dbPath == "" && *kind == "instance" && *backing != registry.BackingUnit {
		*dbPath = provision.DefaultDBPath(dataDir(), host)
	}
	// On a box whose layout intent is nested, a row this verb would make
	// activator-managed gets the nested defaults (design §8.6 step 5,
	// plan 5): it starts out isolated instead of running flat for a pass
	// and being moved. Explicit --socket or --db is the operator's own
	// choice and is left alone (the pass reports it foreign).
	if *kind == "instance" && *backing == "" && *hibernate && *addr == "" &&
		*socket == provision.DefaultSocket(host) && *dbPath == provision.DefaultDBPath(dataDir(), host) {
		store, _ := deploymentReadSeam()
		var evidence []registry.Route
		if r, err := registry.Open(*reg); err == nil {
			evidence, _ = r.List()
			r.Close()
		}
		nested, err := addMintsNested(context.Background(), store, boxLabelForAdd, evidence, dataDir())
		if err != nil {
			log.Fatalf("add %s: %v", host, err)
		}
		if nested {
			*socket, *dbPath = provision.NestedSocket(host), provision.NestedDBPath(dataDir(), host)
		}
	}
	// A hibernating unit tenant stops and starts with systemd, not the
	// agent — box-wide litestream (root) is the ONLY thing that can
	// replicate it, so skipping --db here is a genuine data-loss risk
	// (not just a missed nicety) the moment the unit is ever stopped for
	// idleness. Require it. A non-hibernating unit tenant is always running
	// and so has no equivalent forced moment of data being unreachable if
	// unreplicated — --db is still strongly recommended there but stays the
	// operator's call, unenforced.
	if *backing == registry.BackingUnit && *hibernate && *dbPath == "" {
		log.Fatalf("add %s: --backing unit --hibernate requires --db <path> — the box-wide litestream unit can't find a unit tenant's database on its own (see ops.FleetDBs)", host)
	}
	// An agent-exec child gets no STATE_DIRECTORY — the activation
	// contract's "relative --db resolves inside $STATE_DIRECTORY" convention
	// is systemd's half, real only under --backing unit. On the exec path the
	// activator consumes the stored path verbatim (Restore, Seed, Start's
	// --db flag, the replicator), so a relative path resolves against a cwd
	// nobody chose and every wake dies as an edge 503 discoverable only in
	// the journal (issue #70). Refuse it here, at the only door that can
	// introduce one — every other mint path builds absolute paths.
	if *backing != registry.BackingUnit && *dbPath != "" && !filepath.IsAbs(*dbPath) {
		log.Fatalf("add %s: --db %q is relative, and agent-exec children get no STATE_DIRECTORY to resolve it in — pass an absolute path (the default here is %s), or use --backing unit, whose StateDirectory= makes relative paths real",
			host, *dbPath, provision.DefaultDBPath(dataDir(), host))
	}
	account := resolveAccount(*reg, *acct, *acctID)
	// A `daemon` route may not hibernate: the capability means the platform
	// owns the process (enable, start, restart-on-adopt), and hibernation is
	// the activator stopping the very process that declaration exists to keep
	// up. Refused HERE when the comparison resolves — but, unlike `carlos
	// route --hibernate`, an UNRESOLVABLE comparison proceeds silently: this
	// verb runs at provision time on a box with no release pubkey, no
	// deployment store, and often no release shipped yet, and refusing there
	// would break provisioning. The authoritative refusal is the converge
	// pass, which sees the manifest every hour on a box that has one.
	if *hibernate {
		store, pub := deploymentReadSeam()
		daemon, note, proven := routeDaemonCheck(context.Background(), store, pub, account,
			registry.Route{Host: host, App: *app, Channel: *channel})
		if proven && daemon {
			log.Fatalf("add %s: %s (%s)", host, daemonHibernateRefusal, note)
		}
	}
	r, err := registry.Open(*reg)
	if err != nil {
		log.Fatal(err)
	}
	// The account is stamped on the row: it is what adopt and the edge build
	// this app's bucket keys from. Reads still fall back to the legacy
	// unqualified keys, so an unmigrated app added here keeps working.
	//
	// An upsert over an existing host is a rewrite, not a creation, and the
	// verb the operator reads back must say which one happened (issue #69):
	// "added" printed over a rewrite is how vitogo's operator was left
	// believing a fresh row existed while a running edge kept the old one.
	existing, err := r.Get(host)
	if err != nil {
		log.Fatal(err)
	}
	if err := refuseAddNested(host, existing, dataDir()); err != nil {
		log.Fatalf("%v", err)
	}
	route := registry.Route{
		Host: host, Slug: *slug, Kind: *kind, Socket: *socket, Addr: *addr,
		App: *app, Account: account,
		Channel: *channel, Hibernate: *hibernate, StatusNamed: *statusNamed, Backing: *backing,
		Unit: *unit, DBPath: *dbPath, Status: "ready",
	}
	if existing != nil {
		// Carried forward from Task 3's review (registry.go's Upsert doc names
		// this exact gap at cli.go:194-198): this verb has no flags for a
		// multi-upstream set or its probe, so it must not be the thing that
		// blanks one — Upsert's addrs/health columns are plain overwrites, and
		// a Route literal built above with no Addrs/Health always overwrites
		// with "". Preserving what this box-local verb doesn't manage mirrors
		// Upsert's own CASE-preserved columns (environment, caps, tunnel, ...):
		// an absent flag keeps what was already there.
		//
		// An explicit --addr is different, and deliberately NOT folded into
		// that same "flag wins" story when the row already carries an
		// established Addrs set (Task 8 review, IMPORTANT 1): silently
		// clearing Addrs here to let --addr win would build a row
		// ValidateShape forbids in every other door ("addr and addrs both
		// set"), and even if it didn't, AddrList and the edge already prefer
		// Addrs over Addr — the --addr would be dead on arrival, and the next
		// converge tick would revert it anyway (convergeOwned re-derives Addr
		// FROM the record's Addrs), the exact "success printed over nothing"
		// shape issue #69 named. The record — not this box verb — owns
		// multi-upstream shape, so the fix is a refusal that names the door
		// that does: `carlos instances set-upstreams`.
		if *addr != "" && existing.Addrs != "" {
			log.Fatalf("add %s: --addr %s would coexist with the established multi-upstream set (%s) on this row — "+
				"a route picks one upstream shape, and the record owns it, not this box verb. Repoint the set with "+
				"`carlos instances set-upstreams --app %s --host %s --addrs ...` instead of --addr here", host, *addr, existing.Addrs, *app, host)
		}
		route.Addrs, route.Health = existing.Addrs, existing.Health
		if *addr == "" {
			route.Addr = existing.Addr
		}
		// Steering (multi-region ingress design Part 4) carries no -steering
		// flag on this box verb either, and Upsert overwrites it exactly like
		// Addrs/Health (registry.go's Upsert doc comment) — so the same gap
		// applies: a bare re-add would silently clear a route's Route53
		// steering opt-in the moment the steering housekeeper duty (Task 6)
		// starts converging DNS from it. Preserve it the same way.
		route.Steering = existing.Steering
		// Replicas (#162) has the same shape and a different preserve rule:
		// the column CONVERGES from the record in both directions (the one
		// deliberate widening), so the next converge tick repairs a re-add's
		// blank anyway — but between this write and that tick the edges file
		// would publish the account without its replica placements, a window
		// of residence lying for seconds-to-minutes. Preserve it here so the
		// window does not exist. The replica MARK is deliberately NOT
		// preserved: it rides the record's Role, which this verb cannot
		// express, and a re-add minting an unmarked row over a replica's is
		// a shape change the next converge repairs — the mark is a routing
		// annotation, not a loss of state.
		route.Replicas = existing.Replicas
	}
	if err := r.Upsert(route); err != nil {
		log.Fatal(err)
	}
	if existing != nil {
		fmt.Printf("%s %s (existing route)\n", colorFor(os.Stdout).ok("updated"), host)
	} else {
		fmt.Printf("%s %s\n", colorFor(os.Stdout).ok("added"), host)
	}
}

func runRemove(args []string) {
	fs := flag.NewFlagSet("remove", flag.ExitOnError)
	reg := fs.String("registry", defaultRegistry(), "registry database")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	if fs.NArg() != 1 {
		log.Fatal("usage: carlos remove <host>")
	}
	r, err := registry.Open(*reg)
	if err != nil {
		log.Fatal(err)
	}
	// Data-safe by contract: stops routing the host, never touches its DB.
	if err := r.Delete(fs.Arg(0)); err != nil {
		log.Fatal(err)
	}
	fmt.Printf("%s %s\n", colorFor(os.Stdout).ok("removed"), fs.Arg(0))
}

func runInstances(args []string) {
	// The subcommands are positional words, not flags, and must never reach
	// fs.Parse: enable is the app's provisioning opt-in (instances_cli.go),
	// create/list/delete are the console-API record fronts (issue #66).
	// Every other invocation is the bare listing form below — THIS box's
	// registry routes, operator-side — unchanged.
	if len(args) > 0 {
		switch args[0] {
		case "enable":
			runInstancesEnable(args[1:])
			return
		case "create":
			runInstancesCreate(args[1:])
			return
		case "import":
			runInstancesImport(args[1:])
			return
		case "list":
			runInstancesList(args[1:])
			return
		case "delete":
			runInstancesDelete(args[1:])
			return
		case "door-key":
			runInstancesDoorKey(args[1:])
			return
		case "set":
			runInstancesSet(args[1:])
			return
		case "update":
			runInstancesUpdate(args[1:])
			return
		case "set-upstreams":
			runInstancesSetUpstreams(args[1:])
			return
		case "set-channel":
			runInstancesSetChannel(args[1:])
			return
		case "set-environment":
			runInstancesSetEnvironment(args[1:])
			return
		case "retry":
			runInstancesRetry(args[1:])
			return
		}
	}
	fs := flag.NewFlagSet("instances", flag.ExitOnError)
	reg := fs.String("registry", defaultRegistry(), "registry database")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	r, err := registry.Open(*reg)
	if err != nil {
		log.Fatal(err)
	}
	routes, err := r.List()
	if err != nil {
		log.Fatal(err)
	}
	w := newTable(os.Stdout)
	// BACKING and UNIT are the two halves of one answer — who owns this
	// route's process — so they print side by side: a route with a UNIT is
	// cycled by that unit on adoption, whatever BACKING says (and BACKING is
	// empty for every such route, which is the shape that made issue #110
	// invisible from this listing).
	//
	// ENV is the named config bundle the route is bound to (issue #64). It
	// decides which secrets this route's process is delivered, and reading
	// it used to mean sqlite3 against the registry — the one step of
	// titogo's secrets procedure that still needed box access. It sits
	// beside CHANNEL because the two are the same kind of fact about a
	// route and are set by the same verb (`carlos route`). Empty means the
	// app's default bundle, which is a real answer, so it prints blank
	// rather than being singled out. Off-box, `carlos routes` answers the
	// same question through the console.
	//
	// REPL is the rung of the replication ladder this route's database is
	// replicated on (discussion #10). It prints beside HIB because the two
	// are one fact: the class is read only where the activator replicates,
	// which is a hibernating route. An empty cell means the pre-ladder
	// default — change-driven ~1s sync — and until `carlos route
	// --replication` existed it was the honest answer for every route on
	// every box, which is precisely why the column had to become visible at
	// the same moment it became settable.
	fmt.Fprintln(w, "HOST\tAPP\tKIND\tCHANNEL\tENV\tSTATUS\tVERSION\tHIB\tREPL\tBACKING\tUNIT")
	for _, t := range routes {
		hib := ""
		if t.Hibernate {
			hib = "yes"
		}
		fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n",
			t.Host, t.App, t.Kind, t.Channel, t.Environment, t.Status, t.Version, hib, t.Replication, t.Backing, t.Unit)
	}
	w.Flush()
}

// runRoute changes which release channel and/or which named config
// environment a route follows — the verb that exists so "move keymail.dev
// off canary/keymail" or "bind demo.go.tito.io to the demo environment" is a
// command instead of raw sqlite against a production registry. The two axes
// are independent (2026-08-02-pipelines-design.md §2): --channel says how
// mature a build is, --environment says what glue it's wired to
// (internal/appconfig's config.json default bundle, optionally overlaid by
// a named apps/<acct>/<app>/environments/<name>.json bundle — see
// appconfig.LoadMerged). At least one of the two flags is required; either
// may be given alone, or both together.
//
// The channel safety rail is the whole point of that half: adoption reads a
// route's channel and converges it onto whatever version that channel
// currently points at, so repointing a route to a channel holding a
// DIFFERENT version than the route is actually serving changes what
// production serves on the next adopt/recheck pass. Before writing
// anything, this resolves what the TARGET channel's signed pointer says
// (same env/store plumbing as runAccountsMigrate/verifyMigratedPointers:
// CARLOS_RELEASE_PUBKEY, the deployment store) and compares it — see
// routeVersionCheck — against what the route is actually serving. Two
// outcomes can be written: the versions agree (proceed unconditionally), or
// they genuinely disagree and the operator passed --allow-version-change
// (the one thing that flag means). Anything that leaves the comparison
// UNPROVEN — no pubkey, an unresolvable channel, a bad signature — is
// refused with NO override: --allow-version-change means "I know it's
// different, do it anyway," never "skip finding out."
//
// The environment rail is simpler and has no override: --environment names a
// bundle that must already exist in the bucket for the route's app+account
// (checked via appconfig.ListEnvironments, the same Store the channel rail
// uses — unavailable store fails closed exactly like the channel rail's
// missing-pubkey/missing-store cases), or the write is refused, naming
// `carlos env set --app <a> --environment <name> ...` as the way to create
// it first. "--environment -" (a literal single dash) clears the binding
// back to the app's default bundle and needs no bucket check — there is
// nothing to prove exists.
//
// A third, independent axis, --tunnel, names the reverse channel a route is
// bound to at the edge (registry.Route.Tunnel — backing for later work in
// the remote-host-agent feature, issue #53). It is simpler than --environment
// in one way and matches it in another: shape is checked the same way
// (registry.ValidTunnelName, same [a-z0-9-]{1,32} pattern as
// ValidEnvironmentName), but there is no bucket to check EXISTENCE against
// (a tunnel only becomes real once a host-agent dials in under that name) —
// and it is --host only, since there is no bulk sibling to SetTunnel, so --app
// + --tunnel is refused outright rather than silently doing nothing. Same
// "--tunnel -" clear sentinel and the same platformCanaryOwnedRow refusal as
// --channel/--environment, for the same structural reason. It writes through
// its own SetTunnel call, independent of the --channel/--environment switch
// below — with only ever one candidate, there is no combined transaction for
// it to join.
//
// When BOTH --channel and --environment are given, both rails must resolve cleanly before EITHER
// axis writes: a single combined transaction (SetChannelsAndEnvironments)
// commits both columns for every candidate together, so a combined move can
// never leave a route repointed on one axis but not the other.
//
// --app's bulk form is all-or-nothing (2026-08-02 review, CRITICAL 1): every
// matched route's rail(s) are resolved BEFORE any write, and a single
// refusal — on EITHER axis — aborts the whole batch with every failure
// printed — never a half-moved app. It also excludes Platform routes with a
// printed reason, since a bulk sweep touching every tenant's minted route in
// one command is a different, riskier operation than "move this app's
// ordinary routes" — Platform routes stay opt-in, one --host at a time.
//
// A Platform+Canary route's Channel is a harder boundary: the domains
// reconciler's canary-minting pass (internal/domains/reconcile.go) is
// idempotent via `followed`, a check computed from THAT ROUTE'S OWN Channel
// column — not from the `byHost` map the same loop also sets but never
// reads for that purpose (a claim this comment got wrong in an earlier
// revision; re-review NEW 1 caught it). So ANY change to a Platform+Canary
// route's channel — not only a move off canary/*, moving BETWEEN two
// canary/* channels is just as reachable — makes `followed` false for the
// channel it left, and the next reconcile pass re-Upserts the same host
// from `firstReadyUpstream`, silently reverting this verb's write and
// blanking version/status/dns/cert/ENVIRONMENT columns to mintedRoute's
// zero values (mintedRoute never sets Environment — see its doc comment).
// platformCanaryOwnedRow refuses this unconditionally, structurally, for
// every candidate route in pass 1 below — not only in the --host branch —
// so it holds even if a future change lets --app include Platform routes.
// The SAME function guards the --environment axis (2026-08-02 review,
// IMPORTANT 4): the re-mint above rebuilds the whole ROW from mintedRoute,
// not just the channel it derives, so an environment binding on a
// Platform+Canary route is a binding on a row this verb does not own.
// registry.Upsert now preserves a non-empty Environment, which makes such a
// binding survivable rather than silently reverted — but "survives the
// reconciler" is not the same as "is owned by the operator", and an earlier
// revision of this comment argued from "reconcile does not DERIVE
// Environment", which was true and not the point. A plain Platform apex
// (non-Canary) is still bindable via --host on either axis.
// (A static, non-Canary Platform apex has no such trap: its idempotency
// guard, `hasApex`, never reads Channel at all, so --host may freely repoint
// one — though doing so while it happens to sit on a canary/* channel does
// make that canary's own `followed` check false, which mints a fresh canary
// host next pass. Additive, not reverted, but worth knowing.)
//
// This verb also carries the four SINGLE-ROUTE axes, each of which stands
// alone and short-circuits everything above (see runRouteGrant,
// runRouteHibernate, runRouteBacking and runRouteUnit):
//
//	carlos route --host keymail.dev --grant port:25
//	carlos route --host platform.carlosframework.com --grant host-role
//	carlos route --host gotenberg.docs.example.com --grant doc-engine
//	carlos route --host keymail.dev --revoke port:25
//	carlos route --host keymail.dev --hibernate=false
//	carlos route --host console.carloku.com --unit console.service
//
// --unit is how an EXISTING route — the production consoles, addr-backed under
// a bespoke console.service — tells the adopt pass which unit to cycle when
// its version moves. Without it adoption stamped Version (and therefore the
// edge's X-Carlos-Version header) over a process nothing had restarted, and
// the deploy needed a hand-run `systemctl restart console.service` to become
// true (issue #110). Like the other three it is --host only.
//
// --grant/--revoke are the operator's door to the two operator-consented
// capabilities (spec §2 "Grants"): a manifest DECLARES `ports`/`host_role`,
// and nothing materializes until an operator has granted it on THIS box.
// They require --host and never accept --app — grants are per route per box by
// design, and a bulk grant is precisely the operation nobody should be able
// to type by accident. --hibernate is here because a `daemon` route may not
// hibernate, and that refusal belongs at the same door as the flag.
func runRoute(args []string) {
	fs := flag.NewFlagSet("route", flag.ExitOnError)
	regPath := fs.String("registry", defaultRegistry(), "registry database")
	host := fs.String("host", "", "the one route to repoint")
	app := appFlag(fs, "repoint every non-platform route of this app")
	acctFilter := fs.String("account", "ops", `account (sqid or name) to filter --app routes by; "all" for every account`)
	acctIDFilter := fs.String("account-id", "", "account sqid to filter --app routes by, matched as a sqid ONLY — supersedes --account")
	channel := channelFlag(fs, "release channel to subscribe the route(s) to")
	environment := environmentFlag(fs,
		`named config environment to bind the route(s) to (must already exist in the bucket); "-" clears to default`)
	grant := fs.String("grant", "", "capability token to grant this route ON THIS BOX: port:<n> (privileged, never 22/80/443) | host-role | doc-engine")
	revoke := fs.String("revoke", "", "capability token to revoke from this route on this box")
	hibernate := fs.Bool("hibernate", false, "hibernate this route when idle; --hibernate=false stops it hibernating")
	backing := fs.String("backing", "", `who owns this route's process: "unit" (a carlos-app@<host> systemd unit) | "" (an exec child of the agent)`)
	unit := fs.String("unit", "",
		`systemd unit that owns this route's process (e.g. console.service), so adoption can cycle it; "-" clears the stamp (host only, no --app)`)
	tunnel := fs.String("tunnel", "",
		`name of the reverse channel backing this route at the edge; "-" clears to a Socket/Addr-routed upstream (host only, no --app)`)
	replication := fs.String("replication", "",
		`replication ladder rung for this route's database: `+strings.Join(registry.ReplicationClasses(), " | ")+`; "-" clears to the default (host only, no --app)`)
	allowVersionChange := fs.Bool("allow-version-change", false,
		"allow repointing to a channel that RESOLVES but holds a different version than the route currently serves "+
			"(never overrides an unresolvable target, a bad signature, or a missing pubkey)")
	dryRun := fs.Bool("dry-run", false, "run the full safety-rail check and print the plan, but write nothing")
	const usage = "usage: carlos route --host <host> [--channel <channel>] [--environment <name>] [--tunnel <name>] [--registry <db>] [--allow-version-change] [--dry-run]\n" +
		"   or: carlos route --app <app> [--channel <channel>] [--environment <name>] [--account <sqid|name|all>] [--account-id <sqid>] [--registry <db>] [--allow-version-change] [--dry-run]\n" +
		"       (--account-id names ONE account by sqid and supersedes --account, \"all\" included; --tunnel is --host only)\n" +
		"   or: carlos route --host <host> (--grant | --revoke) <port:n|host-role|doc-engine> [--registry <db>] [--dry-run]\n" +
		"   or: carlos route --host <host> --hibernate[=false] [--registry <db>] [--dry-run]\n" +
		"   or: carlos route --host <host> --backing <unit|\"\"> [--registry <db>] [--dry-run]\n" +
		"   or: carlos route --host <host> --unit <name.service|-> [--registry <db>] [--dry-run]\n" +
		`   or: carlos route --host <host> --replication <stream|batched|daily|-> [--registry <db>] [--dry-run]`
	// -h names the remote verbs, because this one only ever finds the box
	// it runs on: someone holding just this verb could not tell that a
	// signed-in CLI changes a route's channel or environment from anywhere
	// (#264).
	fs.Usage = flagUsage(fs, usage+"\n\n"+
		"This edits the routing table on the box it runs on, so it has to run there. To change a route's channel or environment from your own machine, sign in and use carlos instances set-channel or carlos instances set-environment.")
	fs.Parse(args)
	// --hibernate has to be TRI-state: a plain flag.Bool cannot tell "the
	// operator did not mention hibernation" from "the operator asked for
	// --hibernate=false", and this verb must never flip a column nobody named.
	// fs.Visit reports only the flags actually present on the command line,
	// which keeps the operator's syntax the natural `--hibernate` /
	// `--hibernate=false` rather than a quoted tri-state string.
	// --backing is tri-state for the same reason, and more sharply: "" is a
	// LEGAL VALUE of that column (an exec child), so a flag left unmentioned
	// and a flag set to "" cannot be told apart by the value alone. --unit is
	// tri-state on the same grounds — "" is that column's default, meaning
	// "no unit owns this route's process" — though its CLEAR spelling is
	// "--unit -", the sentinel --environment and --tunnel already use, so that a
	// bare `--unit ""` reads as the mistake it almost certainly is rather than
	// as a silent unstamp.
	// --replication is tri-state on --unit's exact grounds: "" is this
	// column's default (the pre-ladder every-second sync), so a flag left
	// unmentioned and a flag set to "" cannot be told apart by value, and its
	// CLEAR spelling is the same "-" sentinel.
	hibernateSet, backingSet, unitSet, replicationSet := false, false, false, false
	fs.Visit(func(f *flag.Flag) {
		switch f.Name {
		case "hibernate":
			hibernateSet = true
		case "backing":
			backingSet = true
		case "unit":
			unitSet = true
		case "replication":
			replicationSet = true
		}
	})

	if *host == "" && *app == "" {
		log.Fatal(usage)
	}
	if *host != "" && *app != "" {
		log.Fatal("route: --host and --app are mutually exclusive — name the one route, or the app whose routes to move\n\n" + usage)
	}
	// The three single-route axes. Each is ITS OWN operation: a grant change
	// is not a repoint, and combining two would put two unrelated writes
	// behind two unrelated rails in one command. --channel and --environment
	// keep their existing combinability (one transaction, both rails resolved
	// first); everything below stands alone.
	if *grant != "" && *revoke != "" {
		log.Fatal("route: --grant and --revoke are mutually exclusive — a grant and a revoke of the same route are two operations, and typing both leaves it ambiguous which one you meant to happen")
	}
	grantAxis := *grant != "" || *revoke != ""
	// --backing is a single-route axis exactly like the other two, and it is
	// held to the same exclusions (2026-08-03 review, I2). It was not, and
	// each gap was a SILENT PARTIAL: the switch below runs grantAxis first,
	// then hibernateSet, then --backing, so `--hibernate=false --backing unit`
	// wrote the hibernate column and dropped the backing stamp on the floor
	// without a word, and `--backing unit --channel canary/x` stamped the row
	// and never repointed it. An operator who typed both wanted both; a
	// command that does one of them and reports success is worse than one
	// that refuses.
	if grantAxis && hibernateSet {
		log.Fatal("route: --grant/--revoke and --hibernate are mutually exclusive — run them as two commands, so each meets its own rail alone")
	}
	if backingSet && (grantAxis || hibernateSet) {
		log.Fatal("route: --backing is mutually exclusive with --grant/--revoke and --hibernate — run them as two commands, so each meets its own rail alone. " +
			"The order is not arbitrary either: --backing unit comes FIRST, because a grant onto a route that is not unit-backed is refused (it would materialize nothing)")
	}
	// --replication is held to the same exclusions as the other single-route
	// axes, and for the reason recorded above --backing: the switch below
	// runs one arm, so a command naming two axes would write one of them and
	// drop the other on the floor without a word.
	if replicationSet && (grantAxis || hibernateSet || backingSet || unitSet) {
		log.Fatal("route: --replication is mutually exclusive with --grant/--revoke, --hibernate, --backing and --unit — run them as two commands, so each meets its own rail alone. " +
			"--hibernate in particular decides whether this route's class is read AT ALL (the activator replicates hibernating routes; /etc/litestream.yml covers the rest), so the two in one command would leave the answer depending on which arm ran")
	}
	if unitSet && (grantAxis || hibernateSet || backingSet) {
		log.Fatal("route: --unit is mutually exclusive with --grant/--revoke, --hibernate and --backing — run them as two commands, so each meets its own rail alone. " +
			"--unit and --backing in particular are two different answers to the same question (who owns this route's process), and a command that wrote both would leave the row saying both")
	}
	if (grantAxis || hibernateSet || backingSet || unitSet || replicationSet) && (*channel != "" || *environment != "" || *tunnel != "") {
		log.Fatal("route: --grant/--revoke/--hibernate/--backing/--unit/--replication cannot be combined with --channel, --environment or --tunnel — a capability grant and a repoint are different operations with different rails, and one command that did both would commit them together")
	}
	if (grantAxis || hibernateSet || backingSet || unitSet || replicationSet) && *host == "" {
		log.Fatal("route: --grant/--revoke/--hibernate/--backing/--unit/--replication require --host, never --app: grants, process backing and the owning unit are per route per box by design, and a bulk grant is precisely the operation nobody should be able to type by accident\n\n" + usage)
	}
	// The token is checked BEFORE the registry is opened for writing: a typo
	// should cost nothing and reach nothing.
	if grantAxis {
		tok := *grant
		if tok == "" {
			tok = *revoke
		}
		if err := caps.ValidGrantToken(tok); err != nil {
			log.Fatalf("route: %v", err)
		}
	}
	// And so is --backing's value, for the same reason and at the same moment
	// (2026-08-03 review, I1): runRouteBacking's doc comment already claimed
	// this check ("the value is \"\" or BackingUnit, checked at the flag
	// before the registry is even opened") while nothing performed it, and
	// registry.SetBacking writes the column verbatim. `--backing unti` was
	// therefore accepted, wrote "unti", and left the route neither unit-backed
	// (release.Adoptable compares against BackingUnit exactly) nor visibly
	// wrong — the migration's one write, silently a no-op. Same wording as
	// runAdd's check, which is the other door onto this column.
	if backingSet && *backing != "" && *backing != registry.BackingUnit {
		log.Fatalf("route: backing %q: want %q (a carlos-app@<host> systemd unit) or \"\" (an exec child of the agent)", *backing, registry.BackingUnit)
	}
	// --unit's value gets the same treatment at the same moment, and for the
	// reason the column exists: an unvalidated name reaches the row verbatim
	// and is discovered months later as an adopt pass that restarted nothing.
	// "-" is the clear sentinel (checked first, exactly as envClear precedes
	// ValidEnvironmentName), and it resolves to "" — the column's own default,
	// which ValidUnitName accepts. A bare `--unit ""` is refused rather than
	// treated as a clear: the doors that CLEAR say so out loud.
	unitClear := *unit == "-"
	resolvedUnit := *unit
	if unitClear {
		resolvedUnit = ""
	}
	if unitSet && !unitClear {
		if *unit == "" {
			log.Fatal(`route: --unit "" is refused — use "--unit -" to clear the stamp, or name the unit that owns this route's process (e.g. --unit console.service)`)
		}
		if !registry.ValidUnitName(*unit) {
			log.Fatalf("route: unit %q: want a systemd service unit name (e.g. console.service), or \"-\" to clear the stamp", *unit)
		}
	}
	// --replication's value is checked at the flag, before the registry is
	// opened, on --unit's exact grounds: an unknown class reaches
	// replicationSyncInterval, which answers anything it does not recognise
	// with litestream's default — so `--replication dayly` would be written,
	// reported as done, and discovered as a route that quietly never left the
	// most expensive rung. "-" is the clear sentinel and resolves to "", the
	// column's own default; a bare `--replication ""` is refused, because the
	// doors that CLEAR say so out loud (see --unit).
	replicationClear := *replication == "-"
	resolvedReplication := *replication
	if replicationClear {
		resolvedReplication = ""
	}
	if replicationSet && !replicationClear {
		if *replication == "" {
			log.Fatalf(`route: --replication "" is refused — use "--replication -" to clear the rung, or name one: %s`, strings.Join(registry.ReplicationClasses(), ", "))
		}
		if !registry.ValidReplication(*replication) {
			log.Fatalf("route: replication %q: want one of %s, or \"-\" to clear to the default", *replication, strings.Join(registry.ReplicationClasses(), ", "))
		}
	}
	if *tunnel != "" && *app != "" {
		log.Fatal("route: --tunnel is only supported with --host — bulk --app tunnel repointing is not implemented\n\n" + usage)
	}
	if *channel == "" && *environment == "" && *tunnel == "" && !grantAxis && !hibernateSet && !backingSet && !unitSet && !replicationSet {
		log.Fatal(`route: at least one of --channel, --environment, --tunnel, --grant, --revoke, --hibernate, --backing, --unit or --replication is required. "" only means stable-by-default at route CREATION (carlos add) — ` +
			"this verb sets an explicit binding, and needs an explicit target to set it to.")
	}
	if *channel != "" && !release.ValidChannel(*channel) {
		log.Fatalf("route: %q is not a valid channel", *channel)
	}
	// "-" is the CLI-level "clear to default" sentinel, checked before
	// ValidEnvironmentName (which would otherwise accept it — a bare hyphen
	// fits [a-z0-9-]{1,32} — and treat it as a real bundle name). A real
	// environment named "-" is unreachable through this verb as a result;
	// that's an accepted asymmetry, not a bug (matches "-" as a stdin/stdout
	// sentinel elsewhere in Unix tooling).
	envClear := *environment == "-"
	// "default" gets its own refusal rather than falling through to the
	// generic one (2026-08-02 review, MINOR 7): it is a well-formed name that
	// is reserved, not a malformed one, and the generic message sent the
	// operator looking for a typo. It also keeps this refusal pinned to the
	// RESERVATION — with the reservation gone, the existence rail below would
	// still refuse "default" for an unrelated reason and a test asserting
	// only "refused" would not notice. Same wording as checkEnvironmentFlag's
	// (appconfig_cli.go), which every `carlos env`/`secrets` verb uses.
	if *environment == "default" {
		log.Fatal(`route: --environment default is refused: "default" already names the base bundle ` +
			"(config.json) every route gets when nothing is bound — use \"--environment -\" to clear a binding back to it")
	}
	if *environment != "" && !envClear && !appconfig.ValidEnvironmentName(*environment) {
		log.Fatalf("route: %q is not a valid environment name — want [a-z0-9-]{1,32}", *environment)
	}
	// resolvedEnv is what actually lands in the registry column: "" either
	// because --environment was never given (untouched — see the write step)
	// or because it was explicitly cleared.
	resolvedEnv := *environment
	if envClear {
		resolvedEnv = ""
	}
	// tunnelClear is the same "-" clear sentinel as --environment's, checked
	// before ValidTunnelName for the same reason envClear precedes
	// ValidEnvironmentName above: a bare hyphen fits [a-z0-9-]{1,32} and
	// would otherwise be treated as a real tunnel name. Unlike --environment
	// there is no reserved name and no bucket to validate EXISTENCE against
	// here — a tunnel name only becomes real once the host-agent dials in
	// under it (later work in this feature) — but the shape itself is
	// checked, same as --environment's, since this name is later matched
	// against a dialed-in identity and may build keys.
	tunnelClear := *tunnel == "-"
	if *tunnel != "" && !tunnelClear && !registry.ValidTunnelName(*tunnel) {
		log.Fatalf("route: %q is not a valid tunnel name — want [a-z0-9-]{1,32}", *tunnel)
	}
	resolvedTunnel := *tunnel
	if tunnelClear {
		resolvedTunnel = ""
	}

	r, err := registry.Open(*regPath)
	if err != nil {
		log.Fatal(err)
	}
	defer r.Close()

	// The single-route axes short-circuit here, before the channel/environment
	// machinery below: they write one column of one row, through one narrow
	// setter, and share none of that batch's all-or-nothing plumbing.
	if grantAxis || hibernateSet || backingSet || unitSet || replicationSet {
		rt, err := r.Get(*host)
		if err != nil {
			log.Fatal(err)
		}
		if rt == nil {
			log.Fatalf("route: no route for host %q", *host)
		}
		if err := suffixRail(*rt); err != nil {
			log.Fatalf("route: %v", err)
		}
		switch {
		case grantAxis:
			runRouteGrant(r, *rt, *grant, *revoke, *dryRun)
		case hibernateSet:
			runRouteHibernate(r, *rt, *hibernate, *dryRun)
		case backingSet:
			runRouteBacking(r, *rt, *backing, *dryRun)
		case replicationSet:
			runRouteReplication(r, *rt, resolvedReplication, *dryRun)
		default:
			runRouteUnit(r, *rt, resolvedUnit, *dryRun)
		}
		return
	}

	var candidates []registry.Route
	if *host != "" {
		rt, err := r.Get(*host)
		if err != nil {
			log.Fatal(err)
		}
		if rt == nil {
			log.Fatalf("route: no route for host %q", *host)
		}
		if err := suffixRail(*rt); err != nil {
			log.Fatalf("route: %v", err)
		}
		candidates = []registry.Route{*rt}
	} else {
		routes, err := r.List()
		if err != nil {
			log.Fatal(err)
		}
		// --account-id supersedes --account here as everywhere, including
		// over the "all" escape hatch: someone who names one account by
		// its sqid means that account, not every account, whatever --account
		// happens to still hold from a script's defaults.
		var wantAcct string
		if strings.TrimSpace(*acctIDFilter) != "" {
			wantAcct = resolveAccount(*regPath, "", *acctIDFilter)
		} else if !strings.EqualFold(*acctFilter, "all") {
			wantAcct = resolveAccount(*regPath, *acctFilter, "")
		}
		for _, rt := range routes {
			if rt.App != *app {
				continue
			}
			if wantAcct != "" && release.RouteAccount(rt) != wantAcct {
				continue
			}
			if rt.Platform {
				// A bulk sweep changing every tenant's platform-minted route
				// in one command is a different, riskier operation than
				// "move this app's ordinary routes" — Platform routes stay
				// opt-in, one --host at a time. (Platform+Canary routes are
				// refused even via --host — see platformCanaryOwnedRow,
				// applied structurally below for every candidate, not just
				// here — so this exclusion is belt-and-braces for those, but
				// load-bearing for a plain Platform apex.)
				warnf(os.Stdout, "%s: skipped — platform-minted route; use --host %s to change it explicitly\n", rt.Host, rt.Host)
				continue
			}
			if err := suffixRail(rt); err != nil {
				// A real provisioned sidecar's App equals its owning app's, so
				// this is not hypothetical: an -app sweep legitimately meets
				// these rows. Skipped like a Platform route, not fatal — one
				// row under the reserved suffix must not abort the rest of
				// the batch.
				warnf(os.Stdout, "%s: skipped — %v; use -host %s to see the refusal explicitly\n", rt.Host, err, rt.Host)
				continue
			}
			candidates = append(candidates, rt)
		}
		if len(candidates) == 0 {
			log.Fatalf("route: no non-platform routes for app %q (with the given --account filter, if any)", *app)
		}
	}

	// Pass 1: resolve every candidate's rail(s) WITHOUT writing anything. A
	// check-then-write-per-route loop can leave an --app bulk half-repointed
	// if a later route fails after earlier ones already wrote (CRITICAL 1) —
	// so every route's verdict, on EVERY axis touched, is known before the
	// first write happens.
	//
	// The Platform+Canary refusal is decided HERE, for every candidate,
	// rather than only in the --host branch above — a shared function both
	// forms go through — so it holds structurally even though only --host can
	// reach a Platform+Canary route today (--app already excludes every
	// Platform route). See runRoute's doc comment for why this refusal has
	// no override and applies to ANY channel change, not only leaving
	// canary/* — and why it is Channel-specific, never blocking the
	// --environment axis.
	type verdictKind int
	const (
		vNeedsCheck    verdictKind = iota // routeVersionCheck resolves this below
		vNoop                             // already on the target channel
		vSafe                             // resolved: proceed unconditionally
		vVersionChange                    // resolved, differs: overridable
		vUnproven                         // resolved, but nothing proves what would change: never overridable
		vPlatformOwned                    // Platform+Canary: reconciler owns this column, never overridable
		vRecordOwned                      // Provisioned: the instance record owns this column, never overridable
	)
	type verdict struct {
		rt   registry.Route
		from string
		kind verdictKind
		note string
	}
	// chanVerdicts stays nil (len 0) when --channel was not given, so every
	// loop below that ranges over it is naturally a no-op — that IS the
	// "channel axis untouched" state, not a special case to guard against.
	var chanVerdicts []verdict
	if *channel != "" {
		chanVerdicts = make([]verdict, len(candidates))
		anyNeedsCheck := false
		for i, rt := range candidates {
			from := rt.Channel
			if from == "" {
				from = "stable"
			}
			switch {
			case from == *channel:
				chanVerdicts[i] = verdict{rt: rt, from: from, kind: vNoop}
			case platformCanaryOwnedRow(rt):
				chanVerdicts[i] = verdict{rt: rt, from: from, kind: vPlatformOwned, note: fmt.Sprintf(
					"Platform+Canary route on %s — its channel is derived and re-asserted by the domains\n"+
						"reconciler's canary-minting pass (internal/domains/reconcile.go), which is idempotent via a\n"+
						"check on THIS route's own Channel column; any change here is silently reverted on the next\n"+
						"reconcile pass. This verb does not own this column", from)}
			case recordOwnedRow(rt):
				chanVerdicts[i] = verdict{rt: rt, from: from, kind: vRecordOwned, note: fmt.Sprintf(
					"provisioned route on %s — its channel is owned by the instance record at\n"+
						"apps/%s/%s/instances/%s.json, which the provisioning reconciler converges onto this row\n"+
						"on every pass; a change here is reverted within seconds of the next serial bump.\n"+
						"Change it where it lives:\n"+
						"  carlos instances set-channel --app %s --host %s --channel %s",
					from, release.RouteAccount(rt), rt.App, rt.Host,
					rt.App, rt.Host, *channel)}
			default:
				chanVerdicts[i] = verdict{rt: rt, from: from, kind: vNeedsCheck}
				anyNeedsCheck = true
			}
		}
		if anyNeedsCheck {
			// Resolved once, the same way verifyMigratedPointers does:
			// byte/version comparison needs the public key, not the signing key
			// — this verb never writes a release object, only the registry row.
			// A run needing no check at all (every candidate a no-op or
			// Platform+Canary-refused) never reaches here, so it needs no
			// store/pubkey configured.
			var pub ed25519.PublicKey
			if b := strings.TrimSpace(os.Getenv("CARLOS_RELEASE_PUBKEY")); b != "" {
				if p, perr := release.ParsePublicKey(b, "CARLOS_RELEASE_PUBKEY"); perr == nil {
					pub = p
				}
			}
			var store release.Store
			if pub != nil {
				store = openStore()
			}
			ctx := context.Background()
			for i := range chanVerdicts {
				if chanVerdicts[i].kind != vNeedsCheck {
					continue
				}
				rt := chanVerdicts[i].rt
				note, safe, overridable := routeVersionCheck(ctx, store, pub, release.RouteAccount(rt), rt, chanVerdicts[i].from, *channel)
				chanVerdicts[i].note = note
				switch {
				case safe:
					chanVerdicts[i].kind = vSafe
				case overridable:
					chanVerdicts[i].kind = vVersionChange
				default:
					chanVerdicts[i].kind = vUnproven
				}
			}
		}
	}

	// The environment rail: an existence check against the bucket, via the
	// same Store the channel rail uses (unavailable store fails closed
	// exactly like the channel rail's missing-pubkey/missing-store cases —
	// openStore's own log.Fatal). No override exists for a missing bundle:
	// unlike a version mismatch, there is no "I know it's different, do it
	// anyway" reading of binding a route to config that isn't there.
	type envVerdictKind int
	const (
		eNoop          envVerdictKind = iota // already bound to the target
		eWillWrite                           // clearing, or the bundle exists — proceed
		eMissing                             // named but absent in the bucket — refuse, no override
		ePlatformOwned                       // Platform+Canary: the reconciler owns this row, never overridable
		eRecordOwned                         // Provisioned: the instance record owns this column
	)
	type envVerdict struct {
		rt   registry.Route
		from string
		kind envVerdictKind
		note string
	}
	var envVerdicts []envVerdict
	if *environment != "" {
		envVerdicts = make([]envVerdict, len(candidates))
		var store release.Store
		// existsByApp caches one ListEnvironments call per distinct
		// (account, app) among the candidates (an --app --account all sweep can
		// span several accounts), mirroring the channel rail's "resolve once"
		// pattern — never opened at all when every candidate is already bound
		// to the target or the operator is clearing (envClear needs no bucket
		// check per the doc comment above). The key is BOTH halves of the
		// query it caches: keyed on account alone (2026-08-02 review, MINOR
		// 6) it would approve a bundle that exists for a different app in the
		// same account. Unreachable today — --app fixes rt.App and --host has
		// one candidate — which is exactly why it is worth keying correctly
		// now rather than after a future widening.
		existsByApp := map[string]map[string]bool{}
		ctx := context.Background()
		for i, rt := range candidates {
			from := rt.Environment
			switch {
			case from == resolvedEnv:
				envVerdicts[i] = envVerdict{rt: rt, from: from, kind: eNoop}
			case platformCanaryOwnedRow(rt):
				// Refused on the same terms as the channel axis, and for the
				// same structural reason: the reconciler rebuilds this row.
				// Placed before envClear so CLEARING is refused too — a
				// half-owned column would be the worst of both.
				envVerdicts[i] = envVerdict{rt: rt, from: from, kind: ePlatformOwned, note: "Platform+Canary route — its whole row is rebuilt by the domains\n" +
					"reconciler's canary-minting pass (internal/domains/reconcile.go) whenever that pass's `followed`\n" +
					"check goes false, from values it derives itself. This verb does not own this row.\n" +
					"A binding already on such a route (made before this rail existed) survives re-mints and cannot be\n" +
					"cleared here either: `carlos remove " + rt.Host + "` and let the next reconcile pass re-mint it clean"}
			case recordOwnedRow(rt):
				envVerdicts[i] = envVerdict{rt: rt, from: from, kind: eRecordOwned, note: fmt.Sprintf(
					"provisioned route — its environment is owned by the instance record at\n"+
						"apps/%s/%s/instances/%s.json, which the provisioning reconciler converges onto this row on every\n"+
						"pass. A binding made here (or cleared here) is reverted within seconds of the next serial bump.",
					release.RouteAccount(rt), rt.App, rt.Host)}
			case envClear:
				envVerdicts[i] = envVerdict{rt: rt, from: from, kind: eWillWrite}
			default:
				acct := release.RouteAccount(rt)
				ck := acct + "/" + rt.App
				names, cached := existsByApp[ck]
				if !cached {
					if store == nil {
						store = openStore()
					}
					list, lerr := appconfig.ListEnvironments(ctx, store, acct, rt.App)
					if lerr != nil {
						log.Fatalf("route: list environments for %s/%s: %v", acct, rt.App, lerr)
					}
					names = make(map[string]bool, len(list))
					for _, n := range list {
						names[n] = true
					}
					existsByApp[ck] = names
				}
				if names[*environment] {
					envVerdicts[i] = envVerdict{rt: rt, from: from, kind: eWillWrite}
				} else {
					envVerdicts[i] = envVerdict{rt: rt, from: from, kind: eMissing, note: fmt.Sprintf(
						"environment %q does not exist for %s/%s — create it first with `carlos env set --app %s --environment %s ...`",
						*environment, acct, rt.App, rt.App, *environment)}
				}
			}
		}
	}

	// Pass 2: any refusal, on EITHER axis, aborts the WHOLE batch — print
	// every one, write nothing. This is what makes the combined
	// --channel/--environment invocation atomic: neither axis's write call
	// below is reached unless every candidate cleared both rails.
	var refusals []string
	var toWrite []string
	for _, v := range chanVerdicts {
		switch v.kind {
		case vNoop:
			// handled in the final print loop
		case vSafe:
			toWrite = append(toWrite, v.rt.Host)
		case vVersionChange:
			if *allowVersionChange {
				toWrite = append(toWrite, v.rt.Host)
			} else {
				refusals = append(refusals, fmt.Sprintf("%s: %s (use --allow-version-change to override)", v.rt.Host, v.note))
			}
		case vUnproven:
			refusals = append(refusals, fmt.Sprintf(
				"%s: %s — refusing regardless of --allow-version-change: the comparison itself is unproven", v.rt.Host, v.note))
		case vPlatformOwned, vRecordOwned:
			refusals = append(refusals, fmt.Sprintf(
				"%s: %s — refusing regardless of --allow-version-change: no flag overrides a column this verb does not own",
				v.rt.Host, v.note))
		}
	}
	var envToWrite []string
	for _, v := range envVerdicts {
		switch v.kind {
		case eNoop:
			// handled in the final print loop
		case eWillWrite:
			envToWrite = append(envToWrite, v.rt.Host)
		case eMissing:
			refusals = append(refusals, fmt.Sprintf("%s: %s", v.rt.Host, v.note))
		case ePlatformOwned, eRecordOwned:
			refusals = append(refusals, fmt.Sprintf(
				"%s: %s — no flag overrides a row this verb does not own", v.rt.Host, v.note))
		}
	}
	// The tunnel rail: --host only (the --app refusal above guarantees exactly
	// one candidate whenever --tunnel is set), and unlike --channel/--environment
	// there is no existence check — a tunnel only becomes real once the
	// host-agent dials in under that name, so there is no bucket store to
	// consult yet. Same Platform+Canary refusal as the other two axes, for
	// the same structural reason (platformCanaryOwnedRow's doc comment covers
	// all three columns it guards).
	var tunnelFrom string
	tunnelNoop, tunnelWillWrite := false, false
	if *tunnel != "" {
		rt := candidates[0]
		tunnelFrom = rt.Tunnel
		switch {
		case tunnelFrom == resolvedTunnel:
			tunnelNoop = true
		case platformCanaryOwnedRow(rt):
			refusals = append(refusals, fmt.Sprintf(
				"%s: Platform+Canary route — its whole row is rebuilt by the domains reconciler's canary-minting "+
					"pass (internal/domains/reconcile.go) — this verb does not own this row, no flag overrides that",
				rt.Host))
		default:
			tunnelWillWrite = true
		}
	}
	if len(refusals) > 0 {
		for _, msg := range refusals {
			fmt.Fprintln(os.Stderr, msg)
		}
		fmt.Fprintf(os.Stderr, "%s %d of %d route(s) refused — nothing written\n",
			colorFor(os.Stderr).bad("route:"), len(refusals), len(candidates))
		os.Exit(1)
	}

	switch {
	case *channel != "" && *environment != "":
		// Combined: one transaction, both columns, for every candidate that
		// cleared both rails above — including per-axis no-ops, which is
		// harmless (an UPDATE to the value already there) and keeps this
		// branch from having to reconcile two possibly-different host lists
		// into one atomic write.
		if !*dryRun {
			hosts := make([]string, len(candidates))
			for i, rt := range candidates {
				hosts[i] = rt.Host
			}
			if err := r.SetChannelsAndEnvironments(hosts, *channel, resolvedEnv); err != nil {
				log.Fatalf("route: write: %v — nothing written; the batch is one transaction", err)
			}
		}
	case *channel != "":
		if len(toWrite) > 0 && !*dryRun {
			if err := r.SetChannels(toWrite, *channel); err != nil {
				log.Fatalf("route: write: %v — nothing written; the batch is one transaction", err)
			}
		}
	case *environment != "":
		if len(envToWrite) > 0 && !*dryRun {
			if err := r.SetEnvironments(envToWrite, resolvedEnv); err != nil {
				log.Fatalf("route: write: %v — nothing written; the batch is one transaction", err)
			}
		}
	}
	// The tunnel axis writes independently of the channel/environment switch
	// above — it has no bulk sibling to combine into their transactions (it
	// is always exactly one candidate; see the --app refusal above), so there
	// is nothing for it to be atomic WITH.
	if tunnelWillWrite && !*dryRun {
		if err := r.SetTunnel(candidates[0].Host, resolvedTunnel); err != nil {
			log.Fatalf("route: write: %v", err)
		}
	}

	prefix := ""
	if *dryRun {
		prefix = colorFor(os.Stdout).warn("(dry run)") + " "
	}
	envLabel := func(e string) string {
		if e == "" {
			return "default"
		}
		return e
	}
	tunnelLabel := func(t string) string {
		if t == "" {
			return "none"
		}
		return t
	}
	for i := range candidates {
		if chanVerdicts != nil {
			v := chanVerdicts[i]
			if v.kind == vNoop {
				mutedf(os.Stdout, "%s: already following %s — nothing to do\n", v.rt.Host, *channel)
			} else {
				fmt.Printf("%s%s: %s -> %s (%s)\n", prefix, v.rt.Host, v.from, *channel, v.note)
			}
		}
		if envVerdicts != nil {
			v := envVerdicts[i]
			if v.kind == eNoop {
				mutedf(os.Stdout, "%s: environment already %s — nothing to do\n", v.rt.Host, envLabel(resolvedEnv))
			} else {
				fmt.Printf("%s%s: environment %s -> %s\n", prefix, v.rt.Host, envLabel(v.from), envLabel(resolvedEnv))
			}
		}
		if *tunnel != "" && i == 0 {
			if tunnelNoop {
				mutedf(os.Stdout, "%s: tunnel already %s — nothing to do\n", candidates[0].Host, tunnelLabel(resolvedTunnel))
			} else {
				fmt.Printf("%s%s: tunnel %s -> %s\n", prefix, candidates[0].Host, tunnelLabel(tunnelFrom), tunnelLabel(resolvedTunnel))
			}
		}
	}
}

// platformCanaryOwnedRow reports whether the domains reconciler, not this
// verb, owns rt's row: true for any Platform+Canary route. See runRoute's doc
// comment for why — the canary-minting pass in internal/domains/reconcile.go
// is idempotent via `followed`, a check computed from THAT ROUTE'S OWN
// Channel column, not from the `byHost` map the same loop also sets but never
// reads for that purpose. Any change to the channel (not only a move off
// canary/*) makes `followed` false, and the next pass re-Upserts the whole
// row from mintedRoute's derived values.
//
// One function, both axes: --channel because the reconciler re-derives that
// column, --environment because it rebuilds the row the column lives on
// (2026-08-02 review, IMPORTANT 4).
func platformCanaryOwnedRow(rt registry.Route) bool {
	return rt.Platform && rt.Canary
}

// recordOwnedRow reports whether the instance-provisioning reconciler, not
// this verb, owns this route's Channel and Environment columns.
//
// Provisioned is set only by that pass — on mint, and on adoption after it
// matched the row's shape against the bucket record (internal/provision).
// The pass converges both columns from the record on every tick, so a write
// here lands and is reverted within seconds of the next serial bump, with
// nothing said anywhere. Same structural refusal as platformCanaryOwnedRow,
// one more owner: the door is the record.
func recordOwnedRow(rt registry.Route) bool { return rt.Provisioned }

// runRouteGrant is the operator's door to a transfer of box authority: one
// token, onto or off one route, on this box. Everything it can refuse it
// refuses before writing, and the write itself is registry.SetCaps — the
// only writer of that column in the codebase.
//
// A revoke is not the mirror image of a grant in one respect: it needs no
// conflict check, because removing a grant can never collide with anything.
// Everything else — the row-shape rails, the hand-edited-column refusal — is
// shared, since a route whose grants this verb refuses to widen is not a
// route whose grants it should be quietly rewriting either.
func runRouteGrant(r *registry.Registry, rt registry.Route, grantTok, revokeTok string, dryRun bool) {
	tok, granting := grantTok, true
	if grantTok == "" {
		tok, granting = revokeTok, false
	}
	if err := grantRail(rt, tok); err != nil {
		log.Fatalf("route: %s: %v", rt.Host, err)
	}
	cur, err := caps.ParseGrants(rt.Caps)
	if err != nil {
		// Refuse rather than repair: this column is written only by SetCaps,
		// so an unparseable value means a hand edit, and rewriting it here
		// would silently decide which half of somebody's edit survived.
		log.Fatalf("route: %s: %v — fix the row with `carlos route --host %s --grant/--revoke` from a known-good state, never by hand",
			rt.Host, err, rt.Host)
	}

	prefix := ""
	if dryRun {
		prefix = colorFor(os.Stdout).warn("(dry run)") + " "
	}
	var next string
	if granting {
		routes, lerr := r.List()
		if lerr != nil {
			log.Fatal(lerr)
		}
		if cerr := portGrantConflict(routes, rt.Host, tok); cerr != nil {
			log.Fatalf("route: %s: %v", rt.Host, cerr)
		}
		if next, err = caps.Grant(rt.Caps, tok); err != nil {
			log.Fatalf("route: %s: %v", rt.Host, err)
		}
		if cur.Has(tok) {
			mutedf(os.Stdout, "%s: %s already granted — nothing to do\n", rt.Host, tok)
			return
		}
	} else {
		if next, err = caps.Revoke(rt.Caps, tok); err != nil {
			log.Fatalf("route: %s: %v", rt.Host, err)
		}
		if !cur.Has(tok) {
			// A quiet success: the caller's goal is "not granted", not "I
			// removed something".
			mutedf(os.Stdout, "%s: %s was not granted — nothing to revoke\n", rt.Host, tok)
			return
		}
	}
	if !dryRun {
		// Compare-and-swap on the value this command computed `next` FROM:
		// losing a race against another operator's grant/revoke refuses with
		// the current value rather than merging a stale read over it.
		if err := r.SetCaps(rt.Host, rt.Caps, next); err != nil {
			log.Fatalf("route: write: %v", err)
		}
	}
	verb := "granted"
	if !granting {
		verb = "revoked"
	}
	fmt.Printf("%s%s: %s %s (caps: %s)\n", prefix, rt.Host, verb, tok, capsLabel(next))
	// Deliberately no restart, no unit rewrite, and no advisory here: a grant
	// is a row, and the converge pass is what materializes (or unmakes) the
	// drop-in on its next run.
}

// capsLabel renders a caps column for a human — "none" rather than an empty
// string, so a revoke that emptied the column reads as an answer.
func capsLabel(col string) string {
	if col == "" {
		return "none"
	}
	return col
}

// addRail refuses the two things `carlos add` must never mint: a host under
// the reserved sidecar suffix, and a sidecar row itself. Sidecars are
// provisioned through their own record family (carlos sidecars create), so
// an operator minting one by hand would produce a row no reconcile pass
// owns and no record explains. Defence in depth behind Task 3's structural
// refusal to SERVE a Kind=="sidecar" row (edge.go's Reload): this rail
// guards the OTHER direction — an ordinary-kind row planted under a
// sidecar's reserved name, which edge would serve without a second thought.
func addRail(host, kind string) error {
	if strings.HasSuffix(host, provision.SidecarSuffix) {
		return fmt.Errorf("%s is under the reserved suffix %s — sidecar hosts are minted by `carlos sidecars create`, never by hand", host, provision.SidecarSuffix)
	}
	if kind == "sidecar" {
		return errors.New("carlos add does not mint sidecars — use `carlos sidecars create -app <app> <name>`")
	}
	return nil
}

// appRail is `carlos add`'s side of the tmp- reservation (temporary static
// sites spec §3): a temporary app's routes come only from its instance
// records, minted through the console, so a hand-added route may never wear
// a tmp- name. Upgrading a box's binary upgrades this writer; there is no
// other copy of it.
func appRail(app string) error {
	if release.IsTemporaryName(app) {
		return fmt.Errorf("--app %s: names starting with tmp- belong to temporary apps, which get their routes from the console. carlos add cannot route one", app)
	}
	return nil
}

// suffixRail is `carlos route`'s side of the same guarantee addRail closes
// at `carlos add`: this verb cannot MINT a row under the reserved sidecar
// suffix (r.Get returns nil for one, and every axis below refuses before
// opening the registry for writing) — but it CAN keep an existing one alive,
// repoint its channel, or grant it capabilities, and that is the door this
// rail closes. It applies to every axis this verb writes (-channel,
// -environment, -tunnel, -grant/-revoke, -hibernate, -backing, -unit): the
// row it guards against is not a genuine Kind=="sidecar" row (edge.go's
// Reload never serves those regardless) but an ordinary-kind row wearing a
// sidecar's reserved name, which this verb must never keep converging.
func suffixRail(rt registry.Route) error {
	if strings.HasSuffix(rt.Host, provision.SidecarSuffix) {
		return fmt.Errorf("%s is under the reserved suffix %s — this verb does not touch routes under it; sidecar hosts are owned by their own record family, never this one", rt.Host, provision.SidecarSuffix)
	}
	return nil
}

// grantRail is everything a --grant/--revoke can be refused for from the ROW
// alone: the token's own validity, plus the two structural refusals. Pure, so
// every refusal is table-testable without a registry.
//
// Both refusals apply to a REVOKE as well as a grant, which is deliberate: on
// a Platform+Canary row the reconciler rebuilds the row this verb would be
// writing, and on a non-unit route there is nothing the platform would
// materialize a grant into, so a "revoke" there would be theatre — a column
// edit that changes nothing while reading like a de-escalation.
func grantRail(rt registry.Route, tok string) error {
	if err := caps.ValidGrantToken(tok); err != nil {
		return err
	}
	if platformCanaryOwnedRow(rt) {
		return fmt.Errorf("Platform+Canary route — the domains reconciler's canary-minting pass rebuilds this whole row " +
			"(internal/domains/reconcile.go) whenever its `followed` check goes false, so this verb does not own it. " +
			"registry.Upsert preserves a non-empty caps column, which makes a grant here survivable — but survivable is not owned, " +
			"and no flag overrides a row this verb does not own")
	}
	if rt.Kind != "instance" || rt.Backing != registry.BackingUnit {
		return fmt.Errorf("capabilities are only honoured for unit-backed instance routes (kind %q, backing %q) — "+
			"a grant here would materialize nothing; see the backing advisory", rt.Kind, rt.Backing)
	}
	return nil
}

// portGrantConflict refuses a port grant already held by a DIFFERENT route on
// this box, naming the other host. `:25` can be held by exactly one process on
// one machine, so two routes granted it is a promise the box cannot keep — and
// the operator finding out here, at the moment they type it, is the whole
// point: converge's own CheckClaims runs hourly, long after the typing.
//
// host-role is deliberately not conflict-checked: the box's instance role is
// not consumed by being used, and two apps holding it is two independent
// operator decisions, not a collision. doc-engine is not conflict-checked for
// the same reason and a sharper one: the capability is per-route unit text,
// not a box resource, and a dedicated sidecar box running several engine
// routes is the expected shape.
//
// A row whose caps column does not parse fails the check CLOSED, naming that
// row: "I could not find out whether this port is already claimed" is never a
// licence to claim it.
func portGrantConflict(routes []registry.Route, host, tok string) error {
	if tok == caps.HostRoleToken || tok == caps.DocEngineToken {
		return nil
	}
	for _, other := range routes {
		if other.Host == host || other.Caps == "" {
			continue
		}
		g, err := caps.ParseGrants(other.Caps)
		if err != nil {
			return fmt.Errorf("cannot prove %s is unclaimed: route %s has an unreadable caps column (%v)", tok, other.Host, err)
		}
		if g.Has(tok) {
			return fmt.Errorf("%s is already granted to %s on this box — a privileged port can be held by exactly one process on one machine; "+
				"revoke it there first (`carlos route --host %s --revoke %s`)", tok, other.Host, other.Host, tok)
		}
	}
	return nil
}

// refuseUnflagNested stops `carlos route --hibernate=false` on a row whose
// paths are nested. Clearing the flag takes the row out of the activator's
// management, and the layout revert only quiesces and moves rows the
// activator manages, so the row's files would be stranded under the
// instance tree (design §8.2, downgrade rules). The test is by path shape,
// whatever else the row says, or by a layout journal, which means the row
// is mid-move and recovery needs it managed.
func refuseUnflagNested(rt registry.Route, dataDir string) error {
	if provision.NestedPaths(dataDir, rt) || provision.LayoutJournalExists(dataDir, rt.Host) {
		return fmt.Errorf("route: %s: its paths are nested. revert the layout first; a nested instance must stay managed so the revert can stop it", rt.Host)
	}
	return nil
}

// runRouteHibernate flips the per-route hibernation opt-in — the first CLI
// caller registry.SetHibernate has ever had — behind the one rail that makes
// the capability model coherent: a route whose release declares `daemon` may
// not hibernate.
//
// Resolution follows this verb's existing doctrine exactly. A PROVEN daemon
// refuses; an UNPROVEN comparison (no pubkey, no store, an unresolvable
// channel, a bad signature, an unreadable manifest) also refuses, because "I
// could not find out" is never a licence. --hibernate=false needs no proof and
// always proceeds: stopping a route from hibernating can never contradict a
// declaration.
func runRouteHibernate(r *registry.Registry, rt registry.Route, on, dryRun bool) {
	if platformCanaryOwnedRow(rt) {
		// mintedRoute (internal/domains/reconcile.go) never sets Hibernate,
		// and Upsert does not preserve it, so a flip here is reverted to
		// false by the next re-mint — the same structural refusal the other
		// axes carry, for the same reason.
		log.Fatalf("route: %s: Platform+Canary route — the domains reconciler rebuilds this row from values it derives itself, "+
			"and a hibernation flip here is reverted on the next reconcile pass. This verb does not own this row", rt.Host)
	}
	if rt.IsSidecar() && rt.Provisioned {
		// Task 11 item 5 (carried from Task 9's review), the same lesson
		// this verb has already paid for on the Channel and Environment
		// axes (recordOwnedRow) — one column later. Unlike an ORDINARY
		// provisioned instance route, where Hibernate is the instance
		// record's OPTIONAL, tri-state column (absent means this verb is
		// still free to set it — internal/provision/reconcile.go's
		// convergeSidecar-adjacent instance path), a sidecar row's
		// Hibernate is unconditional: convergeSidecar
		// (internal/provision/sidecar_reconcile.go) re-stamps it from
		// SidecarManifest.HibernateOrDefault EVERY pass, with no "the
		// record is silent, this column is free" escape. A flip here lands
		// and is reverted within seconds of the next serial bump, with
		// nothing said anywhere.
		//
		// Defence in depth, not the only door: every genuine sidecar host
		// is minted under provision.SidecarSuffix, so suffixRail already
		// refuses this axis (and every other) before this function is ever
		// reached in practice. This check is what fires for a row that
		// reaches this far anyway — a Kind=="sidecar" row outside the
		// reserved suffix, which suffixRail's host-string check alone
		// cannot see — and it is what gives an operator who does meet it
		// the actual reason, not just "reserved suffix".
		log.Fatalf("route: %s: provisioned sidecar row — its hibernation is owned by the app's sidecars manifest at\n%s, "+
			"which the sidecar reconcile pass converges onto this row on every pass; a flip here is reverted within\n"+
			"seconds of the next serial bump. This verb does not own this column",
			rt.Host, provision.SidecarManifestKey(release.RouteAccount(rt), rt.App))
	}
	if on {
		store, pub := deploymentReadSeam()
		daemon, note, proven := routeDaemonCheck(context.Background(), store, pub, release.RouteAccount(rt), rt)
		if !proven {
			log.Fatalf("route: %s: --hibernate refused: %s — refusing regardless: the comparison itself is unproven, and %q is never a licence",
				rt.Host, note, "I could not find out")
		}
		if daemon {
			log.Fatalf("route: %s: %s (%s)", rt.Host, daemonHibernateRefusal, note)
		}
	}
	if !on {
		if err := refuseUnflagNested(rt, dataDir()); err != nil {
			log.Fatalf("%v", err)
		}
	}
	prefix := ""
	if dryRun {
		prefix = colorFor(os.Stdout).warn("(dry run)") + " "
	}
	if rt.Hibernate == on {
		mutedf(os.Stdout, "%s: hibernate already %v — nothing to do\n", rt.Host, on)
		return
	}
	if !dryRun {
		if err := r.SetHibernate(rt.Host, on); err != nil {
			log.Fatalf("route: write: %v", err)
		}
	}
	fmt.Printf("%s%s: hibernate %v -> %v\n", prefix, rt.Host, rt.Hibernate, on)
	// Said at the moment of typing, because the write does two things and
	// only one of them was asked for. registry.SetHibernate clears the
	// replication class when hibernation goes off — a rung is only read on
	// the activator's wake path, and a non-hibernating database is covered
	// by box-wide /etc/litestream.yml at its own interval, so keeping the
	// column would leave `carlos routes` printing a rung nothing honours.
	// An operator who turns hibernation back on has to name the rung again,
	// and being told that here is the difference between a decision and a
	// discovery.
	if !on && rt.Replication != "" {
		fmt.Printf("  replication %s cleared — the rung is read only on the activator's wake path, and this route no longer sleeps.\n", replicationWord(rt.Replication))
		fmt.Printf("  its database is now covered by box-wide /etc/litestream.yml at that unit's own interval; re-`--hibernate` then set the rung again if you want it back\n")
	}
}

// runRouteBacking stamps who owns a route's process — the door design §5b
// and §5c reference and the code did not have: backing was choosable at
// `carlos add` and never afterwards, so the migration's stamp step (stop the
// hand-provisioned unit, copy the data, stamp the row, let the next adopt
// pass converge its capabilities) had no command, and neither did the
// reverse move the migration's reversibility depends on.
//
// Three rails, all structural, none overridable:
//
//   - a Platform route is refused. A platform ALIAS materializes nothing by
//     design (a second carlos-app@ unit claiming :25 would break the real
//     one), and the domains reconciler rebuilds these rows from mintedRoute
//     — which sets backing from its own derivation, so a stamp here is
//     reverted on the next pass. Same refusal, same reason, as the other
//     single-route axes.
//   - kind != "instance" is refused: only an instance has a process for
//     anything to back. A service or static route has no ExecStart to move.
//   - a UNIT-STAMPED row may not be moved TO unit backing. That is the same
//     forbidden state unitRail guards from the other side, reached in the
//     other order (`--unit X` first, `--backing unit` second), and it has to be
//     refused at both ends or the pair is only half-guarded.
//   - the value is "" or BackingUnit, checked at the flag before the
//     registry is even opened.
//
// It does NOT check whether the data has been copied, or whether a unit
// exists — it cannot, and pretending to would be worse than not trying. The
// migration checklist is the operator's; this verb is the one write in it,
// and it prints what the next adopt pass will do with the new value so the
// operator can decide whether they are ready for it.
func runRouteBacking(r *registry.Registry, rt registry.Route, backing string, dryRun bool) {
	if err := backingRail(rt, backing); err != nil {
		log.Fatalf("route: %s: %v", rt.Host, err)
	}
	prefix := ""
	if dryRun {
		prefix = colorFor(os.Stdout).warn("(dry run)") + " "
	}
	if rt.Backing == backing {
		mutedf(os.Stdout, "%s: backing already %s — nothing to do\n", rt.Host, backingWord(backing))
		return
	}
	if !dryRun {
		if err := r.SetBacking(rt.Host, backing); err != nil {
			log.Fatalf("route: write: %v", err)
		}
	}
	fmt.Printf("%s%s: backing %s -> %s\n", prefix, rt.Host, backingWord(rt.Backing), backingWord(backing))
	// The consequence, named at the moment of typing rather than discovered
	// in an hourly log: the two halves of the platform that branch on this
	// column change sides together.
	if backing == registry.BackingUnit {
		fmt.Printf("  the next adopt pass converges this route's capability drop-ins and config sync writes /etc/carlos/apps/%s.env; the agent stops materializing /data/%s.config.env\n", rt.Host, rt.Host)
		fmt.Printf("  for a daemon route that pass also enables and STARTS carlos-app@%s.service — the adopt timer is hourly, so stamp this only once the data it will serve is in place\n", rt.Host)
	} else {
		fmt.Printf("  the next adopt pass stops converging this route's capabilities (any drop-ins already written are left exactly as they are, for an operator to remove) and config delivery returns to the agent's /data/%s.config.env\n", rt.Host)
	}
}

// backingRail is the row-shape half of runRouteBacking's refusals, split out
// as a pure function for the same reason grantRail is: a rail nobody can
// exercise without running the process is a rail nobody tests.
//
// Refusing EVERY Platform row (not only Platform+Canary, as the channel and
// environment axes do) is deliberate and stricter. Those axes refuse the
// canary-minted rows because the reconciler re-derives them; this one refuses
// a plain platform apex too, because backing is not merely rebuilt for those
// rows — it is meaningless. A platform alias is a second name for one
// process, and a second carlos-app@ unit under it would claim ports and a
// state directory the real route already owns.
//
// It takes the TARGET value as well as the row because one of its refusals
// is directional: moving a unit-stamped row TO unit backing creates the state
// unitRail forbids from the other side, while moving one AWAY from unit
// backing is always fine (it can only reduce the number of things claiming to
// own the process). Refusing both directions would strand any row that
// somehow reached the forbidden state with no way back out.
func backingRail(rt registry.Route, backing string) error {
	if backing == registry.BackingUnit && rt.Unit != "" {
		return fmt.Errorf("route names the unit %s — a unit-backed route's process is the carlos-app@%s unit the platform derives and already restarts, "+
			"so a second unit name would be two answers to one question. Clear the stamp first: `carlos route --host %s --unit -`", rt.Unit, rt.Host, rt.Host)
	}
	if rt.Platform {
		return fmt.Errorf("Platform route — a platform alias is a second name for one process, so a unit of its own would " +
			"claim ports and a state directory the real route already holds; and the domains reconciler rebuilds these rows " +
			"from values it derives itself. This verb does not own this row")
	}
	if rt.Kind != "instance" {
		return fmt.Errorf("backing is only meaningful for an instance route (kind %q) — a %s route has no process of its own to back", rt.Kind, rt.Kind)
	}
	return nil
}

// runRouteUnit stamps the systemd unit that owns a route's process — the door
// issue #110 needed. A console is an addr-backed route (`--addr
// 127.0.0.1:8090`) under an operator-run console.service: the adopt pass's
// restart dispatch could reach it by neither of its arms, so promote → adopt
// stamped a new Version, and with it the edge's X-Carlos-Version header,
// behind a process still running the old build. The rule was "promote →
// adopt → ssh in and systemctl restart console.service", which is exactly the
// kind of follow-up step a `carlos` command is not allowed to require.
//
// This is a stamp on an EXISTING row on a live box, deliberately: the
// production consoles already exist, and the fix must not need them recreated.
//
// Two rails, both structural, neither overridable:
//
//   - a unit-backed route is refused. Its process is a carlos-app@<host>
//     unit the platform derives and already restarts; a second unit name on
//     the same row is two answers to one question, and the dispatch would
//     honour this one and stop cycling the tenant.
//   - the name's shape is checked at the flag, before the registry is even
//     opened (registry.ValidUnitName), because a typo here is invisible
//     afterwards — a restart that reaches nothing looks exactly like a
//     restart that worked.
//
// It does NOT check that the unit exists. It cannot: on the box where it
// matters the command may well run before the unit is installed, and a probe
// that has to succeed would make the stamp unavailable exactly when it is
// most needed. What it does instead is print what the next adopt pass will do
// with the value, at the moment of typing.
func runRouteUnit(r *registry.Registry, rt registry.Route, unit string, dryRun bool) {
	if err := unitRail(rt); err != nil {
		log.Fatalf("route: %s: %v", rt.Host, err)
	}
	prefix := ""
	if dryRun {
		prefix = colorFor(os.Stdout).warn("(dry run)") + " "
	}
	if rt.Unit == unit {
		mutedf(os.Stdout, "%s: unit already %s — nothing to do\n", rt.Host, unitWord(unit))
		return
	}
	if !dryRun {
		if err := r.SetUnit(rt.Host, unit); err != nil {
			log.Fatalf("route: write: %v", err)
		}
	}
	fmt.Printf("%s%s: unit %s -> %s\n", prefix, rt.Host, unitWord(rt.Unit), unitWord(unit))
	if unit != "" {
		fmt.Printf("  every adopt pass that moves this route's version now runs `systemctl try-restart %s` instead of cycling nothing — try-restart, so a unit you have stopped stays stopped\n", unit)
	} else {
		fmt.Printf("  adoption goes back to the backing dispatch for this route: a carlos-app@%s unit if it is unit-backed, the activator's own child otherwise — and neither exists for an addr-backed route, which is what leaves its process running the old build\n", rt.Host)
	}
}

// unitRail is the row-shape half of runRouteUnit's refusals, split out as a
// pure function for the same reason backingRail is: a rail nobody can
// exercise without running the process is a rail nobody tests.
//
// Deliberately NARROWER than backingRail. It does not refuse Platform rows:
// this column is not derived by the domains reconciler (mintedRoute never
// sets it) and registry.Upsert preserves a non-empty value, so a stamp there
// survives a re-mint rather than being silently reverted — and the
// deployment's own console can legitimately be a platform-minted row. It does
// not refuse non-instance kinds either: a `service` route is precisely a
// sibling process the platform only proxies, and naming the unit that owns it
// is a true statement, even though adoption has no version to move for one
// today.
func unitRail(rt registry.Route) error {
	if rt.Backing == registry.BackingUnit {
		return fmt.Errorf("unit-backed route — its process is the carlos-app@%s unit the platform derives and already restarts, "+
			"so a second unit name here would be two answers to one question (drop the backing first with `carlos route --backing \"\"` if this route really is owned by a bespoke service)", rt.Host)
	}
	return nil
}

// unitWord renders a unit column value for an operator — the empty string is
// this column's default and a real answer ("nothing named"), and printing it
// as "" reads as a bug.
func unitWord(unit string) string {
	if unit == "" {
		return `"" (no unit named — adoption dispatches on backing)`
	}
	return unit
}

// runRouteReplication puts one route on a rung of the replication ladder.
//
// The ladder was built and honoured and had no door: three named rungs, one
// of them reachable, and it was the most expensive one. Every route on the
// fleet therefore ran on the every-second default, which for one tenant was
// about 100x the S3 traffic the alternative would have cost — a 610 MiB
// database, low write volume, compacted several times a day (discussion
// #10). This verb is the smallest thing that closes that: the activator
// already reads the column on wake and hands the interval to litestream, so
// nothing new happens on the box.
//
// It is deliberately the OPERATOR's door and not a member-facing knob. The
// loss window is a durability promise the platform makes — `daily` means up
// to a day of writes lost with the box — and whether a member may choose it,
// or whether it belongs to a plan tier, is a design question #10 filed
// rather than answered. Nothing here forecloses either answer.
//
// Two rails, both structural, and they are one premise read whole. The class
// is read in exactly one place — the activator's wake path — and
// /etc/litestream.yml covers everything box-wide litestream is NOT told to
// leave alone, at its own interval. What it is told to leave alone is
// `r.Hibernate && r.DBPath != ""` (internal/ops/litestream.go), so BOTH
// halves are required: a non-hibernating route, and a hibernating route
// whose row names no database, are each swept into the box-wide config and
// each would have the column written, reported as set, and read by nothing —
// the same silent no-op the missing door was.
//
// The column also cannot outlive the first half. registry.SetHibernate
// clears the rung when hibernation goes off, and Upsert refuses to carry one
// onto a non-hibernating row, so `carlos routes` cannot come to print a rung
// this rail would no longer let anyone set.
//
// The value's shape is checked at the flag, before the registry is opened.
func runRouteReplication(r *registry.Registry, rt registry.Route, class string, dryRun bool) {
	if err := replicationRail(rt); err != nil {
		log.Fatalf("route: %s: %v", rt.Host, err)
	}
	prefix := ""
	if dryRun {
		prefix = colorFor(os.Stdout).warn("(dry run)") + " "
	}
	if rt.Replication == class {
		mutedf(os.Stdout, "%s: replication already %s — nothing to do\n", rt.Host, replicationWord(class))
		return
	}
	if !dryRun {
		if err := r.SetReplication(rt.Host, class); err != nil {
			log.Fatalf("route: write: %v", err)
		}
	}
	fmt.Printf("%s%s: replication %s -> %s\n", prefix, rt.Host, replicationWord(rt.Replication), replicationWord(class))
	// Said every time, because the write is not the change: the activator
	// hands the interval to litestream when it wakes the instance, so an
	// instance that is awake right now keeps its current interval until it
	// next sleeps and wakes.
	fmt.Printf("  takes effect on this instance's next wake — the sync interval is handed to litestream at wake, so an instance awake now keeps the old one until it sleeps\n")
}

// replicationRail is runRouteReplication's pair of refusals, split out as a
// pure function for the reason unitRail and backingRail are: a rail nobody
// can exercise without running the process is a rail nobody tests.
func replicationRail(rt registry.Route) error {
	if !rt.Hibernate {
		return fmt.Errorf("route does not hibernate — the replication class is read only where the activator replicates, on wake, and /etc/litestream.yml (which covers every NON-hibernating database on this box) has its own interval. "+
			"Setting a rung here would write a column nothing reads. Use `carlos route --host %s --hibernate` first if this route should be activator-managed", rt.Host)
	}
	// Hibernation alone was never the premise — the exclusion it stands on
	// is `r.Hibernate && r.DBPath != ""` (ops.FleetDBsWithSkipped), and
	// provision.routeFor's comment already records this trap from the other
	// side. A hibernating route whose row names no database is NOT excluded
	// from box-wide litestream: the dataDir glob claims its file anyway and
	// replicates it at that unit's interval, so a rung set here would be
	// written, reported by `carlos routes`, and honoured by nothing — the
	// same silent no-op the missing door was, reached through the door.
	//
	// Empty db_path on a hibernating row is not hypothetical: `carlos add`
	// defaults the path only for a single-upstream `instance` that is not
	// unit-backed, so a static or service route, or an instance record
	// carrying an upstream SET (provision.Record.DerivesPaths is false
	// there), hibernates with the column empty.
	if rt.DBPath == "" {
		return fmt.Errorf("route hibernates but its row names no database (db_path is empty) — box-wide litestream excludes a hibernating route only when it knows WHERE its database is (`r.Hibernate && r.DBPath != \"\"`, ops.FleetDBs), "+
			"so this route's file is still being replicated at /etc/litestream.yml's own interval and a rung set here would be honoured by nothing. "+
			"Give the row its path first (`carlos add --host %s --db <absolute path>`, which rewrites the existing route), then set the rung", rt.Host)
	}
	return nil
}

// replicationWord renders a replication column value for an operator — "" is
// this column's default and a real answer, and printing it as "" reads as a
// bug rather than as the rung every route was on before there was a door.
func replicationWord(class string) string {
	if class == "" {
		return `"" (the default: change-driven ~1s sync, same as stream)`
	}
	switch class {
	case registry.ReplicationStream:
		return "stream (change-driven ~1s — loss window seconds)"
	case registry.ReplicationBatched:
		return "batched (at most every 5m — loss window minutes)"
	case registry.ReplicationDaily:
		return "daily (at wake and every 24h — loss window up to a day)"
	}
	return class
}

// backingWord renders a backing column value for an operator — the empty
// string is a real value of this column, and printing it as "" reads as a
// bug rather than as an answer.
func backingWord(backing string) string {
	if backing == "" {
		return `"" (exec child of the agent)`
	}
	return backing
}

// daemonHibernateRefusal is the one sentence both doors say — `carlos route
// --hibernate` and `carlos add --hibernate` — so an operator who meets it at
// one of them recognises it at the other.
const daemonHibernateRefusal = "--hibernate is refused for a route whose release declares the `daemon` capability: " +
	"a daemon is long-running BY DECLARATION — the platform enables it, starts it and restarts it on adopt — " +
	"and hibernating it would have the activator stop the very process that declaration exists to keep up"

// routeDaemonCheck reports whether the release this route runs declares the
// `daemon` capability, and whether that answer is PROVEN. The three returns
// are the same shape routeVersionCheck uses, and for the same reason: the
// caller must be able to tell "no daemon" from "I could not find out", and
// treat them differently.
//
// It checks BOTH the version the route is currently serving and the version
// its channel points at, because both are "what this route runs": the first
// is what is running now, the second is what the next adopt converges onto.
// Either declaring daemon is a refusal.
//
// A route naming no app is proven NOT a daemon rather than unproven — there
// is no release, so there is nothing that could declare anything, and that is
// a proof rather than an absence of one.
func routeDaemonCheck(ctx context.Context, store release.Store, pub ed25519.PublicKey, acct string, rt registry.Route) (daemon bool, note string, proven bool) {
	if rt.App == "" {
		return false, "route names no app, so no release can declare a capability for it", true
	}
	if pub == nil {
		return false, "cannot verify what this route's release declares: CARLOS_RELEASE_PUBKEY is not set", false
	}
	if store == nil {
		return false, "cannot verify what this route's release declares: no deployment store is configured " +
			"(set CARLOS_DEPLOYMENT_BUCKET or CARLOS_DEPLOYMENT_DIR)", false
	}
	ch := rt.Channel
	if ch == "" {
		ch = "stable"
	}
	ptr, err := release.Channel(ctx, store, pub, acct, rt.App, ch)
	if err != nil {
		return false, fmt.Sprintf("channel %s (%s) does not resolve: %v", ch, release.ChannelKey(acct, rt.App, ch), err), false
	}
	versions := []string{ptr.Version}
	if rt.Version != "" && rt.Version != ptr.Version {
		versions = append(versions, rt.Version)
	}
	for _, v := range versions {
		m, merr := release.GetManifest(ctx, store, acct, rt.App, v)
		if merr != nil {
			return false, fmt.Sprintf("manifest for %s %s does not read: %v", rt.App, v, merr), false
		}
		if m.Capabilities != nil && m.Capabilities.Daemon {
			return true, fmt.Sprintf("%s %s declares it", rt.App, v), true
		}
	}
	return false, fmt.Sprintf("no daemon capability in %s %s", rt.App, strings.Join(versions, " or ")), true
}

// deploymentReadSeam resolves the public key and store used for READ-ONLY
// release checks, returning nils instead of exiting when either is absent.
// openStore's log.Fatal is right for a verb that cannot do its job without a
// store; it is wrong here, where "no store configured" is an ANSWER the
// caller has to weigh — `carlos add` proceeds on it (provision time, no
// bucket yet), `carlos route --hibernate` refuses on it.
func deploymentReadSeam() (release.Store, ed25519.PublicKey) {
	var pub ed25519.PublicKey
	if b := strings.TrimSpace(os.Getenv("CARLOS_RELEASE_PUBKEY")); b != "" {
		if p, err := release.ParsePublicKey(b, "CARLOS_RELEASE_PUBKEY"); err == nil {
			pub = p
		}
	}
	if pub == nil {
		return nil, nil
	}
	if b := os.Getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" {
		s, err := release.NewDeploymentS3Store(context.Background(), b)
		if err != nil {
			return nil, pub
		}
		return s, pub
	}
	if d := os.Getenv("CARLOS_DEPLOYMENT_DIR"); d != "" {
		ds := release.NewDirStore(d)
		ds.EnableChangeIndex()
		return ds, pub
	}
	return nil, pub
}

// routeVersionCheck resolves the rail for one route: does moving it from
// fromChannel to toChannel change what gets served?
//
//   - safe=true: the comparison resolved and the two sides agree — proceed
//     unconditionally, nothing changes.
//   - safe=false, overridable=true: the comparison resolved and the two
//     sides genuinely disagree — --allow-version-change is the intended door.
//   - safe=false, overridable=false: nothing here PROVES what would change
//     (no pubkey, an unresolvable channel, a bad signature) — never
//     overridable.
//
// rt.Version — stamped by the last adopt pass — is the honest "what is this
// route actually serving" for an instance route. SetVersions is only ever
// called for Kind == instance, though, so a static route (or an instance
// never yet adopted) always has Version == "" — comparing that against a
// resolved target would either garble the message or force
// --allow-version-change unconditionally. For those, the only available proxy
// is what the route's CURRENT channel resolves to, compared against the
// target — both sides read through the identical release.Channel path the
// runtime itself uses.
func routeVersionCheck(ctx context.Context, store release.Store, pub ed25519.PublicKey, acct string, rt registry.Route, fromChannel, toChannel string) (note string, safe, overridable bool) {
	if pub == nil {
		return "cannot verify what the target channel serves: CARLOS_RELEASE_PUBKEY is not set", false, false
	}
	target, err := release.Channel(ctx, store, pub, acct, rt.App, toChannel)
	if err != nil {
		return fmt.Sprintf("target channel %s (%s) does not resolve: %v",
			toChannel, release.ChannelKey(acct, rt.App, toChannel), err), false, false
	}
	if rt.Version != "" {
		if target.Version == rt.Version {
			return fmt.Sprintf("version unchanged: %s", target.Version), true, false
		}
		return fmt.Sprintf("version changing: %s -> %s (route is running %s, %s currently serves %s)",
			rt.Version, target.Version, rt.Version, toChannel, target.Version), false, true
	}
	current, err := release.Channel(ctx, store, pub, acct, rt.App, fromChannel)
	if err != nil {
		return fmt.Sprintf("route has no recorded version, and its current channel %s (%s) does not resolve\n"+
			"either, so there is nothing to compare against: %v",
			fromChannel, release.ChannelKey(acct, rt.App, fromChannel), err), false, false
	}
	if current.Version == target.Version {
		return fmt.Sprintf("version unchanged: %s (route has no adopted Version; compared %s's pointer against %s's)",
			target.Version, fromChannel, toChannel), true, false
	}
	return fmt.Sprintf("version changing: %s -> %s (route has no adopted Version; %s currently serves %s, %s currently serves %s)",
		current.Version, target.Version, fromChannel, current.Version, toChannel, target.Version), false, true
}

// accountFlags registers the --account/--account-id pair (and the --registry
// seam) every app verb carries. --account takes a sqid OR a name, resolved
// symmetrically against the registry's materialized accounts — neither space
// wins, so a name that collides with some other account's sqid is refused,
// not silently redirected. --account-id is the way out of that refusal, and
// the reference a script should carry: a sqid, matched as a sqid only, never
// looked up as a name. --account defaults to the built-in ops account, so a
// single-tenant deployment never types either flag. Reads fall back to the
// legacy unqualified keys either way, so an unmigrated app keeps working.
//
// Returns the raw flag values alongside a resolver thunk, rather than the
// resolved value alone, because resolution has to happen AFTER fs.Parse —
// and it opens the registry, which a verb that never uses the account should
// not pay for. The raw pointer exists for API mode (cli.go's four
// console-API verbs): they must send the --account value verbatim (a
// developer laptop has no local registry to resolve it against, and the
// console does its own resolution against the caller's memberships), and
// reading it straight off this return value keeps that path from being
// string-coupled to this function's internal flag name the way
// fs.Lookup("account") would be.
func accountFlags(fs *flag.FlagSet) (raw, rawID *string, resolve func() string) {
	raw, rawID, _, resolve = accountFlagsWithRegistry(fs)
	return raw, rawID, resolve
}

// accountFlagsWithRegistry is accountFlags for the handful of verbs that
// need the registry PATH as well as the account it resolves — not the flag's
// default, but whatever --registry actually says.
//
// It exists so that those verbs cannot end up reading a different database
// than the one their own --account was resolved against. A verb that called
// defaultRegistry() directly would do exactly that the moment anyone passed
// --registry, and would do it silently, which for `env set`'s restart notice
// (appconfig_cli.go) would mean answering "which routes did this reach" from
// the wrong box's registry.
func accountFlagsWithRegistry(fs *flag.FlagSet) (raw, rawID, regPath *string, resolve func() string) {
	// Empty default, two meanings by mode: API mode resolves it to the
	// caller's own account (resolvedAccount, apimode.go); direct mode's
	// resolveAccount reads "" as ops exactly as it always has, so box and
	// script behavior is unchanged.
	acct := fs.String("account", "", "account (sqid or name) the app belongs to (default: your only account, or the configured/chosen one; ops in direct mode)")
	acctID := fs.String("account-id", "", "account sqid, matched as a sqid ONLY — supersedes --account; this is the flag an ambiguity error tells you to use")
	reg := fs.String("registry", defaultRegistry(), "registry database (resolves --account)")
	return acct, acctID, reg, func() string { return resolveAccount(*reg, *acct, *acctID) }
}

func openStore() release.Store {
	if b := os.Getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" {
		s, err := release.NewDeploymentS3Store(context.Background(), b)
		if err != nil {
			log.Fatal(err)
		}
		return s
	}
	if d := os.Getenv("CARLOS_DEPLOYMENT_DIR"); d != "" {
		ds := release.NewDirStore(d)
		ds.EnableChangeIndex()
		return ds
	}
	log.Fatal("set CARLOS_DEPLOYMENT_BUCKET (S3) or CARLOS_DEPLOYMENT_DIR (local)")
	return nil
}

func signingKey() ed25519.PrivateKey {
	k, err := release.ParseSigningKey(os.Getenv("CARLOS_RELEASE_KEY"), "CARLOS_RELEASE_KEY")
	if err != nil {
		log.Fatal(err)
	}
	return k
}

// releaseSigner is the identity the write verbs sign with: the root key, or a
// scoped key travelling with the grant that authorises it.
//
// The grant is discovered from CARLOS_RELEASE_GRANT, exactly as the key is
// discovered from CARLOS_RELEASE_KEY — secret manager to shell variable to
// process memory, never disk. A scoped key with no grant beside it is fatal
// here rather than three machines away: a scoped seed alone proves nothing,
// so it would sign pointers nothing in the fleet will ever accept, and the
// failure would surface as an unexplained signature error on a box.
//
// That check is only possible because a scoped seed is tagged (release's
// ScopedKeyPrefix). Both kinds of key are otherwise 32 random bytes.
func releaseSigner() release.Signer {
	raw := os.Getenv("CARLOS_RELEASE_KEY")
	key, err := release.ParseSigningKey(raw, "CARLOS_RELEASE_KEY")
	if err != nil {
		log.Fatal(err)
	}
	blob := strings.TrimSpace(os.Getenv("CARLOS_RELEASE_GRANT"))
	if blob == "" {
		if release.IsScopedSeed(raw) {
			log.Fatal("CARLOS_RELEASE_KEY holds a scoped key but CARLOS_RELEASE_GRANT is not set — a scoped key alone proves nothing; set the grant blob that was printed beside the key by `carlos release-keygen --scope`")
		}
		return release.RootSigner(key)
	}
	// A grant is only meaningful against the key the fleet trusts, so scoped
	// signing needs the deployment's public key too — the same value every box
	// is configured with, not a new secret.
	if strings.TrimSpace(os.Getenv("CARLOS_RELEASE_PUBKEY")) == "" {
		log.Fatal("CARLOS_RELEASE_GRANT is set, so this is a scoped key: signing also needs the deployment's root public key in CARLOS_RELEASE_PUBKEY, which is what the grant is checked against")
	}
	sg, err := release.ScopedSigner(key, blob, publicKey())
	if err != nil {
		log.Fatal(err)
	}
	return sg
}

func publicKey() ed25519.PublicKey {
	k, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		log.Fatal(err)
	}
	return k
}

// shipNow is the clock direct-mode `carlos ship` stamps BuiltAt with. A test
// seam only: a re-ship of an existing version is byte-compared with the
// stored manifest, BuiltAt included, so a test of the re-ship path pins it.
var shipNow = time.Now

const (
	shipSwapFlagHelp = "hold: this release stops each instance before starting it, for a migration the previous build cannot share the database with"
	shipBadSwap      = "--swap takes hold, or leave it off (%q is not a swap demand)"
	shipHoldNote     = "  this release swaps in hold mode: each instance stops before the new build starts"
)

// validSwapFlag is --swap's client-side gate on ship and deploy (spec §4.2):
// a release may demand hold, or nothing.
func validSwapFlag(s string) error {
	if s == "" || s == release.DemandHold {
		return nil
	}
	return fmt.Errorf(shipBadSwap, s)
}

func runShip(args []string) {
	fs := flag.NewFlagSet("ship", flag.ExitOnError)
	messageFlags := addMutationMessageFlags(fs)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw, acctOf := accountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to ship through (API mode only; default: the only logged-in one)")
	kind := fs.String("kind", "binary", "binary | static")
	ver := versionFlag(fs, "version (git short sha); omit to mint the next iteration under the app's version target")
	target := fs.String("target", "linux-arm64", "GOOS-GOARCH the binary was built for (binary kind, single file)")
	capsFile := fs.String("capabilities", "", "path to a JSON capability block (internal/caps) to embed in the manifest")
	label := fs.String("label", "", "one-line human label for this release (\"fixing modals\")")
	notes := fs.String("notes", "", "changelog fragment for this release")
	notesFile := fs.String("notes-file", "", "file holding the changelog fragment (mutually exclusive with --notes)")
	swap := fs.String("swap", "", shipSwapFlagHelp)
	programChange := fs.Bool("program-change", false, "accept that this release's Go main package differs from the app's previous release (recorded in the manifest)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	// --version is no longer required: with a version target set, the next
	// iteration under it is inferable, and inferring it is the whole point of
	// the target (release versioning spec, 2026-08-08). Everything else about
	// this check is unchanged — an app and at least one file are still the
	// minimum a ship can mean.
	usage := "usage: carlos ship --app <a> [--version <sha>] [--label <one line>] [--notes <text> | --notes-file <file>] [--kind binary|static] [--account <sqid-or-name>] [--account-id <sqid>] [--console <origin>] [--capabilities <file.json>] [--program-change] [--swap hold] <files-or-site-dir>"
	if fs.NArg() == 0 {
		log.Fatal(usage)
	}

	// --notes and --notes-file are two spellings of ONE manifest field, so both
	// together is a refusal rather than a precedence rule nobody remembers.
	if *notes != "" && *notesFile != "" {
		log.Fatal("pass --notes or --notes-file, not both")
	}
	noteText := *notes
	if *notesFile != "" {
		b, err := os.ReadFile(*notesFile)
		if err != nil {
			log.Fatal(err)
		}
		noteText = string(b)
	}
	// Bounded HERE, before the mode split, for the same reason the capability
	// block is: both doors must refuse the same bytes with the same message
	// at the operator's terminal. release.ValidateLabelNotes is the one home
	// for the limits and their wording — direct mode would reach it again
	// inside release.Ship, API mode again in the console, but neither of
	// those is where a human is standing.
	if err := release.ValidateLabelNotes(*label, noteText); err != nil {
		log.Fatal(err)
	}
	if err := validSwapFlag(*swap); err != nil {
		log.Fatal(err)
	}

	// Read and decode the declaration BEFORE the mode split, so both doors
	// refuse the same bytes with the same message at the operator's terminal
	// — the strict decoder for an unknown capability name, ValidCapabilities
	// for anything semantically wrong. Direct mode would find out again in
	// release.Ship and API mode again in the console, but neither of those is
	// where a human is standing.
	// Validated against the app NAME, which may still be empty here when
	// --app is left to resolution — so with an omitted --app the read happens
	// again after resolution (below), and this pre-split pass only runs
	// when the name is already known. Both doors still refuse the same
	// bytes with the same message; an omitted --app just moves the moment.
	var capsSet *caps.Set
	if *app != "" {
		var err error
		capsSet, err = readCapabilities(*capsFile, *app)
		if err != nil {
			log.Fatal(err)
		}
	}

	sess, err := resolveMode(*consoleFlag)
	if err != nil {
		log.Fatal(err)
	}
	if sess != nil {
		message, merr := resolveMutationMessage(fs, messageFlags)
		if merr != nil {
			log.Fatal(merr)
		}
		// The banner prints BEFORE the account resolves: resolution may put
		// a muted "account: …" line — or the whole interactive chooser — on
		// the terminal, and a prompt with no console named yet is a question
		// with no context. The account is still resolved HERE, not inside
		// runShipAPI: resolvedAccount fatals (on a name in --account-id, or
		// an ambiguous omitted --account), and log.Fatal skips every deferred
		// function — which is the one thing runShipAPI exists to avoid (its
		// static branch defers the cleanup of a scratch tarball up to
		// 256MiB). Resolved before the call, the refusal lands before
		// runShipAPI's body has registered a defer at all.
		printUsing(stdout, sess)
		acct := resolvedAccount(sess, acctRaw, acctIDRaw)
		app2 := resolvedApp(sess, acct, *app, usage)
		if *app == "" {
			// --app came from resolution: the capability block is validated
			// now, against the real name.
			capsSet, err = readCapabilities(*capsFile, app2)
			if err != nil {
				log.Fatal(err)
			}
		}
		sess.shipSwap = *swap
		if _, err := runShipAPI(sess, acct, fs, app2, *ver, *kind, *target, *label, noteText, capsSet, false, *programChange, message); err != nil {
			log.Fatal(err)
		}
		return
	}

	if *app == "" {
		log.Fatal(usage)
	}
	files := map[string][]byte{}
	arts := map[string]release.Artifact{}
	addFile := func(name string, data []byte) {
		files[name] = data
		sum := sha256.Sum256(data)
		arts[name] = release.Artifact{File: name, SHA256: hex.EncodeToString(sum[:]), Size: int64(len(data))}
	}
	if *kind == "static" {
		// A static release is one deterministic artifact: site.tar.gz of the
		// given directory, so re-shipping an unchanged tree is byte-identical.
		if fs.NArg() != 1 {
			log.Fatal("static ship takes exactly one site directory")
		}
		data, err := edge.PackDir(fs.Arg(0))
		if err != nil {
			log.Fatal(err)
		}
		addFile("site.tar.gz", data)
	} else if fs.NArg() == 1 {
		// The adopt pass looks artifacts up by GOOS-GOARCH, so a binary's
		// artifact key is its target, not its build filename (learned live
		// 2026-08-01: a basename-keyed artifact is invisible to every host).
		data, err := os.ReadFile(fs.Arg(0))
		if err != nil {
			log.Fatal(err)
		}
		addFile(*target, data)
	} else {
		// Multiple files: pre-named per target (linux-arm64, linux-amd64, ...).
		for _, path := range fs.Args() {
			data, err := os.ReadFile(path)
			if err != nil {
				log.Fatal(err)
			}
			addFile(filepath.Base(path), data)
		}
	}
	// Direct mode opens the store ONCE, here: the mint below reads it (the
	// release dirs are the iteration counter — there is no side counter to
	// drift from them), and release.Ship writes through the same handle.
	st := openStore()
	acct := acctOf()
	shipVersion := *ver
	if shipVersion == "" {
		vt, _, err := release.GetTarget(context.Background(), st, acct, *app)
		if err != nil {
			log.Fatal(err)
		}
		if vt == nil {
			log.Fatal("no --version and no version target set — set one with `carlos version target --app " + *app + " <semver>`, or pass --version")
		}
		t, ok := release.ParseSemVer(vt.Target)
		if !ok {
			// GetTarget already refuses a stored target that isn't a bare
			// semver, so this is unreachable through that door — kept because
			// the alternative to a named refusal here is a silent zero SemVer
			// minting "0.0.0-1" against an app nobody meant to touch.
			log.Fatalf("stored version target %q is not a semver — set it again", vt.Target)
		}
		shipVersion, err = release.NextIteration(context.Background(), st, acct, *app, t)
		if err != nil {
			log.Fatal(err)
		}
	}
	m := release.Manifest{
		Schema: 1, App: *app, Version: shipVersion, Kind: *kind,
		BuiltBy: "carlos " + version.Build, BuiltAt: shipNow().UTC().Format(time.RFC3339),
		Artifacts: arts,
		// Empty when the flags were not given: both are omitempty, so the
		// manifest bytes stay byte-identical to what this verb wrote before
		// labels and notes existed.
		Label: *label,
		Notes: noteText,
		// nil when --capabilities was not given: omitempty then keeps the
		// manifest bytes byte-identical to what this verb wrote yesterday.
		Capabilities: capsSet,
		// "" when --swap was not given: omitempty keeps the bytes as before.
		Swap: *swap,
	}
	// The wrong-program rule, shared with the console's API door: refuse a
	// release whose Go main package differs from the app's previous release
	// unless --program-change says it is meant. Only a NEW version meets the
	// comparison: re-shipping a version already on record stays idempotent.
	// A re-ship is still STAMPED the way its first ship was (Program from the
	// bytes, ProgramChange from the flag), because release.Ship below decides
	// between identical and conflict by comparing manifest bytes, and an
	// unstamped manifest never matches a stamped one.
	if _, gerr := release.GetManifest(context.Background(), st, acct, *app, shipVersion); gerr != nil {
		if err := release.CheckShipProgram(context.Background(), st, acct, &m, files, *programChange); err != nil {
			log.Fatal(err)
		}
	} else if err := release.StampProgram(&m, files, *programChange); err != nil {
		log.Fatal(err)
	}
	if err := release.Ship(context.Background(), st, acct, m, files); err != nil {
		log.Fatal(err)
	}
	fmt.Fprintf(stdout, "%s %s %s (%d artifacts)\n", colorFor(stdout).ok("shipped"), *app, shipVersion, len(files))
	if *swap == release.DemandHold {
		mutedf(stdout, shipHoldNote)
	}
}

// readCapabilities is the ship-time declaration surface: a file holding
// EXACTLY the manifest's "capabilities" object (not a whole manifest, not a
// wrapper), decoded through caps.Set's own strict door and then through the
// same ValidCapabilities gate the box runs at adopt.
//
// The file is the whole surface on purpose. There are no per-capability
// flags, because a flag per capability is a vocabulary that drifts from the
// one in internal/caps, and the block belongs with the app's source anyway —
// it is a property of the build, versioned with it.
//
// An empty path means no declaration: nil, and a manifest byte-identical to
// yesterday's. An explicit `null` in the file is refused rather than folded
// into nil — a human who passed --capabilities meant to declare something, and
// silently shipping nothing is the config-that-lies failure mode.
func readCapabilities(path, app string) (*caps.Set, error) {
	if path == "" {
		return nil, nil
	}
	raw, err := os.ReadFile(path)
	if err != nil {
		return nil, fmt.Errorf("ship: --capabilities: %w", err)
	}
	var set *caps.Set
	if err := json.Unmarshal(raw, &set); err != nil {
		return nil, fmt.Errorf("ship: --capabilities %s: %w", path, err)
	}
	if set == nil {
		return nil, fmt.Errorf("ship: --capabilities %s: the file declares null — omit the flag to ship no capabilities", path)
	}
	if err := caps.ValidCapabilities(app, set); err != nil {
		return nil, fmt.Errorf("ship: --capabilities %s: %w", path, err)
	}
	for _, n := range set.Notes() {
		fmt.Fprintf(stdout, "%s\n", n)
	}
	return set, nil
}

// shipTargetRe validates a binary artifact's part name client-side, against
// the exact shape the server enforces (targetRe, internal/console/apicli.go)
// — checked here too so a malformed name (a build filename shipped without
// --target, say) is refused before a single byte crosses the wire, rather
// than after paying for the whole upload and hitting the server's own 400.
var shipTargetRe = regexp.MustCompile(`^[a-z0-9]+-[a-z0-9]+$`)

// runShipAPI is runShip's API-mode body: acct is the caller's --account /
// --account-id as apiAccount read them, and goes straight to the console
// (never acctOf/resolveAccount — there is no local registry to resolve
// against on a developer laptop). It arrives already resolved because
// apiAccount can fatal, and nothing that fatals may run inside this
// function — see the deferred-cleanup note below. Unlike the direct-mode
// body above, artifacts are streamed from disk by path (apiSession.ship
// reads them lazily) rather than buffered into memory here first — for a
// static release, whose bytes only exist in memory (edge.PackDir's tarball),
// that means writing them to a scratch temp file first so there is a path to
// hand over at all.
//
// Returns an error rather than calling log.Fatal itself (the caller does
// that): log.Fatal calls os.Exit immediately, which skips every deferred
// function — including the static kind's scratch temp file cleanup below,
// up to 256MiB per invocation. Returning lets that defer actually run on
// every exit path, fatal or not.
//
// printBanner controls only the "using <console> as <you>" line: true for
// runShip's own call (unchanged), false for a caller — `carlos deploy` —
// that already printed its own banner earlier in the same run (before its
// instance-list call, so a listing failure still names the console) and
// would otherwise print it twice. The "shipped"/"already shipped" line
// below is never suppressed; it is the one true account of what ship did,
// and deploy relies on that rather than printing its own.
func runShipAPI(sess *apiSession, acct apiAcct, fs *flag.FlagSet, app, ver, kind, target, label, notes string, capsSet *caps.Set, printBanner, programChange bool, messages ...string) (apiShipResult, error) {
	message := firstMutationMessage(messages)
	files := map[string]string{}
	if kind == "static" {
		if fs.NArg() != 1 {
			return apiShipResult{}, errors.New("static ship takes exactly one site directory")
		}
		data, err := edge.PackDir(fs.Arg(0))
		if err != nil {
			return apiShipResult{}, err
		}
		tmp, err := os.CreateTemp("", "carlos-ship-*.tar.gz")
		if err != nil {
			return apiShipResult{}, err
		}
		defer os.Remove(tmp.Name())
		if _, err := tmp.Write(data); err != nil {
			tmp.Close()
			return apiShipResult{}, err
		}
		if err := tmp.Close(); err != nil {
			return apiShipResult{}, err
		}
		files["site.tar.gz"] = tmp.Name()
	} else if fs.NArg() == 1 {
		files[target] = fs.Arg(0)
	} else {
		// Multiple files: pre-named per target (linux-arm64, linux-amd64,
		// ...) by basename, exactly like the direct-mode body above.
		for _, path := range fs.Args() {
			files[filepath.Base(path)] = path
		}
	}

	// Client-side name check, binary kind only (a static release's one part
	// is always named "site.tar.gz" above, never a caller-supplied name) —
	// same shape the server enforces, checked here so a bad basename is
	// named and refused before anything is stat'd or dialed.
	if kind != "static" {
		for name, path := range files {
			if !shipTargetRe.MatchString(name) {
				return apiShipResult{}, fmt.Errorf("ship: invalid artifact name %q (from %s) — want a GOOS-GOARCH shape like linux-arm64 "+
					"(pass --target to name a single file explicitly)", name, path)
			}
		}
	}

	// Pre-flight: every artifact path must exist before the console is ever
	// dialed or a byte moves — a typo'd path must fail clean here, not after
	// the "using ..." line already printed.
	for name, path := range files {
		if _, err := os.Stat(path); err != nil {
			return apiShipResult{}, fmt.Errorf("ship: artifact %q: %w", name, err)
		}
	}

	if printBanner {
		printUsing(stdout, sess)
	}
	res, err := sess.ship(context.Background(), acct, app, ver, kind, label, notes, files, capsSet, programChange, message)
	if err != nil {
		return apiShipResult{}, err
	}
	// res.Version, never ver: when the caller passed no --version the CONSOLE
	// minted one from the app's target, and the response is the only place
	// that name exists. It echoes the sent version verbatim otherwise, so
	// this is the right thing to print on both paths rather than two.
	if res.NoOp {
		mutedf(stdout, "already shipped %s %s — identical content, nothing to do\n", app, res.Version)
	} else {
		fmt.Fprintf(stdout, "%s %s %s\n", colorFor(stdout).ok("shipped"), app, res.Version)
	}
	if sess.shipSwap == release.DemandHold {
		mutedf(stdout, shipHoldNote)
	}
	// res carries more than the print above needs: BuildProbe is the app's
	// build-report path, which `carlos deploy` turns into its watch's exit
	// criterion. Returned rather than acted on here — ship itself has no
	// watch.
	return res, nil
}

// runPromote points a channel at a shipped version, with the ladder enforced:
// beta is entered from edge and stable from beta, so "it was signed off on
// beta" means something. Canary channels are never refused (the fleet runs on
// them), a repeated command is a no-op success, and --hotfix is the one
// bypass — explicit, so it can be seen.
func runPromote(args []string) {
	fs := flag.NewFlagSet("promote", flag.ExitOnError)
	messageFlags := addMutationMessageFlags(fs)
	app := appFlag(fs, "")
	hotfix := fs.Bool("hotfix", false, "bypass the channel ladder (recorded as a hotfix)")
	retry := fs.Bool("retry", false, "re-sign the channel's current promotion of this version with a fresh time, so a box that refused it tries again")
	acctRaw, acctIDRaw, acctOf := accountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to promote through (API mode only; default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos promote --app <a> [--hotfix] [--account <sqid-or-name>] [--account-id <sqid>] [--console <origin>] <version> [channel]"
	if fs.NArg() != 1 && fs.NArg() != 2 {
		log.Fatal(usage)
	}
	version, channel := fs.Arg(0), fs.Arg(1) // Arg(1) is "" for the one-arg form

	sess, err := resolveMode(*consoleFlag)
	if err != nil {
		log.Fatal(err)
	}
	if channel == "" && sess == nil {
		// Direct mode is the operator's, against a bucket rather than a
		// console: there is no channels door to ask, and an operator
		// promoting by hand is the one caller who always knows the name.
		log.Fatal(usage)
	}
	if sess != nil {
		// --hotfix now travels over the wire (release pipelines v2, Task 10):
		// handleAPIPromote reads it strictly after its own rules decision
		// (its doc comment), so this is never itself a way past a channel's
		// declared rules — only past the ladder's ORDER check, exactly as it
		// is in direct mode below.
		printUsing(stdout, sess)
		acct := resolvedAccount(sess, acctRaw, acctIDRaw)
		app2 := resolvedApp(sess, acct, *app, usage)
		if channel == "" {
			resolved, cerr := promoteChannel(context.Background(), sess, acct, app2)
			if cerr != nil {
				log.Fatal(cerr)
			}
			channel = resolved
		}
		message, merr := resolveMutationMessage(fs, messageFlags)
		if merr != nil {
			log.Fatal(merr)
		}
		res, err := sess.promote(context.Background(), acct, app2, version, channel, *hotfix, *retry, message)
		if err != nil {
			var prop *apiProposedError
			if errors.As(err, &prop) {
				printProposed(stdout, prop)
				return
			}
			log.Fatal(err)
		}
		if res.NoOp {
			mutedf(stdout, "%s %s is already on %s — nothing to do\n", app2, version, channel)
		} else {
			fmt.Fprintf(stdout, "%s %s %s -> %s\n", colorFor(stdout).ok("promoted"), app2, version, channel)
		}
		return
	}

	if *app == "" {
		log.Fatal(usage)
	}
	res, err := releaseSigner().PromoteOnLadder(context.Background(), openStore(),
		acctOf(), *app, channel, version,
		release.PromoteOptions{Hotfix: *hotfix, Retry: *retry, Actor: promoteActor()})
	if err != nil {
		log.Fatal(err)
	}
	switch {
	case res.NoOp:
		mutedf(stdout, "%s %s is already on %s — nothing to do\n", *app, version, channel)
	case res.Hotfix:
		c := colorFor(stdout)
		fmt.Fprintf(stdout, "%s %s %s -> %s %s\n", c.ok("promoted"), *app, version, channel,
			c.warn("(HOTFIX: ladder bypassed, recorded)"))
	default:
		fmt.Fprintf(stdout, "%s %s %s -> %s\n", colorFor(stdout).ok("promoted"), *app, version, channel)
	}
	// The tag a stable promotion cut, printed after EVERY branch above rather
	// than only the fresh ones: PromoteOnLadder cuts the tag on the no-op
	// path too (a repeat promote heals a version that reached stable before
	// tags existed, or whose tag write failed the first time), and a name
	// that was just minted with nothing said about it is a name the operator
	// never learns. Empty off stable, so nothing new prints anywhere else.
	//
	// TagWarning is a warning, not an error, for the reason release.stableTag
	// gives: the pointer has already moved, so a promotion that happened must
	// report success. It is printed verbatim.
	if res.Tag != "" {
		fmt.Fprintf(stdout, "%s %s\n", colorFor(stdout).ok("tagged"), res.Tag)
	}
	if res.TagWarning != "" {
		fmt.Fprintf(stdout, "%s\n", colorFor(stdout).warn(res.TagWarning))
	}
}

// promoteChannel resolves the channel a one-arg `carlos promote` lands on,
// from the app's own channels rather than from the ladder in --help.
//
// The ladder printed by --help (canary → edge → beta → stable) is the LEGACY
// frozen one, and a pipelined app's channels are its own names, in its own
// order. An agent reading that help and promoting onto "beta" is reading a
// document about a different app — which is exactly what happened
// (skills#29): a two-channel app, a confident guess at a third name, and a
// refusal that (to its credit) named the real one.
//
// So the rule is the same one staticDeployChannel uses: resolve when the
// answer is unambiguous, and when it is not, say what this app actually has
// and let a human pick. A pipeline's declared list wins over the promoted
// rows — a channel with no pointer yet is still a channel you may promote
// onto, and it is the rows that are the incomplete picture, not the pipeline.
//
// canary/* is excluded from the count, not from the choice: it is a reserved
// per-session namespace outside any pipeline (a dead end by design, always
// allowed, zero bake), so a canary someone cut this morning must not turn
// tomorrow's ordinary promote into an ambiguity. Naming one explicitly works
// exactly as it always did.
func promoteChannel(ctx context.Context, sess *apiSession, acct apiAcct, app string) (string, error) {
	res, err := sess.channels(ctx, acct, app)
	if err != nil {
		return "", fmt.Errorf("promote: list channels: %w (name the channel to promote onto)", err)
	}
	return pickPromoteChannel(res, app)
}

// pickPromoteChannel is promoteChannel's decision, split from its round trip
// so the rules above can be read — and tested — without a console. Same shape
// as staticDeployChannel, for the same reason.
func pickPromoteChannel(res apiChannelsResult, app string) (string, error) {
	if len(res.PipelineChannels) == 1 {
		return res.PipelineChannels[0], nil
	}
	if len(res.PipelineChannels) > 1 {
		return "", fmt.Errorf("promote: %s releases through %s — name which one to promote onto "+
			"(its entry channel is %s; see carlos pipeline)",
			app, strings.Join(res.PipelineChannels, " → "), res.PipelineEntry)
	}
	var named []string
	for _, ch := range res.Channels {
		if rung, ok := release.Rung(ch.Channel); ok && rung == release.RungCanary {
			continue
		}
		named = append(named, ch.Channel)
	}
	switch len(named) {
	case 1:
		return named[0], nil
	case 0:
		return "", fmt.Errorf("promote: %s has no channels yet — name the channel to promote onto", app)
	default:
		return "", fmt.Errorf("promote: %s has %d channels (%s) — name which one to promote onto",
			app, len(named), strings.Join(named, ", "))
	}
}

// promoteActor names who is promoting, for the pointer's audit trail: the
// logged-in console identity when this terminal has one, and otherwise the
// operator's user@host. Never fatal and never blocking — a release verb runs
// from CI as often as from a laptop, and "we don't know who" is a worse
// answer than "someone on this machine", but neither is worth refusing a
// promotion over.
func promoteActor() string {
	if c, err := loggedInCredential(""); err == nil && c.Address != "" {
		return c.Address
	}
	return defaultLabel()
}

func runChannels(args []string) {
	fs := flag.NewFlagSet("channels", flag.ExitOnError)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw, acctOf := accountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to read through (API mode only; default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos channels --app <a> [--account <sqid-or-name>] [--account-id <sqid>] [--console <origin>]"

	sess, err := resolveMode(*consoleFlag)
	if err != nil {
		log.Fatal(err)
	}
	if sess != nil {
		printUsing(stdout, sess)
		acct := resolvedAccount(sess, acctRaw, acctIDRaw)
		app2 := resolvedApp(sess, acct, *app, usage)
		chansRes, err := sess.channels(context.Background(), acct, app2)
		if err != nil {
			log.Fatal(err)
		}
		chans := chansRes.Channels
		// A separate read rather than a new field on the channels response:
		// the channels endpoint is what the box-facing world already parses,
		// and a display join belongs on the display side. A tag read that
		// fails degrades to a tag-less table — the question this verb answers
		// is "what is each channel serving", and no tag has ever been part of
		// that answer.
		tags, err := sess.tags(context.Background(), acct, app2)
		if err != nil {
			mutedf(stdout, "(tags unreadable: %v)\n", err)
		}
		byRelease := make(map[string]string, len(tags))
		for _, t := range tags {
			byRelease[t.Release] = t.Version
		}
		w := newTable(stdout)
		fmt.Fprintln(w, textChannelsHeader)
		for _, ch := range chans {
			if ch.Err != "" {
				fmt.Fprintf(w, "%s\tERROR: %s\t\t\t\n", ch.Channel, ch.Err)
				continue
			}
			fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", ch.Channel, ch.Version, ch.PromotedAt, byRelease[ch.Version], digestCell(ch.Digests, ch.Straggler))
		}
		w.Flush()
		return
	}

	if *app == "" {
		log.Fatal(usage)
	}
	acct := acctOf()
	s := openStore()
	prefix := release.ChannelsPrefix(acct, *app)
	keys, err := s.List(context.Background(), prefix)
	if err != nil {
		log.Fatal(err)
	}
	// Same join as API mode above, from the store rather than the console,
	// and degrading the same way.
	tags, err := release.ListTags(context.Background(), s, acct, *app)
	if err != nil {
		mutedf(stdout, "(tags unreadable: %v)\n", err)
	}
	byRelease := make(map[string]string, len(tags))
	for _, t := range tags {
		byRelease[t.Release] = t.Version
	}
	pub := publicKey()
	marker, _ := release.ReadDigestMarker(context.Background(), s, pub, acct)
	w := newTable(stdout)
	fmt.Fprintln(w, textChannelsHeader)
	for _, k := range keys {
		name := strings.TrimSuffix(strings.TrimPrefix(k, prefix), ".json")
		cp, err := release.Channel(context.Background(), s, pub, acct, *app, name)
		if err != nil {
			fmt.Fprintf(w, "%s\tERROR: %v\t\t\t\n", name, err)
			continue
		}
		st := release.DigestStateOf(cp, marker)
		fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", name, cp.Version, cp.PromotedAt.Format(time.RFC3339), byRelease[cp.Version], digestCell(st.HeadDigest, st.Straggler))
	}
	w.Flush()
}

// runVersionTarget is `carlos version target`: the per-app semver that
// `carlos ship` counts iterations under (release versioning spec,
// 2026-08-08). No positional argument reads it, one sets it — rather than a
// -set flag, which would make a mistyped read (a stray value, a shell glob)
// into a write of the one number every future version of this app hangs off.
//
// This is a SUB-verb of `carlos version` (main.go), which on its own still
// prints the binary's build id and nothing else: the oldest output in this
// program, typed by every install script there is.
func runVersionTarget(args []string) {
	fs := flag.NewFlagSet("version target", flag.ExitOnError)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw, acctOf := accountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to read or set through (API mode only; default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos version target --app <a> [--account <sqid-or-name>] [--account-id <sqid>] [--console <origin>] [<semver>]"
	if fs.NArg() > 1 {
		log.Fatal(usage)
	}

	sess, err := resolveMode(*consoleFlag)
	if err != nil {
		log.Fatal(err)
	}
	if sess != nil {
		printUsing(stdout, sess)
		acct := resolvedAccount(sess, acctRaw, acctIDRaw)
		app2 := resolvedApp(sess, acct, *app, usage)
		// unset is what both modes print when no target has ever been set, and
		// it names the way to set one: an operator meeting this line is one
		// command away from the feature, and the command is not guessable
		// from "none".
		unset := func() {
			mutedf(stdout, "no version target set\n")
			mutedf(stdout, "  set one with: carlos version target --app %s 0.1.0\n", app2)
		}
		if fs.NArg() == 0 {
			vt, err := sess.versionTarget(context.Background(), acct, app2)
			if err != nil {
				log.Fatal(err)
			}
			if vt.Target == "" {
				unset()
				return
			}
			fmt.Fprintf(stdout, "%s %s\n", colorFor(stdout).ok("version target"), vt.Target)
			if vt.SetBy != "" {
				mutedf(stdout, "  set by %s at %s\n", vt.SetBy, vt.SetAt)
			}
			return
		}
		stored, err := sess.setVersionTarget(context.Background(), acct, app2, fs.Arg(0))
		if err != nil {
			log.Fatal(err)
		}
		fmt.Fprintf(stdout, "%s %s\n", colorFor(stdout).ok("version target"), stored)
		return
	}

	if *app == "" {
		log.Fatal(usage)
	}
	unset := func() {
		mutedf(stdout, "no version target set\n")
		mutedf(stdout, "  set one with: carlos version target --app %s 0.1.0\n", *app)
	}
	acct := acctOf()
	s := openStore()
	if fs.NArg() == 0 {
		vt, _, err := release.GetTarget(context.Background(), s, acct, *app)
		if err != nil {
			log.Fatal(err)
		}
		if vt == nil {
			unset()
			return
		}
		fmt.Fprintf(stdout, "%s %s\n", colorFor(stdout).ok("version target"), vt.Target)
		if vt.SetBy != "" {
			mutedf(stdout, "  set by %s at %s\n", vt.SetBy, vt.SetAt.Format(time.RFC3339))
		}
		return
	}
	// SetTarget carries both refusals — a target that isn't a bare semver,
	// and one that is not above every tag already cut — and its messages name
	// the offending value, so they pass through untouched.
	target := fs.Arg(0)
	if err := release.SetTarget(context.Background(), s, acct, *app, target, promoteActor(), time.Now().UTC()); err != nil {
		log.Fatal(err)
	}
	fmt.Fprintf(stdout, "%s %s\n", colorFor(stdout).ok("version target"), target)
}

// releaseRow is one row of `carlos releases`, in either mode: the console
// joins the tags server-side (handleAPIReleasesList) and direct mode joins
// them here, but both arrive at the same five columns so the table renders
// once below. Err is not a column — it is the direct-mode "this manifest
// would not read" row, printed in the LABEL cell the way runChannels prints
// an unverifiable pointer.
type releaseRow struct {
	Version, Label, Kind, BuiltAt, Tag, Err string
}

// releasesCap bounds the list, matching the console's own versionsCap: a
// long-lived app has thousands of releases and nobody reads past the first
// screen of them.
const releasesCap = 20

// localReleases is the direct-mode half of `carlos releases`: pick the
// releases worth reading, read those manifests, join the tags, newest first.
//
// It is a local sibling of console.Versions rather than a call to it,
// deliberately: the CLI does not import internal/console — that package
// carries templates, sessions, a registry and a whole HTTP server. What the
// two DO share is release.VersionCandidates, so both surfaces pick the same
// releasesCap versions (by manifest mod time, when the store carries stamps
// — a bare *S3Store or *DirStore, which is all openStore hands out) and read
// nothing else; the two modes of this one command agree about what "the
// latest release" is. A manifest that will not read is a row with Err set,
// exactly as it is on the console's app page and in runChannels above, not a
// failure of the whole listing.
func localReleases(ctx context.Context, s release.Store, acct, app string) ([]releaseRow, error) {
	cands, _, err := release.VersionCandidates(ctx, s, acct, app, releasesCap)
	if err != nil {
		return nil, err
	}
	tags, err := release.ListTags(ctx, s, acct, app)
	if err != nil {
		return nil, err
	}
	byRelease := make(map[string]string, len(tags))
	for _, t := range tags {
		byRelease[t.Release] = t.Version
	}
	var out []releaseRow
	for _, v := range cands {
		m, err := release.GetManifest(ctx, s, acct, app, v)
		if err != nil {
			out = append(out, releaseRow{Version: v, Tag: byRelease[v], Err: err.Error()})
			continue
		}
		out = append(out, releaseRow{
			Version: v, Label: m.Label, Kind: m.Kind, BuiltAt: m.BuiltAt, Tag: byRelease[v],
		})
	}
	// Newest first by build time, version-descending within a tie — the same
	// order console.Versions sorts by, so the two surfaces agree about what
	// "the latest release" is. An unparseable BuiltAt sorts as the zero time
	// (last), which is where a manifest that could not be read belongs.
	parse := func(s string) time.Time {
		t, _ := time.Parse(time.RFC3339, s)
		return t
	}
	sort.SliceStable(out, func(i, j int) bool {
		ti, tj := parse(out[i].BuiltAt), parse(out[j].BuiltAt)
		if !ti.Equal(tj) {
			return ti.After(tj)
		}
		return out[i].Version > out[j].Version
	})
	if len(out) > releasesCap {
		out = out[:releasesCap]
	}
	return out, nil
}

// runReleases is `carlos releases`: what has been shipped, newest first,
// with the label its author typed and the tag it earned if it reached
// stable. `carlos channels` answers "what is running"; this answers "what is
// there to run".
func runReleases(args []string) {
	fs := flag.NewFlagSet("releases", flag.ExitOnError)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw, acctOf := accountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to read through (API mode only; default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos releases --app <a> [--account <sqid-or-name>] [--account-id <sqid>] [--console <origin>]"

	sess, err := resolveMode(*consoleFlag)
	if err != nil {
		log.Fatal(err)
	}
	var rows []releaseRow
	appName := *app
	if sess != nil {
		printUsing(stdout, sess)
		acct := resolvedAccount(sess, acctRaw, acctIDRaw)
		appName = resolvedApp(sess, acct, *app, usage)
		rels, err := sess.releases(context.Background(), acct, appName)
		if err != nil {
			log.Fatal(err)
		}
		for _, r := range rels {
			rows = append(rows, releaseRow{Version: r.Version, Label: r.Label, Kind: r.Kind, BuiltAt: r.BuiltAt, Tag: r.Tag})
		}
	} else {
		if appName == "" {
			log.Fatal(usage)
		}
		rows, err = localReleases(context.Background(), openStore(), acctOf(), appName)
		if err != nil {
			log.Fatal(err)
		}
	}

	if len(rows) == 0 {
		mutedf(stdout, "no releases yet — carlos ship --app %s <files>\n", appName)
		return
	}
	w := newTable(stdout)
	fmt.Fprintln(w, "VERSION\tLABEL\tKIND\tBUILT\tTAG")
	for _, r := range rows {
		if r.Err != "" {
			fmt.Fprintf(w, "%s\tERROR: %s\t\t\t%s\n", r.Version, r.Err, r.Tag)
			continue
		}
		fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", r.Version, r.Label, r.Kind, r.BuiltAt, r.Tag)
	}
	w.Flush()
}

func runRollback(args []string) {
	fs := flag.NewFlagSet("rollback", flag.ExitOnError)
	messageFlags := addMutationMessageFlags(fs)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw, acctOf := accountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to roll back through (API mode only; default: the only logged-in one)")
	usage := "usage: carlos rollback --app <a> [--account <sqid-or-name>] [--account-id <sqid>] [--console <origin>] <channel>"
	fs.Usage = flagUsage(fs, usage)
	fs.Parse(args)
	if fs.NArg() != 1 {
		log.Fatal(usage)
	}

	sess, err := resolveMode(*consoleFlag)
	if err != nil {
		log.Fatal(err)
	}
	if sess != nil {
		printUsing(stdout, sess)
		acct := resolvedAccount(sess, acctRaw, acctIDRaw)
		app2 := resolvedApp(sess, acct, *app, usage)
		message, merr := resolveMutationMessage(fs, messageFlags)
		if merr != nil {
			log.Fatal(merr)
		}
		if err := sess.rollback(context.Background(), acct, app2, fs.Arg(0), message); err != nil {
			var prop *apiProposedError
			if errors.As(err, &prop) {
				printProposed(stdout, prop)
				return
			}
			log.Fatal(err)
		}
		fmt.Fprintf(stdout, "%s %s %s\n", colorFor(stdout).ok("rolled back"), app2, fs.Arg(0))
		return
	}

	if *app == "" {
		log.Fatal(usage)
	}
	if err := releaseSigner().Rollback(context.Background(), openStore(), acctOf(), *app, fs.Arg(0)); err != nil {
		log.Fatal(err)
	}
	fmt.Fprintf(stdout, "%s %s %s\n", colorFor(stdout).ok("rolled back"), *app, fs.Arg(0))
}

// ledgerUsage lists every `carlos ledger` sub-verb — printed for a bare
// `carlos ledger`, an unrecognized sub, and `-h`, matching RunAuth/
// RunAccounts/RunEnv's own bare/unknown/-h behaviour (auth.go, accounts.go,
// appconfig_cli.go): "-h" is never special-cased inside a sub's own
// flag.FlagSet, it just isn't a recognized args[0] here either, so it falls
// through to this same text.
const ledgerUsage = `usage: carlos ledger <command>

commands:
  append  --app <a> [--account <acct>] [--account-id <sqid>] <ledger> <entry.json|->
                                       append one JSON entry to the chain —
                                       reads the file, or stdin with "-";
                                       validated as JSON before it is sent
  head    --app <a> [--account <acct>] [--account-id <sqid>] <ledger>
                                       print the chain's HEAD.json verbatim
  list    --app <a> [--account <acct>] [--account-id <sqid>]
                                       list this app's ledger names and its
                                       publish manifest (what's public, if
                                       anything)
  publish --app <a> [--account <acct>] [--account-id <sqid>] [--public=true|false] [--title <t>] <ledger>...
                                       add these ledgers to what this app
                                       serves publicly (additive: the ones
                                       already public stay public, and an
                                       omitted --title keeps the stored one);
                                       --public=false removes just these,
                                       taking the app private only when
                                       nothing is left
  blob    --app <a> [--account <acct>] [--account-id <sqid>] <ledger> <file>...
                                       upload one or more content-addressed
                                       blobs (create-only; re-upload of
                                       identical bytes is a no-op)
  verify  --app <a> --account <sqid> --url <https://status-host> <ledger>
                                       walk a PUBLISHED chain over plain
                                       HTTPS and re-hash every entry — needs
                                       NO credentials, so anyone can check a
                                       published ledger. --account here is the
                                       account SQID, not a name: the public
                                       status-host path is keyed by sqid.

Every sub above except "verify --url" goes through the console API — this
terminal never touches bucket credentials directly (open-ledgers design
spec §2.3). [--console
<origin>] picks which logged-in console when more than one is configured.
`

// RunLedger dispatches the ledger sub-verbs.
func RunLedger(args []string) {
	if len(args) == 0 {
		fmt.Fprint(os.Stderr, ledgerUsage)
		os.Exit(1)
	}
	switch args[0] {
	case "append":
		runLedgerAppend(args[1:])
	case "head":
		runLedgerHead(args[1:])
	case "list":
		runLedgerList(args[1:])
	case "publish":
		runLedgerPublish(args[1:])
	case "blob":
		runLedgerBlob(args[1:])
	case "verify":
		runLedgerVerify(args[1:])
	default:
		fmt.Fprint(os.Stderr, ledgerUsage)
		os.Exit(1)
	}
}

// ledgerAPISession resolves this terminal's console session for a ledger
// sub-verb that writes or reads through the console (every sub except
// `verify --url`, which never calls this at all). Unlike promote/channels/
// rollback/ship, ledgers have no direct-bucket mode to fall back to — spec
// §2.3 is unconditional: a contributor machine gets no bucket credentials,
// full stop. resolveMode returning (nil, nil) means CARLOS_DEPLOYMENT_BUCKET
// or _DIR is set, i.e. direct mode is what this terminal is configured for —
// which for every OTHER release verb is a valid path to take, but for ledger
// is refused here rather than silently reinterpreted or left for a caller
// to nil-dereference.
func ledgerAPISession(consoleFlag string) *apiSession {
	sess, err := resolveMode(consoleFlag)
	if err != nil {
		log.Fatal(err)
	}
	if sess == nil {
		log.Fatal("carlos ledger always goes through the console API — no bucket credentials in a terminal " +
			"(spec §2.3); run carlos auth login --console <url>, or unset CARLOS_DEPLOYMENT_BUCKET/CARLOS_DEPLOYMENT_DIR")
	}
	return sess
}

// ledgerAccountFlags registers the account flags a ledger verb has, in
// place of accountFlags. accountFlags also registers --registry, the local
// registry database that resolves an account name in DIRECT-bucket mode —
// and ledgerAPISession refuses direct mode outright (spec §2.3: no bucket
// credentials in a terminal). A --registry flag on these verbs is therefore
// dead: nothing here ever opens a registry, and offering the flag advertises
// a mode that does not exist. The account is resolved server-side, by the
// console, against the caller's own memberships — which is why --account
// takes a sqid OR a name here and needs no local database to do it —
// and why --account-id needs no local database either: it rides the wire as
// ?exact=sqid, the console-side switch that makes the segment match as a
// sqid only, so the console's own symmetric resolution can never name-match
// it.
func ledgerAccountFlags(fs *flag.FlagSet) (raw, rawID *string) {
	return fs.String("account", "", "account (sqid or name) the app belongs to — resolved by the console (default: your only account, or the configured/chosen one)"),
		fs.String("account-id", "", "account sqid, matched as a sqid ONLY — supersedes --account")
}

func runLedgerAppend(args []string) {
	fs := flag.NewFlagSet("ledger append", flag.ExitOnError)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw := ledgerAccountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to append through (default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos ledger append --app <a> [--account <acct>] [--account-id <sqid>] <ledger> <entry.json|->"
	if fs.NArg() != 2 {
		log.Fatal(usage)
	}
	name, src := fs.Arg(0), fs.Arg(1)
	if !ledger.ValidName(name) {
		log.Fatalf("ledger append: invalid ledger name %q", name)
	}

	// Read and validate the payload BEFORE ever dialing the console — a
	// malformed entry is the caller's mistake to see immediately, not a
	// round trip and a server-side 400.
	var raw []byte
	var err error
	if src == "-" {
		raw, err = io.ReadAll(os.Stdin)
	} else {
		raw, err = os.ReadFile(src)
	}
	if err != nil {
		log.Fatalf("ledger append: %v", err)
	}
	if !json.Valid(raw) {
		log.Fatalf("ledger append: %s is not valid JSON", src)
	}

	sess := ledgerAPISession(*consoleFlag)
	printUsing(stdout, sess)
	acct := resolvedAccount(sess, acctRaw, acctIDRaw)
	app2 := resolvedApp(sess, acct, *app, usage)
	a, err := sess.ledgerAppend(context.Background(), acct, app2, name, json.RawMessage(raw))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Fprintf(stdout, "%s %s: entry #%d (sha %s)\n", colorFor(stdout).ok("appended to"), name, a.Count, a.SHA256[:12])
}

// runLedgerHead prints HEAD.json verbatim — no "using ... as ..." banner
// line first, unlike every other sub here, so the output stays valid JSON a
// script can pipe straight into `jq` rather than a human-readable line plus
// a JSON blob.
func runLedgerHead(args []string) {
	fs := flag.NewFlagSet("ledger head", flag.ExitOnError)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw := ledgerAccountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to read through (default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos ledger head --app <a> [--account <acct>] [--account-id <sqid>] <ledger>"
	if fs.NArg() != 1 {
		log.Fatal(usage)
	}
	name := fs.Arg(0)
	if !ledger.ValidName(name) {
		log.Fatalf("ledger head: invalid ledger name %q", name)
	}

	sess := ledgerAPISession(*consoleFlag)
	acct := resolvedAccount(sess, acctRaw, acctIDRaw)
	app2 := resolvedApp(sess, acct, *app, usage)
	h, err := sess.ledgerHead(context.Background(), acct, app2, name)
	if err != nil {
		log.Fatal(err)
	}
	b, err := json.MarshalIndent(h, "", "  ")
	if err != nil {
		log.Fatal(err)
	}
	fmt.Fprintln(stdout, string(b))
}

func runLedgerList(args []string) {
	fs := flag.NewFlagSet("ledger list", flag.ExitOnError)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw := ledgerAccountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to read through (default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos ledger list --app <a> [--account <acct>] [--account-id <sqid>]"

	sess := ledgerAPISession(*consoleFlag)
	printUsing(stdout, sess)
	acct := resolvedAccount(sess, acctRaw, acctIDRaw)
	app2 := resolvedApp(sess, acct, *app, usage)
	body, err := sess.ledgerList(context.Background(), acct, app2)
	if err != nil {
		log.Fatal(err)
	}
	var res struct {
		Manifest *ledger.Manifest `json:"manifest"`
		Ledgers  []string         `json:"ledgers"`
	}
	if err := json.Unmarshal(body, &res); err != nil {
		log.Fatalf("ledger list: unreadable response: %v", err)
	}

	switch {
	case res.Manifest == nil:
		mutedf(stdout, "not published — nothing is public for this app yet")
	case res.Manifest.Public:
		title := res.Manifest.Title
		if title == "" {
			title = app2
		}
		fmt.Fprintf(stdout, "published %q: %s\n", title, strings.Join(res.Manifest.Ledgers, ", "))
	default:
		mutedf(stdout, "unpublished (public=false)")
	}
	if len(res.Ledgers) == 0 {
		mutedf(stdout, "no ledgers yet")
		return
	}
	w := newTable(stdout)
	fmt.Fprintln(w, "LEDGER")
	for _, n := range res.Ledgers {
		fmt.Fprintln(w, n)
	}
	w.Flush()
}

func runLedgerPublish(args []string) {
	fs := flag.NewFlagSet("ledger publish", flag.ExitOnError)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw := ledgerAccountFlags(fs)
	public := fs.Bool("public", true, "serve these ledgers publicly (--public=false unpublishes)")
	title := fs.String("title", "", "display title for the published set")
	consoleFlag := consoleOption(fs, "console origin to publish through (default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos ledger publish --app <a> [--account <acct>] [--account-id <sqid>] [--public=true|false] [--title <t>] <ledger>..."
	if fs.NArg() == 0 {
		log.Fatal(usage)
	}
	names := fs.Args()
	for _, n := range names {
		if !ledger.ValidName(n) {
			log.Fatalf("ledger publish: invalid ledger name %q", n)
		}
	}

	sess := ledgerAPISession(*consoleFlag)
	printUsing(stdout, sess)
	acct := resolvedAccount(sess, acctRaw, acctIDRaw)
	app2 := resolvedApp(sess, acct, *app, usage)
	m := ledger.Manifest{Public: *public, Ledgers: names, Title: *title}
	got, err := sess.ledgerPublish(context.Background(), acct, app2, m)
	if err != nil {
		log.Fatal(err)
	}
	// Report the RESULTING public set, never the argument: publish is
	// additive and unpublish removes by name, so the app's public set after
	// this call is something only the server can say. Echoing `names` here
	// would tell a caller who published `prompts` that prompts is what is
	// public — when carbon is public too, or, on an unpublish, when carbon is
	// what is left.
	verb := "published"
	if !*public {
		verb = "unpublished"
	}
	if len(got.Ledgers) == 0 {
		fmt.Fprintf(stdout, "%s %s: no public ledgers\n", verb, app2)
		return
	}
	fmt.Fprintf(stdout, "%s %s: public ledgers now [%s]\n", verb, app2, strings.Join(got.Ledgers, " "))
}

func runLedgerBlob(args []string) {
	fs := flag.NewFlagSet("ledger blob", flag.ExitOnError)
	app := appFlag(fs, "")
	acctRaw, acctIDRaw := ledgerAccountFlags(fs)
	consoleFlag := consoleOption(fs, "console origin to upload through (default: the only logged-in one)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	usage := "usage: carlos ledger blob --app <a> [--account <acct>] [--account-id <sqid>] <ledger> <file>..."
	if fs.NArg() < 2 {
		log.Fatal(usage)
	}
	name := fs.Arg(0)
	if !ledger.ValidName(name) {
		log.Fatalf("ledger blob: invalid ledger name %q", name)
	}
	files := fs.Args()[1:]

	sess := ledgerAPISession(*consoleFlag)
	printUsing(stdout, sess)
	acct := resolvedAccount(sess, acctRaw, acctIDRaw)
	app2 := resolvedApp(sess, acct, *app, usage)
	for _, path := range files {
		data, err := os.ReadFile(path)
		if err != nil {
			log.Fatalf("ledger blob: %v", err)
		}
		if len(data) > ledger.MaxBlob {
			log.Fatalf("ledger blob: %s is %d bytes, exceeds the %d byte cap", path, len(data), ledger.MaxBlob)
		}
		sha, err := sess.ledgerBlob(context.Background(), acct, app2, name, data)
		if err != nil {
			log.Fatal(err)
		}
		fmt.Fprintf(stdout, "%s: %s\n", path, sha)
	}
}

// runLedgerVerify never calls resolveMode/ledgerAPISession — verifying a
// published chain needs no console session and no credential at all (spec
// §2.5): it is a plain HTTPS walk against the public status-host, which is
// the entire point of publishing one. --account here is the account SQID,
// not a name — the public path is keyed by sqid (apimode.go's apiPath, by
// contrast, accepts either, because the console resolves a name against the
// caller's own memberships; there is no caller and no membership check
// here).
func runLedgerVerify(args []string) {
	fs := flag.NewFlagSet("ledger verify", flag.ExitOnError)
	app := appFlag(fs, "")
	account := fs.String("account", "", "account SQID (not name) — the public status-host path is keyed by sqid")
	urlBase := fs.String("url", "", "public status-host base URL, e.g. https://status.example.com — no credentials needed")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)
	if *app == "" || *account == "" || *urlBase == "" || fs.NArg() != 1 {
		log.Fatal("usage: carlos ledger verify --app <a> --account <sqid> --url <https://status-host> <ledger>")
	}
	name := fs.Arg(0)
	if !ledger.ValidName(name) {
		log.Fatalf("ledger verify: invalid ledger name %q", name)
	}

	g := ledger.HTTPGetter(http.DefaultClient, strings.TrimRight(*urlBase, "/"))
	res, err := ledger.Verify(context.Background(), g, *account, *app, name)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Fprintf(stdout, "%s %d entries, head %s\n", colorFor(stdout).ok("verified:"), res.Count, res.HeadSHA256[:12])
}

// defaultGrantTTL is how long a scoped key lives when the operator does not
// say: a quarter. Short enough that a forgotten builder key stops working
// while someone still remembers what it was for, long enough that rotating it
// is a calendar item rather than a chore.
const defaultGrantTTL = 90 * 24 * time.Hour

// runKeygen has two modes in one verb, because they mint the same kind of
// thing for the same deployment.
//
// Without --scope it is what it has always been: a fresh root pair, printed as
// the two variables a deployment is configured with. Nothing about that
// output or its meaning changes.
//
// With --scope it mints a SCOPED key: a fresh pair plus a grant signed by the
// root key read from CARLOS_RELEASE_KEY (same env discipline as promote — the
// root key is read, never printed, never written). What comes out is the
// scoped seed and the grant blob, which travel together. The deployment's
// CARLOS_RELEASE_PUBKEY does NOT change and is deliberately not reprinted
// here: the fleet's trust anchor stays exactly one key, and a second pubkey
// in this output would invite someone to install it on a box.
func runKeygen(args []string) {
	fs := flag.NewFlagSet("release-keygen", flag.ExitOnError)
	scope := fs.String("scope", "", "mint a SCOPED key for these rungs, comma separated (canary,edge,beta) — stable is never delegable")
	name := fs.String("name", "", "label for the scoped key; it appears in every refusal it causes")
	expires := fs.Duration("expires", defaultGrantTTL, "how long the scoped key may sign for (max 8760h)")
	fs.Usage = flagUsage(fs, "")
	fs.Parse(args)

	if *scope == "" {
		var stray []string
		fs.Visit(func(f *flag.Flag) { stray = append(stray, "--"+f.Name) })
		if len(stray) > 0 {
			log.Fatalf("%s only apply with --scope; without it this mints the deployment's ROOT key", strings.Join(stray, " and "))
		}
		seed, pub, err := release.GenerateKey()
		if err != nil {
			log.Fatal(err)
		}
		fmt.Printf("CARLOS_RELEASE_KEY=%s\nCARLOS_RELEASE_PUBKEY=%s\n", seed, pub)
		fmt.Fprintln(os.Stderr, "store the KEY in your secret manager; the PUBKEY goes in host config")
		return
	}

	if *name == "" {
		log.Fatal("usage: carlos release-keygen --scope <rungs> --name <label> [--expires <dur>]")
	}
	if release.IsScopedSeed(os.Getenv("CARLOS_RELEASE_KEY")) {
		log.Fatal("CARLOS_RELEASE_KEY holds a scoped key: only the ROOT key can mint a grant, which is what makes a scoped key unable to widen itself")
	}
	var rungs []string
	for _, r := range strings.Split(*scope, ",") {
		if r = strings.TrimSpace(r); r != "" {
			rungs = append(rungs, r)
		}
	}
	seed, grant, err := release.MintKeyGrant(signingKey(), *name, rungs, *expires)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("CARLOS_RELEASE_KEY=%s\nCARLOS_RELEASE_GRANT=%s\n", seed, grant)
	fmt.Fprintf(os.Stderr, `scoped key %q for %s, expiring %s

  The KEY is the secret — store it wherever the builder's secrets live.
  The GRANT is not secret (public material only) but must travel WITH the
  key: a scoped key alone signs pointers nothing will accept.
  CARLOS_RELEASE_PUBKEY does not change; the fleet still trusts one key.

  Deploy order: fleet binaries that understand grants FIRST, this key
  second. An older binary reads a grant-signed pointer as a bad signature,
  which is the right way round to fail but will stop that box adopting.
`, *name, strings.Join(rungs, ", "), time.Now().UTC().Add(*expires).Format(time.RFC3339))
}

// statusUsage is shared between "no flag given" and "-h/--help" so an
// operator gets the same explanation either way: this verb reads and checks,
// it never publishes. The fleet status page's PUBLISH loop (CollectBox ->
// WriteBox on every box, BuildFleet -> PublishFleet on the lease holder) runs
// inside the box agent's own 5-minute tick — there is deliberately no
// "carlos status publish" here; --live only reads what those ticks wrote.
const statusUsage = `usage: carlos status --dry-run <dir>
       carlos status --live
       carlos status --instances [--account <sqid>]
       carlos status --layout

  --dry-run <dir>  build a throwaway fleet stuffed with fixture secrets
                   (hostnames, a machine id, mid-band fractions), run it
                   through the real CollectBox/WriteBox/BuildFleet/
                   PublishFleet path into <dir>, then read every published
                   object back and fail if any fixture leaked. The self-
                   hoster's "what would I publish?" check, run against a
                   scratch directory before ever pointing this binary at a
                   real bucket.

  --live            read status/boxes/*.json out of the status bucket and print
                   one line per host: release, tunnel state, memory, load and
                   free disk — raw figures, never published on the public
                   status page. A host that has stopped reporting prints as
                   UNREACHABLE with how long it has been dark and none of its
                   last readings. Needs bucket credentials and
                   CARLOS_STATUS_BUCKET (or CARLOS_DEPLOYMENT_BUCKET, the same
                   defaulting the agent's own status loop applies).

  --instances       who is using what right now: one row per AWAKE instance
                    across the fleet, with the account and app it belongs to
                    and what it is consuming. Signed in through the console
                    (carlos auth login), operator only — a non-operator token
                    gets the same bare 404 an unknown route gets. --account
                    <sqid> narrows it to one account. A sleeping instance has
                    no process and therefore no reading, so it does not appear;
                    and accounts are joined against the console's own box's
                    route registry, so an instance on another box in a fleet
                    can land under (platform). Both limits print with the
                    table.

  --layout          which layout each instance is on, one row per host, box by
                    box: flat (shared /data), nested (its own directory), fenced
                    (mid-move, will not wake), conflict (two paths claim it,
                    will not wake), foreign (paths the pass will not touch) or
                    external (another supervisor owns it). Operator only, through
                    the console, like --instances.
`

// RunStatus is the "carlos status" verb: a fixture dry-run and a live read of
// what the fleet's boxes last reported. Modeled on ops.Run's FlagSet shape
// (main.go's "ops" case), but flatter — two independent modes, no sub-verb
// dispatch.
func RunStatus(args []string) {
	fs := flag.NewFlagSet("status", flag.ExitOnError)
	dryRunDir := fs.String("dry-run", "", "build a fixture fleet, publish it into this directory, and verify no leaks")
	liveFlag := fs.Bool("live", false, "read status/boxes/*.json from the status bucket and print per-host metrics")
	instancesFlag := fs.Bool("instances", false, "one row per awake instance across the fleet, with the account and app it belongs to (operator only, through the console)")
	acctFilter := fs.String("account", "", "narrow --instances to one account (sqid)")
	layoutFlag := fs.Bool("layout", false, "which layout each instance is on, one row per host, box by box (operator only, through the console, like --instances)")
	consoleFlag := consoleOption(fs, "console to ask (default: the only logged-in one)")
	fs.Usage = flagUsage(fs, statusUsage)
	fs.Parse(args)

	if *instancesFlag {
		// Console-backed, unlike --live: the per-unit readings live in the
		// deployment's metrics objects, which the console already sweeps, and
		// the account each one belongs to is a registry join only the console
		// can do. --live reads the status bucket directly with the operator's
		// own bucket credentials; this deliberately needs none, which is the
		// whole point (discussion #11: the alternative was `aws s3 cp`).
		if err := statusInstances(os.Stdout, *consoleFlag, *acctFilter); err != nil {
			log.Fatal(err)
		}
		return
	}
	if *layoutFlag {
		// Console-backed for the identical reason --instances is: the
		// layout report lives in each box's own private snapshot, and the
		// console already reads it for its operator page.
		if err := statusLayout(os.Stdout, *consoleFlag); err != nil {
			log.Fatal(err)
		}
		return
	}
	if *liveFlag {
		if err := statusLive(os.Stdout); err != nil {
			log.Fatal(err)
		}
		return
	}
	if *dryRunDir == "" {
		fmt.Fprint(os.Stderr, statusUsage)
		os.Exit(1)
	}
	if err := status.DryRun(*dryRunDir, os.Stdout); err != nil {
		log.Fatal(err)
	}
}

// statusLive prints what every box in the fleet last reported about itself,
// read straight from status/boxes/ — the private snapshots, raw numbers and
// all, which is why this needs the operator's own bucket credentials and why
// nothing here is ever served publicly (see internal/status's package
// comment). Bucket resolution is host.ResolveStatusBucket, the identical
// CARLOS_STATUS_BUCKET-then-CARLOS_DEPLOYMENT_BUCKET rule the agent's status
// loop uses, so this reads exactly the bucket the agents write.
func statusLive(w io.Writer) error {
	ctx := context.Background()
	bucket := host.ResolveStatusBucket(os.Getenv)
	if bucket == "" {
		return fmt.Errorf("set CARLOS_STATUS_BUCKET or CARLOS_DEPLOYMENT_BUCKET")
	}
	s, err := release.NewDeploymentS3Store(ctx, bucket)
	if err != nil {
		return err
	}
	boxes, err := status.LoadBoxes(ctx, s)
	if err != nil {
		return err
	}
	printLiveBoxes(w, boxes, time.Now())
	return nil
}

// printLiveBoxes is statusLive's rendering half, split out so it can be
// asserted on without a bucket. One line per box, label-sorted so two runs
// diff on the fleet rather than on list order.
//
// A stale box (status.BoxStale — the same rule the console and the published
// boxes table use) prints UNREACHABLE and how long it has been dark, and
// deliberately NOT its last readings: those numbers described a machine that
// has since stopped answering, and printing them in the same columns as a
// live host's is how an outage gets read as fine. A fresh box that carries no
// metrics block at all is a host running code from before the block existed —
// which is a fact worth stating, not a row of zeros.
func printLiveBoxes(w io.Writer, boxes []status.Box, now time.Time) {
	sorted := append([]status.Box(nil), boxes...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Label < sorted[j].Label })
	for _, b := range sorted {
		if status.BoxStale(b, now, status.SnapshotPeriod) {
			fmt.Fprintf(w, "%-24s %-12s UNREACHABLE (last heartbeat %s ago)\n",
				b.Label, orDash(b.Region), now.Sub(b.GeneratedAt).Round(time.Minute))
			continue
		}
		m := b.Metrics
		if m == nil {
			fmt.Fprintf(w, "%-24s %-12s ok (no metrics block — pre-upgrade host)\n", b.Label, orDash(b.Region))
			continue
		}
		fmt.Fprintf(w, "%-24s %-12s release=%s tunnel=%s mem=%s/%s load=%s disk-free=%s alerts=%s\n",
			b.Label, orDash(b.Region), orDash(m.Release), orDash(m.Tunnel),
			humanBytes(m.MemAvailBytes), humanBytes(m.MemTotalBytes),
			humanFloat(m.Load1), humanBytes(m.DiskFreeBytes), alertsWord(m.Alerts))
		for _, r := range m.ReleaseRefusals {
			if r.Kind == release.RefusalWrongProgram {
				fmt.Fprintf(w, "  refused %s %s since %s: wrong program (%s)\n", r.Host, r.Version, r.RefusedAt.UTC().Format(time.RFC3339), strings.TrimPrefix(r.Reason, "wrong program: "))
				continue
			}
			if r.Kind == release.RefusalNotProven {
				fmt.Fprintf(w, "  refused %s %s since %s: staged bytes do not match dev's proof (%s)\n", r.Host, r.Version, r.RefusedAt.UTC().Format(time.RFC3339), strings.TrimPrefix(r.Reason, "not proven: "))
				continue
			}
			failed := ""
			if r.RecoveryError != "" {
				failed = " (rollback failed: " + r.RecoveryError + ")"
			}
			fmt.Fprintf(w, "  refused %s %s since %s: %s%s\n", r.Host, r.Version, r.RefusedAt.UTC().Format(time.RFC3339), r.Reason, failed)
		}
		for _, pw := range m.ProofWaits {
			sha := pw.SHA256
			if len(sha) > 12 {
				sha = sha[:12]
			}
			// No prover means the wait is the policy file's, not a missing
			// proof: say so, with the load error exactly as given.
			what := "a " + pw.Prover + " proof"
			if pw.Prover == "" {
				what = "a proof (policy: " + pw.Reason + ")"
			}
			line := fmt.Sprintf("  waiting for %s: %s %s (sha256 %s) since %s", what, pw.Host, pw.Version, orDash(sha), pw.Since.UTC().Format(time.RFC3339))
			if pw.Prover != "" && pw.Reason != "" {
				line += ": " + pw.Reason
			}
			fmt.Fprintln(w, line)
		}
	}
}

// alertsWord renders a box's alert-sink answer for the live listing. "not
// delivered" is spelled out because it is the state discussion #192 found
// silent for weeks: a box whose own disk, memory and cpu rules have nowhere
// to send. A box that predates the field did not say, so it is a dash.
func alertsWord(a *status.AlertDelivery) string {
	if a == nil {
		return "-"
	}
	var sinks []string
	if a.Webhook {
		sinks = append(sinks, "webhook")
	}
	if a.SNS {
		sinks = append(sinks, "sns")
	}
	if len(sinks) == 0 && !a.SNSUnavailable {
		return "not-delivered"
	}
	out := strings.Join(sinks, "+")
	if a.SNSUnavailable {
		// A topic is configured but its client could not be built. Alone it
		// is the whole answer; next to a working sink it is a warning.
		if out == "" {
			out = "sns-unavailable"
		} else {
			out += " (sns-unavailable)"
		}
	}
	if a.WebhookFailures > 0 {
		out += fmt.Sprintf(" (webhook failed %d)", a.WebhookFailures)
	}
	if a.SNSFailures > 0 {
		out += fmt.Sprintf(" (sns failed %d)", a.SNSFailures)
	}
	return out
}

// orDash, humanBytes and humanFloat render one field of a metrics block.
// Unknown is a dash in all three — never a zero, never a blank column: the
// status package carries "I couldn't measure this" as a nil pointer all the
// way to here precisely so the reader can tell it apart from a measured zero.
func orDash(s string) string {
	if s == "" {
		return "-"
	}
	return s
}

func humanBytes(v *int64) string {
	if v == nil {
		return "-"
	}
	b := float64(*v)
	switch {
	case b >= 1<<30:
		return fmt.Sprintf("%.1fGB", b/(1<<30))
	case b >= 1<<20:
		return fmt.Sprintf("%.1fMB", b/(1<<20))
	case b >= 1<<10:
		return fmt.Sprintf("%.1fKB", b/(1<<10))
	default:
		return fmt.Sprintf("%dB", *v)
	}
}

func humanFloat(v *float64) string {
	if v == nil {
		return "-"
	}
	return fmt.Sprintf("%.2f", *v)
}