wassname/pi-goals
1
0
Fork 0
mirror of https://github.com/wassname/pi-goals.git synced 2026-09-12 12:50:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

WIP: preserve startup failure and stale-pane recovery

Browse Source 
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.
This commit is contained in:
wassname2
2026-09-09 19:48:10 +08:00
parent 097349632c
commit 43fc97aebd
6 changed files with 114 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/index.ts
+8 -5
View File
@@ -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,
src/internal/supervisor/index.ts
+28 -4
View File
@@ -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<SupervisorBinding>(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;
}
src/internal/supervisor/protocol.ts
+2
View File
@@ -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";
src/supervisor.ts
+42 -7
View File
@@ -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<SupervisorStatus>;
prepare(binding: SupervisorBinding, signal?: AbortSignal): Promise<void>;
bootstrap(bootstrap: Bootstrap, signal?: AbortSignal): Promise<SupervisorBinding>;
attached(bindingId: string, signal?: AbortSignal): Promise<SupervisorBinding>;
attached(bindingId: string, signal?: AbortSignal, expected?: SupervisorBinding): Promise<SupervisorBinding>;
activate(bindingId: string, signal?: AbortSignal): Promise<void>;
review(bindingId: string, goal: string, hash: string, signal?: AbortSignal): Promise<SupervisorDecision>;
stop(bindingId: string): Promise<void>;
@@ -73,12 +74,21 @@ export function pendingReply<T>(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<Record<string, any>> {
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<string, any>;
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);
test/internal-supervisor/plan.test.ts
+18
View File
@@ -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 {
test/supervisor-integration.test.ts
+16 -3
View File
@@ -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" });