diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 49fa481f4eb..a9ce3e931f0 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1486,6 +1486,51 @@ it("refuses a terminal call that arrives before a held call's result (restart ch } }, 20_000); +it("refuses a finish once for a returned call that never reaches the agent, then accepts the next finish", async () => { + const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-unseen-once-")); + const OriginalCore = durableControlPlane.DurablePrpControlPlane; + let callbacks!: ConstructorParameters[0]; + let core!: DurablePrpControlPlane; + const coreSpy = vi.spyOn(durableControlPlane, "DurablePrpControlPlane").mockImplementation(function (options) { + callbacks = options; + return core = new OriginalCore(options); + }); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(stateDirectory), stateDirectory, + }); + const handler = vi.fn(async () => ({ success: true, contentItems: [{ type: "inputText" as const, text: "{\"done\":true}" }] })); + bundle.transport.setServerRequestHandler(handler); + try { + await bundle.transport.request("thread/start", { cwd: tmpdir() }); + const identity = core.store.state.identity; + let seq = 9200; + const commit = (eventType: string, payload: Record) => { + seq += 1; + core.store.state.committedEvents.push({ sourceSeq: seq, sourceEventId: `unseen-once-${seq}`, eventType, priority: 1, + envelope: { payload: { eventType, turnId: identity.turnId, payload } }, deliveryCount: 1, logicalEffectCount: 1 }); + return `unseen-once-${seq}`; + }; + commit("tool.execution.started", { executionId: "exec-search", transport: "dynamic", name: "paperclip_search_assigned_tools", status: "running" }); + core.queueCommand("semantic_tool.result", { callId: "exec-search", operationId: "paperclip_search_assigned_tools", + inputDigest: "sha256:search", result: { tools: [] }, isError: false }, "command_tool_unseen_once_search", true); + const finish = (n: number) => { + const sourceEventId = commit("tool.execution.started", { executionId: `exec-finish-${n}`, transport: "dynamic", name: "paperclip_finish", status: "running" }); + return callbacks.onSemanticToolInput!({ callId: `exec-finish-${n}`, operationId: "paperclip_finish", + input: { summary: "Done" }, sourceEventId, sourceEventType: "semantic_tool.input", correlation: identity }); + }; + expect(await finish(1)).toMatchObject({ isError: true, result: { error: { code: "terminal_call_before_results" } } }); + expect(handler).not.toHaveBeenCalled(); + commit("tool.execution.completed", { executionId: "exec-finish-1", transport: "dynamic", name: "paperclip_finish", status: "failed" }); + expect((await finish(2)).isError).not.toBe(true); + expect(handler).toHaveBeenCalledTimes(1); + } finally { + coreSpy.mockRestore(); + await bundle.transport.close().catch(() => undefined); + await rm(stateDirectory, { recursive: true, force: true }); + } +}, 20_000); + it("answers a held call before starting the next turn (restart chaos t101)", async () => { const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-held-before-turn-")); const OriginalCore = durableControlPlane.DurablePrpControlPlane; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index e28ca0fa2c1..130cbc2878c 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -3941,6 +3941,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ], }); #handlerInstalled = false; + readonly #terminalCallRefusals = new Map(); readonly #handlerReady: Promise; #resolveHandlerReady!: () => void; #core: DurablePrpControlPlane | null = null; @@ -7288,12 +7289,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // (restart chaos t05) before judging what the agent had seen. const settled = await settleEarlierToolCalls( () => this.#closed || core !== this.#core ? null : { events: core.store.state.committedEvents, commands: core.store.state.commands }, - { terminalSourceEventId: call.sourceEventId, turnId: call.correlation.turnId }, + { terminalSourceEventId: call.sourceEventId, turnId: call.correlation.turnId, refusals: this.#terminalCallRefusals }, ); if (!settled || this.#closed || core !== this.#core || epoch !== this.#turnStartResponseEpoch || threadId !== this.#threadId) { throw new SemanticToolNotDispatchedError(); } if (settled.unseen.length || settled.pending.length) { + this.#terminalCallRefusals.set(call.callId, settled.unseenCallIds); return staleTerminalCallResult(call.operationId, settled.unseen, settled.pending); } } diff --git a/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts b/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts index 6fe1cb72eda..0931458021f 100644 --- a/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts +++ b/packages/paperclip-runner/src/live/terminal-call-freshness.test.ts @@ -53,6 +53,50 @@ it("forgets calls of a provider turn lost on restore once a new turn starts (res expect(unseenReturnedToolCalls(events, commands, { terminalSourceEventId: finish.sourceEventId, turnId: "turn-1" })).toEqual([]); }); +describe("a call whose result never reaches the agent (two staging Codex runs, 2026-11-11)", () => { + const commands = [{ type: "semantic_tool.result", payload: { callId: "s", operationId: "paperclip_search_assigned_tools" } }]; + + it("names the returned call once; a finish decided after that refusal is judged without it", async () => { + const first = started("f1", "paperclip_finish"); + const second = started("f2", "paperclip_finish"); + const events = [ + event("turn.started", {}), started("s", "paperclip_search_assigned_tools"), + started("r", "read_document"), result("r"), completed("r", "read_document"), + first, completed("f1", "paperclip_finish"), second, + ]; + const refused = await settleEarlierToolCalls(() => ({ events, commands }), { terminalSourceEventId: first.sourceEventId, turnId: "turn-1" }); + expect(refused).toEqual({ unseen: ["paperclip_search_assigned_tools"], unseenCallIds: ["s"], pending: [] }); + const refusals = new Map([["f1", refused!.unseenCallIds]]); + expect(unseenReturnedToolCalls(events, commands, { terminalSourceEventId: second.sourceEventId, turnId: "turn-1" })) + .toEqual(["paperclip_search_assigned_tools"]); + expect(unseenReturnedToolCalls(events, commands, { terminalSourceEventId: second.sourceEventId, turnId: "turn-1", refusals })) + .toEqual([]); + }); + + it("still refuses a terminal call queued before the agent received the refusal", () => { + const second = started("f2", "paperclip_block"); + const events = [event("turn.started", {}), started("s", "paperclip_search_assigned_tools"), started("f1", "paperclip_finish"), + second, completed("f1", "paperclip_finish")]; + expect(unseenReturnedToolCalls(events, commands, { terminalSourceEventId: second.sourceEventId, turnId: "turn-1", + refusals: new Map([["f1", ["s"]]]) })).toEqual(["paperclip_search_assigned_tools"]); + }); + + it("still refuses for a write that was pending at the earlier refusal and returned unseen during the next wait", async () => { + const second = started("f2", "paperclip_finish"); + const events = [event("turn.started", {}), started("w", "write_document"), started("f1", "paperclip_finish"), + completed("f1", "paperclip_finish"), second]; + const writeCommands: { type: string; payload: Record }[] = []; + const first = await settleEarlierToolCalls(() => ({ events, commands: writeCommands }), + { terminalSourceEventId: events[2]!.sourceEventId, turnId: "turn-1" }, { timeoutMs: 20, pollMs: 5 }); + expect(first).toMatchObject({ unseen: [], unseenCallIds: [], pending: ["write_document"] }); + const refusals = new Map([["f1", first!.unseenCallIds]]); + setTimeout(() => writeCommands.push({ type: "semantic_tool.result", payload: { callId: "w", operationId: "write_document" } }), 20); + const next = await settleEarlierToolCalls(() => ({ events, commands: writeCommands }), + { terminalSourceEventId: second.sourceEventId, turnId: "turn-1", refusals }, { timeoutMs: 5_000, pollMs: 5 }); + expect(next).toEqual({ unseen: ["write_document"], unseenCallIds: ["w"], pending: [] }); + }); +}); + it("counts the controller's semantic_tool.result command as the returned result", () => { const block = started("b", "paperclip_block"); const commands = [{ type: "semantic_tool.result", payload: { callId: "w", operationId: "write_document" } }]; @@ -78,7 +122,7 @@ describe("settleEarlierToolCalls", () => { setTimeout(() => commands.push({ type: "semantic_tool.result", payload: { callId: "w", operationId: "write_document" } }), 30); const settled = await settleEarlierToolCalls(() => ({ events, commands }), { terminalSourceEventId: block.sourceEventId }, { timeoutMs: 5_000, pollMs: 5 }); - expect(settled).toEqual({ unseen: ["write_document"], pending: [] }); + expect(settled).toEqual({ unseen: ["write_document"], unseenCallIds: ["w"], pending: [] }); }); it("reports a call still pending after the bound", async () => { @@ -86,7 +130,7 @@ describe("settleEarlierToolCalls", () => { const events = [started("w", "write_document"), block]; const settled = await settleEarlierToolCalls(() => ({ events, commands: [] }), { terminalSourceEventId: block.sourceEventId }, { timeoutMs: 20, pollMs: 5 }); - expect(settled).toEqual({ unseen: [], pending: ["write_document"] }); + expect(settled).toEqual({ unseen: [], unseenCallIds: [], pending: ["write_document"] }); const outcome = staleTerminalCallResult("paperclip_block", settled!.unseen, settled!.pending); expect(outcome).toMatchObject({ isError: true, result: { error: { code: "terminal_call_before_results", retryable: true } } }); expect(outcome.result.error.message).toContain("has not returned yet"); @@ -97,7 +141,7 @@ describe("settleEarlierToolCalls", () => { const events = [started("w", "write_document"), result("w"), completed("w", "write_document"), block]; const startedAt = Date.now(); expect(await settleEarlierToolCalls(() => ({ events, commands: [] }), { terminalSourceEventId: block.sourceEventId })) - .toEqual({ unseen: [], pending: [] }); + .toEqual({ unseen: [], unseenCallIds: [], pending: [] }); expect(Date.now() - startedAt).toBeLessThan(1_000); expect(await settleEarlierToolCalls(() => null, {})).toBeNull(); }); diff --git a/packages/paperclip-runner/src/live/terminal-call-freshness.ts b/packages/paperclip-runner/src/live/terminal-call-freshness.ts index 0d8ce22cf9d..f6db8555a6f 100644 --- a/packages/paperclip-runner/src/live/terminal-call-freshness.ts +++ b/packages/paperclip-runner/src/live/terminal-call-freshness.ts @@ -38,8 +38,12 @@ type Command = { type: string; payload: Record }; const record = (value: unknown): Record => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; -type TerminalCallScope = { terminalSourceEventId?: string | null; turnId?: string | null }; -type ToolCallState = { name: string; returned: boolean; seen: boolean }; +type TerminalCallScope = { + terminalSourceEventId?: string | null; + turnId?: string | null; + refusals?: ReadonlyMap; +}; +type ToolCallState = { name: string; returned: boolean; seen: boolean; named: boolean }; /** Names of calls that returned before the terminal call but that the agent had not yet seen. */ export function unseenReturnedToolCalls( @@ -47,7 +51,12 @@ export function unseenReturnedToolCalls( commands: readonly Command[], input: TerminalCallScope, ): string[] { - return callNames(earlierToolCalls(events, commands, input), call => call.returned && !call.seen); + return callNames(unseenReturnedCalls(events, commands, input)); +} + +function unseenReturnedCalls(events: readonly CommittedEvent[], commands: readonly Command[], input: TerminalCallScope) { + return [...earlierToolCalls(events, commands, input)] + .filter(([, call]) => call.returned && !call.seen && !call.named); } /** @@ -62,11 +71,11 @@ export function pendingEarlierToolCalls( commands: readonly Command[], input: TerminalCallScope, ): string[] { - return callNames(earlierToolCalls(events, commands, input), call => !call.returned && !call.seen); + return callNames([...earlierToolCalls(events, commands, input)].filter(([, call]) => !call.returned && !call.seen)); } -function callNames(calls: Map, keep: (call: ToolCallState) => boolean): string[] { - return [...new Set([...calls.values()].filter(call => keep(call) && !TERMINAL_OPERATIONS.has(call.name)).map(call => call.name))]; +function callNames(calls: [string, ToolCallState][]): string[] { + return [...new Set(calls.filter(([, call]) => !TERMINAL_OPERATIONS.has(call.name)).map(([, call]) => call.name))]; } function earlierToolCalls( @@ -90,8 +99,14 @@ function earlierToolCalls( if ((type === "tool.execution.started" || type === "tool.execution.completed") && payload.transport === "dynamic") { const id = typeof payload.executionId === "string" ? payload.executionId : null; if (!id) continue; - const call = calls.get(id) ?? { name: typeof payload.name === "string" ? payload.name : "tool", returned: false, seen: false }; - if (type === "tool.execution.completed") call.seen = true; + const call = calls.get(id) ?? { name: typeof payload.name === "string" ? payload.name : "tool", returned: false, seen: false, named: false }; + if (type === "tool.execution.completed") { + call.seen = true; + for (const named of input.refusals?.get(id) ?? []) { + const earlier = calls.get(named); + if (earlier) earlier.named = true; + } + } calls.set(id, call); continue; } @@ -118,14 +133,15 @@ export async function settleEarlierToolCalls( read: () => { events: readonly CommittedEvent[]; commands: readonly Command[] } | null, input: TerminalCallScope, options: { timeoutMs?: number; pollMs?: number } = {}, -): Promise<{ unseen: string[]; pending: string[] } | null> { +): Promise<{ unseen: string[]; unseenCallIds: string[]; pending: string[] } | null> { const deadline = Date.now() + (options.timeoutMs ?? TERMINAL_CALL_SETTLE_TIMEOUT_MS); for (;;) { const state = read(); if (!state) return null; const pending = pendingEarlierToolCalls(state.events, state.commands, input); if (pending.length === 0 || Date.now() >= deadline) { - return { unseen: unseenReturnedToolCalls(state.events, state.commands, input), pending }; + const unseen = unseenReturnedCalls(state.events, state.commands, input); + return { unseen: callNames(unseen), unseenCallIds: unseen.filter(([, call]) => !TERMINAL_OPERATIONS.has(call.name)).map(([id]) => id), pending }; } await new Promise(resolve => setTimeout(resolve, options.pollMs ?? 50)); } diff --git a/server/src/__tests__/runtime-skill-preservation.test.ts b/server/src/__tests__/runtime-skill-preservation.test.ts index d0546257902..2539879ce35 100644 --- a/server/src/__tests__/runtime-skill-preservation.test.ts +++ b/server/src/__tests__/runtime-skill-preservation.test.ts @@ -4,13 +4,13 @@ import os from "node:os"; import path from "node:path"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { activityLog, agents, companies, heartbeatRuns, issues, companySkillRepositorySnapshots as snapshots, +import { activityLog, agentInstructionRevisions, agents, companies, heartbeatRuns, issues, companySkillRepositorySnapshots as snapshots, nativeRunFinalizations, createDb } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import * as storageModule from "../storage/index.js"; import { createStorageService } from "../storage/service.js"; import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; -import { materializeAsset, rehydratePinnedSkillAssets } from "../services/native-runtime/runtime-context.js"; +import { materializeAsset, rehydratePinnedInstructionAsset, rehydratePinnedSkillAssets } from "../services/native-runtime/runtime-context.js"; import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js"; import { canonicalNativeRuntimeContextDigest, parseNativeExecutionInput } from "../vendor/paperclip-runner/index.js"; import { preserveRuntimeSkillAssets, hasLocalRuntimeSkillObligations, migrateRuntimeSkillAssets } from "../services/native-runtime/runtime-skill-preservation.js"; @@ -241,4 +241,26 @@ describe("admitted unversioned runtime skill preservation", () => { expect(await db.select().from(snapshots).where(and(eq(snapshots.companyId, f.companyId), eq(snapshots.id, snapshot!.id)))) .toHaveLength(1); }); + + it("rebuilds an admitted instruction bundle that is missing on a new controller host", async () => { + const companyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Instruction rehydration", issuePrefix: `I${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Rehydrated" }); + const admitted = Buffer.from("# Admitted instructions\n"); + for (const content of [admitted, Buffer.from("# Edited after admission\n")]) { + await db.insert(agentInstructionRevisions).values({ companyId, agentId, entryFile: "AGENTS.md", contentBase64: content.toString("base64"), + contentHash: "unused", byteLength: content.byteLength, source: "board" }); + } + const bundle = await materializeAsset([{ path: "AGENTS.md", content: admitted, mode: 0o444 }]); + const context = nativeRuntimeContextFixture(); + context.instructions = { entryPath: "AGENTS.md", bundle }; + await removeRuntimeAssetTree(bundle.rootPath); + await fs.rm(path.join(home, "instances", "default", "runtime-context-assets", "manifests", `${bundle.digest}.json`), { force: true }); + await expect(fs.lstat(bundle.rootPath)).rejects.toMatchObject({ code: "ENOENT" }); + const rehydrated = await rehydratePinnedInstructionAsset(db, { companyId, agentId }, context); + expect(rehydrated.instructions.bundle).toEqual(bundle); + expect(await fs.readFile(path.join(bundle.rootPath, "AGENTS.md"), "utf8")).toBe("# Admitted instructions\n"); + const elsewhere = { ...context, instructions: { ...context.instructions, bundle: { ...bundle, rootPath: "/other-home/runtime-context-assets/bundles/x" } } }; + expect(await rehydratePinnedInstructionAsset(db, { companyId, agentId }, elsewhere)).toBe(elsewhere); + }); }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6c7b4b5685d..95019c2a012 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -25,7 +25,7 @@ import { assertNativeHarnessBackupOwner, durableNativeHarnessStorage, NATIVE_HAR import { restoreNativeProviderArtifacts, retainNativeProviderArtifacts, retainExistingNativeProviderArtifacts, type NativeProviderArtifactReceipt } from "./native-provider-artifacts.js"; import { prepareRemoteOpenCodeArtifacts, type OpenCodeLaunchArtifacts } from "./remote-opencode-artifacts.js"; -import { rehydratePinnedSkillAssets } from "./runtime-context.js"; +import { rehydratePinnedInstructionAsset, rehydratePinnedSkillAssets } from "./runtime-context.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; import { createNativeRecoveryUsageDelivery } from "./native-recovery-usage-delivery.js"; import { createNativeTurnAccounting, NATIVE_LOST_TURN_USAGE_EVIDENCE, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent, isNativeUsageDeliveryEvent } from "./native-usage-accounting.js"; @@ -15220,6 +15220,9 @@ async function createRunnerdBackendWithinSessionClaim( retainedTransition?: VerifiedWarmTransitionBinding, ): Promise { const timeController = createRunPreparationTiming(input.execution.binding.runId, "native.controller"); + if ("runtimeContext" in input.execution) { + await rehydratePinnedInstructionAsset(input.db, input.execution.binding, input.execution.runtimeContext).catch(() => undefined); + } let recoveryPending = retainedTransition !== undefined; const target = input.runnerExecutionTarget ?? { kind: "local" as const }; const remoteTarget = target.kind === "remote" ? target : null; diff --git a/server/src/services/native-runtime/runtime-context.ts b/server/src/services/native-runtime/runtime-context.ts index 4426ce58990..e69381a677e 100644 --- a/server/src/services/native-runtime/runtime-context.ts +++ b/server/src/services/native-runtime/runtime-context.ts @@ -11,8 +11,8 @@ import { restConnectorFor } from "../connectors/rest-registry.js"; import { createHash, randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; -import { and, eq } from "drizzle-orm"; -import { companySkillVersions, heartbeatRuns, type Db } from "@paperclipai/db"; +import { and, desc, eq } from "drizzle-orm"; +import { agentInstructionRevisions, agents, companySkillVersions, heartbeatRuns, type Db } from "@paperclipai/db"; import type { PaperclipSkillEntry } from "@paperclipai/adapter-utils/server-utils"; import { isToolConnectionAttentionHealth } from "@paperclipai/shared"; import { @@ -199,6 +199,51 @@ async function materializeSkillAsset(source: string) { return bundle; } +async function localRuntimeAssetPresent(asset: NativeRuntimeAssetReference) { + const root = path.join(resolvePaperclipInstanceRoot(), "runtime-context-assets"); + const manifest = await fs.readFile(path.join(root, "manifests", `${asset.digest}.json`), "utf8").catch(() => null); + return manifest !== null && sha256(manifest) === asset.manifestDigest + && (await fs.lstat(path.join(root, "bundles", asset.digest)).catch(() => null))?.isDirectory() === true; +} + +function singleFileAssetDigest(entryPath: string, content: Buffer) { + return sha256(JSON.stringify([{ path: entryPath, sha256: sha256(content), mode: 0o444, size: content.byteLength }])); +} + +export async function rehydratePinnedInstructionAsset( + db: Db, + binding: { companyId: string; agentId: string }, + context: NativeRuntimeContextSnapshot, +): Promise { + const pinned = context.instructions.bundle; + if (pinned.rootPath !== path.join(resolvePaperclipInstanceRoot(), "runtime-context-assets", "bundles", pinned.digest)) return context; + if (await localRuntimeAssetPresent(pinned)) return context; + const matches = (bundle: NativeRuntimeAssetReference) => bundle.digest === pinned.digest + && bundle.manifestDigest === pinned.manifestDigest && bundle.fileCount === pinned.fileCount && bundle.totalBytes === pinned.totalBytes; + if (pinned.fileCount === 1) { + const rows = await db.select({ contentBase64: agentInstructionRevisions.contentBase64 }).from(agentInstructionRevisions) + .where(and(eq(agentInstructionRevisions.companyId, binding.companyId), eq(agentInstructionRevisions.agentId, binding.agentId), + eq(agentInstructionRevisions.entryFile, context.instructions.entryPath))) + .orderBy(desc(agentInstructionRevisions.createdAt)).limit(500); + for (const row of rows) { + const content = Buffer.from(row.contentBase64, "base64"); + if (singleFileAssetDigest(context.instructions.entryPath, content) !== pinned.digest) continue; + const bundle = await materializeAsset([{ path: context.instructions.entryPath, content, mode: 0o444 }]); + if (matches(bundle)) return { ...context, instructions: { ...context.instructions, bundle } }; + } + } + const [agent] = await db.select().from(agents).where(and(eq(agents.companyId, binding.companyId), eq(agents.id, binding.agentId))); + if (agent) { + for (const agentFiles of [true, false]) { + const rebuilt = await materializeInstructionBundle(db, agent, agentFiles).catch(() => null); + if (rebuilt && rebuilt.entryPath === context.instructions.entryPath && matches(rebuilt.bundle)) { + return { ...context, instructions: { ...context.instructions, bundle: rebuilt.bundle } }; + } + } + } + return context; +} + export async function rehydratePinnedSkillAssets( db: Db, binding: { companyId: string; runId: string; agentId: string }, diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index a8e06c3219a..e1d815cf974 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -752,6 +752,24 @@ A live turn is seen only when a poll (`--poll-ms`, 2–5 s) lands inside it, so poll interval can still be missed. A runner replacement or stop clears the live-turn mark, as it does for an open tool. +**A tool result that never reaches the agent (`tool_completed`, Codex only).** +`--provider-fault-message tool_completed` drops (PV2), holds (PV1) or repeats (PV3) Codex's +`item/completed` for the turn's first Paperclip tool call other than `paperclip_finish` and +`paperclip_block`. The model has the result, but the runner never records that it reached the model, +as when Codex cancels the nested exec cell that made the call. In two staging Codex runs +(2026-11-11) every later `paperclip_finish` was then refused with +`terminal_call_before_results`, the turn ended with no semantic result, and the same-run recovery +failed the run. No turn is interrupted, so the workloads are those that save a document and then +finish (A1, C1): + +```sh +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex --workloads A1 --faults PV2 --provider-fault-message tool_completed --concurrency 1 --tasks 1 --seed 301 +``` + +The task passes when the fault applied once and the ordinary A1 grade passes (the brief is saved and +the task is `done`). + | Adapter | `terminal` | `interrupt_ack` | | --- | --- | --- | | Codex | the `turn/completed` after a `turn/interrupt` | the response to that `turn/interrupt` | diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index 04856f7d269..f4b29526727 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -11,7 +11,7 @@ import { parseStorageTrigger } from "./storage-trigger.js"; import { parseNativeTrigger, parseNativeTriggerOutage } from "./native-trigger.js"; import { parseRepeatedRestartOptions } from "./repeated-restarts.js"; import { daytonaRetainsSandboxes } from "./ttft-baseline-stats.js"; -import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, isProviderFault, parseProviderDelays, parseProviderMessages, type ProviderFaultKind } from "./provider-faults.js"; +import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, TOOL_COMPLETED_WORKLOADS, UNINTERRUPTED_PROVIDER_MESSAGES, isProviderFault, parseProviderDelays, parseProviderMessages, type ProviderFaultKind } from "./provider-faults.js"; import { PROVIDER_FAULT_ADAPTERS, parseProviderFaultRestart } from "./acpx-provider-faults.js"; import { DISK_FULL_FAULT_KINDS, isDiskFullFault, type DiskFullFaultKind } from "./disk-full-faults.js"; import { @@ -233,10 +233,18 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC if (adapters.some(adapter => !PROVIDER_FAULT_ADAPTERS.includes(adapter))) throw new Error(`Provider message faults (PV*) cover ${PROVIDER_FAULT_ADAPTERS.join(" and ")}; remove the other adapters from --adapters`); // Every PV task gets its own restart through --provider-fault-restart; a planned K cell would be a different task. if (options.faults.length) throw new Error("Provider message faults (PV*) do not share a wave with planned restart faults (K*); use --provider-fault-restart K1 to restart after each provider fault"); - // Every planned task gets the fault; only these workloads send the turn/interrupt the rules target. - const idle = options.workloads.filter(id => !PROVIDER_FAULT_WORKLOADS.includes(id)); - if (idle.length) throw new Error(`Provider message faults (PV*) need workloads that interrupt a provider turn (${PROVIDER_FAULT_WORKLOADS.join(", ")}); remove ${idle.join(", ")} from --workloads`); const messages = parseProviderMessages(values["provider-fault-message"]); + const messageFor = (id: WorkloadId) => messages.byWorkload?.[id] ?? messages.message ?? "terminal"; + const idle = options.workloads.filter(id => !(UNINTERRUPTED_PROVIDER_MESSAGES.includes(messageFor(id)) + ? TOOL_COMPLETED_WORKLOADS : PROVIDER_FAULT_WORKLOADS).includes(id)); + if (idle.length) { + throw new Error(idle.some(id => UNINTERRUPTED_PROVIDER_MESSAGES.includes(messageFor(id))) + ? `--provider-fault-message tool_completed needs workloads that save a document before finishing (${TOOL_COMPLETED_WORKLOADS.join(", ")}); remove ${idle.join(", ")} from --workloads` + : `Provider message faults (PV*) need workloads that interrupt a provider turn (${PROVIDER_FAULT_WORKLOADS.join(", ")}); remove ${idle.join(", ")} from --workloads`); + } + if (options.workloads.some(id => UNINTERRUPTED_PROVIDER_MESSAGES.includes(messageFor(id))) && adapters.some(adapter => adapter !== "codex")) { + throw new Error("--provider-fault-message tool_completed shims Codex only; run it with --adapters codex"); + } if (messages.message) options.providerFaultMessage = messages.message; if (messages.byWorkload) options.providerFaultMessageByWorkload = messages.byWorkload; const delays = parseProviderDelays(values["provider-fault-delays"]); diff --git a/tests/restart-chaos/provider-faults.test.ts b/tests/restart-chaos/provider-faults.test.ts index 3c31a00f3f9..796ecbf3835 100644 --- a/tests/restart-chaos/provider-faults.test.ts +++ b/tests/restart-chaos/provider-faults.test.ts @@ -30,6 +30,9 @@ process.stdin.on("data", chunk => { process.stdout.write(delta.subarray(cut)); send({ id: message.id, result: { turn: { id: "turn-" + message.id } } }); send({ method: "turn/started", params: { threadId: message.params.threadId, turn: { id: "turn-" + message.id } } }); + for (const tool of ["write_document", "paperclip_finish"]) { + send({ method: "item/completed", params: { threadId: message.params.threadId, item: { type: "dynamicToolCall", id: "call-" + tool, tool, status: "completed" } } }); + } }, 20); } if (message.method === "turn/interrupt") { @@ -91,6 +94,7 @@ async function turnThroughShim(root: string, prompt: string, waitMs = 600) { terminals: lines.filter(line => line.message.method === "turn/completed"), starts: lines.filter(line => line.message.id === 1), deltas: lines.filter(line => line.message.method === "item/agentMessage/delta").map(line => line.message.params.delta as string), + toolCompletions: lines.filter(line => line.message.method === "item/completed").map(line => line.message.params.item.tool as string), }; } @@ -148,6 +152,18 @@ describe("provider message fault shim", () => { expect(result.terminals).toHaveLength(1); }); + it("tool_completed drops the first Paperclip tool completion, never a finish or block (staging, 2026-11-11)", async () => { + const { seam: installed, root } = await seam(); + await installed.arm([task("eeee4444", { kind: "PV2", message: "tool_completed" })]); + const result = await turnThroughShim(root, "nonce eeee4444"); + expect(result.toolCompletions).toEqual(["paperclip_finish"]); + expect(result.terminals).toHaveLength(1); + expect((await installed.eventsFor("eeee4444")).filter(event => event.event === "applied")) + .toMatchObject([{ kind: "PV2", message: "tool_completed", action: "drop", tool: "write_document" }]); + expect((await turnThroughShim(root, "nonce eeee4444")).toolCompletions).toEqual(["write_document", "paperclip_finish"]); + expect((await turnThroughShim(root, "nonce ffff5555")).toolCompletions).toEqual(["write_document", "paperclip_finish"]); + }); + it("discards a held message when the provider exits first, as a dead provider would", async () => { const { seam: installed, root } = await seam(); await installed.arm([task("eeee4444", { kind: "PV1", message: "terminal", delayMs: 5_000 })]); @@ -203,6 +219,13 @@ describe("provider fault planning", () => { expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV2", "--provider-fault-delays", "18"])).toThrow(/PV1/); expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--provider-fault-message", "terminal"])).toThrow(/provider fault/); expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV1", "--provider-fault-message", "result"])).toThrow(/terminal/); + const toolCompleted = parseCommand(["--adapters", "codex", "--workloads", "A1", "--faults", "PV2", "--provider-fault-message", "tool_completed", "--tasks", "1", "--seed", "1"]); + if (toolCompleted.mode !== "wave") throw new Error("expected a wave"); + expect(toolCompleted.options).toMatchObject({ providerFaults: ["PV2"], workloads: ["A1"], providerFaultMessage: "tool_completed" }); + expect(planWave("w-20261111T000000-s1", toolCompleted.options).tasks[0]!.providerFault).toEqual({ kind: "PV2", message: "tool_completed" }); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "Q1", "--faults", "PV2", "--provider-fault-message", "tool_completed"])).toThrow(/save a document/); + expect(() => parseCommand(["--adapters", "claude", "--workloads", "A1", "--faults", "PV2", "--provider-fault-message", "tool_completed"])).toThrow(/Codex only/); + expect(() => parseCommand(["--adapters", "codex", "--workloads", "A1", "--faults", "PV2"])).toThrow(/interrupt a provider turn/); expect(() => parseProviderDelays("18,abc")).toThrow(/not a number/); expect(() => parseProviderDelays("601")).toThrow(/PV2/); }); diff --git a/tests/restart-chaos/provider-faults.ts b/tests/restart-chaos/provider-faults.ts index 59808caf906..5b76505420d 100644 --- a/tests/restart-chaos/provider-faults.ts +++ b/tests/restart-chaos/provider-faults.ts @@ -25,9 +25,11 @@ export const PROVIDER_FAULT_KINDS = ["PV1", "PV2", "PV3"] as const; export type ProviderFaultKind = (typeof PROVIDER_FAULT_KINDS)[number]; export const PROVIDER_FAULT_ACTIONS: Record = { PV1: "delay", PV2: "drop", PV3: "duplicate" }; -/** `terminal`: the turn/completed that follows a turn/interrupt. `interrupt_ack`: the response to that turn/interrupt. */ -export const PROVIDER_MESSAGES = ["terminal", "interrupt_ack"] as const; +/** `terminal`: the turn/completed that follows a turn/interrupt. `interrupt_ack`: the response to that turn/interrupt. `tool_completed`: Codex's item/completed for the first non-terminal Paperclip tool call. */ +export const PROVIDER_MESSAGES = ["terminal", "interrupt_ack", "tool_completed"] as const; export type ProviderMessage = (typeof PROVIDER_MESSAGES)[number]; +export const UNINTERRUPTED_PROVIDER_MESSAGES: readonly ProviderMessage[] = ["tool_completed"]; +export const TOOL_COMPLETED_WORKLOADS: readonly WorkloadId[] = ["A1", "C1"]; /** Restarts a provider fault may straddle: K1 kill -9, K2 SIGTERM graceful restart, K4 kill -9 then an outage hold. */ export const PROVIDER_RESTART_KINDS = ["K1", "K2", "K4"] as const; @@ -126,7 +128,8 @@ export function parseProviderMessages(raw: string | undefined): { message?: Prov if (result.message) throw new Error("--provider-fault-message takes one default message"); result.message = right as ProviderMessage; } else { - if (!PROVIDER_FAULT_WORKLOADS.includes(left as WorkloadId)) throw new Error(`--provider-fault-message: ${left} is not one of ${PROVIDER_FAULT_WORKLOADS.join(", ")}`); + const allowed = UNINTERRUPTED_PROVIDER_MESSAGES.includes(right as ProviderMessage) ? TOOL_COMPLETED_WORKLOADS : PROVIDER_FAULT_WORKLOADS; + if (!allowed.includes(left as WorkloadId)) throw new Error(`--provider-fault-message: ${left} is not one of ${allowed.join(", ")}`); result.byWorkload = { ...result.byWorkload, [left]: right as ProviderMessage }; } } diff --git a/tests/restart-chaos/provider-shim/codex-shim.mjs b/tests/restart-chaos/provider-shim/codex-shim.mjs index ef7f374afba..b1b3b103f48 100644 --- a/tests/restart-chaos/provider-shim/codex-shim.mjs +++ b/tests/restart-chaos/provider-shim/codex-shim.mjs @@ -7,7 +7,8 @@ // except for the one provider message an armed rule names: // - "terminal": the `turn/completed` notification that follows a `turn/interrupt` // request (a governed wait or a board Stop); -// - "interrupt_ack": the response to that `turn/interrupt` request. +// - "interrupt_ack": the response to that `turn/interrupt` request; +// - "tool_completed": Codex's `item/completed` for a non-terminal Paperclip tool call. // Rules are armed per task in the control file, by the task's nonce. A process is // bound to a nonce once the nonce appears in its traffic (the task's prompt or a // tool result). Every rule applies at most once per task (an exclusive marker file @@ -118,8 +119,16 @@ child.stdin.on("error", () => undefined); const pendingTimers = new Map(); function emit(line) { process.stdout.write(`${line}\n`); } +const TERMINAL_TOOLS = new Set(["paperclip_finish", "paperclip_block"]); + function classify(message) { if (!message) return null; + if (message.method === "item/completed" && message.params?.item?.type === "dynamicToolCall" + && typeof message.params.item.tool === "string" && !TERMINAL_TOOLS.has(message.params.item.tool)) { + const threadId = message.params?.threadId; + return { message: "tool_completed", afterInterrupt: true, threadId: typeof threadId === "string" ? threadId : null, + turnStatus: null, tool: message.params.item.tool }; + } if (message.method === "turn/completed") { const threadId = message.params?.threadId; const afterInterrupt = typeof threadId === "string" && interruptedThreads.size ? interruptedThreads.has(threadId) : interruptSeen; @@ -146,7 +155,7 @@ function relay(line) { const rule = target ? ruleFor(target) : null; if (!target || !rule || !claim(rule)) { emit(line); return; } const base = { event: "applied", nonce: rule.nonce, kind: rule.kind, message: target.message, turnStatus: target.turnStatus, - requestId: message.id ?? null, sinceSpawnMs: Date.now() - startedAt }; + requestId: message.id ?? null, sinceSpawnMs: Date.now() - startedAt, ...(target.tool ? { tool: target.tool } : {}) }; if (rule.kind === "PV2") { log({ ...base, action: "drop" }); return; } if (rule.kind === "PV3") { emit(line); emit(line); log({ ...base, action: "duplicate" }); return; } const delayMs = Number(rule.delayMs ?? 0);