From 43fc97aebdc10f1f5699d89632364f017ab8e5e8 Mon Sep 17 00:00:00 2001 From: wassname2 Date: Wed, 9 Sep 2026 19:48:10 +0800 Subject: [PATCH] WIP: preserve startup failure and stale-pane recovery Saved before switching to the alternate implementation at user request. npm test run: 107 passed, 1 failed, 2 skipped. The recorded-pane test still expects one Herdr call; the new discovery path makes two. Not a functional acceptance claim. --- src/index.ts | 13 ++++--- src/internal/supervisor/index.ts | 32 ++++++++++++++--- src/internal/supervisor/protocol.ts | 2 ++ src/supervisor.ts | 49 +++++++++++++++++++++++---- test/internal-supervisor/plan.test.ts | 18 ++++++++++ test/supervisor-integration.test.ts | 19 +++++++++-- 6 files changed, 114 insertions(+), 19 deletions(-) diff --git a/src/index.ts b/src/index.ts index d90de20..8f5e258 100644 --- a/src/index.ts +++ b/src/index.ts @@ -152,6 +152,7 @@ interface PlanState { legacyCompletionClaims: string[]; /** Ready captured its fork, but worker model recovery is still pending (also across reload). */ modelRecovery: "worker" | null; + startupError?: string; /** Optional model ref for the sign-off judge; unset => current session model, else pi's default. */ judgeModel: string | null; planVersion: number | null; @@ -309,8 +310,8 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { return; } if (state.phase === "planning") { - ctx.ui.setStatus(STATUS_KEY, ctx.ui.theme.fg("warning", state.modelRecovery ? models.ready ? "retry Ready" : "worker model paused" : "planning")); - ctx.ui.setWidget(WIDGET_KEY, ["pi-goals: drafting goals"]); + ctx.ui.setStatus(STATUS_KEY, ctx.ui.theme.fg("warning", state.startupError ? "supervisor startup failed" : state.modelRecovery ? models.ready ? "retry Ready" : "worker model paused" : "planning")); + ctx.ui.setWidget(WIDGET_KEY, state.startupError ? [`pi-goals: ${state.startupError}`, "No work started. /goals stop or /goals exit leaves startup; inspect the supervisor before retrying Ready."] : ["pi-goals: drafting goals"]); return; } if (state.phase === "starting") { @@ -367,14 +368,14 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { const approvedDraft = planHash(readPlan(ctx)); const handoff = workMessage(ctx); const recoveringWorker = state.modelRecovery === "worker"; - state = { ...state, phase: "starting" }; + state = { ...state, phase: "starting", startupError: undefined }; persist(); updateWidget(ctx); try { // A stopped, never-attached bootstrap cannot be rejoined. A new explicit Ready // may replace it; reconnect/resume never create another fork. if (state.supervisor) { const previous = await supervisor.status(signal); - if (previous.binding?.paused && previous.role === "none") { + if ((previous.binding?.paused || previous.binding?.startupFailure) && previous.role === "none") { await supervisor.stop(state.supervisor.id); state = { ...state, supervisor: null }; persist(); } @@ -413,7 +414,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { pi.sendUserMessage(handoff, { deliverAs: "followUp" }); } catch (error) { if (signal.aborted) return; - state = { ...state, phase: "planning", modelRecovery: null }; persist(); updateWidget(ctx); + state = { ...state, phase: "planning", modelRecovery: null, reviewRequested: false, startupError: String(error) }; persist(); updateWidget(ctx); ctx.ui.notify(`Could not initialize the supervisor: ${String(error)}. Use /goals supervisor to inspect startup, or /goals steward off and retry Ready.`, "error"); } finally { if (!lifetime.signal.aborted && state.phase !== "working" && !state.modelRecovery && !state.pausedFrom && !state.exited) { @@ -636,6 +637,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { state = { ...state, phase: "planning", + startupError: undefined, pausedFrom: undefined, resumeHash: undefined, exited: false, modelRecovery: null, reviewRequested: false, @@ -878,6 +880,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { signedOffGoals: last?.data?.signedOffGoals ?? [], legacyCompletionClaims: last?.data?.legacyCompletionClaims ?? [], modelRecovery: last?.data?.modelRecovery ?? null, + startupError: saved?.startupError ?? (saved?.phase === "starting" ? "Supervisor startup interrupted by reload. Inspect its pane before retrying Ready." : undefined), judgeModel: last?.data?.judgeModel ?? null, planVersion: last?.data?.planVersion ?? null, autoIntervalMs: useNewDefaults || saved?.autoIntervalMs === undefined ? AUTO_DEFAULT_INTERVAL_MS : saved.autoIntervalMs, diff --git a/src/internal/supervisor/index.ts b/src/internal/supervisor/index.ts index e6795d5..923ece2 100644 --- a/src/internal/supervisor/index.ts +++ b/src/internal/supervisor/index.ts @@ -308,7 +308,7 @@ export default function (pi: any, modelReady: () => boolean = () => true, planPr return { workerId: available ? await resolveOwnId() : ownId, role: state.role, binding: state.plan, connected: available && !!state.pairedId && !peerDisconnected && live.some(s => s.id === state.pairedId), activity: state.plan?.stopped ? "ended" : state.plan?.paused ? "stopped by user" : compacting ? "compacting" : bootstrapPending ? "starting" : activeAssessment ? "reviewing" : refreshInFlight ? "requesting overview" : state.plan?.active ? "monitoring" : "inactive", - lastFailure }; + lastFailure: lastFailure ?? state.plan?.startupFailure }; }, pause(exit = false) { pauseLocally(exit); @@ -378,7 +378,15 @@ export default function (pi: any, modelReady: () => boolean = () => true, planPr await startPair(worker, state.goal, ctx, binding); return binding; } catch (error) { - if (!stopping && generation === pairingGeneration) reset("Supervisor initialization failed; retry from the worker"); + if (!stopping && generation === pairingGeneration) { + const reason = `Supervisor initialization failed: ${String(error)}`.slice(0, 2000); + lastFailure = reason; + // Readiness of the terminal is not readiness of the pairing. Notify the + // worker's attachment wait of this terminal failure before resetting locally. + try { send({ t: "plan_failed", to: bootstrap.workerId, bindingId: binding.id, sessionFile: binding.supervisorSession, reason }); } + catch { ctx.ui.notify("Startup failed; the worker could not be notified. Stop startup in the worker pane.", "warning"); } + reset(reason); + } throw error; } finally { bootstrapPending = false; } }, @@ -392,10 +400,20 @@ export default function (pi: any, modelReady: () => boolean = () => true, planPr reset("Plan supervision stopped", true); if (!connected) throw new Error("Stopped locally; Intercom disconnected so the supervisor may not have received stop. Inspect its recorded pane."); }, - async attached(bindingId, signal) { + async attached(bindingId, signal, expected) { await ready(signal); if (!state.plan || bindingId !== state.plan.id) throw new Error("Plan pairing changed or is unavailable"); - if (state.role === "worker" && (await channel!.listSessions()).some(s => s.id === state.pairedId)) return state.plan; + if (state.plan.startupFailure) throw new Error(state.plan.startupFailure); + if (expected) { + if (expected.id !== bindingId || expected.workerSession !== state.plan.workerSession) throw new Error("Startup identity changed"); + state = { ...state, plan: { ...state.plan, supervisorSession: expected.supervisorSession, supervisorPane: expected.supervisorPane } }; save(); + } + if (state.role === "worker") { + const live = await channel!.listSessions(); + if (!state.plan || state.plan.id !== bindingId) throw new Error("Plan pairing changed while checking attachment"); + if (state.plan.startupFailure) throw new Error(state.plan.startupFailure); + if (live.some(s => s.id === state.pairedId)) return state.plan; + } if (attached) throw new Error("Already waiting for the supervisor"); const pending = pendingReply(signal, () => { attached = undefined; }); attached = pending; @@ -771,6 +789,12 @@ ${latestView}` }, if (from !== me && state.role === "none" && !process.env[SUBAGENT_ENV]) send({ t: "here", to: from }); return; } + if (wire.t === "plan_failed" && wire.to === me && state.plan?.id === wire.bindingId && state.plan.supervisorSession === wire.sessionFile && !state.plan.active && attached) { + lastFailure = wire.reason; + state = { ...state, plan: { ...state.plan, startupFailure: wire.reason, active: false } }; save(); + attached.finish(undefined, new Error(wire.reason)); attached = undefined; + return; + } if (wire.t === "plan_stop" && state.role === "supervisor" && state.plan?.id === wire.bindingId && from === state.pairedId) { ctx.abort(); reset("Plan supervision stopped", true); return; } diff --git a/src/internal/supervisor/protocol.ts b/src/internal/supervisor/protocol.ts index 07c8802..18cc070 100644 --- a/src/internal/supervisor/protocol.ts +++ b/src/internal/supervisor/protocol.ts @@ -56,6 +56,7 @@ export interface GoalDecision extends GoalReview { } export type PlanWire = | { t: "plan_hello" | "plan_hello_ack"; to: string; bindingId: string; role: "worker" | "supervisor"; sessionFile: string; paused?: boolean; pauseId?: string } + | { t: "plan_failed"; to: string; bindingId: string; sessionFile: string; reason: string } | { t: "plan_pause"; to: string; bindingId: string; exit: boolean; pauseId: string } | { t: "plan_resume"; to: string; bindingId: string; requestId: string; planHash: string; pauseId?: string } | { t: "plan_resumed"; to: string; bindingId: string; requestId: string; accepted: boolean } @@ -66,6 +67,7 @@ export type PlanWire = export function validPlanWire(value: any): value is PlanWire { if (!value || typeof value.to !== "string" || typeof value.bindingId !== "string") return false; if (value.t === "plan_hello" || value.t === "plan_hello_ack") return ["worker", "supervisor"].includes(value.role) && typeof value.sessionFile === "string" && (value.paused === undefined || typeof value.paused === "boolean") && (value.pauseId === undefined || typeof value.pauseId === "string"); + if (value.t === "plan_failed") return typeof value.sessionFile === "string" && typeof value.reason === "string" && value.reason.length > 0 && value.reason.length <= 2000; if (value.t === "plan_pause") return typeof value.exit === "boolean" && typeof value.pauseId === "string"; if (value.t === "plan_resume") return typeof value.requestId === "string" && typeof value.planHash === "string" && (value.pauseId === undefined || typeof value.pauseId === "string"); if (value.t === "plan_resumed") return typeof value.requestId === "string" && typeof value.accepted === "boolean"; diff --git a/src/supervisor.ts b/src/supervisor.ts index a73e1c0..4b0cf52 100644 --- a/src/supervisor.ts +++ b/src/supervisor.ts @@ -12,6 +12,7 @@ export interface SupervisorBinding { supervisorSession?: string; active?: boolean; stopped?: boolean; + startupFailure?: string; /** User pause: retain the pair, but no autonomous work until explicit resume. */ paused?: boolean; pauseId?: string; @@ -31,7 +32,7 @@ export interface SupervisorController { status(signal?: AbortSignal): Promise; prepare(binding: SupervisorBinding, signal?: AbortSignal): Promise; bootstrap(bootstrap: Bootstrap, signal?: AbortSignal): Promise; - attached(bindingId: string, signal?: AbortSignal): Promise; + attached(bindingId: string, signal?: AbortSignal, expected?: SupervisorBinding): Promise; activate(bindingId: string, signal?: AbortSignal): Promise; review(bindingId: string, goal: string, hash: string, signal?: AbortSignal): Promise; stop(bindingId: string): Promise; @@ -73,12 +74,21 @@ export function pendingReply(signal: AbortSignal | undefined, cancel: () => v return { requestId: randomUUID(), promise, finish }; } +class HerdrFailure extends Error { + constructor(message: string, readonly code?: string) { super(message); } +} + async function herdr(pi: ExtensionAPI, args: string[], signal?: AbortSignal): Promise> { if (process.env.HERDR_ENV !== "1") throw new Error("Start Pi inside Herdr to launch or focus the supervisor. No pane was created."); const result = await pi.exec("herdr", args, { timeout: 45_000, signal }); - if (result.code !== 0) throw new Error(`Herdr: ${result.stderr || result.stdout}`); + if (result.code !== 0) { + const text = result.stderr || result.stdout; + let code: string | undefined; + try { code = JSON.parse(text).error?.code; } catch { /* Non-JSON transport failures remain non-recoverable. */ } + throw new HerdrFailure(`Herdr: ${text}`, code); + } const parsed = JSON.parse(result.stdout); - if (parsed.error) throw new Error(`Herdr: ${parsed.error.message}`); + if (parsed.error) throw new HerdrFailure(`Herdr: ${parsed.error.message}`, parsed.error.code); return parsed.result ?? parsed; } @@ -86,7 +96,7 @@ export async function focusSupervisor(pi: ExtensionAPI, binding: SupervisorBindi const pane = target === "worker" ? binding.workerPane : binding.supervisorPane; if (!pane) throw new Error("No supervisor pane is recorded. Select Ready to start it."); try { await herdr(pi, target === "zoom" ? ["pane", "zoom", "--pane", pane, "--toggle"] : ["agent", "focus", pane]); } - catch (error) { throw new Error(`${String(error)}. Session location/liveness is unknown. Locate the existing supervisor first; only after confirming it is no longer running, reopen pi --session ${JSON.stringify(binding.supervisorSession)}.`); } + catch (error) { throw new Error(`${String(error)}. Session location/liveness is unknown. Locate the existing supervisor first; only after confirming it is no longer running, reopen pi --session ${JSON.stringify(binding.supervisorSession)}.`, { cause: error }); } } export function supervisorBootstrap(ctx: ExtensionContext): Bootstrap | undefined { @@ -152,9 +162,34 @@ export async function startSupervisor( } if (binding.supervisorPane) { // An existing occupant is not permission to start another process on the same session file. - await focusSupervisor(pi, binding, "supervisor"); + try { await focusSupervisor(pi, binding, "supervisor"); } + catch (error) { + const cause = (error as Error).cause; + if (!(cause instanceof HerdrFailure) || !["agent_not_found", "pane_not_found"].includes(cause.code ?? "")) throw error; + // Only an explicit Ready reaches this path. A missing pane is not a reason + // to reopen its session blindly: first locate any moved/resumed occupant. + let roster: Record; + try { roster = await herdr(pi, ["agent", "list"], signal); } catch { throw error; } + signal.throwIfAborted(); + if (!Array.isArray(roster.agents)) throw error; + const matches = roster.agents.filter((agent: any) => agent.agent_session?.value === binding.supervisorSession); + if (matches.length > 1) throw new Error("Multiple panes report this supervisor session; resolve the duplicate before Ready."); + if (matches.length === 1) { + if (typeof matches[0].pane_id !== "string") throw error; + binding = { ...binding, supervisorPane: matches[0].pane_id }; save(binding); + await focusSupervisor(pi, binding, "supervisor"); + } else { + if (roster.agents.some((agent: any) => agent.agent === "pi" && !agent.agent_session?.value)) throw new Error("A Pi pane has unknown session identity; inspect it before replacing the supervisor."); + // End the old binding before creating a DISTINCT fork. Late traffic from + // the old process cannot authorize or steer the replacement worker pair. + await supervisor.stop(binding.id); + signal.throwIfAborted(); + ctx.ui.notify("Recorded supervisor is no longer in Herdr. Ready is creating a replacement; the old session is retained.", "info"); + return startSupervisor(pi, supervisor, ctx, planPath, null, save, signal); + } + } signal.throwIfAborted(); - return await supervisor.attached(binding.id, signal); + return await supervisor.attached(binding.id, signal, binding); } const split = await herdr(pi, ["pane", "split", "--current", "--direction", "right", "--cwd", ctx.cwd, "--env", `PI_CODING_AGENT_DIR=${getAgentDir()}`, "--no-focus"], signal); const pane = split.pane?.pane_id; @@ -164,7 +199,7 @@ export async function startSupervisor( signal.throwIfAborted(); // Keep bootstrap and worker binding identical, including the returned pane identity. SessionManager.open(binding.supervisorSession!).appendCustomEntry(SUPERVISOR_ROLE, { binding, workerId: status.workerId }); - const waiting = supervisor.attached(binding.id, signal); + const waiting = supervisor.attached(binding.id, signal, binding); void waiting.catch(() => {}); try { await herdr(pi, ["agent", "start", `supervisor-${binding.id.slice(0, 8)}`, "--kind", "pi", "--pane", pane, "--", "--session", binding.supervisorSession!, ...supervisorResourceArgs(process.argv.slice(2))], signal); diff --git a/test/internal-supervisor/plan.test.ts b/test/internal-supervisor/plan.test.ts index 369437e..0b57a28 100644 --- a/test/internal-supervisor/plan.test.ts +++ b/test/internal-supervisor/plan.test.ts @@ -9,6 +9,24 @@ import { type SupervisorBinding as PlanBinding, planHash } from "../../src/super const tick = () => new Promise(resolve => setImmediate(resolve)); +test("bootstrap failure reaches only its matching attachment wait", async () => { + const h = pairHarness(); + try { + await h.worker.hook("session_start"); await h.supervisor.hook("session_start"); + await h.worker.controller.prepare(h.binding); + const wait = h.worker.controller.attached(h.binding.id); + let settled = false; void wait.then(() => { settled = true; }, () => { settled = true; }); + const rejection = assert.rejects(wait, /Native compaction timeout/); + h.worker.receive({ type: "message", fromSessionId: "other", payload: { t: "plan_failed", to: "worker", bindingId: h.binding.id, sessionFile: "/wrong/session", reason: "wrong attempt" } }); + await tick(); assert.equal(settled, false); + h.supervisor.ctx.compact = ({ onError }: any) => onError(new Error("Native compaction timeout")); + await assert.rejects(h.supervisor.controller.bootstrap({ binding: h.binding, workerId: "worker" }), /Native compaction timeout/); + await rejection; + assert.match((await h.worker.controller.status()).lastFailure, /Native compaction timeout/); + await assert.rejects(h.worker.controller.attached(h.binding.id), /Native compaction timeout/, "retry cannot hang on a known failed attachment"); + } finally { await h.close(); } +}); + test("human pause survives reload/reconnect; only explicit resume reactivates the same pair", async () => { const h = pairHarness(); try { diff --git a/test/supervisor-integration.test.ts b/test/supervisor-integration.test.ts index 365b97c..fd7c5e1 100644 --- a/test/supervisor-integration.test.ts +++ b/test/supervisor-integration.test.ts @@ -22,7 +22,7 @@ const tick = () => new Promise(resolve => setImmediate(resolve)); afterEach(() => { vi.unstubAllEnvs(); judge.calls = []; }); describe("actual goals and supervisor package hooks (Herdr and judge mocked)", () => { - it.each(["completion", "replacement during activation", "steward off during activation", "missing worker model", "unauthenticated worker model", "clear while model unavailable", "off while model unavailable", "clear during recovery restore", "replacement during recovery restore"])("Ready forks once and preserves lifecycle ownership: %s", async (scenario) => { + it.each(["completion", "compaction failure", "replacement during activation", "steward off during activation", "missing worker model", "unauthenticated worker model", "clear while model unavailable", "off while model unavailable", "clear during recovery restore", "replacement during recovery restore"])("Ready forks once and preserves lifecycle ownership: %s", async (scenario) => { const cwd = mkdtempSync(join(tmpdir(), "goals-supervisor-integration-")); const peers: any[] = []; const wires: any[] = []; @@ -82,8 +82,8 @@ describe("actual goals and supervisor package hooks (Herdr and judge mocked)", ( }; const ctx: any = { cwd, hasUI: true, isIdle: () => true, model: { provider: "offline", id: (manager.getBranch().findLast((entry: any) => entry.type === "model_change") as any)?.modelId ?? "test", contextWindow: 200_000 }, sessionManager: manager, modelRegistry: { find: (provider: string, id: string) => peer.missing && id === "worker" ? undefined : ({ provider, id, contextWindow: 200_000 }) }, - getContextUsage: () => ({ tokens: 50_000 }), compact({ onComplete }: any) { peer.compactions++; peer.compactionModels.push(ctx.model.id); onComplete({}); }, abort() { peer.aborts++; }, - ui: { theme: { fg: (_: string, text: string) => text }, setWidget() {}, setStatus() {}, notify: vi.fn(), select: async () => "Ready" }, + getContextUsage: () => ({ tokens: 50_000 }), compact({ onComplete, onError }: any) { peer.compactions++; peer.compactionModels.push(ctx.model.id); if (scenario === "compaction failure" && id === "supervisor") onError(new Error("Native compaction timeout")); else onComplete({}); }, abort() { peer.aborts++; }, + ui: { theme: { fg: (_: string, text: string) => text }, setWidget: vi.fn(), setStatus: vi.fn(), notify: vi.fn(), select: async () => "Ready" }, }; peer.pi = pi; peer.ctx = ctx; peer.hook = async (name: string, event = {}) => { for (const fn of hooks.get(name) ?? []) await fn(event, ctx); }; peers.push(peer); goals(pi); return peer; @@ -129,6 +129,19 @@ describe("actual goals and supervisor package hooks (Herdr and judge mocked)", ( } await starting; await tick(); const supervisor = peers[1]; + if (scenario === "compaction failure") { + const saved = manager.getBranch().findLast((entry: any) => entry.customType === "pi-goals-state") as any; + expect(saved.data).toMatchObject({ phase: "planning", reviewRequested: false, startupError: expect.stringContaining("Native compaction timeout") }); + expect(worker.ctx.ui.setStatus).toHaveBeenCalledWith("pi-goals", "supervisor startup failed"); + expect(worker.messages.some((m: string) => m.startsWith("Work the goals"))).toBe(false); + expect(wires.some(w => w.t === "plan_failed")).toBe(true); + expect(wires.some(w => w.t === "plan_activate")).toBe(false); + await worker.hook("agent_settled"); + expect(herdrCalls.filter(args => args[1] === "start")).toHaveLength(1); + await worker.hook("session_start", { reason: "reload" }); + expect(worker.ctx.ui.setStatus).toHaveBeenCalledWith("pi-goals", "supervisor startup failed"); + return; + } if (worker.missing || worker.noAuth) { const saved = manager.getBranch().findLast((entry: any) => entry.customType === "pi-goals-state") as any; expect(saved.data).toMatchObject({ phase: "planning", modelRecovery: "worker" });