Merge steward orders 1 (r3): an order is one systemd job; refusal reads the named unit only

Soul cut-steward-orders-1.s5 holds at 77e2295. Not installed: steward-orders-2 installs it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Cargo.lock

@@ -482,6 +482,7 @@ dependencies = [
  "libc",
  "rmp-serde",
  "serde",
+ "serde_json",
  "sha2",
  "tempfile",
  "toml",

Cargo.toml

@@ -38,6 +38,7 @@ serde = { version = "1", features = ["derive"] }
 ed25519-dalek = "2.2"
 toml = "0.8"
 uuid = { version = "1", features = ["v4"] }
+serde_json = "1"
 
 [dev-dependencies]
 tempfile = "3"

README.md

@@ -58,6 +58,64 @@ idunn status --command <id>
 the bindings, so `idunn up profile:full-gamecult` brings up everything bound
 into that profile.
 
+### Steward orders
+
+A second root unit from the same binary, `idunn-orders` (`idunn orders serve`),
+takes typed orders from one steward Mind over `/run/idunn-orders/steward.sock`.
+It has its own store, `orders.cc`, and its own binding, `steward.toml`; the
+survival daemon takes no order. There is still no command string:
+
+```bash
+idunn order package-install <pkg>[=version]... --reason TEXT   # apt-get update, install named packages, check them
+idunn order unit <start|stop|restart|reload> <unit> --reason TEXT
+idunn order deploy <service|profile:name> --reason TEXT        # the same commit as `idunn up`
+idunn order show <id>                                          # status and output tails
+idunn order list
+```
+
+A unit order submits exactly one job, for the unit it names: every verb runs
+`systemctl <verb> --job-mode=ignore-requirements -- <unit>`, so systemd builds a
+transaction of one job. It does not start what the unit requires or restart what
+is part of it: order the dependency first, and order `postgresql@16-main.service`
+rather than the `postgresql.service` wrapper. Targets are not orderable. The job is one; what the unit does afterwards is
+not: systemd still runs full transactions of its own for `Restart=` after a
+failure, for a `StopWhenUnneeded=` dependency left without a user, and for a
+`Conflicts=` partner of a unit that fails, so an ordered stop of a unit
+that something restarts does not last, and a start of a `Restart=always`
+unit can report failure and be active seconds later. A unit
+order may act on any unit except Idunn's own (`idunn-*`) and the minds' own
+(`mind-self@*`, `mind-persona@*`), and the units that end or take the host
+(`systemd-suspend`, `systemd-hibernate`, `systemd-hybrid-sleep`,
+`systemd-suspend-then-hibernate`, `rescue`, `emergency`). That set is compiled
+into the binary and read from no configuration. An order is refused, recorded
+and posted when the ordered unit, or an alias it resolves from, is one of them;
+when it has an emergency action (`SuccessAction=`, `FailureAction=`,
+`StartLimitAction=`, `JobTimeoutAction=`) other than `none` or an `OnFailure=`
+or `OnSuccess=` unit (not for a reload); or, for a stop or restart, when another
+unit is `BoundBy=` it, because systemd stops that unit with it whatever the job
+mode; for a start or restart, when it has an `Upholds=` unit, which systemd
+starts in a full transaction whatever the job mode; or, for a stop, restart or
+reload, when it has a `ConsistsOf=` member, a wrapper such as
+`postgresql.service` whose verbs would exit 0 and touch nothing. A unit
+`systemctl show` cannot resolve fails the order. Slices, scopes,
+targets and globs are not units an order can name.
+
+A package order installs the packages it names and nothing else: apt runs with
+`--no-remove`, each name is passed as an exact-name pattern so `.` and `+`
+match themselves, and a last step fails the order unless every named package
+is installed.
+
+An order runs once, after the steward's uid is checked and the instance's
+brake check (`mind-control --instance <principal> check`) exits 0. The check
+that the order's principal and uid are the bound steward's runs again at every
+execution, so an order accepted before `steward.toml` changed is refused and
+recorded, not run. The socket belongs to the steward user alone (mode 0600);
+root may `show` and `list` orders but never issue one. An order that had
+started when the service died is recorded as lost and never run again. No order releases a brake. Acceptance and result are posted to
+`#operator` as a retried projection of `orders.cc`: a chat outage delays the
+posts and not the orders. A refusal names the field and a code, never the value
+it refused.
+
 ## Authority
 
 **Owner.** Deployment admission, and the continuity of installed bodies.

deploy/idunn-orders.service

@@ -0,0 +1,47 @@
+[Unit]
+Description=Idunn steward orders: typed package, unit and deploy orders from the steward
+# Deliberately no After= or Requires= on any managed target, Synapse or
+# idunn-yggdrasil: orders are how an operator repairs those, so none of them
+# may be a condition of this unit starting. Posts are a retried projection and
+# wait for Synapse on their own.
+
+[Service]
+Type=simple
+
+# Root, from the same binary as idunn-yggdrasil, with its own store and socket.
+# idunn-yggdrasil keeps its no-imperative-input invariant and its write set:
+# no order code runs in `idunn serve`.
+#
+# The unit name matches idunn-*, which is in the always-refused set, so no
+# order can stop or restart this service.
+User=root
+# No Group= and no UMask=: the service gives its socket to the steward user
+# (mode 0600) itself and sets Idunn's own umask for the files it writes, so
+# nothing here reaches control.cc. The steward user is the only peer the
+# service answers: it checks SO_PEERCRED on every connection.
+ExecStart=/usr/local/bin/idunn orders serve \
+  --store /var/lib/gamecult/idunn-orders \
+  --state-store /var/lib/gamecult/idunn/control.cc \
+  --steward-binding /etc/gamecult/idunn/steward.toml \
+  --socket /run/idunn-orders/steward.sock
+Restart=always
+RestartSec=5
+TimeoutStopSec=20
+
+RuntimeDirectory=idunn-orders
+RuntimeDirectoryMode=0755
+
+# The Synapse access token of @idunn, copied from a root-only file, and read
+# through $CREDENTIALS_DIRECTORY; the service names it in steward.toml.
+LoadCredential=idunn-matrix-token:/etc/gamecult/idunn/matrix-token
+
+PrivateTmp=yes
+ProtectHome=yes
+ProtectSystem=strict
+# Its own order store, and the control store a deploy order commits a command
+# to. Transient units started with systemd-run run in the manager's namespace,
+# so packages and units are not limited by this list.
+ReadWritePaths=/var/lib/gamecult/idunn-orders /var/lib/gamecult/idunn
+
+[Install]
+WantedBy=multi-user.target

docs/deploy-provenance-map.md

@@ -199,3 +199,196 @@ on Yggdrasil: a 3.5 GiB tree, its staging copy, downloads and pip cache. It
 removes it at the end. Revision 2 runs it on the trees as shipped, with the 300
 terminfo links (T13). If a tree holds a link the T14 rule refuses, the job
 prints it and stops, and the link is reported to Self, not deleted.
+
+## Cut `isolation-credential-evidence`
+
+Spec `idunn-deploy-provenance:cut_spec:cut-cut-isolation-credential-evidence.r1`,
+at `112a7a5` (origin/main, live on Yggdrasil since 2026-10-24 11:29 UTC). It
+serves Vili's admission (`motion-pipeline:cut_report:cut-vili-admission.h4`,
+first deviation): every `idunn up vili` is refused before fencing.
+
+### Body facts (probed read-only on Yggdrasil, 2026-10-24)
+
+- I1. `idunn up vili` runs 6 and 7 failed with
+  `systemd workload isolation properties differ from the Idunn contract`, the
+  message of `observe_unit` (`src/drivers.rs:2975-3005`). Of its conjuncts,
+  only one differs on the failed Vili unit: `LoadCredential`. Idunn expects
+  `""` because Vili has no service credential besides the presence identity
+  (`service_credential_names` empty); systemd shows `[unprintable]`. Every
+  other conjunct matches, InaccessiblePaths included (one value across all
+  five idunn-vili/odin units).
+- I2. systemd 257.13 `systemctl show -p LoadCredential` prints
+  `LoadCredential=[unprintable]` for every unit, whatever the credentials:
+  `ssh.service` and `systemd-journald.service` (not DynamicUser, zero
+  LoadCredential), the four failed `idunn-vili-*` units (zero), and the Odin
+  unit (one). The display form carries no information, so the conjunct was
+  vacuous whenever credentials existed and always false when none did. Odin
+  and StreamPixels passed only because each carries a regular credential.
+- I3. The exact value is readable: D-Bus
+  `org.freedesktop.systemd1.Service.LoadCredential` (`busctl get-property`)
+  answers `a(ss) 0` for the Vili units and ssh, `a(ss) 1
+  "ODIN_TOPOLOGY_IDENTITY" "<path>"` for Odin.
+- I4. What the workload can actually read is the delivered directory. On the
+  host, `/run/credentials/` holds a directory only for units with credentials
+  (getty, journald, Odin, StreamPixels service and web). Inside Odin's MainPID
+  root, `/run/credentials/<unit>/` holds exactly `ODIN_TOPOLOGY_IDENTITY`
+  (`-r--r-----+ root root`). The presence identity is a parent-only `OpenFile`
+  descriptor, never a credential.
+- I5. The rule arrived whole in `1fc0dad` (2026-09-03, "Rebuild Idunn
+  deployment transaction authority"), with its twin clause in
+  `validate_launch_observation` (`:3468-3472`, `:3513-3516`); no later commit
+  touched it. No ruling or finding in any campaign concerns workload isolation
+  verification; gamecult-chat's foreign-workloads cut touched
+  `CREDENTIALS_DIRECTORY` reservation, not this check.
+- I6. `SystemdWorkloadObservation` is persisted by previous Idunns in both
+  MessagePack encodings, compact (positional) included
+  (`persisted_systemd_observations_decode_as_the_systemd_variant`,
+  `:9915-9930`), so deleting the `load_credential` field shifts every later
+  slot of a live incumbent's record. The field is also in the drift list
+  (`:2462`) that `stop` (`:4064-4070`) uses to refuse stopping a changed
+  workload: changing what the field holds without leaving that list would make
+  the next Idunn refuse to stop the live Odin and StreamPixels incumbents.
+- I7. Fixtures: the verbatim 257.13 `systemctl show` output of base's 27
+  properties for a Vili unit (presence identity only) and the Odin unit
+  (presence identity plus one service credential), captured 2026-10-24 into
+  this Imagination's scratchpad `fixtures/`.
+- I8. The new rule holds for every live systemd incumbent: inside each
+  MainPID root the delivered entries equal its binding's `secret_files` keys
+  minus `GAMECULT_RUNTIME_PRESENCE_IDENTITY` (Odin 1, StreamPixels service 11,
+  StreamPixels web 1; key names compared, no values read). An Idunn restart onto
+  this cut re-observes them (`:3955-3970`) and passes.
+
+### Decision
+
+A property systemd cannot display does not belong in the comparison. The
+invariant it reached for, that the workload holds exactly the credentials Idunn
+delivered, is observed where it is decided: the entries of
+`/run/credentials/<unit>` in the MainPID's root equal Idunn's service
+credential names, absent directory equal to none. That covers
+`LoadCredential`, `SetCredential`, `ImportCredential` and drop-ins alike, and
+needs no new program. The D-Bus read (I3) is exact but sees only one of those
+properties and adds a second systemd client and parser; rejected.
+
+### Authority
+
+Owner: `observe_unit` decides the credential rule, from the delivered
+directory. `LoadCredential`'s display form is no longer read or compared; the
+`load_credential` field is a retired wire slot, written empty, out of the drift
+list, kept only so persisted records decode. Forbidden writers: the
+`observe_unit` conjunct and the `validate_launch_observation` clause. Dropping
+the slot waits on the next schema revision of the observation.
+
+### Deploy
+
+Idunn's body changes only through gamecult-ops
+`scripts/install-idunn-yggdrasil-release.sh` (stage `idunn` only; unit and
+bindings unchanged). After it, Vili's Hands reruns from step 4 of
+`motion-pipeline:cut_report:cut-vili-admission.h4`'s undone list.
+
+## Cut `runner-failure-evidence`
+
+Spec `idunn-deploy-provenance:cut_spec:cut-runner-failure-evidence.r1`, at
+`804ed75` (origin/main, live on Yggdrasil). It serves the StreamPixels-area
+Self: `idunn up vili` (command `up-4866467c-ee98-4a2b-b4a9-a008d7c44acd`,
+transaction `tx-a04b68c7-c742-4d7d-8220-8aa08b9febae`, 2026-10-25 00:50-00:55
+UTC) failed in Sealing with `Docker runner exited with exit status: 1: ` and
+nobody could tell which recipe step failed.
+
+### Body facts (source read at `804ed75`; host read-only, 2026-10-25)
+
+- R1. The runner is `docker run --rm` (`src/drivers.rs:2088-2090`), launched by
+  `DockerRunnerDriver::docker` (`:2173-2195`) with stdin null and
+  `Command::output()`: stdout and stderr are each buffered whole in the
+  daemon's memory, unbounded. On a non-zero exit the error is
+  `"Docker runner exited with {status}: {stderr}"`; stdout is dropped. The
+  container is gone the moment it exits, so `docker ps -a` and `docker logs`
+  hold nothing.
+- R2. The empty message: the failing process wrote nothing to stderr, and
+  stdout, the only stream that carried its report, is discarded at
+  `:2182-2193`. The same interpolation echoes whatever a step does write to
+  stderr into `last_error`, `pre_fencing_abort.error` and
+  `completion.error` (`bounded_detail`, 2048 bytes), and from there into
+  `idunn up`'s `deployment failed:` line and `idunn status`.
+- R3. No step identity reaches the error. `materialize` (`:2368-2391`) loops
+  over `declaration.steps` and calls `run_in_workspace` with argv only;
+  `step.id` is used only in the environment precondition. External inputs
+  (`curl` in the runner, `:2247-2258`) fail through the same helper with the
+  same anonymous message.
+- R4. Which step failed on 2026-10-25, inferred from the failed transaction's
+  staging root, which survives (926 MiB,
+  `/var/lib/gamecult/idunn/staging/tx-a04b68c7-.../`): in `.runner-node-build`,
+  `node_modules/.package-lock.json` was written at 00:51:24 (so
+  `install-vili-dependencies`, `npm ci`, finished), nothing outside
+  `node_modules` is newer, and `release/vili` (the next step's output) is
+  absent; the python workspace holds only fetched inputs. The failing step was
+  `test-daemon` (`node --test ... scripts/test-vili-daemon.mjs`), whose test
+  reporter writes to stdout: exactly the stream R1 discards. Inference from
+  residue, not from output.
+- R5. The failure path: `resume_one_transaction` (`src/control_plane.rs:6291-6320`)
+  sends a pre-fence error to `begin_pre_fencing_abort` (`:9916-9935`), which
+  persists `bounded_detail(format!("{error:#}"))`. Nothing reaches the journal
+  on this path: the idunn-yggdrasil journal for 00:45-01:10 holds only
+  provider-exclusion lines.
+- R6. Stores. `control.cc` (root 0640) holds commands, transactions
+  (`idunn.deployment_transaction.v4`, keys 0-36), generations, supervision and
+  expiry requests; `ControlSnapshot::read` refuses any other type
+  (`:3420`). Records must re-encode byte-exactly (`decode_record`,
+  `:3717-3727`), so a new transaction field is a schema bump and a v4 lift
+  (`:4100-4110`). Terminal transactions move verbatim to `history.cc`
+  (`:4366-4396`; 39.8 MB on the host), which status reads for `--command`.
+  `incidents.cc` lives in the published projection directory and by its own
+  doc (`src/incident.rs:1-11`) carries no free text and is never `control.cc`,
+  because a type added there stops continuity for every target on a rollback.
+- R7. The daemon runs as root (`User=root`). Runner secrets reach the
+  container as read-only file mounts named by env (`:2041-2047`), so any text
+  a step derives from them can land in its output.
+- R8. Staging residue: 475 transaction roots, 55 GiB under
+  `/var/lib/gamecult/idunn/staging`; a failed Sealing leaves its workspaces
+  because `materialize` returns before `remove_tree_inside`.
+
+### Decision
+
+The failure becomes a typed value at the runner, `RunnerStepFailure`: the
+subject (recipe step or external input, by id), the runner id, the exit code or
+signal, and the last 8 KiB of stdout and of stderr with each stream's total
+byte count. Its `Display` carries the subject and status, never output.
+Capture streams both pipes through bounded tails, so a verbose build no longer
+buffers whole in the daemon.
+
+The tail lives in a new private store, `runner-evidence.cc`, beside
+`control.cc` (mode 0600), one record per transaction
+(`idunn.runner_failure.v1`), newest 32 kept by the writer. Not in the
+transaction record: that is a v5 bump whose rollback makes every older Idunn
+refuse its control store (R6), the scar `incident.rs` already names. Not in
+`incidents.cc`: published, and free text is barred there. Not in the journal:
+journal readers are a wider set than root, so the journal line names the
+transaction, the subject, the status and the store, and no output.
+
+No redaction. Matching secret bytes would make the daemon read every secret,
+and still misses any encoding, prefix or transform of one; a claimed redaction
+that leaks is worse than a stated rule. The rule instead is the reader set: the
+store's only readers (root) can already read every secret the runner was
+given.
+
+Keeping the failed container (dropping `--rm`) is worse: its output then lives
+in Docker's json-file log, unbounded, untyped, outside Idunn's state and
+readable by the docker group; containers accumulate with no owner to reap them;
+`docker system prune` erases the evidence; and nothing ties a container to the
+transaction in `idunn status`.
+
+### Authority
+
+Owner: the runner decides what failed and keeps the bounded tail; the engine's
+pre-fence failure path is the one writer of `runner-evidence.cc`.
+`last_error`, `pre_fencing_abort.error` and `completion.error` keep only the
+failure's `Display`. Forbidden writers: the `stderr` interpolation in
+`DockerRunnerDriver::docker`. Retention is by count inside the writer, so the
+store needs no external owner; staging roots (R8) stay with
+`ghostlight-verse:follow_up:idunn-generation-retention`.
+
+### Deploy
+
+Idunn's body changes only through gamecult-ops
+`scripts/install-idunn-yggdrasil-release.sh` (stage `idunn` only; unit and
+bindings unchanged). The next failed Sealing then shows its step and tail on
+`sudo idunn status --command <id>`.

src/control_plane.rs

@@ -42,11 +42,14 @@ use crate::drivers::{
     ChallengeFailure, CultCacheTopologyDriver, CultCacheWriteLeaseDriver, DockerRunnerDriver, FrozenSourceReceipt,
     GitSourceDriver, InstalledReleaseObservation, IsolationEvidence, NginxRouteDriver,
     LockContended, ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators,
-    RouteObservation, RoutePreflightReceipt, RunnerPort, SourcePort,
+    RouteObservation, RoutePreflightReceipt, RunnerPort, RunnerStepFailure, SourcePort,
     SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort,
     WriteLeasePort,
 };
 use crate::incident::{CloseReason, IncidentCondition, IncidentStore, render_incidents};
+use crate::runner_evidence::{
+    self, Kept, RunnerFailureRecord, runner_evidence_store_path, unreadable_store_path,
+};
 use crate::host_actuator::{
     HostActuatorAccess, HostActuatorHub, HostActuatorRunnerDriver, HostActuatorWorkloadDriver,
     HostUnobservable, IdunnHostActuatorIdentity, SharedHostActuatorHub, spawn_hub_service,
@@ -2931,6 +2934,10 @@ enum Command {
         recipe: PathBuf,
         binding: Option<PathBuf>,
     },
+    #[cfg(unix)]
+    OrdersServe(crate::steward_orders::ServeOptions),
+    #[cfg(unix)]
+    Order(crate::steward_orders::OrderCommand),
 }
 
 pub fn run(args: impl Iterator<Item = String>) -> Result<()> {
@@ -2970,6 +2977,10 @@ pub fn run(args: impl Iterator<Item = String>) -> Result<()> {
             requested_by,
         } => expire(&state_store, &command_id, &requested_by),
         Command::Validate { recipe, binding } => validate(&recipe, binding.as_deref()),
+        #[cfg(unix)]
+        Command::OrdersServe(options) => crate::steward_orders::serve(options),
+        #[cfg(unix)]
+        Command::Order(order) => crate::steward_orders::run_order(order),
     }
 }
 
@@ -3022,6 +3033,10 @@ fn parse(args: impl Iterator<Item = String>) -> Result<Command> {
         "cancel" => parse_cancel(args),
         "expire" => parse_expire(args),
         "validate" => parse_validate(args),
+        #[cfg(unix)]
+        "orders" => crate::steward_orders::parse_serve(args).map(Command::OrdersServe),
+        #[cfg(unix)]
+        "order" => crate::steward_orders::parse_order(args).map(Command::Order),
         "--help" | "-h" | "help" => bail!(usage()),
         _ => bail!("unknown Idunn command {command:?}\n\n{}", usage()),
     }
@@ -4727,20 +4742,21 @@ fn resolve_selector(
     Ok(targets)
 }
 
-fn submit(
+/// The one commit primitive for a deployment command: `idunn up` and a steward
+/// deploy order both write theirs here. Returns the commit time.
+pub(crate) fn commit_deployment_command(
     store_path: &Path,
+    command_id: &str,
     selector: &str,
     requested_by: &str,
-    wait: bool,
-    timeout_seconds: u64,
-) -> Result<()> {
+) -> Result<u64> {
     if let Some(parent) = store_path.parent() {
         fs::create_dir_all(parent)?;
     }
     let now = now_millis()?;
     let command = DeploymentCommand {
         schema_version: DEPLOYMENT_COMMAND_SCHEMA.into(),
-        command_id: format!("up-{}", Uuid::new_v4()),
+        command_id: command_id.into(),
         kind: CommandKind::Deploy,
         selector: selector.into(),
         requested_by: requested_by.into(),
@@ -4758,7 +4774,33 @@ fn submit(
         )?,
         "deployment command id collided"
     );
-    println!("{}", command.command_id);
+    Ok(now)
+}
+
+/// A command's present state as `idunn status --command` derives it, for the
+/// steward order poster: `queued` and `running` are not terminal.
+#[cfg(unix)]
+pub(crate) fn command_state(
+    state_store: &Path,
+    command_id: &str,
+) -> Result<(&'static str, String)> {
+    let snapshot = ControlSnapshot::read(state_store)?;
+    let archived = read_history_transactions(state_store);
+    let mut transactions = snapshot.transaction_for_command(command_id);
+    transactions.extend(archived.iter().filter(|value| value.command_id == command_id));
+    Ok(derived_command_status(&transactions))
+}
+
+fn submit(
+    store_path: &Path,
+    selector: &str,
+    requested_by: &str,
+    wait: bool,
+    timeout_seconds: u64,
+) -> Result<()> {
+    let command_id = format!("up-{}", Uuid::new_v4());
+    let now = commit_deployment_command(store_path, &command_id, selector, requested_by)?;
+    println!("{command_id}");
     if !wait {
         return Ok(());
     }
@@ -4769,10 +4811,10 @@ fn submit(
             snapshot
                 .commands
                 .iter()
-                .any(|stored| stored.value.command_id == command.command_id),
+                .any(|stored| stored.value.command_id == command_id),
             "submitted deployment command disappeared"
         );
-        let transactions = snapshot.transaction_for_command(&command.command_id);
+        let transactions = snapshot.transaction_for_command(&command_id);
         if !transactions.is_empty() && transactions.iter().all(|value| value.is_terminal()) {
             if let Some(error) =
                 transactions
@@ -4797,7 +4839,7 @@ fn submit(
     }
 }
 
-fn status(
+pub(crate) fn status(
     out: &mut impl Write,
     store_path: &Path,
     incident_store: &Path,
@@ -4877,6 +4919,13 @@ fn status(
                 }
                 if let Some(abort) = &transaction.pre_fencing_abort {
                     writeln!(out, "    pre-fencing abort {abort:?}")?;
+                    // The step that failed and its output tail live in their
+                    // own private store. An unreadable store costs its lines.
+                    runner_evidence::show_status(
+                        out,
+                        &runner_evidence_store_path(store_path),
+                        &transaction.transaction_id,
+                    )?;
                 }
                 if let Some(abort) = &transaction.post_fencing_abort {
                     writeln!(out, "    post-fencing abort {abort:?}")?;
@@ -5374,6 +5423,32 @@ impl Engine {
         IncidentStore::new(&self.options.incident_store)
     }
 
+    /// A failed runner step keeps its output tail in the runner evidence
+    /// store and says so on the journal, naming no output. The store write is
+    /// best effort: its failure is named in the line and never stops the
+    /// abort that follows.
+    fn keep_runner_failure(
+        &self,
+        transaction: &DeploymentTransaction,
+        error: &anyhow::Error,
+    ) -> Option<String> {
+        let failure = error.downcast_ref::<RunnerStepFailure>()?;
+        let path = runner_evidence_store_path(&self.options.state_store);
+        let recorded = now_millis().and_then(|now| {
+            runner_evidence::record(
+                &path,
+                &RunnerFailureRecord::from_failure(
+                    &transaction.transaction_id,
+                    &transaction.command_id,
+                    &transaction.target,
+                    failure,
+                    now,
+                ),
+            )
+        });
+        Some(runner_failure_journal_line(transaction, failure, &path, &recorded))
+    }
+
     /// Settle the outcome of one incident operation. A failure never leaves
     /// this function: it is traced once per episode, under the operation and
     /// the subject it was decided for (`*` for a pass over the whole store),
@@ -6315,6 +6390,9 @@ impl Engine {
                 if latest.value.pre_fencing_abort.is_none()
                     && error.downcast_ref::<LockContended>().is_none()
                 {
+                    if let Some(line) = self.keep_runner_failure(&latest.value, &error) {
+                        eprintln!("{line}");
+                    }
                     self.begin_pre_fencing_abort(latest, error)?;
                 } else {
                     self.record_resumable_error(latest, &error)?;
@@ -10553,7 +10631,13 @@ fn required<'a, T>(value: &'a Option<T>, label: &str) -> Result<&'a T> {
     value.as_ref().with_context(|| format!("{label} is absent"))
 }
 
-fn now_millis() -> Result<u64> {
+/// The control store `idunn up` and `idunn serve` use unless told otherwise.
+#[cfg(unix)]
+pub(crate) fn default_state_store() -> PathBuf {
+    RuntimeOptions::default().state_store
+}
+
+pub(crate) fn now_millis() -> Result<u64> {
     Ok(SystemTime::now()
         .duration_since(UNIX_EPOCH)
         .context("system clock predates Unix epoch")?
@@ -10573,12 +10657,12 @@ fn sibling_path(path: &Path, suffix: &str) -> PathBuf {
     PathBuf::from(value)
 }
 
-fn string_value(args: &mut impl Iterator<Item = String>, name: &str) -> Result<String> {
+pub(crate) fn string_value(args: &mut impl Iterator<Item = String>, name: &str) -> Result<String> {
     args.next()
         .ok_or_else(|| anyhow!("{name} requires a value"))
 }
 
-fn path_value(args: &mut impl Iterator<Item = String>, name: &str) -> Result<PathBuf> {
+pub(crate) fn path_value(args: &mut impl Iterator<Item = String>, name: &str) -> Result<PathBuf> {
     Ok(PathBuf::from(string_value(args, name)?))
 }
 
@@ -10594,7 +10678,7 @@ fn u64_value(args: &mut impl Iterator<Item = String>, name: &str) -> Result<u64>
         .with_context(|| format!("{name} requires a u64"))
 }
 
-fn require_selector(value: &str) -> Result<()> {
+pub(crate) fn require_selector(value: &str) -> Result<()> {
     let value = value.strip_prefix("profile:").unwrap_or(value);
     require_id(value, "deployment selector")
 }
@@ -10641,6 +10725,43 @@ const DETAIL_EMPTY: &str = "(no detail)";
 /// only judge, and this output always satisfies it. NUL is dropped, empty
 /// input gets a fixed placeholder, and an over-long value keeps its head and
 /// its tail (where a build log's error is) cut on char boundaries.
+/// The journal line for a failed runner step: the transaction, the subject and
+/// status (the failure's `Display`), and where the evidence is. Never output.
+fn runner_failure_journal_line(
+    transaction: &DeploymentTransaction,
+    failure: &RunnerStepFailure,
+    store_path: &Path,
+    recorded: &Result<Kept>,
+) -> String {
+    let head = format!(
+        "Idunn transaction {} (command {}, target {}) {failure}",
+        transaction.transaction_id, transaction.command_id, transaction.target
+    );
+    match recorded {
+        Ok(kept) => {
+            let aside = match kept {
+                Kept::Stored => String::new(),
+                Kept::SetAside { replaced } => format!(
+                    "; an undecodable evidence store was set aside as {}{}",
+                    unreadable_store_path(store_path).display(),
+                    match replaced {
+                        Some(last_written) => format!(
+                            ", replacing an earlier set-aside store last written at unix millis {last_written}"
+                        ),
+                        None => String::new(),
+                    }
+                ),
+            };
+            format!(
+                "{head}; evidence in {}{aside}; read it with idunn status --command {}",
+                store_path.display(),
+                transaction.command_id
+            )
+        }
+        Err(error) => format!("{head}; evidence was not kept: {error:#}"),
+    }
+}
+
 fn bounded_detail(value: &str) -> String {
     let value = value.replace('', "");
     if value.is_empty() {
@@ -10681,7 +10802,10 @@ fn usage() -> &'static str {
      idunn status [--state-store PATH] [--incident-store PATH] [--command ID]\n\
      idunn cancel <command-id> [--state-store PATH]\n\
      idunn expire <command-id> [--state-store PATH]\n\
-     idunn validate --recipe PATH [--binding PATH]\n\n\
+     idunn validate --recipe PATH [--binding PATH]\n\
+     idunn orders serve [--store DIR] [--steward-binding PATH] [--socket PATH]\n\
+     idunn order <package-install PKG... | unit VERB UNIT | deploy SELECTOR> --reason TEXT\n\
+     idunn order show ID | idunn order list\n\n\
      Recipes describe capability and process requirements. Idunn seals exact\n\
      source and artifacts, admits one incarnation, and delegates execution to\n\
      systemd and routing mechanics to the configured proxy."
@@ -11893,9 +12017,45 @@ mod tests {
             vec!["serve", "--swarm-profile", "yggdrasil-local"],
             vec!["serve", "--restart-command", "bad"],
             vec!["validate-runtime-admission"],
+            vec!["order", "unit", "restart", "nginx.service", "--reason", "r", "--command", "sh -c bad"],
+            vec!["order", "unit", "restart", "nginx.service", "--reason", "r", "--shell"],
+            vec!["order", "run", "sh", "--reason", "r"],
+            vec!["order", "package-install", "git-lfs", "--reason", "r", "--deploy-command", "x"],
+            vec!["order", "unit", "restart", "nginx.service", "extra", "--reason", "r"],
+            vec!["order", "deploy", "vili", "--reason", "r", "--requested-by", "someone"],
+            vec!["orders", "serve", "--allow-units", "nginx.service"],
+            vec!["orders", "run"],
         ] {
             assert!(parse(arguments.into_iter().map(str::to_owned)).is_err());
         }
+        // The declarative order forms parse, and only as order commands.
+        #[cfg(unix)]
+        for arguments in [
+            vec!["order", "unit", "restart", "nginx.service", "--reason", "r"],
+            vec!["order", "package-install", "git-lfs", "--reason", "r"],
+            vec!["order", "deploy", "vili", "--reason", "r"],
+            vec!["order", "list"],
+            vec!["orders", "serve"],
+        ] {
+            let parsed = parse(arguments.into_iter().map(str::to_owned)).unwrap();
+            assert!(matches!(parsed, Command::Order(_) | Command::OrdersServe(_)));
+        }
+    }
+
+    #[test]
+    fn deployment_commit_primitive_is_shared_and_never_overwrites() -> Result<()> {
+        let temp = TempDir::new()?;
+        let store = temp.path().join("control.cc");
+        commit_deployment_command(&store, "up-one", "vili", "steward:eureka order ord-1")?;
+        let snapshot = ControlSnapshot::read(&store)?;
+        assert_eq!(snapshot.commands.len(), 1);
+        let command = &snapshot.commands[0].value;
+        assert_eq!(command.command_id, "up-one");
+        assert_eq!(command.selector, "vili");
+        assert_eq!(command.requested_by, "steward:eureka order ord-1");
+        assert!(commit_deployment_command(&store, "up-one", "other", "x").is_err());
+        assert_eq!(ControlSnapshot::read(&store)?.commands.len(), 1);
+        Ok(())
     }
 
     #[test]
@@ -24657,4 +24817,435 @@ GAMECULT_RUNTIME_PRESENCE_IDENTITY = "/etc/gamecult/service/runtime-presence-ide
             Ok(())
         }
     }
+
+    // BEGIN runner-failure-evidence tests
+    // A failed runner step is named, its output tail is kept in
+    // runner-evidence.cc, and no output reaches an error, control.cc or the
+    // journal line. The runner is the real DockerRunnerDriver driving a fake
+    // docker program, through the engine's Sealing path.
+
+    const RUNNER_CANARY: &str = "CANARY-7c41e0d9";
+
+    /// A source that freezes a fixed tree, so a deployment reaches the runner.
+    struct TreeSource {
+        recipe: String,
+        root: PathBuf,
+    }
+
+    impl SourcePort for TreeSource {
+        fn resolve(
+            &self,
+            _: &OperatorBinding,
+            _: &str,
+            _: u64,
+        ) -> Result<crate::drivers::ResolvedSource> {
+            Ok(crate::drivers::ResolvedSource {
+                facts: crate::deployment_plan::tests::source(&self.recipe),
+                recipe_bytes: self.recipe.clone().into_bytes(),
+            })
+        }
+
+        fn freeze(&self, transaction_id: &str, plan: &CompiledDeploymentPlan) -> Result<FrozenSourceReceipt> {
+            Ok(FrozenSourceReceipt {
+                transaction_id: transaction_id.to_owned(),
+                plan_id: plan.plan_id.clone(),
+                snapshot_sha256: format!("sha256-{}", "a".repeat(64)),
+            })
+        }
+
+        fn observe_frozen(
+            &self,
+            plan: &CompiledDeploymentPlan,
+            receipt: &FrozenSourceReceipt,
+        ) -> Result<crate::drivers::FrozenSource> {
+            Ok(crate::drivers::FrozenSource::fixture(
+                receipt.clone(),
+                plan.source.clone(),
+                plan.recipe_blob.clone(),
+                self.root.clone(),
+            ))
+        }
+
+        fn cleanup(&self, _: &str, _: Option<&FrozenSourceReceipt>) -> Result<()> {
+            Ok(())
+        }
+    }
+
+    /// The unrouted recipe with two steps ahead of its build, and no external
+    /// input: the second step, `test-daemon`, is the one the fake docker fails.
+    fn two_step_recipe() -> String {
+        let recipe = unrouted_recipe().replace(
+            "capability = \"service.runtime\"",
+            &format!("capability = \"{ODIN_RENDEZVOUS_CAPABILITY}\""),
+        );
+        let (head, rest) = recipe
+            .split_once("[[external_inputs]]")
+            .expect("the recipe declares an external input");
+        let (_, tail) = rest.split_once("[[artifacts]]").expect("the recipe declares artifacts");
+        let head = head.replacen(
+            "[[steps]]\nid = \"build\"",
+            "[[steps]]\nid = \"install\"\nphase = \"prepare\"\nrunner = \"rust\"\nargv = [\"cargo\", \"fetch\"]\n\n\
+             [[steps]]\nid = \"test-daemon\"\nphase = \"test\"\nrunner = \"rust\"\nargv = [\"cargo\", \"test\"]\n\n\
+             [[steps]]\nid = \"build\"",
+            1,
+        );
+        format!("{head}[[artifacts]]{tail}")
+    }
+
+    /// A Sealing world whose runner is the real driver over a fake docker: the
+    /// first call succeeds, the second prints the canary and an escape
+    /// sequence on both streams and exits 3.
+    #[cfg(unix)]
+    fn failing_runner_world() -> Result<(EngineFixture, TempDir)> {
+        use std::os::unix::fs::PermissionsExt;
+        let tools = TempDir::new()?;
+        let docker = tools.path().join("docker");
+        std::fs::write(
+            &docker,
+            format!(
+                "#!/bin/sh\nn=$(cat \"$0.count\" 2>/dev/null || echo 0)\nn=$((n+1))\necho $n > \"$0.count\"\n\
+                 [ \"$n\" -lt 2 ] && exit 0\n\
+                 printf 'out {RUNNER_CANARY} \\033[2J\\n'\nprintf 'err {RUNNER_CANARY}\\n' >&2\nexit 3\n"
+            ),
+        )?;
+        std::fs::set_permissions(&docker, std::fs::Permissions::from_mode(0o755))?;
+        let recipe = two_step_recipe();
+        let mut world = sealing_world(&recipe)?;
+        let tree = tools.path().join("tree");
+        std::fs::create_dir_all(&tree)?;
+        std::fs::write(tree.join("README"), b"source")?;
+        world.engine.source = Arc::new(TreeSource { recipe, root: tree });
+        world.engine.docker_runner = DockerRunnerDriver { docker_program: docker };
+        Ok((world, tools))
+    }
+
+    /// Scheduler ticks until the deployment ends (or the budget is spent).
+    fn tick_until_ended(world: &EngineFixture) -> Result<DeploymentTransaction> {
+        for _ in 0..24 {
+            world.engine.resume_backoff.lock().unwrap().clear();
+            let _ = world.engine.run_scheduler_tick();
+            if latest(world)?.completion.is_some() {
+                break;
+            }
+        }
+        latest(world)
+    }
+
+    fn printed_status(world: &EngineFixture) -> Result<String> {
+        let mut printed = Vec::new();
+        status(&mut printed, &world.state_store, &world.engine.options.incident_store, Some("up-service"))?;
+        Ok(String::from_utf8(printed)?)
+    }
+
+    fn canary_failure(stdout: &[u8]) -> RunnerStepFailure {
+        RunnerStepFailure {
+            subject: crate::drivers::RunnerSubject::Step("test-daemon".into()),
+            runner: "rust".into(),
+            exit_code: Some(3),
+            signal: None,
+            stdout_tail: stdout.to_vec(),
+            stdout_total_bytes: stdout.len() as u64,
+            stderr_tail: stdout.to_vec(),
+            stderr_total_bytes: stdout.len() as u64,
+        }
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn a_failed_runner_step_is_named_kept_and_never_echoed() -> Result<()> {
+        use std::os::unix::fs::PermissionsExt;
+        let (world, _tools) = failing_runner_world()?;
+        let ended = tick_until_ended(&world)?;
+        let Some(TransactionCompletion::FailedBeforeFencing { error }) = &ended.completion else {
+            bail!("the deployment did not fail before fencing: {:?} {:?}", ended.phase, ended.last_error);
+        };
+        // Named: the second step, on its runner, with the exit status.
+        assert!(
+            error.contains("runner step test-daemon on runner rust exited with exit status 3"),
+            "{error}"
+        );
+        // Never echoed: no error text, no control.cc byte, carries the output.
+        let abort = ended.pre_fencing_abort.as_ref().context("no pre-fencing abort")?;
+        for text in [error.as_str(), abort.error.as_str(), &format!("{ended:?}")] {
+            assert!(!text.contains(RUNNER_CANARY), "runner output reached a record: {text}");
+        }
+        // The terminal transaction has moved to history.cc: both stores.
+        for store in [world.state_store.clone(), history_store_path(&world.state_store)] {
+            let bytes = std::fs::read(&store)?;
+            assert!(!bytes.is_empty(), "{} is empty", store.display());
+            assert!(
+                !bytes.windows(RUNNER_CANARY.len()).any(|window| window == RUNNER_CANARY.as_bytes()),
+                "runner output reached {}",
+                store.display()
+            );
+        }
+        // Kept: in the private store, tails of both streams, and on status.
+        let evidence = runner_evidence_store_path(&world.state_store);
+        assert_eq!(std::fs::metadata(&evidence)?.permissions().mode() & 0o777, 0o600);
+        let record = runner_evidence::for_transaction(&evidence, &ended.transaction_id)?
+            .context("no evidence for the failed transaction")?;
+        assert_eq!(record.subject_id, "test-daemon");
+        assert_eq!(record.exit_code, Some(3));
+        assert!(String::from_utf8_lossy(&record.stdout_tail).contains(RUNNER_CANARY));
+        assert!(String::from_utf8_lossy(&record.stderr_tail).contains(RUNNER_CANARY));
+        let printed = printed_status(&world)?;
+        assert!(printed.contains("failed step test-daemon"), "{printed}");
+        assert!(printed.contains(RUNNER_CANARY), "{printed}");
+        assert!(!printed.contains('\u{1b}'), "status printed a raw escape byte");
+        Ok(())
+    }
+
+    #[test]
+    fn the_journal_line_names_the_store_and_never_the_output() -> Result<()> {
+        let world = EngineFixture::new()?;
+        let (transaction, _) = terminal_transaction_with_command("ghostlight")?;
+        let failure = canary_failure(RUNNER_CANARY.as_bytes());
+        let line = world
+            .engine
+            .keep_runner_failure(&transaction, &anyhow::Error::new(failure.clone()).context("sealing"))
+            .context("a runner failure produced no journal line")?;
+        assert!(!line.contains(RUNNER_CANARY), "{line}");
+        for needle in [
+            transaction.transaction_id.as_str(),
+            "up-ghostlight",
+            "ghostlight",
+            "runner step test-daemon on runner rust exited with exit status 3",
+            "runner-evidence.cc",
+            "idunn status --command up-ghostlight",
+        ] {
+            assert!(line.contains(needle), "{needle} missing from {line}");
+        }
+        // Not a runner failure: nothing is kept and nothing is said.
+        assert!(
+            world
+                .engine
+                .keep_runner_failure(&transaction, &anyhow::anyhow!("a different failure"))
+                .is_none()
+        );
+        // The line a failed write produces says so, with no output.
+        let unwritable = runner_failure_journal_line(
+            &transaction,
+            &failure,
+            Path::new("/var/lib/gamecult/idunn/runner-evidence.cc"),
+            &Err(anyhow::anyhow!("disk full")),
+        );
+        assert!(unwritable.contains("evidence was not kept: disk full"), "{unwritable}");
+        assert!(!unwritable.contains(RUNNER_CANARY));
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn a_failed_evidence_write_does_not_stop_the_abort() -> Result<()> {
+        let (world, _tools) = failing_runner_world()?;
+        // A directory where the store should be: every write to it fails.
+        std::fs::create_dir(runner_evidence_store_path(&world.state_store))?;
+        let ended = tick_until_ended(&world)?;
+        let Some(TransactionCompletion::FailedBeforeFencing { error }) = &ended.completion else {
+            bail!("the unwritable store stopped the abort: {:?} {:?}", ended.phase, ended.last_error);
+        };
+        assert!(error.contains("runner step test-daemon on runner rust"), "{error}");
+        let line = world
+            .engine
+            .keep_runner_failure(&ended, &anyhow::Error::new(canary_failure(b"")))
+            .context("no journal line")?;
+        assert!(line.contains("evidence was not kept"), "{line}");
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn an_undecodable_evidence_store_does_not_stop_the_next_failure_keeping_its_evidence() -> Result<()> {
+        let damage = format!("\u{c1} not a store {RUNNER_CANARY}-in-the-damage");
+        let (world, _tools) = failing_runner_world()?;
+        let evidence = runner_evidence_store_path(&world.state_store);
+        std::fs::write(&evidence, damage.as_bytes())?;
+        let ended = tick_until_ended(&world)?;
+        // The transaction's own outcome is the runner failure, not the store.
+        let Some(TransactionCompletion::FailedBeforeFencing { error }) = &ended.completion else {
+            bail!("the damaged store changed the outcome: {:?} {:?}", ended.phase, ended.last_error);
+        };
+        assert!(error.contains("runner step test-daemon on runner rust"), "{error}");
+        assert!(!error.contains("damage"), "{error}");
+        // The damage is kept aside; the failure's evidence is in a fresh store.
+        assert_eq!(std::fs::read(unreadable_store_path(&evidence))?, damage.as_bytes());
+        let record = runner_evidence::for_transaction(&evidence, &ended.transaction_id)?
+            .context("the failure kept no evidence beside a damaged store")?;
+        assert_eq!(record.subject_id, "test-daemon");
+        assert!(printed_status(&world)?.contains("failed step test-daemon"));
+        // The journal says it, naming the file and none of its content.
+        std::fs::write(&evidence, damage.as_bytes())?;
+        let line = world
+            .engine
+            .keep_runner_failure(&ended, &anyhow::Error::new(canary_failure(b"x")))
+            .context("no journal line")?;
+        assert!(line.contains("an undecodable evidence store was set aside as"), "{line}");
+        assert!(line.contains("runner-evidence.unreadable.cc"), "{line}");
+        assert!(!line.contains("damage") && !line.contains("not a store"), "{line}");
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn an_unreadable_evidence_store_costs_only_its_lines() -> Result<()> {
+        let (world, _tools) = failing_runner_world()?;
+        let ended = tick_until_ended(&world)?;
+        std::fs::write(runner_evidence_store_path(&world.state_store), b"\xc1not a store")?;
+        let printed = printed_status(&world)?;
+        assert!(printed.contains(&format!("transaction {}", ended.transaction_id)), "{printed}");
+        assert!(printed.contains("target service phase"), "{printed}");
+        assert!(printed.contains("pre-fencing abort"), "{printed}");
+        assert_eq!(printed.matches("runner evidence unreadable").count(), 1, "{printed}");
+        Ok(())
+    }
+
+    /// Everything status prints from the evidence store goes through one
+    /// allowlist, so a hostile build log reaches the terminal as visible
+    /// escapes whatever its class.
+    #[cfg(unix)]
+    #[test]
+    fn status_prints_hostile_runner_output_as_visible_escapes_only() -> Result<()> {
+        let (world, _tools) = failing_runner_world()?;
+        let ended = tick_until_ended(&world)?;
+        let hostile: Vec<(&str, &str)> = vec![
+            ("\u{1b}[2J", "\\u{1b}[2J"),
+            ("\u{7}", "\\u{7}"),
+            ("\r", "\\u{d}"),
+            ("\u{202e}", "\\u{202e}"),
+            ("\u{2066}", "\\u{2066}"),
+            ("\u{85}", "\\u{85}"),
+            ("\u{e0041}", "\\u{e0041}"),
+            ("\u{200b}", "\\u{200b}"),
+            ("\u{feff}", "\\u{feff}"),
+        ];
+        let mut tail: Vec<u8> = Vec::new();
+        for &(raw, _) in &hostile {
+            tail.extend_from_slice(b"a");
+            tail.extend_from_slice(raw.as_bytes());
+            tail.extend_from_slice(b"b\n");
+        }
+        tail.extend_from_slice(b"bad \xff\xc1 bytes\n");
+        runner_evidence::record(
+            &runner_evidence_store_path(&world.state_store),
+            &RunnerFailureRecord::from_failure(
+                &ended.transaction_id,
+                &ended.command_id,
+                &ended.target,
+                &canary_failure(&tail),
+                9,
+            ),
+        )?;
+        let printed = printed_status(&world)?;
+        for &(raw, escaped) in &hostile {
+            assert!(!printed.contains(raw), "{escaped} reached the terminal raw");
+            assert!(printed.contains(&format!("a{escaped}b")), "{escaped} missing: {printed}");
+        }
+        assert!(printed.contains("bad \\xff\\xc1 bytes"), "{printed}");
+        let section = printed.find("    failed step").context("no evidence section")?;
+        assert!(printed[section..].is_ascii(), "non-ASCII reached the terminal: {printed}");
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn status_prints_the_unreadable_store_line_escaped() -> Result<()> {
+        let (world, _tools) = failing_runner_world()?;
+        tick_until_ended(&world)?;
+        let evidence = runner_evidence_store_path(&world.state_store);
+        // The decode error of this store quotes the stored subject kind.
+        let mut bytes = std::fs::read(&evidence)?;
+        let at = bytes
+            .windows(4)
+            .position(|window| window == b"step")
+            .context("no step marker in the evidence store")?;
+        bytes[at..at + 4].copy_from_slice(b"\x1b[2J");
+        std::fs::write(&evidence, bytes)?;
+        let printed = printed_status(&world)?;
+        assert_eq!(printed.matches("runner evidence unreadable").count(), 1, "{printed}");
+        assert!(!printed.contains('\u{1b}'), "an escape byte reached the terminal");
+        assert!(printed.contains("\\u{1b}[2J"), "{printed}");
+        Ok(())
+    }
+
+    #[test]
+    fn the_journal_line_names_a_displaced_set_aside_store_by_time_only() -> Result<()> {
+        let (transaction, _) = terminal_transaction_with_command("ghostlight")?;
+        let failure = canary_failure(RUNNER_CANARY.as_bytes());
+        let path = Path::new("/var/lib/gamecult/idunn/runner-evidence.cc");
+        let first = runner_failure_journal_line(
+            &transaction,
+            &failure,
+            path,
+            &Ok(Kept::SetAside { replaced: None }),
+        );
+        assert!(first.contains("was set aside as"), "{first}");
+        assert!(!first.contains("replacing"), "{first}");
+        let second = runner_failure_journal_line(
+            &transaction,
+            &failure,
+            path,
+            &Ok(Kept::SetAside { replaced: Some(1234) }),
+        );
+        assert!(
+            second.contains("replacing an earlier set-aside store last written at unix millis 1234"),
+            "{second}"
+        );
+        assert!(!second.contains(RUNNER_CANARY));
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn control_store_types_are_unchanged_by_a_runner_failure() -> Result<()> {
+        let (world, _tools) = failing_runner_world()?;
+        tick_until_ended(&world)?;
+        // The store a rolled-back Idunn reads: it refuses any foreign type.
+        ControlSnapshot::read(&world.state_store)?;
+        let mut transactions = 0;
+        for store in [world.state_store.clone(), history_store_path(&world.state_store)] {
+            for envelope in SingleFileMessagePackBackingStore::new(&store).pull_all_read_only_snapshot()? {
+                assert!(!envelope.r#type.contains("runner_failure"), "{}", envelope.r#type);
+                if envelope.r#type == DeploymentTransaction::TYPE {
+                    transactions += 1;
+                    assert_eq!(envelope.schema_id.as_deref(), Some("idunn.deployment_transaction.v4"));
+                }
+            }
+        }
+        assert_eq!(transactions, 1, "the failed transaction is in neither store");
+        Ok(())
+    }
+
+    /// The records control.cc holds re-encode byte for byte. The hex was
+    /// captured at base 804ed75, before this cut touched anything: a runner
+    /// failure adds no field, no type and no schema to the control store.
+    #[test]
+    fn control_records_encode_exactly_as_they_did_at_base() -> Result<()> {
+        let command = DeploymentCommand {
+            schema_version: DEPLOYMENT_COMMAND_SCHEMA.into(),
+            command_id: "up-service".into(),
+            kind: CommandKind::Deploy,
+            selector: "service".into(),
+            requested_by: "test".into(),
+            requested_at_unix_millis: 100,
+        };
+        let mut transaction = DeploymentTransaction::new(&command, "service".into(), 0, None, 100)?;
+        transaction.transaction_id = "tx-pinned".into();
+        transaction.last_error = Some("runner step test-daemon on runner rust exited with exit status 3".into());
+        transaction.pre_fencing_abort = Some(PreFencingAbort {
+            error: "runner step test-daemon on runner rust exited with exit status 3".into(),
+            candidate_cleanup: CleanupEvidence::Skipped,
+            topology_reconciliation: CleanupEvidence::Skipped,
+            source_cleanup: CleanupEvidence::Complete,
+        });
+        let hex = |bytes: &[u8]| bytes.iter().map(|byte| format!("{byte:02x}")).collect::<String>();
+        let command_hex = hex(&command_envelope(&command, 100)?.payload);
+        let transaction_hex = hex(&transaction_envelope(&transaction, 100)?.payload);
+        assert_eq!(command_hex, PINNED_COMMAND_HEX, "command bytes changed");
+        assert_eq!(transaction_hex, PINNED_TRANSACTION_HEX, "transaction bytes changed");
+        Ok(())
+    }
+
+    const PINNED_COMMAND_HEX: &str = "96bb6964756e6e2e6465706c6f796d656e745f636f6d6d616e642e7632aa75702d73657276696365a66465706c6f79a773657276696365a47465737464";
+    const PINNED_TRANSACTION_HEX: &str = "dc0025bf6964756e6e2e6465706c6f796d656e745f7472616e73616374696f6e2e7634a974782d70696e6e6564aa75702d73657276696365a66465706c6f79a77365727669636500a77365616c696e676464c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c000d94072756e6e6572207374657020746573742d6461656d6f6e206f6e2072756e6e657220727573742065786974656420776974682065786974207374617475732033c094d94072756e6e6572207374657020746573742d6461656d6f6e206f6e2072756e6e657220727573742065786974656420776974682065786974207374617475732033a7736b6970706564a7736b6970706564a8636f6d706c657465c0c0c0c0";
+    // END runner-failure-evidence tests
 }

src/drivers.rs

@@ -157,6 +157,24 @@ pub struct FrozenSourceReceipt {
     pub snapshot_sha256: String,
 }
 
+#[cfg(test)]
+impl FrozenSource {
+    /// A frozen source over an existing tree, for tests that drive a runner.
+    pub(crate) fn fixture(
+        receipt: FrozenSourceReceipt,
+        facts: SourceSelectionFacts,
+        recipe_bytes: Vec<u8>,
+        root: PathBuf,
+    ) -> Self {
+        Self {
+            receipt,
+            facts,
+            recipe_bytes,
+            root,
+        }
+    }
+}
+
 impl FrozenSourceReceipt {
     pub fn validate_against(&self, plan: &CompiledDeploymentPlan) -> Result<()> {
         plan.validate()?;
@@ -587,7 +605,9 @@ pub trait WorkloadPort {
     ///
     /// Idunn launches candidates as transient units with `Restart=no`, so a
     /// unit that has failed, or that systemd no longer knows, will not come
-    /// back on its own. Past the fence every error is otherwise treated as
+    /// back on its own. Those units are collected once they fail
+    /// (`INCARNATION_UNIT_LIFECYCLE`), so a failed candidate usually reads as
+    /// forgotten rather than failed. Past the fence every error is otherwise treated as
     /// resumable and retried forever, which is right while a candidate can
     /// still recover and a trap once it cannot.
     fn is_permanently_stopped(&self, observation: &WorkloadObservation) -> Result<bool>;
@@ -2152,6 +2172,115 @@ pub(crate) fn docker_run_args(
     Ok(args)
 }
 
+/// Bytes of each output stream that a failed runner step keeps. The daemon
+/// never holds more than this per stream while the step runs.
+pub(crate) const RUNNER_OUTPUT_TAIL_BYTES: usize = 8192;
+
+/// What a runner container was running: a recipe step or an external input,
+/// by the id the recipe gave it.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) enum RunnerSubject {
+    Step(String),
+    ExternalInput(String),
+}
+
+impl RunnerSubject {
+    pub(crate) fn id(&self) -> &str {
+        match self {
+            Self::Step(id) | Self::ExternalInput(id) => id,
+        }
+    }
+
+    fn noun(&self) -> &'static str {
+        match self {
+            Self::Step(_) => "step",
+            Self::ExternalInput(_) => "external input",
+        }
+    }
+}
+
+/// A runner container that did not exit successfully: what failed, how, and
+/// the last `RUNNER_OUTPUT_TAIL_BYTES` of each stream with its total size.
+/// The tails are evidence for `runner-evidence.cc`. `Display` and `Debug`
+/// carry none of them, so no error chain, record or log line renders output.
+#[derive(Clone, PartialEq, Eq)]
+pub(crate) struct RunnerStepFailure {
+    pub(crate) subject: RunnerSubject,
+    pub(crate) runner: String,
+    pub(crate) exit_code: Option<i32>,
+    pub(crate) signal: Option<i32>,
+    pub(crate) stdout_tail: Vec<u8>,
+    pub(crate) stdout_total_bytes: u64,
+    pub(crate) stderr_tail: Vec<u8>,
+    pub(crate) stderr_total_bytes: u64,
+}
+
+impl RunnerStepFailure {
+    /// The status as an operator reads it.
+    pub(crate) fn status(&self) -> String {
+        runner_status(self.exit_code, self.signal)
+    }
+}
+
+/// How a container ended, as an operator reads it.
+pub(crate) fn runner_status(exit_code: Option<i32>, signal: Option<i32>) -> String {
+    match (exit_code, signal) {
+        (Some(code), _) => format!("exit status {code}"),
+        (None, Some(signal)) => format!("signal {signal}"),
+        (None, None) => "an unknown status".to_owned(),
+    }
+}
+
+impl std::fmt::Display for RunnerStepFailure {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            formatter,
+            "runner {} {} on runner {} exited with {}; its output tail is kept in runner-evidence.cc",
+            self.subject.noun(),
+            self.subject.id(),
+            self.runner,
+            self.status()
+        )
+    }
+}
+
+impl std::fmt::Debug for RunnerStepFailure {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("RunnerStepFailure")
+            .field("subject", &self.subject)
+            .field("runner", &self.runner)
+            .field("exit_code", &self.exit_code)
+            .field("signal", &self.signal)
+            .field("stdout_total_bytes", &self.stdout_total_bytes)
+            .field("stderr_total_bytes", &self.stderr_total_bytes)
+            .finish_non_exhaustive()
+    }
+}
+
+impl std::error::Error for RunnerStepFailure {}
+
+/// Read `reader` to its end keeping only the last `cap` bytes and counting
+/// every byte, so a verbose step never buffers whole in the daemon.
+fn drain_tail(mut reader: impl Read, cap: usize) -> std::io::Result<(Vec<u8>, u64)> {
+    let mut tail = Vec::new();
+    let mut total = 0u64;
+    let mut chunk = [0u8; 8192];
+    loop {
+        let read = match reader.read(&mut chunk) {
+            Ok(0) => return Ok((tail, total)),
+            Ok(read) => read,
+            Err(error) if error.kind() == ErrorKind::Interrupted => continue,
+            Err(error) => return Err(error),
+        };
+        total += read as u64;
+        tail.extend_from_slice(&chunk[..read]);
+        if tail.len() > cap {
+            tail.drain(..tail.len() - cap);
+        }
+    }
+}
+
 /// Docker is only a runner substrate. Recipe argv stays an argv vector, the
 /// operator binding selects the exact image/network/mount affordances, and the
 /// driver returns complete materialization receipts rather than build truth by
@@ -2170,7 +2299,7 @@ impl Default for DockerRunnerDriver {
 }
 
 impl DockerRunnerDriver {
-    fn docker<I, S>(&self, args: I) -> Result<Output>
+    fn docker<I, S>(&self, subject: RunnerSubject, runner_id: &str, args: I) -> Result<()>
     where
         I: IntoIterator<Item = S>,
         S: AsRef<OsStr>,
@@ -2179,23 +2308,50 @@ impl DockerRunnerDriver {
             self.docker_program.is_absolute(),
             "Docker runner program is not absolute"
         );
-        let output = Command::new(&self.docker_program)
+        let mut child = Command::new(&self.docker_program)
             .args(args)
             .stdin(Stdio::null())
-            .output()
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .spawn()
             .with_context(|| format!("starting Docker runner {}", self.docker_program.display()))?;
-        if !output.status.success() {
-            bail!(
-                "Docker runner exited with {}: {}",
-                output.status,
-                String::from_utf8_lossy(&output.stderr).trim()
-            );
+        let stdout = child.stdout.take().context("Docker runner has no stdout pipe")?;
+        let stderr = child.stderr.take().context("Docker runner has no stderr pipe")?;
+        let stdout = thread::spawn(move || drain_tail(stdout, RUNNER_OUTPUT_TAIL_BYTES));
+        let stderr = thread::spawn(move || drain_tail(stderr, RUNNER_OUTPUT_TAIL_BYTES));
+        let status = child.wait().context("waiting for the Docker runner")?;
+        let (stdout_tail, stdout_total_bytes) = stdout
+            .join()
+            .map_err(|_| anyhow!("the runner stdout reader panicked"))?
+            .context("reading the runner stdout")?;
+        let (stderr_tail, stderr_total_bytes) = stderr
+            .join()
+            .map_err(|_| anyhow!("the runner stderr reader panicked"))?
+            .context("reading the runner stderr")?;
+        if status.success() {
+            return Ok(());
         }
-        Ok(output)
+        #[cfg(unix)]
+        let signal = std::os::unix::process::ExitStatusExt::signal(&status);
+        #[cfg(not(unix))]
+        let signal = None;
+        Err(RunnerStepFailure {
+            subject,
+            runner: runner_id.to_owned(),
+            exit_code: status.code(),
+            signal,
+            stdout_tail,
+            stdout_total_bytes,
+            stderr_tail,
+            stderr_total_bytes,
+        }
+        .into())
     }
 
     fn run_in_workspace(
         &self,
+        step_id: &str,
+        runner_id: &str,
         runner: &DockerRunnerBinding,
         workspace: &Path,
         working_directory: &Path,
@@ -2212,8 +2368,7 @@ impl DockerRunnerDriver {
         build_machine_id_file(workspace)?;
         let spec = ContainerSpec::for_step(runner, required_environment, source_stamp)?;
         let args = docker_run_args(&spec, workspace, working_directory, argv)?;
-        self.docker(args)?;
-        Ok(())
+        self.docker(RunnerSubject::Step(step_id.to_owned()), runner_id, args)
     }
 
     fn materialize_external_input(
@@ -2255,7 +2410,11 @@ impl DockerRunnerDriver {
             input.url.clone(),
         ];
         let args = docker_run_args(&spec, workspace, Path::new("."), &curl_argv)?;
-        self.docker(args)?;
+        self.docker(
+            RunnerSubject::ExternalInput(input.id.clone()),
+            &input.runner,
+            args,
+        )?;
         let bytes = fs::read(&destination).with_context(|| {
             format!(
                 "reading materialized external input {}",
@@ -2377,6 +2536,8 @@ impl RunnerPort for DockerRunnerDriver {
                 );
             }
             self.run_in_workspace(
+                &step.id,
+                &step.runner,
                 runner,
                 workspace,
                 &step.working_directory,
@@ -2415,6 +2576,13 @@ impl RunnerPort for DockerRunnerDriver {
     }
 }
 
+/// How every incarnation unit lives and dies. It runs once (`Restart=no`)
+/// and systemd forgets it once it is inactive or failed (`--collect`): the
+/// outcome is Idunn's to record in its own store, and a failed transient unit
+/// left loaded is only clutter that marks the host degraded.
+const INCARNATION_UNIT_LIFECYCLE: [&str; 3] =
+    ["--collect", "--property=Type=exec", "--property=Restart=no"];
+
 /// systemd owns process execution. This driver lowers one validated launch
 /// contract into a transient unit and then proves the native process and
 /// executable that systemd actually started. It does not decide admission,
@@ -3605,8 +3773,9 @@ impl SystemdTransientWorkloadDriver {
             OsString::from("--no-ask-password"),
             OsString::from("--expand-environment=no"),
             OsString::from(format!("--unit={unit}")),
-            OsString::from("--property=Type=exec"),
-            OsString::from("--property=Restart=no"),
+        ];
+        args.extend(INCARNATION_UNIT_LIFECYCLE.map(OsString::from));
+        args.extend([
             OsString::from("--property=KillMode=mixed"),
             OsString::from("--property=DynamicUser=yes"),
             OsString::from("--property=SupplementaryGroups="),
@@ -3642,7 +3811,7 @@ impl SystemdTransientWorkloadDriver {
             OsString::from(format!("--working-directory={}", installed.display())),
             OsString::from(format!("--property=ReadOnlyPaths={}", installed.display())),
             OsString::from(format!("--property=ReadOnlyPaths={}", bundle.display())),
-        ];
+        ]);
         for open_file in parent_only_open_files {
             args.push(OsString::from(format!("--property=OpenFile={open_file}")));
         }
@@ -9783,6 +9952,16 @@ Content-Le".to_vec()), |_| {}),
         Ok(())
     }
 
+    /// An incarnation unit runs once and is collected when it ends, so a
+    /// failed candidate does not stay loaded and mark the host degraded.
+    /// Collection is safe because a forgotten unit already reads as
+    /// permanently stopped (the test below).
+    #[test]
+    fn incarnation_units_run_once_and_are_collected() {
+        assert!(INCARNATION_UNIT_LIFECYCLE.contains(&"--collect"));
+        assert!(INCARNATION_UNIT_LIFECYCLE.contains(&"--property=Restart=no"));
+    }
+
     /// Admission audit, claim 3: `is_permanently_stopped` now lets the
     /// Isolation gate skip `prove_isolation` against the incumbent. Enumerate
     /// what it answers `true` for, using a fake `systemctl show` that replays
@@ -16231,4 +16410,292 @@ processes 9
         );
         Ok(())
     }
+
+    // BEGIN runner-failure-evidence tests
+    // The runner decides what failed and what it kept: these drive the real
+    // DockerRunnerDriver over a fake docker program, never the capture alone.
+
+    const RUNNER_CANARY: &str = "CANARY-5d2e91b7";
+
+    #[cfg(unix)]
+    fn fake_docker(dir: &Path, body: &str) -> Result<DockerRunnerDriver> {
+        use std::os::unix::fs::PermissionsExt;
+        let program = dir.join("docker");
+        fs::write(&program, format!("#!/bin/sh\n{body}\n"))?;
+        fs::set_permissions(&program, fs::Permissions::from_mode(0o755))?;
+        Ok(DockerRunnerDriver {
+            docker_program: program,
+        })
+    }
+
+    /// One recipe step through the real `run_in_workspace` path.
+    #[cfg(unix)]
+    fn run_step(driver: &DockerRunnerDriver, dir: &Path, step_id: &str) -> Result<()> {
+        let workspace = dir.join("workspace");
+        fs::create_dir_all(&workspace)?;
+        driver.run_in_workspace(
+            step_id,
+            "rust",
+            &fixture_docker_runner_binding(),
+            &workspace,
+            Path::new("."),
+            &["cargo".to_owned(), "test".to_owned()],
+            &std::collections::BTreeSet::new(),
+            ("IDUNN_STAMP", "rev"),
+        )
+    }
+
+    #[cfg(unix)]
+    fn failure_of(result: Result<()>) -> Result<RunnerStepFailure> {
+        let error = result.expect_err("the step did not fail");
+        error
+            .downcast_ref::<RunnerStepFailure>()
+            .cloned()
+            .with_context(|| format!("the failure is not typed: {error:#}"))
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn a_failed_step_is_named_with_both_tails() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        // The first step passes, the second fails: the count is the fake's.
+        let driver = fake_docker(
+            temp.path(),
+            &format!(
+                "n=$(cat \"$0.count\" 2>/dev/null || echo 0); n=$((n+1)); echo $n > \"$0.count\"\n\
+                 [ \"$n\" -lt 2 ] && exit 0\n\
+                 echo 'stdout line one'; echo 'stdout report {RUNNER_CANARY}'\n\
+                 echo 'stderr report' >&2\nexit 3"
+            ),
+        )?;
+        run_step(&driver, temp.path(), "first")?;
+        let failure = failure_of(run_step(&driver, temp.path(), "second"))?;
+        assert_eq!(failure.subject, RunnerSubject::Step("second".into()));
+        assert_eq!(failure.runner, "rust");
+        assert_eq!(failure.exit_code, Some(3));
+        assert_eq!(failure.signal, None);
+        assert!(
+            failure
+                .stdout_tail
+                .ends_with(format!("stdout report {RUNNER_CANARY}\n").as_bytes()),
+            "stdout was dropped: {:?}",
+            String::from_utf8_lossy(&failure.stdout_tail)
+        );
+        assert_eq!(failure.stderr_tail, b"stderr report\n");
+        assert_eq!(failure.stdout_total_bytes, failure.stdout_tail.len() as u64);
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn a_step_killed_by_a_signal_is_named_by_its_signal() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let driver = fake_docker(temp.path(), "kill -9 $$")?;
+        let failure = failure_of(run_step(&driver, temp.path(), "build"))?;
+        assert_eq!((failure.exit_code, failure.signal), (None, Some(9)));
+        assert!(failure.to_string().contains("exited with signal 9"), "{failure}");
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn a_failed_external_input_is_named_as_one() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let driver = fake_docker(temp.path(), "echo curl failed >&2\nexit 22")?;
+        let declaration = TargetDeclaration::parse(crate::deployment_plan::tests::RECIPE)?;
+        let binding = OperatorBinding::parse(crate::deployment_plan::tests::BINDING)?;
+        let input_id = declaration.external_inputs[0].id.clone();
+        let workspaces = BTreeMap::from([("rust".to_owned(), temp.path().join("workspace"))]);
+        fs::create_dir_all(&workspaces["rust"])?;
+        let error = driver
+            .materialize_external_input(&declaration, &binding, &workspaces, &input_id)
+            .expect_err("the download did not fail");
+        let failure = error
+            .downcast_ref::<RunnerStepFailure>()
+            .with_context(|| format!("the failure is not typed: {error:#}"))?;
+        assert_eq!(failure.subject, RunnerSubject::ExternalInput(input_id.clone()));
+        assert_eq!(failure.exit_code, Some(22));
+        assert!(
+            failure.to_string().contains(&format!("runner external input {input_id} on runner rust")),
+            "{failure}"
+        );
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn runner_output_never_reaches_the_error_text() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let driver = fake_docker(
+            temp.path(),
+            &format!("echo 'out {RUNNER_CANARY}'\necho 'err {RUNNER_CANARY}' >&2\nexit 1"),
+        )?;
+        let failure = failure_of(run_step(&driver, temp.path(), "build"))?;
+        assert!(String::from_utf8_lossy(&failure.stdout_tail).contains(RUNNER_CANARY));
+        assert!(String::from_utf8_lossy(&failure.stderr_tail).contains(RUNNER_CANARY));
+        let error = anyhow::Error::new(failure.clone()).context("sealing");
+        for text in [
+            format!("{error:#}"),
+            format!("{error:?}"),
+            format!("{failure}"),
+            format!("{failure:?}"),
+        ] {
+            assert!(!text.contains(RUNNER_CANARY), "runner output reached text: {text}");
+        }
+        Ok(())
+    }
+
+    /// A reader that answers from a script, so a read error the OS rarely
+    /// produces (an interrupted call) is produced on demand.
+    struct ScriptedReader(Vec<std::io::Result<Vec<u8>>>);
+
+    impl Read for ScriptedReader {
+        fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
+            if self.0.is_empty() {
+                return Ok(0);
+            }
+            let bytes = self.0.remove(0)?;
+            buffer[..bytes.len()].copy_from_slice(&bytes);
+            Ok(bytes.len())
+        }
+    }
+
+    struct BrokenReader;
+
+    impl Read for BrokenReader {
+        fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
+            Err(std::io::Error::other("persistent read failure"))
+        }
+    }
+
+    #[test]
+    fn an_interrupted_read_is_retried_and_loses_no_byte() {
+        let reader = ScriptedReader(vec![
+            Ok(b"ab".to_vec()),
+            Err(std::io::Error::from(ErrorKind::Interrupted)),
+            Ok(b"cd".to_vec()),
+            Err(std::io::Error::from(ErrorKind::Interrupted)),
+        ]);
+        let drained = drain_tail(reader, 3).expect("an interrupted read ended the drain");
+        assert_eq!(drained, (b"bcd".to_vec(), 4));
+    }
+
+    #[test]
+    fn a_persistent_read_error_ends_the_drain_instead_of_spinning() {
+        let (sender, receiver) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let _ = sender.send(drain_tail(BrokenReader, 8).is_err());
+        });
+        assert_eq!(
+            receiver.recv_timeout(Duration::from_secs(5)),
+            Ok(true),
+            "a persistent read error spun or was swallowed"
+        );
+    }
+
+    #[test]
+    fn a_non_interrupted_read_error_after_data_is_returned_not_retried() {
+        let reader = ScriptedReader(vec![
+            Ok(b"ab".to_vec()),
+            Err(std::io::Error::from(ErrorKind::BrokenPipe)),
+            Ok(b"cd".to_vec()),
+        ]);
+        let error = drain_tail(reader, 8).expect_err("a broken pipe was retried past");
+        assert_eq!(error.kind(), ErrorKind::BrokenPipe);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn each_tail_is_the_last_bound_bytes_with_the_total() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let data = temp.path().join("data");
+        for size in [
+            0,
+            1,
+            RUNNER_OUTPUT_TAIL_BYTES - 1,
+            RUNNER_OUTPUT_TAIL_BYTES,
+            RUNNER_OUTPUT_TAIL_BYTES + 1,
+            1024 * 1024,
+        ] {
+            // Distinct neighbours, so a shifted or head-keeping tail differs.
+            let bytes: Vec<u8> = (0..size).map(|index| (index % 251) as u8).collect();
+            fs::write(&data, &bytes)?;
+            let driver = fake_docker(
+                temp.path(),
+                &format!("cat '{}'\ncat '{}' >&2\nexit 1", data.display(), data.display()),
+            )?;
+            let failure = failure_of(run_step(&driver, temp.path(), "build"))?;
+            let expected = &bytes[size.saturating_sub(RUNNER_OUTPUT_TAIL_BYTES)..];
+            assert_eq!(failure.stdout_tail, expected, "stdout tail at {size} bytes");
+            assert_eq!(failure.stderr_tail, expected, "stderr tail at {size} bytes");
+            assert_eq!(failure.stdout_total_bytes, size as u64);
+            assert_eq!(failure.stderr_total_bytes, size as u64);
+        }
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn a_successful_step_and_a_missing_program_keep_their_outcomes() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        run_step(&fake_docker(temp.path(), "echo fine")?, temp.path(), "build")?;
+        let missing = DockerRunnerDriver {
+            docker_program: temp.path().join("no-such-docker"),
+        };
+        let error = run_step(&missing, temp.path(), "build").expect_err("a missing program ran");
+        assert!(error.downcast_ref::<RunnerStepFailure>().is_none());
+        assert!(format!("{error:#}").contains("starting Docker runner"), "{error:#}");
+        Ok(())
+    }
+
+    /// The 2026-10-25 vili failure: `node --test` reported on stdout alone,
+    /// stderr stayed empty (the runner's /tmp is a noexec tmpfs) and the step
+    /// exited 1. The old capture dropped stdout and printed an empty message.
+    #[cfg(unix)]
+    #[test]
+    fn a_step_that_reports_only_on_stdout_keeps_that_report_and_its_name() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let driver = fake_docker(
+            temp.path(),
+            &format!("echo 'not ok 1 - daemon {RUNNER_CANARY}'\necho '# fail 1'\nexit 1"),
+        )?;
+        let error = run_step(&driver, temp.path(), "test-daemon").expect_err("the step passed");
+        // The error names the step and carries no output.
+        let text = format!("{error:#}");
+        assert!(text.contains("runner step test-daemon on runner rust exited with exit status 1"), "{text}");
+        assert!(!text.contains(RUNNER_CANARY) && !text.contains("not ok"), "{text}");
+        // The evidence keeps the stdout tail; the empty stderr is a recorded 0.
+        let failure = error
+            .downcast_ref::<RunnerStepFailure>()
+            .context("the failure is not typed")?;
+        let store = temp.path().join("runner-evidence.cc");
+        let record = crate::runner_evidence::RunnerFailureRecord::from_failure(
+            "tx-vili",
+            "up-vili",
+            "vili",
+            failure,
+            1,
+        );
+        crate::runner_evidence::record(&store, &record)?;
+        let kept = crate::runner_evidence::for_transaction(&store, "tx-vili")?.context("no evidence kept")?;
+        assert_eq!(kept.subject_id, "test-daemon");
+        assert_eq!(kept.exit_code, Some(1));
+        assert_eq!(
+            String::from_utf8_lossy(&kept.stdout_tail),
+            format!("not ok 1 - daemon {RUNNER_CANARY}\n# fail 1\n")
+        );
+        assert_eq!(kept.stdout_total_bytes, kept.stdout_tail.len() as u64);
+        assert!(kept.stderr_tail.is_empty());
+        assert_eq!(kept.stderr_total_bytes, 0);
+        let mut shown = Vec::new();
+        crate::runner_evidence::show_status(&mut shown, &store, "tx-vili")?;
+        let rendered = String::from_utf8(shown)?;
+        assert!(rendered.contains("stderr (last 0 of 0 bytes)"), "{rendered}");
+        assert!(rendered.contains(&format!("| not ok 1 - daemon {RUNNER_CANARY}")), "{rendered}");
+        // Debug names the failure and prints no tail.
+        let debug = format!("{failure:?}");
+        assert!(debug.starts_with("RunnerStepFailure") && !debug.contains("not ok"), "{debug}");
+        Ok(())
+    }
+    // END runner-failure-evidence tests
 }

src/host_actuator.rs

@@ -1151,7 +1151,7 @@ mod tests {
         let answerer = thread::spawn(move || -> Result<()> {
             for _ in 0..500 {
                 client.poll_resends()?;
-                while let Some(frame) = client.receive_once()? {
+                if let Some(frame) = client.receive_once()? {
                     let envelope = HostActuatorEnvelope::decode(&frame.payload)?;
                     let request = envelope.open_request("raven", &idunn_anchor)?;
                     assert_eq!(envelope.session_nonce, nonce);

src/lib.rs

@@ -7,4 +7,7 @@ pub mod host_actuator;
 #[cfg(unix)]
 pub mod launcher;
 mod incident;
+mod runner_evidence;
 pub mod provisioning;
+#[cfg(unix)]
+mod steward_orders;

src/runner_evidence.rs

@@ -0,0 +1,805 @@
+//! The output tail of a failed runner step.
+//!
+//! One file, `runner-evidence.cc`, beside `control.cc`, holds one type: the
+//! last bytes of a failed runner container's stdout and stderr, one record per
+//! transaction. The engine's pre-fence failure path is its only writer, and
+//! `idunn status --command` its only reader. It is never `control.cc`, which
+//! refuses any type it does not know, and never `incidents.cc`, which is
+//! published and carries no free text.
+//!
+//! The tail is raw runner output. It is kept unredacted and readable by root
+//! alone (mode 0600): root can already read every secret the runner was given.
+//! It never enters an error, a transaction record or the journal. The status
+//! reader is a terminal, so everything it prints from this store goes through
+//! one allowlist sink (`emit`): printable ASCII passes, all else is escaped.
+
+use std::path::{Path, PathBuf};
+
+use anyhow::{Context, Result, bail, ensure};
+use cultcache_rs::{
+    CultCacheEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore,
+    TryCompareExchangeSnapshotOutcome,
+};
+use serde::{Deserialize, Serialize};
+
+use crate::control_plane::{decode_record, require_id, typed_envelope};
+use crate::drivers::{
+    LockContended, RUNNER_OUTPUT_TAIL_BYTES, RunnerStepFailure, RunnerSubject,
+    exchange_behind_private_lock, runner_status,
+};
+
+const RUNNER_FAILURE_SCHEMA: &str = "idunn.runner_failure.v1";
+
+/// Records kept; the writer drops the oldest past this, so the store needs no
+/// other owner.
+pub(crate) const RUNNER_EVIDENCE_RECORDS: usize = 32;
+
+/// `runner-evidence.cc`, beside the control store.
+pub(crate) fn runner_evidence_store_path(state_store: &Path) -> PathBuf {
+    state_store.with_file_name("runner-evidence.cc")
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub(crate) enum SubjectKind {
+    Step,
+    ExternalInput,
+}
+
+#[derive(Clone, PartialEq, Eq, DatabaseEntry)]
+#[cultcache(type = "idunn.runner_failure", schema = "idunn.runner_failure.v1")]
+pub(crate) struct RunnerFailureRecord {
+    #[cultcache(key = 0)]
+    pub(crate) schema_version: String,
+    #[cultcache(key = 1)]
+    pub(crate) transaction_id: String,
+    #[cultcache(key = 2)]
+    pub(crate) command_id: String,
+    #[cultcache(key = 3)]
+    pub(crate) target: String,
+    #[cultcache(key = 4)]
+    pub(crate) subject_kind: SubjectKind,
+    #[cultcache(key = 5)]
+    pub(crate) subject_id: String,
+    #[cultcache(key = 6)]
+    pub(crate) runner: String,
+    #[cultcache(key = 7)]
+    pub(crate) exit_code: Option<i32>,
+    #[cultcache(key = 8)]
+    pub(crate) signal: Option<i32>,
+    #[cultcache(key = 9)]
+    pub(crate) stdout_tail: Vec<u8>,
+    #[cultcache(key = 10)]
+    pub(crate) stdout_total_bytes: u64,
+    #[cultcache(key = 11)]
+    pub(crate) stderr_tail: Vec<u8>,
+    #[cultcache(key = 12)]
+    pub(crate) stderr_total_bytes: u64,
+    #[cultcache(key = 13)]
+    pub(crate) recorded_at_unix_millis: u64,
+}
+
+/// Tails stay out of `Debug`, so a failed assertion or a logged record never
+/// prints runner output.
+impl std::fmt::Debug for RunnerFailureRecord {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("RunnerFailureRecord")
+            .field("transaction_id", &self.transaction_id)
+            .field("subject_id", &self.subject_id)
+            .finish_non_exhaustive()
+    }
+}
+
+impl RunnerFailureRecord {
+    pub(crate) fn from_failure(
+        transaction_id: &str,
+        command_id: &str,
+        target: &str,
+        failure: &RunnerStepFailure,
+        now: u64,
+    ) -> Self {
+        let (subject_kind, subject_id) = match &failure.subject {
+            RunnerSubject::Step(id) => (SubjectKind::Step, id),
+            RunnerSubject::ExternalInput(id) => (SubjectKind::ExternalInput, id),
+        };
+        Self {
+            schema_version: RUNNER_FAILURE_SCHEMA.to_owned(),
+            transaction_id: transaction_id.to_owned(),
+            command_id: command_id.to_owned(),
+            target: target.to_owned(),
+            subject_kind,
+            subject_id: subject_id.clone(),
+            runner: failure.runner.clone(),
+            exit_code: failure.exit_code,
+            signal: failure.signal,
+            stdout_tail: failure.stdout_tail.clone(),
+            stdout_total_bytes: failure.stdout_total_bytes,
+            stderr_tail: failure.stderr_tail.clone(),
+            stderr_total_bytes: failure.stderr_total_bytes,
+            recorded_at_unix_millis: now,
+        }
+    }
+
+    pub(crate) fn validate(&self) -> Result<()> {
+        ensure!(
+            self.schema_version == RUNNER_FAILURE_SCHEMA,
+            "runner failure schema is unsupported"
+        );
+        require_id(&self.transaction_id, "runner failure transaction")?;
+        require_id(&self.command_id, "runner failure command")?;
+        require_id(&self.target, "runner failure target")?;
+        require_id(&self.subject_id, "runner failure subject")?;
+        require_id(&self.runner, "runner failure runner")?;
+        for (tail, total, stream) in [
+            (&self.stdout_tail, self.stdout_total_bytes, "stdout"),
+            (&self.stderr_tail, self.stderr_total_bytes, "stderr"),
+        ] {
+            ensure!(
+                tail.len() <= RUNNER_OUTPUT_TAIL_BYTES,
+                "runner failure {stream} tail exceeds its bound"
+            );
+            ensure!(
+                tail.len() as u64 <= total,
+                "runner failure {stream} tail exceeds its stream"
+            );
+        }
+        Ok(())
+    }
+
+    fn envelope(&self) -> Result<CultCacheEnvelope> {
+        self.validate()?;
+        typed_envelope(
+            &self.transaction_id,
+            Self::TYPE,
+            RUNNER_FAILURE_SCHEMA,
+            self,
+            self.recorded_at_unix_millis,
+        )
+    }
+}
+
+/// One read of the store: the envelopes exactly as stored, and the records
+/// decoded from them in the same order.
+fn snapshot(path: &Path) -> Result<(Vec<CultCacheEnvelope>, Vec<RunnerFailureRecord>)> {
+    let envelopes = SingleFileMessagePackBackingStore::new(path)
+        .pull_all_read_only_snapshot()
+        .context("reading the runner evidence store")?;
+    let mut records = Vec::new();
+    for envelope in &envelopes {
+        ensure!(
+            envelope.r#type == RunnerFailureRecord::TYPE
+                && envelope.schema_id.as_deref() == Some(RUNNER_FAILURE_SCHEMA),
+            "the runner evidence store holds a document that is not a runner failure"
+        );
+        let record: RunnerFailureRecord = decode_record(envelope)?;
+        record.validate()?;
+        ensure!(
+            envelope.key == record.transaction_id,
+            "runner evidence store key differs from the record's transaction"
+        );
+        records.push(record);
+    }
+    Ok((envelopes, records))
+}
+
+/// How a write left the store.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum Kept {
+    Stored,
+    /// The store held bytes that do not decode; it was moved to
+    /// `unreadable_store_path` and the record opened a fresh one. `replaced`
+    /// is the last-written time (unix milliseconds) of an earlier set-aside
+    /// store this one displaced, which is gone: one slot, and it says so.
+    SetAside { replaced: Option<u64> },
+}
+
+/// Where an undecodable store is kept. One slot: the next one replaces it, so
+/// the damage a bad store can cost is one extra file.
+pub(crate) fn unreadable_store_path(path: &Path) -> PathBuf {
+    path.with_file_name("runner-evidence.unreadable.cc")
+}
+
+/// Moves the store over the previous set-aside one, root-only, and reports
+/// what it displaced. Nothing of the store's content is named: a decode error
+/// can quote stored values, and none is carried here.
+fn set_aside_unreadable(path: &Path) -> Result<Option<u64>> {
+    let aside = unreadable_store_path(path);
+    let replaced = std::fs::symlink_metadata(&aside).ok().map(|earlier| {
+        earlier
+            .modified()
+            .ok()
+            .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
+            .map_or(0, |since| since.as_millis() as u64)
+    });
+    std::fs::rename(path, &aside)
+        .context("the runner evidence store is undecodable and could not be set aside")?;
+    keep_private(&aside)?;
+    Ok(replaced)
+}
+
+/// How the writer reads the store's bytes; tests inject an I/O error here.
+type ReadStore<'a> = &'a dyn Fn(&Path) -> std::io::Result<Vec<u8>>;
+
+fn unreadable_fault(error: &std::io::Error) -> anyhow::Error {
+    anyhow::anyhow!("the runner evidence store could not be read: {}", error.kind())
+}
+
+/// The one writer. Upsert by transaction id; past `RUNNER_EVIDENCE_RECORDS` the
+/// oldest records go in the same exchange. The file ends at mode 0600.
+///
+/// A store is set aside only when bytes that were read do not decode. An I/O
+/// error leaves the store in place, keeps no evidence for that failure, and
+/// the error names its kind for the journal.
+pub(crate) fn record(path: &Path, record: &RunnerFailureRecord) -> Result<Kept> {
+    record_reading(path, record, &|path| std::fs::read(path))
+}
+
+fn record_reading(path: &Path, record: &RunnerFailureRecord, read: ReadStore) -> Result<Kept> {
+    let first = match read(path) {
+        Ok(bytes) => Some(bytes),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+        Err(error) => return Err(unreadable_fault(&error)),
+    };
+    let (envelopes, records, how) = match snapshot(path) {
+        Ok((envelopes, records)) => (envelopes, records, Kept::Stored),
+        Err(_) => {
+            let Some(first) = first else {
+                bail!("the runner evidence store could not be decoded");
+            };
+            // Undecodable only if the same bytes are still there to read: a
+            // second read that fails, or finds other bytes, is not a verdict.
+            match read(path) {
+                Ok(second) if second == first => {}
+                Ok(_) => bail!("the runner evidence store changed while it was being read"),
+                Err(error) => return Err(unreadable_fault(&error)),
+            }
+            let replaced = set_aside_unreadable(path)?;
+            (Vec::new(), Vec::new(), Kept::SetAside { replaced })
+        }
+    };
+    let mut kept: Vec<(u64, CultCacheEnvelope)> = envelopes
+        .iter()
+        .zip(&records)
+        .filter(|(_, stored)| stored.transaction_id != record.transaction_id)
+        .map(|(envelope, stored)| (stored.recorded_at_unix_millis, envelope.clone()))
+        .collect();
+    kept.push((record.recorded_at_unix_millis, record.envelope()?));
+    // Stable: records of one instant keep their stored order, so the record
+    // being written survives a tie.
+    kept.sort_by_key(|(recorded_at, _)| *recorded_at);
+    let excess = kept.len().saturating_sub(RUNNER_EVIDENCE_RECORDS);
+    let replacements: Vec<_> = kept
+        .into_iter()
+        .skip(excess)
+        .map(|(_, envelope)| envelope)
+        .collect();
+    match exchange_behind_private_lock(path, &envelopes, &replacements)? {
+        TryCompareExchangeSnapshotOutcome::Exchanged => {}
+        TryCompareExchangeSnapshotOutcome::Mismatch => {
+            bail!("the runner evidence store changed while it was being written")
+        }
+        TryCompareExchangeSnapshotOutcome::LockContended => {
+            return Err(LockContended {
+                store: "runner evidence store",
+            }
+            .into());
+        }
+    }
+    keep_private(path)?;
+    Ok(how)
+}
+
+#[cfg(unix)]
+fn keep_private(path: &Path) -> Result<()> {
+    use std::os::unix::fs::PermissionsExt;
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
+        .context("making the runner evidence store private")
+}
+
+#[cfg(not(unix))]
+fn keep_private(_path: &Path) -> Result<()> {
+    Ok(())
+}
+
+/// The one reader. An absent store holds nothing.
+pub(crate) fn for_transaction(
+    path: &Path,
+    transaction_id: &str,
+) -> Result<Option<RunnerFailureRecord>> {
+    Ok(snapshot(path)?
+        .1
+        .into_iter()
+        .find(|stored| stored.transaction_id == transaction_id))
+}
+
+/// The record as `idunn status --command` shows it, as raw lines. They are
+/// not text yet: `show_status` is the only caller, and `emit` the only
+/// writer.
+fn render(record: &RunnerFailureRecord) -> Vec<Vec<u8>> {
+    let noun = match record.subject_kind {
+        SubjectKind::Step => "step",
+        SubjectKind::ExternalInput => "external input",
+    };
+    let mut lines: Vec<Vec<u8>> = vec![
+        format!("    failed {noun} {}", record.subject_id).into_bytes(),
+        format!("    runner {}", record.runner).into_bytes(),
+        format!("    status {}", runner_status(record.exit_code, record.signal)).into_bytes(),
+    ];
+    for (stream, tail, total) in [
+        ("stdout", &record.stdout_tail, record.stdout_total_bytes),
+        ("stderr", &record.stderr_tail, record.stderr_total_bytes),
+    ] {
+        lines.push(format!("      {stream} (last {} of {total} bytes)", tail.len()).into_bytes());
+        let text = tail.strip_suffix(b"\n").unwrap_or(tail);
+        if !text.is_empty() {
+            lines.extend(text.split(|byte| *byte == b'\n').map(|line| {
+                let mut shown = b"      | ".to_vec();
+                shown.extend_from_slice(line);
+                shown
+            }));
+        }
+    }
+    lines
+}
+
+/// What `idunn status --command` prints for a transaction's runner evidence:
+/// its record, or one line naming why the store is unreadable. Nothing is
+/// written when the transaction has no record.
+pub(crate) fn show_status(
+    out: &mut impl std::io::Write,
+    store: &Path,
+    transaction_id: &str,
+) -> std::io::Result<()> {
+    let lines = match for_transaction(store, transaction_id) {
+        Ok(Some(record)) => render(&record),
+        Ok(None) => return Ok(()),
+        Err(error) => {
+            vec![format!("    runner evidence unreadable: {}", error.root_cause()).into_bytes()]
+        }
+    };
+    emit(out, &lines)
+}
+
+/// The one sink: every byte the status read prints from the evidence store,
+/// or from an error about it, is written here and nowhere else. An allowlist:
+/// printable ASCII and tab pass; every other code point is a visible
+/// `\u{..}` escape and every invalid UTF-8 byte a `\x..` escape. A newline
+/// inside a line is escaped too: this function writes the only ones.
+pub(crate) fn emit(out: &mut impl std::io::Write, lines: &[Vec<u8>]) -> std::io::Result<()> {
+    for line in lines {
+        out.write_all(escape(line).as_bytes())?;
+        out.write_all(b"\n")?;
+    }
+    Ok(())
+}
+
+pub(crate) fn escape(bytes: &[u8]) -> String {
+    let mut shown = String::with_capacity(bytes.len());
+    for chunk in bytes.utf8_chunks() {
+        for character in chunk.valid().chars() {
+            match character {
+                ' '..='~' | '\t' => shown.push(character),
+                _ => shown.push_str(&format!("\\u{{{:x}}}", u32::from(character))),
+            }
+        }
+        for byte in chunk.invalid() {
+            shown.push_str(&format!("\\x{byte:02x}"));
+        }
+    }
+    shown
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+    use std::os::unix::fs::PermissionsExt;
+
+    fn failure(subject: RunnerSubject, stdout: &[u8], stderr: &[u8]) -> RunnerStepFailure {
+        RunnerStepFailure {
+            subject,
+            runner: "rust".into(),
+            exit_code: Some(3),
+            signal: None,
+            stdout_tail: stdout.to_vec(),
+            stdout_total_bytes: stdout.len() as u64,
+            stderr_tail: stderr.to_vec(),
+            stderr_total_bytes: stderr.len() as u64,
+        }
+    }
+
+    /// What status writes for a record, through the one sink.
+    fn shown(record: &RunnerFailureRecord) -> String {
+        let mut written = Vec::new();
+        emit(&mut written, &render(record)).expect("a Vec takes any write");
+        String::from_utf8(written).expect("the sink writes ASCII only")
+    }
+
+    fn shown_lines(record: &RunnerFailureRecord) -> Vec<String> {
+        shown(record).lines().map(str::to_owned).collect()
+    }
+
+    fn record_at(transaction: &str, now: u64) -> RunnerFailureRecord {
+        RunnerFailureRecord::from_failure(
+            transaction,
+            "up-service",
+            "service",
+            &failure(RunnerSubject::Step("build".into()), b"out\n", b"err\n"),
+            now,
+        )
+    }
+
+    #[test]
+    fn the_evidence_store_keeps_the_newest_32() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        // The literal, not the constant: the spec's bound is 32.
+        for index in 0..=32 {
+            record(&path, &record_at(&format!("tx-{index:02}"), 1000 + index as u64))?;
+        }
+        let kept: Vec<_> = snapshot(&path)?.1.into_iter().map(|stored| stored.transaction_id).collect();
+        assert_eq!(kept.len(), 32);
+        assert!(!kept.contains(&"tx-00".to_owned()), "the oldest survived");
+        assert!(kept.contains(&"tx-32".to_owned()), "the newest was evicted");
+        assert!(kept.contains(&"tx-01".to_owned()), "the second oldest was evicted early");
+        // Rewriting a resident transaction replaces it and drops nothing.
+        let mut again = record_at("tx-05", 5000);
+        again.stdout_tail = b"rewritten\n".to_vec();
+        again.stdout_total_bytes = 10;
+        record(&path, &again)?;
+        let kept = snapshot(&path)?.1;
+        assert_eq!(kept.len(), 32);
+        assert!(kept.iter().any(|stored| stored.transaction_id == "tx-01"));
+        let stored = for_transaction(&path, "tx-05")?.context("rewritten record is gone")?;
+        assert_eq!(stored.stdout_tail, b"rewritten\n");
+        assert_eq!(for_transaction(&path, "tx-00")?, None);
+        Ok(())
+    }
+
+    #[test]
+    fn a_record_round_trips_byte_for_byte() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        let written = RunnerFailureRecord::from_failure(
+            "tx-a",
+            "up-service",
+            "service",
+            &failure(RunnerSubject::ExternalInput("toolchain".into()), &[0, 1, 255, 10], b""),
+            77,
+        );
+        record(&path, &written)?;
+        assert_eq!(for_transaction(&path, "tx-a")?, Some(written));
+        assert_eq!(for_transaction(&path, "tx-b")?, None);
+        Ok(())
+    }
+
+    #[test]
+    fn the_evidence_store_is_private() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        // Whatever the umask, the mode is the store's own rule, on every write.
+        for index in 0..2 {
+            record(&path, &record_at(&format!("tx-{index}"), 10 + index))?;
+            assert_eq!(std::fs::metadata(&path)?.permissions().mode() & 0o777, 0o600);
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn a_tail_over_its_bound_or_its_stream_is_refused() {
+        let mut oversize = record_at("tx-a", 1);
+        oversize.stdout_tail = vec![b'x'; RUNNER_OUTPUT_TAIL_BYTES + 1];
+        oversize.stdout_total_bytes = u64::MAX;
+        assert!(oversize.validate().is_err());
+        let mut impossible = record_at("tx-a", 1);
+        impossible.stderr_total_bytes = 1;
+        assert!(impossible.validate().is_err());
+        let mut exact = record_at("tx-a", 1);
+        exact.stdout_tail = vec![b'x'; RUNNER_OUTPUT_TAIL_BYTES];
+        exact.stdout_total_bytes = RUNNER_OUTPUT_TAIL_BYTES as u64;
+        assert!(exact.validate().is_ok());
+    }
+
+    #[test]
+    fn status_renders_the_tail_with_control_bytes_escaped() {
+        let mut stored = record_at("tx-a", 1);
+        stored.stdout_tail = b"first\n\x1b[2Jsecond\x00end\tTab\nlast\n".to_vec();
+        stored.stdout_total_bytes = 5000;
+        stored.stderr_tail = b"bad\r\x07".to_vec();
+        stored.stderr_total_bytes = 6;
+        let lines = shown_lines(&stored);
+        let text = lines.join("\n");
+        assert!(!text.contains('\u{1b}') && !text.contains('\0') && !text.contains('\r'));
+        assert!(!text.contains('\u{7}'));
+        assert_eq!(
+            lines,
+            vec![
+                "    failed step build",
+                "    runner rust",
+                "    status exit status 3",
+                "      stdout (last 30 of 5000 bytes)",
+                "      | first",
+                "      | \\u{1b}[2Jsecond\\u{0}end\tTab",
+                "      | last",
+                "      stderr (last 5 of 6 bytes)",
+                "      | bad\\u{d}\\u{7}",
+            ]
+        );
+    }
+
+    #[test]
+    fn debug_names_the_record_and_prints_no_tail() {
+        let mut stored = record_at("tx-a", 1);
+        stored.stdout_tail = b"SECRET-OUTPUT".to_vec();
+        stored.stdout_total_bytes = 13;
+        let debug = format!("{stored:?}");
+        assert!(debug.contains("tx-a") && !debug.contains("SECRET"), "{debug}");
+    }
+
+    #[test]
+    fn an_empty_tail_renders_only_its_header() {
+        let mut stored = record_at("tx-a", 1);
+        stored.stdout_tail = Vec::new();
+        stored.stdout_total_bytes = 0;
+        let lines = shown_lines(&stored);
+        assert!(lines.contains(&"      stdout (last 0 of 0 bytes)".to_owned()));
+        assert!(!lines.iter().any(|line| line == "      | "));
+    }
+
+    #[test]
+    fn the_reader_names_an_unreadable_store_as_an_error() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        std::fs::write(&path, b"\xc1 not a store")?;
+        assert!(for_transaction(&path, "tx-a").is_err());
+        Ok(())
+    }
+
+    fn files_beside(path: &Path) -> Result<Vec<String>> {
+        let mut names: Vec<String> = std::fs::read_dir(path.parent().context("no parent")?)?
+            .map(|entry| Ok(entry?.file_name().to_string_lossy().into_owned()))
+            .collect::<Result<_>>()?;
+        names.retain(|name| name.starts_with("runner-evidence") && !name.contains("lock"));
+        names.sort();
+        Ok(names)
+    }
+
+    /// A store a newer Idunn wrote: well-formed, but a schema this one does not know.
+    fn write_future_store(path: &Path) -> Result<()> {
+        let mut future = record_at("tx-future", 1);
+        future.schema_version = "idunn.runner_failure.v2".into();
+        let envelope = typed_envelope(
+            &future.transaction_id,
+            RunnerFailureRecord::TYPE,
+            "idunn.runner_failure.v2",
+            &future,
+            1,
+        )?;
+        let outcome = exchange_behind_private_lock(path, &[], &[envelope])?;
+        ensure!(
+            matches!(outcome, TryCompareExchangeSnapshotOutcome::Exchanged),
+            "the future store was not written"
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn a_corrupt_store_is_set_aside_and_the_next_failure_keeps_its_evidence() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        let damage = b"\xc1 not a store \x00 damage";
+        std::fs::write(&path, damage)?;
+        let kept = record(&path, &record_at("tx-a", 5))?;
+        assert_eq!(kept, Kept::SetAside { replaced: None });
+        // Kept, not deleted: the damage is where the journal says, root-only.
+        let aside = unreadable_store_path(&path);
+        assert_eq!(std::fs::read(&aside)?, damage);
+        assert_eq!(std::fs::metadata(&aside)?.permissions().mode() & 0o777, 0o600);
+        // The failure's evidence is in a fresh, private store.
+        assert_eq!(
+            for_transaction(&path, "tx-a")?.map(|stored| stored.transaction_id),
+            Some("tx-a".to_owned())
+        );
+        assert_eq!(std::fs::metadata(&path)?.permissions().mode() & 0o777, 0o600);
+        // A healthy store is not set aside again.
+        assert_eq!(record(&path, &record_at("tx-b", 6))?, Kept::Stored);
+        assert_eq!(std::fs::read(&aside)?, damage);
+        Ok(())
+    }
+
+    #[test]
+    fn a_future_version_store_is_set_aside_and_the_next_failure_keeps_its_evidence() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        write_future_store(&path)?;
+        let before = std::fs::read(&path)?;
+        assert!(for_transaction(&path, "tx-future").is_err());
+        assert_eq!(
+            record(&path, &record_at("tx-a", 5))?,
+            Kept::SetAside { replaced: None }
+        );
+        assert_eq!(std::fs::read(unreadable_store_path(&path))?, before);
+        assert!(for_transaction(&path, "tx-a")?.is_some());
+        assert_eq!(for_transaction(&path, "tx-future")?, None);
+        Ok(())
+    }
+
+    #[test]
+    fn at_most_one_unreadable_store_is_kept() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        let damages: [&[u8]; 3] = [b"\xc1 first", b"\xc1 second", b"\xc1 third"];
+        for (index, damage) in damages.into_iter().enumerate() {
+            std::fs::write(&path, damage)?;
+            record(&path, &record_at(&format!("tx-{index}"), 5 + index as u64))?;
+            assert_eq!(
+                std::fs::read(unreadable_store_path(&path))?,
+                damage,
+                "the newest damage is the one kept"
+            );
+        }
+        assert_eq!(
+            files_beside(&path)?,
+            vec![
+                "runner-evidence.cc".to_owned(),
+                "runner-evidence.unreadable.cc".to_owned()
+            ]
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn a_store_that_cannot_be_read_is_left_in_place_and_the_write_fails() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        // A directory where the store should be: unreadable, not undecodable.
+        std::fs::create_dir(&path)?;
+        assert!(record(&path, &record_at("tx-a", 1)).is_err());
+        assert!(path.is_dir());
+        assert!(!unreadable_store_path(&path).exists());
+        Ok(())
+    }
+
+    #[test]
+    fn every_code_point_but_printable_ascii_is_a_visible_escape() {
+        // The allowlist, exhaustively: no class list decides anything.
+        for code in 0..=0x10ffffu32 {
+            let Some(character) = char::from_u32(code) else {
+                continue;
+            };
+            let text = character.to_string();
+            let shown = escape(text.as_bytes());
+            if matches!(character, ' '..='~' | '\t') {
+                assert_eq!(shown, text, "U+{code:04X} should pass");
+            } else {
+                assert_eq!(shown, format!("\\u{{{code:x}}}"), "U+{code:04X}");
+            }
+        }
+        // Invalid UTF-8 is escaped byte by byte; valid text round it passes.
+        assert_eq!(escape(b"a\xc1\xffb\xe2\x82"), "a\\xc1\\xffb\\xe2\\x82");
+        assert_eq!(escape(b"ok \xe2\x82\xac"), "ok \\u{20ac}");
+    }
+
+    #[test]
+    fn a_newline_inside_a_line_cannot_forge_a_line() {
+        let mut stored = record_at("tx-a", 1);
+        stored.subject_id = "build\n    status exit status 0".into();
+        let lines = shown_lines(&stored);
+        assert_eq!(lines[0], "    failed step build\\u{a}    status exit status 0");
+        assert_eq!(
+            lines.iter().filter(|line| line.starts_with("    status ")).count(),
+            1
+        );
+    }
+
+    /// A store whose first record is well-formed except that its subject kind
+    /// holds `marker`: the decode error quotes it.
+    fn store_with_hostile_subject_kind(path: &Path, marker: &[u8; 4]) -> Result<()> {
+        record(path, &record_at("tx-a", 1))?;
+        let mut bytes = std::fs::read(path)?;
+        let at = bytes
+            .windows(4)
+            .position(|window| window == b"step")
+            .context("no step marker")?;
+        bytes[at..at + 4].copy_from_slice(marker);
+        std::fs::write(path, bytes)?;
+        Ok(())
+    }
+
+    #[test]
+    fn the_unreadable_store_line_is_escaped_with_the_rest() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        store_with_hostile_subject_kind(&path, b"\x1b[2J")?;
+        let mut written = Vec::new();
+        show_status(&mut written, &path, "tx-a")?;
+        assert!(!written.contains(&0x1b), "an escape byte reached the terminal");
+        assert!(written.iter().all(|byte| matches!(byte, b' '..=b'~' | b'\t' | b'\n')));
+        let text = String::from_utf8(written)?;
+        assert!(text.starts_with("    runner evidence unreadable: "), "{text}");
+        assert!(text.contains("\\u{1b}[2J"), "the quoted value was not shown escaped: {text}");
+        Ok(())
+    }
+
+    #[test]
+    fn a_failed_set_aside_names_itself_keeps_the_store_and_echoes_nothing() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        store_with_hostile_subject_kind(&path, b"CNRY")?;
+        let damage = std::fs::read(&path)?;
+        // Sanity: this damage decodes to an error that quotes the canary.
+        assert!(format!("{:#}", snapshot(&path).unwrap_err()).contains("CNRY"));
+        // A non-empty directory where the aside goes: rename fails even as root.
+        let aside = unreadable_store_path(&path);
+        std::fs::create_dir(&aside)?;
+        std::fs::write(aside.join("occupied"), b"x")?;
+        let error = record(&path, &record_at("tx-b", 2)).unwrap_err();
+        for text in [format!("{error:#}"), format!("{error:?}")] {
+            assert!(text.contains("could not be set aside"), "{text}");
+            assert!(!text.contains("CNRY"), "stored content reached the error: {text}");
+        }
+        // The failure cost nothing: the store is as it was, with no evidence
+        // written over it.
+        assert_eq!(std::fs::read(&path)?, damage);
+        assert!(aside.join("occupied").exists());
+        Ok(())
+    }
+
+    #[test]
+    fn an_io_error_leaves_the_store_in_place_and_keeps_no_evidence() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        record(&path, &record_at("tx-a", 1))?;
+        let before = std::fs::read(&path)?;
+        let denied = || std::io::Error::from(std::io::ErrorKind::PermissionDenied);
+        // The first read fails on a healthy store.
+        let error = record_reading(&path, &record_at("tx-b", 2), &|_| Err(denied())).unwrap_err();
+        assert!(format!("{error:#}").contains("permission denied"), "{error:#}");
+        assert_eq!(std::fs::read(&path)?, before, "the store changed");
+        assert!(!unreadable_store_path(&path).exists());
+        assert!(for_transaction(&path, "tx-b")?.is_none());
+        // The second read, the one that confirms the bytes do not decode, fails.
+        std::fs::write(&path, b"\xc1 not a store")?;
+        let calls = std::cell::Cell::new(0usize);
+        let read = |path: &Path| {
+            calls.set(calls.get() + 1);
+            if calls.get() == 2 {
+                return Err(denied());
+            }
+            std::fs::read(path)
+        };
+        let error = record_reading(&path, &record_at("tx-b", 2), &read).unwrap_err();
+        assert!(format!("{error:#}").contains("permission denied"), "{error:#}");
+        assert_eq!(std::fs::read(&path)?, b"\xc1 not a store");
+        assert!(!unreadable_store_path(&path).exists());
+        // Bytes that changed between the reads are no verdict either.
+        let calls = std::cell::Cell::new(0usize);
+        let read = |path: &Path| -> std::io::Result<Vec<u8>> {
+            calls.set(calls.get() + 1);
+            let mut bytes = std::fs::read(path)?;
+            bytes.push(calls.get() as u8);
+            Ok(bytes)
+        };
+        assert!(record_reading(&path, &record_at("tx-b", 2), &read).is_err());
+        assert_eq!(std::fs::read(&path)?, b"\xc1 not a store");
+        assert!(!unreadable_store_path(&path).exists());
+        Ok(())
+    }
+
+    #[test]
+    fn a_second_set_aside_says_it_replaced_the_first() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let path = runner_evidence_store_path(&temp.path().join("control.cc"));
+        std::fs::write(&path, b"\xc1 first")?;
+        assert_eq!(
+            record(&path, &record_at("tx-a", 1))?,
+            Kept::SetAside { replaced: None }
+        );
+        std::fs::write(&path, b"\xc1 second")?;
+        let Kept::SetAside { replaced: Some(_) } = record(&path, &record_at("tx-b", 2))? else {
+            bail!("a displaced set-aside store went unreported");
+        };
+        assert_eq!(std::fs::read(unreadable_store_path(&path))?, b"\xc1 second");
+        Ok(())
+    }
+}