Fix reviewed recovery edge cases

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.

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.

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);

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"); }

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 {

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;
 			},
 		},

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<string, any>();
 	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");