alex/flybrain
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

session: resolve uncertain calls, and measure each mode in its own process

Browse source 
Review fixes for SESSION-02.

An expired caller deadline was becoming a failed epoch without the
ipc-v1 section 6 resolution. That procedure existed and was correct and
had exactly one caller, a test injection, so the deadline this slice
introduced bypassed it and a merely slow participant lost its epoch.
Deadlines is now the two-stage shape section 6 describes -- a probe,
then a bounded resolve budget and attempt count -- call_owned returns a
typed CallOutcome so an expiry is distinguishable from a refusal, and
Prepare, Commit, Advance and the lifecycle calls all query the same
request id against the same incarnation before the epoch can fail. This
is also step-v1 section 7's Advance row, which was imperative about it.

The coordinator peak-RSS column was measuring the measuring process.
VmHWM never falls and every row shared one process, so the column was
cumulative and the mode ranking reversed when the rows were reordered.
Each row now runs in a measure-row child of its own. The corrected
numbers say the opposite of what the first report claimed: the
coordinator's own peak is roughly flat across the modes and lowest in
process mode, and the cost of the split is the children.

workers-v1 section 2 bounded Agent.Initialize's workerThreads by
"within launcher allocation" and named no wire for it. Dated amendment:
HelloResult.limits gains workerThreads, the worker reports what its
launcher gave it, and the launcher refuses one that disagrees. The
schema set, the shared fixtures and the TypeScript package move
together; contractDigest changes, which ipc-v1 section 4 provides for.

Also: the stale-epoch row now reaches the stale-epoch path against a
live agent process and asserts exact codes on both halves; the
router-restart row asserts the handle drop it claimed; frames are
counted from the behaviour trace instead of calculated; the README says
which suites run over which transports; bootstrap is fence-guarded; the
shutdown reason is an Id rather than a silent fallback; and
agent_mutations returns None rather than zero where the counter lives
in another process.
This commit is contained in:
acamilo 2026-09-22 15:41:42 +00:00
parent 438eaf3130
commit 77c8ee4558
22 changed files with 855 additions and 186 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
docs/design/session-framework/ipc-v1.md
View file

@ -94,10 +94,14 @@ interface HelloResult {
workerId: Id; incarnationId: Id; role: "agent" | "environment" | "coordinator"; workerId: Id; incarnationId: Id; role: "agent" | "environment" | "coordinator";
buildDigest: Digest; contractDigest: Digest; buildDigest: Digest; contractDigest: Digest;
capabilities: Id[]; capabilities: Id[];
limits: { maxAgents: number; maxPorts: number }; limits: { maxAgents: number; maxPorts: number; workerThreads: number };
} }
``` ```
`limits.workerThreads` is the thread allocation the worker's launcher started it within; it
is an integer >=1 and its rule belongs to [worker interfaces](workers-v1.md) section 2, whose
2026-09-22 amendment added it.
The bus supplies caller identity; do not accept a forged caller in params. Bind a worker's The bus supplies caller identity; do not accept a forged caller in params. Bind a worker's
session authority to the expected coordinator identity/incarnation during negotiation and session authority to the expected coordinator identity/incarnation during negotiation and
initialization. Wrong worker/role, no common major or missing required capability refuses initialization. Wrong worker/role, no common major or missing required capability refuses

10
docs/design/session-framework/workers-v1.md
View file

@ -99,6 +99,16 @@ interface AgentInitializeResult {
} }
``` ```
**Amendment, 2026-09-22 (SESSION-02).** `HelloResult.limits` gains `workerThreads`, an
integer >=1 reporting the allocation the launcher started that worker within, because
"within launcher allocation" above had no wire-level proof: the launcher passes the number to
the worker out of band, and a coordinator that is not also its own launcher had no contract
path to it. Hello is where a worker already proves its identity and reports its limits, so the
allocation belongs there. A caller asking for more than the worker reports is refused with
`BUSY` before the model is constructed, which this section already required; the amendment
only makes the number visible to whoever must respect it. It changes `contractDigest`, which
[session RPC](ipc-v1.md) section 4 already provides for.
The profile fixes warm-up/calibration behavior and supported schema versions. Validate inputs The profile fixes warm-up/calibration behavior and supported schema versions. Validate inputs
and required roles before model construction. Install the initial sensory input, warm the and required roles before model construction. Install the initial sensory input, warm the
brain with learning disabled, calibrate the fixed readout and establish Ready(0). Do not brain with learning disabled, calibrate the fixed readout and establish Ready(0). Do not

12
packages/session-types/src/workers.ts
View file

@ -43,6 +43,8 @@ export const MAX_ACKNOWLEDGE = 16;
export const MAX_ENGINE_FRAME_LEN = 64; export const MAX_ENGINE_FRAME_LEN = 64;
/** Not stated by a document; this crate's choice, published in the schema set. */ /** Not stated by a document; this crate's choice, published in the schema set. */
export const MAX_CAPABILITIES = 32; export const MAX_CAPABILITIES = 32;
/** The largest `workerThreads` a launcher may allocate to one worker (workers-v1 2). */
export const MAX_WORKER_THREADS = 4096;
export const MAX_SUPPORTED_MAJORS = 8; export const MAX_SUPPORTED_MAJORS = 8;
export const MAX_MESSAGE_CODE_POINTS = 512; export const MAX_MESSAGE_CODE_POINTS = 512;
@ -305,7 +307,7 @@ export function readAgentInitializeParams(value: unknown): AgentInitializeParams
seed: reader.int('seed', -2_147_483_648, 2_147_483_647), seed: reader.int('seed', -2_147_483_648, 2_147_483_647),
initialInput: readSensoryInput(reader.value('initialInput')), initialInput: readSensoryInput(reader.value('initialInput')),
initialDecisionContext: readTypedValue(reader.value('initialDecisionContext')), initialDecisionContext: readTypedValue(reader.value('initialDecisionContext')),
workerThreads: reader.int('workerThreads', 1, 4_096), workerThreads: reader.int('workerThreads', 1, MAX_WORKER_THREADS),
}; };
reader.finish(); reader.finish();
return params; return params;
@ -776,7 +778,12 @@ export interface HelloResult {
buildDigest: Digest; buildDigest: Digest;
contractDigest: Digest; contractDigest: Digest;
capabilities: Id[]; capabilities: Id[];
limits: { maxAgents: number; maxPorts: number }; /**
* `workerThreads` is the thread allocation this worker was launched within, added by the
* 2026-09-22 amendment to workers-v1 section 2: the bound "within launcher allocation" had
* no wire on which a caller could learn the allocation.
*/
limits: { maxAgents: number; maxPorts: number; workerThreads: number };
} }
export interface StatusResult { export interface StatusResult {
@ -860,6 +867,7 @@ export function readHelloResult(value: unknown): HelloResult {
const limits = { const limits = {
maxAgents: limitsReader.int('maxAgents', 1, MAX_AGENTS), maxAgents: limitsReader.int('maxAgents', 1, MAX_AGENTS),
maxPorts: limitsReader.int('maxPorts', 1, MAX_PORTS), maxPorts: limitsReader.int('maxPorts', 1, MAX_PORTS),
workerThreads: limitsReader.int('workerThreads', 1, MAX_WORKER_THREADS),
}; };
limitsReader.finish(); limitsReader.finish();
reader.finish(); reader.finish();

6
services/flysim/crates/fly-session-types/fixtures/contract-digest.json
View file

@ -1,9 +1,9 @@
{ {
"description": "contractDigest is the SHA-256 of the canonical schema set in schema-set.json.", "description": "contractDigest is the SHA-256 of the canonical schema set in schema-set.json.",
"contractDigest": "7932aef30c4d2d16e428081affc4e0ad187987f5b361138d553e54fd7f843b50", "contractDigest": "d8f29a49b5df05ad8f75f7f5790a3f8cde9c5ad23a685137474c649c3c9da36d",
"schemaSetVersion": 1, "schemaSetVersion": 1,
"schemaSetBytes": 26685, "schemaSetBytes": 26814,
"types": 53, "types": 53,
"enums": 11, "enums": 11,
"limits": 25 "limits": 26
} }

11
services/flysim/crates/fly-session-types/fixtures/invalid.json
View file

@ -2802,7 +2802,8 @@
], ],
"limits": { "limits": {
"maxAgents": 4, "maxAgents": 4,
"maxPorts": 4 "maxPorts": 4,
"workerThreads": 1
} }
}, },
"reason": "an agent must advertise agent-step-v1" "reason": "an agent must advertise agent-step-v1"
@ -2823,7 +2824,8 @@
], ],
"limits": { "limits": {
"maxAgents": 5, "maxAgents": 5,
"maxPorts": 4 "maxPorts": 4,
"workerThreads": 1
} }
}, },
"reason": "the first composition allows four agents" "reason": "the first composition allows four agents"
@ -2844,7 +2846,8 @@
], ],
"limits": { "limits": {
"maxAgents": 4, "maxAgents": 4,
"maxPorts": 4 "maxPorts": 4,
"workerThreads": 1
} }
}, },
"reason": "v1 selects major 1" "reason": "v1 selects major 1"
@ -3897,4 +3900,4 @@
"reason": "a commit acknowledges the transition's next boundary" "reason": "a commit acknowledges the transition's next boundary"
} }
] ]
} }

2
services/flysim/crates/fly-session-types/fixtures/schema-set.json
View file

File diff suppressed because one or more lines are too long

14
services/flysim/crates/fly-session-types/fixtures/valid.json
View file

