session: parallel processes, a launcher and the fault behaviour

SESSION-02: one agent process per fly and one environment process under
the coordinator over the Unix-socket transport, compared against the
in-process composition and a dedicated-thread variant. The mode is the
only thing that changes; the composition, the coordinator, the workers
and the router are the same code in all three.

The launcher is the configured supervisor. It owns a total thread budget
with one allocation per participant, refused as BUSY before anything
starts when the total cannot cover it; the configured client, service,
worker and port identities, proved in Worker.Hello before the
coordinator pins a registration; Worker.Status health on the
supervisor's own monotonic clock at the ipc-v1 section 6 budgets; and
reaping, where Worker.Shutdown is the request and the operating system
is the guarantee.

The worker executable is a subcommand of this crate's one binary, which
is what implementation.md section 2 allows in place of a separate
worker crate.

The coordinator's fault behaviour: every failure names the participant
it is attributed to, every domain call has a caller-side deadline so a
dead participant is a diagnosed outcome rather than a hang, and failing
fences the epoch -- the boundary stops, the handles drop, and no further
transition or publication is allowed. Agent.Initialize now carries the
launcher's allocation, and an agent refuses one asking for more.

tests/processes.rs proves every acceptance bullet once per execution
mode, and the two section 4 rows SESSION-01 could not reach in one
process: a router restart during a world advance, and an old worker's
reply after a restart. measure compares the three modes at one, two and
four agents; its table is in the crate README, and it is not a capacity
claim.

services/flysim/crates/fly-session/Cargo.toml

@@ -11,6 +11,13 @@ description = "The lockstep session coordinator, its phase machine and a synthet
 name = "fly_session"
 path = "src/lib.rs"
 
+# One binary, one role per subcommand. `implementation.md` section 2 allows worker
+# executables to be subcommands of one binary rather than separate crates, and the launcher
+# starts this one with `agent` or `environment` for a participant in its own process.
+[[bin]]
+name = "fly-session"
+path = "src/bin/fly-session.rs"
+
 [dependencies]
 # The domain contract (scalars, payloads, canonical digests, the trace format) and the bus.
 # Everything else this crate needs is std or Tokio.
@@ -18,7 +25,9 @@ fly-session-types = { path = "../fly-session-types" }
 flybus = { path = "../flybus" }
 
 serde_json = { workspace = true }
-tokio = { version = "1", features = ["rt", "sync", "time", "macros"] }
+# `rt-multi-thread` is not only for the tests: a worker process and a dedicated-thread
+# worker each build their own runtime sized to the launcher's thread allocation.
+tokio = { version = "1", features = ["rt", "rt-multi-thread", "sync", "time", "macros", "io-util"] }
 
 [dev-dependencies]
 tempfile = "3"

services/flysim/crates/fly-session/README.md

@@ -3,10 +3,12 @@
 The lockstep session coordinator, its phase machine and a synthetic composition over
 [`flybus`](../flybus).
 
-This crate is the SESSION-01 slice of the session-framework implementation guide: the
+This crate is the SESSION-01 and SESSION-02 slices of the session-framework implementation
-sequential transaction of `step-v1`, driven over the Flybus router, with small fake workers
+guide: the transaction of `step-v1`, driven over the Flybus router, with small fake workers
-standing in for a brain and an emulator. It contains no public controller API, no implicit
+standing in for a brain and an emulator, run either in the coordinator's process, on dedicated
-best-effort retry, no real emulator and no real brain.
+threads, or as one agent process per fly and one environment process under a launcher. It
+contains no public controller API, no implicit best-effort retry, no real emulator and no real
+brain.
 
 The domain scalars, method payloads, their validation, the canonical digests and the trace
 format all come from [`fly-session-types`](../fly-session-types), the CONTRACT-01 crate. This
@@ -38,7 +40,51 @@ Ready(k) ─ Prepare all agents concurrently ───────────
 | `task` | The task and executor traits, the deterministic counter task, the identity executor |
 | `rpc` | Domain calls: `req-<U64>` serials, incarnation pinning, the retry rule |
 | `coordinator` | The transaction, the trace, the failure rules and the publication boundary |
-| `harness` | The runnable composition: router, two agents, one arena, one coordinator |
+| `launcher` | The supervisor: thread budget, identities, start, health check, reap |
+| `metrics` | Latency percentiles and the machine's core and memory counters |
+| `measure` | The execution-mode comparison of the guide's section 5 |
+| `cli` | The binary's subcommands: `agent`, `environment`, `measure` |
+| `harness` | The runnable composition: router, the flies, one arena, one coordinator |
+
+## Execution modes and the launcher
+
+A participant runs in one of three places, and the same composition code starts it in any of
+them. The separate-process mode is the SESSION-02 subject; the other two are what it is
+compared against.
+
+| Mode | Where each participant runs | Transport |
+| --- | --- | --- |
+| `InProcess` | A task on the coordinator's runtime | in-memory or Unix socket |
+| `Thread` | Its own OS thread, with its own runtime | Unix socket |
+| `Process` | Its own process: one per fly, one for the world | Unix socket |
+
+The launcher is the configured supervisor. It owns four things:
+
+- **The thread budget.** A total allocation, one slice of it reserved for the coordinator and
+  its router, and one allocation per participant. A request the total cannot cover is refused
+  as `BUSY` before anything starts. `Agent.Initialize` carries exactly the allocation the
+  launcher handed out, and an agent refuses an Initialize asking for more than its own, which
+  is what `workers-v1` means by "within launcher allocation".
+- **Identity.** The bus client id, the service name, the worker id and an agent's port binding
+  are launcher configuration. The launcher says `Worker.Hello` with the identity it configured
+  and refuses anything that answers as another worker, role or incarnation -- before the
+  coordinator has pinned a registration. The registration the coordinator pins is the one that
+  hello returned, never one that was assumed.
+- **Health.** `Worker.Status` on the supervisor's own monotonic clock, with the `ipc-v1`
+  section 6 prototype budgets: probe at two seconds, fail at ten, a separate budget for boot.
+  A status answer never waits for a mutation, so a busy participant is still a healthy one.
+- **Reaping.** `Worker.Shutdown` is the request and the operating system is the guarantee. A
+  participant that does not stop inside the budget is terminated, and the supervisor reports
+  which of the two happened. A launcher that is dropped takes its children with it.
+
+A separate-process participant is a subcommand of this crate's one binary, which is what
+`implementation.md` section 2 allows instead of separate worker crates:
+
+```sh
+fly-session agent       --socket S --store-root D --client-id C --service N --threads T ...
+fly-session environment --socket S --store-root D --client-id C --service N --threads T ...
+fly-session measure     --steps 300 --agents 1,2,4
+```
 
 ## What it implements
 
@@ -62,6 +108,14 @@ Ready(k) ─ Prepare all agents concurrently ───────────
 - **The failure rules.** A partial commit fails the epoch; an uncertain Advance is resolved
   against its original domain request id and never becomes a second batch; a worker
   incarnation change invalidates the epoch.
+- **A failure stops the epoch rather than neutralising a player.** Every failure carries the
+  participant it is attributed to, and failing fences the session: the committed boundary
+  stops moving, the artifact handles are dropped, and no further transition or publication is
+  allowed. Lifting the fence is a coherent group restore, which is STATE-01's.
+- **A bounded diagnosed outcome.** A caller-side deadline on every domain call, on the
+  coordinator's own clock, so a participant that dies or stops answering produces a typed
+  failure naming it rather than a hang. An expired deadline is `unknown`, never `none`: a
+  caller-side timeout is not evidence that nothing was mutated.
 - **Domain deduplication over bus calls.** Same key, request and body replays its cached
   reply with fresh delivery ownership over retained artifacts; a changed body is `CONFLICT`; a
   duplicate of a running operation is `IN_PROGRESS` for that bus call while the original
@@ -125,17 +179,41 @@ harness.shutdown().await;
 - **No state methods.** `State.Capture`, `State.StageRestore` and `State.ActivateRestore` are
   STATE-01. The phase machine has their edges (`Capturing`, `Restoring`) and the workers do not
   advertise them as implemented methods.
-- **One process.** SESSION-01 runs every participant in one process over the same router.
-  SESSION-02 is the per-fly process split.
 - **No audience input.** The admitted pre-step stimulation list exists and is always empty.
 - **Pacing is coarse.** The pacing deadline rounds one step to whole nanoseconds for sleeping
   only; simulation time stays rational and that rounding never re-enters the accumulator.
 
+## Measurements
+
+`fly-session measure` runs the same composition in each mode at one, two and four agents and
+reports the thread allocation, the RPC and critical-path percentiles, the memory peaks and the
+router's owner, collection and queue counters. **These are local synthetic timings on one
+machine and no host capacity claim follows from any of them**; they exist so the three modes
+can be compared with each other. Pacing is off for the run, so the samples are work rather
+than sleep, and the run report carries the full table.
+
+What the numbers said on a four-core development box, at 300 transitions per row:
+
+- A process boundary costs little at the median and shows up in the tail. Two agents: the
+  critical path was about 7.8 ms p50 in-process, 8.6 ms on threads and 12.7 ms across
+  processes, while p99 went 12.4 / 12.7 / 26.0 ms. The medians are within a small multiple of
+  each other; the tails are where a scheduler with more runnable threads than cores appears.
+- Four agents needs six threads, which that box does not have, and every mode's tail widens
+  together. That is the budget being honest, not a property of the process split.
+- Memory is the clearest difference: one coordinator at about 14 MiB peak RSS plus roughly
+  5.6 MiB per participant process, against a single 11 MiB process for the threaded variant.
+- Ownership and queues stayed bounded in every mode and at every agent count: at most 15 live
+  owners, 11 artifact roots and one queue entry per agent, with the store holding two sealed
+  frames and 128 bytes at rest. Of 311 frames produced, 309 were collected -- the current and
+  previous boundary are the two that are still owned.
+
 ## Tests
 
 ```text
-cargo test -p fly-session                                    # unit + both integration suites
+cargo test -p fly-session                                    # unit + all three integration suites
 cargo run -p fly-session --example session                   # the runnable synthetic session
+cargo build -p fly-session --bin fly-session                 # the worker binary the launcher starts
+cargo run -p fly-session --example processes                 # the same session in all three modes
 ```
 
 Every integration test runs over both transports, through the same router code: all but one
@@ -150,6 +228,13 @@ because it compares their behaviour traces against each other.
   transition that just ended; a terminal episode pausing at its own boundary; `Worker.Status`
   during a session; and sequential, concurrent and reversed dispatch producing one behaviour
   trace.
+- `tests/processes.rs`: the SESSION-02 acceptance bullets, each generated once per execution
+  mode -- a delayed one-agent result holding the world, a worker or helper death with a
+  bounded diagnosed outcome, an uncertain Advance that creates no second batch, a partial
+  Commit that permits no next-step play, supervision and identity, and the launcher thread
+  allocation -- plus the sequential/reversed/parallel trace comparison across all three modes
+  and the two process-mode section 4 rows: a router restart during a world advance, and an old
+  worker's reply after a restart.
 - `tests/failures.rs`: a duplicate Prepare after a lost reply; a duplicate Commit; the same
   batch with altered controls; a lost Advance result; a cached artifact consumed by its first
   caller; one Commit failing after another succeeded; a replaced registration; a reply from

services/flysim/crates/fly-session/examples/processes.rs

@@ -0,0 +1,75 @@
+//! The same synthetic session in all three execution modes, printing the one behaviour trace
+//! they agree on.
+//!
+//! ```sh
+//! cargo run -p fly-session --example processes
+//! ```
+//!
+//! The separate-process run starts one agent process per fly and one environment process
+//! through this crate's own binary, so it needs that binary built:
+//!
+//! ```sh
+//! cargo build -p fly-session --bin fly-session
+//! ```
+
+use fly_session::harness::{ExecutionMode, HarnessConfig, SessionHarness, Via};
+use fly_session::launcher::default_worker_program;
+
+#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let program = default_worker_program();
+    println!("worker program: {}", program.display());
+    let mut agreed: Option<Vec<String>> = None;
+
+    for mode in ExecutionMode::all() {
+        let dir = tempfile::tempdir()?;
+        let config = HarnessConfig { mode, ..HarnessConfig::default() };
+        println!(
+            "\n=== {} : {} threads for {} participants plus the coordinator",
+            mode.label(),
+            config.budget()?.total(),
+            config.agents.len() + 1
+        );
+        let mut harness = SessionHarness::start(Via::Unix, dir.path(), config).await?;
+        harness.coordinator.bootstrap().await?;
+        let reports = harness.coordinator.run(3).await?;
+        for report in &reports {
+            println!("  committed boundary {}", report.boundary);
+        }
+        for (worker_id, status) in harness.launcher.health_check_all().await {
+            match status {
+                Ok(status) => println!(
+                    "  {worker_id}: {:?}, progress {}",
+                    status.state, status.progress_counter
+                ),
+                Err(e) => println!("  {worker_id}: unhealthy: {e}"),
+            }
+        }
+        let behaviour = harness.coordinator.trace.behavior();
+        match &agreed {
+            None => {
+                println!("  behaviour trace, {} transitions:", behaviour.len());
+                for line in &behaviour {
+                    println!("    {line}");
+                }
+                agreed = Some(behaviour);
+            }
+            Some(first) => {
+                assert_eq!(
+                    &behaviour, first,
+                    "{} produced a different behaviour trace",
+                    mode.label()
+                );
+                println!("  behaviour trace: identical to the first run");
+            }
+        }
+        let reaped = harness.launcher.reap_all("example").await;
+        for (worker_id, outcome) in reaped {
+            println!("  reaped {worker_id}: {outcome:?}");
+        }
+        harness.shutdown().await;
+        drop(dir);
+    }
+    println!("\nall three execution modes produced one behaviour trace");
+    Ok(())
+}

services/flysim/crates/fly-session/src/agent.rs

@@ -203,6 +203,9 @@ pub struct AgentConfig {
     pub incarnation_id: Id,
     pub tick_duration: RationalNs,
     pub warmup_ticks: u64,
+    /// The thread allocation the launcher started this worker within. `workers-v1` requires
+    /// `Agent.Initialize`'s `workerThreads` to lie inside it.
+    pub worker_threads: usize,
     pub faults: AgentFaults,
 }
 
