From 805b8e00d4067e2ce3a8f44e15d25a95cb3c874f Mon Sep 17 00:00:00 2001 From: JB Date: Tue, 25 Aug 2026 19:42:45 -0400 Subject: [PATCH] fix(pi-workflow): preserve recovery gate ownership --- docs/pi-workflow/README.md | 2 +- docs/pi-workflow/design.md | 2 +- extensions/pi-workflow/engine.test.ts | 141 ++++++++++++++++++++++++++ extensions/pi-workflow/engine.ts | 32 +++++- 4 files changed, 172 insertions(+), 5 deletions(-) diff --git a/docs/pi-workflow/README.md b/docs/pi-workflow/README.md index 9c2b4b3..6cb50ae 100644 --- a/docs/pi-workflow/README.md +++ b/docs/pi-workflow/README.md @@ -70,7 +70,7 @@ An analysis run uses `action: "start"` with `workflow: { mode: "analysis-script" ## Recovery -Interrupted work opens a Recovery Gate with retry, inspect, and abort choices. Completed Assignments and Decisions remain persisted. Potentially mutating interrupted Assignments are preserved and are not replayed automatically; an explicit retry creates newly fenced work where policy permits it. +Interrupted work opens a Recovery Gate with retry, inspect, and abort choices. Completed Assignments and Decisions remain persisted. Potentially mutating interrupted Assignments are preserved and are not replayed automatically; an explicit retry creates newly fenced work where policy permits it. Startup recovery leaves an idle run already waiting at Recovery under its current parent instead of letting another Pi session claim it. If that parent's process has exited, an explicit retry from a previously attached parent session may reclaim the stale lease before continuing. ## Artifacts diff --git a/docs/pi-workflow/design.md b/docs/pi-workflow/design.md index 66c7847..6d847d8 100644 --- a/docs/pi-workflow/design.md +++ b/docs/pi-workflow/design.md @@ -508,7 +508,7 @@ Writes are atomic and mode `0600`. Serialization, transcript, source, argument, The snapshot contains Decisions and prior Assignment outcomes, so a separate event store is unnecessary. Every `start`, `decide`, and `abort` command requires an idempotency key. The launch key and initial result are persisted before workspace creation. State-changing commands use revisions and idempotency keys to reject stale or duplicate transitions. Repeating a completed key returns its recorded result without another transition. -An active run owns a persisted lease containing process identity, active parent-session identity, heartbeat time, and worktree identity. The run retains every parent session ID attached through a handoff so `/workflow` can discover it from either session, but only the active ID may mutate the run. Startup and session recovery reconcile stale leases, child processes, workspace-creation intent, Git identity, context handoffs, compaction intents, and assignments left `running`. +An active run owns a persisted lease containing process identity, active parent-session identity, heartbeat time, and worktree identity. The run retains every parent session ID attached through a handoff so `/workflow` can discover it from either session, but only the active ID may mutate the run. The narrow exception is an explicit Recovery retry from a retained parent ID after the recorded lease process has died; that retry reclaims the stale lease before applying the Decision. Startup and session recovery reconcile stale leases, child processes, workspace-creation intent, Git identity, context handoffs, compaction intents, and assignments left `running`. An idle run that already has an open Recovery Gate is left attached to its current parent so unrelated Pi sessions cannot steal ownership during startup. Fresh-session handoff intent is durable before `newSession()`. A `session_shutdown` with reason `new` bypasses ordinary interruption recovery only when its source session matches the pending intent; that handler atomically claims the event's `targetSessionFile` and marks the intent `claimed-target-awaiting-withSession`. The claimed target's immediately following `session_start` is exempt from recovery only when its reason is `new` and `previousSessionFile` matches the source. The replacement `withSession` callback must then match the claimed target before it atomically consumes the intent, rotates the active parent-session ID, records both session IDs, and resumes delivery. Any later `session_start` caused by `startup`, `reload`, `resume`, or `fork`, or a bounded claim expiry before consumption, follows normal recovery and leaves implementation blocked. diff --git a/extensions/pi-workflow/engine.test.ts b/extensions/pi-workflow/engine.test.ts index 6275081..ef83c4b 100644 --- a/extensions/pi-workflow/engine.test.ts +++ b/extensions/pi-workflow/engine.test.ts @@ -1508,6 +1508,147 @@ test("Git or ownership safety failures open Recovery before remediation", async }); }); +test("Recovery retry reruns failed verification after the remediation limit", async () => { + await withTempDir(async (directory) => { + const setup = await fixture(directory); + let commandRuns = 0; + const hostCommands: HostCommandExecutor = { + async run() { + commandRuns++; + return { + result: { + code: commandRuns === 1 ? 1 : 0, + stdout: commandRuns === 1 ? "failed" : "passed", + stderr: "", + }, + }; + }, + }; + const effectiveConfig = { + ...structuredClone(DEFAULT_WORKFLOW_CONFIG), + maxReviewCycles: 1, + }; + setup.executor.enqueue("discover", { + ...discoveryWithQuestions(), + questions: [], + }); + setup.executor.enqueue("design", design("minor")); + setup.executor.enqueue("implement", { + summary: "Initial implementation", + changedPaths: ["src/index.ts"], + focusedTests: ["initial test"], + }); + setup.executor.enqueue("verify", { + succeeded: true, + focusedBehavioralTests: true, + summary: "Corrected implementation verified", + }); + setup.executor.enqueue("review", { + specification: { verdict: "approved", findings: [] }, + standards: { verdict: "approved", findings: [] }, + }); + + const planned = await setup.engine("session-1", { + hostCommands, + effectiveConfig, + }).execute(startCommand()); + const approved = await setup.engine("session-1", { + hostCommands, + effectiveConfig, + }).execute({ + action: "decide", + runId: planned.runId, + gateId: planned.openGate?.id ?? "missing", + expectedRevision: planned.revision, + decision: { + kind: "approve-plan", + approved: true, + acceptedUiImpact: "minor", + }, + idempotencyKey: "approve-failed-verification-retry", + }); + const waiting = await setup.engine("session-1", { + hostCommands, + effectiveConfig, + }).execute({ + action: "decide", + runId: approved.runId, + gateId: approved.openGate?.id ?? "missing", + expectedRevision: approved.revision, + decision: { + kind: "select-implementation-context", + strategy: "current-session", + }, + idempotencyKey: "context-failed-verification-retry", + }); + assert.equal(waiting.openGate?.kind, "recovery"); + assert.equal(commandRuns, 1); + + const transientRecovery = setup.engine("transient-session", { + hostCommands, + effectiveConfig, + processId: 200, + isProcessAlive: () => false, + }); + assert.deepEqual(await transientRecovery.recover(), []); + const untouched = await setup.store.loadRun(waiting.runId); + assert.equal(untouched.revision, waiting.revision); + assert.equal(untouched.lease?.activeParentSessionId, "session-1"); + + const liveStolen = structuredClone(untouched); + liveStolen.revision++; + liveStolen.parentSessionIds.push("transient-session"); + liveStolen.lease = { + processId: 200, + activeParentSessionId: "transient-session", + heartbeatAt: 0, + worktreeIdentity: liveStolen.workspace.root, + }; + await setup.store.saveRun(liveStolen, untouched.revision); + + await assert.rejects( + setup.engine("session-1", { + hostCommands, + effectiveConfig, + processId: 300, + isProcessAlive: (processId) => processId === 200, + }).execute({ + action: "decide", + runId: waiting.runId, + gateId: waiting.openGate?.id ?? "missing", + expectedRevision: liveStolen.revision, + decision: { kind: "recover", resolution: "retry" }, + idempotencyKey: "live-owner-retry-failed-verification", + }), + /Historical parent sessions/, + ); + + const staleStolen = structuredClone(liveStolen); + staleStolen.revision++; + staleStolen.lease!.processId = 201; + await setup.store.saveRun(staleStolen, liveStolen.revision); + + const completed = await setup.engine("session-1", { + hostCommands, + effectiveConfig, + processId: 300, + isProcessAlive: () => false, + }).execute({ + action: "decide", + runId: waiting.runId, + gateId: waiting.openGate?.id ?? "missing", + expectedRevision: staleStolen.revision, + decision: { kind: "recover", resolution: "retry" }, + idempotencyKey: "retry-failed-verification", + }); + + assert.equal(completed.status, "completed"); + assert.equal(commandRuns, 2); + const persisted = await setup.store.loadRun(completed.runId); + assert.equal(persisted.verificationEvidence?.commands[0]?.status, "succeeded"); + }); +}); + test("changes requested trigger serialized remediation, verification, and fresh review", async () => { await withTempDir(async (directory) => { const setup = await fixture(directory); diff --git a/extensions/pi-workflow/engine.ts b/extensions/pi-workflow/engine.ts index 6de1ffb..a6d8d43 100644 --- a/extensions/pi-workflow/engine.ts +++ b/extensions/pi-workflow/engine.ts @@ -1261,7 +1261,7 @@ export class WorkflowEngine { validateWorkflowCommand(command); if (command.action === "start") return this.start(command); if (command.action === "start-analysis") return this.startAnalysis(command); - const run = await this.options.store.loadRun(command.runId); + let run = await this.options.store.loadRun(command.runId); if (command.action === "inspect") return view(run); if (command.action === "consume-context-handoff") { return this.consumeContextHandoff(run, command); @@ -1269,6 +1269,7 @@ export class WorkflowEngine { if (command.action === "recover-context-intent") { return this.recoverContextIntent(run, command); } + run = this.reclaimStaleRecoveryLease(run, command); this.assertActiveParent(run); if (command.action === "abort") return this.abort(run, command); if (command.action === "decide") return this.decide(run, command); @@ -1332,6 +1333,8 @@ export class WorkflowEngine { assignment.status === "running" || (assignment.status === "queued" && Boolean(assignment.executionToken)), ); + const alreadyWaiting = activeGate(run)?.kind === "recovery"; + if (running.length === 0 && alreadyWaiting) continue; if (await this.hasActiveAssignmentLease(run.runId, running)) continue; if (!this.leaseIsStale(run)) continue; run = await this.transition(run, (next) => { @@ -1362,8 +1365,6 @@ export class WorkflowEngine { ); } } - const alreadyWaiting = activeGate(run)?.kind === "recovery"; - if (running.length === 0 && alreadyWaiting) continue; run = await this.transition(run, (next) => { const interruptedIds = new Set(running.map((assignment) => assignment.id)); for (const assignment of next.assignments) { @@ -4519,6 +4520,31 @@ export class WorkflowEngine { }); } + private reclaimStaleRecoveryLease( + run: WorkflowRun, + command: WorkflowCommand, + ) { + if ( + command.action !== "decide" || + command.decision.kind !== "recover" || + command.decision.resolution !== "retry" || + activeGate(run)?.id !== command.gateId || + !run.parentSessionIds.includes(this.options.parentSessionId) || + !run.lease || + this.isProcessAlive(run.lease.processId) + ) { + return run; + } + const claimed = structuredClone(run); + claimed.lease = { + processId: this.processId, + activeParentSessionId: this.options.parentSessionId, + heartbeatAt: this.now(), + worktreeIdentity: claimed.workspace.root, + }; + return claimed; + } + private assertActiveParent(run: WorkflowRun) { if ( run.lease &&