@ -1758,12 +1758,13 @@
], ],
"limits": { "limits": {
"maxAgents": 4, "maxAgents": 4,
"maxPorts": 4 "maxPorts": 4,
"workerThreads": 1
} }
}, },
"note": "", "note": "",
"canonical": "{\"buildDigest\":\"44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66\",\"capabilities\":[\"agent-step-v1\",\"checkpoint-v1\",\"pixel-observation-v1\"],\"contractDigest\":\"cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf\",\"incarnationId\":\"inc-1\",\"limits\":{\"maxAgents\":4,\"maxPorts\":4},\"role\":\"agent\",\"selectedMajor\":1,\"selectedMinor\":0,\"workerId\":\"fly-a\"}", "canonical": "{\"buildDigest\":\"44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66\",\"capabilities\":[\"agent-step-v1\",\"checkpoint-v1\",\"pixel-observation-v1\"],\"contractDigest\":\"cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf\",\"incarnationId\":\"inc-1\",\"limits\":{\"maxAgents\":4,\"maxPorts\":4,\"workerThreads\":1},\"role\":\"agent\",\"selectedMajor\":1,\"selectedMinor\":0,\"workerId\":\"fly-a\"}",
"digest": "9f8cbc9dfe7a7532c2e41b53c4ce001973ca95703a81aad30ee2b1c8b640bc13" "digest": "262ea112e24dcdbdc1a5050b6dd6d031eace6a6a4cdca814913ac89fc9e39666"
}, },
{ {
"name": "hello result for an environment", "name": "hello result for an environment",
@ -1782,12 +1783,13 @@
], ],
"limits": { "limits": {
"maxAgents": 1, "maxAgents": 1,
"maxPorts": 1 "maxPorts": 1,
"workerThreads": 1
} }
}, },
"note": "", "note": "",
"canonical": "{\"buildDigest\":\"44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66\",\"capabilities\":[\"world-step-v1\",\"checkpoint-v1\"],\"contractDigest\":\"cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf\",\"incarnationId\":\"inc-2\",\"limits\":{\"maxAgents\":1,\"maxPorts\":1},\"role\":\"environment\",\"selectedMajor\":1,\"selectedMinor\":0,\"workerId\":\"world\"}", "canonical": "{\"buildDigest\":\"44575cf5b28512d75644bf54a517dcef304ff809fd511747621b4d64f19aac66\",\"capabilities\":[\"world-step-v1\",\"checkpoint-v1\"],\"contractDigest\":\"cc8321d6375c494d043fdd0260f21bc0ec51dacc9f6abb7f909cdcd3041b78bf\",\"incarnationId\":\"inc-2\",\"limits\":{\"maxAgents\":1,\"maxPorts\":1,\"workerThreads\":1},\"role\":\"environment\",\"selectedMajor\":1,\"selectedMinor\":0,\"workerId\":\"world\"}",
"digest": "0dc33a6cbf6bf66c4d28dc41b17b2eb7918b55c65469f4161e009f662f11bca2" "digest": "06ced6fff0d810005ab2598cc9fb72a138303daef0787232871c9afa43b0eac0"
}, },
{ {
"name": "status result before initialization", "name": "status result before initialization",

9
services/flysim/crates/fly-session-types/src/schema.rs
View file

@ -239,6 +239,11 @@ pub const LIMITS: &[LimitSchema] = &[
value: crate::workers::MAX_CAPABILITIES as u64, value: crate::workers::MAX_CAPABILITIES as u64,
source: "crate", source: "crate",
}, },
LimitSchema {
name: "maxWorkerThreads",
value: crate::workers::MAX_WORKER_THREADS,
source: "workers-v1 2",
},
LimitSchema { LimitSchema {
name: "maxSupportedMajors", name: "maxSupportedMajors",
value: crate::workers::MAX_SUPPORTED_MAJORS as u64, value: crate::workers::MAX_SUPPORTED_MAJORS as u64,
@ -651,8 +656,8 @@ pub const SCHEMAS: &[TypeSchema] = &[
), ),
req( req(
"limits", "limits",
"{maxAgents:int,maxPorts:int}", "{maxAgents:int,maxPorts:int,workerThreads:int}",
"1..=4 agents and 1..=4 ports", "1..=4 agents, 1..=4 ports, and the launcher allocation this worker runs in",
), ),
], ],
}, },

19
services/flysim/crates/fly-session-types/src/workers.rs
View file

@ -33,6 +33,9 @@ pub const MAX_ACKNOWLEDGE: usize = 16;
pub const MAX_ENGINE_FRAME_LEN: usize = 64; pub const MAX_ENGINE_FRAME_LEN: usize = 64;
/// Negotiated capability ids. Not a stated bound; recorded in the schema set. /// Negotiated capability ids. Not a stated bound; recorded in the schema set.
pub const MAX_CAPABILITIES: usize = 32; pub const MAX_CAPABILITIES: usize = 32;
/// The largest `workerThreads` a launcher may allocate to one worker (`workers-v1` 2).
pub const MAX_WORKER_THREADS: u64 = 4_096;
/// Supported majors in Worker.Hello. Not a stated bound; recorded in the schema set. /// Supported majors in Worker.Hello. Not a stated bound; recorded in the schema set.
pub const MAX_SUPPORTED_MAJORS: usize = 8; pub const MAX_SUPPORTED_MAJORS: usize = 8;
/// Domain error messages are <=512 code points (ipc-v1 section 7). /// Domain error messages are <=512 code points (ipc-v1 section 7).
@ -651,7 +654,7 @@ impl DomainType for AgentInitializeParams {
let seed = i32_field(&mut f, "seed")?; let seed = i32_field(&mut f, "seed")?;
let initial_input = SensoryInput::from_json(f.value("initialInput")?)?; let initial_input = SensoryInput::from_json(f.value("initialInput")?)?;
let initial_decision_context = TypedValue::from_json(f.value("initialDecisionContext")?)?; let initial_decision_context = TypedValue::from_json(f.value("initialDecisionContext")?)?;
let worker_threads = f.int("workerThreads", 1, 4_096)?; let worker_threads = f.int("workerThreads", 1, MAX_WORKER_THREADS)?;
f.finish()?; f.finish()?;
let p = AgentInitializeParams { let p = AgentInitializeParams {
agent_id, agent_id,
@ -1958,6 +1961,13 @@ pub struct HelloResult {
pub capabilities: Vec<String>, pub capabilities: Vec<String>,
pub max_agents: u64, pub max_agents: u64,
pub max_ports: u64, pub max_ports: u64,
/// The thread allocation this worker was launched within.
///
/// `workers-v1` bounds `Agent.Initialize`'s `workerThreads` by "within launcher
/// allocation" and, before the 2026-09-22 amendment, named no wire on which a caller could
/// learn it. This is that wire: the worker reports what its launcher gave it, and a caller
/// that meant to ask for more finds out here rather than after the model exists.
pub worker_threads: u64,
} }
impl HelloResult { impl HelloResult {
@ -1990,13 +2000,14 @@ impl DomainType for HelloResult {
let build_digest = f.string("buildDigest")?.to_owned(); let build_digest = f.string("buildDigest")?.to_owned();
let contract_digest = f.string("contractDigest")?.to_owned(); let contract_digest = f.string("contractDigest")?.to_owned();
let capabilities = id_list(&mut f, "capabilities", 0, MAX_CAPABILITIES)?; let capabilities = id_list(&mut f, "capabilities", 0, MAX_CAPABILITIES)?;
let (max_agents, max_ports) = { let (max_agents, max_ports, worker_threads) = {
let v = f.value("limits")?; let v = f.value("limits")?;
let mut l = Fields::new(v, "HelloResult.limits")?; let mut l = Fields::new(v, "HelloResult.limits")?;
let max_agents = l.int("maxAgents", 1, MAX_AGENTS as u64)?; let max_agents = l.int("maxAgents", 1, MAX_AGENTS as u64)?;
let max_ports = l.int("maxPorts", 1, MAX_PORTS as u64)?; let max_ports = l.int("maxPorts", 1, MAX_PORTS as u64)?;
let worker_threads = l.int("workerThreads", 1, MAX_WORKER_THREADS)?;
l.finish()?; l.finish()?;
(max_agents, max_ports) (max_agents, max_ports, worker_threads)
}; };
f.finish()?; f.finish()?;
let r = HelloResult { let r = HelloResult {
@ -2008,6 +2019,7 @@ impl DomainType for HelloResult {
capabilities, capabilities,
max_agents, max_agents,
max_ports, max_ports,
worker_threads,
}; };
r.validate()?; r.validate()?;
Ok(r) Ok(r)
@ -2031,6 +2043,7 @@ impl DomainType for HelloResult {
obj(vec![ obj(vec![
("maxAgents", Value::from(self.max_agents)), ("maxAgents", Value::from(self.max_agents)),
("maxPorts", Value::from(self.max_ports)), ("maxPorts", Value::from(self.max_ports)),
("workerThreads", Value::from(self.worker_threads)),
]), ]),
), ),
]) ])

80
services/flysim/crates/fly-session/README.md
View file

@ -64,12 +64,15 @@ The launcher is the configured supervisor. It owns four things:
its router, and one allocation per participant. A request the total cannot cover is refused 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 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 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". is what `workers-v1` means by "within launcher allocation". The allocation is on the wire,
not only in the launcher's own record: `HelloResult.limits.workerThreads` reports it, under
the dated 2026-09-22 amendment to `workers-v1` section 2 that this slice added, so a
coordinator that is not also its own launcher can read the bound it has to respect.
- **Identity.** The bus client id, the service name, the worker id and an agent's port binding - **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 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 and refuses anything that answers as another worker, role, incarnation or thread allocation
coordinator has pinned a registration. The registration the coordinator pins is the one that -- before the coordinator has pinned a registration. The registration the coordinator pins
hello returned, never one that was assumed. 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` - **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. 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. A status answer never waits for a mutation, so a busy participant is still a healthy one.
@ -112,10 +115,19 @@ fly-session measure --steps 300 --agents 1,2,4
participant it is attributed to, and failing fences the session: the committed boundary 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 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. 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 - **The `ipc-v1` section 6 procedure, on the path that reaches it.** A call that goes without
coordinator's own clock, so a participant that dies or stops answering produces a typed a terminal reply for the probe budget is *uncertain*, not failed. The coordinator then
failure naming it rather than a hang. An expired deadline is `unknown`, never `none`: a queries the same operation -- a fresh bus call carrying the original domain request id and
caller-side timeout is not evidence that nothing was mutated. body, pinned to the same incarnation, with its retained attachments -- for a bounded number
of attempts within a bounded budget, absorbing `IN_PROGRESS` while the original is still
running. Only when that ends without a definite answer, or the incarnation is gone, or the
retained result expired, is the epoch failed. A merely slow participant therefore finishes
its step, and `step-v1` section 7's "query/retransmit same request to same incarnation;
never new batch" is the same code path for a slow Advance.
- **A bounded diagnosed outcome.** Those budgets are the coordinator's own, on its 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 - **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 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 duplicate of a running operation is `IN_PROGRESS` for that bus call while the original
@ -192,20 +204,34 @@ machine and no host capacity claim follows from any of them**; they exist so the
can be compared with each other. Pacing is off for the run, so the samples are work rather 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. than sleep, and the run report carries the full table.
Every row runs in a child process of its own. A peak-memory figure is a high-water mark that
never falls, so rows sharing one process would each report where that process had already
been: the column would sort itself by row position rather than by mode, and the mode ranking
would reverse when the rows were reordered. One child per row is what makes the number belong
to the row.
What the numbers said on a four-core development box, at 300 transitions per row: 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 - A process boundary costs about a fifth of the critical path at the median. Two agents: 10.0
critical path was about 7.8 ms p50 in-process, 8.6 ms on threads and 12.7 ms across ms p50 in-process, 12.2 ms on threads, 12.1 ms across processes, with p99 at 21.4 / 19.4 /
processes, while p99 went 12.4 / 12.7 / 26.0 ms. The medians are within a small multiple of 21.4 ms. The dedicated-thread and separate-process variants are within noise of each other,
each other; the tails are where a scheduler with more runnable threads than cores appears. so what is being paid for is leaving the coordinator's runtime, not crossing a socket.
- `Worker.Status` -- an RPC answered from a cell with no domain work behind it -- is the
router and transport floor: 0.60 / 0.89 / 1.17 ms p50 for the three modes at two agents.
- Four agents needs six threads, which that box does not have, and every mode's tail widens - 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. together. That is the budget being honest about oversubscription, not a property of the
- Memory is the clearest difference: one coordinator at about 14 MiB peak RSS plus roughly process split.
5.6 MiB per participant process, against a single 11 MiB process for the threaded variant. - Memory is where the split really shows, but not where the first version of this note said.
- Ownership and queues stayed bounded in every mode and at every agent count: at most 15 live The coordinator's own peak is roughly the same in all three modes and is *lowest* in process
owners, 11 artifact roots and one queue entry per agent, with the store holding two sealed mode -- 8.9 / 9.2 / 7.9 MiB at one agent -- because the workers are no longer inside it.
frames and 128 bytes at rest. Of 311 frames produced, 309 were collected -- the current and What the split costs is the children: about 5.7 MiB per participant process, so the whole
previous boundary are the two that are still owned. composition is roughly 9 MiB on threads against 39 MiB across processes at four agents.
- Ownership, collection and queues stayed bounded in every mode and at every agent count: at
most 15 live owners, 11 artifact roots and one queued entry per agent, with the store at
rest holding two sealed frames and 128 bytes. Of 311 frames observed, 309 were collected --
the two still owned are the current and previous boundary. The frame count is taken from the
behaviour trace's observation boundaries rather than calculated from the step count, so a
backend sealing two frames per boundary would show up instead of being hidden.
## Tests ## Tests
@ -216,10 +242,15 @@ cargo build -p fly-session --bin fly-session # the worker binary
cargo run -p fly-session --example processes # the same session in all three modes 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 The three integration suites do not all run over both transports, and cannot:
are generated twice by `both_transports!`, and
`sequential_concurrent_and_reversed_orders_agree` walks both transports inside one test - `tests/session.rs` and `tests/failures.rs` are in-process compositions and run over both,
because it compares their behaviour traces against each other. through the same router code. All but one test in them is generated twice by
`both_transports!`; `sequential_concurrent_and_reversed_orders_agree` walks both transports
inside one test, because it compares their behaviour traces against each other.
- `tests/processes.rs` runs over the Unix socket only, in all three execution modes. A
participant in a process of its own has no in-memory transport to reach the router by, so
the mode is the axis that suite varies and the transport is fixed.
- `tests/session.rs`: one world advance per complete batch; every agent Prepared before the - `tests/session.rs`: one world advance per complete batch; every agent Prepared before the
advance; one task evaluation per transition; every agent committed before the next Prepare or advance; one task evaluation per transition; every agent committed before the next Prepare or
@ -229,7 +260,8 @@ because it compares their behaviour traces against each other.
during a session; and sequential, concurrent and reversed dispatch producing one behaviour during a session; and sequential, concurrent and reversed dispatch producing one behaviour
trace. trace.
- `tests/processes.rs`: the SESSION-02 acceptance bullets, each generated once per execution - `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 mode -- a slow participant resolved rather than failed, 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 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 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 allocation -- plus the sequential/reversed/parallel trace comparison across all three modes

2
services/flysim/crates/fly-session/examples/processes.rs
View file

@ -63,7 +63,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!(" behaviour trace: identical to the first run"); println!(" behaviour trace: identical to the first run");
} }
} }
let reaped = harness.launcher.reap_all("example").await; let reaped = harness.launcher.reap_all(&fly_session::types::id("example")).await;
for (worker_id, outcome) in reaped { for (worker_id, outcome) in reaped {
println!(" reaped {worker_id}: {outcome:?}"); println!(" reaped {worker_id}: {outcome:?}");
} }

