diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index 81a26008517..1918bbbf758 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -74,6 +74,14 @@ the process dies inside the effect window, recovery records an indeterminate result and refuses to execute that command again. Recent results are bounded; commands older than the compacted controller cursor fail closed. +A controller can die after it sends `turn.start` but before it checkpoints the +accepted turn. The provider keeps running that turn. On reattach, the host +driver adopts the single active turn only when the latest completed +`turn.start` result for this run accepted that exact turn id, and no run +attachment came after it. The driver then emits the missing `turn.accepted` +and waits for that turn instead of starting another. Any other active turn +still fails closed as an unexpected turn. + State written before complete-command fingerprints existed is migrated by compacting its legacy command journal through the last recorded controller sequence. The runner can recover, but it rejects redelivery of those older diff --git a/packages/paperclip-runner/src/control-plane/acpx-controller-recovery.test.ts b/packages/paperclip-runner/src/control-plane/acpx-controller-recovery.test.ts index 9c6e8924fea..1e605b0bc84 100644 --- a/packages/paperclip-runner/src/control-plane/acpx-controller-recovery.test.ts +++ b/packages/paperclip-runner/src/control-plane/acpx-controller-recovery.test.ts @@ -13,6 +13,9 @@ import { GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION } from "../drivers/acpx/generat import { createCapabilityRunnerdCodexTransport } from "../live/runnerd-codex-transport.js"; import { bindAcpxCandidateAdmission, type AcpxCandidateStartupAdmission } from "../drivers/acpx/candidate-admission.js"; import { ANTIGRAVITY_DISTRIBUTION_PINS } from "../drivers/acpx/generated-profiles.js"; +import { CodexAppServerDriver } from "../drivers/codex/codex-app-server-driver.js"; +import { createCodexTaskEnvelope } from "../contracts/codex.js"; +import type { PrpEvent } from "../protocol/replay-contract.js"; const packageRoot = fileURLToPath(new URL("../..", import.meta.url)); const binary = process.env.PAPERCLIP_ACPX_REPRO_BINARY ?? resolve(packageRoot, "runner/target/debug/paperclip-runnerd"); @@ -57,7 +60,7 @@ function sidecarSource(root: string, selectedProfile = profile, agent = "claude" " case 'session.open': identity={kind:'acpx',normalizedSessionId:p.normalizedSessionId,acpxRecordId:'record-1',backendSessionId:'backend-1',agentSessionId:'agent-1',profileDigest,workspaceDigest:'sha256:'+'2'.repeat(64),requestedModel:p.model,effectiveModel:p.model,permissionMode:p.permissionMode,mode:p.mode,piThinkingLevel:p.piThinkingLevel,claudeEffort:p.claudeEffort,grokEffort:p.grokEffort,grokDisableWebSearch:p.grokDisableWebSearch,providerLifetimeFenceCandidates:ports};reply(r,{sidecarPid:process.pid,identity,status:{},turnControls:{steering:false,queuedFollowUp:false}});break;", " case 'run.attach': runId=p.runId; reply(r,{runId,catalogRevision:p.catalogRevision});break;", " case 'session.goal.get': if(fs.existsSync(root+'/hold-goal')){const timer=setInterval(()=>{if(fs.existsSync(root+'/release-goal')){clearInterval(timer);reply(r,goalSnapshot());}},10);}else reply(r,goalSnapshot());break;", - " case 'turn.start': turnId=p.turnId;reply(r,{turnId,turnControls:{steering:false,queuedFollowUp:false}});event('runtime.event',{type:'text_delta',text:'started'});break;", + " case 'turn.start': {const start=()=>{turnId=p.turnId;reply(r,{turnId,turnControls:{steering:false,queuedFollowUp:false}});event('runtime.event',{type:'text_delta',text:'started'});fs.writeFileSync(root+'/turn-start-receipt',String(process.pid));};if(fs.existsSync(root+'/hold-turn-start')){const timer=setInterval(()=>{if(fs.existsSync(root+'/release-turn-start')){clearInterval(timer);start();}},10);}else start();break;}", " case 'session.snapshot': reply(r,{identity,runId,turnId,pendingRuntimeRequests:pendingPermission?[{requestId:'permission-1',type:'permission',turnId}]:[]});break;", " case 'permission.resolve': if(!pendingPermission)throw new Error('duplicate approval effect');pendingPermission=false;fs.appendFileSync(root+'/approval-effects','approved\\n');reply(r,{resolved:true});break;", " case 'turn.cancel': reply(r,{cancelled:true});event('runtime.turn_terminal',{status:'interrupted'});turnId=null;break;", @@ -320,6 +323,75 @@ it("adopts the live ACPX execution through the production native transport",asyn } },60000); +it.each(["running","finished"] as const)("adopts the %s turn the runner accepted for a controller that died before checkpointing it (RC-59)",async(phase)=>{ + const run=await fixture(); + let adopted:ReturnType|undefined; + try{ + await run.ready(); + // The dying controller sends turn.start and dies before the provider + // accepts it. The runner accepts and starts the turn while no controller + // is attached, so the replacement receives the result and the start. + writeFileSync(resolve(run.root,"hold-turn-start"),"1"); + run.controller().queueCommand("turn.start",{turnId:"provider-turn-1",text:"Continue the follow-up"},"command-dying-turn",true); + await vi.waitFor(()=>expect(readFileSync(resolve(run.root,"commands.jsonl"),"utf8")).toContain("turn.start"),{timeout:5000}); + await run.stopController();run.restoreCheckpoint(); + writeFileSync(resolve(run.root,"release-turn-start"),"1"); + await vi.waitFor(()=>expect(existsSync(resolve(run.root,"turn-start-receipt"))).toBe(true),{timeout:5000}); + if(phase==="finished"){ + // The provider also finishes the turn before any controller returns. + writeFileSync(resolve(run.root,"action"),"complete"); + await vi.waitFor(()=>expect(existsSync(resolve(run.root,"complete-receipt"))).toBe(true),{timeout:5000}); + } + const options:Parameters[0]={ + provider:"acpx",acpxAgent:"claude",stateDirectory:run.root, + runnerBinary:binary,runnerArtifact:{version:expectedRunnerVersion,digest:expectedRunnerDigest},prpIdentity:identity, + resumeProviderSession:{driverSessionId:"record-1",providerSessionId:"agent-1"}, + runnerProcessLauncher:()=>{throw new Error("duplicate runner launch");},controlPlaneRegistration:run.adoptController, + runnerReconnectGraceMs:10000,environment:{PAPERCLIP_WORKSPACE_CWD:run.root}, + adoptExistingRunner:{pid:run.child.pid!,processGroupId:run.child.pid!,startedAt:run.startedAt, + isAlive:()=>{try{process.kill(run.child.pid!,0);return true;}catch{return false;}}, + signal:signal=>{try{process.kill(-run.child.pid!,signal);return true;}catch{return false;}}}, + }; + adopted=createCapabilityRunnerdCodexTransport(options); + expect(adopted.transport.controllerAcceptedTurnId?.()).toBeNull(); + const driver=new CodexAppServerDriver({conversationMode:"direct", + taskEnvelope:createCodexTaskEnvelope({objective:"Complete the fixture turn",contractRevision:"fixture",criteria:[]}), + transportFactory:()=>adopted!.transport,workingDirectoryAuthority:"remote_runner",environment:{PAPERCLIP_WORKSPACE_CWD:run.root}, + capabilities:{goals:false,threadLineage:false},driverIdentity:{kind:"acpx_runtime",displayName:"Claude via ACPX",version:"fixture"}}); + const recovery=await driver.recoverSession({driverKind:"acpx_runtime",driverSessionId:"record-1",providerSessionId:"agent-1", + runId:identity.runId,normalizedSessionId:identity.normalizedSessionId,activeTurnId:null}); + expect(recovery.recovered,recovery.reason).toBe(true); + const session=recovery.session!; + expect(adopted.transport.controllerAcceptedTurnId?.()).toBe("provider-turn-1"); + if(phase==="running"){ + expect((await session.snapshot()).activeTurnId).toBe("provider-turn-1"); + writeFileSync(resolve(run.root,"action"),"complete"); + } + const events:PrpEvent[]=[]; + for await(const event of session.events()){ + events.push(event); + if(event.eventType.startsWith("turn.")&&!["turn.accepted","turn.started"].includes(event.eventType))break; + } + expect(events.filter(event=>event.eventType.startsWith("turn.")).map(event=>[event.eventType,event.turnId])).toEqual([ + ["turn.accepted","provider-turn-1"],["turn.started","provider-turn-1"],["turn.completed","provider-turn-1"]]); + expect(events.some(event=>event.eventType==="session.failed")).toBe(false); + // Neither the turn's replayed notifications nor its terminal fail the session afterwards. + await new Promise(settle=>setTimeout(settle,500)); + const settled=await session.snapshot(); + expect(settled.terminalTurns?.map(turn=>turn.turnId)).toEqual(["provider-turn-1"]); + expect(settled.activeTurnId??null).toBeNull(); + // The session passed protocol validation, so the next turn reaches the + // Runner. (The scripted sidecar has no second-turn suspension support.) + const next=await session.startTurn({message:{role:"user",text:"Next turn"}}).then(()=>null,(error:unknown)=>String(error)); + expect(next??"").not.toMatch(/unexpected turn|turn_binding_mismatch|session cannot start/); + expect(run.launches()).toHaveLength(1); + expect(readFileSync(resolve(run.root,"commands.jsonl"),"utf8").split("\n").filter(command=>command==="turn.start")).toHaveLength(1); + }finally{ + await adopted?.transport.close().catch(()=>undefined); + await run.cleanup(); + } +},60000); + it.each(["legacy", "stale"])("reconnects before requesting a %s sidecar terminal goal",async(mode)=>{ const run=await fixture(); try{ diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts index 32ae39d837f..2262a3494ca 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -51,6 +51,13 @@ export interface CodexAppServerTransport { prepareConversationResume?(params: Record): Promise; /** One-shot requests authenticated by an adopted live provider, never a saved checkpoint alone. */ takeRestoredRuntimeRequests?(): CodexRpcServerRequest[]; + /** + * The provider turn that this run's latest durable `turn.start` command + * result accepted, or null when no such authenticated result exists. A + * controller can die after the runner accepts its turn/start but before the + * controller checkpoints the turn; recovery uses this to adopt that turn. + */ + controllerAcceptedTurnId?(): string | null; /** * Optional provider-neutral resolution path used when a transport has * already normalized a native server request behind another PRP boundary. diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts index c3ea63d6b03..09cc8880108 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts @@ -634,6 +634,36 @@ export class CodexAppServerDriver implements HarnessDriver { dispositionOnlyRecoveryConsumed = false; dispositionOnlyRecoveryTurnId = null; } + // A controller can die after the runner accepted its turn/start but + // before it checkpointed the turn or emitted turn.accepted. The provider + // then keeps running that turn. Adopt it only when the runner's + // authenticated command result for this run's latest turn.start names + // that turn and no other turn is active; any other active turn stays + // foreign and fails closed when its notifications arrive. + let adoptedAcceptedTurnId: string | null = null; + if ( + !recoveringAutonomousGoal && + recoveredActiveTurnId === null && + !reconcileUncheckpointedDispositionTurn && + snapshot.semanticResult == null + ) { + const turns = Array.isArray(existingThread.turns) + ? existingThread.turns.map(record) + : []; + const acceptedTurnId = transport.controllerAcceptedTurnId?.() ?? ""; + const active = turns.filter((turn) => text(turn.status) === "inProgress"); + // The accepted turn may still run, or it may have finished while no + // controller was attached. Either way it must be the only live turn. + if ( + acceptedTurnId.length > 0 && + turns.some((turn) => text(turn.id) === acceptedTurnId) && + active.every((turn) => text(turn.id) === acceptedTurnId) && + !(snapshot.terminalTurns ?? []).some((turn) => turn.turnId === acceptedTurnId) + ) { + recoveredActiveTurnId = acceptedTurnId; + adoptedAcceptedTurnId = acceptedTurnId; + } + } if (opened.context.liveConsole) opened.context.liveConsole.goals = this.#caps.goals; const session = this.#session({ @@ -653,6 +683,7 @@ export class CodexAppServerDriver implements HarnessDriver { codexUsageBaseline: snapshot.codexUsageBaseline, dispositionOnlyRecoveryConsumed, dispositionOnlyRecoveryTurnId, + adoptedAcceptedTurnId, stalePendingRuntimeRequests: snapshot.pendingRuntimeRequests ?? [], lineage: snapshot.lineage, sourceSequence: snapshot.lastSourceSequence ?? 0, @@ -979,6 +1010,7 @@ export class CodexAppServerDriver implements HarnessDriver { codexUsageBaseline?: PersistedHarnessSession["codexUsageBaseline"]; dispositionOnlyRecoveryConsumed?: boolean; dispositionOnlyRecoveryTurnId?: string | null; + adoptedAcceptedTurnId?: string | null; stalePendingRuntimeRequests?: HarnessRuntimeRequest[]; lineage?: HarnessThreadLineageEntry[]; sourceSequence: number; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts index 65473f50383..df682da37e7 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts @@ -387,6 +387,68 @@ describe("Codex app-server Codex driver", () => { expect((await recovery?.session?.snapshot())?.activeTurnId).toBe("turn-1"); }); + describe("a turn accepted for a controller that died before checkpointing it", () => { + async function recoverAfterLostAcceptance(acceptedTurnId: string | null) { + const first = new FakeCodexTransport(); + const second = Object.assign(new FakeCodexTransport(), { + controllerAcceptedTurnId: () => acceptedTurnId, + }); + second.readResponse = { + thread: { + id: "thread-1", + sessionId: "provider-session-1", + cwd: WORKSPACE, + turns: [{ id: "turn-1", status: "inProgress", items: [] }], + }, + }; + const driver = makeDriver([first, second]); + const original = await driver.openSession({ + runId: "run-lost-acceptance", + normalizedSessionId: "normalized-lost-acceptance", + workingDirectory: WORKSPACE, + }); + // RC-59: the dying controller checkpointed the resumed session, sent + // turn/start, and died before checkpointing the accepted turn. + const snapshot = await original.snapshot(); + expect(snapshot.activeTurnId ?? null).toBeNull(); + await original.close({ reason: "controller restarted" }); + const recovery = await driver.recoverSession!(snapshot); + expect(recovery).toMatchObject({ recovered: true }); + // The runner releases the turn's start, held until the thread read + // bound the recovered turn, then the turn's own terminal. + second.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress", items: [] } }); + second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] } }); + return recovery.session!; + } + + it("adopts the turn when the runner's durable turn.start result names it", async () => { + const recovered = await recoverAfterLostAcceptance("turn-1"); + expect((await recovered.snapshot()).activeTurnId).toBe("turn-1"); + const events = await collectUntilTerminal(recovered.events()); + const turnEvents = events.filter(event => event.eventType.startsWith("turn.")); + expect(turnEvents.map(event => [event.eventType, event.turnId])).toEqual([ + ["turn.accepted", "turn-1"], + ["turn.started", "turn-1"], + ["turn.completed", "turn-1"], + ]); + await expect(recovered.startTurn({ message: { role: "user", text: "Again." } })) + .rejects.toThrow(); + await recovered.close({ reason: "test complete" }); + }); + + it.each([ + ["no accepted turn.start result", null], + ["a turn.start result for a different turn", "turn-other"], + ])("still fails closed on a foreign active turn with %s", async (_label, acceptedTurnId) => { + const recovered = await recoverAfterLostAcceptance(acceptedTurnId); + expect((await recovered.snapshot()).activeTurnId ?? null).toBeNull(); + await expect(collectUntilTerminal(recovered.events())).resolves.toBeDefined(); + await expect(recovered.startTurn({ message: { role: "user", text: "Work." } })) + .rejects.toThrow("Provider started an unexpected turn."); + await recovered.close({ reason: "test complete" }); + }); + }); + it("rejects a recovered provider thread outside the assigned workspace", async () => { const first = new FakeCodexTransport(); const second = new FakeCodexTransport(); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts index 28f305421b3..b2f797a573c 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts @@ -197,6 +197,8 @@ export class CodexSessionState { codexUsageBaseline?: CodexUsageBaseline; dispositionOnlyRecoveryConsumed?: boolean; dispositionOnlyRecoveryTurnId?: string | null; + /** A recovered turn whose turn.accepted the previous controller never emitted. */ + adoptedAcceptedTurnId?: string | null; stalePendingRuntimeRequests?: HarnessRuntimeRequest[]; lineage?: HarnessThreadLineageEntry[]; goal?: HarnessThreadGoal | null; @@ -625,6 +627,12 @@ export function initializeCodexSessionEvents( state.emitGoalEvent("session.goal.snapshot", state.currentGoal, { workingNow: state.activeTurnId !== null, }); + // Recovery adopted a turn the runner accepted for the previous controller, + // which died before emitting turn.accepted. Emit it before any of that + // turn's notifications so its later terminal keeps the normal order. + if (input.adoptedAcceptedTurnId && input.adoptedAcceptedTurnId === state.activeTurnId) { + state.emit("turn.accepted", { turnId: input.adoptedAcceptedTurnId }, { turnId: input.adoptedAcceptedTurnId }); + } for (const stale of input.stalePendingRuntimeRequests ?? []) { state.emit( "runtime_request.cancelled", 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 12845bcac4f..92d14217627 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -68,6 +68,7 @@ import { unwrapRunnerdProviderNotifications, resolveRunnerdCodexSkillInputs, withCodexCollaborationRuntimeInstructions, + controllerAcceptedTurnStartId, } from "./runnerd-codex-transport.js"; import { defaultCapabilityRunnerdBinary, @@ -77,6 +78,33 @@ import { assignedRuntimeContext, } from "../../test/support/runnerd-transport.js"; +describe("controllerAcceptedTurnStartId", () => { + // Shapes copied from the RC-59 durable controller journal. + const attach = { commandId: "command_prp_00000001", controllerSeq: 1, type: "run.attach", status: "completed", result: {} }; + const start = (overrides: Record = {}) => ({ + commandId: "command_lab_start", controllerSeq: 4, type: "turn.start", status: "completed", + payload: { text: "Continue", turnId: "turn_lab_1" }, + result: { commandType: "turn.start", controllerSeq: 4, status: "completed", result: { providerTurnId: "turn_lab_1", status: "accepted" } }, + ...overrides, + }); + + it("names the turn this run's latest completed turn.start accepted", () => { + expect(controllerAcceptedTurnStartId([attach, start()])).toBe("turn_lab_1"); + }); + + it.each([ + ["no turn.start", [attach]], + ["a pending turn.start", [attach, start({ status: "pending", result: undefined })]], + ["a failed turn.start", [attach, start({ status: "failed" })]], + ["a result that did not accept the turn", [attach, start({ result: { result: { providerTurnId: "turn_lab_1", status: "rejected" } } })]], + ["a result for a different turn than requested", [attach, start({ result: { result: { providerTurnId: "turn_lab_2", status: "accepted" } } })]], + ["a later run attachment", [start({ controllerSeq: 1 }), { ...attach, controllerSeq: 2 }]], + ["a later pending turn.start", [attach, start(), start({ controllerSeq: 5, status: "pending", payload: { turnId: "turn_lab_2" }, result: undefined })]], + ])("returns null for %s", (_label, commands) => { + expect(controllerAcceptedTurnStartId(commands)).toBeNull(); + }); +}); + it("attributes opened candidate sessions to the harness biller instead of the model vendor", () => { const provider = runnerdLaunchProfileInternals.openedThreadModelProvider; for (const model of ["gpt-5.6-luna[context=272k,reasoning=medium,fast=false]", "claude-sonnet-5"]) { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index bf1fb16079b..8b7031df883 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -606,6 +606,26 @@ function retargetRunAttachPayload( return payload; } +/** + * The provider turn accepted by this run's latest durable `turn.start` + * command. Only a completed command whose authenticated result accepted the + * same controller-assigned turn id counts, and it must follow the latest run + * attachment so a prior run's turn on a reused Runner is never claimed. + */ +export function controllerAcceptedTurnStartId(commands: readonly unknown[]): string | null { + const ordered = commands.map(record).filter(command => typeof command.controllerSeq === "number") + .sort((left, right) => (left.controllerSeq as number) - (right.controllerSeq as number)); + const latestStart = ordered.findLast(command => command.type === "turn.start"); + if (!latestStart || latestStart.status !== "completed") return null; + const latestAttachment = ordered.findLast(command => command.type === "run.attach" || command.type === "run.prepare"); + if (latestAttachment && (latestAttachment.controllerSeq as number) > (latestStart.controllerSeq as number)) return null; + const accepted = record(record(latestStart.result).result); + const providerTurnId = accepted.providerTurnId; + if (accepted.status !== "accepted" || typeof providerTurnId !== "string" || providerTurnId.length === 0) return null; + const requestedTurnId = record(latestStart.payload).turnId; + return requestedTurnId === undefined || requestedTurnId === providerTurnId ? providerTurnId : null; +} + function recoveredRunAttachment(state: { commands: readonly { commandId: string; @@ -3867,6 +3887,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { return typeof model === "string" && model.startsWith("openrouter/"); } + controllerAcceptedTurnId(): string | null { + return controllerAcceptedTurnStartId(this.#core?.store.state.commands ?? []); + } + turnControlCapabilities(): NativeTurnControlCapabilities | null { if (this.options.provider !== "acpx") return null; if (this.#closed || this.#failure) return { steering: false, queuedFollowUp: false };