diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 27a3999f5e0..defd7e6b3e2 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4081,6 +4081,9 @@ are retained during shutdown so recovery can replay a contiguous journal. They carry no permission to start work. Stream closure, stream failure or drain timeout after complete usage leaves the saved wait unfinalized until a provider terminal is retained; complete token counters alone cannot certify a stopped provider. +Codex governed-wait settlement allows up to 30 seconds for its interruption +terminal, matching the transport command window. The caller's execution deadline +can shorten that window. An interrupt acknowledgement alone is not terminal proof. A timeout, missing accounting, or an unproven provider terminal remains a failure; the saved wait does not certify successful provider execution. diff --git a/packages/paperclip-runner/src/backends/codex-native-backend.ts b/packages/paperclip-runner/src/backends/codex-native-backend.ts index 27188fb6f0f..630c14f07c3 100644 --- a/packages/paperclip-runner/src/backends/codex-native-backend.ts +++ b/packages/paperclip-runner/src/backends/codex-native-backend.ts @@ -223,7 +223,10 @@ function createTransportBackedNativeSessionBackend( requireProviderSessionIdentity: options.transportFactory !== undefined, }), preparedContext ? constraints : undefined, - input.provider.kind === "cursor_cloud" ? CURSOR_CLOUD_TERMINAL_DRAIN_TIMEOUT_MS : undefined, + // Codex interruption uses a bounded 30-second transport command. Keep + // listening for its terminal through that window before quarantining a wait. + input.provider.kind === "cursor_cloud" ? CURSOR_CLOUD_TERMINAL_DRAIN_TIMEOUT_MS + : isCodex ? 30_000 : undefined, ); if (input.provider.kind === "cursor_cloud") { backend.descriptor = async () => { diff --git a/packages/paperclip-runner/src/live/runnerd-governed-wait.test.ts b/packages/paperclip-runner/src/live/runnerd-governed-wait.test.ts new file mode 100644 index 00000000000..e0413ee00b3 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-governed-wait.test.ts @@ -0,0 +1,113 @@ +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, it, vi } from "vitest"; +import { createRunnerdNativeSessionBackend } from "../backends/codex-native-backend.js"; +import type { NativeExecutionInputV1 } from "../contracts/native-execution.js"; +import type { PersistedNativeSession } from "../contracts/native-session-backend.js"; +import type { PrpEvent, PrpStructuredRunResult } from "../protocol/replay-contract.js"; +import { executeNativeSession } from "../native-session-runtime.js"; +import { createRunnerdCodexTransport } from "./runnerd-codex-transport.js"; +import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; + +it.each([0, 18_000, null])("requires a proven Codex terminal when parking a wait (delay: %s)", async terminalDelayMs => { + const root = await mkdtemp(join(tmpdir(), "runner-governed-wait-")); + const input: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { + runId: "run-wait", companyId: "company-wait", issueId: "issue-wait", + agentId: "agent-wait", executionWorkspaceId: "workspace-wait", + }, + session: { + normalizedSessionId: "session-wait", driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, + }, + task: { + identifier: "TEST-1", title: "Ask for access", description: null, + prompt: "Ask for required access.", workMode: "standard", + }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + provider: { kind: "codex", model: null }, + completionContract: { + id: "contract-wait", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", + contract: { + revision: "1", objective: "Ask for access", + criteria: [{ id: "access", requirement: "Obtain access" }], + }, + }, + interactionResponses: [], credentialBindings: [], + }; + const bundle = createRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root, "--hold-turn", "--emit-tool-call", "--call-log", join(root, "calls.log"), + ...(terminalDelayMs === null + ? ["--accept-interrupt-without-terminal"] + : ["--interrupt-terminal-delay-ms", String(terminalDelayMs)])), + stateDirectory: root, + prpIdentity: { + runnerInstanceId: "runner-wait", environmentLeaseId: "lease-wait", runId: "run-wait", + normalizedSessionId: "session-wait", turnId: "turn-wait", itemId: "item-wait", + }, + }); + // The controlled provider calls its standard dynamic-tool fixture. The port + // below supplies the server's already-resolved governed-wait disposition. + const tool = vi.fn(async () => ({ ok: true, task: { id: "task-1" } })); + const backend = createRunnerdNativeSessionBackend(input, { + runnerInstanceId: "runner-wait", transportFactory: () => bundle.transport, + dynamicTools: [{ + name: "get_task_context", description: "Read the task.", + inputSchema: { type: "object", properties: {}, additionalProperties: false }, + }], + dynamicToolHandler: tool, + }); + const yielded: PrpStructuredRunResult = { + schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", summary: "Waiting for access.", + completionClaim: { + contractRevision: "1", objectiveSatisfied: false, + criteria: [{ criterionId: "access", status: "unknown", evidenceRefs: [] }], + remainingWork: [{ description: "Await access.", blocksCompletion: true }], + }, + evidence: [], verification: [], attentionRequests: [], artifacts: [], + continuation: { kind: "response_wake", summary: "Continue after access.", idempotencyKey: "question-1" }, + }; + const history: PrpEvent[] = []; + const completeRun = vi.fn(async () => undefined); + let checkpoint: PersistedNativeSession | undefined; + try { + const execution = executeNativeSession({ + input, backend, runnerInstanceId: "runner-wait", controlPlaneInstanceId: "controller-wait", timeoutMs: 120_000, + controlPlane: { + async openRun() {}, + async checkpointSession(value) { checkpoint = structuredClone(value); }, + completeRun, + async appendEvent(event: PrpEvent) { + history.push(event); + return { cursor: history.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; + }, + async replayEvents(request) { + return { + events: history.filter(event => event.sourceInstanceId === request.sourceInstanceId && event.sourceSeq > request.afterSourceSeq), + highestContiguousSourceSeq: 0, + }; + }, + }, + resolveGovernedWait: ({ event }) => event.eventType === "item.completed" && event.payload.kind === "dynamicToolCall" ? yielded : null, + }); + if (terminalDelayMs === null) { + await expect(execution).rejects.toMatchObject({ providerCode: "governed_wait_provider_terminal_unproven" }); + expect(completeRun).not.toHaveBeenCalled(); + expect(history.some(event => event.eventType === "run.terminal")).toBe(false); + } else { + await expect(execution).resolves.toMatchObject({ result: yielded }); + expect(completeRun).toHaveBeenCalledOnce(); + expect(history.some(event => event.eventType === "turn.interrupted")).toBe(true); + } + expect(checkpoint?.governedWait?.result).toEqual(yielded); + expect(tool).toHaveBeenCalledOnce(); + const calls = (await readFile(join(root, "calls.log"), "utf8")).split("\n"); + expect(calls.filter(call => call === "turn/start")).toHaveLength(1); + } finally { + await bundle.transport.close().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } +}, 45_000);