4
services/flysim/crates/fly-session/src/agent.rs
View file

@ -650,6 +650,10 @@ impl WorkerEndpoint for FakeAgentWorker {
self.status.clone() self.status.clone()
} }
fn worker_threads(&self) -> u64 {
self.config.worker_threads as u64
}
fn methods(&self) -> Vec<&'static str> { fn methods(&self) -> Vec<&'static str> {
vec!["Agent.Initialize", "Agent.Prepare", "Agent.Commit"] vec!["Agent.Initialize", "Agent.Prepare", "Agent.Commit"]
} }

51
services/flysim/crates/fly-session/src/cli.rs
View file

@ -30,6 +30,7 @@ fly-session <command> [options]
agent serve one agent worker on a launcher-created endpoint agent serve one agent worker on a launcher-created endpoint
environment serve the environment 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 measure compare the execution modes and print the measurement table
measure-row measure one row and print it as JSON (one child per row)
Worker options (agent and environment): Worker options (agent and environment):
--socket PATH the launcher's endpoint for this participant --socket PATH the launcher's endpoint for this participant
@ -48,9 +49,14 @@ Worker options (agent and environment):
Measure options: Measure options:
--steps N transitions per run (default 200) --steps N transitions per run (default 200)
--warmup-steps N transitions run before sampling starts (default 10)
--agents 1,2,4 agent counts to compare (default 1,2,4) --agents 1,2,4 agent counts to compare (default 1,2,4)
--modes LIST in-process, thread, process (default all three) --modes LIST in-process, thread, process (default all three)
--worker-threads N within-agent worker threads (default 1) --worker-threads N within-agent worker threads (default 1)
measure-row options: --mode NAME --agents N, plus the measure options above. Each row runs in
a process of its own, so its memory peak is its own rather than the peak of the rows before
it.
"; ";
/// The binary's entry point. /// The binary's entry point.
@ -65,6 +71,7 @@ pub fn main() -> ExitCode {
let result = match command.as_str() { let result = match command.as_str() {
"agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)), "agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)),
"measure" => Options::parse(&rest).and_then(|o| measure(&o)), "measure" => Options::parse(&rest).and_then(|o| measure(&o)),
"measure-row" => Options::parse(&rest).and_then(|o| measure_row(&o)),
"--help" | "-h" | "help" => { "--help" | "-h" | "help" => {
print!("{USAGE}"); print!("{USAGE}");
return ExitCode::SUCCESS; return ExitCode::SUCCESS;
@ -215,10 +222,10 @@ fn parse_ports(value: &str) -> Result<Vec<Id>, String> {
.collect() .collect()
} }
/// Runs the execution-mode comparison and prints its table. fn measure_config(options: &Options) -> Result<crate::measure::MeasureConfig, String> {
fn measure(options: &Options) -> Result<(), String> {
let mut config = crate::measure::MeasureConfig { let mut config = crate::measure::MeasureConfig {
steps: options.u64("steps", 200)?, steps: options.u64("steps", 200)?,
warmup_steps: options.u64("warmup-steps", 10)?,
worker_threads: options.usize("worker-threads", 1)?, worker_threads: options.usize("worker-threads", 1)?,
..crate::measure::MeasureConfig::default() ..crate::measure::MeasureConfig::default()
}; };
@ -233,19 +240,43 @@ fn measure(options: &Options) -> Result<(), String> {
config.modes = list config.modes = list
.split(',') .split(',')
.filter(|p| !p.is_empty()) .filter(|p| !p.is_empty())
.map(|p| match p { .map(parse_mode)
"in-process" => Ok(ExecutionMode::InProcess),
"thread" => Ok(ExecutionMode::Thread),
"process" => Ok(ExecutionMode::Process),
other => Err(format!("--modes: {other:?}")),
})
.collect::<Result<Vec<ExecutionMode>, String>>()?; .collect::<Result<Vec<ExecutionMode>, String>>()?;
} }
Ok(config)
}
fn parse_mode(name: &str) -> Result<ExecutionMode, String> {
match name {
"in-process" => Ok(ExecutionMode::InProcess),
"thread" => Ok(ExecutionMode::Thread),
"process" => Ok(ExecutionMode::Process),
other => Err(format!("unknown mode {other:?}")),
}
}
/// Runs the execution-mode comparison and prints its table.
///
/// One child per row: a peak-memory figure is only that row's if nothing else ran in the
/// process that produced it.
fn measure(options: &Options) -> Result<(), String> {
let config = measure_config(options)?;
let program = std::env::current_exe().map_err(|e| format!("current exe: {e}"))?;
let rows = crate::measure::run(&config, &program)?;
print!("{}", crate::measure::table(&rows));
Ok(())
}
/// Measures exactly one row and prints it as one JSON object. The parent's child.
fn measure_row(options: &Options) -> Result<(), String> {
let config = measure_config(options)?;
let mode = parse_mode(options.required("mode")?)?;
let agents = options.usize("agents", 2)?;
let runtime = tokio::runtime::Builder::new_multi_thread() let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all() .enable_all()
.build() .build()
.map_err(|e| format!("runtime: {e}"))?; .map_err(|e| format!("runtime: {e}"))?;
let rows = runtime.block_on(crate::measure::run(&config))?; let row = runtime.block_on(crate::measure::one(&config, mode, agents))?;
print!("{}", crate::measure::table(&rows)); println!("{}", row.to_json());
Ok(()) Ok(())
} }

363
services/flysim/crates/fly-session/src/coordinator.rs
View file

