diff --git a/docs/design/session-framework/ipc-v1.md b/docs/design/session-framework/ipc-v1.md index 083a6ce..fddd8ee 100644 --- a/docs/design/session-framework/ipc-v1.md +++ b/docs/design/session-framework/ipc-v1.md @@ -94,10 +94,14 @@ interface HelloResult { workerId: Id; incarnationId: Id; role: "agent" | "environment" | "coordinator"; buildDigest: Digest; contractDigest: Digest; 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 session authority to the expected coordinator identity/incarnation during negotiation and initialization. Wrong worker/role, no common major or missing required capability refuses diff --git a/docs/design/session-framework/workers-v1.md b/docs/design/session-framework/workers-v1.md index 45b45eb..996d5cb 100644 --- a/docs/design/session-framework/workers-v1.md +++ b/docs/design/session-framework/workers-v1.md @@ -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 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 diff --git a/packages/session-types/src/workers.ts b/packages/session-types/src/workers.ts index 1b1de22..6f93a9d 100644 --- a/packages/session-types/src/workers.ts +++ b/packages/session-types/src/workers.ts @@ -43,6 +43,8 @@ export const MAX_ACKNOWLEDGE = 16; export const MAX_ENGINE_FRAME_LEN = 64; /** Not stated by a document; this crate's choice, published in the schema set. */ 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_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), initialInput: readSensoryInput(reader.value('initialInput')), initialDecisionContext: readTypedValue(reader.value('initialDecisionContext')), - workerThreads: reader.int('workerThreads', 1, 4_096), + workerThreads: reader.int('workerThreads', 1, MAX_WORKER_THREADS), }; reader.finish(); return params; @@ -776,7 +778,12 @@ export interface HelloResult { buildDigest: Digest; contractDigest: Digest; 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 { @@ -860,6 +867,7 @@ export function readHelloResult(value: unknown): HelloResult { const limits = { maxAgents: limitsReader.int('maxAgents', 1, MAX_AGENTS), maxPorts: limitsReader.int('maxPorts', 1, MAX_PORTS), + workerThreads: limitsReader.int('workerThreads', 1, MAX_WORKER_THREADS), }; limitsReader.finish(); reader.finish(); diff --git a/services/flysim/crates/fly-session-types/fixtures/contract-digest.json b/services/flysim/crates/fly-session-types/fixtures/contract-digest.json index bae6c95..8845d36 100644 --- a/services/flysim/crates/fly-session-types/fixtures/contract-digest.json +++ b/services/flysim/crates/fly-session-types/fixtures/contract-digest.json @@ -1,9 +1,9 @@ { "description": "contractDigest is the SHA-256 of the canonical schema set in schema-set.json.", - "contractDigest": "7932aef30c4d2d16e428081affc4e0ad187987f5b361138d553e54fd7f843b50", + "contractDigest": "d8f29a49b5df05ad8f75f7f5790a3f8cde9c5ad23a685137474c649c3c9da36d", "schemaSetVersion": 1, - "schemaSetBytes": 26685, + "schemaSetBytes": 26814, "types": 53, "enums": 11, - "limits": 25 + "limits": 26 } diff --git a/services/flysim/crates/fly-session-types/fixtures/invalid.json b/services/flysim/crates/fly-session-types/fixtures/invalid.json index cb5fa10..7305201 100644 --- a/services/flysim/crates/fly-session-types/fixtures/invalid.json +++ b/services/flysim/crates/fly-session-types/fixtures/invalid.json @@ -2802,7 +2802,8 @@ ], "limits": { "maxAgents": 4, - "maxPorts": 4 + "maxPorts": 4, + "workerThreads": 1 } }, "reason": "an agent must advertise agent-step-v1" @@ -2823,7 +2824,8 @@ ], "limits": { "maxAgents": 5, - "maxPorts": 4 + "maxPorts": 4, + "workerThreads": 1 } }, "reason": "the first composition allows four agents" @@ -2844,7 +2846,8 @@ ], "limits": { "maxAgents": 4, - "maxPorts": 4 + "maxPorts": 4, + "workerThreads": 1 } }, "reason": "v1 selects major 1" @@ -3897,4 +3900,4 @@ "reason": "a commit acknowledges the transition's next boundary" } ] -} \ No newline at end of file +} diff --git a/services/flysim/crates/fly-session-types/fixtures/schema-set.json b/services/flysim/crates/fly-session-types/fixtures/schema-set.json index 2973a86..e1a175f 100644 --- a/services/flysim/crates/fly-session-types/fixtures/schema-set.json +++ b/services/flysim/crates/fly-session-types/fixtures/schema-set.json @@ -1 +1 @@ -{"contract":"fly-session-types","enums":[{"members":["f32le-interleaved"],"name":"AudioFormat","source":"state-media-v1 2"},{"members":["bipolar","unit"],"name":"AxisRange","source":"workers-v1 3"},{"members":["fixed-build","unverified"],"name":"Determinism","source":"workers-v1 3"},{"members":["terminal"],"name":"EpisodeRequestKind","source":"workers-v1 4"},{"members":["INVALID_ARGUMENT","UNSUPPORTED","IDENTITY_MISMATCH","STALE_EPOCH","STALE_STEP","FUTURE_STEP","INVALID_PHASE","CONFLICT","IN_PROGRESS","BUSY","BUFFER_INVALID","RESULT_EXPIRED","INCOMPATIBLE_STATE","BACKEND_FAILURE","INTERNAL"],"name":"ErrorCode","source":"ipc-v1 7"},{"members":["none","applied","unknown"],"name":"MutationCertainty","source":"ipc-v1 3"},{"members":["exact-checkpoint","episode-restart"],"name":"Recovery","source":"workers-v1 3"},{"members":["agent","environment","coordinator"],"name":"Role","source":"ipc-v1 4"},{"members":["lockstep-v1"],"name":"SchedulerId","source":"publishing-v1 3"},{"members":["rgba8"],"name":"ViewFormat","source":"state-media-v1 2"},{"members":["uninitialized","ready","preparing","prepared","advancing","committing","capturing","staged-restore","restoring","failed","stopping"],"name":"WorkerState","source":"ipc-v1 4"}],"limits":[{"name":"maxAcknowledge","source":"ipc-v1 5","value":16},{"name":"maxAgents","source":"ipc-v1 2","value":4},{"name":"maxAssets","source":"crate","value":64},{"name":"maxAttachments","source":"bus-v1 4","value":32},{"name":"maxAudioStreams","source":"crate","value":8},{"name":"maxAxes","source":"workers-v1 3","value":16},{"name":"maxButtons","source":"workers-v1 3","value":32},{"name":"maxCapabilities","source":"crate","value":32},{"name":"maxEngineFrameLength","source":"workers-v1 3","value":64},{"name":"maxEnvelopeBytes","source":"bus-v1 4","value":65536},{"name":"maxMessageCodePoints","source":"ipc-v1 7","value":512},{"name":"maxObservationDelaySteps","source":"state-media-v1 2","value":8},{"name":"maxPixelAspectPart","source":"state-media-v1 2","value":65535},{"name":"maxPorts","source":"ipc-v1 2","value":4},{"name":"maxRateRoles","source":"ipc-v1 2","value":64},{"name":"maxRewardsPerOperation","source":"workers-v1 1","value":64},{"name":"maxSampleFrames","source":"state-media-v1 2","value":192000},{"name":"maxSchemaVersion","source":"ipc-v1 2","value":65535},{"name":"maxSnapshotEvents","source":"crate","value":64},{"name":"maxStimuliPerOperation","source":"workers-v1 1","value":64},{"name":"maxSupportedMajors","source":"crate","value":8},{"name":"maxSupportedStimuli","source":"crate","value":64},{"name":"maxTypedValueBytes","source":"ipc-v1 2","value":32768},{"name":"maxViewDimension","source":"state-media-v1 2","value":4096},{"name":"maxViews","source":"workers-v1 1","value":8}],"scalars":{"ArtifactIdentity":"storeId, artifactId and generation of a bus ArtifactRef","BusCallId":"call-","Digest":"64 lowercase hexadecimal digits (SHA-256)","DomainRequestId":"req-","Id":"^[a-z0-9][a-z0-9._-]{0,63}$","OwnerToken":"dlv- or own-","U64":"\"0\" or [1-9][0-9]*, <= 18446744073709551615"},"types":[{"fields":[{"constraint":"1..=16, unique","kind":"array","name":"requestIds","required":true}],"name":"AcknowledgeParams","source":"ipc-v1 5"},{"fields":[{"constraint":"<= 16, unique, a subset of the request","kind":"array","name":"acknowledged","required":true}],"name":"AcknowledgeResult","source":"ipc-v1 5"},{"fields":[{"constraint":"","kind":"Id","name":"restoreToken","required":true}],"name":"ActivateRestoreParams","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"U64","name":"committedStep","required":true},{"constraint":"","kind":"Id","name":"checkpointId","required":true},{"constraint":"required from an environment, null from an agent","kind":"WorldObservation|null","name":"observation","required":false}],"name":"ActivateRestoreResult","source":"state-media-v1 5"},{"fields":[{"constraint":"unique within an epoch","kind":"Id","name":"batchId","required":true},{"constraint":"one complete batch: every declared port once, descriptor order","kind":"array","name":"controls","required":true}],"name":"AdvanceParams","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"k+1","kind":"U64","name":"committedStep","required":true},{"constraint":"","kind":"Digest","name":"decisionContextDigest","required":true},{"constraint":"","kind":"AgentTelemetry","name":"telemetry","required":true}],"name":"AgentCommitResult","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"Id","name":"portId","required":true},{"constraint":"","kind":"Digest","name":"profileDigest","required":true},{"constraint":"","kind":"Digest","name":"datasetDigest","required":true},{"constraint":"geometry mapping needs this, not neuronCount","kind":"Digest","name":"indexDigest","required":true},{"constraint":"","kind":"U64","name":"neuronCount","required":true},{"constraint":"<= 64, unique","kind":"array","name":"rateRoles","required":true},{"constraint":"unique","kind":"array","name":"supportedStimuli","required":true}],"name":"AgentDescriptor","source":"publishing-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"AssetRef","name":"profile","required":true},{"constraint":"signed 32-bit","kind":"int","name":"seed","required":true},{"constraint":"","kind":"SensoryInput","name":"initialInput","required":true},{"constraint":"","kind":"TypedValue","name":"initialDecisionContext","required":true},{"constraint":">= 1","kind":"int","name":"workerThreads","required":true}],"name":"AgentInitializeParams","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"Digest","name":"profileDigest","required":true},{"constraint":"positive","kind":"RationalNs","name":"tickDuration","required":true},{"constraint":"","kind":"U64","name":"warmupTicks","required":true},{"constraint":"\"0\"","kind":"U64","name":"committedStep","required":true},{"constraint":"","kind":"Digest","name":"decisionContextDigest","required":true},{"constraint":"","kind":"AgentTelemetry","name":"telemetry","required":true}],"name":"AgentInitializeResult","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"U64","name":"brainTicks","required":true},{"constraint":"finite and nonnegative","kind":"number","name":"populationRateHz","required":true},{"constraint":"<= 64, unique roleId, profile order, finite nonnegative hz","kind":"array<{roleId:Id,hz:number}>","name":"rates","required":true},{"constraint":"changed <= updates; signal finite","kind":"{enabled:bool,updates:U64,changed:U64,signal:number}","name":"learning","required":true}],"name":"AgentTelemetry","source":"workers-v1 1"},{"fields":[{"constraint":"","kind":"Id","name":"id","required":true},{"constraint":"","kind":"Digest","name":"digest","required":true},{"constraint":"positive","kind":"U64","name":"byteLength","required":true},{"constraint":"","kind":"Id","name":"format","required":true}],"name":"AssetRef","source":"workers-v1 1"},{"fields":[{"constraint":"","kind":"Id","name":"streamId","required":true},{"constraint":"8000..=192000","kind":"int","name":"sampleRate","required":true},{"constraint":"1..=8","kind":"int","name":"channels","required":true},{"constraint":"","kind":"AudioFormat","name":"format","required":true}],"name":"AudioDescriptor","source":"state-media-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"streamId","required":true},{"constraint":"no overlap or rewind within an epoch","kind":"U64","name":"firstSample","required":true},{"constraint":"0..=192000","kind":"int","name":"sampleFrames","required":true},{"constraint":"byteLength == sampleFrames x channels x 4, finite f32","kind":"ArtifactRef","name":"samples","required":true},{"constraint":"true on the first chunk after restore","kind":"bool","name":"discontinuity","required":true}],"name":"AudioRef","source":"state-media-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"checkpointId","required":true}],"name":"CaptureParams","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"Id","name":"checkpointId","required":true},{"constraint":"the committed boundary","kind":"U64","name":"boundary","required":true},{"constraint":"","kind":"Digest","name":"compatibilityDigest","required":true},{"constraint":"listed attachment; digest required","kind":"ArtifactRef","name":"payload","required":true}],"name":"CaptureResult","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"DomainRequestId","name":"preparedRequestId","required":true},{"constraint":"boundary == scope.step + 1","kind":"SensoryInput","name":"nextInput","required":true},{"constraint":"","kind":"TypedValue","name":"nextDecisionContext","required":true},{"constraint":"<= 64, unique eventId, order retained","kind":"array","name":"rewards","required":true},{"constraint":"<= 64, unique id, order retained","kind":"array","name":"taskStimulations","required":true}],"name":"CommitParams","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"U64","name":"descriptorRevision","required":true},{"constraint":"","kind":"Id","name":"publisherIncarnation","required":true},{"constraint":"the committed boundary","kind":"Scope","name":"scope","required":true},{"constraint":"","kind":"Id","name":"episodeId","required":true},{"constraint":"monotonic within publisherIncarnation","kind":"U64","name":"sequence","required":true},{"constraint":"","kind":"RationalNs","name":"worldTime","required":true},{"constraint":"1..=4, unique agentId, telemetry in profile role order","kind":"array","name":"agents","required":true},{"constraint":"","kind":"TypedValue","name":"progress","required":true},{"constraint":"declared attachments held through publication admission","kind":"{views:array,audio:array}","name":"media","required":true},{"constraint":"unique, task order","kind":"array","name":"eventIds","required":true}],"name":"CommittedSnapshot","source":"publishing-v1 3"},{"fields":[{"constraint":"","kind":"SchemaRef","name":"schema","required":true},{"constraint":"<= 32, unique, fixed order","kind":"array","name":"buttons","required":true},{"constraint":"<= 16, unique id, neutral inside its range","kind":"array<{id:Id,range:AxisRange,neutral:number}>","name":"axes","required":true}],"name":"ControllerSchema","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"Digest","name":"backendDigest","required":true},{"constraint":"","kind":"Digest","name":"contentDigest","required":true},{"constraint":"","kind":"Digest","name":"configurationDigest","required":true},{"constraint":"fixed, reduced, positive","kind":"RationalNs","name":"stepDuration","required":true},{"constraint":"1..=4, unique portId, fixed order","kind":"array<{portId:Id,controls:ControllerSchema}>","name":"ports","required":true},{"constraint":"","kind":"SchemaRef","name":"inspectionSchema","required":true},{"constraint":"<= 8, unique viewId","kind":"array","name":"views","required":true},{"constraint":"unique streamId","kind":"array","name":"audio","required":true},{"constraint":"","kind":"Recovery","name":"recovery","required":true},{"constraint":"","kind":"Determinism","name":"determinism","required":true}],"name":"EnvironmentDescriptor","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"AssetRef","name":"backendConfig","required":true},{"constraint":"","kind":"AssetRef","name":"taskConfig","required":true},{"constraint":"","kind":"Id","name":"episodeId","required":true},{"constraint":"1..=4, unique portId and unique agentId","kind":"array<{portId:Id,agentId:Id}>","name":"portBindings","required":true}],"name":"EnvironmentInitializeParams","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"EnvironmentDescriptor","name":"descriptor","required":true},{"constraint":"boundary 0 and worldTime 0/1","kind":"WorldObservation","name":"observation","required":true}],"name":"EnvironmentInitializeResult","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"EpisodeRequestKind","name":"kind","required":true},{"constraint":"","kind":"Id","name":"reason","required":true},{"constraint":"","kind":"TypedValue","name":"outcome","required":true}],"name":"EpisodeRequest","source":"workers-v1 4"},{"fields":[{"constraint":"","kind":"Id","name":"sessionId","required":true},{"constraint":"","kind":"Id","name":"expectedWorkerId","required":true},{"constraint":"","kind":"Role","name":"role","required":true},{"constraint":"nonempty, unique, 1..=65535","kind":"array","name":"supportedMajors","required":true}],"name":"HelloParams","source":"ipc-v1 4"},{"fields":[{"constraint":"1","kind":"const","name":"selectedMajor","required":true},{"constraint":"0","kind":"const","name":"selectedMinor","required":true},{"constraint":"","kind":"Id","name":"workerId","required":true},{"constraint":"","kind":"Id","name":"incarnationId","required":true},{"constraint":"","kind":"Role","name":"role","required":true},{"constraint":"","kind":"Digest","name":"buildDigest","required":true},{"constraint":"","kind":"Digest","name":"contractDigest","required":true},{"constraint":"unique; agent-step-v1 or world-step-v1 required for that role","kind":"array","name":"capabilities","required":true},{"constraint":"1..=4 agents and 1..=4 ports","kind":"{maxAgents:int,maxPorts:int}","name":"limits","required":true}],"name":"HelloResult","source":"ipc-v1 4"},{"fields":[{"constraint":"","kind":"Id","name":"portId","required":true},{"constraint":"every declared button, descriptor order, no extras","kind":"array<{id:Id,down:bool}>","name":"buttons","required":true},{"constraint":"every declared axis in order; bipolar [-1,1], unit [0,1], refused not clamped","kind":"array<{id:Id,value:number}>","name":"axes","required":true}],"name":"PortControl","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"Digest","name":"profileDigest","required":true},{"constraint":"positive","kind":"RationalNs","name":"interval","required":true},{"constraint":"","kind":"Digest","name":"decisionContextDigest","required":true},{"constraint":"<= 64, unique id, supplied order retained","kind":"array","name":"preStepStimulations","required":true}],"name":"PrepareParams","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"<= brainTicks","kind":"U64","name":"ticksAdvanced","required":true},{"constraint":"","kind":"U64","name":"brainTicks","required":true},{"constraint":">= 0 and < one model tick","kind":"RationalNs","name":"remainder","required":true},{"constraint":"the profile's registered intent schema","kind":"TypedValue","name":"decision","required":true}],"name":"PreparedDecision","source":"workers-v1 2"},{"fields":[{"constraint":"reduced against denominator","kind":"U64","name":"numerator","required":true},{"constraint":"positive; zero is encoded 0/1; arithmetic is checked","kind":"U64","name":"denominator","required":true}],"name":"RationalNs","source":"ipc-v1 2"},{"fields":[{"constraint":"unique within its outcome namespace","kind":"Id","name":"eventId","required":true},{"constraint":"","kind":"Id","name":"ruleId","required":true},{"constraint":"finite; positive-only profiles reject negatives","kind":"number","name":"value","required":true}],"name":"Reward","source":"workers-v1 1"},{"fields":[{"constraint":"","kind":"Id","name":"id","required":true},{"constraint":"1..=65535","kind":"int","name":"version","required":true},{"constraint":"64 lowercase hex digits","kind":"Digest","name":"digest","required":true}],"name":"SchemaRef","source":"ipc-v1 2"},{"fields":[{"constraint":"^[a-z0-9][a-z0-9._-]{0,63}$","kind":"Id","name":"sessionId","required":true},{"constraint":"^[a-z0-9][a-z0-9._-]{0,63}$","kind":"Id","name":"epoch","required":true},{"constraint":"decimal string, <= 18446744073709551615","kind":"U64","name":"step","required":true}],"name":"Scope","source":"ipc-v1 2"},{"fields":[{"constraint":"the observed environment boundary","kind":"U64","name":"boundary","required":true},{"constraint":"<= 8, unique viewId, producedStep <= boundary","kind":"array","name":"views","required":true},{"constraint":"a pixel-only profile rejects non-null","kind":"TypedValue|null","name":"structured","required":false}],"name":"SensoryInput","source":"workers-v1 1"},{"fields":[{"constraint":"","kind":"Id","name":"sessionId","required":true},{"constraint":"","kind":"U64","name":"revision","required":true},{"constraint":"","kind":"Digest","name":"compositionDigest","required":true},{"constraint":"","kind":"SchedulerId","name":"schedulerId","required":true},{"constraint":"","kind":"EnvironmentDescriptor","name":"environment","required":true},{"constraint":"","kind":"SchemaRef","name":"taskSchema","required":true},{"constraint":"1..=4, unique agentId and portId, each portId declared by the environment","kind":"array","name":"agents","required":true},{"constraint":"unique id","kind":"array","name":"assets","required":true}],"name":"SessionDescriptor","source":"publishing-v1 3"},{"fields":[{"constraint":"\"error\"","kind":"const","name":"type","required":true},{"constraint":"echoes the request","kind":"DomainRequestId","name":"requestId","required":true},{"constraint":"","kind":"Id","name":"workerId","required":true},{"constraint":"","kind":"Id","name":"incarnationId","required":true},{"constraint":"echoes the request scope","kind":"Scope|null","name":"scope","required":false},{"constraint":"","kind":"ErrorCode","name":"error.code","required":true},{"constraint":"<= 512 code points","kind":"string","name":"error.message","required":true},{"constraint":"none for every code raised before mutation","kind":"MutationCertainty","name":"error.mutation","required":true}],"name":"SessionRpcFailure","source":"ipc-v1 3"},{"fields":[{"constraint":"req-","kind":"DomainRequestId","name":"requestId","required":true},{"constraint":"null for lifecycle calls","kind":"Scope|null","name":"scope","required":false},{"constraint":"no bus callId/deliveryId/ownerId keys","kind":"object","name":"params","required":true}],"name":"SessionRpcRequest","source":"ipc-v1 2"},{"fields":[{"constraint":"\"result\"","kind":"const","name":"type","required":true},{"constraint":"echoes the request","kind":"DomainRequestId","name":"requestId","required":true},{"constraint":"","kind":"Id","name":"workerId","required":true},{"constraint":"","kind":"Id","name":"incarnationId","required":true},{"constraint":"echoes the request scope","kind":"Scope|null","name":"scope","required":false},{"constraint":"","kind":"object","name":"result","required":true}],"name":"SessionRpcSuccess","source":"ipc-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"reason","required":true}],"name":"ShutdownParams","source":"ipc-v1 7"},{"fields":[{"constraint":"true","kind":"const","name":"stopping","required":true}],"name":"ShutdownResult","source":"ipc-v1 7"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"AgentTelemetry","name":"telemetry","required":true},{"constraint":"null exactly at boundary 0","kind":"TypedValue|null","name":"selectedDecision","required":false},{"constraint":"null exactly at boundary 0; the agent's assigned port","kind":"PortControl|null","name":"appliedControls","required":false}],"name":"SnapshotAgent","source":"publishing-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"checkpointId","required":true},{"constraint":"provenance, not the new handles","kind":"Scope","name":"sourceScope","required":true},{"constraint":"","kind":"Digest","name":"compatibilityDigest","required":true},{"constraint":"newly imported; digest required","kind":"ArtifactRef","name":"payload","required":true}],"name":"StageRestoreParams","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"Id","name":"checkpointId","required":true},{"constraint":"activates once, bound to scope and payload","kind":"Id","name":"restoreToken","required":true}],"name":"StageRestoreResult","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"WorkerState","name":"state","required":true},{"constraint":"null before initialization","kind":"Scope|null","name":"currentScope","required":false},{"constraint":"","kind":"DomainRequestId|null","name":"activeRequestId","required":false},{"constraint":"","kind":"DomainRequestId|null","name":"lastCompletedRequestId","required":false},{"constraint":"","kind":"Id|null","name":"lastBatchId","required":false},{"constraint":"advances on progress, not on status queries","kind":"U64","name":"progressCounter","required":true}],"name":"StatusResult","source":"ipc-v1 4"},{"fields":[{"constraint":"echoes the request","kind":"Id","name":"batchId","required":true},{"constraint":"k","kind":"U64","name":"appliedFromStep","required":true},{"constraint":"appliedFromStep + 1","kind":"U64","name":"nextStep","required":true},{"constraint":"over validated canonical requested controls","kind":"Digest","name":"appliedControlsDigest","required":true},{"constraint":"boundary == nextStep","kind":"WorldObservation","name":"observation","required":true}],"name":"StepResult","source":"workers-v1 3"},{"fields":[{"constraint":"unique within its command namespace","kind":"Id","name":"id","required":true},{"constraint":"resolved through a profile capability","kind":"Id","name":"kindId","required":true},{"constraint":"finite and > 0","kind":"number","name":"durationMs","required":true}],"name":"Stimulus","source":"workers-v1 1"},{"fields":[{"constraint":"derived from epoch, source step, rule and ordinal","kind":"Id","name":"id","required":true},{"constraint":"","kind":"Id","name":"kindId","required":true},{"constraint":"the newly reached boundary, 0 for bootstrap","kind":"U64","name":"sourceStep","required":true},{"constraint":"","kind":"Id|null","name":"agentId","required":false},{"constraint":"","kind":"TypedValue","name":"payload","required":true}],"name":"TaskEvent","source":"workers-v1 4"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"Digest","name":"profileDigest","required":true},{"constraint":"","kind":"U64","name":"ticksAdvanced","required":true},{"constraint":"","kind":"U64","name":"brainTicks","required":true},{"constraint":"","kind":"RationalNs","name":"remainder","required":true},{"constraint":"","kind":"Digest","name":"decisionDigest","required":true},{"constraint":"the commit acknowledgment","kind":"U64","name":"committedStep","required":true}],"name":"TraceAgent","source":"step-v1 8"},{"fields":[{"constraint":"","kind":"Scope","name":"scope","required":true},{"constraint":"sorted by agentId, independent of dispatch order","kind":"array","name":"agents","required":true},{"constraint":"","kind":"Id","name":"batchId","required":true},{"constraint":"","kind":"Digest","name":"controlDigest","required":true},{"constraint":"the world boundary the environment acknowledged","kind":"U64","name":"acknowledgedBoundary","required":true},{"constraint":"sorted by viewId","kind":"array<{viewId:Id,producedStep:U64}>","name":"observationBoundaries","required":true},{"constraint":"task outcome ids in task order","kind":"array","name":"outcomeIds","required":true},{"constraint":"task event ids in task order","kind":"array","name":"eventIds","required":true},{"constraint":"","kind":"U64","name":"publishedBoundary","required":true}],"name":"TraceBehaviour","source":"step-v1 8"},{"fields":[{"constraint":"","kind":"U64","name":"wallTimeNs","required":true},{"constraint":"","kind":"array<{agentId:Id,requestId:DomainRequestId}>","name":"prepareRequestIds","required":true},{"constraint":"","kind":"DomainRequestId","name":"advanceRequestId","required":true},{"constraint":"","kind":"array<{agentId:Id,requestId:DomainRequestId}>","name":"commitRequestIds","required":true},{"constraint":"call-","kind":"array","name":"busCallIds","required":true},{"constraint":"dlv- or own-","kind":"array","name":"deliveryIds","required":true}],"name":"TraceOperational","source":"step-v1 8"},{"fields":[{"constraint":"compared between runs","kind":"TraceBehaviour","name":"behaviour","required":true},{"constraint":"recorded, never compared: wall time and transport identities","kind":"TraceOperational","name":"operational","required":true}],"name":"TransitionTrace","source":"step-v1 8"},{"fields":[{"constraint":"","kind":"SchemaRef","name":"schema","required":true},{"constraint":"canonical JSON of the whole TypedValue <= 32768 bytes","kind":"object","name":"value","required":true}],"name":"TypedValue","source":"ipc-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"viewId","required":true},{"constraint":"1..=4096","kind":"int","name":"width","required":true},{"constraint":"1..=4096","kind":"int","name":"height","required":true},{"constraint":"","kind":"ViewFormat","name":"format","required":true},{"constraint":"exactly 4 x width","kind":"int","name":"rowStride","required":true},{"constraint":"positive integers <= 65535","kind":"{numerator:int,denominator:int}","name":"pixelAspect","required":true},{"constraint":"0..=8","kind":"int","name":"observationDelaySteps","required":true}],"name":"ViewDescriptor","source":"state-media-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"viewId","required":true},{"constraint":"max(0, boundary - observationDelaySteps) for required sensory views","kind":"U64","name":"producedStep","required":true},{"constraint":"listed attachment; byteLength == rowStride x height","kind":"ArtifactRef","name":"pixels","required":true}],"name":"ViewRef","source":"state-media-v1 2"},{"fields":[{"constraint":"","kind":"U64","name":"boundary","required":true},{"constraint":"logical time since episode start","kind":"RationalNs","name":"worldTime","required":true},{"constraint":"<= 64 characters","kind":"string|null","name":"engineFrame","required":false},{"constraint":"<= 8, unique viewId","kind":"array","name":"sensoryViews","required":true},{"constraint":"the descriptor's inspectionSchema","kind":"TypedValue","name":"inspection","required":true},{"constraint":"<= 8, unique viewId","kind":"array","name":"broadcastViews","required":true},{"constraint":"unique streamId","kind":"array","name":"audio","required":true}],"name":"WorldObservation","source":"workers-v1 3"}],"version":1} +{"contract":"fly-session-types","enums":[{"members":["f32le-interleaved"],"name":"AudioFormat","source":"state-media-v1 2"},{"members":["bipolar","unit"],"name":"AxisRange","source":"workers-v1 3"},{"members":["fixed-build","unverified"],"name":"Determinism","source":"workers-v1 3"},{"members":["terminal"],"name":"EpisodeRequestKind","source":"workers-v1 4"},{"members":["INVALID_ARGUMENT","UNSUPPORTED","IDENTITY_MISMATCH","STALE_EPOCH","STALE_STEP","FUTURE_STEP","INVALID_PHASE","CONFLICT","IN_PROGRESS","BUSY","BUFFER_INVALID","RESULT_EXPIRED","INCOMPATIBLE_STATE","BACKEND_FAILURE","INTERNAL"],"name":"ErrorCode","source":"ipc-v1 7"},{"members":["none","applied","unknown"],"name":"MutationCertainty","source":"ipc-v1 3"},{"members":["exact-checkpoint","episode-restart"],"name":"Recovery","source":"workers-v1 3"},{"members":["agent","environment","coordinator"],"name":"Role","source":"ipc-v1 4"},{"members":["lockstep-v1"],"name":"SchedulerId","source":"publishing-v1 3"},{"members":["rgba8"],"name":"ViewFormat","source":"state-media-v1 2"},{"members":["uninitialized","ready","preparing","prepared","advancing","committing","capturing","staged-restore","restoring","failed","stopping"],"name":"WorkerState","source":"ipc-v1 4"}],"limits":[{"name":"maxAcknowledge","source":"ipc-v1 5","value":16},{"name":"maxAgents","source":"ipc-v1 2","value":4},{"name":"maxAssets","source":"crate","value":64},{"name":"maxAttachments","source":"bus-v1 4","value":32},{"name":"maxAudioStreams","source":"crate","value":8},{"name":"maxAxes","source":"workers-v1 3","value":16},{"name":"maxButtons","source":"workers-v1 3","value":32},{"name":"maxCapabilities","source":"crate","value":32},{"name":"maxEngineFrameLength","source":"workers-v1 3","value":64},{"name":"maxEnvelopeBytes","source":"bus-v1 4","value":65536},{"name":"maxMessageCodePoints","source":"ipc-v1 7","value":512},{"name":"maxObservationDelaySteps","source":"state-media-v1 2","value":8},{"name":"maxPixelAspectPart","source":"state-media-v1 2","value":65535},{"name":"maxPorts","source":"ipc-v1 2","value":4},{"name":"maxRateRoles","source":"ipc-v1 2","value":64},{"name":"maxRewardsPerOperation","source":"workers-v1 1","value":64},{"name":"maxSampleFrames","source":"state-media-v1 2","value":192000},{"name":"maxSchemaVersion","source":"ipc-v1 2","value":65535},{"name":"maxSnapshotEvents","source":"crate","value":64},{"name":"maxStimuliPerOperation","source":"workers-v1 1","value":64},{"name":"maxSupportedMajors","source":"crate","value":8},{"name":"maxSupportedStimuli","source":"crate","value":64},{"name":"maxTypedValueBytes","source":"ipc-v1 2","value":32768},{"name":"maxViewDimension","source":"state-media-v1 2","value":4096},{"name":"maxViews","source":"workers-v1 1","value":8},{"name":"maxWorkerThreads","source":"workers-v1 2","value":4096}],"scalars":{"ArtifactIdentity":"storeId, artifactId and generation of a bus ArtifactRef","BusCallId":"call-","Digest":"64 lowercase hexadecimal digits (SHA-256)","DomainRequestId":"req-","Id":"^[a-z0-9][a-z0-9._-]{0,63}$","OwnerToken":"dlv- or own-","U64":"\"0\" or [1-9][0-9]*, <= 18446744073709551615"},"types":[{"fields":[{"constraint":"1..=16, unique","kind":"array","name":"requestIds","required":true}],"name":"AcknowledgeParams","source":"ipc-v1 5"},{"fields":[{"constraint":"<= 16, unique, a subset of the request","kind":"array","name":"acknowledged","required":true}],"name":"AcknowledgeResult","source":"ipc-v1 5"},{"fields":[{"constraint":"","kind":"Id","name":"restoreToken","required":true}],"name":"ActivateRestoreParams","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"U64","name":"committedStep","required":true},{"constraint":"","kind":"Id","name":"checkpointId","required":true},{"constraint":"required from an environment, null from an agent","kind":"WorldObservation|null","name":"observation","required":false}],"name":"ActivateRestoreResult","source":"state-media-v1 5"},{"fields":[{"constraint":"unique within an epoch","kind":"Id","name":"batchId","required":true},{"constraint":"one complete batch: every declared port once, descriptor order","kind":"array","name":"controls","required":true}],"name":"AdvanceParams","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"k+1","kind":"U64","name":"committedStep","required":true},{"constraint":"","kind":"Digest","name":"decisionContextDigest","required":true},{"constraint":"","kind":"AgentTelemetry","name":"telemetry","required":true}],"name":"AgentCommitResult","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"Id","name":"portId","required":true},{"constraint":"","kind":"Digest","name":"profileDigest","required":true},{"constraint":"","kind":"Digest","name":"datasetDigest","required":true},{"constraint":"geometry mapping needs this, not neuronCount","kind":"Digest","name":"indexDigest","required":true},{"constraint":"","kind":"U64","name":"neuronCount","required":true},{"constraint":"<= 64, unique","kind":"array","name":"rateRoles","required":true},{"constraint":"unique","kind":"array","name":"supportedStimuli","required":true}],"name":"AgentDescriptor","source":"publishing-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"AssetRef","name":"profile","required":true},{"constraint":"signed 32-bit","kind":"int","name":"seed","required":true},{"constraint":"","kind":"SensoryInput","name":"initialInput","required":true},{"constraint":"","kind":"TypedValue","name":"initialDecisionContext","required":true},{"constraint":">= 1","kind":"int","name":"workerThreads","required":true}],"name":"AgentInitializeParams","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"Digest","name":"profileDigest","required":true},{"constraint":"positive","kind":"RationalNs","name":"tickDuration","required":true},{"constraint":"","kind":"U64","name":"warmupTicks","required":true},{"constraint":"\"0\"","kind":"U64","name":"committedStep","required":true},{"constraint":"","kind":"Digest","name":"decisionContextDigest","required":true},{"constraint":"","kind":"AgentTelemetry","name":"telemetry","required":true}],"name":"AgentInitializeResult","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"U64","name":"brainTicks","required":true},{"constraint":"finite and nonnegative","kind":"number","name":"populationRateHz","required":true},{"constraint":"<= 64, unique roleId, profile order, finite nonnegative hz","kind":"array<{roleId:Id,hz:number}>","name":"rates","required":true},{"constraint":"changed <= updates; signal finite","kind":"{enabled:bool,updates:U64,changed:U64,signal:number}","name":"learning","required":true}],"name":"AgentTelemetry","source":"workers-v1 1"},{"fields":[{"constraint":"","kind":"Id","name":"id","required":true},{"constraint":"","kind":"Digest","name":"digest","required":true},{"constraint":"positive","kind":"U64","name":"byteLength","required":true},{"constraint":"","kind":"Id","name":"format","required":true}],"name":"AssetRef","source":"workers-v1 1"},{"fields":[{"constraint":"","kind":"Id","name":"streamId","required":true},{"constraint":"8000..=192000","kind":"int","name":"sampleRate","required":true},{"constraint":"1..=8","kind":"int","name":"channels","required":true},{"constraint":"","kind":"AudioFormat","name":"format","required":true}],"name":"AudioDescriptor","source":"state-media-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"streamId","required":true},{"constraint":"no overlap or rewind within an epoch","kind":"U64","name":"firstSample","required":true},{"constraint":"0..=192000","kind":"int","name":"sampleFrames","required":true},{"constraint":"byteLength == sampleFrames x channels x 4, finite f32","kind":"ArtifactRef","name":"samples","required":true},{"constraint":"true on the first chunk after restore","kind":"bool","name":"discontinuity","required":true}],"name":"AudioRef","source":"state-media-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"checkpointId","required":true}],"name":"CaptureParams","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"Id","name":"checkpointId","required":true},{"constraint":"the committed boundary","kind":"U64","name":"boundary","required":true},{"constraint":"","kind":"Digest","name":"compatibilityDigest","required":true},{"constraint":"listed attachment; digest required","kind":"ArtifactRef","name":"payload","required":true}],"name":"CaptureResult","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"DomainRequestId","name":"preparedRequestId","required":true},{"constraint":"boundary == scope.step + 1","kind":"SensoryInput","name":"nextInput","required":true},{"constraint":"","kind":"TypedValue","name":"nextDecisionContext","required":true},{"constraint":"<= 64, unique eventId, order retained","kind":"array","name":"rewards","required":true},{"constraint":"<= 64, unique id, order retained","kind":"array","name":"taskStimulations","required":true}],"name":"CommitParams","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"U64","name":"descriptorRevision","required":true},{"constraint":"","kind":"Id","name":"publisherIncarnation","required":true},{"constraint":"the committed boundary","kind":"Scope","name":"scope","required":true},{"constraint":"","kind":"Id","name":"episodeId","required":true},{"constraint":"monotonic within publisherIncarnation","kind":"U64","name":"sequence","required":true},{"constraint":"","kind":"RationalNs","name":"worldTime","required":true},{"constraint":"1..=4, unique agentId, telemetry in profile role order","kind":"array","name":"agents","required":true},{"constraint":"","kind":"TypedValue","name":"progress","required":true},{"constraint":"declared attachments held through publication admission","kind":"{views:array,audio:array}","name":"media","required":true},{"constraint":"unique, task order","kind":"array","name":"eventIds","required":true}],"name":"CommittedSnapshot","source":"publishing-v1 3"},{"fields":[{"constraint":"","kind":"SchemaRef","name":"schema","required":true},{"constraint":"<= 32, unique, fixed order","kind":"array","name":"buttons","required":true},{"constraint":"<= 16, unique id, neutral inside its range","kind":"array<{id:Id,range:AxisRange,neutral:number}>","name":"axes","required":true}],"name":"ControllerSchema","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"Digest","name":"backendDigest","required":true},{"constraint":"","kind":"Digest","name":"contentDigest","required":true},{"constraint":"","kind":"Digest","name":"configurationDigest","required":true},{"constraint":"fixed, reduced, positive","kind":"RationalNs","name":"stepDuration","required":true},{"constraint":"1..=4, unique portId, fixed order","kind":"array<{portId:Id,controls:ControllerSchema}>","name":"ports","required":true},{"constraint":"","kind":"SchemaRef","name":"inspectionSchema","required":true},{"constraint":"<= 8, unique viewId","kind":"array","name":"views","required":true},{"constraint":"unique streamId","kind":"array","name":"audio","required":true},{"constraint":"","kind":"Recovery","name":"recovery","required":true},{"constraint":"","kind":"Determinism","name":"determinism","required":true}],"name":"EnvironmentDescriptor","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"AssetRef","name":"backendConfig","required":true},{"constraint":"","kind":"AssetRef","name":"taskConfig","required":true},{"constraint":"","kind":"Id","name":"episodeId","required":true},{"constraint":"1..=4, unique portId and unique agentId","kind":"array<{portId:Id,agentId:Id}>","name":"portBindings","required":true}],"name":"EnvironmentInitializeParams","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"EnvironmentDescriptor","name":"descriptor","required":true},{"constraint":"boundary 0 and worldTime 0/1","kind":"WorldObservation","name":"observation","required":true}],"name":"EnvironmentInitializeResult","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"EpisodeRequestKind","name":"kind","required":true},{"constraint":"","kind":"Id","name":"reason","required":true},{"constraint":"","kind":"TypedValue","name":"outcome","required":true}],"name":"EpisodeRequest","source":"workers-v1 4"},{"fields":[{"constraint":"","kind":"Id","name":"sessionId","required":true},{"constraint":"","kind":"Id","name":"expectedWorkerId","required":true},{"constraint":"","kind":"Role","name":"role","required":true},{"constraint":"nonempty, unique, 1..=65535","kind":"array","name":"supportedMajors","required":true}],"name":"HelloParams","source":"ipc-v1 4"},{"fields":[{"constraint":"1","kind":"const","name":"selectedMajor","required":true},{"constraint":"0","kind":"const","name":"selectedMinor","required":true},{"constraint":"","kind":"Id","name":"workerId","required":true},{"constraint":"","kind":"Id","name":"incarnationId","required":true},{"constraint":"","kind":"Role","name":"role","required":true},{"constraint":"","kind":"Digest","name":"buildDigest","required":true},{"constraint":"","kind":"Digest","name":"contractDigest","required":true},{"constraint":"unique; agent-step-v1 or world-step-v1 required for that role","kind":"array","name":"capabilities","required":true},{"constraint":"1..=4 agents, 1..=4 ports, and the launcher allocation this worker runs in","kind":"{maxAgents:int,maxPorts:int,workerThreads:int}","name":"limits","required":true}],"name":"HelloResult","source":"ipc-v1 4"},{"fields":[{"constraint":"","kind":"Id","name":"portId","required":true},{"constraint":"every declared button, descriptor order, no extras","kind":"array<{id:Id,down:bool}>","name":"buttons","required":true},{"constraint":"every declared axis in order; bipolar [-1,1], unit [0,1], refused not clamped","kind":"array<{id:Id,value:number}>","name":"axes","required":true}],"name":"PortControl","source":"workers-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"Digest","name":"profileDigest","required":true},{"constraint":"positive","kind":"RationalNs","name":"interval","required":true},{"constraint":"","kind":"Digest","name":"decisionContextDigest","required":true},{"constraint":"<= 64, unique id, supplied order retained","kind":"array","name":"preStepStimulations","required":true}],"name":"PrepareParams","source":"workers-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"<= brainTicks","kind":"U64","name":"ticksAdvanced","required":true},{"constraint":"","kind":"U64","name":"brainTicks","required":true},{"constraint":">= 0 and < one model tick","kind":"RationalNs","name":"remainder","required":true},{"constraint":"the profile's registered intent schema","kind":"TypedValue","name":"decision","required":true}],"name":"PreparedDecision","source":"workers-v1 2"},{"fields":[{"constraint":"reduced against denominator","kind":"U64","name":"numerator","required":true},{"constraint":"positive; zero is encoded 0/1; arithmetic is checked","kind":"U64","name":"denominator","required":true}],"name":"RationalNs","source":"ipc-v1 2"},{"fields":[{"constraint":"unique within its outcome namespace","kind":"Id","name":"eventId","required":true},{"constraint":"","kind":"Id","name":"ruleId","required":true},{"constraint":"finite; positive-only profiles reject negatives","kind":"number","name":"value","required":true}],"name":"Reward","source":"workers-v1 1"},{"fields":[{"constraint":"","kind":"Id","name":"id","required":true},{"constraint":"1..=65535","kind":"int","name":"version","required":true},{"constraint":"64 lowercase hex digits","kind":"Digest","name":"digest","required":true}],"name":"SchemaRef","source":"ipc-v1 2"},{"fields":[{"constraint":"^[a-z0-9][a-z0-9._-]{0,63}$","kind":"Id","name":"sessionId","required":true},{"constraint":"^[a-z0-9][a-z0-9._-]{0,63}$","kind":"Id","name":"epoch","required":true},{"constraint":"decimal string, <= 18446744073709551615","kind":"U64","name":"step","required":true}],"name":"Scope","source":"ipc-v1 2"},{"fields":[{"constraint":"the observed environment boundary","kind":"U64","name":"boundary","required":true},{"constraint":"<= 8, unique viewId, producedStep <= boundary","kind":"array","name":"views","required":true},{"constraint":"a pixel-only profile rejects non-null","kind":"TypedValue|null","name":"structured","required":false}],"name":"SensoryInput","source":"workers-v1 1"},{"fields":[{"constraint":"","kind":"Id","name":"sessionId","required":true},{"constraint":"","kind":"U64","name":"revision","required":true},{"constraint":"","kind":"Digest","name":"compositionDigest","required":true},{"constraint":"","kind":"SchedulerId","name":"schedulerId","required":true},{"constraint":"","kind":"EnvironmentDescriptor","name":"environment","required":true},{"constraint":"","kind":"SchemaRef","name":"taskSchema","required":true},{"constraint":"1..=4, unique agentId and portId, each portId declared by the environment","kind":"array","name":"agents","required":true},{"constraint":"unique id","kind":"array","name":"assets","required":true}],"name":"SessionDescriptor","source":"publishing-v1 3"},{"fields":[{"constraint":"\"error\"","kind":"const","name":"type","required":true},{"constraint":"echoes the request","kind":"DomainRequestId","name":"requestId","required":true},{"constraint":"","kind":"Id","name":"workerId","required":true},{"constraint":"","kind":"Id","name":"incarnationId","required":true},{"constraint":"echoes the request scope","kind":"Scope|null","name":"scope","required":false},{"constraint":"","kind":"ErrorCode","name":"error.code","required":true},{"constraint":"<= 512 code points","kind":"string","name":"error.message","required":true},{"constraint":"none for every code raised before mutation","kind":"MutationCertainty","name":"error.mutation","required":true}],"name":"SessionRpcFailure","source":"ipc-v1 3"},{"fields":[{"constraint":"req-","kind":"DomainRequestId","name":"requestId","required":true},{"constraint":"null for lifecycle calls","kind":"Scope|null","name":"scope","required":false},{"constraint":"no bus callId/deliveryId/ownerId keys","kind":"object","name":"params","required":true}],"name":"SessionRpcRequest","source":"ipc-v1 2"},{"fields":[{"constraint":"\"result\"","kind":"const","name":"type","required":true},{"constraint":"echoes the request","kind":"DomainRequestId","name":"requestId","required":true},{"constraint":"","kind":"Id","name":"workerId","required":true},{"constraint":"","kind":"Id","name":"incarnationId","required":true},{"constraint":"echoes the request scope","kind":"Scope|null","name":"scope","required":false},{"constraint":"","kind":"object","name":"result","required":true}],"name":"SessionRpcSuccess","source":"ipc-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"reason","required":true}],"name":"ShutdownParams","source":"ipc-v1 7"},{"fields":[{"constraint":"true","kind":"const","name":"stopping","required":true}],"name":"ShutdownResult","source":"ipc-v1 7"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"AgentTelemetry","name":"telemetry","required":true},{"constraint":"null exactly at boundary 0","kind":"TypedValue|null","name":"selectedDecision","required":false},{"constraint":"null exactly at boundary 0; the agent's assigned port","kind":"PortControl|null","name":"appliedControls","required":false}],"name":"SnapshotAgent","source":"publishing-v1 3"},{"fields":[{"constraint":"","kind":"Id","name":"checkpointId","required":true},{"constraint":"provenance, not the new handles","kind":"Scope","name":"sourceScope","required":true},{"constraint":"","kind":"Digest","name":"compatibilityDigest","required":true},{"constraint":"newly imported; digest required","kind":"ArtifactRef","name":"payload","required":true}],"name":"StageRestoreParams","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"Id","name":"checkpointId","required":true},{"constraint":"activates once, bound to scope and payload","kind":"Id","name":"restoreToken","required":true}],"name":"StageRestoreResult","source":"state-media-v1 5"},{"fields":[{"constraint":"","kind":"WorkerState","name":"state","required":true},{"constraint":"null before initialization","kind":"Scope|null","name":"currentScope","required":false},{"constraint":"","kind":"DomainRequestId|null","name":"activeRequestId","required":false},{"constraint":"","kind":"DomainRequestId|null","name":"lastCompletedRequestId","required":false},{"constraint":"","kind":"Id|null","name":"lastBatchId","required":false},{"constraint":"advances on progress, not on status queries","kind":"U64","name":"progressCounter","required":true}],"name":"StatusResult","source":"ipc-v1 4"},{"fields":[{"constraint":"echoes the request","kind":"Id","name":"batchId","required":true},{"constraint":"k","kind":"U64","name":"appliedFromStep","required":true},{"constraint":"appliedFromStep + 1","kind":"U64","name":"nextStep","required":true},{"constraint":"over validated canonical requested controls","kind":"Digest","name":"appliedControlsDigest","required":true},{"constraint":"boundary == nextStep","kind":"WorldObservation","name":"observation","required":true}],"name":"StepResult","source":"workers-v1 3"},{"fields":[{"constraint":"unique within its command namespace","kind":"Id","name":"id","required":true},{"constraint":"resolved through a profile capability","kind":"Id","name":"kindId","required":true},{"constraint":"finite and > 0","kind":"number","name":"durationMs","required":true}],"name":"Stimulus","source":"workers-v1 1"},{"fields":[{"constraint":"derived from epoch, source step, rule and ordinal","kind":"Id","name":"id","required":true},{"constraint":"","kind":"Id","name":"kindId","required":true},{"constraint":"the newly reached boundary, 0 for bootstrap","kind":"U64","name":"sourceStep","required":true},{"constraint":"","kind":"Id|null","name":"agentId","required":false},{"constraint":"","kind":"TypedValue","name":"payload","required":true}],"name":"TaskEvent","source":"workers-v1 4"},{"fields":[{"constraint":"","kind":"Id","name":"agentId","required":true},{"constraint":"","kind":"Digest","name":"profileDigest","required":true},{"constraint":"","kind":"U64","name":"ticksAdvanced","required":true},{"constraint":"","kind":"U64","name":"brainTicks","required":true},{"constraint":"","kind":"RationalNs","name":"remainder","required":true},{"constraint":"","kind":"Digest","name":"decisionDigest","required":true},{"constraint":"the commit acknowledgment","kind":"U64","name":"committedStep","required":true}],"name":"TraceAgent","source":"step-v1 8"},{"fields":[{"constraint":"","kind":"Scope","name":"scope","required":true},{"constraint":"sorted by agentId, independent of dispatch order","kind":"array","name":"agents","required":true},{"constraint":"","kind":"Id","name":"batchId","required":true},{"constraint":"","kind":"Digest","name":"controlDigest","required":true},{"constraint":"the world boundary the environment acknowledged","kind":"U64","name":"acknowledgedBoundary","required":true},{"constraint":"sorted by viewId","kind":"array<{viewId:Id,producedStep:U64}>","name":"observationBoundaries","required":true},{"constraint":"task outcome ids in task order","kind":"array","name":"outcomeIds","required":true},{"constraint":"task event ids in task order","kind":"array","name":"eventIds","required":true},{"constraint":"","kind":"U64","name":"publishedBoundary","required":true}],"name":"TraceBehaviour","source":"step-v1 8"},{"fields":[{"constraint":"","kind":"U64","name":"wallTimeNs","required":true},{"constraint":"","kind":"array<{agentId:Id,requestId:DomainRequestId}>","name":"prepareRequestIds","required":true},{"constraint":"","kind":"DomainRequestId","name":"advanceRequestId","required":true},{"constraint":"","kind":"array<{agentId:Id,requestId:DomainRequestId}>","name":"commitRequestIds","required":true},{"constraint":"call-","kind":"array","name":"busCallIds","required":true},{"constraint":"dlv- or own-","kind":"array","name":"deliveryIds","required":true}],"name":"TraceOperational","source":"step-v1 8"},{"fields":[{"constraint":"compared between runs","kind":"TraceBehaviour","name":"behaviour","required":true},{"constraint":"recorded, never compared: wall time and transport identities","kind":"TraceOperational","name":"operational","required":true}],"name":"TransitionTrace","source":"step-v1 8"},{"fields":[{"constraint":"","kind":"SchemaRef","name":"schema","required":true},{"constraint":"canonical JSON of the whole TypedValue <= 32768 bytes","kind":"object","name":"value","required":true}],"name":"TypedValue","source":"ipc-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"viewId","required":true},{"constraint":"1..=4096","kind":"int","name":"width","required":true},{"constraint":"1..=4096","kind":"int","name":"height","required":true},{"constraint":"","kind":"ViewFormat","name":"format","required":true},{"constraint":"exactly 4 x width","kind":"int","name":"rowStride","required":true},{"constraint":"positive integers <= 65535","kind":"{numerator:int,denominator:int}","name":"pixelAspect","required":true},{"constraint":"0..=8","kind":"int","name":"observationDelaySteps","required":true}],"name":"ViewDescriptor","source":"state-media-v1 2"},{"fields":[{"constraint":"","kind":"Id","name":"viewId","required":true},{"constraint":"max(0, boundary - observationDelaySteps) for required sensory views","kind":"U64","name":"producedStep","required":true},{"constraint":"listed attachment; byteLength == rowStride x height","kind":"ArtifactRef","name":"pixels","required":true}],"name":"ViewRef","source":"state-media-v1 2"},{"fields":[{"constraint":"","kind":"U64","name":"boundary","required":true},{"constraint":"logical time since episode start","kind":"RationalNs","name":"worldTime","required":true},{"constraint":"<= 64 characters","kind":"string|null","name":"engineFrame","required":false},{"constraint":"<= 8, unique viewId","kind":"array","name":"sensoryViews","required":true},{"constraint":"the descriptor's inspectionSchema","kind":"TypedValue","name":"inspection","required":true},{"constraint":"<= 8, unique viewId","kind":"array","name":"broadcastViews","required":true},{"constraint":"unique streamId","kind":"array","name":"audio","required":true}],"name":"WorldObservation","source":"workers-v1 3"}],"version":1} diff --git a/services/flysim/crates/fly-session-types/fixtures/valid.json b/services/flysim/crates/fly-session-types/fixtures/valid.json index c316f95..dea3a64 100644 --- a/services/flysim/crates/fly-session-types/fixtures/valid.json +++ b/services/flysim/crates/fly-session-types/fixtures/valid.json @@ -1758,12 +1758,13 @@ ], "limits": { "maxAgents": 4, - "maxPorts": 4 + "maxPorts": 4, + "workerThreads": 1 } }, "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\"}", - "digest": "9f8cbc9dfe7a7532c2e41b53c4ce001973ca95703a81aad30ee2b1c8b640bc13" + "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": "262ea112e24dcdbdc1a5050b6dd6d031eace6a6a4cdca814913ac89fc9e39666" }, { "name": "hello result for an environment", @@ -1782,12 +1783,13 @@ ], "limits": { "maxAgents": 1, - "maxPorts": 1 + "maxPorts": 1, + "workerThreads": 1 } }, "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\"}", - "digest": "0dc33a6cbf6bf66c4d28dc41b17b2eb7918b55c65469f4161e009f662f11bca2" + "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": "06ced6fff0d810005ab2598cc9fb72a138303daef0787232871c9afa43b0eac0" }, { "name": "status result before initialization", diff --git a/services/flysim/crates/fly-session-types/src/schema.rs b/services/flysim/crates/fly-session-types/src/schema.rs index 9a122c0..6317620 100644 --- a/services/flysim/crates/fly-session-types/src/schema.rs +++ b/services/flysim/crates/fly-session-types/src/schema.rs @@ -239,6 +239,11 @@ pub const LIMITS: &[LimitSchema] = &[ value: crate::workers::MAX_CAPABILITIES as u64, source: "crate", }, + LimitSchema { + name: "maxWorkerThreads", + value: crate::workers::MAX_WORKER_THREADS, + source: "workers-v1 2", + }, LimitSchema { name: "maxSupportedMajors", value: crate::workers::MAX_SUPPORTED_MAJORS as u64, @@ -651,8 +656,8 @@ pub const SCHEMAS: &[TypeSchema] = &[ ), req( "limits", - "{maxAgents:int,maxPorts:int}", - "1..=4 agents and 1..=4 ports", + "{maxAgents:int,maxPorts:int,workerThreads:int}", + "1..=4 agents, 1..=4 ports, and the launcher allocation this worker runs in", ), ], }, diff --git a/services/flysim/crates/fly-session-types/src/workers.rs b/services/flysim/crates/fly-session-types/src/workers.rs index a62cdb7..7ff4d07 100644 --- a/services/flysim/crates/fly-session-types/src/workers.rs +++ b/services/flysim/crates/fly-session-types/src/workers.rs @@ -33,6 +33,9 @@ pub const MAX_ACKNOWLEDGE: usize = 16; pub const MAX_ENGINE_FRAME_LEN: usize = 64; /// Negotiated capability ids. Not a stated bound; recorded in the schema set. 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. pub const MAX_SUPPORTED_MAJORS: usize = 8; /// 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 initial_input = SensoryInput::from_json(f.value("initialInput")?)?; 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()?; let p = AgentInitializeParams { agent_id, @@ -1958,6 +1961,13 @@ pub struct HelloResult { pub capabilities: Vec, pub max_agents: 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 { @@ -1990,13 +2000,14 @@ impl DomainType for HelloResult { let build_digest = f.string("buildDigest")?.to_owned(); let contract_digest = f.string("contractDigest")?.to_owned(); 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 mut l = Fields::new(v, "HelloResult.limits")?; let max_agents = l.int("maxAgents", 1, MAX_AGENTS 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()?; - (max_agents, max_ports) + (max_agents, max_ports, worker_threads) }; f.finish()?; let r = HelloResult { @@ -2008,6 +2019,7 @@ impl DomainType for HelloResult { capabilities, max_agents, max_ports, + worker_threads, }; r.validate()?; Ok(r) @@ -2031,6 +2043,7 @@ impl DomainType for HelloResult { obj(vec![ ("maxAgents", Value::from(self.max_agents)), ("maxPorts", Value::from(self.max_ports)), + ("workerThreads", Value::from(self.worker_threads)), ]), ), ]) diff --git a/services/flysim/crates/fly-session/README.md b/services/flysim/crates/fly-session/README.md index bc98c78..f5de173 100644 --- a/services/flysim/crates/fly-session/README.md +++ b/services/flysim/crates/fly-session/README.md @@ -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 as `BUSY` before anything starts. `Agent.Initialize` carries exactly the allocation the launcher handed out, and an agent refuses an Initialize asking for more than its own, which - is what `workers-v1` means by "within launcher allocation". + 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 are launcher configuration. The launcher says `Worker.Hello` with the identity it configured - and refuses anything that answers as another worker, role or incarnation -- before the - coordinator has pinned a registration. The registration the coordinator pins is the one that - hello returned, never one that was assumed. + and refuses anything that answers as another worker, role, incarnation or thread allocation + -- before the coordinator has pinned a registration. The registration the coordinator pins + is the one that hello returned, never one that was assumed. - **Health.** `Worker.Status` on the supervisor's own monotonic clock, with the `ipc-v1` section 6 prototype budgets: probe at two seconds, fail at ten, a separate budget for boot. A status answer never waits for a mutation, so a busy participant is still a healthy one. @@ -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 stops moving, the artifact handles are dropped, and no further transition or publication is allowed. Lifting the fence is a coherent group restore, which is STATE-01's. -- **A bounded diagnosed outcome.** A caller-side deadline on every domain call, on the - coordinator's own clock, so a participant that dies or stops answering produces a typed - failure naming it rather than a hang. An expired deadline is `unknown`, never `none`: a - caller-side timeout is not evidence that nothing was mutated. +- **The `ipc-v1` section 6 procedure, on the path that reaches it.** A call that goes without + a terminal reply for the probe budget is *uncertain*, not failed. The coordinator then + queries the same operation -- a fresh bus call carrying the original domain request id and + 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 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 @@ -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 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: -- A process boundary costs little at the median and shows up in the tail. Two agents: the - critical path was about 7.8 ms p50 in-process, 8.6 ms on threads and 12.7 ms across - processes, while p99 went 12.4 / 12.7 / 26.0 ms. The medians are within a small multiple of - each other; the tails are where a scheduler with more runnable threads than cores appears. +- A process boundary costs about a fifth of the critical path at the median. Two agents: 10.0 + ms p50 in-process, 12.2 ms on threads, 12.1 ms across processes, with p99 at 21.4 / 19.4 / + 21.4 ms. The dedicated-thread and separate-process variants are within noise of each other, + 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 - together. That is the budget being honest, not a property of the process split. -- Memory is the clearest difference: one coordinator at about 14 MiB peak RSS plus roughly - 5.6 MiB per participant process, against a single 11 MiB process for the threaded variant. -- Ownership and queues stayed bounded in every mode and at every agent count: at most 15 live - owners, 11 artifact roots and one queue entry per agent, with the store holding two sealed - frames and 128 bytes at rest. Of 311 frames produced, 309 were collected -- the current and - previous boundary are the two that are still owned. + together. That is the budget being honest about oversubscription, not a property of the + process split. +- Memory is where the split really shows, but not where the first version of this note said. + The coordinator's own peak is roughly the same in all three modes and is *lowest* in process + mode -- 8.9 / 9.2 / 7.9 MiB at one agent -- because the workers are no longer inside it. + What the split costs is the children: about 5.7 MiB per participant process, so the whole + 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 @@ -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 ``` -Every integration test runs over both transports, through the same router code: all but one -are generated twice by `both_transports!`, and -`sequential_concurrent_and_reversed_orders_agree` walks both transports inside one test -because it compares their behaviour traces against each other. +The three integration suites do not all run over both transports, and cannot: + +- `tests/session.rs` and `tests/failures.rs` are in-process compositions and run over both, + 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 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 trace. - `tests/processes.rs`: the SESSION-02 acceptance bullets, each generated once per execution - mode -- a delayed one-agent result holding the world, a worker or helper death with a + 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 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 diff --git a/services/flysim/crates/fly-session/examples/processes.rs b/services/flysim/crates/fly-session/examples/processes.rs index 54b7721..0391f0b 100644 --- a/services/flysim/crates/fly-session/examples/processes.rs +++ b/services/flysim/crates/fly-session/examples/processes.rs @@ -63,7 +63,7 @@ async fn main() -> Result<(), Box> { 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 { println!(" reaped {worker_id}: {outcome:?}"); } diff --git a/services/flysim/crates/fly-session/src/agent.rs b/services/flysim/crates/fly-session/src/agent.rs index 7ccdfe9..187e3b1 100644 --- a/services/flysim/crates/fly-session/src/agent.rs +++ b/services/flysim/crates/fly-session/src/agent.rs @@ -650,6 +650,10 @@ impl WorkerEndpoint for FakeAgentWorker { self.status.clone() } + fn worker_threads(&self) -> u64 { + self.config.worker_threads as u64 + } + fn methods(&self) -> Vec<&'static str> { vec!["Agent.Initialize", "Agent.Prepare", "Agent.Commit"] } diff --git a/services/flysim/crates/fly-session/src/cli.rs b/services/flysim/crates/fly-session/src/cli.rs index f13f44f..d492144 100644 --- a/services/flysim/crates/fly-session/src/cli.rs +++ b/services/flysim/crates/fly-session/src/cli.rs @@ -30,6 +30,7 @@ fly-session [options] agent serve one agent worker on a launcher-created endpoint environment serve the environment worker on a launcher-created endpoint measure compare the execution modes and print the measurement table + measure-row measure one row and print it as JSON (one child per row) Worker options (agent and environment): --socket PATH the launcher's endpoint for this participant @@ -48,9 +49,14 @@ Worker options (agent and environment): Measure options: --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) --modes LIST in-process, thread, process (default all three) --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. @@ -65,6 +71,7 @@ pub fn main() -> ExitCode { let result = match command.as_str() { "agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)), "measure" => Options::parse(&rest).and_then(|o| measure(&o)), + "measure-row" => Options::parse(&rest).and_then(|o| measure_row(&o)), "--help" | "-h" | "help" => { print!("{USAGE}"); return ExitCode::SUCCESS; @@ -215,10 +222,10 @@ fn parse_ports(value: &str) -> Result, String> { .collect() } -/// Runs the execution-mode comparison and prints its table. -fn measure(options: &Options) -> Result<(), String> { +fn measure_config(options: &Options) -> Result { let mut config = crate::measure::MeasureConfig { steps: options.u64("steps", 200)?, + warmup_steps: options.u64("warmup-steps", 10)?, worker_threads: options.usize("worker-threads", 1)?, ..crate::measure::MeasureConfig::default() }; @@ -233,19 +240,43 @@ fn measure(options: &Options) -> Result<(), String> { config.modes = list .split(',') .filter(|p| !p.is_empty()) - .map(|p| match p { - "in-process" => Ok(ExecutionMode::InProcess), - "thread" => Ok(ExecutionMode::Thread), - "process" => Ok(ExecutionMode::Process), - other => Err(format!("--modes: {other:?}")), - }) + .map(parse_mode) .collect::, String>>()?; } + Ok(config) +} + +fn parse_mode(name: &str) -> Result { + 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() .enable_all() .build() .map_err(|e| format!("runtime: {e}"))?; - let rows = runtime.block_on(crate::measure::run(&config))?; - print!("{}", crate::measure::table(&rows)); + let row = runtime.block_on(crate::measure::one(&config, mode, agents))?; + println!("{}", row.to_json()); Ok(()) } diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index 22f4221..b72d4b8 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -107,21 +107,41 @@ impl std::error::Error for SessionFailure {} type Outcome = Result; -/// 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 -/// prototype values: ten seconds without progress is a failure, with a separate budget for a -/// long boot. They are failure-detection values, not a gameplay latency goal. Without them a -/// dead participant is a hang rather than a diagnosed outcome. +/// Section 6 is a two-stage procedure, and these are its two stages. `probe` is how long a +/// call may go without a terminal reply before it is *uncertain*; an uncertain call is not a +/// failed one, so the coordinator then runs the section 6 resolution -- a fresh bus call +/// 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)] 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, } impl Default for 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, /// How many times an exact duplicate met IN_PROGRESS while resolving an uncertain call. 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. pub deadlines: Deadlines, /// Per-method and critical-path latency samples. Local synthetic timings, never a @@ -289,6 +311,7 @@ impl Coordinator { injections: Injections::default(), injection_log: Vec::new(), in_progress_replies: 0, + resolutions: 0, deadlines: Deadlines::default(), metrics: Metrics::default(), blame: None, @@ -429,6 +452,15 @@ impl Coordinator { 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> { 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. 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?; for index in 0..self.agents.len() { self.hello_agent(index).await?; @@ -790,14 +836,30 @@ struct Job { 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), + /// 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. /// /// The deadline is the caller's, on the caller's monotonic clock. A participant that has died /// mid-call is usually reported by the bus itself, because its connection took its /// registration with it; this bound is what makes the remaining cases -- a live process that -/// stopped answering -- a diagnosed outcome rather than a hang. An expired deadline is -/// deliberately `unknown`: `ipc-v1` section 6 forbids reading a caller-side timeout as proof -/// that nothing was mutated. +/// stopped answering -- a diagnosed outcome rather than a hang. #[allow(clippy::too_many_arguments)] async fn call_owned( bus: flybus::Client, @@ -809,30 +871,46 @@ async fn call_owned( request_id: DomainRequestId, want: Vec, deadline: Duration, -) -> Result { +) -> CallOutcome { let refs: Vec<(&str, &flybus::Artifact)> = attachments.iter().map(|(n, a)| (n.as_str(), a)).collect(); let call = rpc::call(&bus, &worker, method, scope, params, &refs, request_id, &want); match tokio::time::timeout(deadline, call).await { - Ok(result) => result, - Err(_) => Err(DomainError::new( - ErrorCode::BackendFailure, - format!( - "{method}: {} did not answer within {:?}", - worker.worker_id, deadline - ), - MutationCertainty::Unknown, - )), + Ok(Ok(reply)) => CallOutcome::Answered(Box::new(reply)), + Ok(Err(e)) => CallOutcome::Refused(e), + Err(_) => CallOutcome::Expired, } } +/// 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. +/// +/// 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 { agent_id: Id, - reply: Result, + outcome: CallOutcome, scope: Option, worker: WorkerRef, method: &'static str, + request_id: DomainRequestId, + params: Map, + attachments: Vec<(String, flybus::Artifact)>, /// How long the call took, for the section 5 percentiles. elapsed: Duration, } @@ -859,25 +937,31 @@ impl Coordinator { let deadline = if method.ends_with("Initialize") || method == "Worker.Hello" { self.deadlines.boot } else { - self.deadlines.call + self.deadlines.probe }; let started = Instant::now(); - let reply = call_owned( + let outcome = call_owned( self.bus.clone(), worker.clone(), method, scope.clone(), - params, - owned, - request_id, + params.clone(), + owned.clone(), + request_id.clone(), want.to_vec(), deadline, ) .await; self.metrics.record(method, started.elapsed()); - let reply = match reply { - Ok(reply) => reply, - Err(e) => return Err(self.fail_now(e, method)), + let reply = match outcome { + CallOutcome::Answered(reply) => *reply, + // 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)?; match reply.result() { @@ -945,9 +1029,22 @@ impl Coordinator { Ok(()) } - /// Resolves an uncertain operation: a fresh bus call with the original domain request id - /// and body, pinned to the same service incarnation. It never issues a new request id and - /// never recomputes a decision. + /// The `ipc-v1` section 6 resolution of an uncertain call. + /// + /// 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)] async fn resolve( &mut self, @@ -963,9 +1060,17 @@ impl Coordinator { // starts no second mutation. Waiting and asking again is the resolution, not a retry // of the operation. self.blame(Some(worker.worker_id.clone())); - let deadline = self.deadlines.call; - for attempt in 0..200u32 { - let reply = call_owned( + let probe = self.deadlines.probe; + let budget = self.deadlines.resolve; + 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(), worker.clone(), method, @@ -974,12 +1079,19 @@ impl Coordinator { attachments.clone(), request_id.clone(), want.to_vec(), - deadline, + probe, ) .await; - let reply = match reply { - Ok(reply) => reply, - Err(e) => return Err(self.fail_now(e, method)), + let reply = match outcome { + CallOutcome::Answered(reply) => *reply, + // 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)?; match reply.result() { @@ -988,21 +1100,14 @@ impl Coordinator { return Ok(reply); } Err(e) if e.code == ErrorCode::InProgress => { - let _ = attempt; self.in_progress_replies += 1; 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(self.fail_now( - DomainError::new( - ErrorCode::BackendFailure, - "the uncertain operation never resolved", - MutationCertainty::Unknown, - ), - method, - )) + Err(self.fail_now(unresolved(method, worker, budget), method)) } /// 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>, what: &str, ) { - let deadline = self.deadlines.call; + let deadline = self.deadlines.probe; let reply = call_owned( self.bus.clone(), worker.clone(), @@ -1033,7 +1138,7 @@ impl Coordinator { ) .await; let outcome = match reply { - Ok(reply) => match reply.result() { + CallOutcome::Answered(reply) => match reply.result() { Ok(result) => InjectionOutcome { what: what.to_owned(), code: None, @@ -1045,11 +1150,18 @@ impl Coordinator { identical: false, }, }, - Err(e) => InjectionOutcome { + CallOutcome::Refused(e) => InjectionOutcome { what: what.to_owned(), code: Some(e.code), 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); } @@ -1071,7 +1183,7 @@ impl Coordinator { _ => Map::new(), }; let request_id = self.serials.next(&worker.service); - let reply = call_owned( + let outcome = call_owned( self.bus.clone(), worker.clone(), method, @@ -1080,10 +1192,14 @@ impl Coordinator { Vec::new(), request_id, Vec::new(), - self.deadlines.call, + self.deadlines.probe, ) - .await?; - reply.result().cloned() + .await; + 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`. @@ -1102,8 +1218,10 @@ impl Coordinator { } /// 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, order: DispatchOrder) -> Vec { - let deadline = self.deadlines.call; + let deadline = self.deadlines.probe; let mut out = Vec::new(); match order { DispatchOrder::Sequential | DispatchOrder::Reversed => { @@ -1113,24 +1231,27 @@ impl Coordinator { } for job in jobs { let started = Instant::now(); - let reply = call_owned( + let outcome = call_owned( self.bus.clone(), job.worker.clone(), job.method, job.scope.clone(), - job.params, - job.attachments, - job.request_id, + job.params.clone(), + job.attachments.clone(), + job.request_id.clone(), Vec::new(), deadline, ) .await; out.push(JobResult { agent_id: job.agent_id, - reply, + outcome, scope: job.scope, worker: job.worker, method: job.method, + request_id: job.request_id, + params: job.params, + attachments: job.attachments, elapsed: started.elapsed(), }); } @@ -1145,24 +1266,27 @@ impl Coordinator { let method = job.method; tasks.push(tokio::spawn(async move { let started = Instant::now(); - let reply = call_owned( + let outcome = call_owned( bus, job.worker, job.method, job.scope, - job.params, - job.attachments, - job.request_id, + job.params.clone(), + job.attachments.clone(), + job.request_id.clone(), Vec::new(), deadline, ) .await; JobResult { agent_id, - reply, + outcome, scope, worker, method, + request_id: job.request_id, + params: job.params, + attachments: job.attachments, elapsed: started.elapsed(), } })); @@ -1414,13 +1538,31 @@ impl Coordinator { } let results = self.run_jobs(jobs, self.dispatch).await; 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())); - let reply = match reply { - Ok(reply) => reply, + let reply = match outcome { + 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 // 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)?; let decision: PreparedDecision = match reply.parse() { @@ -1600,7 +1742,7 @@ impl Coordinator { let want = vec!["view.arena".to_owned()]; self.audit.push(format!("advance:{k}")); self.blame(Some(worker.worker_id.clone())); - let advance_deadline = self.deadlines.call; + let advance_deadline = self.deadlines.probe; let advance_started = Instant::now(); let injected = self.injections.at_step == k; @@ -1672,7 +1814,7 @@ impl Coordinator { ) .await? } else { - let reply = call_owned( + let outcome = call_owned( self.bus.clone(), worker.clone(), "Environment.Advance", @@ -1685,14 +1827,32 @@ impl Coordinator { ) .await; self.metrics.record("Environment.Advance", advance_started.elapsed()); - let reply = match reply { - Ok(reply) => reply, - Err(e) => return Err(self.fail_now(e, "advance")), - }; - self.check_reply(&worker, &reply, &Some(scope.clone()), "advance")?; - match reply.result() { - Ok(_) => reply, - Err(e) => return Err(self.fail_now(e, "advance")), + match outcome { + CallOutcome::Answered(reply) => { + let reply = *reply; + self.check_reply(&worker, &reply, &Some(scope.clone()), "advance")?; + match reply.result() { + Ok(_) => reply, + 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 first_failure = None; let mut blamed: Option = 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())); - match reply { - Ok(reply) => { + // Uncertain: resolve the same Commit against the same incarnation first. A Commit + // 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)?; match reply.result() { Ok(_) => {} @@ -2003,12 +2197,13 @@ impl Coordinator { self.blame(None); commits.push((agent_id, result)); } - Err(e) => { + CallOutcome::Refused(e) => { if first_failure.is_none() { blamed = Some(agent_id.clone()); } first_failure = Some(first_failure.unwrap_or(e)); } + CallOutcome::Expired => unreachable!("an expiry was resolved just above"), } } if let Some(error) = first_failure { diff --git a/services/flysim/crates/fly-session/src/environment.rs b/services/flysim/crates/fly-session/src/environment.rs index deb8d21..cf72b21 100644 --- a/services/flysim/crates/fly-session/src/environment.rs +++ b/services/flysim/crates/fly-session/src/environment.rs @@ -36,6 +36,8 @@ pub struct EnvironmentConfig { /// The world's fixed reduced step duration. 60 Hz is `1/60` s. pub step_duration: RationalNs, pub ports: Vec, + /// The thread allocation the launcher started this worker within. + pub worker_threads: usize, pub faults: EnvironmentFaults, } @@ -425,6 +427,10 @@ impl WorkerEndpoint for CounterEnvironment { self.status.clone() } + fn worker_threads(&self) -> u64 { + self.config.worker_threads as u64 + } + fn methods(&self) -> Vec<&'static str> { vec!["Environment.Initialize", "Environment.Advance"] } diff --git a/services/flysim/crates/fly-session/src/harness.rs b/services/flysim/crates/fly-session/src/harness.rs index b4a635e..6840f8c 100644 --- a/services/flysim/crates/fly-session/src/harness.rs +++ b/services/flysim/crates/fly-session/src/harness.rs @@ -390,18 +390,19 @@ impl SessionHarness { 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 - /// [`SessionHarness::progress_of`], which reads it over the bus in every mode. - pub fn agent_mutations(&self, agent_id: &Id) -> u64 { + /// `None` means "not observable from here", not "nothing happened": a participant with a + /// process of its own keeps its counter there. [`SessionHarness::progress_of`] reads it + /// over the bus and works in every mode. + pub fn agent_mutations(&self, agent_id: &Id) -> Option { self.launcher .worker(agent_id) .and_then(crate::launcher::LaunchedWorker::progress_counter) - .unwrap_or_default() } - pub fn environment_mutations(&self) -> u64 { + pub fn environment_mutations(&self) -> Option { self.agent_mutations(&id(ENV_WORKER)) } @@ -420,7 +421,7 @@ impl SessionHarness { pub async fn shutdown(self) { let SessionHarness { coordinator, mut launcher, observers, .. } = self; drop(coordinator); - launcher.reap_all("shutdown").await; + launcher.reap_all(&id("shutdown")).await; for observer in observers.into_inner().expect("not poisoned") { observer.close().await; } diff --git a/services/flysim/crates/fly-session/src/launcher.rs b/services/flysim/crates/fly-session/src/launcher.rs index e1bb474..2537236 100644 --- a/services/flysim/crates/fly-session/src/launcher.rs +++ b/services/flysim/crates/fly-session/src/launcher.rs @@ -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)); } Err(e) if handover(&e) && Instant::now() < deadline => { @@ -969,17 +983,21 @@ impl Launcher { /// 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. /// A participant that does not stop within the budget is terminated, which is reported as /// 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 { return ReapOutcome::AlreadyGone; }; let service = worker.identity.service.clone(); let incarnation = worker.service_incarnation.clone(); 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( SessionRpcRequest { request_id, @@ -1025,7 +1043,7 @@ impl Launcher { } /// 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(); for worker_id in self.worker_ids() { 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(), step_duration: spec.step_duration, ports: spec.ports.clone(), + worker_threads: spec.worker_threads, faults: spec.faults.clone(), } } diff --git a/services/flysim/crates/fly-session/src/measure.rs b/services/flysim/crates/fly-session/src/measure.rs index 18d32b7..963139f 100644 --- a/services/flysim/crates/fly-session/src/measure.rs +++ b/services/flysim/crates/fly-session/src/measure.rs @@ -9,11 +9,19 @@ //! question this slice has to answer is what a process boundary costs, not how fast anything //! is. Pacing is switched off for the run, so a transition follows the one before it as fast //! as the participants answer and the samples are work rather than sleep. +//! +//! **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::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use serde_json::{Value, json}; + use crate::coordinator::DispatchOrder; use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via}; use crate::launcher::ExecutionMode; @@ -108,29 +116,169 @@ pub struct Row { pub sealed_final: usize, pub store_bytes_max: u64, pub store_bytes_final: u64, - /// One sealed frame per boundary, boundary zero included. - pub frames_produced: u64, + /// Frames this run actually observed, counted from the behaviour trace: every sensory + /// 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 { - /// 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 { - 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 { + let u = |name: &str| -> Result { + 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 { + 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. -pub async fn run(config: &MeasureConfig) -> Result, String> { +/// Runs the comparison, one child process per row. +/// +/// `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, String> { let mut rows = Vec::new(); for mode in &config.modes { 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) } -async fn one(config: &MeasureConfig, mode: ExecutionMode, agents: usize) -> Result { +fn row_in_a_child( + config: &MeasureConfig, + program: &std::path::Path, + mode: ExecutionMode, + agents: usize, +) -> Result { + 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 { let dir = std::env::temp_dir().join(format!( "fly-session-measure-{}-{agents}-{}", mode.label(), @@ -247,7 +395,15 @@ async fn measure_in( sealed_final: harness.router_stats().sealed_artifacts, store_bytes_max, store_bytes_final: harness.router_stats().store_bytes, - frames_produced: config.steps + config.warmup_steps + 1, + // 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::(), }; harness.shutdown().await; Ok(row) @@ -261,7 +417,8 @@ pub fn table(rows: &[Row]) -> String { ); if let Some(first) = rows.first() { 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 )); } @@ -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_str( "| mode | agents | coordinator peak RSS KiB | participant peak RSS KiB | owners max/final | \ -roots max | queued max | sealed max/final | store bytes max/final | frames produced/collected |\n", +roots max | queued max | sealed max/final | store bytes max/final | frames observed/collected |\n", ); out.push_str("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n"); 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.store_bytes_max, r.store_bytes_final, - r.frames_produced, + r.frames_observed, r.collected(), )); } diff --git a/services/flysim/crates/fly-session/src/worker.rs b/services/flysim/crates/fly-session/src/worker.rs index aeb6ed2..327d66b 100644 --- a/services/flysim/crates/fly-session/src/worker.rs +++ b/services/flysim/crates/fly-session/src/worker.rs @@ -188,6 +188,12 @@ pub trait WorkerEndpoint: Send + 'static { fn capabilities(&self) -> Vec; 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. /// Anything else returns UNSUPPORTED without entering the endpoint. fn methods(&self) -> Vec<&'static str>; @@ -275,7 +281,7 @@ async fn run( ) { // 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. - 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; ( e.worker_id(), @@ -285,6 +291,7 @@ async fn run( e.capabilities(), e.status_cell(), e.methods(), + e.worker_threads(), ) }; let mut running: Vec> = Vec::new(); @@ -321,6 +328,7 @@ async fn run( &incarnation_id, role, &capabilities, + threads, ); let _ = responder.reply(outcome.to_outcome(), &[]).await; continue; @@ -622,6 +630,7 @@ fn failure( failure_outcome(request_id, worker_id, incarnation_id, scope, error) } +#[allow(clippy::too_many_arguments)] fn hello( request: &SessionRpcRequest, session_id: &Id, @@ -629,6 +638,7 @@ fn hello( incarnation_id: &Id, role: Role, capabilities: &[Id], + worker_threads: u64, ) -> SessionRpcOutcome { let params: HelloParams = match HelloParams::from_json(&request.params) { Ok(params) => params, @@ -684,6 +694,7 @@ fn hello( capabilities: capabilities.to_vec(), max_agents: MAX_AGENTS as u64, max_ports: MAX_PORTS as u64, + worker_threads, }; success( request, diff --git a/services/flysim/crates/fly-session/tests/failures.rs b/services/flysim/crates/fly-session/tests/failures.rs index 60e7b66..7de2bc1 100644 --- a/services/flysim/crates/fly-session/tests/failures.rs +++ b/services/flysim/crates/fly-session/tests/failures.rs @@ -33,6 +33,15 @@ both_transports!( const STEPS: u64 = 4; 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. struct Run { behaviour: Vec, @@ -51,8 +60,8 @@ async fn run_with(via: Via, injections: Injections) -> Run { let run = Run { behaviour: f.harness.coordinator.trace.behavior(), mutations: vec![ - (fly_a(), f.harness.agent_mutations(&fly_a())), - (fly_b(), f.harness.agent_mutations(&fly_b())), + (fly_a(), local_mutations(&f, &fly_a())), + (fly_b(), local_mutations(&f, &fly_b())), ], counter: f .harness diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs index 248a060..1c3f6a4 100644 --- a/services/flysim/crates/fly-session/tests/processes.rs +++ b/services/flysim/crates/fly-session/tests/processes.rs @@ -22,6 +22,7 @@ use fly_session::phase::Phase; use fly_session::types::*; all_modes!( + a_slow_participant_is_resolved_rather_than_failed, a_delayed_one_agent_result_holds_the_world, a_worker_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 = 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 @@ -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(); 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. assert_eq!( 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); // 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!(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; } @@ -470,6 +586,10 @@ async fn a_router_restart_during_a_world_advance_fences_the_epoch() { }); within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); 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 started = Instant::now(); @@ -499,6 +619,11 @@ async fn a_router_restart_during_a_world_advance_fences_the_epoch() { f.harness.coordinator.is_fenced(), "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!(count(&f.harness.coordinator.audit, "publish:1"), 0); 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; } -/// The other half of the same row: the replacement process is live and refuses an operation -/// naming the epoch the old process belonged to, rather than applying it to a fresh brain. +/// The other half of the same row, in two parts, because the two refusals are different +/// 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)] async fn a_restarted_worker_refuses_an_operation_from_the_old_epoch() { let mode = ExecutionMode::Process; let mut f = mode_fixture(mode, two_agents(mode)).await; within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); within("step", f.harness.coordinator.step()).await.unwrap(); - let restarted = f.harness.restart_agent(&fly_b()).await.unwrap(); + // 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( &restarted.service, &restarted.service_incarnation, &fly_b(), ); - let params = serde_json::json!({ - "agentId": "fly-b", - "profileDigest": digest_of_bytes(b"whatever"), - "interval": {"numerator": "16666667", "denominator": "1"}, - "decisionContextDigest": digest_of_bytes(b"whatever"), - "preStepStimulations": [], - }); let err = within( - "stale epoch", + "uninitialized replacement", f.harness.coordinator.probe_raw( &replacement, "Agent.Prepare", Some(scope_at("demo", "e1", 1)), - params, + prepare_params("fly-b"), ), ) .await .expect_err("an uninitialized replacement has no epoch to prepare in"); - assert!( - matches!(err.code, ErrorCode::StaleEpoch | ErrorCode::InvalidPhase), - "a replacement refuses the old epoch's work: {err}" + assert_eq!( + err.code, + 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!(f.harness.coordinator.stats().advances, 1); 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": [], + }) +} diff --git a/services/flysim/crates/fly-session/tests/session.rs b/services/flysim/crates/fly-session/tests/session.rs index bc7c7bf..752693b 100644 --- a/services/flysim/crates/fly-session/tests/session.rs +++ b/services/flysim/crates/fly-session/tests/session.rs @@ -35,7 +35,7 @@ const STEPS: u64 = 3; async fn one_world_advance_per_complete_batch(via: Via) { let mut f = default_fixture(via).await; 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(); assert_eq!(reports.len() as u64, 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 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!(f.harness.coordinator.trace.transitions.len() as u64, STEPS); 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. for agent in [fly_a(), fly_b()] { 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" ); }