@@ -369,6 +372,18 @@ impl FakeAgentWorker {
         if params.worker_threads == 0 {
             return Err(DomainError::invalid("workerThreads must be >= 1"));
         }
+        // `workers-v1`: workerThreads is "within launcher allocation". This worker was started
+        // with that allocation, so a request for more than it is a capacity refusal made
+        // before the model is constructed, not a silent reduction to what is available.
+        if params.worker_threads > self.config.worker_threads as u64 {
+            return Err(DomainError::before(
+                ErrorCode::Busy,
+                format!(
+                    "Agent.Initialize asks for {} worker threads; the launcher allocated {}",
+                    params.worker_threads, self.config.worker_threads
+                ),
+            ));
+        }
         params.initial_decision_context.validate().map_err(DomainError::invalid)?;
         let available = FakeAgentWorker::available_actions(&params.initial_decision_context)?;
         // Everything is validated before the model is constructed.

services/flysim/crates/fly-session/src/bin/fly-session.rs

@@ -0,0 +1,8 @@
+//! This crate's one binary. Every role it can take is a subcommand of it.
+//!
+//! `implementation.md` section 2: "Worker executables can be subcommands of one binary
+//! initially; process boundaries do not require separate repos."
+
+fn main() -> std::process::ExitCode {
+    fly_session::cli::main()
+}

services/flysim/crates/fly-session/src/cli.rs

@@ -0,0 +1,251 @@
+//! The subcommands of this crate's one binary.
+//!
+//! `implementation.md` section 2 allows worker executables to be subcommands of one binary
+//! rather than separate crates, and that is what these are: `agent` and `environment` are the
+//! two worker roles a launcher starts as separate processes, and `measure` runs the execution
+//! modes against each other.
+//!
+//! ```text
+//! fly-session agent       --socket S --store-root D --client-id C --service N --threads T ...
+//! fly-session environment --socket S --store-root D --client-id C --service N --threads T ...
+//! fly-session measure     [--steps N] [--agents 1,2,4] [--modes in-process,thread,process]
+//! ```
+//!
+//! A worker process is told exactly which participant it is. It proves that identity in
+//! `Worker.Hello`, so a process started under another one is refused by its own supervisor
+//! before the coordinator has pinned anything.
+
+use std::collections::BTreeMap;
+use std::path::PathBuf;
+use std::process::ExitCode;
+
+use crate::agent::AgentFaults;
+use crate::environment::EnvironmentFaults;
+use crate::launcher::{AgentLaunch, EnvironmentLaunch, ExecutionMode, Started, serve_one};
+use crate::types::*;
+
+const USAGE: &str = "\
+fly-session <command> [options]
+
+  agent          serve one agent worker on a launcher-created endpoint
+  environment    serve the environment worker on a launcher-created endpoint
+  measure        compare the execution modes and print the measurement table
+
+Worker options (agent and environment):
+  --socket PATH        the launcher's endpoint for this participant
+  --store-root PATH    the router's artifact store root
+  --client-id ID       the configured bus client identity
+  --service NAME       the one service name this worker registers
+  --threads N          the launcher's thread allocation for this worker
+  --session ID         the session this worker belongs to
+  --incarnation ID     this worker's domain incarnation
+
+  agent:        --agent ID --port ID --tick-numerator N --tick-denominator N
+                --warmup-ticks N [--prepare-delay-ms N] [--commit-delay-ms N]
+                [--fail-commit-at-step N]
+  environment:  --worker ID --ports p1,p2 --step-numerator N --step-denominator N
+                [--advance-delay-ms N] [--omit-view-at-boundary N]
+
+Measure options:
+  --steps N            transitions per run (default 200)
+  --agents 1,2,4       agent counts to compare (default 1,2,4)
+  --modes LIST         in-process, thread, process (default all three)
+  --worker-threads N   within-agent worker threads (default 1)
+";
+
+/// The binary's entry point.
+pub fn main() -> ExitCode {
+    let mut args = std::env::args_os().skip(1);
+    let Some(command) = args.next() else {
+        eprint!("{USAGE}");
+        return ExitCode::from(2);
+    };
+    let command = command.to_string_lossy().into_owned();
+    let rest: Vec<String> = args.map(|a| a.to_string_lossy().into_owned()).collect();
+    let result = match command.as_str() {
+        "agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)),
+        "measure" => Options::parse(&rest).and_then(|o| measure(&o)),
+        "--help" | "-h" | "help" => {
+            print!("{USAGE}");
+            return ExitCode::SUCCESS;
+        }
+        other => Err(format!("unknown command {other:?}\n\n{USAGE}")),
+    };
+    match result {
+        Ok(()) => ExitCode::SUCCESS,
+        Err(e) => {
+            eprintln!("fly-session {command}: {e}");
+            ExitCode::FAILURE
+        }
+    }
+}
+
+/// `--flag value` options. The launcher builds this argv, so the grammar stays small.
+#[derive(Debug, Default)]
+struct Options(BTreeMap<String, String>);
+
+impl Options {
+    fn parse(args: &[String]) -> Result<Options, String> {
+        let mut out = BTreeMap::new();
+        let mut iter = args.iter();
+        while let Some(flag) = iter.next() {
+            let Some(name) = flag.strip_prefix("--") else {
+                return Err(format!("expected an option, found {flag:?}"));
+            };
+            let value = iter
+                .next()
+                .ok_or_else(|| format!("option --{name} needs a value"))?;
+            if out.insert(name.to_owned(), value.clone()).is_some() {
+                return Err(format!("option --{name} was given twice"));
+            }
+        }
+        Ok(Options(out))
+    }
+
+    fn required(&self, name: &str) -> Result<&str, String> {
+        self.0
+            .get(name)
+            .map(String::as_str)
+            .ok_or_else(|| format!("option --{name} is required"))
+    }
+
+    fn optional(&self, name: &str) -> Option<&str> {
+        self.0.get(name).map(String::as_str)
+    }
+
+    fn id(&self, name: &str) -> Result<Id, String> {
+        parse_id(self.required(name)?).map_err(|e| format!("--{name}: {e}"))
+    }
+
+    fn u64(&self, name: &str, default: u64) -> Result<u64, String> {
+        match self.0.get(name) {
+            None => Ok(default),
+            Some(value) => value.parse().map_err(|_| format!("--{name}: {value:?} is not a number")),
+        }
+    }
+
+    fn opt_u64(&self, name: &str) -> Result<Option<u64>, String> {
+        match self.0.get(name) {
+            None => Ok(None),
+            Some(value) => value
+                .parse()
+                .map(Some)
+                .map_err(|_| format!("--{name}: {value:?} is not a number")),
+        }
+    }
+
+    fn usize(&self, name: &str, default: usize) -> Result<usize, String> {
+        Ok(self.u64(name, default as u64)? as usize)
+    }
+
+    fn path(&self, name: &str) -> Result<PathBuf, String> {
+        Ok(PathBuf::from(self.required(name)?))
+    }
+
+    fn rational(&self, numerator: &str, denominator: &str) -> Result<RationalNs, String> {
+        let n = self.u64(numerator, 0)?;
+        let d = self.u64(denominator, 1)?;
+        RationalNs::new(n, d).map_err(|e| format!("--{numerator}/--{denominator}: {}", e.0))
+    }
+}
+
+/// Serves one worker until `Worker.Shutdown`, then exits.
+fn serve(role: &str, options: &Options) -> Result<(), String> {
+    let socket = options.path("socket")?;
+    let store_root = options.path("store-root")?;
+    let client_id = options.required("client-id")?.to_owned();
+    let service = options.required("service")?.to_owned();
+    let threads = options.usize("threads", 1)?;
+    if threads == 0 {
+        return Err("--threads must be at least 1".to_owned());
+    }
+    let session_id = options.id("session")?;
+    let incarnation_id = options.id("incarnation")?;
+    let what = match role {
+        "agent" => Started::Agent(AgentLaunch {
+            session_id,
+            agent_id: options.id("agent")?,
+            port_id: options.id("port")?,
+            incarnation_id,
+            tick_duration: options.rational("tick-numerator", "tick-denominator")?,
+            warmup_ticks: options.u64("warmup-ticks", 0)?,
+            worker_threads: threads,
+            faults: AgentFaults {
+                fail_commit_at_step: options.opt_u64("fail-commit-at-step")?,
+                prepare_delay_ms: options.u64("prepare-delay-ms", 0)?,
+                commit_delay_ms: options.u64("commit-delay-ms", 0)?,
+            },
+            client_id: client_id.clone(),
+            service: service.clone(),
+        }),
+        _ => Started::Environment(EnvironmentLaunch {
+            session_id,
+            worker_id: options.id("worker")?,
+            incarnation_id,
+            step_duration: options.rational("step-numerator", "step-denominator")?,
+            ports: parse_ports(options.required("ports")?)?,
+            worker_threads: threads,
+            faults: EnvironmentFaults {
+                advance_delay_ms: options.u64("advance-delay-ms", 0)?,
+                omit_view_at_boundary: options.opt_u64("omit-view-at-boundary")?,
+            },
+            client_id: client_id.clone(),
+            service: service.clone(),
+        }),
+    };
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(threads)
+        .enable_all()
+        .build()
+        .map_err(|e| format!("runtime: {e}"))?;
+    runtime.block_on(async move {
+        let handle = serve_one(&socket, &client_id, &service, &store_root, &what, threads).await?;
+        // The worker serves until its supervisor's Worker.Shutdown, which it answers before
+        // it stops. Exiting is then one event, not a race between a reply and a signal.
+        handle.join().await;
+        Ok(())
+    })
+}
+
+fn parse_ports(value: &str) -> Result<Vec<Id>, String> {
+    value
+        .split(',')
+        .filter(|part| !part.is_empty())
+        .map(|part| parse_id(part).map_err(|e| format!("--ports: {e}")))
+        .collect()
+}
+
+/// Runs the execution-mode comparison and prints its table.
+fn measure(options: &Options) -> Result<(), String> {
+    let mut config = crate::measure::MeasureConfig {
+        steps: options.u64("steps", 200)?,
+        worker_threads: options.usize("worker-threads", 1)?,
+        ..crate::measure::MeasureConfig::default()
+    };
+    if let Some(list) = options.optional("agents") {
+        config.agent_counts = list
+            .split(',')
+            .filter(|p| !p.is_empty())
+            .map(|p| p.parse::<usize>().map_err(|_| format!("--agents: {p:?}")))
+            .collect::<Result<Vec<usize>, String>>()?;
+    }
+    if let Some(list) = options.optional("modes") {
+        config.modes = list
+            .split(',')
+            .filter(|p| !p.is_empty())
+            .map(|p| match p {
+                "in-process" => Ok(ExecutionMode::InProcess),
+                "thread" => Ok(ExecutionMode::Thread),
+                "process" => Ok(ExecutionMode::Process),
+                other => Err(format!("--modes: {other:?}")),
+            })
+            .collect::<Result<Vec<ExecutionMode>, String>>()?;
+    }
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .map_err(|e| format!("runtime: {e}"))?;
+    let rows = runtime.block_on(crate::measure::run(&config))?;
+    print!("{}", crate::measure::table(&rows));
+    Ok(())
+}

services/flysim/crates/fly-session/src/coordinator.rs

@@ -10,10 +10,12 @@
 //! domain request id, and anything that cannot be resolved fails the epoch.
 
 use std::collections::{BTreeMap, BTreeSet};
+use std::time::{Duration, Instant};
 
 use serde_json::{Map, Value, json};
 
 use crate::clock::Pacing;
+use crate::metrics::Metrics;
 use crate::phase::{Phase, PhaseMachine};
 use crate::rpc::{self, DomainReply, Serials, WorkerRef};
 use crate::task::{ActionExecutor, Task};
@@ -84,11 +86,20 @@ pub struct SessionFailure {
     pub error: DomainError,
     pub phase: String,
     pub detail: String,
+    /// The participant the failure is attributed to, where one is.
+    ///
+    /// `step-v1` section 7 stops the epoch rather than neutralising a player, so a diagnosed
+    /// outcome has to say which participant it was: the coordinator names it here instead of
+    /// leaving a caller to read it out of a message.
+    pub participant: Option<Id>,
 }
 
 impl std::fmt::Display for SessionFailure {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{} at {}: {}", self.detail, self.phase, self.error)
+        match &self.participant {
+            Some(who) => write!(f, "{} at {} ({who}): {}", self.detail, self.phase, self.error),
+            None => write!(f, "{} at {}: {}", self.detail, self.phase, self.error),
+        }
     }
 }
 
@@ -96,6 +107,24 @@ impl std::error::Error for SessionFailure {}
 
 type Outcome<T> = Result<T, SessionFailure>;
 