@ -107,21 +107,41 @@ impl std::error::Error for SessionFailure {}
type Outcome<T> = Result<T, SessionFailure>; type Outcome<T> = Result<T, SessionFailure>;
/// How long the coordinator waits for a participant before it calls the call uncertain. /// The caller-side failure-detection budgets of `ipc-v1` section 6, on the coordinator's own
/// monotonic clock.
/// ///
/// `ipc-v1` section 6 measures these on the caller's own monotonic clock and gives the /// Section 6 is a two-stage procedure, and these are its two stages. `probe` is how long a
/// prototype values: ten seconds without progress is a failure, with a separate budget for a /// call may go without a terminal reply before it is *uncertain*; an uncertain call is not a
/// long boot. They are failure-detection values, not a gameplay latency goal. Without them a /// failed one, so the coordinator then runs the section 6 resolution -- a fresh bus call
/// dead participant is a hang rather than a diagnosed outcome. /// carrying the original domain request id and body, pinned to the same incarnation -- for at
/// most `resolve` and at most `resolve_attempts` tries. Only when that ends without a definite
/// answer, or the incarnation is gone, or the retained result expired, is the epoch failed.
///
/// The prototype values follow section 6: probe at two seconds without a reply, give up at
/// ten seconds without progress, with a separate budget for a long boot. They are
/// failure-detection values, not a gameplay latency goal. The attempt count is explicit
/// because section 6 forbids filling this gap with an implicit best-effort policy.
#[derive(Clone, Copy, Debug)] #[derive(Clone, Copy, Debug)]
pub struct Deadlines { pub struct Deadlines {
pub call: Duration, /// Without a terminal reply for this long, the call is uncertain.
pub probe: Duration,
/// The resolution's own budget, measured from its first attempt.
pub resolve: Duration,
/// How many times the resolution may re-ask. Bounded, and never a retry of the operation:
/// every attempt carries the original request id and body.
pub resolve_attempts: u32,
/// A separate, larger budget for `Worker.Hello` and the `Initialize` methods.
pub boot: Duration, pub boot: Duration,
} }
impl Default for Deadlines { impl Default for Deadlines {
fn default() -> Deadlines { fn default() -> Deadlines {
Deadlines { call: Duration::from_secs(10), boot: Duration::from_secs(30) } Deadlines {
probe: Duration::from_secs(2),
resolve: Duration::from_secs(8),
resolve_attempts: 512,
boot: Duration::from_secs(30),
}
} }
} }
@ -231,6 +251,8 @@ pub struct Coordinator {
pub injection_log: Vec<InjectionOutcome>, pub injection_log: Vec<InjectionOutcome>,
/// How many times an exact duplicate met IN_PROGRESS while resolving an uncertain call. /// How many times an exact duplicate met IN_PROGRESS while resolving an uncertain call.
pub in_progress_replies: u64, pub in_progress_replies: u64,
/// How many uncertain calls ran the `ipc-v1` section 6 resolution.
pub resolutions: u64,
/// The caller-side failure-detection budgets of `ipc-v1` section 6. /// The caller-side failure-detection budgets of `ipc-v1` section 6.
pub deadlines: Deadlines, pub deadlines: Deadlines,
/// Per-method and critical-path latency samples. Local synthetic timings, never a /// Per-method and critical-path latency samples. Local synthetic timings, never a
@ -289,6 +311,7 @@ impl Coordinator {
injections: Injections::default(), injections: Injections::default(),
injection_log: Vec::new(), injection_log: Vec::new(),
in_progress_replies: 0, in_progress_replies: 0,
resolutions: 0,
deadlines: Deadlines::default(), deadlines: Deadlines::default(),
metrics: Metrics::default(), metrics: Metrics::default(),
blame: None, blame: None,
@ -429,6 +452,15 @@ impl Coordinator {
self.fenced self.fenced
} }
/// How many artifact handles this session still owns: the committed boundary's views plus
/// any the world has produced but not yet committed.
///
/// A fence drops both sets, which is what "old handles are invalid" means on this side of
/// the router. It is exposed so that can be asserted rather than described.
pub fn live_view_handles(&self) -> usize {
self.views.len() + self.pending_views.len()
}
fn agent(&self, agent_id: &Id) -> Option<&AgentSlot> { fn agent(&self, agent_id: &Id) -> Option<&AgentSlot> {
self.agents.iter().find(|a| a.agent_id == *agent_id) self.agents.iter().find(|a| a.agent_id == *agent_id)
} }
@ -441,6 +473,20 @@ impl Coordinator {
/// ///
/// Nothing here advances the environment or produces a gameplay reward. /// Nothing here advances the environment or produces a gameplay reward.
pub async fn bootstrap(&mut self) -> Outcome<()> { pub async fn bootstrap(&mut self) -> Outcome<()> {
if self.fenced {
// Defence in depth: the phase machine refuses `Failed -> Ready(0)` anyway, but a
// fenced session should not be sending Hello and Initialize to live workers on the
// way to finding that out.
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,
});
}
self.hello_environment().await?; self.hello_environment().await?;
for index in 0..self.agents.len() { for index in 0..self.agents.len() {
self.hello_agent(index).await?; self.hello_agent(index).await?;
@ -790,14 +836,30 @@ struct Job {
request_id: DomainRequestId, request_id: DomainRequestId,
} }
/// Issues one domain call with owned arguments, so it can run in its own task.
///
/// What one bus call came back as.
///
/// The expiry is its own variant rather than an error, because `ipc-v1` section 6 treats the
/// two differently: a refusal is an answer and can fail the epoch, while an expired deadline
/// is only an *uncertain* call and owes the resolution procedure first. Collapsing them into
/// one error is how a merely slow participant loses an epoch.
enum CallOutcome {
// Boxed: a `DomainReply` carries its artifact handles, and the other two variants are a
// unit and one error. Without the box every caller's `Result` is sized for the reply.
Answered(Box<DomainReply>),
/// The caller's deadline expired with no terminal reply.
Expired,
/// The bus or the reply itself refused. This is an answer, even when it is a bad one.
Refused(DomainError),
}
/// Issues one domain call with owned arguments, so it can run in its own task. /// 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 /// 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 /// 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 /// 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 /// stopped answering -- a diagnosed outcome rather than a hang.
/// deliberately `unknown`: `ipc-v1` section 6 forbids reading a caller-side timeout as proof
/// that nothing was mutated.
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
async fn call_owned( async fn call_owned(
bus: flybus::Client, bus: flybus::Client,
@ -809,30 +871,46 @@ async fn call_owned(
request_id: DomainRequestId, request_id: DomainRequestId,
want: Vec<String>, want: Vec<String>,
deadline: Duration, deadline: Duration,
) -> Result<DomainReply, DomainError> { ) -> CallOutcome {
let refs: Vec<(&str, &flybus::Artifact)> = let refs: Vec<(&str, &flybus::Artifact)> =
attachments.iter().map(|(n, a)| (n.as_str(), a)).collect(); attachments.iter().map(|(n, a)| (n.as_str(), a)).collect();
let call = rpc::call(&bus, &worker, method, scope, params, &refs, request_id, &want); let call = rpc::call(&bus, &worker, method, scope, params, &refs, request_id, &want);
match tokio::time::timeout(deadline, call).await { match tokio::time::timeout(deadline, call).await {
Ok(result) => result, Ok(Ok(reply)) => CallOutcome::Answered(Box::new(reply)),
Err(_) => Err(DomainError::new( Ok(Err(e)) => CallOutcome::Refused(e),
ErrorCode::BackendFailure, Err(_) => CallOutcome::Expired,
format!(
"{method}: {} did not answer within {:?}",
worker.worker_id, deadline
),
MutationCertainty::Unknown,
)),
} }
} }
/// The error an unresolved uncertain call finally becomes.
///
/// Deliberately `unknown`: `ipc-v1` section 6 forbids reading a caller-side timeout as proof
/// that nothing was mutated.
fn unresolved(method: &str, worker: &WorkerRef, budget: Duration) -> DomainError {
DomainError::new(
ErrorCode::BackendFailure,
format!(
"{method}: {} never resolved within {:?}; the operation's outcome is unknown",
worker.worker_id, budget
),
MutationCertainty::Unknown,
)
}
/// One per-agent call's result, in the shape the phase loops read it. /// One per-agent call's result, in the shape the phase loops read it.
///
/// The request is carried back beside the outcome so an uncertain call can be resolved
/// against its *original* domain request id and body. Recomputing it would be a second
/// operation, which `step-v1` section 7 forbids.
struct JobResult { struct JobResult {
agent_id: Id, agent_id: Id,
reply: Result<DomainReply, DomainError>, outcome: CallOutcome,
scope: Option<Scope>, scope: Option<Scope>,
worker: WorkerRef, worker: WorkerRef,
method: &'static str, method: &'static str,
request_id: DomainRequestId,
params: Map<String, Value>,
attachments: Vec<(String, flybus::Artifact)>,
/// How long the call took, for the section 5 percentiles. /// How long the call took, for the section 5 percentiles.
elapsed: Duration, elapsed: Duration,
} }
@ -859,25 +937,31 @@ impl Coordinator {
let deadline = if method.ends_with("Initialize") || method == "Worker.Hello" { let deadline = if method.ends_with("Initialize") || method == "Worker.Hello" {
self.deadlines.boot self.deadlines.boot
} else { } else {
self.deadlines.call self.deadlines.probe
}; };
let started = Instant::now(); let started = Instant::now();
let reply = call_owned( let outcome = call_owned(
self.bus.clone(), self.bus.clone(),
worker.clone(), worker.clone(),
method, method,
scope.clone(), scope.clone(),
params, params.clone(),
owned, owned.clone(),
request_id, request_id.clone(),
want.to_vec(), want.to_vec(),
deadline, deadline,
) )
.await; .await;
self.metrics.record(method, started.elapsed()); self.metrics.record(method, started.elapsed());
let reply = match reply { let reply = match outcome {
Ok(reply) => reply, CallOutcome::Answered(reply) => *reply,
Err(e) => return Err(self.fail_now(e, method)), // Uncertain, not failed: section 6 owes this call its resolution first.
CallOutcome::Expired => {
return self
.resolve(worker, method, scope.clone(), params, owned, request_id, want)
.await;
}
CallOutcome::Refused(e) => return Err(self.fail_now(e, method)),
}; };
self.check_reply(worker, &reply, &scope, method)?; self.check_reply(worker, &reply, &scope, method)?;
match reply.result() { match reply.result() {
@ -945,9 +1029,22 @@ impl Coordinator {
Ok(()) Ok(())
} }
/// Resolves an uncertain operation: a fresh bus call with the original domain request id /// The `ipc-v1` section 6 resolution of an uncertain call.
/// and body, pinned to the same service incarnation. It never issues a new request id and ///
/// never recomputes a decision. /// Step 2 of that section: while the same bus, service and worker incarnation still exist,
/// issue a fresh bus call carrying the **original** domain request id and body with its
/// retained input attachments. The worker's own deduplication answers it from the record
/// of the first attempt, so this queries rather than repeats: it never issues a new request
/// id, never recomputes a decision, and never becomes a second batch.
///
/// Step 3: only a matching terminal result resolves it. `IN_PROGRESS` means the original is
/// still running and no second mutation was started, so the procedure waits and asks again.
/// An expiry inside the procedure is likewise not an answer.
///
/// Step 4: the epoch fails when the routes or ownership were lost, the incarnation changed,
/// the retained result expired, or the procedure's own bounded budget ran out. Those bounds
/// are [`Deadlines`] and are explicit, because section 6 refuses an implicit best-effort
/// policy here.
#[allow(clippy::too_many_arguments)] #[allow(clippy::too_many_arguments)]
async fn resolve( async fn resolve(
&mut self, &mut self,
@ -963,9 +1060,17 @@ impl Coordinator {
// starts no second mutation. Waiting and asking again is the resolution, not a retry // starts no second mutation. Waiting and asking again is the resolution, not a retry
// of the operation. // of the operation.
self.blame(Some(worker.worker_id.clone())); self.blame(Some(worker.worker_id.clone()));
let deadline = self.deadlines.call; let probe = self.deadlines.probe;
for attempt in 0..200u32 { let budget = self.deadlines.resolve;
let reply = call_owned( let attempts = self.deadlines.resolve_attempts;
let started = Instant::now();
self.resolutions += 1;
self.audit.push(format!("resolve:{}:{method}", worker.worker_id));
for _ in 0..attempts {
if started.elapsed() >= budget {
break;
}
let outcome = call_owned(
self.bus.clone(), self.bus.clone(),
worker.clone(), worker.clone(),
method, method,
@ -974,12 +1079,19 @@ impl Coordinator {
attachments.clone(), attachments.clone(),
request_id.clone(), request_id.clone(),
want.to_vec(), want.to_vec(),
deadline, probe,
) )
.await; .await;
let reply = match reply { let reply = match outcome {
Ok(reply) => reply, CallOutcome::Answered(reply) => *reply,
Err(e) => return Err(self.fail_now(e, method)), // Still no answer. The original may simply be slow; asking again is the
// procedure, and the request id it carries is unchanged.
CallOutcome::Expired => {
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
continue;
}
// Step 4: routes or ownership lost, or the incarnation is gone.
CallOutcome::Refused(e) => return Err(self.fail_now(e, method)),
}; };
self.check_reply(worker, &reply, &scope, method)?; self.check_reply(worker, &reply, &scope, method)?;
match reply.result() { match reply.result() {
@ -988,21 +1100,14 @@ impl Coordinator {
return Ok(reply); return Ok(reply);
} }
Err(e) if e.code == ErrorCode::InProgress => { Err(e) if e.code == ErrorCode::InProgress => {
let _ = attempt;
self.in_progress_replies += 1; self.in_progress_replies += 1;
tokio::time::sleep(std::time::Duration::from_millis(2)).await; tokio::time::sleep(std::time::Duration::from_millis(2)).await;
} }
// A terminal refusal, including RESULT_EXPIRED: definite, so the epoch fails.
Err(e) => return Err(self.fail_now(e, method)), Err(e) => return Err(self.fail_now(e, method)),
} }
} }
Err(self.fail_now( Err(self.fail_now(unresolved(method, worker, budget), method))
DomainError::new(
ErrorCode::BackendFailure,
"the uncertain operation never resolved",
MutationCertainty::Unknown,
),
method,
))
} }
/// Sends the same domain request again on a fresh bus call and reports what came back, /// Sends the same domain request again on a fresh bus call and reports what came back,
@ -1019,7 +1124,7 @@ impl Coordinator {
expected: Option<&Value>, expected: Option<&Value>,
what: &str, what: &str,
) { ) {
let deadline = self.deadlines.call; let deadline = self.deadlines.probe;
let reply = call_owned( let reply = call_owned(
self.bus.clone(), self.bus.clone(),
worker.clone(), worker.clone(),
@ -1033,7 +1138,7 @@ impl Coordinator {
) )
.await; .await;
let outcome = match reply { let outcome = match reply {
Ok(reply) => match reply.result() { CallOutcome::Answered(reply) => match reply.result() {
Ok(result) => InjectionOutcome { Ok(result) => InjectionOutcome {
what: what.to_owned(), what: what.to_owned(),
code: None, code: None,
@ -1045,11 +1150,18 @@ impl Coordinator {
identical: false, identical: false,
}, },
}, },
Err(e) => InjectionOutcome { CallOutcome::Refused(e) => InjectionOutcome {
what: what.to_owned(), what: what.to_owned(),
code: Some(e.code), code: Some(e.code),
identical: false, identical: false,
}, },
// A probe is a diagnostic, not a phase of the transaction: it reports what it saw
// and never resolves anything on the session's behalf.
CallOutcome::Expired => InjectionOutcome {
what: what.to_owned(),
code: Some(ErrorCode::BackendFailure),
identical: false,
},
}; };
self.injection_log.push(outcome); self.injection_log.push(outcome);
} }
@ -1071,7 +1183,7 @@ impl Coordinator {
_ => Map::new(), _ => Map::new(),
}; };
let request_id = self.serials.next(&worker.service); let request_id = self.serials.next(&worker.service);
let reply = call_owned( let outcome = call_owned(
self.bus.clone(), self.bus.clone(),
worker.clone(), worker.clone(),
method, method,
@ -1080,10 +1192,14 @@ impl Coordinator {
Vec::new(), Vec::new(),
request_id, request_id,
Vec::new(), Vec::new(),
self.deadlines.call, self.deadlines.probe,
) )
.await?; .await;
reply.result().cloned() match outcome {
CallOutcome::Answered(reply) => reply.result().cloned(),
CallOutcome::Refused(e) => Err(e),
CallOutcome::Expired => Err(unresolved(method, worker, self.deadlines.probe)),
}
} }
/// The sensory input one agent is permitted to consume at `boundary`. /// The sensory input one agent is permitted to consume at `boundary`.
@ -1102,8 +1218,10 @@ impl Coordinator {
} }
/// Runs one set of per-agent jobs in the configured dispatch order. /// Runs one set of per-agent jobs in the configured dispatch order.
/// A job's deadline is the section 6 probe, not the failure point: an expiry here starts
/// the resolution, which the phase loops run one at a time with the session in hand.
async fn run_jobs(&mut self, jobs: Vec<Job>, order: DispatchOrder) -> Vec<JobResult> { async fn run_jobs(&mut self, jobs: Vec<Job>, order: DispatchOrder) -> Vec<JobResult> {
let deadline = self.deadlines.call; let deadline = self.deadlines.probe;
let mut out = Vec::new(); let mut out = Vec::new();
match order { match order {
DispatchOrder::Sequential | DispatchOrder::Reversed => { DispatchOrder::Sequential | DispatchOrder::Reversed => {
@ -1113,24 +1231,27 @@ impl Coordinator {
} }
for job in jobs { for job in jobs {
let started = Instant::now(); let started = Instant::now();
let reply = call_owned( let outcome = call_owned(
self.bus.clone(), self.bus.clone(),
job.worker.clone(), job.worker.clone(),
job.method, job.method,
job.scope.clone(), job.scope.clone(),
job.params, job.params.clone(),
job.attachments, job.attachments.clone(),
job.request_id, job.request_id.clone(),
Vec::new(), Vec::new(),
deadline, deadline,
) )
.await; .await;
out.push(JobResult { out.push(JobResult {
agent_id: job.agent_id, agent_id: job.agent_id,
reply, outcome,
scope: job.scope, scope: job.scope,
worker: job.worker, worker: job.worker,
method: job.method, method: job.method,
request_id: job.request_id,
params: job.params,
attachments: job.attachments,
elapsed: started.elapsed(), elapsed: started.elapsed(),
}); });
} }
@ -1145,24 +1266,27 @@ impl Coordinator {
let method = job.method; let method = job.method;
tasks.push(tokio::spawn(async move { tasks.push(tokio::spawn(async move {
let started = Instant::now(); let started = Instant::now();
let reply = call_owned( let outcome = call_owned(
bus, bus,
job.worker, job.worker,
job.method, job.method,
job.scope, job.scope,
job.params, job.params.clone(),
job.attachments, job.attachments.clone(),
job.request_id, job.request_id.clone(),
Vec::new(), Vec::new(),
deadline, deadline,
) )
.await; .await;
JobResult { JobResult {
agent_id, agent_id,
reply, outcome,
scope, scope,
worker, worker,
method, method,
request_id: job.request_id,
params: job.params,
attachments: job.attachments,
elapsed: started.elapsed(), elapsed: started.elapsed(),
} }
})); }));
@ -1414,13 +1538,31 @@ impl Coordinator {
} }
let results = self.run_jobs(jobs, self.dispatch).await; let results = self.run_jobs(jobs, self.dispatch).await;
let mut prepared = Vec::new(); let mut prepared = Vec::new();
for JobResult { agent_id, reply, scope, worker, method, elapsed: _ } in results { for job in results {
let JobResult {
agent_id,
outcome,
scope,
worker,
method,
request_id,
params,
attachments,
elapsed: _,
} = job;
self.blame(Some(agent_id.clone())); self.blame(Some(agent_id.clone()));
let reply = match reply { let reply = match outcome {
Ok(reply) => reply, CallOutcome::Answered(reply) => *reply,
// Uncertain: the agent may be slow rather than gone. Query the same operation
// against the same incarnation before the epoch is failed. A Prepare that
// already ran answers from its own record, so no tick is repeated.
CallOutcome::Expired => {
self.resolve(&worker, method, scope.clone(), params, attachments, request_id, &[])
.await?
}
// Some agents are already Prepared. Dispatch stops and the epoch fails; a // Some agents are already Prepared. Dispatch stops and the epoch fails; a
// prepared agent is never asked to prepare again. // prepared agent is never asked to prepare again.
Err(e) => return Err(self.fail_now(e, method)), CallOutcome::Refused(e) => return Err(self.fail_now(e, method)),
}; };
self.check_reply(&worker, &reply, &scope, method)?; self.check_reply(&worker, &reply, &scope, method)?;
let decision: PreparedDecision = match reply.parse() { let decision: PreparedDecision = match reply.parse() {
@ -1600,7 +1742,7 @@ impl Coordinator {
let want = vec!["view.arena".to_owned()]; let want = vec!["view.arena".to_owned()];
self.audit.push(format!("advance:{k}")); self.audit.push(format!("advance:{k}"));
self.blame(Some(worker.worker_id.clone())); self.blame(Some(worker.worker_id.clone()));
let advance_deadline = self.deadlines.call; let advance_deadline = self.deadlines.probe;
let advance_started = Instant::now(); let advance_started = Instant::now();
let injected = self.injections.at_step == k; let injected = self.injections.at_step == k;
@ -1672,7 +1814,7 @@ impl Coordinator {
) )
.await? .await?
} else { } else {
let reply = call_owned( let outcome = call_owned(
self.bus.clone(), self.bus.clone(),
worker.clone(), worker.clone(),
"Environment.Advance", "Environment.Advance",
@ -1685,14 +1827,32 @@ impl Coordinator {
) )
.await; .await;
self.metrics.record("Environment.Advance", advance_started.elapsed()); self.metrics.record("Environment.Advance", advance_started.elapsed());
let reply = match reply { match outcome {
Ok(reply) => reply, CallOutcome::Answered(reply) => {
Err(e) => return Err(self.fail_now(e, "advance")), let reply = *reply;
}; self.check_reply(&worker, &reply, &Some(scope.clone()), "advance")?;
self.check_reply(&worker, &reply, &Some(scope.clone()), "advance")?; match reply.result() {
match reply.result() { Ok(_) => reply,
Ok(_) => reply, Err(e) => return Err(self.fail_now(e, "advance")),
Err(e) => return Err(self.fail_now(e, "advance")), }
}
// `step-v1` section 7 is imperative for this row: "Advance acknowledgment
// lost | Query/retransmit same request to same incarnation; never new batch."
// The resolution carries the original batch id and controls, so the world is
// asked about the operation it already has rather than given another one.
CallOutcome::Expired => {
self.resolve(
&worker,
"Environment.Advance",
Some(scope.clone()),
params.clone(),
Vec::new(),
request_id.clone(),
&want,
)
.await?
}
CallOutcome::Refused(e) => return Err(self.fail_now(e, "advance")),
} }
}; };
@ -1957,10 +2117,44 @@ impl Coordinator {
let mut commits = Vec::new(); let mut commits = Vec::new();
let mut first_failure = None; let mut first_failure = None;
let mut blamed: Option<Id> = None; let mut blamed: Option<Id> = None;
for JobResult { agent_id, reply, scope, worker, method, elapsed: _ } in results { for job in results {
let JobResult {
agent_id,
outcome,
scope,
worker,
method,
request_id,
params,
attachments: job_attachments,
elapsed: _,
} = job;
self.blame(Some(agent_id.clone())); self.blame(Some(agent_id.clone()));
match reply { // Uncertain: resolve the same Commit against the same incarnation first. A Commit
Ok(reply) => { // that already ran replays its cached reply, so no reward is applied twice.
let outcome = match outcome {
CallOutcome::Expired => {
match self
.resolve(
&worker,
method,
scope.clone(),
params,
job_attachments,
request_id,
&[],
)
.await
{
Ok(reply) => CallOutcome::Answered(Box::new(reply)),
Err(failure) => return Err(failure),
}
}
other => other,
};
match outcome {
CallOutcome::Answered(reply) => {
let reply = *reply;
self.check_reply(&worker, &reply, &scope, method)?; self.check_reply(&worker, &reply, &scope, method)?;
match reply.result() { match reply.result() {
Ok(_) => {} Ok(_) => {}
@ -2003,12 +2197,13 @@ impl Coordinator {
self.blame(None); self.blame(None);
commits.push((agent_id, result)); commits.push((agent_id, result));
} }
Err(e) => { CallOutcome::Refused(e) => {
if first_failure.is_none() { if first_failure.is_none() {
blamed = Some(agent_id.clone()); blamed = Some(agent_id.clone());
} }
first_failure = Some(first_failure.unwrap_or(e)); first_failure = Some(first_failure.unwrap_or(e));
} }
CallOutcome::Expired => unreachable!("an expiry was resolved just above"),
} }
} }
if let Some(error) = first_failure { if let Some(error) = first_failure {

6
services/flysim/crates/fly-session/src/environment.rs
View file

@ -36,6 +36,8 @@ pub struct EnvironmentConfig {
/// The world's fixed reduced step duration. 60 Hz is `1/60` s. /// The world's fixed reduced step duration. 60 Hz is `1/60` s.
pub step_duration: RationalNs, pub step_duration: RationalNs,
pub ports: Vec<Id>, pub ports: Vec<Id>,
/// The thread allocation the launcher started this worker within.
pub worker_threads: usize,
pub faults: EnvironmentFaults, pub faults: EnvironmentFaults,
} }
@ -425,6 +427,10 @@ impl WorkerEndpoint for CounterEnvironment {
self.status.clone() self.status.clone()
} }
fn worker_threads(&self) -> u64 {
self.config.worker_threads as u64
}
fn methods(&self) -> Vec<&'static str> { fn methods(&self) -> Vec<&'static str> {
vec!["Environment.Initialize", "Environment.Advance"] vec!["Environment.Initialize", "Environment.Advance"]
} }

15
services/flysim/crates/fly-session/src/harness.rs
View file

@ -390,18 +390,19 @@ impl SessionHarness {
id(ENV_WORKER) id(ENV_WORKER)
} }
/// The agent worker's progress counter, which is its fake model's mutation count. /// The agent worker's progress counter, which is its fake model's mutation count, when
/// this process is where that counter lives.
/// ///
/// A participant in another process keeps its counter there; use /// `None` means "not observable from here", not "nothing happened": a participant with a
/// [`SessionHarness::progress_of`], which reads it over the bus in every mode. /// process of its own keeps its counter there. [`SessionHarness::progress_of`] reads it
pub fn agent_mutations(&self, agent_id: &Id) -> u64 { /// over the bus and works in every mode.
pub fn agent_mutations(&self, agent_id: &Id) -> Option<u64> {
self.launcher self.launcher
.worker(agent_id) .worker(agent_id)
.and_then(crate::launcher::LaunchedWorker::progress_counter) .and_then(crate::launcher::LaunchedWorker::progress_counter)
.unwrap_or_default()
} }
pub fn environment_mutations(&self) -> u64 { pub fn environment_mutations(&self) -> Option<u64> {
self.agent_mutations(&id(ENV_WORKER)) self.agent_mutations(&id(ENV_WORKER))
} }
@ -420,7 +421,7 @@ impl SessionHarness {
pub async fn shutdown(self) { pub async fn shutdown(self) {
let SessionHarness { coordinator, mut launcher, observers, .. } = self; let SessionHarness { coordinator, mut launcher, observers, .. } = self;
drop(coordinator); drop(coordinator);
launcher.reap_all("shutdown").await; launcher.reap_all(&id("shutdown")).await;
for observer in observers.into_inner().expect("not poisoned") { for observer in observers.into_inner().expect("not poisoned") {
observer.close().await; observer.close().await;
} }

25
services/flysim/crates/fly-session/src/launcher.rs
View file

@ -895,6 +895,20 @@ impl Launcher {
), ),
)); ));
} }
// The 2026-09-22 `workers-v1` amendment puts the allocation on the wire,
// so the launcher checks that the worker it started agrees about what it
// was given rather than trusting the argv it sent.
if hello.worker_threads != identity.worker_threads as u64 {
return Err(DomainError::before(
ErrorCode::IdentityMismatch,
format!(
"{} reports a {}-thread allocation; the launcher gave it {}",
identity.worker_id,
hello.worker_threads,
identity.worker_threads
),
));
}
return Ok((service_incarnation, hello.incarnation_id)); return Ok((service_incarnation, hello.incarnation_id));
} }
Err(e) if handover(&e) && Instant::now() < deadline => { Err(e) if handover(&e) && Instant::now() < deadline => {
@ -969,17 +983,21 @@ impl Launcher {
/// Asks one participant to stop, then makes sure it has. /// Asks one participant to stop, then makes sure it has.
/// ///
/// The reason is an `Id` rather than a string, so a caller that got it wrong is a
/// compile-time or `parse_id` error at its own call site instead of a substitution the
/// supervisor makes silently.
///
/// `Worker.Shutdown` is the supervisor's request; the operating system is its guarantee. /// `Worker.Shutdown` is the supervisor's request; the operating system is its guarantee.
/// A participant that does not stop within the budget is terminated, which is reported as /// A participant that does not stop within the budget is terminated, which is reported as
/// such rather than as a clean stop. /// such rather than as a clean stop.
pub async fn reap(&mut self, worker_id: &Id, reason: &str) -> ReapOutcome { pub async fn reap(&mut self, worker_id: &Id, reason: &Id) -> ReapOutcome {
let Some(worker) = self.workers.get(worker_id) else { let Some(worker) = self.workers.get(worker_id) else {
return ReapOutcome::AlreadyGone; return ReapOutcome::AlreadyGone;
}; };
let service = worker.identity.service.clone(); let service = worker.identity.service.clone();
let incarnation = worker.service_incarnation.clone(); let incarnation = worker.service_incarnation.clone();
let request_id = DomainRequestId::from_serial(self.next_serial()); let request_id = DomainRequestId::from_serial(self.next_serial());
let params = ShutdownParams { reason: parse_id(reason).unwrap_or_else(|_| id("stop")) }; let params = ShutdownParams { reason: reason.clone() };
let payload = object( let payload = object(
SessionRpcRequest { SessionRpcRequest {
request_id, request_id,
@ -1025,7 +1043,7 @@ impl Launcher {
} }
/// Reaps every participant. Used at the end of a session and on every failure path. /// Reaps every participant. Used at the end of a session and on every failure path.
pub async fn reap_all(&mut self, reason: &str) -> Vec<(Id, ReapOutcome)> { pub async fn reap_all(&mut self, reason: &Id) -> Vec<(Id, ReapOutcome)> {
let mut out = Vec::new(); let mut out = Vec::new();
for worker_id in self.worker_ids() { for worker_id in self.worker_ids() {
let outcome = self.reap(&worker_id, reason).await; let outcome = self.reap(&worker_id, reason).await;
@ -1270,6 +1288,7 @@ pub(crate) fn environment_config(spec: &EnvironmentLaunch) -> EnvironmentConfig
incarnation_id: spec.incarnation_id.clone(), incarnation_id: spec.incarnation_id.clone(),
step_duration: spec.step_duration, step_duration: spec.step_duration,
ports: spec.ports.clone(), ports: spec.ports.clone(),
worker_threads: spec.worker_threads,
faults: spec.faults.clone(), faults: spec.faults.clone(),
} }
} }

181
services/flysim/crates/fly-session/src/measure.rs
View file

@ -9,11 +9,19 @@
//! question this slice has to answer is what a process boundary costs, not how fast anything //! 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 //! 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. //! as the participants answer and the samples are work rather than sleep.
//!
//! **Every row runs in a process of its own.** The coordinator's memory figure is a peak --
//! `VmHWM` never falls -- so several rows sharing one process would each report where that
//! process had already been rather than what its own mode costs, and the column would order
//! itself by row position instead of by mode. The parent spawns one `measure-row` child per
//! row and reads its result back, so each figure belongs to the row that produced it.
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::sync::Arc; use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use serde_json::{Value, json};
use crate::coordinator::DispatchOrder; use crate::coordinator::DispatchOrder;
use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via}; use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via};
use crate::launcher::ExecutionMode; use crate::launcher::ExecutionMode;
@ -108,29 +116,169 @@ pub struct Row {
pub sealed_final: usize, pub sealed_final: usize,
pub store_bytes_max: u64, pub store_bytes_max: u64,
pub store_bytes_final: u64, pub store_bytes_final: u64,
/// One sealed frame per boundary, boundary zero included. /// Frames this run actually observed, counted from the behaviour trace: every sensory
pub frames_produced: u64, /// view of every transition, plus the one the environment sealed for boundary zero.
///
/// Counted rather than calculated, so a backend that sealed two frames per boundary would
/// show up here instead of being hidden by arithmetic.
pub frames_observed: u64,
} }
impl Row { impl Row {
/// Frames the store collected: produced, minus the ones still owned at the end. /// Frames the store collected: observed, minus the ones still owned at the end.
pub fn collected(&self) -> u64 { pub fn collected(&self) -> u64 {
self.frames_produced.saturating_sub(self.sealed_final as u64) self.frames_observed.saturating_sub(self.sealed_final as u64)
}
/// The row as one JSON object, for the child that measured it to hand back.
pub fn to_json(&self) -> Value {
let p = |x: &Percentiles| {
json!({"count": x.count, "p50": x.p50_ns, "p95": x.p95_ns, "p99": x.p99_ns,
"max": x.max_ns})
};
json!({
"mode": self.mode.label(),
"agents": self.agents,
"workerThreads": self.worker_threads,
"physicalCores": self.physical_cores,
"budgetTotal": self.budget_total,
"budgetUsed": self.budget_used,
"steps": self.steps,
"prepare": p(&self.prepare),
"commit": p(&self.commit),
"advance": p(&self.advance),
"status": p(&self.status),
"step": p(&self.step),
"coordinatorPeakRssKib": self.coordinator_peak_rss_kib,
"participantsPeakRssKib": self.participants_peak_rss_kib,
"ownersMax": self.owners_max,
"ownersFinal": self.owners_final,
"artifactRootsMax": self.artifact_roots_max,
"queuedMax": self.queued_max,
"sealedMax": self.sealed_max,
"sealedFinal": self.sealed_final,
"storeBytesMax": self.store_bytes_max,
"storeBytesFinal": self.store_bytes_final,
"framesObserved": self.frames_observed,
})
}
/// Reads back what a `measure-row` child printed.
pub fn from_json(value: &Value) -> Result<Row, String> {
let u = |name: &str| -> Result<u64, String> {
value
.get(name)
.and_then(Value::as_u64)
.ok_or_else(|| format!("measure row: {name} is missing or not a number"))
};
let p = |name: &str| -> Result<Percentiles, String> {
let v = value
.get(name)
.ok_or_else(|| format!("measure row: {name} is missing"))?;
let f = |k: &str| v.get(k).and_then(Value::as_u64).unwrap_or_default();
Ok(Percentiles {
count: f("count") as usize,
p50_ns: f("p50"),
p95_ns: f("p95"),
p99_ns: f("p99"),
max_ns: f("max"),
})
};
let mode = match value.get("mode").and_then(Value::as_str) {
Some("in-process") => ExecutionMode::InProcess,
Some("thread") => ExecutionMode::Thread,
Some("process") => ExecutionMode::Process,
other => return Err(format!("measure row: unknown mode {other:?}")),
};
Ok(Row {
mode,
agents: u("agents")? as usize,
worker_threads: u("workerThreads")? as usize,
physical_cores: u("physicalCores")? as usize,
budget_total: u("budgetTotal")? as usize,
budget_used: u("budgetUsed")? as usize,
steps: u("steps")?,
prepare: p("prepare")?,
commit: p("commit")?,
advance: p("advance")?,
status: p("status")?,
step: p("step")?,
coordinator_peak_rss_kib: u("coordinatorPeakRssKib")?,
participants_peak_rss_kib: u("participantsPeakRssKib")?,
owners_max: u("ownersMax")? as usize,
owners_final: u("ownersFinal")? as usize,
artifact_roots_max: u("artifactRootsMax")?,
queued_max: u("queuedMax")? as usize,
sealed_max: u("sealedMax")? as usize,
sealed_final: u("sealedFinal")? as usize,
store_bytes_max: u("storeBytesMax")?,
store_bytes_final: u("storeBytesFinal")?,
frames_observed: u("framesObserved")?,
})
} }
} }
/// Runs the comparison. Every row is one composition in one mode. /// Runs the comparison, one child process per row.
pub async fn run(config: &MeasureConfig) -> Result<Vec<Row>, String> { ///
/// `program` is this crate's binary; each row is measured by a `measure-row` invocation of it
/// so that the row's memory peak is its own and not the accumulated peak of the rows before
/// it. The order of the rows therefore cannot change any of their numbers.
pub fn run(config: &MeasureConfig, program: &std::path::Path) -> Result<Vec<Row>, String> {
let mut rows = Vec::new(); let mut rows = Vec::new();
for mode in &config.modes { for mode in &config.modes {
for agents in &config.agent_counts { for agents in &config.agent_counts {
rows.push(one(config, *mode, *agents).await?); rows.push(row_in_a_child(config, program, *mode, *agents)?);
} }
} }
Ok(rows) Ok(rows)
} }
async fn one(config: &MeasureConfig, mode: ExecutionMode, agents: usize) -> Result<Row, String> { fn row_in_a_child(
config: &MeasureConfig,
program: &std::path::Path,
mode: ExecutionMode,
agents: usize,
) -> Result<Row, String> {
let output = std::process::Command::new(program)
.arg("measure-row")
.arg("--mode")
.arg(mode.label())
.arg("--agents")
.arg(agents.to_string())
.arg("--steps")
.arg(config.steps.to_string())
.arg("--warmup-steps")
.arg(config.warmup_steps.to_string())
.arg("--worker-threads")
.arg(config.worker_threads.to_string())
.stdin(std::process::Stdio::null())
.output()
.map_err(|e| format!("measure-row {}: {e}", mode.label()))?;
if !output.status.success() {
return Err(format!(
"measure-row {} {agents}: exited {}: {}",
mode.label(),
output.status,
String::from_utf8_lossy(&output.stderr).trim()
));
}
let text = String::from_utf8_lossy(&output.stdout);
let line = text
.lines()
.rev()
.find(|l| l.trim_start().starts_with('{'))
.ok_or_else(|| format!("measure-row {}: no row on stdout", mode.label()))?;
let value: Value = serde_json::from_str(line)
.map_err(|e| format!("measure-row {}: unreadable row: {e}", mode.label()))?;
Row::from_json(&value)
}
/// Measures exactly one row, in this process. The `measure-row` subcommand's body.
pub async fn one(
config: &MeasureConfig,
mode: ExecutionMode,
agents: usize,
) -> Result<Row, String> {
let dir = std::env::temp_dir().join(format!( let dir = std::env::temp_dir().join(format!(
"fly-session-measure-{}-{agents}-{}", "fly-session-measure-{}-{agents}-{}",
mode.label(), mode.label(),
@ -247,7 +395,15 @@ async fn measure_in(
sealed_final: harness.router_stats().sealed_artifacts, sealed_final: harness.router_stats().sealed_artifacts,
store_bytes_max, store_bytes_max,
store_bytes_final: harness.router_stats().store_bytes, store_bytes_final: harness.router_stats().store_bytes,
frames_produced: config.steps + config.warmup_steps + 1, // Counted from the behaviour trace: every sensory view of every transition this run
// recorded, plus the frame the environment sealed for boundary zero.
frames_observed: 1 + harness
.coordinator
.trace
.transitions
.iter()
.map(|t| t.behaviour.observation_boundaries.len() as u64)
.sum::<u64>(),
}; };
harness.shutdown().await; harness.shutdown().await;
Ok(row) Ok(row)
@ -261,7 +417,8 @@ pub fn table(rows: &[Row]) -> String {
); );
if let Some(first) = rows.first() { if let Some(first) = rows.first() {
out.push_str(&format!( out.push_str(&format!(
"Physical cores: {}. Coordinator reservation: 1 thread.\n\n", "Physical cores: {}. Coordinator reservation: 1 thread. Every row was measured in \
a process of its own, so no figure depends on the order of the rows.\n\n",
first.physical_cores first.physical_cores
)); ));
} }
@ -298,7 +455,7 @@ Commit p50/p95/p99 us | Advance p50/p95/p99 us | Status p50/p99 us | step p50/p9
out.push('\n'); out.push('\n');
out.push_str( out.push_str(
"| mode | agents | coordinator peak RSS KiB | participant peak RSS KiB | owners max/final | \ "| 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", roots max | queued max | sealed max/final | store bytes max/final | frames observed/collected |\n",
); );
out.push_str("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n"); out.push_str("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n");
for r in rows { for r in rows {
@ -316,7 +473,7 @@ roots max | queued max | sealed max/final | store bytes max/final | frames produ
r.sealed_final, r.sealed_final,
r.store_bytes_max, r.store_bytes_max,
r.store_bytes_final, r.store_bytes_final,
r.frames_produced, r.frames_observed,
r.collected(), r.collected(),
)); ));
} }

13
services/flysim/crates/fly-session/src/worker.rs
View file

@ -188,6 +188,12 @@ pub trait WorkerEndpoint: Send + 'static {
fn capabilities(&self) -> Vec<Id>; fn capabilities(&self) -> Vec<Id>;
fn status_cell(&self) -> StatusCell; fn status_cell(&self) -> StatusCell;
/// The thread allocation this worker's launcher started it within.
///
/// `Worker.Hello` reports it, so a caller bounded by `workers-v1`'s "within launcher
/// allocation" can read the allocation instead of being told it out of band.
fn worker_threads(&self) -> u64;
/// The domain methods this endpoint implements, beyond the common `Worker.*` set. /// The domain methods this endpoint implements, beyond the common `Worker.*` set.
/// Anything else returns UNSUPPORTED without entering the endpoint. /// Anything else returns UNSUPPORTED without entering the endpoint.
fn methods(&self) -> Vec<&'static str>; fn methods(&self) -> Vec<&'static str>;
@ -275,7 +281,7 @@ async fn run<E: WorkerEndpoint>(
) { ) {
// Identity and capabilities are fixed for the endpoint's lifetime, so the shell reads them // Identity and capabilities are fixed for the endpoint's lifetime, so the shell reads them
// once and never takes the endpoint mutex to answer Hello or Status. // once and never takes the endpoint mutex to answer Hello or Status.
let (worker_id, incarnation_id, session_id, role, capabilities, status, methods) = { let (worker_id, incarnation_id, session_id, role, capabilities, status, methods, threads) = {
let e = endpoint.lock().await; let e = endpoint.lock().await;
( (
e.worker_id(), e.worker_id(),
@ -285,6 +291,7 @@ async fn run<E: WorkerEndpoint>(
e.capabilities(), e.capabilities(),
e.status_cell(), e.status_cell(),
e.methods(), e.methods(),
e.worker_threads(),
) )
}; };
let mut running: Vec<tokio::task::JoinHandle<()>> = Vec::new(); let mut running: Vec<tokio::task::JoinHandle<()>> = Vec::new();
@ -321,6 +328,7 @@ async fn run<E: WorkerEndpoint>(
&incarnation_id, &incarnation_id,
role, role,
&capabilities, &capabilities,
threads,
); );
let _ = responder.reply(outcome.to_outcome(), &[]).await; let _ = responder.reply(outcome.to_outcome(), &[]).await;
continue; continue;
@ -622,6 +630,7 @@ fn failure(
failure_outcome(request_id, worker_id, incarnation_id, scope, error) failure_outcome(request_id, worker_id, incarnation_id, scope, error)
} }
#[allow(clippy::too_many_arguments)]
fn hello( fn hello(
request: &SessionRpcRequest, request: &SessionRpcRequest,
session_id: &Id, session_id: &Id,
@ -629,6 +638,7 @@ fn hello(
incarnation_id: &Id, incarnation_id: &Id,
role: Role, role: Role,
capabilities: &[Id], capabilities: &[Id],
worker_threads: u64,
) -> SessionRpcOutcome { ) -> SessionRpcOutcome {
let params: HelloParams = match HelloParams::from_json(&request.params) { let params: HelloParams = match HelloParams::from_json(&request.params) {
Ok(params) => params, Ok(params) => params,
@ -684,6 +694,7 @@ fn hello(
capabilities: capabilities.to_vec(), capabilities: capabilities.to_vec(),
max_agents: MAX_AGENTS as u64, max_agents: MAX_AGENTS as u64,
max_ports: MAX_PORTS as u64, max_ports: MAX_PORTS as u64,
worker_threads,
}; };
success( success(
request, request,

13
services/flysim/crates/fly-session/tests/failures.rs
View file

@ -33,6 +33,15 @@ both_transports!(
const STEPS: u64 = 4; const STEPS: u64 = 4;
const INJECT_AT: u64 = 2; const INJECT_AT: u64 = 2;
/// The mutation counter of a participant running in this process. These suites are all
/// in-process compositions, so it is always there; SESSION-02's are not, and read it over the
/// bus instead.
fn local_mutations(f: &Fixture, agent_id: &Id) -> u64 {
f.harness
.agent_mutations(agent_id)
.expect("an in-process participant keeps its counter in this process")
}
/// What a run of the standard composition produced. /// What a run of the standard composition produced.
struct Run { struct Run {
behaviour: Vec<String>, behaviour: Vec<String>,
@ -51,8 +60,8 @@ async fn run_with(via: Via, injections: Injections) -> Run {
let run = Run { let run = Run {
behaviour: f.harness.coordinator.trace.behavior(), behaviour: f.harness.coordinator.trace.behavior(),
mutations: vec![ mutations: vec![
(fly_a(), f.harness.agent_mutations(&fly_a())), (fly_a(), local_mutations(&f, &fly_a())),
(fly_b(), f.harness.agent_mutations(&fly_b())), (fly_b(), local_mutations(&f, &fly_b())),
], ],
counter: f counter: f
.harness .harness

189
services/flysim/crates/fly-session/tests/processes.rs
View file

@ -22,6 +22,7 @@ use fly_session::phase::Phase;
use fly_session::types::*; use fly_session::types::*;
all_modes!( all_modes!(
a_slow_participant_is_resolved_rather_than_failed,
a_delayed_one_agent_result_holds_the_world, a_delayed_one_agent_result_holds_the_world,
a_worker_death_has_a_bounded_diagnosed_outcome, a_worker_death_has_a_bounded_diagnosed_outcome,
a_helper_death_has_a_bounded_diagnosed_outcome, a_helper_death_has_a_bounded_diagnosed_outcome,
@ -81,6 +82,91 @@ async fn sequential_reversed_and_parallel_completion_agree() {
} }
} }
// -------------------------------------------------------------------------------------------
// ipc-v1 section 6: an uncertain call is resolved, not failed
/// A participant that is merely slow -- slower than the caller's probe, faster than the
/// resolution's budget -- finishes its step. The epoch is not lost, and the resolution adds no
/// second operation.
///
/// This is the `ipc-v1` section 6 procedure on the path that actually reaches it: the probe
/// expires, the coordinator queries the same request id against the same incarnation, the
/// worker answers `IN_PROGRESS` while its original is still running and then replays its
/// cached reply. `step-v1` section 7's Advance row is the same rule, so the world is slow here
/// too and its batch is never re-sent as a new one.
async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode) {
// A clean run of the same composition, to compare against.
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(2)).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 config = two_agents(mode);
config.agents[1].faults = AgentFaults { prepare_delay_ms: 500, ..AgentFaults::default() };
config.environment_faults =
EnvironmentFaults { advance_delay_ms: 500, ..EnvironmentFaults::default() };
let mut f = mode_fixture(mode, config).await;
// A probe well inside both delays, and a resolution budget well outside them: the point is
// a call that expires and an operation that is nevertheless fine.
f.harness.coordinator.deadlines = fly_session::Deadlines {
probe: Duration::from_millis(120),
resolve: Duration::from_secs(20),
resolve_attempts: 4096,
boot: Duration::from_secs(30),
};
within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
let reports = within("run", f.harness.coordinator.run(2))
.await
.expect("a slow participant is resolved, not failed");
assert_eq!(reports.len(), 2);
assert_eq!(f.harness.coordinator.phase(), Phase::Ready(2));
assert!(!f.harness.coordinator.is_fenced(), "a slow answer is not a lost epoch");
assert!(
f.harness.coordinator.resolutions >= 2,
"both the slow Prepare and the slow Advance must have run the resolution, not {}",
f.harness.coordinator.resolutions
);
assert!(
f.harness.coordinator.in_progress_replies > 0,
"the resolution must have met the original still running"
);
// No second operation anywhere: one advance per transition, one batch id per transition,
// and the same behaviour as the run that never timed out.
assert_eq!(f.harness.coordinator.stats().advances, 2);
let environment = f.harness.environment_id();
let world = within("progress", f.harness.progress_of(&environment)).await.unwrap();
assert_eq!(world, clean.1, "the world moved exactly as often as in the clean run");
assert_eq!(
f.harness.coordinator.trace.behavior(),
clean.0,
"resolving an uncertain call changes no behaviour"
);
let batches: std::collections::BTreeSet<Id> = f
.harness
.coordinator
.trace
.transitions
.iter()
.map(|t| t.behaviour.batch_id.clone())
.collect();
assert_eq!(batches.len(), 2, "one batch id per transition, never a second batch");
// And the agents took exactly the ticks the clean run took: a resolution is a query.
for transition in &f.harness.coordinator.trace.transitions {
for agent in &transition.behaviour.agents {
assert!(agent.ticks_advanced == 16 || agent.ticks_advanced == 17);
}
}
f.shutdown().await;
}
// ------------------------------------------------------------------------------------------- // -------------------------------------------------------------------------------------------
// Acceptance: a delayed one-agent result holds the world // Acceptance: a delayed one-agent result holds the world
@ -339,6 +425,33 @@ async fn every_participant_answers_its_supervisor(mode: ExecutionMode) {
let status = within("health", f.harness.launcher.health_check(&who)).await.unwrap(); let status = within("health", f.harness.launcher.health_check(&who)).await.unwrap();
assert_eq!(status.state, WorkerState::Ready, "{who} is healthy at a boundary"); assert_eq!(status.state, WorkerState::Ready, "{who} is healthy at a boundary");
} }
// Every participant reports the allocation its launcher gave it, which is the wire the
// 2026-09-22 `workers-v1` amendment added. The launcher refused anything else at start,
// so a caller reads it here rather than being told it out of band.
for who in [fly_a(), fly_b(), environment.clone()] {
let worker = f.harness.coordinator.agent_ref(&who).cloned().unwrap_or_else(|| {
f.harness.coordinator.environment_ref().clone()
});
let params = serde_json::json!({
"sessionId": "demo",
"expectedWorkerId": who.as_str(),
"role": if who == environment { "environment" } else { "agent" },
"supportedMajors": [1],
});
let result = within(
"hello",
f.harness.coordinator.probe_raw(&worker, "Worker.Hello", None, params),
)
.await
.expect("a worker answers its own identity");
let reported = result["limits"]["workerThreads"].as_u64();
assert_eq!(
reported,
Some(f.harness.launcher.worker(&who).unwrap().identity.worker_threads as u64),
"{who} must report the allocation its launcher gave it"
);
}
// The agents carry their configured port identities; the environment owns the ports. // The agents carry their configured port identities; the environment owns the ports.
assert_eq!( assert_eq!(
f.harness.launcher.worker(&fly_a()).unwrap().identity.port_id.as_deref(), f.harness.launcher.worker(&fly_a()).unwrap().identity.port_id.as_deref(),
@ -367,9 +480,12 @@ async fn every_participant_answers_its_supervisor(mode: ExecutionMode) {
assert_eq!(err.code, ErrorCode::IdentityMismatch); assert_eq!(err.code, ErrorCode::IdentityMismatch);
// Asking a participant to stop stops it, and the supervisor says which kind of stop it was. // 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; let outcome = f.harness.launcher.reap(&fly_a(), &id("test")).await;
assert_eq!(outcome, ReapOutcome::Stopped, "a live participant answers Worker.Shutdown"); assert_eq!(outcome, ReapOutcome::Stopped, "a live participant answers Worker.Shutdown");
assert_eq!(f.harness.launcher.reap(&fly_a(), "test").await, ReapOutcome::AlreadyGone); assert_eq!(
f.harness.launcher.reap(&fly_a(), &id("test")).await,
ReapOutcome::AlreadyGone
);
f.shutdown().await; f.shutdown().await;
} }
@ -470,6 +586,10 @@ async fn a_router_restart_during_a_world_advance_fences_the_epoch() {
}); });
within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
let boundary_before = f.harness.coordinator.observation().unwrap().boundary; let boundary_before = f.harness.coordinator.observation().unwrap().boundary;
assert!(
f.harness.coordinator.live_view_handles() > 0,
"boundary 0's view is owned before the router goes away"
);
let router = f.harness.router().clone(); let router = f.harness.router().clone();
let started = Instant::now(); let started = Instant::now();
@ -499,6 +619,11 @@ async fn a_router_restart_during_a_world_advance_fences_the_epoch() {
f.harness.coordinator.is_fenced(), f.harness.coordinator.is_fenced(),
"old handles and routes are invalid from here on" "old handles and routes are invalid from here on"
); );
assert_eq!(
f.harness.coordinator.live_view_handles(),
0,
"the fence drops every artifact handle of the old store incarnation"
);
assert_eq!(f.harness.coordinator.stats().advances, 0, "no boundary was committed"); assert_eq!(f.harness.coordinator.stats().advances, 0, "no boundary was committed");
assert_eq!(count(&f.harness.coordinator.audit, "publish:1"), 0); assert_eq!(count(&f.harness.coordinator.audit, "publish:1"), 0);
assert_eq!( assert_eq!(
@ -557,44 +682,74 @@ async fn an_old_worker_reply_after_a_restart_is_rejected_on_stale_epoch_or_incar
f.shutdown().await; f.shutdown().await;
} }
/// The other half of the same row: the replacement process is live and refuses an operation /// The other half of the same row, in two parts, because the two refusals are different
/// naming the epoch the old process belonged to, rather than applying it to a fresh brain. /// refusals and each deserves its own exact code.
///
/// A restarted worker is a *fresh* process: it has no epoch at all, so the old epoch's work is
/// refused on phase, not on timeline. The stale-epoch half of the row needs a worker that has
/// an epoch and has left it, which in process mode is the agent that did not restart.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_restarted_worker_refuses_an_operation_from_the_old_epoch() { async fn a_restarted_worker_refuses_an_operation_from_the_old_epoch() {
let mode = ExecutionMode::Process; let mode = ExecutionMode::Process;
let mut f = mode_fixture(mode, two_agents(mode)).await; let mut f = mode_fixture(mode, two_agents(mode)).await;
within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
within("step", f.harness.coordinator.step()).await.unwrap(); within("step", f.harness.coordinator.step()).await.unwrap();
let restarted = f.harness.restart_agent(&fly_b()).await.unwrap();
// Part one: a live agent process, initialized under epoch e1, meets an operation from
// another epoch. This is the row's stale-epoch half, with a real child process.
let live = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap();
let err = within(
"stale epoch",
f.harness.coordinator.probe_raw(
&live,
"Agent.Prepare",
Some(scope_at("demo", "e0", 1)),
prepare_params("fly-a"),
),
)
.await
.expect_err("an old epoch cannot mutate a worker that belongs to this one");
assert_eq!(err.code, ErrorCode::StaleEpoch);
assert_eq!(err.mutation, MutationCertainty::None, "refused before any mutation");
// Part two: the replacement process. It is a fresh worker with no epoch at all, so the
// same request is refused on phase rather than on timeline -- and, either way, nothing
// from the old epoch is applied to a fresh brain.
let restarted = f.harness.restart_agent(&fly_b()).await.unwrap();
let replacement = fly_session::rpc::WorkerRef::new( let replacement = fly_session::rpc::WorkerRef::new(
&restarted.service, &restarted.service,
&restarted.service_incarnation, &restarted.service_incarnation,
&fly_b(), &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( let err = within(
"stale epoch", "uninitialized replacement",
f.harness.coordinator.probe_raw( f.harness.coordinator.probe_raw(
&replacement, &replacement,
"Agent.Prepare", "Agent.Prepare",
Some(scope_at("demo", "e1", 1)), Some(scope_at("demo", "e1", 1)),
params, prepare_params("fly-b"),
), ),
) )
.await .await
.expect_err("an uninitialized replacement has no epoch to prepare in"); .expect_err("an uninitialized replacement has no epoch to prepare in");
assert!( assert_eq!(
matches!(err.code, ErrorCode::StaleEpoch | ErrorCode::InvalidPhase), err.code,
"a replacement refuses the old epoch's work: {err}" ErrorCode::InvalidPhase,
"a fresh process has no epoch to be stale about: {err}"
); );
assert_eq!(err.mutation, MutationCertainty::None, "nothing was applied to a fresh brain"); assert_eq!(err.mutation, MutationCertainty::None, "nothing was applied to a fresh brain");
assert_eq!(f.harness.coordinator.stats().advances, 1); assert_eq!(f.harness.coordinator.stats().advances, 1);
f.shutdown().await; f.shutdown().await;
} }
/// A well-formed `Agent.Prepare` body, for a probe whose subject is the scope rather than the
/// payload.
fn prepare_params(agent_id: &str) -> serde_json::Value {
serde_json::json!({
"agentId": agent_id,
"profileDigest": digest_of_bytes(b"whatever"),
"interval": {"numerator": "16666667", "denominator": "1"},
"decisionContextDigest": digest_of_bytes(b"whatever"),
"preStepStimulations": [],
})
}

10
services/flysim/crates/fly-session/tests/session.rs
View file

@ -35,7 +35,7 @@ const STEPS: u64 = 3;
async fn one_world_advance_per_complete_batch(via: Via) { async fn one_world_advance_per_complete_batch(via: Via) {
let mut f = default_fixture(via).await; let mut f = default_fixture(via).await;
within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
let before = f.harness.environment_mutations(); let before = f.harness.environment_mutations().expect("an in-process arena");
let reports = within("run", f.harness.coordinator.run(STEPS)).await.unwrap(); let reports = within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
assert_eq!(reports.len() as u64, STEPS); assert_eq!(reports.len() as u64, STEPS);
assert_eq!(f.harness.coordinator.stats().advances, STEPS); assert_eq!(f.harness.coordinator.stats().advances, STEPS);
@ -46,7 +46,10 @@ async fn one_world_advance_per_complete_batch(via: Via) {
"the world is at exactly one boundary per batch" "the world is at exactly one boundary per batch"
); );
// The environment's progress counter moves once per advance and not otherwise. // The environment's progress counter moves once per advance and not otherwise.
assert_eq!(f.harness.environment_mutations() - before, STEPS); assert_eq!(
f.harness.environment_mutations().expect("an in-process arena") - before,
STEPS
);
assert_eq!(count(&f.harness.coordinator.audit, "advance:0"), 1); assert_eq!(count(&f.harness.coordinator.audit, "advance:0"), 1);
assert_eq!(f.harness.coordinator.trace.transitions.len() as u64, STEPS); assert_eq!(f.harness.coordinator.trace.transitions.len() as u64, STEPS);
f.shutdown().await; f.shutdown().await;
@ -227,7 +230,8 @@ async fn bootstrap_cannot_advance_the_world_or_produce_a_reward(via: Via) {
// Warm-up did run, with learning disabled, so the models did mutate. // Warm-up did run, with learning disabled, so the models did mutate.
for agent in [fly_a(), fly_b()] { for agent in [fly_a(), fly_b()] {
assert!( assert!(
f.harness.agent_mutations(&agent) >= f.harness.config.warmup_ticks, f.harness.agent_mutations(&agent).expect("an in-process agent")
>= f.harness.config.warmup_ticks,
"warm-up ticks are real mutations" "warm-up ticks are real mutations"
); );
} }