fix: prevent silent deferral of oversized inbox turns

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:
 

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<void>((resolve) => setTimeout(resolve, ms));
+async function until(check: () => boolean | Promise<boolean>) {
+  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<void>((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(); }
+});

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;
 }

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 };
     },