+/// How long the coordinator waits for a participant before it calls the call uncertain.
+///
+/// `ipc-v1` section 6 measures these on the caller's own monotonic clock and gives the
+/// prototype values: ten seconds without progress is a failure, with a separate budget for a
+/// long boot. They are failure-detection values, not a gameplay latency goal. Without them a
+/// dead participant is a hang rather than a diagnosed outcome.
+#[derive(Clone, Copy, Debug)]
+pub struct Deadlines {
+    pub call: Duration,
+    pub boot: Duration,
+}
+
+impl Default for Deadlines {
+    fn default() -> Deadlines {
+        Deadlines { call: Duration::from_secs(10), boot: Duration::from_secs(30) }
+    }
+}
+
 /// The bus addresses this session publishes on. Chosen by the composition, not the router.
 #[derive(Clone, Debug)]
 pub struct Topics {
@@ -121,6 +150,9 @@ pub struct AgentSlot {
     pub port_id: Id,
     pub profile: AssetRef,
     pub seed: i32,
+    /// The thread allocation the launcher gave this agent, which is what `Agent.Initialize`
+    /// asks for. It is within the launcher allocation by construction.
+    pub worker_threads: u64,
     pub tick_duration: RationalNs,
     pub warmup_ticks: u64,
     pub committed_step: u64,
@@ -144,6 +176,7 @@ impl AgentSlot {
             port_id,
             profile,
             seed,
+            worker_threads: 1,
             tick_duration: RationalNs::ZERO,
             warmup_ticks: 0,
             committed_step: 0,
@@ -198,6 +231,16 @@ pub struct Coordinator {
     pub injection_log: Vec<InjectionOutcome>,
     /// How many times an exact duplicate met IN_PROGRESS while resolving an uncertain call.
     pub in_progress_replies: u64,
+    /// The caller-side failure-detection budgets of `ipc-v1` section 6.
+    pub deadlines: Deadlines,
+    /// Per-method and critical-path latency samples. Local synthetic timings, never a
+    /// capacity claim.
+    pub metrics: Metrics,
+    /// The participant the next failure is attributed to, set around each call to one.
+    blame: Option<Id>,
+    /// Set when the epoch failed: every old handle, route and reply is invalid from here on
+    /// and only a coherent restore may lift it.
+    fenced: bool,
     started: std::time::Instant,
     last_advance_request: Option<DomainRequestId>,
     last_commit_requests: Vec<TraceRequest>,
@@ -246,6 +289,10 @@ impl Coordinator {
             injections: Injections::default(),
             injection_log: Vec::new(),
             in_progress_replies: 0,
+            deadlines: Deadlines::default(),
+            metrics: Metrics::default(),
+            blame: None,
+            fenced: false,
             started: std::time::Instant::now(),
             last_advance_request: None,
             last_commit_requests: Vec::new(),
@@ -312,6 +359,16 @@ impl Coordinator {
         self.pause.load(std::sync::atomic::Ordering::SeqCst)
     }
 
+    /// Stops pacing to the wall clock, so transitions follow one another as fast as the
+    /// participants answer.
+    ///
+    /// Only one pacing authority is ever active; this removes the coordinator's. It is for a
+    /// measurement run, where a 60 Hz sleep would be most of every sample and none of it the
+    /// thing being compared. A session that presents to anyone keeps its pacing.
+    pub fn disable_pacing(&mut self) {
+        self.pacing = None;
+    }
+
     /// Leaves a normal pause at its committed boundary.
     pub fn resume(&mut self) -> Outcome<()> {
         let Phase::Paused(k) = self.phases.phase() else {
@@ -339,12 +396,37 @@ impl Coordinator {
     }
 
     /// Fails the epoch and records the transition to Failed.
+    ///
+    /// The epoch is fenced at the same moment: the session's routes, pinned registrations and
+    /// artifact handles are no longer valid, and nothing but a coherent restore into a new
+    /// epoch may lift that.
     fn fail_now(&mut self, error: DomainError, detail: &str) -> SessionFailure {
         let phase = self.phases.phase().label();
+        let participant = self.blame.take();
         let (from, to) = self.phases.fail();
         self.trace.phase(from, to);
-        self.audit.push(format!("fail:{detail}"));
+        self.fenced = true;
-        SessionFailure { error, phase, detail: detail.to_owned() }
+        self.views.clear();
+        self.pending_views.clear();
+        match &participant {
+            Some(who) => self.audit.push(format!("fail:{detail}:{who}")),
+            None => self.audit.push(format!("fail:{detail}")),
+        }
+        SessionFailure { error, phase, detail: detail.to_owned(), participant }
+    }
+
+    /// Names the participant the next failure belongs to.
+    fn blame(&mut self, who: Option<Id>) {
+        self.blame = who;
+    }
+
+    /// True once the epoch has failed. Old handles and routes are invalid; the session takes
+    /// no further step and publishes nothing.
+    ///
+    /// Lifting the fence is STATE-01's: a group restore into a fresh epoch from a coherent
+    /// checkpoint. This slice only establishes it.
+    pub fn is_fenced(&self) -> bool {
+        self.fenced
     }
 
     fn agent(&self, agent_id: &Id) -> Option<&AgentSlot> {
@@ -571,7 +653,9 @@ impl Coordinator {
             seed: self.agents[index].seed,
             initial_input: self.sensory_input(&observation, 0),
             initial_decision_context: self.agents[index].context.clone(),
-            worker_threads: 1,
+            // The launcher's allocation for this agent. `workers-v1` requires it to lie
+            // within that allocation, and the worker refuses anything larger.
+            worker_threads: self.agents[index].worker_threads,
         };
         let attachments = self.view_attachments();
         let scope = self.scope(0);
@@ -707,6 +791,13 @@ struct Job {
 }
 
 /// Issues one domain call with owned arguments, so it can run in its own task.
+///
+/// The deadline is the caller's, on the caller's monotonic clock. A participant that has died
+/// mid-call is usually reported by the bus itself, because its connection took its
+/// registration with it; this bound is what makes the remaining cases -- a live process that
+/// stopped answering -- a diagnosed outcome rather than a hang. An expired deadline is
+/// deliberately `unknown`: `ipc-v1` section 6 forbids reading a caller-side timeout as proof
+/// that nothing was mutated.
 #[allow(clippy::too_many_arguments)]
 async fn call_owned(
     bus: flybus::Client,
@@ -717,10 +808,33 @@ async fn call_owned(
     attachments: Vec<(String, flybus::Artifact)>,
     request_id: DomainRequestId,
     want: Vec<String>,
+    deadline: Duration,
 ) -> Result<DomainReply, DomainError> {
     let refs: Vec<(&str, &flybus::Artifact)> =
         attachments.iter().map(|(n, a)| (n.as_str(), a)).collect();
-    rpc::call(&bus, &worker, method, scope, params, &refs, request_id, &want).await
+    let call = rpc::call(&bus, &worker, method, scope, params, &refs, request_id, &want);
+    match tokio::time::timeout(deadline, call).await {
+        Ok(result) => result,
+        Err(_) => Err(DomainError::new(
+            ErrorCode::BackendFailure,
+            format!(
+                "{method}: {} did not answer within {:?}",
+                worker.worker_id, deadline
+            ),
+            MutationCertainty::Unknown,
+        )),
+    }
+}
+
+/// One per-agent call's result, in the shape the phase loops read it.
+struct JobResult {
+    agent_id: Id,
+    reply: Result<DomainReply, DomainError>,
+    scope: Option<Scope>,
+    worker: WorkerRef,
+    method: &'static str,
+    /// How long the call took, for the section 5 percentiles.
+    elapsed: Duration,
 }
 
 impl Coordinator {
@@ -740,6 +854,14 @@ impl Coordinator {
             .iter()
             .map(|(n, a)| ((*n).to_owned(), (*a).clone()))
             .collect();
+        // Whatever goes wrong from here until the reply is checked belongs to this worker.
+        self.blame(Some(worker.worker_id.clone()));
+        let deadline = if method.ends_with("Initialize") || method == "Worker.Hello" {
+            self.deadlines.boot
+        } else {
+            self.deadlines.call
+        };
+        let started = Instant::now();
         let reply = call_owned(
             self.bus.clone(),
             worker.clone(),
@@ -749,15 +871,21 @@ impl Coordinator {
             owned,
             request_id,
             want.to_vec(),
+            deadline,
         )
         .await;
+        self.metrics.record(method, started.elapsed());
         let reply = match reply {
             Ok(reply) => reply,
             Err(e) => return Err(self.fail_now(e, method)),
         };
         self.check_reply(worker, &reply, &scope, method)?;
         match reply.result() {
-            Ok(_) => Ok(reply),
+            Ok(reply_value) => {
+                let _ = reply_value;
+                self.blame(None);
+                Ok(reply)
+            }
             Err(e) => Err(self.fail_now(e, method)),
         }
     }
@@ -773,6 +901,7 @@ impl Coordinator {
         scope: &Option<Scope>,
         method: &'static str,
     ) -> Outcome<()> {
+        self.blame(Some(worker.worker_id.clone()));
         let (worker_id, incarnation, echoed) = match &reply.outcome {
             SessionRpcOutcome::Success(s) => {
                 (&s.worker_id, &s.incarnation_id, &s.scope)
@@ -833,6 +962,8 @@ impl Coordinator {
         // The original may still be running, which answers IN_PROGRESS for this bus call and
         // starts no second mutation. Waiting and asking again is the resolution, not a retry
         // of the operation.
+        self.blame(Some(worker.worker_id.clone()));
+        let deadline = self.deadlines.call;
         for attempt in 0..200u32 {
             let reply = call_owned(
                 self.bus.clone(),
@@ -843,6 +974,7 @@ impl Coordinator {
                 attachments.clone(),
                 request_id.clone(),
                 want.to_vec(),
+                deadline,
             )
             .await;
             let reply = match reply {
@@ -851,7 +983,10 @@ impl Coordinator {
             };
             self.check_reply(worker, &reply, &scope, method)?;
             match reply.result() {
-                Ok(_) => return Ok(reply),
+                Ok(_) => {
+                    self.blame(None);
+                    return Ok(reply);
+                }
                 Err(e) if e.code == ErrorCode::InProgress => {
                     let _ = attempt;
                     self.in_progress_replies += 1;
@@ -884,6 +1019,7 @@ impl Coordinator {
         expected: Option<&Value>,
         what: &str,
     ) {
+        let deadline = self.deadlines.call;
         let reply = call_owned(
             self.bus.clone(),
             worker.clone(),
@@ -893,6 +1029,7 @@ impl Coordinator {
             attachments,
             request_id,
             Vec::new(),
+            deadline,
         )
         .await;
         let outcome = match reply {
@@ -943,6 +1080,7 @@ impl Coordinator {
             Vec::new(),
             request_id,
             Vec::new(),
+            self.deadlines.call,
         )
         .await?;
         reply.result().cloned()
@@ -964,11 +1102,8 @@ impl Coordinator {
     }
 
     /// Runs one set of per-agent jobs in the configured dispatch order.
-    async fn run_jobs(
+    async fn run_jobs(&mut self, jobs: Vec<Job>, order: DispatchOrder) -> Vec<JobResult> {
-        &mut self,
+        let deadline = self.deadlines.call;
-        jobs: Vec<Job>,
-        order: DispatchOrder,
-    ) -> Vec<(Id, Result<DomainReply, DomainError>, Option<Scope>, WorkerRef, &'static str)> {
         let mut out = Vec::new();
         match order {
             DispatchOrder::Sequential | DispatchOrder::Reversed => {
@@ -977,6 +1112,7 @@ impl Coordinator {
                     jobs.reverse();
                 }
                 for job in jobs {
+                    let started = Instant::now();
                     let reply = call_owned(
                         self.bus.clone(),
                         job.worker.clone(),
@@ -986,9 +1122,17 @@ impl Coordinator {
                         job.attachments,
                         job.request_id,
                         Vec::new(),
+                        deadline,
                     )
                     .await;
-                    out.push((job.agent_id, reply, job.scope, job.worker, job.method));
+                    out.push(JobResult {
+                        agent_id: job.agent_id,
+                        reply,
+                        scope: job.scope,
+                        worker: job.worker,
+                        method: job.method,
+                        elapsed: started.elapsed(),
+                    });
                 }
             }
             DispatchOrder::Concurrent => {
@@ -1000,6 +1144,7 @@ impl Coordinator {
                     let scope = job.scope.clone();
                     let method = job.method;
                     tasks.push(tokio::spawn(async move {
+                        let started = Instant::now();
                         let reply = call_owned(
                             bus,
                             job.worker,
@@ -1009,9 +1154,17 @@ impl Coordinator {
                             job.attachments,
                             job.request_id,
                             Vec::new(),
+                            deadline,
                         )
                         .await;
-                        (agent_id, reply, scope, worker, method)
+                        JobResult {
+                            agent_id,
+                            reply,
+                            scope,
+                            worker,
+                            method,
+                            elapsed: started.elapsed(),
+                        }
                     }));
                 }
                 for task in tasks {
@@ -1024,7 +1177,10 @@ impl Coordinator {
         }
         // Completion order never affects anything downstream, so the results are put back in
         // sorted agent-id order here and nowhere else.
-        out.sort_by(|a, b| a.0.cmp(&b.0));
+        out.sort_by(|a, b| a.agent_id.cmp(&b.agent_id));
+        for result in &out {
+            self.metrics.record(result.method, result.elapsed);
+        }
         out
     }
 }
@@ -1032,6 +1188,20 @@ impl Coordinator {
 impl Coordinator {
     /// One complete transition `k -> k+1`.
     pub async fn step(&mut self) -> Outcome<StepReport> {
+        if self.fenced {
+            // A failed epoch's routes, registrations and handles are invalid. There is no
+            // partial continuation: only a coherent restore into a new epoch resumes play.
+            return Err(SessionFailure {
+                error: DomainError::before(
+                    ErrorCode::InvalidPhase,
+                    "the epoch is fenced; only a coherent restore resumes play",
+                ),
+                phase: self.phases.phase().label(),
+                detail: "fenced".to_owned(),
+                participant: None,
+            });
+        }
+        let mut step_started = Instant::now();
         let Phase::Ready(k) = self.phases.phase() else {
             return Err(self.fail_now(
                 DomainError::before(
@@ -1062,6 +1232,9 @@ impl Coordinator {
             // Wall time is only pacing. Being late omits the sleep and is reported; it never
             // skips a world step or a neural tick.
             pacing.wait().await;
+            // The critical path is the transaction, not the sleep in front of it: the
+            // deadline the coordinator was waiting for is pacing, and pacing is not work.
+            step_started = Instant::now();
         }
 
         // ---- Phase A: prepare all agents concurrently
@@ -1172,6 +1345,8 @@ impl Coordinator {
             self.pause.store(false, std::sync::atomic::Ordering::SeqCst);
             self.audit.push(format!("pause:{}", k + 1));
         }
+        // The critical path: one whole transition, pacing sleep included.
+        self.metrics.record("step", step_started.elapsed());
         Ok(StepReport { boundary: k + 1, paused, terminal })
     }
 
@@ -1239,7 +1414,8 @@ impl Coordinator {
         }
         let results = self.run_jobs(jobs, self.dispatch).await;
         let mut prepared = Vec::new();
-        for (agent_id, reply, scope, worker, method) in results {
+        for JobResult { agent_id, reply, scope, worker, method, elapsed: _ } in results {
+            self.blame(Some(agent_id.clone()));
             let reply = match reply {
                 Ok(reply) => reply,
                 // Some agents are already Prepared. Dispatch stops and the epoch fails; a
@@ -1271,6 +1447,7 @@ impl Coordinator {
             }
             self.audit.push(format!("prepared:{agent_id}@{k}"));
             self.stats.prepares += 1;
+            self.blame(None);
             prepared.push((agent_id, decision));
         }
         prepared.sort_by(|a, b| a.0.cmp(&b.0));
@@ -1422,6 +1599,9 @@ impl Coordinator {
         self.last_advance_request = Some(request_id.clone());
         let want = vec!["view.arena".to_owned()];
         self.audit.push(format!("advance:{k}"));
+        self.blame(Some(worker.worker_id.clone()));
+        let advance_deadline = self.deadlines.call;
+        let advance_started = Instant::now();
 
         let injected = self.injections.at_step == k;
         let reply = if injected && self.injections.lose_advance_result {
@@ -1501,8 +1681,10 @@ impl Coordinator {
                 Vec::new(),
                 request_id.clone(),
                 want.clone(),
+                advance_deadline,
             )
             .await;
+            self.metrics.record("Environment.Advance", advance_started.elapsed());
             let reply = match reply {
                 Ok(reply) => reply,
                 Err(e) => return Err(self.fail_now(e, "advance")),
@@ -1518,6 +1700,7 @@ impl Coordinator {
             Ok(result) => result,
             Err(e) => return Err(self.fail_now(e, "advance")),
         };
+        self.blame(None);
         self.pending_views = reply.artifacts;
 
         if injected && self.injections.altered_advance_controls {
@@ -1773,13 +1956,18 @@ impl Coordinator {
         let results = self.run_jobs(jobs, self.dispatch).await;
         let mut commits = Vec::new();
         let mut first_failure = None;
-        for (agent_id, reply, scope, worker, method) in results {
+        let mut blamed: Option<Id> = None;
+        for JobResult { agent_id, reply, scope, worker, method, elapsed: _ } in results {
+            self.blame(Some(agent_id.clone()));
             match reply {
                 Ok(reply) => {
                     self.check_reply(&worker, &reply, &scope, method)?;
                     match reply.result() {
                         Ok(_) => {}
                         Err(e) => {
+                            if first_failure.is_none() {
+                                blamed = Some(agent_id.clone());
+                            }
                             first_failure = Some(first_failure.unwrap_or(e));
                             continue;
                         }
@@ -1812,17 +2000,23 @@ impl Coordinator {
                     }
                     self.audit.push(format!("committed:{agent_id}@{k}"));
                     self.stats.commits += 1;
+                    self.blame(None);
                     commits.push((agent_id, result));
                 }
                 Err(e) => {
+                    if first_failure.is_none() {
+                        blamed = Some(agent_id.clone());
+                    }
                     first_failure = Some(first_failure.unwrap_or(e));
                 }
             }
         }
         if let Some(error) = first_failure {
             // One Commit failed after others succeeded. There is no partial-match
-            // continuation: the epoch is failed and the group recovers together.
+            // continuation: the epoch is failed and the group recovers together, and the
+            // failure names the agent whose Commit failed.
             let _ = bodies;
+            self.blame(blamed);
             return Err(self.fail_now(error, "commit"));
         }
         if commits.len() != self.agents.len() {

services/flysim/crates/fly-session/src/harness.rs

@@ -1,36 +1,37 @@
-//! The runnable synthetic composition: one router, two fake agents, one counter arena and one
+//! The runnable synthetic composition: one router, the configured flies, one counter arena and
-//! coordinator, over either transport.
+//! one coordinator, in whichever execution mode the composition asks for.
 //!
-//! All participants use router semantics even when colocated, so the in-memory and
+//! All participants use router semantics even when colocated, so every mode and both
-//! Unix-socket runs exercise the same code. The caller owns the store root directory, which
+//! transports exercise the same code. The caller owns the store root directory, which keeps
-//! keeps this module free of a temporary-directory dependency.
+//! this module free of a temporary-directory dependency.
+//!
+//! The three execution modes are the SESSION-02 comparison:
+//!
+//! | Mode | Where each participant runs | Transport |
+//! | --- | --- | --- |
+//! | [`ExecutionMode::InProcess`] | A task on the coordinator's runtime | either |
+//! | [`ExecutionMode::Thread`] | Its own OS thread and runtime | Unix socket |
+//! | [`ExecutionMode::Process`] | Its own process, one per fly plus one world | Unix socket |
 
 use std::collections::BTreeMap;
-use std::path::{Path, PathBuf};
+use std::path::Path;
 use std::sync::Mutex;
-use std::sync::atomic::{AtomicU64, Ordering};
 
-use flybus::{
+use flybus::{Client, Grants, Pattern, Policy, Router, RouterConfig};
-    Client, ClientConfig, Grants, Pattern, Policy, Router, RouterConfig, ServiceConfig, Transport,
-    UnixListenerHandle,
-};
 
-use crate::agent::{AgentConfig, AgentFaults, FakeAgentWorker, synthetic_profile};
+use crate::agent::{AgentFaults, synthetic_profile};
 use crate::coordinator::{AgentSlot, Coordinator};
-use crate::environment::{CounterEnvironment, EnvironmentConfig, EnvironmentFaults};
+use crate::environment::EnvironmentFaults;
-use crate::rpc::WorkerRef;
+use crate::launcher::{
+    AgentLaunch, EnvironmentLaunch, Launcher, ReapOutcome, SUPERVISOR_CLIENT, ThreadBudget,
+};
 use crate::task::{ActionExecutor, CounterTask, IdentityExecutor, Terminal};
 // `crate::types` is this crate's facade over the shared `fly-session-types` crate; the
 // glob keeps the contract's own names in sight instead of restating them.
 use crate::types::*;
-use crate::worker::{StatusCell, WorkerHandle, serve};
+use crate::worker::StatusCell;
 
-/// Which transport the session runs over. Both must produce the same behaviour.
+pub use crate::launcher::{ExecutionMode, Via};
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum Via {
-    Memory,
-    Unix,
-}
 
 /// One agent in the composition.
 #[derive(Clone, Debug)]
@@ -41,6 +42,22 @@ pub struct AgentSpec {
     /// derivation algorithm is specified before the real agent slice.
     pub seed: i32,
     pub faults: AgentFaults,
+    /// The threads this agent asks the launcher for. `Agent.Initialize` carries exactly what
+    /// the launcher allocated, which `workers-v1` requires it to lie within.
+    pub worker_threads: usize,
+}
+
+impl AgentSpec {
+    /// One agent on one thread, with no injected fault.
+    pub fn new(agent_id: &str, port_id: &str, seed: i32) -> AgentSpec {
+        AgentSpec {
+            agent_id: id(agent_id),
+            port_id: id(port_id),
+            seed,
+            faults: AgentFaults::default(),
+            worker_threads: 1,
+        }
+    }
 }
 
 /// The composition the harness builds.
@@ -56,6 +73,15 @@ pub struct HarnessConfig {
     pub warmup_ticks: u64,
     pub terminal: Terminal,
     pub environment_faults: EnvironmentFaults,
+    /// Where each participant runs.
+    pub mode: ExecutionMode,
+    /// The total thread allocation the launcher may hand out. `None` sizes it from the
+    /// composition and the machine, which is what an ordinary run wants; a test that means to
+    /// exhaust the budget names a number.
+    pub thread_budget: Option<usize>,
+    /// The threads reserved for the coordinator, its router and its store.
+    pub coordinator_threads: usize,
+    pub environment_threads: usize,
 }
 
 impl Default for HarnessConfig {
@@ -65,31 +91,45 @@ impl Default for HarnessConfig {
             epoch: id("e1"),
             episode_id: id("ep1"),
             agents: vec![
-                AgentSpec {
+                AgentSpec { seed: 7, ..AgentSpec::new("fly-a", "p1", 7) },
-                    agent_id: id("fly-a"),
+                AgentSpec { seed: 11, ..AgentSpec::new("fly-b", "p2", 11) },
-                    port_id: id("p1"),
-                    seed: 7,
-                    faults: AgentFaults::default(),
-                },
-                AgentSpec {
-                    agent_id: id("fly-b"),
-                    port_id: id("p2"),
-                    seed: 11,
-                    faults: AgentFaults::default(),
-                },
             ],
             step_hz: 60,
             tick_ms: 1,
             warmup_ticks: 10,
             terminal: Terminal::Never,
             environment_faults: EnvironmentFaults::default(),
+            mode: ExecutionMode::InProcess,
+            thread_budget: None,
+            coordinator_threads: 1,
+            environment_threads: 1,
         }
     }
 }
 
+impl HarnessConfig {
+    /// The threads this composition needs at a minimum: the coordinator, the world and every
+    /// agent's own allocation.
+    pub fn required_threads(&self) -> usize {
+        self.coordinator_threads
+            + self.environment_threads
+            + self.agents.iter().map(|a| a.worker_threads).sum::<usize>()
+    }
+
+    /// The budget the launcher runs under: what was configured, or a budget that covers both
+    /// this composition and this machine's physical cores.
+    pub fn budget(&self) -> Result<ThreadBudget, DomainError> {
+        let total = self
+            .thread_budget
+            .unwrap_or_else(|| crate::metrics::physical_cores().max(self.required_threads()));
+        ThreadBudget::new(total, self.coordinator_threads)
+    }
+}
+
 const ENV_SERVICE: &str = "env.arena";
 const ENV_CLIENT: &str = "environment";
 const ENV_WORKER: &str = "arena";
+const COORDINATOR_CLIENT: &str = "coordinator";
 
 fn agent_service(agent_id: &Id) -> String {
     format!("agent.{agent_id}")
@@ -105,37 +145,6 @@ fn grants(f: impl FnOnce(&mut Grants)) -> Grants {
     g
 }
 
-/// Makes a connection for one launcher-bound participant, over the chosen transport.
-struct Connector {
-    router: Router,
-    via: Via,
-    store_root: PathBuf,
-    sockets: PathBuf,
-    next_socket: AtomicU64,
-    listeners: Mutex<Vec<UnixListenerHandle>>,
-}
-
-impl Connector {
-    async fn client(&self, id: &str) -> Result<Client, flybus::BusError> {
-        let transport = match self.via {
-            Via::Memory => self.router.connect_in_memory_as(id),
-            Via::Unix => {
-                let n = self.next_socket.fetch_add(1, Ordering::Relaxed);
-                let path = self.sockets.join(format!("{id}-{n}.sock"));
-                let listener = self.router.listen_unix_as(&path, id).await.map_err(|e| {
-                    flybus::BusError::new(flybus::ErrorCode::RouterLost, format!("listen: {e}"))
-                })?;
-                let transport = Transport::unix(&path).await.map_err(|e| {
-                    flybus::BusError::new(flybus::ErrorCode::RouterLost, format!("connect: {e}"))
-                })?;
-                self.listeners.lock().expect("not poisoned").push(listener);
-                transport
-            }
-        };
-        Client::connect(transport, ClientConfig::new(id, &self.store_root)).await
-    }
-}
-
 /// What a restarted worker looks like from the outside: a new registration and a new
 /// incarnation, both different from the ones the coordinator pinned.
 #[derive(Clone, Debug)]
@@ -148,16 +157,17 @@ pub struct Restarted {
 /// A running synthetic session.
 pub struct SessionHarness {
     pub coordinator: Coordinator,
-    pub environment: WorkerHandle,
-    pub agents: BTreeMap<Id, WorkerHandle>,
     pub config: HarnessConfig,
     pub via: Via,
-    connector: Connector,
+    pub mode: ExecutionMode,
+    /// The supervisor. It owns every participant's lifetime and thread allocation.
+    pub launcher: Launcher,
     observers: Mutex<Vec<Client>>,
 }
 
 impl SessionHarness {
-    /// Builds the router, the workers and the coordinator. Nothing has stepped yet.
+    /// Builds the router, launches the workers and builds the coordinator. Nothing has
+    /// stepped yet.
     pub async fn start(
         via: Via,
         root: &Path,
@@ -167,16 +177,29 @@ impl SessionHarness {
         let sockets = root.join("sockets");
         std::fs::create_dir_all(&sockets).expect("the caller owns a writable directory");
 
+        // The launcher's policy: who may connect, and what each may do. Naming a target is not
+        // authority to use it, so the supervisor calls but never registers or publishes, and a
+        // worker registers exactly one service and calls nothing.
         let mut policy = Policy::closed()
             .client(
-                "coordinator",
+                COORDINATOR_CLIENT,
                 grants(|g| {
                     g.call = vec![Pattern::prefix("agent."), Pattern::prefix("env.")];
                     g.publish = vec![Pattern::prefix("session.")];
                     g.manage_topics = vec![Pattern::prefix("session.")];
                 }),
             )
+            .client(
+                SUPERVISOR_CLIENT,
+                grants(|g| {
+                    g.call = vec![Pattern::prefix("agent."), Pattern::prefix("env.")];
+                }),
+            )
             .client(ENV_CLIENT, grants(|g| g.register = vec![Pattern::exact(ENV_SERVICE)]))
+            .client(
+                &format!("{ENV_CLIENT}-r2"),
+                grants(|g| g.register = vec![Pattern::exact(ENV_SERVICE)]),
+            )
             .client("observer", grants(|g| g.subscribe = vec![Pattern::prefix("session.")]));
         for spec in &config.agents {
             let service = agent_service(&spec.agent_id);
@@ -196,66 +219,68 @@ impl SessionHarness {
         let router = Router::new(router_config).map_err(|e| {
             flybus::BusError::new(flybus::ErrorCode::StoreFailure, format!("router: {e}"))
         })?;
-        let connector = Connector {
+
-            router,
+        let budget = config.budget().map_err(refusal)?;
-            via,
+        let mut launcher =
-            store_root,
+            Launcher::start(router, config.mode, via, &store_root, &sockets, budget).await?;
-            sockets,
-            next_socket: AtomicU64::new(0),
-            listeners: Mutex::new(Vec::new()),
-        };
 
         let step_duration = hz(config.step_hz).expect("a positive cadence");
         let tick_duration = millis(config.tick_ms).expect("a positive tick");
 
         // The environment first: it owns the world and the descriptor.
-        let env_client = connector.client(ENV_CLIENT).await?;
+        let environment = launcher
-        let env_service = env_client.register(ENV_SERVICE, ServiceConfig::default()).await?;
+            .launch_environment(EnvironmentLaunch {
-        let env_incarnation = env_service.incarnation().to_owned();
-        let environment = serve(
-            env_client,
-            env_service,
-            CounterEnvironment::new(EnvironmentConfig {
                 session_id: config.session_id.clone(),
                 worker_id: id(ENV_WORKER),
                 incarnation_id: id("arena-inc-1"),
                 step_duration,
                 ports: config.agents.iter().map(|a| a.port_id.clone()).collect(),
+                worker_threads: config.environment_threads,
                 faults: config.environment_faults.clone(),
-            }),
+                client_id: ENV_CLIENT.to_owned(),
-        );
+                service: ENV_SERVICE.to_owned(),
+            })
+            .await
+            .map_err(refusal)?;
+        let environment_ref = launcher
+            .worker(&environment.worker_id)
+            .expect("just launched")
+            .worker_ref();
 
         let mut slots = Vec::new();
-        let mut agents = BTreeMap::new();
         for spec in &config.agents {
-            let service_name = agent_service(&spec.agent_id);
+            let identity = launcher
-            let client = connector.client(&agent_client(&spec.agent_id)).await?;
+                .launch_agent(AgentLaunch {
-            let service = client.register(&service_name, ServiceConfig::default()).await?;
-            let incarnation = service.incarnation().to_owned();
-            let handle = serve(
-                client,
-                service,
-                FakeAgentWorker::new(AgentConfig {
                     session_id: config.session_id.clone(),
                     agent_id: spec.agent_id.clone(),
+                    port_id: spec.port_id.clone(),
                     incarnation_id: parse_id(&format!("{}-inc-1", spec.agent_id))
                         .expect("an agent id plus a suffix is an Id"),
                     tick_duration,
                     warmup_ticks: config.warmup_ticks,
+                    worker_threads: spec.worker_threads,
                     faults: spec.faults.clone(),
-                }),
+                    client_id: agent_client(&spec.agent_id),
-            );
+                    service: agent_service(&spec.agent_id),
-            slots.push(AgentSlot::new(
+                })
-                WorkerRef::new(&service_name, &incarnation, &spec.agent_id),
+                .await
+                .map_err(refusal)?;
+            let worker_ref = launcher
+                .worker(&spec.agent_id)
+                .expect("just launched")
+                .worker_ref();
+            let mut slot = AgentSlot::new(
+                worker_ref,
                 spec.agent_id.clone(),
                 spec.port_id.clone(),
                 synthetic_profile(&spec.agent_id, &tick_duration, config.warmup_ticks),
                 spec.seed,
-            ));
+            );
-            agents.insert(spec.agent_id.clone(), handle);
+            slot.worker_threads = identity.worker_threads as u64;
+            slots.push(slot);
         }
 
-        let coordinator_client = connector.client("coordinator").await?;
+        let coordinator_client = launcher.connect(COORDINATOR_CLIENT).await?;
         let executors: BTreeMap<Id, Box<dyn ActionExecutor>> = config
             .agents
             .iter()
@@ -268,7 +293,7 @@ impl SessionHarness {
             config.session_id.clone(),
             config.epoch.clone(),
             config.episode_id.clone(),
-            WorkerRef::new(ENV_SERVICE, &env_incarnation, &id(ENV_WORKER)),
+            environment_ref,
             slots,
             Box::new(CounterTask::new(&config.epoch, config.terminal)),
             executors,
@@ -276,27 +301,42 @@ impl SessionHarness {
 
         Ok(SessionHarness {
             coordinator,
-            environment,
-            agents,
             config,
             via,
-            connector,
+            mode: launcher.mode(),
+            launcher,
             observers: Mutex::new(Vec::new()),
         })
     }
 
     pub fn router(&self) -> &Router {
-        &self.connector.router
+        self.launcher.router()
+    }
+
+    /// The coordinator and its supervisor, borrowed apart.
+    ///
+    /// A supervisor acts while a transition is in flight -- that is what a supervisor is for
+    /// -- so the two have to be reachable at the same time.
+    pub fn parts(&mut self) -> (&mut Coordinator, &mut Launcher) {
+        (&mut self.coordinator, &mut self.launcher)
+    }
+
+    /// The router's own counters: owners, roots, queued messages and store bytes.
+    pub fn router_stats(&self) -> flybus::RouterStats {
+        self.launcher.router().stats()
     }
 
     /// A client for `id`, connected the same way every participant is.
+    ///
+    /// An unconfigured client id is refused by the launcher's policy before it can route, so
+    /// this is not a way around the composition.
     pub async fn client(&self, id: &str) -> Result<Client, flybus::BusError> {
-        self.connector.client(id).await
+        self.launcher.connect(id).await
     }
 
     /// An extra subscriber, for a test that watches the published boundaries.
     pub async fn observer(&self) -> Result<Client, flybus::BusError> {
-        let client = self.connector.client("observer").await?;
+        let client = self.launcher.connect("observer").await?;
         self.observers.lock().expect("not poisoned").push(client.clone());
         Ok(client)
     }
@@ -306,22 +346,6 @@ impl SessionHarness {
     /// The coordinator still pins the old registration, so its next call to that agent fails
     /// rather than silently reaching another brain.
     pub async fn restart_agent(&mut self, agent_id: &Id) -> Result<Restarted, flybus::BusError> {
-        let tick_duration = millis(self.config.tick_ms).expect("a positive tick");
-        if let Some(old) = self.agents.remove(agent_id) {
-            old.stop().await;
-        }
-        let service_name = agent_service(agent_id);
-        let client = self.connector.client(&format!("{}-r2", agent_client(agent_id))).await?;
-        let service = loop {
-            match client.register(&service_name, ServiceConfig::default()).await {
-                Ok(service) => break service,
-                Err(e) if e.code == flybus::ErrorCode::Conflict => {
-                    // The old registration is released when its connection finishes closing.
-                    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
-                }
-                Err(e) => return Err(e),
-            }
-        };
         let spec = self
             .config
             .agents
@@ -329,53 +353,87 @@ impl SessionHarness {
             .find(|spec| spec.agent_id == *agent_id)
             .expect("a configured agent")
             .clone();
+        self.launcher.kill(agent_id).await;
+        let tick_duration = millis(self.config.tick_ms).expect("a positive tick");
         let incarnation_id =
             parse_id(&format!("{agent_id}-inc-2")).expect("an agent id plus a suffix is an Id");
-        let restarted = Restarted {
+        self.launcher
-            service: service_name,
+            .launch_agent(AgentLaunch {
-            service_incarnation: service.incarnation().to_owned(),
-            incarnation_id: incarnation_id.clone(),
-        };
-        let handle = serve(
-            client,
-            service,
-            FakeAgentWorker::new(AgentConfig {
                 session_id: self.config.session_id.clone(),
                 agent_id: agent_id.clone(),
-                incarnation_id,
+                port_id: spec.port_id.clone(),
+                incarnation_id: incarnation_id.clone(),
                 tick_duration,
                 warmup_ticks: self.config.warmup_ticks,
-                faults: spec.faults,
+                worker_threads: spec.worker_threads,
-            }),
+                faults: spec.faults.clone(),
-        );
+                client_id: format!("{}-r2", agent_client(agent_id)),
-        self.agents.insert(agent_id.clone(), handle);
+                service: agent_service(agent_id),
-        Ok(restarted)
+            })
+            .await
+            .map_err(refusal)?;
+        let worker = self.launcher.worker(agent_id).expect("just launched");
+        Ok(Restarted {
+            service: worker.identity.service.clone(),
+            service_incarnation: worker.service_incarnation.clone(),
+            incarnation_id,
+        })
+    }
+
+    /// Ends one participant without asking it, as a crash would.
+    pub async fn kill(&mut self, worker_id: &Id) -> ReapOutcome {
+        self.launcher.kill(worker_id).await
+    }
+
+    /// The worker id the environment answers to.
+    pub fn environment_id(&self) -> Id {
+        id(ENV_WORKER)
     }
 
     /// The agent worker's progress counter, which is its fake model's mutation count.
+    ///
+    /// A participant in another process keeps its counter there; use
+    /// [`SessionHarness::progress_of`], which reads it over the bus in every mode.
     pub fn agent_mutations(&self, agent_id: &Id) -> u64 {
-        self.agents.get(agent_id).map(WorkerHandle::progress_counter).unwrap_or_default()
+        self.launcher
+            .worker(agent_id)
+            .and_then(crate::launcher::LaunchedWorker::progress_counter)
+            .unwrap_or_default()
     }
 
     pub fn environment_mutations(&self) -> u64 {
-        self.environment.progress_counter()
+        self.agent_mutations(&id(ENV_WORKER))
     }
 
+    /// One participant's progress counter, read over the bus. Works in every execution mode.
+    pub async fn progress_of(&mut self, worker_id: &Id) -> Result<u64, DomainError> {
+        Ok(self.launcher.health_check(worker_id).await?.progress_counter)
+    }
+
+    /// The local status cell of a participant in this process, or `None` for one with a
+    /// process of its own.
     pub fn agent_status(&self, agent_id: &Id) -> Option<StatusCell> {
-        self.agents.get(agent_id).map(|handle| handle.status.clone())
+        self.launcher.worker(agent_id).and_then(crate::launcher::LaunchedWorker::status)
     }
 
-    /// Stops every worker and closes the router.
+    /// Reaps every participant and closes the router.
     pub async fn shutdown(self) {
-        let SessionHarness { coordinator, environment, agents, connector, observers, .. } = self;
+        let SessionHarness { coordinator, mut launcher, observers, .. } = self;
         drop(coordinator);
-        environment.stop().await;
+        launcher.reap_all("shutdown").await;
-        for (_, handle) in agents {
-            handle.stop().await;
-        }
         for observer in observers.into_inner().expect("not poisoned") {
             observer.close().await;
         }
-        connector.router.shutdown();
+        launcher.router().shutdown();
     }
 }
+
+/// A launcher refusal, as a bus error: the harness's one error type stays the bus's.
+fn refusal(e: DomainError) -> flybus::BusError {
+    let code = match e.code {
+        ErrorCode::Busy => flybus::ErrorCode::QuotaExceeded,
+        ErrorCode::IdentityMismatch => flybus::ErrorCode::NotAuthorized,
+        _ => flybus::ErrorCode::RouterLost,
+    };
+    flybus::BusError::new(code, e.to_string())
+}

services/flysim/crates/fly-session/src/lib.rs

@@ -22,11 +22,15 @@
 //! [`step-v1`]: https://example.invalid/step-v1
 
 pub mod agent;
+pub mod cli;
 pub mod clock;
 pub mod coordinator;
 pub mod dedup;
 pub mod environment;
 pub mod harness;
+pub mod launcher;
+pub mod measure;
+pub mod metrics;
 pub mod phase;
 pub mod rpc;
 pub mod task;
@@ -37,5 +41,8 @@ pub mod worker;
 pub mod types;
 pub use fly_session_types;
 
-pub use coordinator::{Coordinator, DispatchOrder, Injections, SessionFailure, StepReport};
+pub use coordinator::{
+    Coordinator, Deadlines, DispatchOrder, Injections, SessionFailure, StepReport,
+};
+pub use launcher::{ExecutionMode, Launcher, ReapOutcome, ThreadBudget, Via};
 pub use phase::{Phase, PhaseMachine};

services/flysim/crates/fly-session/src/measure.rs

@@ -0,0 +1,331 @@
+//! The execution-mode comparison the implementation guide's section 5 asks for.
+//!
+//! It runs the same synthetic session in each execution mode, at one, two and four agents,
+//! and reports the thread allocation, the RPC and critical-path percentiles, the memory peaks
+//! and the router's owner, collection and queue counters.
+//!
+//! **These are local synthetic timings on one machine, and no host capacity claim follows
+//! from any of them.** They exist so the three modes can be compared with each other: the
+//! question this slice has to answer is what a process boundary costs, not how fast anything
+//! is. Pacing is switched off for the run, so a transition follows the one before it as fast
+//! as the participants answer and the samples are work rather than sleep.
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+
+use crate::coordinator::DispatchOrder;
+use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via};
+use crate::launcher::ExecutionMode;
+use crate::metrics::{Percentiles, physical_cores};
+
+/// What to compare.
+#[derive(Clone, Debug)]
+pub struct MeasureConfig {
+    /// Transitions per run, after the warm-up ones.
+    pub steps: u64,
+    /// Transitions run before sampling starts, so first-call costs are not in the samples.
+    pub warmup_steps: u64,
+    pub agent_counts: Vec<usize>,
+    pub modes: Vec<ExecutionMode>,
+    /// The within-agent worker count each agent asks its launcher for.
+    pub worker_threads: usize,
+}
+
+impl Default for MeasureConfig {
+    fn default() -> MeasureConfig {
+        MeasureConfig {
+            steps: 200,
+            warmup_steps: 10,
+            agent_counts: vec![1, 2, 4],
+            modes: ExecutionMode::all().to_vec(),
+            worker_threads: 1,
+        }
+    }
+}
+
+/// The highest value each router counter reached while the transitions ran.
+#[derive(Debug, Default)]
+struct Peaks {
+    owners: AtomicU64,
+    roots: AtomicU64,
+    queued: AtomicU64,
+    sealed: AtomicU64,
+    store_bytes: AtomicU64,
+}
+
+impl Peaks {
+    fn observe(&self, stats: &flybus::RouterStats) {
+        raise(&self.owners, stats.owners as u64);
+        raise(&self.roots, stats.artifact_roots);
+        raise(&self.queued, stats.queued as u64);
+        raise(&self.sealed, stats.sealed_artifacts as u64);
+        raise(&self.store_bytes, stats.store_bytes);
+    }
+
+    fn read(&self) -> (usize, u64, usize, usize, u64) {
+        (
+            self.owners.load(Ordering::Relaxed) as usize,
+            self.roots.load(Ordering::Relaxed),
+            self.queued.load(Ordering::Relaxed) as usize,
+            self.sealed.load(Ordering::Relaxed) as usize,
+            self.store_bytes.load(Ordering::Relaxed),
+        )
+    }
+}
+
+fn raise(slot: &AtomicU64, value: u64) {
+    slot.fetch_max(value, Ordering::Relaxed);
+}
+
+/// One measured composition.
+#[derive(Clone, Debug)]
+pub struct Row {
+    pub mode: ExecutionMode,
+    pub agents: usize,
+    pub worker_threads: usize,
+    pub physical_cores: usize,
+    pub budget_total: usize,
+    pub budget_used: usize,
+    pub steps: u64,
+    /// `Agent.Prepare`, over every agent and every transition.
+    pub prepare: Percentiles,
+    pub commit: Percentiles,
+    pub advance: Percentiles,
+    /// `Worker.Status`: an RPC with no domain work behind it, so it is the router and
+    /// transport floor rather than a measure of the worker.
+    pub status: Percentiles,
+    /// One whole transition, pacing excluded: the critical path.
+    pub step: Percentiles,
+    pub coordinator_peak_rss_kib: u64,
+    /// The sum of the peak resident sets of the participants with processes of their own.
+    pub participants_peak_rss_kib: u64,
+    pub owners_max: usize,
+    pub owners_final: usize,
+    pub artifact_roots_max: u64,
+    pub queued_max: usize,
+    pub sealed_max: usize,
+    pub sealed_final: usize,
+    pub store_bytes_max: u64,
+    pub store_bytes_final: u64,
+    /// One sealed frame per boundary, boundary zero included.
+    pub frames_produced: u64,
+}
+
+impl Row {
+    /// Frames the store collected: produced, minus the ones still owned at the end.
+    pub fn collected(&self) -> u64 {
+        self.frames_produced.saturating_sub(self.sealed_final as u64)
+    }
+}
+
+/// Runs the comparison. Every row is one composition in one mode.
+pub async fn run(config: &MeasureConfig) -> Result<Vec<Row>, String> {
+    let mut rows = Vec::new();
+    for mode in &config.modes {
+        for agents in &config.agent_counts {
+            rows.push(one(config, *mode, *agents).await?);
+        }
+    }
+    Ok(rows)
+}
+
+async fn one(config: &MeasureConfig, mode: ExecutionMode, agents: usize) -> Result<Row, String> {
+    let dir = std::env::temp_dir().join(format!(
+        "fly-session-measure-{}-{agents}-{}",
+        mode.label(),
+        std::process::id()
+    ));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
+    let result = measure_in(config, mode, agents, &dir).await;
+    let _ = std::fs::remove_dir_all(&dir);
+    result
+}
+
+async fn measure_in(
+    config: &MeasureConfig,
+    mode: ExecutionMode,
+    agents: usize,
+    dir: &std::path::Path,
+) -> Result<Row, String> {
+    let specs: Vec<AgentSpec> = (0..agents)
+        .map(|i| AgentSpec {
+            worker_threads: config.worker_threads,
+            ..AgentSpec::new(&format!("fly-{}", (b'a' + i as u8) as char), &format!("p{}", i + 1), 7 + i as i32)
+        })
+        .collect();
+    let harness_config = HarnessConfig {
+        agents: specs,
+        mode,
+        ..HarnessConfig::default()
+    };
+    let budget = harness_config.budget().map_err(|e| e.to_string())?;
+    let (budget_total, budget_used_floor) = (budget.total(), harness_config.required_threads());
+    let mut harness = SessionHarness::start(Via::Unix, dir, harness_config)
+        .await
+        .map_err(|e| format!("{}: start: {}", mode.label(), e.message))?;
+    harness.coordinator.dispatch = DispatchOrder::Concurrent;
+    harness.coordinator.disable_pacing();
+    harness
+        .coordinator
+        .bootstrap()
+        .await
+        .map_err(|e| format!("{}: bootstrap: {e}", mode.label()))?;
+    // The warm-up transitions pay the first-call costs; their samples are then discarded.
+    harness
+        .coordinator
+        .run(config.warmup_steps)
+        .await
+        .map_err(|e| format!("{}: warm-up: {e}", mode.label()))?;
+    harness.coordinator.metrics.clear();
+
+    // The router's counters are sampled *while* transitions run, not between them: a queue
+    // that is empty at every committed boundary says nothing about whether it stayed bounded
+    // during the transaction, which is the thing section 4 asks about.
+    let peaks = Arc::new(Peaks::default());
+    let sampling = Arc::new(AtomicBool::new(true));
+    let sampler = tokio::spawn({
+        let router = harness.router().clone();
+        let peaks = peaks.clone();
+        let sampling = sampling.clone();
+        async move {
+            while sampling.load(Ordering::Relaxed) {
+                peaks.observe(&router.stats());
+                tokio::time::sleep(std::time::Duration::from_micros(200)).await;
+            }
+            peaks.observe(&router.stats());
+        }
+    });
+
+    let environment = harness.environment_id();
+    let mut status = crate::metrics::Metrics::default();
+    for _ in 0..config.steps {
+        harness
+            .coordinator
+            .step()
+            .await
+            .map_err(|e| format!("{}: step: {e}", mode.label()))?;
+        // One status call per transition: an RPC the worker answers from a cell rather than
+        // from its endpoint, so it is the router-and-transport floor the domain calls sit on.
+        let started = std::time::Instant::now();
+        let _ = harness.launcher.health_check(&environment).await;
+        status.record("Worker.Status", started.elapsed());
+    }
+    sampling.store(false, Ordering::Relaxed);
+    let _ = sampler.await;
+    let (owners_max, roots_max, queued_max, sealed_max, store_bytes_max) = peaks.read();
+
+    let metrics = &harness.coordinator.metrics;
+    let zero = Percentiles::default();
+    let row = Row {
+        mode,
+        agents,
+        worker_threads: config.worker_threads,
+        physical_cores: physical_cores(),
+        budget_total,
+        budget_used: budget_used_floor,
+        steps: config.steps,
+        prepare: metrics.percentiles("Agent.Prepare").unwrap_or(zero),
+        commit: metrics.percentiles("Agent.Commit").unwrap_or(zero),
+        advance: metrics.percentiles("Environment.Advance").unwrap_or(zero),
+        status: status.percentiles("Worker.Status").unwrap_or(zero),
+        step: metrics.percentiles("step").unwrap_or(zero),
+        coordinator_peak_rss_kib: crate::metrics::peak_rss_kib().unwrap_or_default(),
+        participants_peak_rss_kib: harness
+            .launcher
+            .peak_rss_kib()
+            .iter()
+            .filter(|(who, _)| who.as_str() != "coordinator")
+            .map(|(_, kib)| *kib)
+            .sum(),
+        owners_max,
+        owners_final: harness.router_stats().owners,
+        artifact_roots_max: roots_max,
+        queued_max,
+        sealed_max,
+        sealed_final: harness.router_stats().sealed_artifacts,
+        store_bytes_max,
+        store_bytes_final: harness.router_stats().store_bytes,
+        frames_produced: config.steps + config.warmup_steps + 1,
+    };
+    harness.shutdown().await;
+    Ok(row)
+}
+
+/// The measurement table, as Markdown.
+pub fn table(rows: &[Row]) -> String {
+    let mut out = String::new();
+    out.push_str(
+        "Local synthetic timings on one machine. Not a capacity claim, and not a latency goal.\n\n",
+    );
+    if let Some(first) = rows.first() {
+        out.push_str(&format!(
+            "Physical cores: {}. Coordinator reservation: 1 thread.\n\n",
+            first.physical_cores
+        ));
+    }
+    out.push_str(
+        "| mode | agents | threads/agent | budget used/total | Prepare p50/p95/p99 us | \
+Commit p50/p95/p99 us | Advance p50/p95/p99 us | Status p50/p99 us | step p50/p95/p99 us |\n",
+    );
+    out.push_str("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n");
+    for r in rows {
+        out.push_str(&format!(
+            "| {} | {} | {} | {}/{} | {:.0}/{:.0}/{:.0} | {:.0}/{:.0}/{:.0} | \
+{:.0}/{:.0}/{:.0} | {:.0}/{:.0} | {:.0}/{:.0}/{:.0} |\n",
+            r.mode.label(),
+            r.agents,
+            r.worker_threads,
+            r.budget_used,
+            r.budget_total,
+            r.prepare.p50_us(),
+            r.prepare.p95_us(),
+            r.prepare.p99_us(),
+            r.commit.p50_us(),
+            r.commit.p95_us(),
+            r.commit.p99_us(),
+            r.advance.p50_us(),
+            r.advance.p95_us(),
+            r.advance.p99_us(),
+            r.status.p50_us(),
+            r.status.p99_us(),
+            r.step.p50_us(),
+            r.step.p95_us(),
+            r.step.p99_us(),
+        ));
+    }
+    out.push('\n');
+    out.push_str(
+        "| mode | agents | coordinator peak RSS KiB | participant peak RSS KiB | owners max/final | \
+roots max | queued max | sealed max/final | store bytes max/final | frames produced/collected |\n",
+    );
+    out.push_str("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n");
+    for r in rows {
+        out.push_str(&format!(
+            "| {} | {} | {} | {} | {}/{} | {} | {} | {}/{} | {}/{} | {}/{} |\n",
+            r.mode.label(),
+            r.agents,
+            r.coordinator_peak_rss_kib,
+            r.participants_peak_rss_kib,
+            r.owners_max,
+            r.owners_final,
+            r.artifact_roots_max,
+            r.queued_max,
+            r.sealed_max,
+            r.sealed_final,
+            r.store_bytes_max,
+            r.store_bytes_final,
+            r.frames_produced,
+            r.collected(),
+        ));
+    }
+    out
+}
+
+/// The rows keyed by mode and agent count, for a caller that wants one of them.
+pub fn by_composition(rows: &[Row]) -> BTreeMap<(String, usize), Row> {
+    rows.iter()
+        .map(|r| ((r.mode.label().to_owned(), r.agents), r.clone()))
+        .collect()
+}

services/flysim/crates/fly-session/src/metrics.rs

@@ -0,0 +1,176 @@
+//! Latency and resource samples, for the measurements the implementation guide's section 5
+//! asks every slice to report.
+//!
+//! These are local synthetic timings on one machine. No host capacity claim follows from any
+//! number this module produces, and nothing here is a gameplay latency goal: the percentiles
+//! exist so that the three execution modes can be compared against each other.
+
+use std::collections::BTreeMap;
+
+/// One sample set's order statistics, by nearest rank.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct Percentiles {
+    pub count: usize,
+    pub p50_ns: u64,
+    pub p95_ns: u64,
+    pub p99_ns: u64,
+    pub max_ns: u64,
+}
+
+impl Percentiles {
+    fn of(sorted: &[u64]) -> Percentiles {
+        let rank = |p: f64| -> u64 {
+            if sorted.is_empty() {
+                return 0;
+            }
+            let n = sorted.len() as f64;
+            let index = (p * n).ceil() as usize;
+            sorted[index.clamp(1, sorted.len()) - 1]
+        };
+        Percentiles {
+            count: sorted.len(),
+            p50_ns: rank(0.50),
+            p95_ns: rank(0.95),
+            p99_ns: rank(0.99),
+            max_ns: sorted.last().copied().unwrap_or_default(),
+        }
+    }
+
+    pub fn p50_us(&self) -> f64 {
+        self.p50_ns as f64 / 1000.0
+    }
+
+    pub fn p95_us(&self) -> f64 {
+        self.p95_ns as f64 / 1000.0
+    }
+
+    pub fn p99_us(&self) -> f64 {
+        self.p99_ns as f64 / 1000.0
+    }
+}
+
+/// Named duration samples. One name is one measured path: a domain method, or the
+/// coordinator's whole critical path for a transition.
+#[derive(Clone, Debug, Default)]
+pub struct Metrics {
+    samples: BTreeMap<String, Vec<u64>>,
+}
+
+impl Metrics {
+    pub fn record(&mut self, what: &str, elapsed: std::time::Duration) {
+        let ns = u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX);
+        self.samples.entry(what.to_owned()).or_default().push(ns);
+    }
+
+    pub fn percentiles(&self, what: &str) -> Option<Percentiles> {
+        let mut values = self.samples.get(what)?.clone();
+        values.sort_unstable();
+        Some(Percentiles::of(&values))
+    }
+
+    pub fn names(&self) -> Vec<String> {
+        self.samples.keys().cloned().collect()
+    }
+
+    pub fn count(&self, what: &str) -> usize {
+        self.samples.get(what).map(Vec::len).unwrap_or_default()
+    }
+
+    pub fn clear(&mut self) {
+        self.samples.clear();
+    }
+}
+
+/// This process's peak resident set, in KiB, from its own status file.
+pub fn peak_rss_kib() -> Option<u64> {
+    peak_rss_of("/proc/self/status")
+}
+
+/// One child process's peak resident set, in KiB. `None` once the child is gone.
+pub fn peak_rss_kib_of(pid: u32) -> Option<u64> {
+    peak_rss_of(&format!("/proc/{pid}/status"))
+}
+
+fn peak_rss_of(path: &str) -> Option<u64> {
+    let text = std::fs::read_to_string(path).ok()?;
+    for line in text.lines() {
+        if let Some(rest) = line.strip_prefix("VmHWM:") {
+            return rest.split_whitespace().next()?.parse().ok();
+        }
+    }
+    None
+}
+
+/// How many physical cores this machine has, counted as distinct (package, core) pairs.
+///
+/// Falls back to the logical count, which is what a thread budget has to use when the
+/// topology cannot be read.
+pub fn physical_cores() -> usize {
+    if let Ok(text) = std::fs::read_to_string("/proc/cpuinfo") {
+        let mut pairs: std::collections::BTreeSet<(String, String)> =
+            std::collections::BTreeSet::new();
+        let (mut package, mut core) = (None, None);
+        for line in text.lines() {
+            if line.trim().is_empty() {
+                if let (Some(p), Some(c)) = (package.take(), core.take()) {
+                    pairs.insert((p, c));
+                }
+                continue;
+            }
+            if let Some((key, value)) = line.split_once(':') {
+                match key.trim() {
+                    "physical id" => package = Some(value.trim().to_owned()),
+                    "core id" => core = Some(value.trim().to_owned()),
+                    _ => {}
+                }
+            }
+        }
+        if let (Some(p), Some(c)) = (package, core) {
+            pairs.insert((p, c));
+        }
+        if !pairs.is_empty() {
+            return pairs.len();
+        }
+    }
+    logical_cores()
+}
+
+/// How many hardware threads this machine reports.
+pub fn logical_cores() -> usize {
+    std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn percentiles_use_nearest_rank() {
+        let mut m = Metrics::default();
+        for ns in 1..=100u64 {
+            m.record("x", std::time::Duration::from_nanos(ns));
+        }
+        let p = m.percentiles("x").expect("recorded");
+        assert_eq!(p.count, 100);
+        assert_eq!(p.p50_ns, 50);
+        assert_eq!(p.p95_ns, 95);
+        assert_eq!(p.p99_ns, 99);
+        assert_eq!(p.max_ns, 100);
+        assert!(m.percentiles("y").is_none());
+    }
+
+    #[test]
+    fn a_single_sample_is_every_percentile() {
+        let mut m = Metrics::default();
+        m.record("x", std::time::Duration::from_nanos(7));
+        let p = m.percentiles("x").expect("recorded");
+        assert_eq!((p.p50_ns, p.p95_ns, p.p99_ns, p.max_ns), (7, 7, 7, 7));
+    }
+
+    #[test]
+    fn the_machine_reports_at_least_one_core() {
+        assert!(physical_cores() >= 1);
+        assert!(logical_cores() >= 1);
+        assert!(peak_rss_kib().unwrap_or(1) > 0);
+    }
+}

services/flysim/crates/fly-session/src/worker.rs

@@ -224,6 +224,22 @@ impl WorkerHandle {
         let _ = self.task.await;
         self.client.close().await;
     }
+
+    /// Stops serving without waiting. The connection closes when the last handle to it is
+    /// dropped, which this does. For a supervisor's `Drop`, where there is no runtime to wait
+    /// on.
+    pub fn abort(self) {
+        self.task.abort();
+    }
+
+    /// Waits until the worker stops serving, which `Worker.Shutdown` makes it do.
+    ///
+    /// A worker process awaits this and then exits, so the supervisor's `Worker.Shutdown` and
+    /// the process's exit are the same event rather than two racing ones.
+    pub async fn join(self) {
+        let _ = self.task.await;
+        self.client.close().await;
+    }
 }
 
 /// Registers `service_name` and serves `endpoint` on it until the service ends or Shutdown.

services/flysim/crates/fly-session/tests/common/mod.rs

@@ -4,7 +4,7 @@
 
 use std::time::Duration;
 
-use fly_session::harness::{HarnessConfig, SessionHarness, Via};
+use fly_session::harness::{ExecutionMode, HarnessConfig, SessionHarness, Via};
 use fly_session::types::*;
 
 pub const WAIT: Duration = Duration::from_secs(20);
@@ -94,3 +94,55 @@ pub async fn within<T>(what: &str, f: impl std::future::Future<Output = T>) -> T
         Err(_) => panic!("{what}: timed out"),
     }
 }
+
+/// Generates one test per execution mode from an `async fn name(mode: ExecutionMode)`.
+///
+/// The separate-process mode is the SESSION-02 subject; the other two are the variants it is
+/// compared against, and a row that holds in one must hold in all three.
+#[macro_export]
+macro_rules! all_modes {
+    ($($name:ident),* $(,)?) => {
+        mod in_process {
+            $(
+                #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+                async fn $name() {
+                    super::$name($crate::common::mode_in_process()).await
+                }
+            )*
+        }
+        mod dedicated_thread {
+            $(
+                #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+                async fn $name() {
+                    super::$name($crate::common::mode_thread()).await
+                }
+            )*
+        }
+        mod separate_process {
+            $(
+                #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+                async fn $name() {
+                    super::$name($crate::common::mode_process()).await
+                }
+            )*
+        }
+    };
+}
+
+pub fn mode_in_process() -> ExecutionMode {
+    ExecutionMode::InProcess
+}
+
+pub fn mode_thread() -> ExecutionMode {
+    ExecutionMode::Thread
+}
+
+pub fn mode_process() -> ExecutionMode {
+    ExecutionMode::Process
+}
+
+/// A fixture in one execution mode. The transport is the mode's own: a separate process
+/// reaches the router only over a socket.
+pub async fn mode_fixture(mode: ExecutionMode, config: HarnessConfig) -> Fixture {
+    fixture(Via::Unix, HarnessConfig { mode, ..config }).await
+}

services/flysim/crates/fly-session/tests/processes.rs

@@ -0,0 +1,600 @@
+//! SESSION-02 acceptance: one agent process per fly and one environment process under the
+//! coordinator, compared with the in-process and dedicated-thread variants.
+//!
+//! Every acceptance bullet is one named test here, generated once per execution mode, so a
+//! rule that holds in one process holds across a process boundary too. The two process-mode
+//! failure rows of section 4 that SESSION-01 could not reach in one process -- a router
+//! restart during a world advance, and an old worker's reply after a restart -- are at the
+//! end and run in the separate-process mode.
+
+mod common;
+
+use std::collections::BTreeMap;
+use std::time::{Duration, Instant};
+
+use common::{at, count, fly_a, fly_b, mode_fixture, within};
+use fly_session::agent::AgentFaults;
+use fly_session::coordinator::{DispatchOrder, Injections};
+use fly_session::environment::EnvironmentFaults;
+use fly_session::harness::{ExecutionMode, HarnessConfig, Via};
+use fly_session::launcher::{ReapOutcome, ThreadBudget};
+use fly_session::phase::Phase;
+use fly_session::types::*;
+
+all_modes!(
+    a_delayed_one_agent_result_holds_the_world,
+    a_worker_death_has_a_bounded_diagnosed_outcome,
+    a_helper_death_has_a_bounded_diagnosed_outcome,
+    an_uncertain_advance_never_creates_a_second_batch,
+    a_partial_commit_never_permits_next_step_play,
+    every_participant_answers_its_supervisor,
+    worker_threads_lie_within_the_launcher_allocation,
+);
+
+const STEPS: u64 = 4;
+
+fn two_agents(mode: ExecutionMode) -> HarnessConfig {
+    HarnessConfig { mode, ..HarnessConfig::default() }
+}
+
+// -------------------------------------------------------------------------------------------
+// Acceptance: sequential, reversed and parallel completion produce equivalent traces
+
+/// `step-v1` section 8, across the process boundary: sequential, concurrent and reversed
+/// dispatch, in all three execution modes, produce one behaviour trace.
+///
+/// This reuses the wave-1 comparator -- the behaviour half of the section 8 trace, with
+/// request ids, bus correlation and wall time excluded -- so "a process behaves like a task"
+/// is the same assertion that "a reordered dispatch behaves like an ordered one" was.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn sequential_reversed_and_parallel_completion_agree() {
+    let mut behaviours: BTreeMap<String, Vec<String>> = BTreeMap::new();
+    for mode in ExecutionMode::all() {
+        for order in [
+            DispatchOrder::Sequential,
+            DispatchOrder::Concurrent,
+            DispatchOrder::Reversed,
+        ] {
+            let mut config = two_agents(mode);
+            // Deliberately unequal completion times, so a concurrent run really does finish
+            // out of dispatch order whichever side of a process boundary the agents are on.
+            config.agents[0].faults =
+                AgentFaults { prepare_delay_ms: 12, ..AgentFaults::default() };
+            config.agents[1].faults = AgentFaults { commit_delay_ms: 9, ..AgentFaults::default() };
+            let mut f = mode_fixture(mode, config).await;
+            f.harness.coordinator.dispatch = order;
+            within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+            within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
+            let behaviour = f.harness.coordinator.trace.behavior();
+            assert_eq!(behaviour.len() as u64, STEPS);
+            behaviours.insert(format!("{}/{order:?}", mode.label()), behaviour);
+            f.shutdown().await;
+        }
+    }
+    let mut iter = behaviours.iter();
+    let (first_name, first) = iter.next().expect("at least one run");
+    for (name, behaviour) in iter {
+        assert_eq!(
+            behaviour, first,
+            "{name} produced a different behaviour trace from {first_name}"
+        );
+    }
+}
+
+// -------------------------------------------------------------------------------------------
+// Acceptance: a delayed one-agent result holds the world
+
+/// One agent takes far longer than the other to prepare. No `Environment.Advance` is sent
+/// until every agent is Prepared, and the world is still at its old boundary while the
+/// coordinator waits.
+async fn a_delayed_one_agent_result_holds_the_world(mode: ExecutionMode) {
+    let mut config = two_agents(mode);
+    config.agents[1].faults = AgentFaults { prepare_delay_ms: 400, ..AgentFaults::default() };
+    let mut f = mode_fixture(mode, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let environment = f.harness.environment_id();
+    let before = within("progress", f.harness.progress_of(&environment)).await.unwrap();
+
+    let (coordinator, launcher) = f.harness.parts();
+    // The supervisor watches the world while the transition is in flight. That is what a
+    // supervisor is for, and `Worker.Status` answers without waiting for a mutation.
+    let (stepped, held) = tokio::join!(
+        async { within("step", coordinator.step()).await },
+        async {
+            tokio::time::sleep(Duration::from_millis(120)).await;
+            within("status", launcher.health_check(&environment)).await
+        }
+    );
+    let report = stepped.expect("the transition completes once the slow agent answers");
+    assert_eq!(report.boundary, 1);
+    let held = held.expect("the environment answers its supervisor during the wait");
+    assert_eq!(
+        held.progress_counter, before,
+        "the world may not advance while one agent is still preparing"
+    );
+    assert_eq!(
+        held.state,
+        WorkerState::Ready,
+        "the environment is at a committed boundary, not advancing"
+    );
+
+    // And the ordering the audit records says the same thing from the coordinator's side.
+    let audit = f.harness.coordinator.audit.clone();
+    let advance = at(&audit, "advance:0");
+    for agent in [fly_a(), fly_b()] {
+        assert!(
+            at(&audit, &format!("prepared:{agent}@0")) < advance,
+            "{agent} must be Prepared before the world advances: {audit:?}"
+        );
+    }
+    assert_eq!(f.harness.coordinator.stats().advances, 1);
+    f.shutdown().await;
+}
+
+// -------------------------------------------------------------------------------------------
+// Acceptance: worker or helper death has a bounded diagnosed outcome
+
+/// One agent dies in the middle of its Prepare. The epoch fails with a typed cause naming
+/// that agent, within the caller's own budget, and nothing continues on the remainder.
+async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
+    let mut config = two_agents(mode);
+    config.agents[1].faults = AgentFaults { prepare_delay_ms: 5_000, ..AgentFaults::default() };
+    let mut f = mode_fixture(mode, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let started = Instant::now();
+
+    let (coordinator, launcher) = f.harness.parts();
+    let (stepped, reaped) = tokio::join!(
+        async { within("step", coordinator.step()).await },
+        async {
+            tokio::time::sleep(Duration::from_millis(80)).await;
+            launcher.kill(&fly_b()).await
+        }
+    );
+    assert_eq!(reaped, ReapOutcome::Terminated);
+    let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one");
+    assert!(
+        started.elapsed() < Duration::from_secs(20),
+        "the outcome must be bounded, not a hang"
+    );
+    assert_eq!(
+        failure.participant.as_deref(),
+        Some(fly_b().as_str()),
+        "the failure names the participant: {failure}"
+    );
+    assert_ne!(
+        failure.error.mutation,
+        MutationCertainty::None,
+        "a participant that died mid-call leaves an uncertain mutation, never a clean none"
+    );
+    assert_eq!(f.harness.coordinator.phase(), Phase::Failed);
+    assert!(f.harness.coordinator.is_fenced());
+    // No partial continuation: no world step, no publication, and no next transition.
+    assert_eq!(f.harness.coordinator.stats().advances, 0);
+    assert_eq!(count(&f.harness.coordinator.audit, "publish:1"), 0);
+    let again = f.harness.coordinator.step().await.expect_err("a fenced epoch takes no step");
+    assert_eq!(again.error.code, ErrorCode::InvalidPhase);
+    f.shutdown().await;
+}
+
+/// The environment helper dies in the middle of the world advance. Same rule: a typed cause
+/// naming it, bounded, and no half-transition afterwards.
+async fn a_helper_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
+    let config = HarnessConfig {
+        environment_faults: EnvironmentFaults {
+            advance_delay_ms: 5_000,
+            ..EnvironmentFaults::default()
+        },
+        ..two_agents(mode)
+    };
+    let mut f = mode_fixture(mode, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let environment = f.harness.environment_id();
+    let started = Instant::now();
+
+    let (coordinator, launcher) = f.harness.parts();
+    let (stepped, reaped) = tokio::join!(
+        async { within("step", coordinator.step()).await },
+        async {
+            tokio::time::sleep(Duration::from_millis(200)).await;
+            launcher.kill(&environment).await
+        }
+    );
+    assert_eq!(reaped, ReapOutcome::Terminated);
+    let failure = stepped.expect_err("a dead world is a failed epoch");
+    assert!(started.elapsed() < Duration::from_secs(20), "bounded, not a hang");
+    assert_eq!(
+        failure.participant.as_deref(),
+        Some(environment.as_str()),
+        "the failure names the participant: {failure}"
+    );
+    assert_ne!(failure.error.mutation, MutationCertainty::None);
+    assert_eq!(f.harness.coordinator.phase(), Phase::Failed);
+    assert!(f.harness.coordinator.is_fenced());
+    assert_eq!(f.harness.coordinator.stats().advances, 0);
+    // The agents prepared and are not asked to prepare again or to commit anything.
+    assert_eq!(f.harness.coordinator.stats().commits, 0);
+    assert_eq!(count(&f.harness.coordinator.audit, "publish:1"), 0);
+    f.shutdown().await;
+}
+
+// -------------------------------------------------------------------------------------------
+// Acceptance: an uncertain Advance never creates a second batch
+
+/// The Advance result is lost after the world already stepped. The coordinator resolves the
+/// same operation against its original domain request id; the world advances once per
+/// transition and the batch is never re-sent as a new one.
+async fn an_uncertain_advance_never_creates_a_second_batch(mode: ExecutionMode) {
+    let clean = {
+        let mut f = mode_fixture(mode, two_agents(mode)).await;
+        within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+        within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
+        let environment = f.harness.environment_id();
+        let world = within("progress", f.harness.progress_of(&environment)).await.unwrap();
+        let out = (f.harness.coordinator.trace.behavior(), world);
+        f.shutdown().await;
+        out
+    };
+
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    f.harness.coordinator.injections = Injections {
+        at_step: 2,
+        lose_advance_result: true,
+        ..Injections::default()
+    };
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
+    let environment = f.harness.environment_id();
+    let world = within("progress", f.harness.progress_of(&environment)).await.unwrap();
+
+    assert_eq!(f.harness.coordinator.stats().advances, STEPS, "one advance per transition");
+    assert_eq!(
+        world, clean.1,
+        "the world moved exactly as often as it did without the loss"
+    );
+    assert_eq!(
+        f.harness.coordinator.trace.behavior(),
+        clean.0,
+        "an uncertain Advance changes no behaviour, so it created no second batch"
+    );
+    // Every transition has exactly one batch, and every batch id is its own.
+    let batches: Vec<Id> = f
+        .harness
+        .coordinator
+        .trace
+        .transitions
+        .iter()
+        .map(|t| t.behaviour.batch_id.clone())
+        .collect();
+    let unique: std::collections::BTreeSet<Id> = batches.iter().cloned().collect();
+    assert_eq!(unique.len(), batches.len(), "one batch id per transition: {batches:?}");
+    let injections = f.harness.coordinator.injection_log.clone();
+    assert!(
+        injections.iter().any(|o| o.what == "lost-advance-result" && o.identical),
+        "the loss must happen after dispatch, so the outcome really is uncertain: {injections:?}"
+    );
+    f.shutdown().await;
+}
+
+// -------------------------------------------------------------------------------------------
+// Acceptance: a partial Commit never permits next-step play
+
+/// One agent's Commit fails after the other's succeeded. The epoch fails naming that agent,
+/// the boundary does not move, nothing is published and there is no next transition.
+async fn a_partial_commit_never_permits_next_step_play(mode: ExecutionMode) {
+    let mut config = two_agents(mode);
+    config.agents[1].faults =
+        AgentFaults { fail_commit_at_step: Some(1), ..AgentFaults::default() };
+    let mut f = mode_fixture(mode, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    within("step", f.harness.coordinator.step()).await.unwrap();
+    let environment = f.harness.environment_id();
+
+    let failure = within("step", f.harness.coordinator.step())
+        .await
+        .expect_err("one failed Commit fails the epoch");
+    assert_eq!(
+        failure.participant.as_deref(),
+        Some(fly_b().as_str()),
+        "the failure names the agent whose Commit failed: {failure}"
+    );
+    assert_eq!(f.harness.coordinator.phase(), Phase::Failed);
+    assert!(f.harness.coordinator.is_fenced());
+
+    // The world moved once inside the failing transition -- the Advance is what the Commit
+    // follows -- and it moves no further. There is no next-step play on a partial commit.
+    let world_before = within("progress", f.harness.progress_of(&environment)).await.unwrap();
+    let again = f.harness.coordinator.step().await.expect_err("no play after a partial commit");
+    assert_eq!(again.error.code, ErrorCode::InvalidPhase);
+    let world_after = within("progress", f.harness.progress_of(&environment)).await.unwrap();
+    assert_eq!(world_after, world_before, "no next world step follows a partial commit");
+    let status = within("status", f.harness.launcher.health_check(&environment)).await.unwrap();
+    assert_eq!(
+        status.current_scope.unwrap().step,
+        2,
+        "the world stays at the boundary the failed transition reached"
+    );
+    let audit = f.harness.coordinator.audit.clone();
+    assert_eq!(count(&audit, "publish:2"), 0);
+    assert_eq!(f.harness.coordinator.committed_boundary(), None);
+    f.shutdown().await;
+}
+
+// -------------------------------------------------------------------------------------------
+// Supervision: identity, health and reaping
+
+/// Every participant answers the supervisor with the identity the launcher configured, and
+/// stops when it is asked to.
+async fn every_participant_answers_its_supervisor(mode: ExecutionMode) {
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    within("run", f.harness.coordinator.run(2)).await.unwrap();
+
+    let environment = f.harness.environment_id();
+    for who in [fly_a(), fly_b(), environment.clone()] {
+        let worker = f.harness.launcher.worker(&who).expect("a launched participant");
+        assert_eq!(worker.identity.worker_id, who);
+        assert_eq!(worker.domain_incarnation, worker.identity.incarnation_id);
+        assert!(!worker.service_incarnation.is_empty());
+        let status = within("health", f.harness.launcher.health_check(&who)).await.unwrap();
+        assert_eq!(status.state, WorkerState::Ready, "{who} is healthy at a boundary");
+    }
+    // The agents carry their configured port identities; the environment owns the ports.
+    assert_eq!(
+        f.harness.launcher.worker(&fly_a()).unwrap().identity.port_id.as_deref(),
+        Some("p1")
+    );
+    assert_eq!(
+        f.harness.launcher.worker(&fly_b()).unwrap().identity.port_id.as_deref(),
+        Some("p2")
+    );
+    assert!(f.harness.launcher.worker(&environment).unwrap().identity.port_id.is_none());
+
+    // A worker that is not the one the caller expects refuses to negotiate at all.
+    let worker = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap();
+    let wrong = serde_json::json!({
+        "sessionId": "demo",
+        "expectedWorkerId": "fly-z",
+        "role": "agent",
+        "supportedMajors": [1],
+    });
+    let err = within(
+        "hello",
+        f.harness.coordinator.probe_raw(&worker, "Worker.Hello", None, wrong),
+    )
+    .await
+    .expect_err("a worker is not whoever a caller says it is");
+    assert_eq!(err.code, ErrorCode::IdentityMismatch);
+
+    // Asking a participant to stop stops it, and the supervisor says which kind of stop it was.
+    let outcome = f.harness.launcher.reap(&fly_a(), "test").await;
+    assert_eq!(outcome, ReapOutcome::Stopped, "a live participant answers Worker.Shutdown");
+    assert_eq!(f.harness.launcher.reap(&fly_a(), "test").await, ReapOutcome::AlreadyGone);
+    f.shutdown().await;
+}
+
+/// `workers-v1`: `Agent.Initialize`'s `workerThreads` lies within the launcher allocation.
+///
+/// The budget refuses an allocation it cannot cover before anything is started, and an agent
+/// refuses an `Agent.Initialize` asking for more threads than its launcher gave it.
+async fn worker_threads_lie_within_the_launcher_allocation(mode: ExecutionMode) {
+    // The budget itself: a total, a coordinator reservation, and a refusal that names both.
+    let mut budget = ThreadBudget::new(4, 1).unwrap();
+    assert_eq!(budget.remaining(), 3);
+    assert_eq!(budget.allocate(&id("arena"), 1).unwrap(), 1);
+    assert_eq!(budget.allocate(&id("fly-a"), 2).unwrap(), 2);
+    let refused = budget.allocate(&id("fly-b"), 1).expect_err("the budget is spent");
+    assert_eq!(refused.code, ErrorCode::Busy);
+    budget.release(&id("fly-a"));
+    assert_eq!(budget.allocate(&id("fly-b"), 1).unwrap(), 1);
+    assert_eq!(budget.allocate(&id("fly-b"), 1).expect_err("already held").code, ErrorCode::Conflict);
+
+    // A composition the configured budget cannot cover never starts.
+    let config = HarnessConfig {
+        mode,
+        thread_budget: Some(2),
+        ..HarnessConfig::default()
+    };
+    let dir = tempfile::tempdir().expect("a temporary directory");
+    let refused = fly_session::harness::SessionHarness::start(Via::Unix, dir.path(), config).await;
+    let refused = refused.err().expect("two threads cannot hold a coordinator, a world and two flies");
+    assert_eq!(refused.code, flybus::ErrorCode::QuotaExceeded, "{}", refused.message);
+    drop(dir);
+
+    // And the worker's own check: it was launched with one thread, so an Initialize asking
+    // for eight is refused before the model is constructed.
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    let worker = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap();
+    let profile = fly_session::agent::synthetic_profile(
+        &fly_a(),
+        &millis(1).unwrap(),
+        f.harness.config.warmup_ticks,
+    );
+    let params = serde_json::json!({
+        "agentId": "fly-a",
+        "profile": profile.to_json(),
+        "seed": 7,
+        "initialInput": {"boundary": "0", "views": [], "structured": null},
+        "initialDecisionContext": {
+            "schema": fly_session::task::context_schema().to_json(),
+            "value": {},
+        },
+        "workerThreads": 8,
+    });
+    let err = within(
+        "initialize",
+        f.harness.coordinator.probe_raw(
+            &worker,
+            "Agent.Initialize",
+            Some(scope_at("demo", "e1", 0)),
+            params,
+        ),
+    )
+    .await
+    .expect_err("eight threads are not within a one-thread allocation");
+    assert_eq!(err.code, ErrorCode::Busy);
+    assert_eq!(err.mutation, MutationCertainty::None, "nothing was constructed");
+    // The allocation the coordinator actually sends is the one the launcher handed out.
+    assert_eq!(f.harness.launcher.worker(&fly_a()).unwrap().identity.worker_threads, 1);
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    f.shutdown().await;
+}
+
+// -------------------------------------------------------------------------------------------
+// Section 4 rows SESSION-01 could not reach in one process
+
+/// Row: "Router restarts during a world advance | Old handles/routes invalid; epoch fails and
+/// restores coherently."
+///
+/// The restore half is STATE-01's. What SESSION-02 establishes is the half before it: the
+/// epoch fails with a typed cause naming the participant the coordinator was talking to, the
+/// session is fenced, every artifact handle of that store incarnation is gone, and no
+/// boundary, publication or further transition follows.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_router_restart_during_a_world_advance_fences_the_epoch() {
+    let mode = ExecutionMode::Process;
+    let config = HarnessConfig {
+        environment_faults: EnvironmentFaults {
+            advance_delay_ms: 3_000,
+            ..EnvironmentFaults::default()
+        },
+        ..two_agents(mode)
+    };
+    let mut f = mode_fixture(mode, config).await;
+    // The router is gone in a moment, so the supervisor must not spend its full budget
+    // asking a participant that can no longer be reached.
+    f.harness.launcher.set_health_policy(fly_session::launcher::HealthPolicy {
+        probe: Duration::from_millis(200),
+        fail: Duration::from_millis(500),
+        boot: Duration::from_secs(30),
+    });
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let boundary_before = f.harness.coordinator.observation().unwrap().boundary;
+    let router = f.harness.router().clone();
+    let started = Instant::now();
+
+    let (coordinator, _launcher) = f.harness.parts();
+    let (stepped, ()) = tokio::join!(
+        async { within("step", coordinator.step()).await },
+        async {
+            // Mid-advance: the world has been asked to move and has not answered yet.
+            tokio::time::sleep(Duration::from_millis(250)).await;
+            router.shutdown();
+        }
+    );
+    let failure = stepped.expect_err("a lost router fails the epoch");
+    assert!(started.elapsed() < Duration::from_secs(20), "bounded, not a hang");
+    assert_eq!(
+        failure.participant.as_deref(),
+        Some(f.harness.environment_id().as_str()),
+        "the failure names the participant the coordinator was waiting for: {failure}"
+    );
+    assert_ne!(
+        failure.error.mutation,
+        MutationCertainty::None,
+        "the world may have stepped; a lost router is never proof that it did not"
+    );
+    assert_eq!(f.harness.coordinator.phase(), Phase::Failed);
+    assert!(
+        f.harness.coordinator.is_fenced(),
+        "old handles and routes are invalid from here on"
+    );
+    assert_eq!(f.harness.coordinator.stats().advances, 0, "no boundary was committed");
+    assert_eq!(count(&f.harness.coordinator.audit, "publish:1"), 0);
+    assert_eq!(
+        f.harness.coordinator.observation().unwrap().boundary,
+        boundary_before,
+        "the committed observation is still the one from before the advance"
+    );
+    // Nothing reconnects into the active epoch: a new call on the old route is refused.
+    let again = f.harness.coordinator.step().await.expect_err("a fenced epoch takes no step");
+    assert_eq!(again.error.code, ErrorCode::InvalidPhase);
+    f.shutdown().await;
+}
+
+/// Row: "Old worker replies after restore | Stale epoch/incarnation rejected", with real
+/// processes.
+///
+/// A restarted agent is a new process, a new registration and a new domain incarnation. The
+/// coordinator pinned the old registration, so its next call fails rather than reaching the
+/// replacement; and the replacement, followed deliberately, refuses an operation from the
+/// epoch the old process belonged to.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn an_old_worker_reply_after_a_restart_is_rejected_on_stale_epoch_or_incarnation() {
+    let mode = ExecutionMode::Process;
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    within("step", f.harness.coordinator.step()).await.unwrap();
+
+    let old = f.harness.coordinator.agent_ref(&fly_b()).cloned().unwrap();
+    let old_pid = f.harness.launcher.worker(&fly_b()).unwrap().pid;
+    assert!(old_pid.is_some(), "a separate-process agent has a process of its own");
+    let restarted = f.harness.restart_agent(&fly_b()).await.unwrap();
+    let new_pid = f.harness.launcher.worker(&fly_b()).unwrap().pid;
+    assert_ne!(old_pid, new_pid, "a restart is a new process");
+    assert_ne!(
+        restarted.service_incarnation, old.bus_incarnation,
+        "a replacement registration is a new incarnation"
+    );
+
+    // Following the new registration while still pinning the old worker's negotiated
+    // incarnation is rejected: this is the shape an old worker's reply would arrive in.
+    let stale = fly_session::rpc::WorkerRef {
+        service: restarted.service.clone(),
+        bus_incarnation: restarted.service_incarnation.clone(),
+        worker_id: fly_b(),
+        domain_incarnation: old.domain_incarnation.clone(),
+    };
+    assert_ne!(old.domain_incarnation, Some(restarted.incarnation_id.clone()));
+    let err = within("status", f.harness.coordinator.status(&stale))
+        .await
+        .expect_err("the replacement is not the incarnation this epoch negotiated");
+    assert_eq!(err.error.code, ErrorCode::IdentityMismatch);
+    assert_eq!(err.participant.as_deref(), Some(fly_b().as_str()));
+    assert_eq!(f.harness.coordinator.phase(), Phase::Failed);
+    assert!(f.harness.coordinator.is_fenced());
+    assert_eq!(f.harness.coordinator.stats().advances, 1, "no world step under a lost pin");
+    f.shutdown().await;
+}
+
+/// The other half of the same row: the replacement process is live and refuses an operation
+/// naming the epoch the old process belonged to, rather than applying it to a fresh brain.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_restarted_worker_refuses_an_operation_from_the_old_epoch() {
+    let mode = ExecutionMode::Process;
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    within("step", f.harness.coordinator.step()).await.unwrap();
+    let restarted = f.harness.restart_agent(&fly_b()).await.unwrap();
+
+    let replacement = fly_session::rpc::WorkerRef::new(
+        &restarted.service,
+        &restarted.service_incarnation,
+        &fly_b(),
+    );
+    let params = serde_json::json!({
+        "agentId": "fly-b",
+        "profileDigest": digest_of_bytes(b"whatever"),
+        "interval": {"numerator": "16666667", "denominator": "1"},
+        "decisionContextDigest": digest_of_bytes(b"whatever"),
+        "preStepStimulations": [],
+    });
+    let err = within(
+        "stale epoch",
+        f.harness.coordinator.probe_raw(
+            &replacement,
+            "Agent.Prepare",
+            Some(scope_at("demo", "e1", 1)),
+            params,
+        ),
+    )
+    .await
+    .expect_err("an uninitialized replacement has no epoch to prepare in");
+    assert!(
+        matches!(err.code, ErrorCode::StaleEpoch | ErrorCode::InvalidPhase),
+        "a replacement refuses the old epoch's work: {err}"
+    );
+    assert_eq!(err.mutation, MutationCertainty::None, "nothing was applied to a fresh brain");
+    assert_eq!(f.harness.coordinator.stats().advances, 1);
+    f.shutdown().await;
+}

services/flysim/crates/fly-session/tests/session.rs

@@ -379,12 +379,7 @@ async fn sequential_concurrent_and_reversed_orders_agree() {
 /// specific.
 async fn a_single_agent_composition_runs_the_same_transaction(via: Via) {
     let config = HarnessConfig {
-        agents: vec![AgentSpec {
+        agents: vec![AgentSpec::new("fly-a", "p1", 7)],
-            agent_id: id("fly-a"),
-            port_id: id("p1"),
-            seed: 7,
-            faults: AgentFaults::default(),
-        }],
         ..HarnessConfig::default()
     };
     let mut f: Fixture = fixture(via, config).await;