mirror of
https://github.com/wassname/pi-plan.git
synced 2026-09-26 14:10:23 +08:00
Record extra subagents as helpers instead of losing the worker binding
A read-only reviewer launch no longer overwrites the implementation worker (reported by maniworker session 01a0809b): the first launch binds, resume of the same session refreshes it, later launches land in state.helpers and show in /goals status. launchPending is now a counter so concurrent launches keep takeover menus honest. Old persisted state migrates with an empty helper list. Co-Authored-By: Pi/OpenAI <288921227+claudypoo@users.noreply.github.com>
This commit is contained in:
+17
-8
@@ -46,11 +46,12 @@ interface State {
|
||||
mode: Mode;
|
||||
plan?: string;
|
||||
worker?: { id?: string; sessionFile: string };
|
||||
helpers: { id?: string; sessionFile: string }[];
|
||||
workerStopped?: boolean;
|
||||
signoffs: Record<string, { evidence: string[]; observation: string }>;
|
||||
child?: boolean;
|
||||
}
|
||||
const initial = (): State => ({ mode: "chat", signoffs: {} });
|
||||
const initial = (): State => ({ mode: "chat", helpers: [], signoffs: {} });
|
||||
const digest = (text: string) => createHash("sha256").update(text).digest("hex");
|
||||
const key = (text: string) => text.trim().toLowerCase();
|
||||
function goals(text: string) {
|
||||
@@ -67,7 +68,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
let state = initial();
|
||||
let generation = 0;
|
||||
let workerRevision = 0;
|
||||
let launchPending = false;
|
||||
let pendingLaunches = 0;
|
||||
let notice = true;
|
||||
let planWatcher: FSWatcher | undefined;
|
||||
let planEditTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
@@ -164,6 +165,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
// Lineage-only workers attach the explicit task path using AttachGoalPlan.
|
||||
save();
|
||||
}
|
||||
state.helpers ??= []; // sessions persisted before helper bookkeeping
|
||||
notice = true;
|
||||
turnsStale = 0;
|
||||
lastWorkingSet = "";
|
||||
@@ -182,7 +184,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
else pi.sendMessage({ customType: "pi-goals-supervision", content, display: true }, { deliverAs: "followUp", triggerTurn: false });
|
||||
}
|
||||
async function confirmOwnership(ctx: ExtensionContext, target: string, text: string, solo = true): Promise<boolean> {
|
||||
if (launchPending) { ctx.ui.notify("Worker launch/resume is still pending; inspect its result before takeover.", "warning"); return false; }
|
||||
if (pendingLaunches > 0) { ctx.ui.notify("A worker launch/resume is still pending; inspect its result before takeover.", "warning"); return false; }
|
||||
const stamp = generation;
|
||||
const revision = workerRevision;
|
||||
const confirmation = solo ? "Worker confirmed stopped" : "Previous supervisor confirmed stopped";
|
||||
@@ -276,14 +278,20 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
if (state.child || !["subagent", "subagent_resume"].includes(event.toolName)) return;
|
||||
// Solo means this chat took over implementation: no concurrent writer may be delegated.
|
||||
if (state.mode === "planning" || state.mode === "paused" || state.mode === "solo") return { block: true, reason: goalToolBlocked(state.mode) };
|
||||
if (state.plan) { launchPending = true; state.workerStopped = false; workerRevision++; save(); }
|
||||
if (state.plan) { pendingLaunches++; state.workerStopped = false; workerRevision++; save(); }
|
||||
});
|
||||
pi.on("tool_result", (event) => {
|
||||
if (state.child || !state.plan || !["subagent", "subagent_resume"].includes(event.toolName)) return;
|
||||
launchPending = false;
|
||||
pendingLaunches = Math.max(0, pendingLaunches - 1);
|
||||
if (event.isError) return;
|
||||
const details = event.details as { id?: string; sessionFile?: string } | undefined;
|
||||
if (details?.id && details.sessionFile) { state.worker = { id: details.id, sessionFile: details.sessionFile }; state.workerStopped = false; workerRevision++; save(); }
|
||||
if (!details?.id || !details.sessionFile) return;
|
||||
const record = { id: details.id, sessionFile: details.sessionFile };
|
||||
if (state.worker?.sessionFile === record.sessionFile) state.worker = record;
|
||||
else if (!state.worker) state.worker = record;
|
||||
// Extra launches stay recorded as helpers; the implementation binding never moves silently.
|
||||
else state.helpers = [...(state.helpers ?? []).filter((h) => h.sessionFile !== record.sessionFile), record];
|
||||
state.workerStopped = false; workerRevision++; save();
|
||||
});
|
||||
|
||||
pi.registerCommand("goals", {
|
||||
@@ -313,6 +321,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
`Plan: ${state.plan ?? "none"}`,
|
||||
`Preferred worker model (plan): ${notedPlanValue("preferred worker model") ?? "not stated; use /goals model <model>"}`,
|
||||
`Recorded worker session: ${state.worker?.sessionFile ?? "not recorded"}`,
|
||||
`Helper subagent sessions: ${state.helpers.length} recorded (liveness via /subagents)`,
|
||||
notedPlanValue("worker session") ? `Worker session noted in plan: ${notedPlanValue("worker session")}` : "",
|
||||
`Hourly check-in: schedule_prompt job ${JSON.stringify(`goals-${ctx.sessionManager.getSessionId()}`)} (list/remove via schedule_prompt; plan-change reviews are the plan-watcher event hook)`,
|
||||
"Liveness is owned by edxeth; inspect /subagents.",
|
||||
@@ -351,7 +360,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
if (!(await confirmOwnership(ctx, target, text, solo))) return;
|
||||
const retained = target === state.plan ? state.signoffs : {};
|
||||
const worker = noted ? { sessionFile: resolve(ctx.cwd, noted) } : state.workerStopped ? state.worker : undefined;
|
||||
state = { mode: solo ? "solo" : "planning", plan: target, signoffs: retained, worker, workerStopped: solo || (!noted && state.workerStopped) };
|
||||
state = { mode: solo ? "solo" : "planning", plan: target, signoffs: retained, worker, helpers: [], workerStopped: solo || (!noted && state.workerStopped) };
|
||||
generation++; notice = true; save(); refresh(ctx); watchPlan(ctx);
|
||||
if (solo) enterSolo(ctx);
|
||||
else send(attachNotice(target, false, noted));
|
||||
@@ -389,7 +398,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
mkdirSync(dirname(path), { recursive: true });
|
||||
// Never overwrite an earlier plan at this session path; the model can revise it after inspection.
|
||||
try { writeFileSync(path, planDocument(objective), { flag: "wx" }); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; }
|
||||
state = { mode: "planning", plan: path, signoffs: {}, worker: state.worker, workerStopped: state.workerStopped }; generation++; notice = true; save(); refresh(ctx); watchPlan(ctx);
|
||||
state = { mode: "planning", plan: path, signoffs: {}, worker: state.worker, helpers: state.helpers, workerStopped: state.workerStopped }; generation++; notice = true; save(); refresh(ctx); watchPlan(ctx);
|
||||
send(planningSeed(objective, path));
|
||||
} catch (error) { ctx.ui.notify(String(error), "error"); }
|
||||
},
|
||||
|
||||
@@ -535,6 +535,36 @@ it.each(["solo", "supervising"])("%s upkeep is turn-driven, folds Log, resets on
|
||||
expect(reminders()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("extra subagent launches are recorded as helpers and never steal the implementation identity", async () => {
|
||||
const f = fixture(); await f.draft(); await f.command("ready");
|
||||
f.hooks.get("tool_result")({ toolName: "subagent", details: { id: "impl", sessionFile: "/tmp/impl.jsonl" } });
|
||||
expect(f.entries.at(-1).data).toMatchObject({ worker: { id: "impl", sessionFile: "/tmp/impl.jsonl" }, helpers: [] });
|
||||
f.hooks.get("tool_result")({ toolName: "subagent", details: { id: "reviewer", sessionFile: "/tmp/review.jsonl" } });
|
||||
expect(f.entries.at(-1).data).toMatchObject({ worker: { id: "impl" }, helpers: [{ id: "reviewer", sessionFile: "/tmp/review.jsonl" }] });
|
||||
// a repeated helper launch updates its record instead of duplicating it
|
||||
f.hooks.get("tool_result")({ toolName: "subagent", details: { id: "reviewer-2", sessionFile: "/tmp/review.jsonl" } });
|
||||
expect(f.entries.at(-1).data.helpers).toEqual([{ id: "reviewer-2", sessionFile: "/tmp/review.jsonl" }]);
|
||||
// resuming the worker keeps the binding and refreshes its id
|
||||
f.hooks.get("tool_result")({ toolName: "subagent_resume", details: { id: "impl-2", sessionFile: "/tmp/impl.jsonl" } });
|
||||
expect(f.entries.at(-1).data).toMatchObject({ worker: { id: "impl-2", sessionFile: "/tmp/impl.jsonl" }, helpers: [{ id: "reviewer-2" }] });
|
||||
});
|
||||
|
||||
it("pending launch counter survives concurrent launches until every result lands", async () => {
|
||||
const f = fixture(); await f.draft(); await f.command("ready");
|
||||
f.hooks.get("tool_call")({ toolName: "subagent" });
|
||||
f.hooks.get("tool_call")({ toolName: "subagent" });
|
||||
f.hooks.get("tool_result")({ toolName: "subagent", details: { id: "a", sessionFile: "/tmp/a.jsonl" } });
|
||||
f.ctx.ui.select.mockResolvedValueOnce("Worker confirmed stopped");
|
||||
await f.command("solo");
|
||||
expect(f.entries.at(-1).data.mode).toBe("supervising"); // one launch still pending
|
||||
expect(f.ctx.notify ?? f.ctx.ui.notify).toHaveBeenLastCalledWith(expect.stringContaining("still pending"), "warning");
|
||||
f.hooks.get("tool_result")({ toolName: "subagent", details: { id: "b", sessionFile: "/tmp/b.jsonl" } });
|
||||
f.ctx.ui.select.mockResolvedValueOnce("Worker confirmed stopped");
|
||||
await f.command("solo");
|
||||
expect(f.entries.at(-1).data.mode).toBe("solo");
|
||||
expect(f.entries.at(-1).data).toMatchObject({ worker: { id: "a" }, helpers: [{ id: "b" }] });
|
||||
});
|
||||
|
||||
it("late worker results invalidate a takeover menu but do not disable plan watching", async () => {
|
||||
const f = fixture(); await f.draft(); await f.command("ready");
|
||||
let answer!: (choice: string) => void;
|
||||
|
||||
Reference in New Issue
Block a user