diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index bd40f581e2..710d6c2bf6 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -785,6 +785,22 @@ Examples: Auto-recovery preserves the existing owner. It does not choose a replacement agent. +### Completion tools and final answers + +A completion tool such as `paperclip_finish` reports task disposition; it does not +end the provider turn. Paperclip continues persisting and displaying provider +events until an authoritative turn terminal arrives. The completion report starts +no interruption timer. Existing execution timeouts, cancellation, governed waits, +and active-goal rules still apply. A later failed or cancelled terminal remains +failed or cancelled even when the agent already reported completed work. + +The final assistant message is the visible task response. Response selection runs +after preceding event persistence completes; the completion summary cannot replace +an available final answer. Existing fallback and explicit-comment precedence still +apply. Stream closure without a turn terminal is not proof of success. Event +replay uses the existing source receipts and never repeats provider work merely +to recover recorded output. + ### Provider continuity and bounded finalization A permanently unusable established provider session may be replaced only with evidence that its predecessor is stopped and fenced, completed results and workspace state are preserved, required task history is available, and pending effects have been reconciled. A provider-native shell command or external write without a reliable outcome receipt is unknown. Unknown effects, integrity failures, and unverified process ownership never authorize speculative replay. Once automatic recovery is ruled out, Paperclip selects a conservative default: preserve recorded work, stop the affected task, and retain a durable no-replay hold. Unknown action outcomes remain unknown. No reconciliation form or user diagnosis is required. diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index e3fd5ae5a7..5598c11664 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -136,7 +136,7 @@ const CODEX_COLLABORATION_RUNTIME_INSTRUCTIONS = `## Codex-style collaboration - Before the first tool call in a turn, send a brief commentary update describing the immediate work you are starting. - During tool-driven work, send concise commentary updates at meaningful transitions so the user can follow progress without opening raw logs. - Reserve \`report_progress\` for meaningful durable milestones on longer work. Do not call it merely to create a completion comment on a short run; Paperclip materializes the final assistant response as the durable completion comment. -- Invoke the semantic completion tool exactly once before the final assistant response. After it succeeds, send one self-contained final response with the outcome and verification, then do not call another tool.`; +- Invoke the semantic completion tool exactly once before the final assistant response. After it succeeds, send one self-contained final response with the outcome and verification, then do not call another tool. The completion tool records task disposition; Paperclip keeps receiving your answer until the provider turn ends.`; export function withCodexCollaborationRuntimeInstructions( instructions: string, diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 475b27ce9a..e28e22bed3 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -529,8 +529,8 @@ describe("executeNativeSession recovery", () => { workingNow: true, }); if (!lateGoal) yield runnerEvent(2, "run.result.proposed", result); - // A newly observed goal must revoke the ordinary semantic-result - // timeout, including when the proposal arrived first. + // A newly observed goal owns the lifetime even when the completion + // proposal arrived first. await new Promise((resolve) => setTimeout(resolve, 25)); yield runnerEvent(3, "session.goal.updated", { goal: { @@ -607,7 +607,6 @@ describe("executeNativeSession recovery", () => { controlPlane: port, runnerInstanceId: "runner-agent-goal", controlPlaneInstanceId: "control-agent-goal", - semanticResultTerminalGraceMs: 5, }); expect(startTurn).toHaveBeenCalledOnce(); @@ -4154,7 +4153,7 @@ describe("executeNativeSession recovery", () => { ]); }); - it("retains a reusable session after its remote semantic-result cancellation settles", async () => { + it("uses the execution timeout when a completion report has no provider terminal", async () => { const lifecycle: string[] = []; let releaseProvider = () => {}; const providerReleased = new Promise((resolve) => { @@ -4261,25 +4260,23 @@ describe("executeNativeSession recovery", () => { controlPlane: port, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", - semanticResultTerminalGraceMs: 0, + timeoutMs: 20, keepSessionOpen: true, onSession: (current) => retainedSessions.push(current), }), - ).resolves.toMatchObject({ result, terminal }); + ).rejects.toThrow("native session timed out after 20ms"); expect(cancel).toHaveBeenCalledWith({ - reason: "Paperclip accepted the durable semantic result.", + reason: "Native session event consumption failed.", signal: expect.any(AbortSignal), }); expect(providerResult).not.toHaveBeenCalled(); expect(events.map((event) => event.eventType)).toEqual([ "run.result.proposed", - "run.result.accepted", - "run.terminal", ]); - expect(lifecycle).toEqual(["cancelled"]); - expect(close).not.toHaveBeenCalled(); - expect(retainedSessions).toEqual([session]); + expect(lifecycle).toContain("cancelled"); + expect(close).toHaveBeenCalledOnce(); + expect(retainedSessions.at(-1)).toBeNull(); }); it("retains a reusable session while a semantic terminal releases its remote subscription", async () => { @@ -4374,112 +4371,139 @@ describe("executeNativeSession recovery", () => { expect(retainedSessions).toEqual([session]); }); - it("retains provider output emitted after a durable semantic result", async () => { - const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })); - const close = vi.fn(async () => undefined); - const events: PrpEvent[] = []; - const session: NativeSession = { - identity: () => identity, - async capabilities() { - return { - resume: true, - typedEvents: true, - steering: false, - interruption: true, - structuredResult: true, - }; - }, - async *events() { - yield runnerEvent(1, "run.result.proposed", result); - yield runnerEvent(2, "item.completed", { - item: { type: "assistant_message", text: "Final response." }, - }); - yield runnerEvent(3, "turn.completed"); - }, - async startTurn() { - return { turnId: "turn-recovery" }; - }, - cancel, - async result() { - return null; - }, - async snapshot() { - return { - backendKind: "mock", - sessionId: identity.sessionId, - identity, - providerSessionId: "provider-recovery", - cursor: "3", - activeTurnId: null, - pendingRuntimeRequests: [], - lineage: [], - }; - }, - close, - }; - const backend: NativeSessionBackend = { - async descriptor() { - return { - kind: "mock", - name: "semantic-result-final-response-backend", - version: "1", - capabilities: await session.capabilities(), - }; - }, - async openSession() { - return session; - }, - }; - const port: ControlPlanePort = { - async openRun() {}, - async checkpointSession() {}, - async appendEvent(event) { - events.push(structuredClone(event as PrpEvent)); - const sourceEvents = events.filter( - (candidate) => candidate.sourceInstanceId === event.sourceInstanceId, - ); - return { - cursor: events.length, - highestContiguousSourceSeq: highestContiguous(sourceEvents), - disposition: "committed", - }; - }, - async replayEvents(replay) { - const sourceEvents = events.filter( - (event) => event.sourceInstanceId === replay.sourceInstanceId, - ); - return { - events: structuredClone( - sourceEvents.filter( - (event) => event.sourceSeq > replay.afterSourceSeq, + it.each([ + ["turn.completed", "succeeded"], + ["turn.failed", "failed"], + ["turn.cancelled", "cancelled"], + ["turn.interrupted", "cancelled"], + ["stream.closed", null], + ] as const)("persists a delayed final answer before settling %s", async (eventType, runTerminalState) => { + vi.useFakeTimers(); + try { + let releasePersistence!: () => void; + const persistence = new Promise((resolve) => { releasePersistence = resolve; }); + let persistingAnswer = false; + const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })); + const close = vi.fn(async () => undefined); + const events: PrpEvent[] = []; + const session: NativeSession = { + identity: () => identity, + async capabilities() { + return { + resume: true, + typedEvents: true, + steering: false, + interruption: true, + structuredResult: true, + }; + }, + async *events() { + yield runnerEvent(1, "run.result.proposed", result); + await new Promise((resolve) => setTimeout(resolve, 6_000)); + yield runnerEvent(2, "item.completed", { + kind: "agentMessage", channel: "final", text: "Final response.", + }); + if (eventType !== "stream.closed") yield runnerEvent(3, eventType); + }, + async startTurn() { + return { turnId: "turn-recovery" }; + }, + cancel, + async result() { + return null; + }, + async snapshot() { + return { + backendKind: "mock", + sessionId: identity.sessionId, + identity, + providerSessionId: "provider-recovery", + cursor: "3", + activeTurnId: null, + pendingRuntimeRequests: [], + lineage: [], + }; + }, + close, + }; + const backend: NativeSessionBackend = { + async descriptor() { + return { + kind: "mock", + name: "semantic-result-final-response-backend", + version: "1", + capabilities: await session.capabilities(), + }; + }, + async openSession() { + return session; + }, + }; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event) { + if (event.eventType === "item.completed") { + persistingAnswer = true; + await persistence; + } + events.push(structuredClone(event as PrpEvent)); + const sourceEvents = events.filter( + (candidate) => candidate.sourceInstanceId === event.sourceInstanceId, + ); + return { + cursor: events.length, + highestContiguousSourceSeq: highestContiguous(sourceEvents), + disposition: "committed", + }; + }, + async replayEvents(replay) { + const sourceEvents = events.filter( + (event) => event.sourceInstanceId === replay.sourceInstanceId, + ); + return { + events: structuredClone( + sourceEvents.filter( + (event) => event.sourceSeq > replay.afterSourceSeq, + ), ), - ), - highestContiguousSourceSeq: highestContiguous(sourceEvents), - }; - }, - async completeRun() {}, - }; + highestContiguousSourceSeq: highestContiguous(sourceEvents), + }; + }, + async completeRun() {}, + }; - await expect( - executeNativeSession({ - input, - backend, - controlPlane: port, - runnerInstanceId: "runner-recovery", - controlPlaneInstanceId: "control-recovery", - semanticResultTerminalGraceMs: 50, - }), - ).resolves.toMatchObject({ result, terminal }); + let completed = false; + const execution = executeNativeSession({ + input, backend, controlPlane: port, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + }).then((value) => { completed = true; return value; }); + await vi.waitFor(() => expect(events).toHaveLength(1)); + await vi.advanceTimersByTimeAsync(6_001); + expect(persistingAnswer).toBe(true); + expect(completed).toBe(false); + expect(events.map((event) => event.eventType)).toEqual(["run.result.proposed"]); + expect(cancel).not.toHaveBeenCalled(); + releasePersistence(); + if (eventType === "stream.closed") { + await expect(execution).rejects.toThrow("native event stream closed before a turn terminal fact"); + expect(events.map((event) => event.eventType)).toEqual(["run.result.proposed", "item.completed"]); + return; + } + await expect(execution).resolves.toMatchObject({ result, terminal: { runTerminalState } }); - expect(cancel).not.toHaveBeenCalled(); - expect(close).toHaveBeenCalledOnce(); - expect(events.map((event) => event.eventType)).toEqual([ - "run.result.proposed", - "item.completed", - "turn.completed", - "run.result.accepted", - "run.terminal", - ]); + expect(cancel).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledOnce(); + expect(events.map((event) => event.eventType)).toEqual([ + "run.result.proposed", + "item.completed", + eventType, + "run.result.accepted", + "run.terminal", + ]); + } finally { + vi.useRealTimers(); + } }); it("rejects a mismatched checkpoint before it mutates control-plane state", async () => { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index f98e1c1d80..c75a30687c 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -37,10 +37,9 @@ import { } from "./live/runnerd-codex-transport.js"; export const DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS = 120_000; -export const DEFAULT_NATIVE_SEMANTIC_RESULT_TERMINAL_GRACE_MS = 5_000; const OPTIONAL_SESSION_CANCELLATION_GRACE_MS = 100; const FAILED_OPERATION_SETTLEMENT_GRACE_MS = 100; -// Retaining a remote provider requires its semantic-result interruption to be +// Retaining a remote provider requires terminal subscription teardown to be // acknowledged before the session is handed to another run. Daytona command // round trips routinely exceed the generic failed-operation grace, but remain // bounded by the transport. Other iterator and handoff cleanup keeps the short @@ -146,8 +145,6 @@ export interface ExecuteNativeSessionOptions { checkpointTimeoutMs?: number; /** Internal test seam; production uses the fixed 120-second platform policy. */ runtimeInputLiveWindowMs?: number; - /** Internal test seam; production gives the provider five seconds to end after a result. */ - semanticResultTerminalGraceMs?: number; onSession?: (session: NativeSession | null) => void; /** Observes why a retained session was removed from warm reuse. */ onSessionQuarantined?: (reason: string) => Promise | void; @@ -794,7 +791,6 @@ async function consumeTurn( input: NativeExecutionInput, timeoutMs: number, runtimeInputLiveWindowMs: number, - semanticResultTerminalGraceMs: number, reusableSessionCancellationGraceMs: number, closeFailedSession: () => Promise, quarantineSession: (reason: string) => void, @@ -807,12 +803,9 @@ async function consumeTurn( initialGoal?: HarnessThreadGoal | null, ) { let timer: ReturnType | undefined; - let semanticResultTimer: ReturnType | undefined; - const semanticResultGraceExpired = Symbol("semantic_result_grace_expired"); const appendAbort = new AbortController(); const governedCleanupOperations = new Set>(); let governedCancellationCommitted = false; - let semanticCancellationCommitted = false; let semanticResultObserved = false; let deferredGovernedCleanupSettlement: Promise | null = null; let deferredSessionCancellationSettlement: Promise | null = null; @@ -859,11 +852,6 @@ async function consumeTurn( let goalControlObserved = false; let latestSessionGoal: HarnessThreadGoal | null = null; let resultSource: "semantic_result" | "governed_wait" | null = null; - let semanticResultEvent: PrpEvent | null = null; - let semanticResultDeadline: Promise< - typeof semanticResultGraceExpired - > | null = null; - let pendingNext: ReturnType | null = null; let providerFailure: NativeProviderTerminalFailure | null = null; const settleDurableResult = ( event: PrpEvent, @@ -878,7 +866,6 @@ async function consumeTurn( signal: appendAbort.signal, }); governedCancellationCommitted = true; - semanticCancellationCommitted = event.eventType === "run.result.proposed"; const cleanup = cancellation.cleanup; governedCleanupOperations.add(cleanup); void cleanup @@ -892,37 +879,11 @@ async function consumeTurn( }; }; while (true) { - pendingNext ??= eventIterator.next().catch(error => { throw providerFailure ?? error; }); - const next = - semanticResultDeadline === null - ? await pendingNext - : await Promise.race([pendingNext, semanticResultDeadline]); - if (next === semanticResultGraceExpired) { - void pendingNext.catch(() => undefined); - if (semanticResultEvent === null || governedResult === null) { - throw new Error("native_semantic_result_grace_lost_result"); - } - return settleDurableResult( - semanticResultEvent, - governedResult, - "Paperclip accepted the durable semantic result.", - ); - } - pendingNext = null; + const next = await eventIterator.next().catch((error) => { + throw providerFailure ?? error; + }); if (stopConsumer) throw new Error("native event consumer stopped"); if (next.done) { - if ( - resultSource === "semantic_result" && - semanticResultEvent !== null && - governedResult !== null - ) { - return { - event: semanticResultEvent, - eventCount, - highestContiguousSourceSeq, - governedResult, - }; - } if (providerFailure) throw providerFailure; throw new Error( "native event stream closed before a turn terminal fact", @@ -969,10 +930,8 @@ async function consumeTurn( sessionGoalObserved = true; latestSessionGoal = eventGoal; // A harness-created goal may arrive after a semantic result proposal. - // The goal owns the durable lifetime; revoke the old grace deadline. + // The goal owns the durable lifetime instead of the completion proposal. if (resultSource === "semantic_result") { - if (semanticResultTimer !== null) clearTimeout(semanticResultTimer); - semanticResultDeadline = null; governedResult = null; resultSource = null; } @@ -1068,17 +1027,7 @@ async function consumeTurn( } governedResult = validation.result; resultSource = "semantic_result"; - semanticResultEvent = event; semanticResultObserved = true; - if (session.cancel !== undefined) { - semanticResultDeadline = new Promise((resolve) => { - semanticResultTimer = setTimeout( - () => resolve(semanticResultGraceExpired), - semanticResultTerminalGraceMs, - ); - semanticResultTimer.unref?.(); - }); - } } if (governedResult === null && resolveGovernedWait) { if (appendAbort.signal.aborted) { @@ -1095,9 +1044,8 @@ async function consumeTurn( } if (governedResult !== null && !isTurnTerminal(event) && !sessionGoalObserved) { if (resultSource === "semantic_result") { - // Give the provider a short grace to publish its final assistant - // message and terminal after the semantic tool returns. If no - // terminal arrives, the deadline above finalizes the durable result. + // A completion tool reports task disposition, not provider termination. + // Persist the final answer and authoritative terminal before finalizing. continue; } return settleDurableResult( @@ -1161,6 +1109,7 @@ async function consumeTurn( } } if (isTurnTerminal(event)) { + if (providerFailure) throw providerFailure; return { event, eventCount, @@ -1281,14 +1230,13 @@ async function consumeTurn( // Iterator and provider cleanup own no control-plane mutation authority. // A reusable session with a semantic result needs a longer bounded // window for the remote event subscription to release. This applies - // both when Paperclip forced an interrupt and when the provider emitted - // its own terminal immediately afterward: the latter still crosses the - // remote PRP acknowledgement boundary and routinely takes longer than + // after the provider terminal, which still crosses the remote PRP + // acknowledgement boundary and routinely takes longer than // the generic local cleanup grace. Governed waits and unrelated stalled // cleanup retain the short fail-closed boundary. const teardownSettled = await settlesWithin( passiveTeardownSettlement, - semanticCancellationCommitted || semanticResultObserved + semanticResultObserved ? reusableSessionCancellationGraceMs : FAILED_OPERATION_SETTLEMENT_GRACE_MS, ); @@ -1296,7 +1244,6 @@ async function consumeTurn( quarantineSession("provider_event_teardown_timed_out"); } if (timer !== undefined) clearTimeout(timer); - if (semanticResultTimer !== undefined) clearTimeout(semanticResultTimer); removeExternalAbort(); } } @@ -2257,8 +2204,6 @@ export async function executeNativeSession( options.timeoutMs ?? 900_000, options.runtimeInputLiveWindowMs ?? DEFAULT_NATIVE_RUNTIME_INPUT_LIVE_WINDOW_MS, - options.semanticResultTerminalGraceMs ?? - DEFAULT_NATIVE_SEMANTIC_RESULT_TERMINAL_GRACE_MS, options.keepSessionOpen ? REUSABLE_SESSION_CANCELLATION_SETTLEMENT_GRACE_MS : FAILED_OPERATION_SETTLEMENT_GRACE_MS, @@ -2381,13 +2326,17 @@ export async function executeNativeSession( ? await session.result() : { result: consumed.governedResult, - terminal: { - schema: "paperclip.prp.terminal.v1", - turnTerminalState: "completed", - runTerminalState: "succeeded", - reportedWorkDisposition: - consumed.governedResult.reportedWorkDisposition, - }, + terminal: isTurnTerminal(terminalEvent) + ? terminalFromEvent( + terminalEvent, + consumed.governedResult.reportedWorkDisposition, + ) + : { + schema: "paperclip.prp.terminal.v1", + turnTerminalState: "completed", + runTerminalState: "succeeded", + reportedWorkDisposition: consumed.governedResult.reportedWorkDisposition, + }, turnId: terminalEvent.turnId ?? null, }; signal.throwIfAborted(); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 046fce2c86..3b11aaf03c 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -3313,13 +3313,11 @@ describe("native warm session supervision", () => { .mockReset() .mockImplementationOnce(async (options) => { expect(options.existingSession).toBeUndefined(); - expect(options.semanticResultTerminalGraceMs).toBe(30_000); options.onSession?.(sharedSession); return result; }) .mockImplementationOnce(async (options) => { expect(options.existingSession).toBe(sharedSession); - expect(options.semanticResultTerminalGraceMs).toBe(30_000); return result; }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index bbd921104e..f84dc707cc 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -195,12 +195,6 @@ const TERMINAL_HEARTBEAT_RUN_STATUSES = new Set([ const NATIVE_SESSION_EXECUTION_LEASE_TTL_MS = 20 * 60_000; const NATIVE_SESSION_EXECUTION_LEASE_RENEW_INTERVAL_MS = 5 * 60_000; const NATIVE_SESSION_CANCELLATION_CLEANUP_GRACE_MS = 2_000; -// A reusable provider must publish its terminal suffix before the next run can -// rotate PRP authority. Remote Codex can take more than the ordinary five-second -// result grace to flush its final answer over Daytona, so retain the bounded -// turn long enough to reach a naturally quiescent, reusable state. This adds no -// delay when the provider terminates normally. -const NATIVE_WARM_SEMANTIC_RESULT_TERMINAL_GRACE_MS = 30_000; const NATIVE_RUNTIME_REQUEST_RESOLUTION_CACHE_MAX = 256; type NativeRuntimeRequestResolution = { runId: string; @@ -4898,10 +4892,6 @@ async function executePaperclipNativeSessionWithinScope( keepSessionOpen: warmSessionId !== null, sessionGoalControl: input.sessionGoalControl, resumeSessionGoalHeartbeat: input.resumeSessionGoalHeartbeat, - semanticResultTerminalGraceMs: - warmSessionId === null - ? undefined - : NATIVE_WARM_SEMANTIC_RESULT_TERMINAL_GRACE_MS, // Every durable runner must finish its bounded suspension before // the next run verifies and rotates the saved authority. requireSessionCloseBeforeReturn: runnerdBackend !== null, diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 99853d98b2..3492cecc27 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -1,5 +1,6 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { eq } from "drizzle-orm"; +import { asc, eq } from "drizzle-orm"; +import { readCompletedAssistantMessageCandidate, resolveHeartbeatRunResponse, selectHeartbeatRunFinalAgentMessage } from "../heartbeat-run-summary.js"; import { activityLog, agentWakeupRequests, @@ -495,7 +496,7 @@ describe("PaperclipControlPlanePort conformance", () => { ]); }); - it("completes one selected Paperclip task through the public package session contract", async () => { + it("persists a delayed final answer and replay before resolving the selected task response", async () => { const identity = CONTROL_PLANE_CONFORMANCE_OPEN.identity; const sessionId = taskSessionId; const evidenceRef = `work_product:${taskWorkProductId}`; @@ -519,9 +520,13 @@ describe("PaperclipControlPlanePort conformance", () => { emittedAt: `2026-08-09T02:59:0${sourceSeq}.000Z`, payload, }); + const finalText = "The launch has two stages. Source: https://example.invalid/launch/STREAM-42"; + const finalAnswer = event(2, "item.completed", { kind: "agentMessage", channel: "final", text: finalText }); const events = [ event(1, "run.result.proposed", taskResult), - event(2, "turn.completed", {}), + finalAnswer, + finalAnswer, // Exact replay must not create another stored answer. + event(3, "turn.completed", {}), ]; const backend: NativeSessionBackend = { async descriptor() { @@ -537,9 +542,13 @@ describe("PaperclipControlPlanePort conformance", () => { return { identity: () => input.identity, async capabilities() { return { resume: false, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, - async *events() { yield* events; }, + async *events() { + yield events[0]!; + await new Promise((resolve) => setTimeout(resolve, 6_000)); + yield* events.slice(1); + }, async startTurn() { return { turnId: "turn-phase6-paperclip-task" }; }, - cancel() { return { cleanup: Promise.resolve() }; }, + cancel() { throw new Error("A completion report must not interrupt the provider"); }, async result() { return { result: taskResult, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, turnId: "turn-phase6-paperclip-task" }; }, async snapshot() { return { backendKind: "mock", sessionId, identity: input.identity, providerSessionId: "provider-phase6-paperclip-task" }; }, async close() {}, @@ -606,6 +615,23 @@ describe("PaperclipControlPlanePort conformance", () => { controlPlaneInstanceId: "phase6-control-plane", }); expect(completed.terminal.runTerminalState).toBe("succeeded"); + const stored = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, taskRunId)).orderBy(asc(heartbeatRunEvents.seq)); + const answers = stored.filter((row) => row.eventType === "item.completed"); + expect(answers).toHaveLength(1); + const finalAgentMessage = selectHeartbeatRunFinalAgentMessage({ + candidates: answers.flatMap((row) => { + const candidate = readCompletedAssistantMessageCandidate({ seq: row.seq, prpEvent: row.payload?.prpEvent }); + return candidate ? [candidate] : []; + }), + }); + const response = resolveHeartbeatRunResponse({ + resultJson: { nativeResult: taskResult }, finalAgentMessage, + }); + expect(resolveHeartbeatRunResponse({ resultJson: { nativeResult: taskResult } }).text).toBe(taskResult.summary); + expect(response.text).toBe(finalText); + expect(response.decision.chosenSource).toBe("final_agent_message"); + expect(stored.findIndex((row) => row.eventType === "run.result.accepted")) + .toBeGreaterThan(stored.findIndex((row) => row.eventType === "item.completed")); await finalizeNativeRun({ db, runId: taskRunId, workspaceFinalizeStatus: "succeeded" }); await expect(port.completeRun({ result: taskResult, diff --git a/tests/e2e/fixtures/completion-stream-bin/codex b/tests/e2e/fixtures/completion-stream-bin/codex new file mode 100755 index 0000000000..8483574aa0 --- /dev/null +++ b/tests/e2e/fixtures/completion-stream-bin/codex @@ -0,0 +1,2 @@ +#!/bin/sh +exec node "$(dirname "$0")/../in-feed-codex.mjs" --completion-stream-fixture "$@" diff --git a/tests/e2e/fixtures/in-feed-codex.mjs b/tests/e2e/fixtures/in-feed-codex.mjs index 60cb30f60a..5f8900a030 100644 --- a/tests/e2e/fixtures/in-feed-codex.mjs +++ b/tests/e2e/fixtures/in-feed-codex.mjs @@ -55,6 +55,23 @@ async function finish(text, evidenceRef) { evidence: [{ ref: evidenceRef }], verification: [{ commandOrCheck: 'Fixture outcome', status: 'passed' }], attentionRequests: [], artifacts: [] }); } async function execute() { + if (process.argv.includes('--completion-stream-fixture')) { + // Report completion before a deliberately slow final answer. The real + // runner and control plane must keep listening after the tool succeeds. + await finish('The fixture work is complete.', 'fixture:STREAM-42'); + // Exceed both former completion deadlines: cold (5s) and reusable (30s). + await new Promise((resolve) => setTimeout(resolve, 31_000)); + const itemId = `answer-${turnId}`; + const text = 'STREAM-42: The launch has two stages. Source: https://example.invalid/launch/STREAM-42'; + send({ method: 'item/started', params: { threadId, turnId, item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text: '' } } }); + send({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId, delta: text.slice(0, 40) } }); + await new Promise((resolve) => setTimeout(resolve, 100)); + send({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId, delta: text.slice(40) } }); + send({ method: 'item/completed', params: { threadId, turnId, item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text } } }); + send({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'completed' } } }); + return; + } + if (recoveryFixture) { send({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId: `progress-${turnId}`, delta: 'Checking the current request and available connections.' } }); } diff --git a/tests/e2e/in-feed-native/completion-stream.spec.ts b/tests/e2e/in-feed-native/completion-stream.spec.ts new file mode 100644 index 0000000000..3095ca5b01 --- /dev/null +++ b/tests/e2e/in-feed-native/completion-stream.spec.ts @@ -0,0 +1,103 @@ +import { expect, test } from '@playwright/test'; +import { execFileSync, spawn, type ChildProcess } from 'node:child_process'; +import { writeFile } from 'node:fs/promises'; +import { resolve } from 'node:path'; + +async function stopTestDrive(child: ChildProcess | undefined): Promise { + if (!child?.pid) return; + const signal = (name: NodeJS.Signals) => { + try { process.kill(-child.pid!, name); } + catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ESRCH') throw error; + } + }; + const waitForExit = (timeoutMs: number) => { + if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve(true); + return new Promise((done) => { + const exited = () => { clearTimeout(timer); done(true); }; + const timer = setTimeout(() => { child.removeListener('exit', exited); done(false); }, timeoutMs); + child.once('exit', exited); + }); + }; + if (child.exitCode !== null || child.signalCode !== null) return; + const stopped = waitForExit(15_000); + signal('SIGTERM'); + if (await stopped) return; + const killed = waitForExit(5_000); + signal('SIGKILL'); + if (!(await killed)) throw new Error('test-drive process did not exit after SIGKILL'); +} + +test('a completion tool does not cut off a delayed final answer', async ({ page }, info) => { + await page.setViewportSize({ width: 1440, height: 1000 }); + const root = resolve(import.meta.dirname, '../../..'); + let child: ChildProcess | undefined; + let logs = ''; + try { + const env = { ...process.env, STREAM_FIXTURE_KEY: 'not-a-real-model-key', NODE_ENV: 'test', + PATH: `${root}/tests/e2e/fixtures/completion-stream-bin:${process.env.PATH}` }; + delete env.DATABASE_URL; + delete env.DATABASE_MIGRATION_URL; + child = spawn(process.execPath, ['cli/node_modules/tsx/dist/cli.mjs', 'cli/src/index.ts', 'test-drive', + '--harness', 'codex', '--api-key-env', 'STREAM_FIXTURE_KEY', '--company-name', 'Stream completion', '--no-browser'], + { cwd: root, env, detached: true, stdio: ['ignore', 'pipe', 'pipe'] }); + child.stdout!.on('data', (chunk) => { logs += chunk.toString(); }); + child.stderr!.on('data', (chunk) => { logs += chunk.toString(); }); + await expect.poll(() => logs.match(/Paperclip is ready at (http:\/\/127\.0\.0\.1:\d+)/)?.[1], { timeout: 100_000 }).toBeTruthy(); + const base = logs.match(/Paperclip is ready at (http:\/\/127\.0\.0\.1:\d+)/)![1]!; + const api = async (path: string) => { + const response = await page.request.get(`${base}/api${path}`); + expect(response.ok(), await response.text()).toBeTruthy(); + return response.json(); + }; + const health = await api('/health'); + expect(health).toMatchObject({ status: 'ok', deploymentMode: 'local_trusted', bootstrapStatus: 'ready', + serverInfo: { git: { branchName: execFileSync('git', ['branch', '--show-current'], { cwd: root, encoding: 'utf8' }).trim() } } }); + const [company] = await api('/companies'); + const [agent] = await api(`/companies/${company.id}/agents`); + expect(await api(`/companies/${company.id}/issues`)).toEqual([]); + expect(await api(`/companies/${company.id}/heartbeat-runs`)).toEqual([]); + const prefix = `/${company.issuePrefix}`; + await page.goto(base + prefix + '/dashboard'); + await page.goto(base + prefix + '/company/settings/instance/experimental'); + if (!(await api('/instance/settings/experimental')).enableNativeRunner) { + await page.getByRole('switch', { name: 'Toggle Paperclip Runner experimental setting' }).click(); + } + await expect.poll(async () => (await api('/instance/settings/experimental')).enableNativeRunner).toBe(true); + await page.goto(base + prefix + `/agents/${agent.id}/configuration`); + await page.getByRole('button', { name: 'Codex', exact: true }).click(); + await page.getByRole('button', { name: /Paperclip Runner/ }).click(); + await page.getByRole('button', { name: 'Save changes', exact: true }).click(); + await expect.poll(async () => (await api(`/agents/${agent.id}`)).adapterType).toBe('paperclip_runner'); + await page.getByRole('link', { name: 'Tasks', exact: true }).click(); + await page.getByRole('button', { name: 'New Task', exact: true }).last().click(); + await page.getByPlaceholder('Task title').fill('Summarize the launch decisions and include the source link'); + await page.getByRole('button', { name: 'Assignee', exact: true }).click(); + await page.getByRole('button', { name: 'CEO', exact: true }).click(); + await page.getByRole('button', { name: 'Create Task', exact: true }).click(); + await expect.poll(async () => (await api(`/companies/${company.id}/issues`)).length).toBe(1); + const [issue] = await api(`/companies/${company.id}/issues`); + await page.goto(`${base}${prefix}/issues/${issue.identifier}`); + await expect.poll(async () => (await api(`/companies/${company.id}/heartbeat-runs`))[0]?.status, { timeout: 60_000 }).toBe('succeeded'); + await expect(page.getByText('STREAM-42: The launch has two stages.', { exact: false }).first()).toBeVisible(); + await expect(page.getByRole('link', { name: 'https://example.invalid/launch/STREAM-42', exact: true }).first()).toBeVisible(); + await expect(page.getByRole('textbox', { name: 'editable markdown', exact: true })).toBeEditable(); + await page.reload(); + await expect(page.getByText('STREAM-42: The launch has two stages.', { exact: false }).first()).toBeVisible(); + const runs = await api(`/companies/${company.id}/heartbeat-runs`); + expect(runs).toHaveLength(1); + expect(runs[0]).toMatchObject({ status: 'succeeded' }); + const run = await api(`/heartbeat-runs/${runs[0].id}`); + expect(run.runtimeMode).toBe('native'); + expect(run.resultJson.presentationDecision.chosenSource).toBe('final_agent_message'); + const comments = await api(`/issues/${issue.id}/comments`); + expect(comments.filter((comment: { body: string }) => comment.body.includes('STREAM-42'))).toHaveLength(1); + await page.screenshot({ path: info.outputPath('delayed-final-answer.png'), fullPage: true }); + await writeFile(info.outputPath('evidence.json'), JSON.stringify({ base, health, companyId: company.id, issueId: issue.id, + runId: run.id, dependency: 'deterministic Codex app-server fixture', delayMs: 31000, + dataDir: logs.match(/Data directory: ([^\n\r]+)/)?.[1], presentation: run.resultJson.presentationDecision }, null, 2)); + } finally { + try { await stopTestDrive(child); } + finally { await writeFile(info.outputPath('test-drive.log'), logs); } + } +});