diff --git a/docs/runtime.md b/docs/runtime.md index 9574a5a..490f08d 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -73,7 +73,21 @@ Attention is opt-in per agent: Attention agents require the durable v2 route. They receive `daimon_inbox` and `daimon_inbox_disposition`; reading does not complete a message, and unmarked -or deferred deliveries stay pending. +or deferred deliveries stay pending. The generated execution prompt must fit +both the 4,096-codepoint and 16,384-byte runtime limits, including the inbox +wrapper. Larger selected payloads remain intact in `daimon_inbox`; the prompt +instructs the agent to read them there. + +Rejected or failed executions also leave unfinished deliveries pending, but +persist a bounded, credential-redacted diagnostic. `GET /v2/availability` +reports that agent's `error` and a `paused` aggregate state, even after restart +or successful work on a different delivery. The diagnostic clears when that +delivery is successfully handled or explicitly deferred by the agent. Retry +still requires new external input; failures do not create a self-wake loop. + +The diagnostic is an optional private `execution_error` field in the stored +receipt. Public receipt schemas are unchanged. Older runtimes cannot read a +store containing that field; retain the newer runtime when recovering it. Version 2 schedules are normalized on the agent: diff --git a/src/runtime/attentionDispatcher.integration.test.ts b/src/runtime/attentionDispatcher.integration.test.ts new file mode 100644 index 0000000..8810118 --- /dev/null +++ b/src/runtime/attentionDispatcher.integration.test.ts @@ -0,0 +1,185 @@ +import assert from "node:assert/strict"; +import { chmod, mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import test from "node:test"; +import type { WakeEvent } from "../core/types.js"; +import { attentionTools, type AttentionRegistry, type AttentionTurn } from "./attention.js"; +import { createOrganizationRuntimeHostForTest } from "./organizationRuntimeHost.js"; +import { createOrganizationRuntimeControlHostWithCoreForTest } from "./organizationRuntimeControl.js"; +import { parseOrganizationRuntimeWakeRequest } from "./organizationRuntime.js"; +import { wakeAcceptanceDigest } from "./wakeAcceptanceTypes.js"; + +const tokenEnv = "DAIMON_ATTENTION_INTEGRATION_TOKEN"; +const token = "attention-integration"; +const pause = (ms = 5) => new Promise((resolve) => setTimeout(resolve, ms)); +async function until(check: () => boolean | Promise) { + for (let n = 0; n < 300; n++) { if (await check()) return; await pause(); } + throw new Error("expected side effect did not appear"); +} +const request = (id: string, text: string, kind = "message") => ({ token, agent_id: "alpha", delivery_id: id, + event: { version: "noopolis.daimon.wake.v2", kind, text, occurred_at: "2026-09-12T11:30:00.000Z" } }); + +async function fixture() { + const root = await mkdtemp(path.join(os.tmpdir(), "daimon-attention-validation-")); await chmod(root, 0o700); + const previousToken = process.env[tokenEnv]; process.env[tokenEnv] = token; + const config = { version: "noopolis.daimon.organization-runtime.v1", host: { bindHost: "127.0.0.1", port: 4318, controlTokenEnv: tokenEnv }, + agents: [{ id: "alpha", name: "Alpha", instructions: "Act", workspacePath: "/workspace/alpha", runtimeHomePath: "/homes/alpha", engine: { kind: "codex" }, attention: {} }] }; + let registry: AttentionRegistry; + const calls: WakeEvent[] = []; + let attempts = 0; + let reject = false; + let onTurn = async (_event: WakeEvent, turn: AttentionTurn) => { + for (const message of turn.messages) await turn.disposition(message.delivery_id, "complete"); + }; + const make = () => { + registry = new Map(); + // Keep the real host, request parser, store, dispatcher, and inbox tools. + // Only cognition is substituted; no provider login or network is involved. + const host = createOrganizationRuntimeHostForTest(config, async () => ({ id: "alpha", + status: () => ({ agentId: "alpha", state: "idle" }), stop: async () => {}, + wake: async (event) => { calls.push(event); await onTurn(event, registry.get("alpha")!); return { agentId: "alpha", text: "done", durationMs: 1 }; } + })); + const wake = host.wake; + host.wake = async (input) => { + attempts++; + if (reject) return { version: "noopolis.daimon.wake-result.v1", status: "rejected", agentId: "alpha", wakeId: input.event.id, code: "invalid_request" }; + return wake(input); + }; + return createOrganizationRuntimeControlHostWithCoreForTest(config, host, { acceptanceStorePath: root, controlToken: token, attentionRegistryForTest: registry, storeOptions: process.platform === "linux" ? {} : { processIdentity: async () => ({ pid: 1, process_start: "test-start", boot_id: "test-boot", pid_namespace_dev: 1, pid_namespace_ino: 1 }), ownerLiveness: async () => true }, fuseEnvironment: { DAIMON_WAKE_FUSE: "off" } }); + }; + let control = make(); await control.start(); + return { root, calls, get attempts() { return attempts; }, get control() { return control; }, get registry() { return registry; }, + set onTurn(value: typeof onTurn) { onTurn = value; }, set reject(value: boolean) { reject = value; }, + async restart() { await control.stop(); control = make(); await control.start(); }, + async cleanup() { await control.stop(); await rm(root, { recursive: true, force: true }); if (previousToken === undefined) delete process.env[tokenEnv]; else process.env[tokenEnv] = previousToken; } + }; +} + +test("a valid long scheduled delivery crosses the real host boundary and completes via its intact inbox", async () => { + const f = await fixture(); + const text = "S".repeat(3540), deliveryId = `schedule:${"a".repeat(180)}:2026-09-12T13:30@GMT+02:00`; + try { + parseOrganizationRuntimeWakeRequest({ token, agentId: "alpha", event: { version: "noopolis.daimon.wake.v1", id: "valid-source", kind: "schedule", text, occurredAt: "2026-09-12T11:30:00.000Z" } }); + f.onTurn = async (event, turn) => { + assert.match(event.text, /read it with daimon_inbox/); + const tool = attentionTools("alpha", f.registry)[0]!; + const result = await tool.execute("read", {}, undefined, undefined, {} as never); + const details = result.details as { messages: Array<{ delivery_id: string; text: string }> }; + assert.equal(details.messages[0]!.text, text); assert.equal(details.messages[0]!.delivery_id, deliveryId); + await turn.disposition(deliveryId, "complete"); + }; + const receipt = await f.control.accept(request(deliveryId, text, "schedule")); assert.equal(receipt.state, "accepted"); + await until(async () => (await f.control.activityV2(token))!.items[0]?.state === "completed"); + assert.equal(f.calls.length, 1); + assert.equal((await f.control.availability(token))!.state, "running"); + } finally { await f.cleanup(); } +}); + +test("a busy inbox batches six large messages without losing payloads or failing the real validator", async () => { + const f = await fixture(); let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + const bodies = Array.from({ length: 6 }, (_, i) => `${i}:` + "R".repeat(1500)); + try { + f.onTurn = async (event, turn) => { + if (turn.messages[0]!.delivery_id === "hold") await held; + else { + assert.equal(turn.messages.length, 6); assert.match(event.text, /read it with daimon_inbox/); + assert.deepEqual(turn.messages.map((message) => message.text), bodies); + } + for (const message of turn.messages) await turn.disposition(message.delivery_id, "complete"); + }; + await f.control.accept(request("hold", "wait")); await until(() => f.calls.length === 1); + for (const [i, body] of bodies.entries()) await f.control.accept(request(`batch-${i}`, body)); + release(); + await until(async () => (await f.control.activityV2(token))!.items.filter((item) => item.state === "completed").length === 7); + assert.equal(f.calls.length, 2); + } finally { release(); await f.cleanup(); } +}); + +test("Unicode is counted in codepoints and a small or multibyte delivery stays inline", async () => { + const f = await fixture(); + try { + for (const text of ["small", "😀".repeat(1900)]) { + f.onTurn = async (event, turn) => { + assert.ok(event.text.includes(text)); assert.doesNotMatch(event.text, /selected payload exceeds/); + await turn.disposition(turn.messages[0]!.delivery_id, "complete"); + }; + await f.control.accept(request(`inline-${f.calls.length}`, text)); + await until(async () => (await f.control.activityV2(token))!.items.every((item) => item.state === "completed")); + } + assert.equal(f.calls.length, 2); + } finally { await f.cleanup(); } +}); + +for (const failure of ["rejected", "engine_failed"] as const) test(`${failure} remains visible after idle and restart, survives unrelated success, and clears on recovery`, async () => { + const f = await fixture(); let fail = true; + try { + f.reject = failure === "rejected"; + f.onTurn = async (_event, turn) => { + if (fail && turn.messages.some((message) => message.delivery_id === "broken")) throw new Error("provider unavailable Bearer secret-review-token"); + for (const message of turn.messages) await turn.disposition(message.delivery_id, "complete"); + }; + await f.control.accept(request("broken", "handle the delivery")); + await until(async () => (await f.control.availability(token))!.state === "paused"); + const availability = (await f.control.availability(token))!; + assert.match(availability.agents[0]!.error!, failure === "rejected" ? /invalid_request/ : /provider unavailable/); + assert.doesNotMatch(JSON.stringify(availability), /secret-review-token/); + const original = (await f.control.activityV2(token))!.items[0]!; + assert.equal(original.state, "accepted"); assert.equal(original.deferred, true); + assert.ok(original.execution_id); + const attempts = f.attempts; await pause(40); assert.equal(f.attempts, attempts); + await f.restart(); await pause(40); assert.equal(f.attempts, attempts); + assert.equal((await f.control.availability(token))!.state, "paused"); + // Let a different delivery complete while the failed delivery still fails. + f.reject = false; + await f.control.accept(request("unrelated", "other work")); + await until(async () => (await f.control.activityV2(token))!.items.some((item) => item.delivery_id === "unrelated" && item.state === "completed")); + assert.equal((await f.control.availability(token))!.state, "paused"); + assert.equal((await f.control.activityV2(token))!.items.find((item) => item.delivery_id === "broken")!.execution_id, original.execution_id); + // Durable diagnostics are redacted at rest, not only at the HTTP surface. + for (const file of await readdir(f.root)) if (file.endsWith(".json")) assert.doesNotMatch(await readFile(path.join(f.root, file), "utf8"), /secret-review-token/); + fail = false; await f.control.accept(request("retry", "new input")); + await until(async () => (await f.control.activityV2(token))!.items.every((item) => item.state === "completed")); + assert.equal((await f.control.availability(token))!.state, "running"); + assert.equal((await f.control.availability(token))!.agents[0]!.error, undefined); + } finally { await f.cleanup(); } +}); + +test("an agent can explicitly defer recovered work without leaving a runtime failure alarm", async () => { + const f = await fixture(); + try { + f.reject = true; + await f.control.accept(request("defer-later", "work")); + await until(async () => (await f.control.availability(token))!.state === "paused"); + f.reject = false; + f.onTurn = async (_event, turn) => { + for (const message of turn.messages) await turn.disposition(message.delivery_id, message.delivery_id === "defer-later" ? "defer" : "complete"); + }; + await f.control.accept(request("fresh", "new input")); + await until(async () => (await f.control.activityV2(token))!.items.some((item) => item.delivery_id === "fresh" && item.state === "completed")); + const availability = (await f.control.availability(token))!; + assert.equal(availability.state, "running"); assert.equal(availability.agents[0]!.error, undefined); + const receipt = (await f.control.activityV2(token))!.items.find((item) => item.delivery_id === "defer-later")!; + assert.equal(receipt.state, "accepted"); assert.equal(receipt.deferred, true); assert.equal(receipt.execution_id, undefined); + await f.restart(); + assert.equal((await f.control.availability(token))!.state, "running"); + } finally { await f.cleanup(); } +}); + + +test("availability stays answerable when durable work belongs to an old agent identity", async () => { + const f = await fixture(); + try { + f.reject = true; await f.control.accept(request("old", "work")); + await until(async () => (await f.control.availability(token))!.state === "paused"); + await until(async () => (await f.control.activityV2(token))!.executions!.length === 0); + const file = (await readdir(f.root)).find((name) => /^[0-9a-f]{64}\.json$/.test(name))!; + const record = JSON.parse(await readFile(path.join(f.root, file), "utf8")); + record.agent_id = "retired"; + record.request_digest = wakeAcceptanceDigest({ token: undefined, agent_id: record.agent_id, delivery_id: record.delivery_id, event: record.event }); + await writeFile(path.join(f.root, file), JSON.stringify(record), { mode: 0o600 }); + const status = await f.control.availability(token); + assert.ok(status); assert.equal(status.agents[0]!.pending, 0); + } finally { await f.cleanup(); } +}); diff --git a/src/runtime/attentionDispatcher.ts b/src/runtime/attentionDispatcher.ts index bc429e7..b1a8f77 100644 --- a/src/runtime/attentionDispatcher.ts +++ b/src/runtime/attentionDispatcher.ts @@ -5,6 +5,7 @@ import { engineFailureDetail } from "./organizationRuntimeHost.js"; import { WakeAcceptanceStore, WakeExecutionClaimLostError, type WakeExecutionClaim } from "./wakeAcceptanceStore.js"; import type { StoredWakeAcceptanceRecord } from "./wakeAcceptanceRecord.js"; import { WakeFuse } from "./wakeFuse.js"; +import { ORGANIZATION_RUNTIME_MAX_STRING_CODEPOINTS, ORGANIZATION_RUNTIME_MAX_WAKE_TEXT_BYTES } from "../contracts/organizationRuntimeContract.js"; type Claimed = { record: StoredWakeAcceptanceRecord; claim: WakeExecutionClaim; done: boolean }; type Options = Readonly<{ store: WakeAcceptanceStore; host: OrganizationRuntimeHost; fuse: WakeFuse; agents: readonly OrganizationRuntimeAgentConfig[]; registry: AttentionRegistry; token: string | undefined; onIdle(agentId: string): void }>; @@ -100,7 +101,7 @@ export class AttentionDispatcher { return; } if (item.done && disposition === "defer") return; - item.record = await store.transitionClaimed(item.record.acceptance_id, item.claim, disposition === "complete" ? "completed" : "accepted", undefined, disposition === "complete" ? "" : undefined, disposition === "defer" ? { deferred: true, clear_execution: true } : { execution_id: executionId, deferred: false }); + item.record = await store.transitionClaimed(item.record.acceptance_id, item.claim, disposition === "complete" ? "completed" : "accepted", undefined, disposition === "complete" ? "" : undefined, disposition === "defer" ? { deferred: true, clear_execution: true, execution_error: null } : { execution_id: executionId, deferred: false, execution_error: null }); item.done = true; }) }); @@ -117,12 +118,14 @@ export class AttentionDispatcher { } finally { clearInterval(heartbeat); } try { await mutation; + 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"); else if (agent.attention !== undefined) { // Successful reading is not completion. A failed execution also keeps // unfinished deliveries, and its execution id for idempotent retry. - await store.transitionClaimed(item.record.acceptance_id, item.claim, "accepted", undefined, undefined, { deferred: true, clear_execution: result.status === "completed" }); + await store.transitionClaimed(item.record.acceptance_id, item.claim, "accepted", undefined, undefined, { deferred: true, clear_execution: result.status === "completed", execution_error: executionError }); } else if (result.status === "completed") await store.transitionClaimed(item.record.acceptance_id, item.claim, "completed", undefined, result.text); else await store.transitionClaimed(item.record.acceptance_id, item.claim, "failed", "engine_failed", result.status === "failed" ? result.detail : undefined); } @@ -170,5 +173,13 @@ export function selectBatch(records: readonly StoredWakeAcceptanceRecord[], agen function inboxPrompt(messages: readonly unknown[], maxBytes = 12000): string { const body = JSON.stringify(messages); - return "Handle this inbox turn. Use daimon_inbox for deliveries and remaining allowances. Explicitly call daimon_inbox_disposition for each handled delivery (complete) or unfinished delivery (defer). Reading or ending this turn never completes a delivery. Deferred work waits for a later external wake.\n" + (Buffer.byteLength(body) > maxBytes ? "The selected payload exceeds the prompt budget; read it with daimon_inbox." : body); + const prefix = "Handle this inbox turn. Use daimon_inbox for deliveries and remaining allowances. Explicitly call daimon_inbox_disposition for each handled delivery (complete) or unfinished delivery (defer). Reading or ending this turn never completes a delivery. Deferred work waits for a later external wake.\n"; + const prompt = prefix + body; + // The inbox budget bounds selection; the v1 execution boundary independently + // bounds the complete prompt, including metadata, escaping, and instructions. + if (Buffer.byteLength(body) > maxBytes || Buffer.byteLength(prompt) > ORGANIZATION_RUNTIME_MAX_WAKE_TEXT_BYTES + || [...prompt].length > ORGANIZATION_RUNTIME_MAX_STRING_CODEPOINTS) { + return prefix + "The selected payload exceeds the prompt budget; read it with daimon_inbox."; + } + return prompt; } diff --git a/src/runtime/organizationRuntimeControl.ts b/src/runtime/organizationRuntimeControl.ts index e204269..e4d9c5c 100644 --- a/src/runtime/organizationRuntimeControl.ts +++ b/src/runtime/organizationRuntimeControl.ts @@ -140,13 +140,16 @@ function createControl(config: OrganizationRuntimeConfig, host: OrganizationRunt async availability(token) { if (!tokensEqual(expectedToken, token) || !store || !fuse) return undefined; await fuse.pollOperatorStop(); - const items = await store.activity(); + const items = await store.activityWithExecutionErrors(); + const executionErrors = new Map(items.filter((record) => (record.state === "accepted" || record.state === "running") + && record.execution_error !== undefined).map((record) => [record.agent_id, record.execution_error!])); const agents = await Promise.all(config.agents.map(async (agent) => ({ agent_id: agent.id, pending: items.filter((item) => item.agent_id === agent.id && (item.state === "accepted" || item.state === "running" && !dispatcher?.activeExecutions().some((execution) => execution.agent_id === agent.id))).length, running: dispatcher?.activeExecutions().some((execution) => execution.agent_id === agent.id) ?? false, deferred: items.filter((item) => item.agent_id === agent.id && item.state === "accepted" && item.deferred).length, budget: await fuse!.snapshot(agent.id, agent.attention), - ...(dispatcher?.failure(agent.id) ? { error: dispatcher.failure(agent.id) } : {}) + ...((dispatcher?.failure(agent.id) ?? executionErrors.get(agent.id)) + ? { error: dispatcher?.failure(agent.id) ?? executionErrors.get(agent.id) } : {}) }))); return { version: "noopolis.daimon.work-availability.v1", state: hardReason() ? "stopped" : agents.some((agent) => agent.budget.state !== "available" || agent.error) ? "paused" : "running", agents }; }, diff --git a/src/runtime/wakeAcceptanceRecord.test.ts b/src/runtime/wakeAcceptanceRecord.test.ts new file mode 100644 index 0000000..c46fe92 --- /dev/null +++ b/src/runtime/wakeAcceptanceRecord.test.ts @@ -0,0 +1,78 @@ +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { chmod, mkdtemp, readFile, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import test from "node:test"; +import { WakeAcceptanceStore } from "./wakeAcceptanceStore.js"; +import { parseStoredWakeAcceptance, sanitizeExecutionError } from "./wakeAcceptanceRecord.js"; +import { parseWakeAcceptanceRequest } from "./wakeAcceptanceTypes.js"; + +type PublicActivityRow = Awaited>[number]; +const publicViewOmitsExecutionError: "execution_error" extends keyof PublicActivityRow ? false : true = true; +void publicViewOmitsExecutionError; + +const storeOptions = process.platform === "linux" ? {} : { + processIdentity: async () => ({ pid: 1, process_start: "test-start", boot_id: "test-boot", pid_namespace_dev: 1, pid_namespace_ino: 1 }), + ownerLiveness: async () => true +}; +async function fixture() { + const root = await mkdtemp(path.join(os.tmpdir(), "daimon-error-record-")); await chmod(root, 0o700); + let store = await WakeAcceptanceStore.open(root, storeOptions); + const { record } = await store.accept(parseWakeAcceptanceRequest({ token: undefined, agent_id: "alpha", delivery_id: "test", + event: { version: "noopolis.daimon.wake.v2", kind: "message", text: "work", occurred_at: "2026-09-12T11:30:00.000Z" } })); + const acquired = await store.acquireClaim(record.acceptance_id, randomUUID(), [record.acceptance_id], randomUUID()); + assert.equal(acquired.state, "acquired"); if (acquired.state !== "acquired") throw new Error("claim missing"); + await store.transitionClaimed(record.acceptance_id, acquired.claim, "running"); + return { root, record, claim: acquired.claim, get store() { return store; }, + async restart() { await store.releaseClaim(acquired.claim); await store.close(); store = await WakeAcceptanceStore.open(root, storeOptions); }, + async cleanup() { await store.close(); await rm(root, { recursive: true, force: true }); } + }; +} + +test("empty execution diagnostics are omitted rather than writing an unreadable record", async () => { + const f = await fixture(); + try { + await f.store.transitionClaimed(f.record.acceptance_id, f.claim, "accepted", undefined, undefined, { deferred: true, execution_error: " \t\n " }); + await f.restart(); + const [row] = await f.store.recoverable(new Set(["alpha"])); + assert.equal(row!.execution_error, undefined); + assert.equal((await f.store.status(f.record.acceptance_id))!.state, "accepted"); + } finally { await f.cleanup(); } +}); + +test("large multibyte diagnostics remain bounded, redacted, and readable after restart", async () => { + const f = await fixture(); + try { + const original = "provider failed Bearer secret-review-token " + "€".repeat(2000); + await f.store.transitionClaimed(f.record.acceptance_id, f.claim, "accepted", undefined, undefined, { deferred: true, execution_error: original }); + await f.restart(); + const [row] = await f.store.recoverable(new Set(["alpha"])); + assert.ok(Buffer.byteLength(row!.execution_error!) <= 2048); + assert.match(row!.execution_error!, /provider failed/); assert.doesNotMatch(row!.execution_error!, /secret-review-token|\uFFFD/); + assert.equal(row!.execution_error, sanitizeExecutionError(original)); + const { readdir } = await import("node:fs/promises"); + for (const file of await readdir(f.root)) if (file.endsWith(".json")) assert.doesNotMatch(await readFile(path.join(f.root, file), "utf8"), /secret-review-token/); + const [publicRow] = await f.store.activity(); assert.equal(Object.hasOwn(publicRow!, "execution_error"), false); + } finally { await f.cleanup(); } +}); + +test("diagnostics are re-sanitized on read without invalidating otherwise valid stored work", async () => { + const f = await fixture(); + try { + const parsed = parseStoredWakeAcceptance({ ...f.record, execution_error: "provider failed Bearer secret-review-token" }); + assert.match(parsed.execution_error!, /provider failed/); assert.doesNotMatch(parsed.execution_error!, /secret-review-token/); + assert.throws(() => parseStoredWakeAcceptance({ ...f.record, execution_error: "x".repeat(2049) }), /execution error/); + } finally { await f.cleanup(); } +}); + +test("running and shutdown transitions preserve an unresolved execution diagnostic", async () => { + const f = await fixture(); + try { + await f.store.transitionClaimed(f.record.acceptance_id, f.claim, "accepted", undefined, undefined, { deferred: true, execution_error: "engine_failed: unavailable" }); + await f.store.transitionClaimed(f.record.acceptance_id, f.claim, "running", undefined, undefined, { deferred: false }); + await f.store.transitionClaimed(f.record.acceptance_id, f.claim, "accepted"); + await f.restart(); + assert.equal((await f.store.recoverable(new Set(["alpha"])))[0]!.execution_error, "engine_failed: unavailable"); + } finally { await f.cleanup(); } +}); diff --git a/src/runtime/wakeAcceptanceRecord.ts b/src/runtime/wakeAcceptanceRecord.ts index 4bfe5bd..ddf544b 100644 --- a/src/runtime/wakeAcceptanceRecord.ts +++ b/src/runtime/wakeAcceptanceRecord.ts @@ -1,3 +1,4 @@ +import { redactCredentialText } from "../core/credentialRedaction.js"; import { WAKE_ACCEPTANCE_VERSION, WAKE_RECEIPT_STATUS_VERSION, @@ -14,7 +15,7 @@ import { export type StoredWakeAcceptanceRecord = Readonly<{ acceptance_id: string; agent_id: string; delivery_id: string; request_digest: string; event: OrganizationRuntimeWakeAcceptanceRequest["event"]; state: WakeReceiptState; - accepted_at: string; updated_at: string; claim_generation?: string; execution_id?: string; deferred?: boolean; code?: WakeReceiptCode; text?: string; + accepted_at: string; updated_at: string; claim_generation?: string; execution_id?: string; deferred?: boolean; execution_error?: string; code?: WakeReceiptCode; text?: string; }>; export function publicAcceptance(record: StoredWakeAcceptanceRecord): OrganizationRuntimeWakeAcceptance { @@ -26,7 +27,7 @@ export function publicStatus(record: StoredWakeAcceptanceRecord): OrganizationRu export function parseStoredWakeAcceptance(value: unknown): StoredWakeAcceptanceRecord { if (value === null || typeof value !== "object" || Array.isArray(value)) throw new Error("wake acceptance record is invalid"); const record = value as Record; - const keys = ["acceptance_id", "agent_id", "delivery_id", "request_digest", "event", "state", "accepted_at", "updated_at", "claim_generation", "execution_id", "deferred", "code", "text"]; + const keys = ["acceptance_id", "agent_id", "delivery_id", "request_digest", "event", "state", "accepted_at", "updated_at", "claim_generation", "execution_id", "deferred", "execution_error", "code", "text"]; if (Object.keys(record).some((key) => !keys.includes(key))) throw new Error("wake acceptance record is invalid"); const parsed = parseWakeAcceptanceRequest({ token: undefined, agent_id: string(record.agent_id), delivery_id: string(record.delivery_id), event: record.event }); const state = string(record.state) as WakeReceiptState; @@ -35,6 +36,8 @@ export function parseStoredWakeAcceptance(value: unknown): StoredWakeAcceptanceR const claimGeneration = record.claim_generation === undefined ? undefined : string(record.claim_generation); const executionId = record.execution_id === undefined ? undefined : string(record.execution_id); if (executionId !== undefined && !uuid(executionId) || record.deferred !== undefined && typeof record.deferred !== "boolean") throw new Error("wake acceptance attention state is invalid"); + const executionError = record.execution_error === undefined ? undefined : sanitizeExecutionError(string(record.execution_error)); + 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"); @@ -42,8 +45,14 @@ export function parseStoredWakeAcceptance(value: unknown): StoredWakeAcceptanceR 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"); - return { acceptance_id: string(record.acceptance_id), agent_id: parsed.agent_id, delivery_id: parsed.delivery_id, request_digest: string(record.request_digest), event: parsed.event, state, accepted_at: timestamp(record.accepted_at), updated_at: timestamp(record.updated_at), ...(claimGeneration === undefined ? {} : { claim_generation: claimGeneration }), ...(executionId === undefined ? {} : { execution_id: executionId }), ...(record.deferred === undefined ? {} : { deferred: record.deferred as boolean }), ...(code === undefined ? {} : { code }), ...(completionText === undefined ? {} : { text: completionText }) }; + return { acceptance_id: string(record.acceptance_id), agent_id: parsed.agent_id, delivery_id: parsed.delivery_id, request_digest: string(record.request_digest), event: parsed.event, state, accepted_at: timestamp(record.accepted_at), updated_at: timestamp(record.updated_at), ...(claimGeneration === undefined ? {} : { claim_generation: claimGeneration }), ...(executionId === undefined ? {} : { execution_id: executionId }), ...(record.deferred === undefined ? {} : { deferred: record.deferred as boolean }), ...(code === undefined ? {} : { code }), ...(completionText === undefined ? {} : { text: completionText }), ...(executionError === undefined ? {} : { execution_error: executionError }) }; } function string(value: unknown): string { if (typeof value !== "string") throw new Error("wake acceptance record is invalid"); return value; } function timestamp(value: unknown): string { const result = string(value); if (Number.isNaN(Date.parse(result)) || new Date(result).toISOString() !== result) throw new Error("wake acceptance record is invalid"); return result; } function uuid(value: string): boolean { return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/iu.test(value); } + +const MAX_EXECUTION_ERROR_BYTES = 2048; +/** Private failure diagnostic, surfaced through the existing availability error. */ +export function sanitizeExecutionError(value: string): string { + return redactCredentialText(value, [], MAX_EXECUTION_ERROR_BYTES).trim(); +} diff --git a/src/runtime/wakeAcceptanceStore.ts b/src/runtime/wakeAcceptanceStore.ts index fc62d95..bad2dd6 100644 --- a/src/runtime/wakeAcceptanceStore.ts +++ b/src/runtime/wakeAcceptanceStore.ts @@ -13,11 +13,12 @@ import { type WakeReceiptState, wakeAcceptanceDigest } from "./wakeAcceptanceTypes.js"; -import { parseStoredWakeAcceptance, publicAcceptance, publicStatus, type StoredWakeAcceptanceRecord } from "./wakeAcceptanceRecord.js"; +import { sanitizeExecutionError, parseStoredWakeAcceptance, publicAcceptance, publicStatus, type StoredWakeAcceptanceRecord } from "./wakeAcceptanceRecord.js"; import { assertOfflineReconciliationLeaseAvailable } from "./wakeAcceptanceReconciliation.js"; import { acquireHostRegistration, releaseHostRegistration, type StoreHostRegistration } from "./storeCoordination.js"; import { MAX_WAKE_ACCEPTANCE_RECORDS, terminalFilesToCompact } from "./wakeAcceptanceRetention.js"; type Stored = StoredWakeAcceptanceRecord; +type ActivityRow = OrganizationRuntimeWakeReceiptStatus & Readonly<{ active: boolean; queue_position?: number; execution_error?: string }>; export type WakeExecutionClaim = Readonly<{ acceptance_id: string; owner_id: string; generation: string; expires_at: string; acceptance_ids?: readonly string[]; execution_id?: string }>; export type WakeExecutionClaimResult = Readonly<{ state: "acquired"; claim: WakeExecutionClaim }> | Readonly<{ state: "held"; retry_at: string }> | Readonly<{ state: "terminal" }>; type DirectoryIdentity = Readonly<{ dev: number; ino: number; uid: number; mode: number }>; @@ -88,7 +89,10 @@ export class WakeAcceptanceStore { const record = await this.findByAcceptanceId(acceptanceId); return record === undefined ? undefined : publicStatus(record); } - async activity(): Promise)[]> { + async activity(): Promise[]> { return await this.activityRows(false); } + /** Internal availability view; diagnostics never enter the public receipt wire. */ + async activityWithExecutionErrors(): Promise { return await this.activityRows(true); } + private async activityRows(includeExecutionErrors: boolean): Promise { const records = await Promise.all((await this.files()).map(async (file) => await this.read(path.join(this.root, file)))); records.sort((left, right) => left.accepted_at.localeCompare(right.accepted_at) || left.acceptance_id.localeCompare(right.acceptance_id)); const queued = new Map(); @@ -98,7 +102,8 @@ export class WakeAcceptanceStore { if (position !== undefined) queued.set(record.agent_id, position); const active = record.state === "running" && !activeExecutions.has(record.agent_id); if (active) activeExecutions.add(record.agent_id); - return { ...publicStatus(record), active, ...(position === undefined ? {} : { queue_position: position }) }; + return { ...publicStatus(record), active, ...(position === undefined ? {} : { queue_position: position }), + ...(includeExecutionErrors && record.execution_error !== undefined ? { execution_error: record.execution_error } : {}) }; }); } async recoverable(agentIds: ReadonlySet): Promise { @@ -174,7 +179,7 @@ export class WakeAcceptanceStore { }); } /** `text` carries the completion output on success and the engine failure cause on failure. */ - transitionClaimed(acceptanceId: string, claim: WakeExecutionClaim, state: WakeReceiptState, code?: WakeReceiptCode, completedText?: string, attention?: { execution_id?: string; deferred?: boolean; clear_execution?: boolean }): Promise { + transitionClaimed(acceptanceId: string, claim: WakeExecutionClaim, state: WakeReceiptState, code?: WakeReceiptCode, completedText?: string, attention?: { execution_id?: string; deferred?: boolean; clear_execution?: boolean; execution_error?: string | null }): Promise { return this.serialize(async () => { if (completedText !== undefined && state !== "completed" && state !== "failed") throw new Error("wake text requires completed or failed state"); const initial = await this.findByAcceptanceId(acceptanceId); @@ -193,7 +198,7 @@ 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 }) }; + 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 }) }; await this.replace(target, next); if (claim.acceptance_ids === undefined && (isTerminal(state) || state === "accepted")) { const currentClaim = await this.readClaimOptional(this.claimFor(record));