diff --git a/src/contracts/runtimeContractManifest.ts b/src/contracts/runtimeContractManifest.ts index 687b104..7bed4e5 100644 --- a/src/contracts/runtimeContractManifest.ts +++ b/src/contracts/runtimeContractManifest.ts @@ -185,5 +185,5 @@ export const RUNTIME_CONTRACT_MANIFEST = { ] }, healthResponseSchema: { type: "object", additionalProperties: false, required: ["version", "state", "agents"], properties: { version: { const: "noopolis.daimon.organization-runtime-health.v1" }, state: { enum: ["starting", "running", "stopping", "stopped"] }, agents: { type: "array", maxItems: ORGANIZATION_RUNTIME_MAX_AGENTS, items: { type: "object", additionalProperties: false, required: ["agentId", "state"], properties: { agentId: text, state: { enum: ["starting", "running", "stopping", "stopped", "idle", "failed"] } } } } } }, activityResponseSchema: { type: "object", additionalProperties: false, required: ["version", "items"], properties: { version: { const: "noopolis.daimon.organization-runtime-activity.v1" }, items: { type: "array", maxItems: 100, items: activityItem }, nextCursor: { type: "string", minLength: 1, maxLength: 16, pattern: "^(0|[1-9][0-9]{0,15})$" } } }, - activityV2ResponseSchema: { type: "object", additionalProperties: false, required: ["version", "items"], properties: { version: { const: ORGANIZATION_RUNTIME_ACTIVITY_V2_VERSION }, state: { enum: ["running", "stopped"] }, executions: { type: "array", maxItems: ORGANIZATION_RUNTIME_MAX_AGENTS, items: { type: "object", additionalProperties: false, required: ["agent_id", "execution_id", "state", "delivery_ids"], properties: { agent_id: text, execution_id: { type: "string" }, state: { const: "running" }, delivery_ids: { type: "array", maxItems: 32, items: text } } } }, items: { type: "array", maxItems: 2_112, items: { type: "object", additionalProperties: false, required: ["version", "acceptance_id", "agent_id", "delivery_id", "request_digest", "state", "accepted_at", "updated_at", "active"], properties: { version: { const: "noopolis.daimon.wake-receipt-status.v2" }, acceptance_id: { type: "string" }, agent_id: text, delivery_id: text, request_digest: { type: "string" }, state: { enum: ["accepted", "running", "completed", "failed", "stopped"] }, accepted_at: timestamp, updated_at: timestamp, active: { type: "boolean" }, execution_id: { type: "string" }, deferred: { type: "boolean" }, text: { type: "string", maxLength: 16384 }, queue_position: { type: "integer", minimum: 1 }, code: { enum: ["engine_failed", "host_stopped", "host_stopping", "queue_full", "unknown_agent"] } } } } } } + activityV2ResponseSchema: { type: "object", additionalProperties: false, required: ["version", "items"], properties: { version: { const: ORGANIZATION_RUNTIME_ACTIVITY_V2_VERSION }, state: { enum: ["running", "stopped"] }, executions: { type: "array", maxItems: ORGANIZATION_RUNTIME_MAX_AGENTS, items: { type: "object", additionalProperties: false, required: ["agent_id", "execution_id", "state", "delivery_ids"], properties: { agent_id: text, execution_id: { type: "string" }, state: { const: "running" }, delivery_ids: { type: "array", maxItems: 32, items: text } } } }, items: { type: "array", maxItems: 2_112, items: { type: "object", additionalProperties: false, required: ["version", "acceptance_id", "agent_id", "delivery_id", "request_digest", "state", "accepted_at", "updated_at", "active"], properties: { version: { const: "noopolis.daimon.wake-receipt-status.v2" }, acceptance_id: { type: "string" }, agent_id: text, delivery_id: text, request_digest: { type: "string" }, state: { enum: ["accepted", "running", "completed", "failed", "stopped"] }, accepted_at: timestamp, updated_at: timestamp, active: { type: "boolean" }, execution_id: { type: "string" }, deferred: { type: "boolean" }, text: { type: "string", maxLength: 16384 }, queue_position: { type: "integer", minimum: 1 }, code: { enum: ["engine_failed", "host_stopped", "host_stopping", "queued_wake_stopped", "active_wake_aborted", "queue_full", "unknown_agent"] } } } } } } } as const; diff --git a/src/runtime/AGENTS.md b/src/runtime/AGENTS.md index f53a605..08250d9 100644 --- a/src/runtime/AGENTS.md +++ b/src/runtime/AGENTS.md @@ -746,3 +746,34 @@ stay absence), and a claim about the store-backed routes beside it — neither settles a closure proof. `state` is optional on the wire for the reason every additive member here is: a projection published before the seal existed must still parse, and its absence means "not stated", never "running". + +A delivery returned to the inbox for restart records the outcome that returned it, +and **only a wake outcome can return one**. `attentionDispatcher` reclaims an +undisposed delivery to `accepted` on exactly one condition — a wake result of +`stopped`, which is also the shape an aborted in-flight wake arrives in +(`organizationRuntimeHost.ts` settles a queued job `queued_wake_stopped` and the +in-flight one `active_wake_aborted`). The dispatcher's own `stopping` latch used to +share that condition, and it is a HOST-LIFECYCLE fact, not a wake outcome: a wake +that *completed* had its evidence discarded because the dispatcher happened to be +halting, and the delivery was recorded `accepted, deferred: false, execution id +retained, no code` — byte-identical to "never ran" and to "ran but forgotten". +Production tolerated that because a restart re-delivers and the agent redoes the +work; a one-shot isolated trial has no restart, so the information was simply lost +and a subject that ran and made a choice reported as an infrastructure failure. It +is the wrong record for production too: an agent that read a delivery and declined +to dispose of it is **deferred**, whichever way the host is heading, and a restart +must not re-deliver it as fresh work. So a completed or failed wake takes the +deferred path regardless of dispatcher state, and `stopping` guards only the +pre-wake path, which is where it belongs — it must never be restored to the +post-wake decision. `WakeReceiptCode` carries `queued_wake_stopped` and +`active_wake_aborted` beside the existing five, because those are the two shapes a +shutdown really gives a wake and neither had an honest name. The wake's own code is +recorded exactly; nothing else names a reclaim, because a plausible name for an +undetermined cause gets acted on and a missing one does not. Two consequences, both +load bearing: `accepted` is the one non-terminal state a record may carry a code in, +since it is the only one reached *from* an ended execution (`running` and +`completed` still refuse one), and `transitionClaimed` no longer carries a code +across a transition — it describes the transition that produced the current state, +and a reclaimed delivery is claimed again later. Widening the enum rotates the +contract manifest digest, so Spawnfile must re-vendor +`contract-manifest.json`/`.sha256` and its pinned constant. diff --git a/src/runtime/attentionDispatcher.ts b/src/runtime/attentionDispatcher.ts index ad204fc..a63c9eb 100644 --- a/src/runtime/attentionDispatcher.ts +++ b/src/runtime/attentionDispatcher.ts @@ -122,7 +122,17 @@ export class AttentionDispatcher { const executionError = result.status === "rejected" ? `wake rejected: ${result.code}` : result.status === "failed" ? `engine_failed: ${result.detail ?? "engine execution failed"}` : null; for (const item of claimed.filter((value) => !value.done)) { - if (this.stopping || result.status === "stopped") await store.transitionClaimed(item.record.acceptance_id, item.claim, "accepted"); + // Returned to the inbox for restart, and recording WHY. Only a WAKE OUTCOME + // decides this: `stopped` — which an aborted in-flight wake also carries, with + // its own code — is the runtime reclaiming work nobody read. The dispatcher's + // own halt is not an outcome and must not stand in for one; it guards the + // pre-wake path, where it belongs. Keying on it here discarded a completed + // wake's evidence because the host happened to be halting, leaving a record + // byte-identical to "never ran". Production tolerated it because a restart + // re-delivers; a one-shot isolated trial has no restart and simply lost it. + if (result.status === "stopped") { + await store.transitionClaimed(item.record.acceptance_id, item.claim, "accepted", result.code); + } else if (agent.attention !== undefined) { // Successful reading is not completion. A failed execution also keeps // unfinished deliveries, and its execution id for idempotent retry. diff --git a/src/runtime/organizationRuntimeClosure.test.ts b/src/runtime/organizationRuntimeClosure.test.ts index 8ec4178..67b1a48 100644 --- a/src/runtime/organizationRuntimeClosure.test.ts +++ b/src/runtime/organizationRuntimeClosure.test.ts @@ -87,3 +87,101 @@ test("an unstarted host seals nothing and a repeated stop does not erase the sea }); async function privateRoot(): Promise { const root = await mkdtemp(path.join(os.tmpdir(), "daimon-closure-")); await chmod(root, 0o700); return root; } + +/** + * A delivery returned to the inbox for restart must say which outcome returned it. + * + * `attentionDispatcher` reclaims an undisposed delivery to `accepted` on two + * conditions — the dispatcher halting, and a wake that came back `stopped` — and it + * recorded neither, so the receipt an evaluator reads was identical for both. A + * live trial closed its execution, spent real money and reported an `accepted` + * delivery with no marker and no reason, and four investigations went into telling + * those two apart from the outside. The wake's own code is exact and is now kept; + * a halt has no code of its own and stays absent, because a plausible name for an + * undetermined cause gets acted on and a missing one does not. + */ +test("a delivery reclaimed for restart records the stopped wake's own code", async () => { + const root = await privateRoot(); + const attention = { version: ORGANIZATION_RUNTIME_VERSION, host: config.host, + agents: [{ ...config.agents[0]!, attention: { maxBatchMessages: 4, maxBatchBytes: 4096, maxExecutions: 8, maxTokens: 100_000 } }] }; + let stopWake = false; + const stopping = { + ...core, + async wake(request: OrganizationRuntimeWakeRequest) { + if (!stopWake) return { version: "noopolis.daimon.wake-result.v1", status: "completed", agentId: request.agentId, wakeId: request.event.id, text: "private", durationMs: 1 } as const; + // Exactly what organizationRuntimeHost settles an in-flight wake with at shutdown. + return { version: "noopolis.daimon.wake-result.v1", status: "stopped", agentId: request.agentId, wakeId: request.event.id, code: "active_wake_aborted" } as const; + } + } as unknown as OrganizationRuntimeHost; + const control = createOrganizationRuntimeControlHostWithCoreForTest(attention, stopping, { acceptanceStorePath: root, controlToken: token, storeOptions }); + try { + await control.start(); + stopWake = true; + const accepted = await control.accept(delivery("restart-delivery")); + assert.equal(accepted.state, "accepted"); + await waitFor(async () => (await control.activityV2(token))?.items.some((item) => item.state === "accepted" && item.code !== undefined) === true); + const item = (await control.activityV2(token))?.items.find((row) => row.delivery_id === "restart-delivery"); + // Returned for restart, undisposed, and no longer silent about which outcome did it. + assert.equal(item?.state, "accepted"); + // Exactly the live shape: the running transition left deferred FALSE and the + // reclaim does not clear it, which is what distinguishes it from a real deferral. + assert.equal(item?.deferred, false); + assert.equal(item?.code, "active_wake_aborted"); + } finally { await control.stop().catch(() => undefined); await rm(root, { recursive: true, force: true }); } +}); + +async function waitFor(predicate: () => Promise, timeoutMs = 5_000): Promise { + const deadline = Date.now() + timeoutMs; + do { if (await predicate()) return; await new Promise((resolve) => setTimeout(resolve, 10)); } while (Date.now() < deadline); + throw new Error("timed out waiting for the reclaimed delivery"); +} + +/** + * The other half, and the one the reclaim path kept getting wrong. A wake that + * COMPLETED is a wake outcome; the dispatcher happening to be halting when it + * lands is not. Keying the reclaim on the host's own `stopping` latch discarded + * that outcome and wrote `accepted, deferred: false, execution retained, no code` + * — a record byte-identical to "never ran" and to "ran but forgotten". Production + * survived it because a restart re-delivers and the agent redoes the work; a + * one-shot isolated trial has no restart, so the evidence was simply lost and a + * subject that really ran and made a choice reported as infrastructure failure. + * An agent that read a delivery and declined to dispose of it is DEFERRED, + * whichever way the host is heading, and a restart must not re-deliver it as + * fresh work. + */ +test("a completed wake under a halting dispatcher is deferred, not reclaimed for restart", async () => { + const root = await privateRoot(); + const attention = { version: ORGANIZATION_RUNTIME_VERSION, host: config.host, + agents: [{ ...config.agents[0]!, attention: { maxBatchMessages: 4, maxBatchBytes: 4096, maxExecutions: 8, maxTokens: 100_000 } }] }; + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + let arrived!: () => void; + const waking = new Promise((resolve) => { arrived = resolve; }); + const blocking = { ...core, + async wake(request: OrganizationRuntimeWakeRequest) { + arrived(); + await held; + return { version: "noopolis.daimon.wake-result.v1", status: "completed", agentId: request.agentId, wakeId: request.event.id, text: "private", durationMs: 1 } as const; + } + } as unknown as OrganizationRuntimeHost; + const control = createOrganizationRuntimeControlHostWithCoreForTest(attention, blocking, { acceptanceStorePath: root, controlToken: token, storeOptions }); + try { + await control.start(); + await control.accept(delivery("halted-delivery")); + await waking; + // The halt lands while the wake is in flight; the wake then completes anyway. + const stopping = control.stop(); + release(); + await stopping; + const item = (await control.activityV2(token))?.items.find((row) => row.delivery_id === "halted-delivery"); + assert.equal(item?.state, "accepted"); + // The wake's own outcome decides the record: read, undisposed, deferred. + assert.equal(item?.deferred, true); + // A completed wake releases its execution, so a restart waits for new input + // instead of replaying the delivery as work nobody has seen. + assert.equal(item?.execution_id, undefined); + // Still silent: a completed wake is no more a named reclaim outcome than a + // halt is, and a plausible name for an undetermined cause gets acted on. + assert.equal(item?.code, undefined); + } finally { await control.stop().catch(() => undefined); await rm(root, { recursive: true, force: true }); } +}); diff --git a/src/runtime/wakeAcceptanceReconciliation.test.ts b/src/runtime/wakeAcceptanceReconciliation.test.ts index fb2bd4e..f7ab14c 100644 --- a/src/runtime/wakeAcceptanceReconciliation.test.ts +++ b/src/runtime/wakeAcceptanceReconciliation.test.ts @@ -48,9 +48,16 @@ test("offline reconciliation blocks untrusted proof, identity mismatch, and conc const mismatch = await reconcileOfflineWakeTransition({ ...request, lock: { ...request.lock, ino: request.lock.ino + 1 } }, { storePath: root, ...testLeaseOptions, verifyDeploymentAttestation: async (context) => ({ request_digest: context.request_digest, nonce: context.nonce, exclusive_store: true, authorized_registration_digests: [] }) }); assert.equal(mismatch.state, "blocked"); let release!: () => void; + let leaseCreated!: () => void; const paused = new Promise((resolve) => { release = resolve; }); - const first = reconcileOfflineWakeTransition(request, { storePath: root, ...testLeaseOptions, verifyDeploymentAttestation: async (context) => { await paused; return { request_digest: context.request_digest, nonce: context.nonce, exclusive_store: true, authorized_registration_digests: [] }; } }); - await new Promise((resolve) => setTimeout(resolve, 5)); + // `verifyDeploymentAttestation` runs only after `acquireLease` has published the + // lease, so signalling from inside it is a real happens-after of that publication. + // A sleep is not: publishing the lease is several fsynced filesystem operations and + // takes ~5 ms even on an idle machine, so a 5 ms timer raced it and the store then + // opened against a store no one had reserved yet. + const leased = new Promise((resolve) => { leaseCreated = resolve; }); + const first = reconcileOfflineWakeTransition(request, { storePath: root, ...testLeaseOptions, verifyDeploymentAttestation: async (context) => { leaseCreated(); await paused; return { request_digest: context.request_digest, nonce: context.nonce, exclusive_store: true, authorized_registration_digests: [] }; } }); + await leased; await assert.rejects(WakeAcceptanceStore.open(root, testStoreOptions), /reserved for offline reconciliation/); const concurrent = await reconcileOfflineWakeTransition(request, { storePath: root, ...testLeaseOptions, verifyDeploymentAttestation: async (context) => ({ request_digest: context.request_digest, nonce: context.nonce, exclusive_store: true, authorized_registration_digests: [] }) }); assert.equal(concurrent.state, "blocked"); diff --git a/src/runtime/wakeAcceptanceRecord.ts b/src/runtime/wakeAcceptanceRecord.ts index ddf544b..7e5ae32 100644 --- a/src/runtime/wakeAcceptanceRecord.ts +++ b/src/runtime/wakeAcceptanceRecord.ts @@ -40,8 +40,12 @@ export function parseStoredWakeAcceptance(value: unknown): StoredWakeAcceptanceR if (executionError === "" || record.execution_error !== undefined && Buffer.byteLength(string(record.execution_error)) > MAX_EXECUTION_ERROR_BYTES) throw new Error("wake acceptance execution error is invalid"); const completionText = record.text === undefined ? undefined : sanitizeWakeCompletionText(string(record.text)); if (claimGeneration !== undefined && !uuid(claimGeneration)) throw new Error("wake acceptance record is invalid"); - if (code !== undefined && !(["engine_failed", "host_stopped", "host_stopping", "queue_full", "unknown_agent"] as const).includes(code)) throw new Error("wake acceptance record is invalid"); - if ((state === "accepted" || state === "running" || state === "completed") && code !== undefined) throw new Error("wake acceptance record is invalid"); + if (code !== undefined && !(["engine_failed", "host_stopped", "host_stopping", "queued_wake_stopped", "active_wake_aborted", "queue_full", "unknown_agent"] as const).includes(code)) throw new Error("wake acceptance record is invalid"); + // `accepted` is the one non-terminal state a record can be arrived at FROM an ended + // execution: the dispatcher returns an undisposed delivery there for restart, and the + // outcome that returned it is the only account of why. `running` and `completed` still + // refuse a code, where one would be nonsense rather than evidence. + if ((state === "running" || state === "completed") && code !== undefined) throw new Error("wake acceptance record is invalid"); if ((state === "failed" || state === "stopped") && code === undefined) throw new Error("wake acceptance record is invalid"); if ((state !== "completed" && state !== "failed" && completionText !== undefined) || completionText !== record.text) throw new Error("wake acceptance record is invalid"); if (string(record.request_digest) !== wakeAcceptanceDigest(parsed) || !uuid(string(record.acceptance_id))) throw new Error("wake acceptance record is invalid"); diff --git a/src/runtime/wakeAcceptanceStore.ts b/src/runtime/wakeAcceptanceStore.ts index bad2dd6..7c456ed 100644 --- a/src/runtime/wakeAcceptanceStore.ts +++ b/src/runtime/wakeAcceptanceStore.ts @@ -198,7 +198,13 @@ export class WakeAcceptanceStore { await this.afterFinalLockAssertion?.(); await this.assertTransitionLock(record, lock); const target = this.fileFor(record.agent_id, record.delivery_id); - const next: Stored = { ...record, state, updated_at: new Date().toISOString(), claim_generation: claim.generation, ...(code === undefined ? {} : { code }), ...(completedText === undefined ? {} : { text: sanitizeWakeCompletionText(completedText) }), ...(attention === undefined ? {} : { execution_id: attention.clear_execution ? undefined : attention.execution_id ?? record.execution_id, deferred: attention.deferred ?? record.deferred, execution_error: attention.execution_error === null ? undefined : attention.execution_error === undefined ? record.execution_error : sanitizeExecutionError(attention.execution_error) || undefined }) }; + // The code explains the transition that produced the CURRENT state, so a + // transition that names none clears it. While codes existed only on terminal + // records this could not matter — a terminal record returns above and is never + // rewritten — but a delivery reclaimed to `accepted` with its wake's outcome is + // claimed again later, and a carried-over code would describe the wrong state. + const { code: _replaced, ...carried } = record; + const next: Stored = { ...carried, state, updated_at: new Date().toISOString(), claim_generation: claim.generation, ...(code === undefined ? {} : { code }), ...(completedText === undefined ? {} : { text: sanitizeWakeCompletionText(completedText) }), ...(attention === undefined ? {} : { execution_id: attention.clear_execution ? undefined : attention.execution_id ?? record.execution_id, deferred: attention.deferred ?? record.deferred, execution_error: attention.execution_error === null ? undefined : attention.execution_error === undefined ? record.execution_error : sanitizeExecutionError(attention.execution_error) || undefined }) }; await this.replace(target, next); if (claim.acceptance_ids === undefined && (isTerminal(state) || state === "accepted")) { const currentClaim = await this.readClaimOptional(this.claimFor(record)); diff --git a/src/runtime/wakeAcceptanceTypes.ts b/src/runtime/wakeAcceptanceTypes.ts index e21de59..8c6ad3a 100644 --- a/src/runtime/wakeAcceptanceTypes.ts +++ b/src/runtime/wakeAcceptanceTypes.ts @@ -25,12 +25,21 @@ export const WAKE_ACCEPTANCE_REQUEST_SCHEMA = { export const WAKE_RECEIPT_STATUS_SCHEMA = { $schema: "https://json-schema.org/draft/2020-12/schema", $id: WAKE_RECEIPT_STATUS_VERSION, type: "object", additionalProperties: false, required: ["version", "acceptance_id", "agent_id", "delivery_id", "request_digest", "state", "accepted_at", "updated_at"], properties: { - version: { const: WAKE_RECEIPT_STATUS_VERSION }, acceptance_id: { type: "string", pattern: "^[0-9a-f-]{36}$" }, agent_id: { type: "string" }, delivery_id: { type: "string" }, request_digest: { type: "string", pattern: "^[a-f0-9]{64}$" }, state: { enum: ["accepted", "running", "completed", "failed", "stopped"] }, accepted_at: { type: "string" }, updated_at: { type: "string" }, execution_id: { type: "string", pattern: "^[0-9a-f-]{36}$" }, deferred: { type: "boolean" }, code: { enum: ["engine_failed", "host_stopped", "host_stopping", "queue_full", "unknown_agent"] }, text: { type: "string", maxLength: MAX_WAKE_COMPLETION_TEXT_BYTES } + version: { const: WAKE_RECEIPT_STATUS_VERSION }, acceptance_id: { type: "string", pattern: "^[0-9a-f-]{36}$" }, agent_id: { type: "string" }, delivery_id: { type: "string" }, request_digest: { type: "string", pattern: "^[a-f0-9]{64}$" }, state: { enum: ["accepted", "running", "completed", "failed", "stopped"] }, accepted_at: { type: "string" }, updated_at: { type: "string" }, execution_id: { type: "string", pattern: "^[0-9a-f-]{36}$" }, deferred: { type: "boolean" }, code: { enum: ["engine_failed", "host_stopped", "host_stopping", "queued_wake_stopped", "active_wake_aborted", "queue_full", "unknown_agent"] }, text: { type: "string", maxLength: MAX_WAKE_COMPLETION_TEXT_BYTES } } } as const; export type WakeReceiptState = "accepted" | "running" | "completed" | "failed" | "stopped"; -export type WakeReceiptCode = "engine_failed" | "host_stopped" | "host_stopping" | "queue_full" | "unknown_agent"; +/** + * Why a receipt reached its state, and every member is a state the runtime really + * produces: `queued_wake_stopped` and `active_wake_aborted` are the two shapes a + * shutdown gives a wake (`organizationRuntimeHost.ts` settles a queued job with the + * first and the in-flight one with the second), and a delivery reclaimed for restart + * used to record neither, because the only caller that could name them passed no + * code at all. A code that cannot be determined stays ABSENT: a plausible name for + * an undetermined cause is worse than no name, because it is acted on. + */ +export type WakeReceiptCode = "engine_failed" | "host_stopped" | "host_stopping" | "queued_wake_stopped" | "active_wake_aborted" | "queue_full" | "unknown_agent"; export type OrganizationRuntimeWakeAcceptanceRequest = Readonly<{ token: string | undefined; agent_id: string;