From c67c15ef9cbd666fbe8460677dc2c75c7a41fd4b Mon Sep 17 00:00:00 2001 From: wassname <1103714+wassname@users.noreply.github.com> Date: Wed, 9 Sep 2026 22:19:32 +0800 Subject: [PATCH] Fix reviewed recovery edge cases --- .../review-round-1-disposition.md | 10 +++++++ .../review-round-1-validation.txt | 23 ++++++++++++++ src/index.ts | 26 +++++++++------- src/intercom.ts | 5 +++- test/goals-flow.test.ts | 30 +++++++++++++++++++ test/intercom-fixture.ts | 19 ++++++------ test/intercom.test.ts | 26 +++++++++++++--- 7 files changed, 115 insertions(+), 24 deletions(-) create mode 100644 slop/reviews/solo-recovery/review-round-1-disposition.md create mode 100644 slop/reviews/solo-recovery/review-round-1-validation.txt diff --git a/slop/reviews/solo-recovery/review-round-1-disposition.md b/slop/reviews/solo-recovery/review-round-1-disposition.md new file mode 100644 index 0000000..fed0430 --- /dev/null +++ b/slop/reviews/solo-recovery/review-round-1-disposition.md @@ -0,0 +1,10 @@ +# Review round 1 disposition + +All four findings in the independent review were independently confirmed and fixed. + +1. **Stale `readyAttempt` after failed Ready:** cleared the token before restoring the planning state. Regression: `restores planning context after a Ready compaction failure` confirms a later compaction re-arms `pi-goals-planning-context`. +2. **Completed plan becoming solo after restart:** `planIsComplete(ctx)` recognizes only cancelled goals and mechanically signed-off `[x]` goals. Completed pairings remain bound and do not arm recovery/solo fallback or paused-worker messaging. Regression: `does not enter solo when a completed pairing resumes without its supervisor` advances beyond the five-minute recovery window without solo state or follow-up. +3. **First channel registration omitted bounded hello retry:** `onReady` now schedules the existing bounded retry. Regression: `retries an unanswered active-binding hello twice after delayed channel registration` covers configuration before the channel becomes ready. +4. **More than one retained disconnected steer:** a newer disconnected steer records each older pending steer as `superseded`, clears it, and restore logic honors that record. Regression: `retains only the newest disconnected steer across a supervisor reload` proves only the newest persisted instruction replays. + +No findings were rejected. diff --git a/slop/reviews/solo-recovery/review-round-1-validation.txt b/slop/reviews/solo-recovery/review-round-1-validation.txt new file mode 100644 index 0000000..c429b04 --- /dev/null +++ b/slop/reviews/solo-recovery/review-round-1-validation.txt @@ -0,0 +1,23 @@ +Review round 1 validation (fresh isolated evidence) + +Environment for every command: + env -u PI_SUBAGENT_CHILD -u PI_GOALS_ROLE PI_GOALS_EVIDENCE_DIR=/tmp/pi-goals-review-round-1-evidence + +npx vitest run test/goals-flow.test.ts test/intercom.test.ts test/supervisor-session.test.ts + PASS: 3 files, 81 tests. + +npm run typecheck + PASS: tsc --noEmit. + +npm run lint + PASS: biome check src/ test/ (42 files; no fixes). + +npm run build + PASS: tsc. + +npm test + PASS: 22 files, 159 tests. + Note: `fatal: not a git repository` is expected stderr from a negative preflight regression. + +git diff --check + PASS. diff --git a/src/index.ts b/src/index.ts index 195c757..b37723b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -198,6 +198,11 @@ export function registerWorker(pi: ExtensionAPI): void { return supervisorPlanReview(claims.map(goal => goal.subject), changes, planDiff(state.previousPlan ?? "", plan)); } + function planIsComplete(ctx: ExtensionContext): boolean { + const goals = scanGoals(readPlan(ctx)); + return goals.length > 0 && goals.every(goal => goal.status === "cancelled" || (goal.status === "done" && state.signedOffGoals.includes(goalKey(goal.subject)))); + } + function enterSolo(ctx: ExtensionContext, reason: string): void { if (state.phase !== "working" || modelError || intercom.ended) return; recoveryAttempt = undefined; @@ -214,7 +219,7 @@ export function registerWorker(pi: ExtensionAPI): void { // Reuse the existing five-minute readiness window; an explicit peer failure ends it early. function recoverSupervisor(ctx: ExtensionContext): void { - if (recoveryAttempt || recoveryCommand || readyAttempt || state.phase !== "working" || state.mode !== "supervised" || modelError || !intercom.bound || intercom.connected) return; + if (recoveryAttempt || recoveryCommand || readyAttempt || state.phase !== "working" || state.mode !== "supervised" || planIsComplete(ctx) || modelError || !intercom.bound || intercom.connected) return; const attempt = {}; recoveryAttempt = attempt; const binding = state.approvalId; @@ -228,10 +233,10 @@ export function registerWorker(pi: ExtensionAPI): void { }); } - function pauseReason(): string | null { + function pauseReason(ctx: ExtensionContext): string | null { if (!state.phase) return null; if (modelError) return `${modelError} Select /model, then run /goals reconnect.`; - if (state.phase === "working" && state.mode === "supervised" && !intercom.connected) return intercom.peerPresent + if (state.phase === "working" && state.mode === "supervised" && !planIsComplete(ctx) && !intercom.connected) return intercom.peerPresent ? "Supervisor is present but not ready. Inspect its pane for startup/compaction or model errors; recover with /model then /goals reconnect in the supervisor pane if needed." : "Supervisor disconnected. Run /goals reconnect, or /goals restart to replace its tracked pane without discarding the plan."; return null; @@ -408,7 +413,7 @@ export function registerWorker(pi: ExtensionAPI): void { function updateWidget(ctx: ExtensionContext): void { refreshSignoffs(ctx); - const paused = pauseReason(); + const paused = pauseReason(ctx); if (paused) { ctx.ui.setStatus(STATUS_KEY, ctx.ui.theme.fg("warning", "goals paused")); ctx.ui.setWidget(WIDGET_KEY, [`pi-goals paused: ${paused}`]); @@ -495,7 +500,7 @@ export function registerWorker(pi: ExtensionAPI): void { } if (!explicitPlan && arg === "solo") { if (state.phase !== "working") { ctx.ui.notify("Solo requires an already-approved plan. A draft still needs Ready.", "warning"); return; } - if (modelError) { ctx.ui.notify(`Cannot enter solo: ${pauseReason()}`, "warning"); return; } + if (modelError) { ctx.ui.notify(`Cannot enter solo: ${pauseReason(ctx)}`, "warning"); return; } if (state.mode === "solo") { ctx.ui.notify("Already UNSUPERVISED; plan preserved, supervisor sign-off unavailable. /goals restart restores supervision.", "warning"); return; } commandAttempt = command; readyAttempt = undefined; @@ -649,7 +654,7 @@ export function registerWorker(pi: ExtensionAPI): void { // The phase snapshot enters context only when planning starts or context was lost. pi.on("before_agent_start", async (_event, ctx) => { - const paused = pauseReason(); + const paused = pauseReason(ctx); if (paused) return { systemPrompt: `${ctx.getSystemPrompt()}\n\nGoal work is paused: ${paused} Do not implement or sign off goals. Human input and read-only diagnosis remain available; wait for recovery before resuming autonomous work.` }; if (state.phase === "working" && state.mode === "solo") return { systemPrompt: `${ctx.getSystemPrompt()}\n\nYou are the UNSUPERVISED implementation worker for ${planRel(ctx)}. Reason: ${state.soloReason} Continue the approved plan, preserve its goals and save evidence and verification results. There is no supervisor; do not wait for steering, call CompleteGoal, or claim supervised sign-off. Report completion as unreviewed. Use /goals restart to restore supervision.` }; if (state.phase === "working") { @@ -700,7 +705,7 @@ export function registerWorker(pi: ExtensionAPI): void { }); pi.on("tool_call", async (event, ctx) => { - const paused = pauseReason(); + const paused = pauseReason(ctx); if (paused && !(["read", "grep", "find", "ls"].includes(event.toolName) || (event.toolName === "bash" && isPlanningReadOnlyCommand(String((event.input as { command?: string }).command))))) { return { block: true, terminate: true, reason: `Goal work is paused: ${paused} Only read-only diagnosis is available.` }; } @@ -735,7 +740,7 @@ export function registerWorker(pi: ExtensionAPI): void { modelError = `Worker model failed after Pi recovery: ${lastAssistantError}`; lastAssistantError = undefined; intercom.markNotReady(); - ctx.ui.notify(`Goal work paused: ${pauseReason()} No mode or model substitution was made.`, "error"); + ctx.ui.notify(`Goal work paused: ${pauseReason(ctx)} No mode or model substitution was made.`, "error"); updateWidget(ctx); return; } @@ -826,6 +831,7 @@ export function registerWorker(pi: ExtensionAPI): void { readyAttempt = undefined; } catch (error) { if (!current()) return; + readyAttempt = undefined; intercom.markNotReady(); stopWorkerTimers(); ctx.ui.notify(`Goal supervisor could not start: ${error instanceof Error ? error.message : String(error)} Use /goals reconnect to retry, or /goals restart to replace the tracked pane.`, "warning"); @@ -891,11 +897,11 @@ export function registerWorker(pi: ExtensionAPI): void { const version = state.planVersion; if (state.phase !== "working") return result("Planning is not approved. Choose Ready before signing off a goal.", true); if (state.mode === "solo") return result("Supervisor sign-off is unavailable in solo mode. Save evidence and use /goals restart for review; no completion recorded.", true); - if (pauseReason()) return result(`Goal sign-off blocked: ${pauseReason()}`, true); + if (pauseReason(ctx)) return result(`Goal sign-off blocked: ${pauseReason(ctx)}`, true); if (!state.approvalId) return result("Goal sign-off blocked: no current supervisor review.", true); const background = await backgroundState(pi); if (signal?.aborted || intercom.ended || state.approvalId !== binding || state.planVersion !== version || state.phase !== "working") return result("Goal sign-off cancelled or superseded; no completion recorded.", true); - if (intercom.ended || !background.quiet || pauseReason()) return result(`Goal sign-off blocked: ${pauseReason() ?? background.description}`, true); + if (intercom.ended || !background.quiet || pauseReason(ctx)) return result(`Goal sign-off blocked: ${pauseReason(ctx) ?? background.description}`, true); const plan = readPlan(ctx); if (!plan.trim()) return result(`No plan file at ${planRel(ctx)}. Run /goals to draft one.`, true); const block = goalBlock(plan, params.goal); diff --git a/src/intercom.ts b/src/intercom.ts index 3e877e7..67bc582 100644 --- a/src/intercom.ts +++ b/src/intercom.ts @@ -99,6 +99,7 @@ export class GoalIntercom { const message = record.message; if (message.binding !== binding) continue; if (record.direction === "out" && message.kind === "steer") this.pending.set(message.id, message); + if (record.direction === "superseded") this.pending.delete(message.id); if (record.direction === "ack") { this.pending.delete(message.id); if (message.through) this.acknowledgedEntry = message.through; @@ -182,6 +183,8 @@ export class GoalIntercom { if (!this.connected) { const retained = [...this.pending.values()].find(message => message.text === text); if (retained) return { id: retained.id, queued: true }; + for (const pending of this.pending.values()) this.record("superseded", pending); + this.pending.clear(); } const message: Message = { binding: this.binding, role: this.role, kind: "steer", id: randomUUID(), text }; this.record("out", message); @@ -322,7 +325,7 @@ export class GoalIntercom { if (this.stopped || this.registered) return; this.pi.events.emit("intercom:extension-register", { namespace: "pi-goals", ownerEligible: false, - onReady: (channel: IntercomExtensionChannel) => { if (this.stopped) return; this.registered = true; this.channel = channel; this.hello(); }, + onReady: (channel: IntercomExtensionChannel) => { if (this.stopped) return; this.registered = true; this.channel = channel; this.hello(); this.schedulePeerRetry(); }, onEvent: (event: IntercomExtensionEvent) => { try { this.receive(event); } catch (error) { if (!this.stopped) this.ctx?.ui.notify(`Goal Intercom error: ${String(error)}`, "error"); } diff --git a/test/goals-flow.test.ts b/test/goals-flow.test.ts index 831bd20..c90cf3a 100644 --- a/test/goals-flow.test.ts +++ b/test/goals-flow.test.ts @@ -406,6 +406,36 @@ it("shows a missing resumed supervisor, pauses writes, and automatically unpause } finally { rmSync(flow.cwd, { recursive: true, force: true }); } }); +it("restores planning context after a Ready compaction failure", async () => { + const flow = setup(["Ready"]); + try { + await flow.commands.get("goals").handler("make the file", flow.ctx); + approvedPlan(flow.cwd); + await flow.hooks.get("before_agent_start")({}, flow.ctx); + flow.ctx.compact.mockImplementationOnce((options: { onError?: (error: Error) => void }) => options.onError?.(new Error("Compaction cancelled"))); + await flow.hooks.get("agent_settled")({}, flow.ctx); + expect(flow.entries.at(-1)?.data).toMatchObject({ phase: "planning" }); + await flow.hooks.get("session_compact")({}, flow.ctx); + expect(await flow.hooks.get("before_agent_start")({}, flow.ctx)).toMatchObject({ message: expect.objectContaining({ customType: "pi-goals-planning-context" }) }); + } finally { rmSync(flow.cwd, { recursive: true, force: true }); } +}); + +it("does not enter solo when a completed pairing resumes without its supervisor", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "setInterval", "clearInterval"] }); + const flow = setup([]); + try { + const path = writePlan(flow.cwd, "# Plan\n\n## Goals\n\n1. [x] goal: make the file\n\n## Log\n"); + flow.entries.push({ type: "custom", customType: "pi-goals-state", data: { phase: "working", mode: "supervised", approvalId: "restored-binding", supervisorPaneId: "owned-pane", planVersion: 1, signedOffGoals: ["make the file"] } }); + flow.transport.replyToHello(false); + await flow.hooks.get("session_start")({}, flow.ctx); + await vi.advanceTimersByTimeAsync(310_000); + expect(readFileSync(path, "utf8")).toContain("[x] goal: make the file"); + expect(flow.entries.at(-1)?.data).toMatchObject({ phase: "working", mode: "supervised", approvalId: "restored-binding" }); + expect(flow.notifications.some(text => text.includes("UNSUPERVISED WORKER"))).toBe(false); + expect(flow.messages.some(message => message.content.startsWith("UNSUPERVISED WORKER"))).toBe(false); + } finally { rmSync(flow.cwd, { recursive: true, force: true }); } +}); + it("exits planning without deleting the draft or approving implementation", async () => { const flow = setup([]); try { diff --git a/test/intercom-fixture.ts b/test/intercom-fixture.ts index 87d3188..b480eda 100644 --- a/test/intercom-fixture.ts +++ b/test/intercom-fixture.ts @@ -1,11 +1,18 @@ -export function intercomFixture(initialAutoHello = true) { +export function intercomFixture(initialAutoHello = true, deferReady = false) { let autoHello = initialAutoHello; let registration: any; const sent: any[] = []; let connected = true; const receive = (payload: any, fromSessionId = "peer") => registration.onEvent({ type: "message", fromSessionId, payload }); + const ready = () => registration.onReady({ + snapshot: () => ({ connected, supported: true }), + publish: (message: any) => { + sent.push(message); + if (message.kind === "hello" && !message.reply && autoHello) queueMicrotask(() => receive({ ...message, role: message.role === "worker" ? "supervisor" : "worker", ready: true, reply: true })); + }, + }); return { - sent, receive, + sent, receive, ready, replyToHello: (value: boolean) => { autoHello = value; }, event: (event: any) => registration.onEvent(event), connect: (value: boolean) => { connected = value; registration.onEvent({ type: "connection", connected: value, supported: true }); }, @@ -14,13 +21,7 @@ export function intercomFixture(initialAutoHello = true) { emit: (name: string, value: any) => { if (name !== "intercom:extension-register") return false; registration = value; - value.onReady({ - snapshot: () => ({ connected, supported: true }), - publish: (message: any) => { - sent.push(message); - if (message.kind === "hello" && !message.reply && autoHello) queueMicrotask(() => receive({ ...message, role: message.role === "worker" ? "supervisor" : "worker", ready: true, reply: true })); - }, - }); + if (!deferReady) ready(); return true; }, }, diff --git a/test/intercom.test.ts b/test/intercom.test.ts index 65020c1..e4cc24d 100644 --- a/test/intercom.test.ts +++ b/test/intercom.test.ts @@ -3,13 +3,14 @@ import { describe, expect, it, vi } from "vitest"; import { GoalIntercom } from "../src/intercom.js"; import { intercomFixture } from "./intercom-fixture.js"; -function setup(role: "worker" | "supervisor", entries: any[] = [], autoHello = true) { - const fixture = intercomFixture(autoHello); +function setup(role: "worker" | "supervisor", entries: any[] = [], autoHello = true, deferReady = false) { + const fixture = intercomFixture(autoHello, deferReady); const hooks = new Map(); const ctx = { isIdle: vi.fn(() => true), hasPendingMessages: vi.fn(() => false), sessionManager: { getEntries: () => entries }, ui: { notify: vi.fn() } }; const api = { events: fixture.events, on: (name: string, hook: any) => hooks.set(name, hook), appendEntry: (customType: string, data: unknown) => entries.push({ type: "custom", customType, data }) }; const link = new GoalIntercom(api as unknown as ExtensionAPI); link.configure("binding", role, ctx as any); + if (deferReady) fixture.ready(); return { link, fixture, entries, ctx, hooks }; } @@ -79,9 +80,9 @@ describe("pi-intercom transport", () => { }); }); -it("retries an unanswered active-binding hello twice, then leaves normal readiness recovery paused", async () => { +it("retries an unanswered active-binding hello twice after delayed channel registration", async () => { vi.useFakeTimers(); - const runtime = setup("worker", [], false); + const runtime = setup("worker", [], false, true); await vi.advanceTimersByTimeAsync(6_000); expect(runtime.fixture.sent.filter(message => message.kind === "hello" && !message.reply)).toHaveLength(3); await vi.advanceTimersByTimeAsync(60_000); @@ -91,6 +92,23 @@ it("retries an unanswered active-binding hello twice, then leaves normal readine vi.useRealTimers(); }); +it("retains only the newest disconnected steer across a supervisor reload", async () => { + const first = setup("supervisor"); + first.link.markReady(); + await first.link.waitReady(); + first.fixture.connect(false); + const old = first.link.steer("Inspect the old output."); + const latest = first.link.steer("Inspect the replacement output."); + await first.hooks.get("session_shutdown")(); + const resumed = setup("supervisor", [...first.entries]); + resumed.link.markReady(); + await resumed.link.waitReady(); + const replayed = resumed.fixture.sent.filter(message => message.kind === "steer"); + expect(replayed.length).toBeGreaterThan(0); + for (const message of replayed) expect(message).toMatchObject({ id: latest.id, text: "Inspect the replacement output." }); + expect(replayed).not.toContainEqual(expect.objectContaining({ id: old.id })); +}); + it("does not acknowledge a synchronous handoff failure, and retries the instruction", async () => { vi.useFakeTimers(); const runtime = setup("worker");