mirror of
https://github.com/wassname/pi-plan.git
synced 2026-09-25 14:00:15 +08:00
feat: add guarded native worker recovery and fresh sessions
This commit is contained in:
wassname2
@@ -149,11 +149,15 @@ pi
|
||||
|
||||
`/goals` shows actions for the current mode. Drafts offer Edit, Discuss and Approve. Discuss returns to chat and waits for your input. Menu New asks for optional instructions before creating a plan; submit blank to use the conversation, or cancel to leave things unchanged. Typed `/goals new <instructions>` still starts directly. Quit (`exit` or `clear`) leaves the original plan unchanged, removes this session's goal check-in, and clears goal state without a model call. Matching check-in names with missing or different session bindings are left unchanged with a warning. Worker processes are unchanged; inspect their native panes and use their exact Intercom identities for steering. New creates a separate draft without overwriting earlier plans, named `.pi/plan/<last-six-session-characters>-vN.md` using the next version after existing files. The title stays inside the plan; old files are not renamed. The widget shows a plain `✓` and the relative plan path (the fallback for unverified terminal links).
|
||||
|
||||
### First-session port limits
|
||||
### Native worker lifecycle and limits
|
||||
|
||||
The parent and worker keep separate native conversations. Worker attachment and stop notices use stock Intercom extension channels; assignments, reports and corrections remain visible Pi messages. A pane-open receipt, idle state or delivery receipt does not approve a goal.
|
||||
|
||||
This port does not yet implement automatic saved-session recovery or later fresh-session replacement. Nico reuses one project binding; an existing conversation is never reset or closed to make room. A recorded worker therefore blocks another `OpenGoalWorker`, including after a confirmed stop. Preserve its saved session and inspect liveness before manual recovery or confirmed solo takeover. `project.open` has no model override; choose a model in the worker's native `/model` UI and verify the resolved choice. — Pi/OpenAI
|
||||
`OpenGoalWorker` opens a blank peer and waits for verified Intercom capability before sending work. For independent work after review, use `action: "fresh"` with the exact inspected `reviewedThrough` entry ID. This uses Pi's new session in the same pane; the previous conversation stays in saved history. Revisions still use the same Intercom session. Drafts, pending input, changed history and a local worker pause block replacement.
|
||||
|
||||
For recovery, use `action: "recover"`, `writersStopped: true` and the owned saved session after inspecting other writers. Recovery restores context without replaying a task or changing its model. A prospective session path is not durable history. A live binding without a responsive, capable Pi peer remains unconfirmed; no shell restart or second backend is invented.
|
||||
|
||||
Without an explicit model preference, a new context uses the current Pi profile's normal defaults. A session-local human choice remains in that earlier session's history; recovery retains it. Requested-model automation currently fails closed: Pi's asynchronous public setter lacks a guard against overwriting a concurrent human selection. No fallback task is launched and no stale preference is reapplied. This model-selection requirement remains unfinished. — Pi/OpenAI
|
||||
|
||||
## Context delivery
|
||||
|
||||
|
||||
+186
-33
@@ -50,19 +50,30 @@ import {
|
||||
|
||||
const STATE = "pi-goals-main-supervisor-v1";
|
||||
const WORKER = "goals-worker";
|
||||
const CONTROL = "goals-worker-control";
|
||||
const WIDGET_GOAL_LIMIT = 3;
|
||||
type Mode = "chat" | "planning" | "supervising" | "paused" | "solo";
|
||||
type GoalStatus = "open" | "active" | "done" | "cancelled";
|
||||
interface Peer {
|
||||
sessionId: string; sessionFile: string; leafId: string | null; paneId: string;
|
||||
durable: boolean; empty: boolean; started?: boolean; parentSession?: string; plan?: string; parentId?: string; requestId?: string; model?: string;
|
||||
}
|
||||
interface WorkerRequest {
|
||||
id: string; action: "start" | "fresh" | "recover"; task?: string; model?: string;
|
||||
reviewedThrough?: string; writersStopped?: boolean; sessionFile?: string; savedDigest?: string; savedId?: string; savedIntercom?: string;
|
||||
phase: "probe" | "control" | "switch"; previous?: Peer;
|
||||
}
|
||||
interface State {
|
||||
mode: Mode;
|
||||
plan?: string;
|
||||
worker?: { sessionFile?: string; intercomId?: string; paneId?: string; requestId?: string; parentId?: string };
|
||||
parent?: { intercomId: string; requestId: string };
|
||||
worker?: { sessionFile?: string; intercomId?: string; paneId?: string; requestId?: string; parentId?: string; identity?: Peer; pending?: WorkerRequest };
|
||||
parent?: { intercomId: string; requestId: string; selfId?: string; started?: boolean };
|
||||
workerStopped?: boolean;
|
||||
pausedFrom?: "solo" | "supervising";
|
||||
signoffs: Record<string, { evidence: string[]; observation: string; signature: string }>;
|
||||
finalReview?: { planDigest: string };
|
||||
child?: boolean;
|
||||
lastControl?: string;
|
||||
}
|
||||
const initial = (): State => ({ mode: "chat", signoffs: {} });
|
||||
const digest = (text: string) => createHash("sha256").update(text).digest("hex");
|
||||
@@ -76,6 +87,15 @@ function goals(text: string) {
|
||||
});
|
||||
}
|
||||
const requirements = (text: string) => goals(text).map(g => goalAcceptanceSignature(text, g.subject)).join("\n");
|
||||
function savedWorker(path: string) {
|
||||
const text = readFileSync(path, "utf8");
|
||||
const entries = text.trim().split("\n").map(line => JSON.parse(line));
|
||||
const header = entries[0];
|
||||
if (header?.type !== "session" || typeof header.id !== "string" || !entries.some(entry => entry.message?.role === "assistant")) throw new Error(nativeMessages.notDurable);
|
||||
const state = entries.filter(entry => entry.type === "custom" && entry.customType === STATE).at(-1)?.data as State | undefined;
|
||||
if (!state?.child || !state.parent) throw new Error(nativeMessages.notOwned);
|
||||
return { header, state, digest: digest(text) };
|
||||
}
|
||||
const result = (text: string) => ({ content: [{ type: "text" as const, text }], details: {} });
|
||||
|
||||
export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
@@ -86,6 +106,10 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
let finalReviewTurnDigest: string | undefined;
|
||||
let opening = false;
|
||||
let channel: IntercomExtensionChannel | undefined;
|
||||
let ownIntercomId: string | undefined;
|
||||
let liveContext: ExtensionContext | undefined;
|
||||
let control: { from: string; request: WorkerRequest; plan: string; expected: Peer; cancelled?: boolean } | undefined;
|
||||
let replacing: "new" | "resume" | undefined;
|
||||
let notice = true;
|
||||
let fullPlanContextDue = true;
|
||||
let planWatcher: FSWatcher | undefined;
|
||||
@@ -223,7 +247,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
} else pi.sendMessage({ customType: "pi-goals-supervision", content, display: true }, { deliverAs: "nextTurn" });
|
||||
}
|
||||
async function confirmOwnership(ctx: ExtensionContext, target: string, text: string, solo = true): Promise<boolean> {
|
||||
if (opening) { ctx.ui.notify("A worker launch/resume is still pending; inspect its result before takeover.", "warning"); return false; }
|
||||
if (opening || state.worker?.pending) { 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";
|
||||
@@ -262,40 +286,148 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
send(`${checkIn(ctx)}\n\n${readyApproved(WORKER, state.plan!, state.worker?.sessionFile, text, ctx.sessionManager.getSessionId())}`);
|
||||
}
|
||||
|
||||
function registerChannel() {
|
||||
function identity(ctx: ExtensionContext): Peer {
|
||||
const sessionFile = ctx.sessionManager.getSessionFile() ?? "";
|
||||
let durable = false;
|
||||
try { durable = savedWorker(sessionFile).header.id === ctx.sessionManager.getSessionId(); } catch { /* A prospective path is not saved history. */ }
|
||||
return { sessionId: ctx.sessionManager.getSessionId(), sessionFile, leafId: ctx.sessionManager.getLeafId(), paneId: process.env.HERDR_PANE_ID ?? "", durable,
|
||||
empty: !ctx.sessionManager.getBranch().some(entry => entry.type === "message"), parentSession: ctx.sessionManager.getHeader()?.parentSession,
|
||||
plan: state.plan, parentId: state.parent?.intercomId, requestId: state.parent?.requestId, started: state.parent?.started, model: ctx.model ? ctx.model.provider + "/" + ctx.model.id : undefined };
|
||||
}
|
||||
const publish = (payload: unknown) => { channel?.publish(payload, { audience: "capable" }); };
|
||||
async function available() {
|
||||
const stamp = generation, ctx = liveContext, current = channel;
|
||||
if (!ctx || !current?.snapshot().connected || !current.snapshot().supported) return;
|
||||
const peers = await current.listSessions().catch(() => []);
|
||||
if (stamp !== generation || current !== channel) return;
|
||||
const self = peers.filter(peer => peer.pid === process.pid);
|
||||
if (self.length === 1 && pi.getCommands().some(command => command.name === CONTROL)) { ownIntercomId = self[0].id; publish({ type: "available", identity: identity(ctx) }); }
|
||||
}
|
||||
function probe() {
|
||||
if (state.mode === "supervising" && state.worker?.pending && state.worker.paneId) publish({ type: "inspect", paneId: state.worker.paneId, requestId: state.worker.pending.id });
|
||||
}
|
||||
function cancelControl() {
|
||||
if (control) control.cancelled = true;
|
||||
if (state.worker?.pending) {
|
||||
try { publish({ type: "cancel", paneId: state.worker.paneId, requestId: state.worker.pending.id }); } catch { /* Local pause still takes effect when the peer is disconnected. */ }
|
||||
state.worker.requestId = state.worker.pending.previous?.requestId ?? state.worker.requestId;
|
||||
state.worker.pending = undefined;
|
||||
}
|
||||
}
|
||||
function registerChannel(ctx: ExtensionContext) {
|
||||
liveContext = ctx;
|
||||
const registration: IntercomExtensionRegistration = {
|
||||
namespace: "pi-goals", ownerEligible: false,
|
||||
onReady: (value) => { channel = value; },
|
||||
onReady: (value) => { channel = value; void available(); },
|
||||
onEvent: (event) => {
|
||||
if (event.type === "session_left" && event.sessionId === state.worker?.intercomId && state.mode === "supervising") {
|
||||
send(workerReview(state.plan!, event.sessionId, nativeMessages.disconnected));
|
||||
if (event.type === "connection" && event.connected) { void available(); probe(); }
|
||||
if (event.type === "session_left" && event.sessionId === state.worker?.intercomId && state.mode === "supervising" && state.worker.pending?.phase !== "switch") send(workerReview(state.plan!, event.sessionId, nativeMessages.disconnected));
|
||||
if (event.type !== "message" || !event.payload || typeof event.payload !== "object") return;
|
||||
const data = event.payload as { type?: string; to?: string; requestId?: string; plan?: string; paneId?: string; sessionFile?: string; text?: string; identity?: Peer; request?: WorkerRequest; expected?: Peer };
|
||||
if (data.type === "inspect" && data.paneId && data.paneId === process.env.HERDR_PANE_ID && typeof data.requestId === "string") {
|
||||
publish({ type: "peer", to: event.fromSessionId, requestId: data.requestId, identity: identity(ctx) }); return;
|
||||
}
|
||||
if (event.type !== "message" || state.child || !state.worker?.requestId) return;
|
||||
const data = event.payload as { type?: string; to?: string; requestId?: string; plan?: string; sessionFile?: string; text?: string } | null;
|
||||
if (!data || data.to !== state.worker.parentId || event.fromSessionId === data.to || data.requestId !== state.worker.requestId || data.plan !== state.plan) return;
|
||||
if (data.type === "attached" && !state.worker.intercomId && typeof data.sessionFile === "string" && isAbsolute(data.sessionFile)) {
|
||||
state.worker.intercomId = event.fromSessionId; state.worker.sessionFile = data.sessionFile; workerRevision++; save();
|
||||
send(workerReview(state.plan!, event.fromSessionId, nativeMessages.attached(data.sessionFile)), false);
|
||||
if (data.type === "cancel" && data.paneId === process.env.HERDR_PANE_ID && control?.from === event.fromSessionId && data.requestId === control.request.id) { control.cancelled = true; return; }
|
||||
if (data.type === "control" && ownIntercomId && data.to === ownIntercomId && data.expected && data.expected.paneId === process.env.HERDR_PANE_ID && data.request && ["start", "fresh", "recover"].includes(data.request.action) && typeof data.request.id === "string" && typeof data.plan === "string") {
|
||||
if (control || !pi.getCommands().some(command => command.name === CONTROL)) return;
|
||||
control = { from: event.fromSessionId, request: data.request, plan: data.plan, expected: data.expected };
|
||||
pi.sendUserMessage("/" + CONTROL, { expandPromptTemplates: true, deliverAs: "followUp" }); return;
|
||||
}
|
||||
if (data.type === "stopped" && event.fromSessionId === state.worker.intercomId && typeof data.text === "string") {
|
||||
send(workerReview(state.plan!, event.fromSessionId, data.text), state.mode === "supervising");
|
||||
const worker = state.worker, pending = worker?.pending;
|
||||
if (state.child || !worker || !state.plan) return;
|
||||
if (data.type === "available" && data.identity?.paneId === worker.paneId && pending) { probe(); return; }
|
||||
if (data.type === "peer" && pending && data.identity && data.to === worker.parentId && data.requestId === pending.id && data.identity.paneId === worker.paneId && state.mode === "supervising") {
|
||||
const peer = data.identity;
|
||||
if (event.fromSessionId === worker.parentId) return;
|
||||
if (peer.parentId === worker.parentId && peer.requestId === pending.id && peer.started) { worker.intercomId = event.fromSessionId; worker.identity = peer; worker.sessionFile = peer.sessionFile; worker.requestId = pending.id; worker.pending = undefined; save(); send(workerReview(state.plan, event.fromSessionId, nativeMessages.actionApplied("observed without replay", peer))); return; }
|
||||
if (pending.phase === "control") return;
|
||||
if (pending.phase === "probe" && pending.action === "fresh" && (peer.sessionId !== pending.previous?.sessionId || peer.sessionFile !== pending.previous?.sessionFile)) { worker.pending = undefined; save(); send(workerReview(state.plan, event.fromSessionId, nativeMessages.controlChanged)); return; }
|
||||
if (pending.phase === "switch") {
|
||||
const arrived = pending.action === "fresh" ? peer.sessionId !== pending.previous?.sessionId && peer.parentSession === pending.previous?.sessionFile && peer.requestId === pending.id : peer.sessionId === pending.savedId && peer.sessionFile === pending.sessionFile;
|
||||
if (!arrived) return;
|
||||
}
|
||||
worker.intercomId = event.fromSessionId; worker.identity = peer; worker.sessionFile = peer.sessionFile;
|
||||
const phase = pending.phase; pending.phase = "control"; workerRevision++; save();
|
||||
publish({ type: "control", to: event.fromSessionId, plan: state.plan, expected: peer, request: { ...pending, action: phase === "switch" || pending.action === "recover" && peer.sessionId === pending.savedId ? "start" : pending.action } }); return;
|
||||
}
|
||||
if (!worker.parentId || !data.requestId || data.to !== worker.parentId || event.fromSessionId === data.to || (data.requestId !== worker.requestId && data.requestId !== pending?.id) || data.plan !== state.plan) return;
|
||||
if (data.type === "switching" && event.fromSessionId === worker.intercomId && pending && data.requestId === pending.id) { pending.phase = "switch"; save(); return; }
|
||||
if (data.type === "rejected" && event.fromSessionId === worker.intercomId && pending && data.requestId === pending.id) { worker.requestId = pending.previous?.requestId ?? worker.requestId; worker.pending = undefined; save(); send(workerReview(state.plan, event.fromSessionId, data.text ?? nativeMessages.controlRejected), state.mode === "supervising"); return; }
|
||||
if (data.type === "attached" && typeof data.sessionFile === "string" && isAbsolute(data.sessionFile) && (!worker.intercomId || worker.intercomId === event.fromSessionId)) {
|
||||
if (worker.pending && data.requestId !== worker.pending.id) { send(workerReview(state.plan, event.fromSessionId, nativeMessages.attached(data.sessionFile)), false); return; }
|
||||
const action = worker.pending?.action;
|
||||
worker.requestId = data.requestId;
|
||||
worker.intercomId = event.fromSessionId; worker.sessionFile = data.sessionFile; if (data.identity) worker.identity = data.identity; worker.pending = undefined; workerRevision++; save();
|
||||
send(workerReview(state.plan, event.fromSessionId, action ? nativeMessages.actionApplied(action, data.identity) : nativeMessages.attached(data.sessionFile)), Boolean(action) && state.mode === "supervising");
|
||||
}
|
||||
if (data.type === "stopped" && event.fromSessionId === worker.intercomId && typeof data.text === "string") {
|
||||
if (data.identity) { worker.identity = data.identity; worker.sessionFile = data.identity.sessionFile; save(); }
|
||||
send(workerReview(state.plan, event.fromSessionId, data.text), state.mode === "supervising");
|
||||
}
|
||||
},
|
||||
};
|
||||
pi.events.emit(INTERCOM_EXTENSION_REGISTER_EVENT, registration);
|
||||
}
|
||||
pi.registerCommand(CONTROL, {
|
||||
description: nativeMessages.controlDescription,
|
||||
handler: async (_args, ctx) => {
|
||||
const operation = control;
|
||||
if (!operation) return; // Never execute arbitrary slash-command payloads.
|
||||
const { request, expected, from, plan } = operation;
|
||||
const controlKey = request.id + ":" + request.action;
|
||||
if (state.lastControl === controlKey) { control = undefined; return; }
|
||||
let authorized = false;
|
||||
const reject = (text: string) => publish({ type: "rejected", to: from, requestId: request.id, plan, text });
|
||||
try {
|
||||
await ctx.waitForIdle();
|
||||
const current = identity(ctx);
|
||||
if (operation.cancelled || state.mode === "paused" && !(request.action === "start" && request.savedId === current.sessionId) || ctx.hasPendingMessages() || ctx.ui.getEditorText().length || current.sessionId !== expected.sessionId || current.leafId !== expected.leafId || current.sessionFile !== expected.sessionFile) throw new Error(nativeMessages.controlChanged);
|
||||
if (!isAbsolute(plan) || !goals(readFileSync(plan, "utf8")).length) throw new Error(messages.invalidAttachment);
|
||||
const peers = await channel?.listSessions();
|
||||
const self = peers?.find(peer => peer.pid === process.pid);
|
||||
if (!self || from === self.id || !peers?.some(peer => peer.id === from)) throw new Error(nativeMessages.parentUnavailable);
|
||||
if (operation.cancelled || ctx.hasPendingMessages() || ctx.ui.getEditorText().length || identity(ctx).leafId !== expected.leafId) throw new Error(nativeMessages.controlChanged);
|
||||
if (state.child && state.parent?.intercomId !== from || !state.child && (state.mode !== "chat" || !current.empty)) throw new Error(nativeMessages.notOwned);
|
||||
authorized = true;
|
||||
if (request.action === "start") {
|
||||
const recovering = request.savedId === current.sessionId;
|
||||
if (!recovering && (!current.empty || state.parent?.started)) throw new Error(nativeMessages.controlChanged);
|
||||
if (request.model) throw new Error(nativeMessages.modelRaceBoundary);
|
||||
state = { ...(recovering ? state : initial()), mode: recovering ? state.mode : "solo", child: true, plan, lastControl: controlKey, parent: { intercomId: from, requestId: request.id, selfId: self.id, started: true } };
|
||||
generation++; notice = true; fullPlanContextDue = true; save();
|
||||
publish({ type: "attached", to: from, requestId: request.id, plan, sessionFile: ctx.sessionManager.getSessionFile(), identity: identity(ctx) });
|
||||
if (!recovering && request.task) pi.sendUserMessage(workerAssignment(plan, from, request.id, request.task));
|
||||
} else {
|
||||
if (!current.empty && (!current.durable || request.reviewedThrough !== current.leafId)) throw new Error(nativeMessages.reviewRequired);
|
||||
if (request.action === "fresh" && (!state.child || !current.durable)) throw new Error(nativeMessages.notDurable);
|
||||
if (request.action === "recover") {
|
||||
if (!request.writersStopped || !request.sessionFile || !request.savedIntercom) throw new Error(nativeMessages.stopRequired);
|
||||
const saved = savedWorker(request.sessionFile);
|
||||
if (saved.digest !== request.savedDigest || saved.header.id !== request.savedId || saved.header.cwd !== ctx.cwd || saved.state.parent?.intercomId !== from || peers.some(peer => peer.id === request.savedIntercom && peer.id !== self.id)) throw new Error(nativeMessages.controlChanged);
|
||||
}
|
||||
publish({ type: "switching", to: from, requestId: request.id, plan });
|
||||
replacing = request.action === "fresh" ? "new" : "resume";
|
||||
const switched = request.action === "fresh"
|
||||
? await ctx.newSession({ parentSession: current.sessionFile, setup: async manager => { manager.appendCustomEntry(STATE, { ...initial(), mode: "solo", child: true, plan, parent: { intercomId: from, requestId: request.id } }); } })
|
||||
: await ctx.switchSession(request.sessionFile!);
|
||||
if (switched.cancelled) { replacing = undefined; state.lastControl = controlKey; save(); reject(nativeMessages.controlCancelled); }
|
||||
}
|
||||
} catch (error) { if (authorized) { state.lastControl = controlKey; save(); } reject(String(error)); }
|
||||
finally { replacing = undefined; control = undefined; }
|
||||
},
|
||||
});
|
||||
const reportStop = (text: string) => {
|
||||
if (!state.child || !state.parent) return;
|
||||
try {
|
||||
if (!channel?.snapshot().connected) throw new Error("disconnected");
|
||||
channel.publish({ type: "stopped", to: state.parent.intercomId, requestId: state.parent.requestId, plan: state.plan, text }, { audience: "capable" });
|
||||
channel.publish({ type: "stopped", to: state.parent.intercomId, requestId: state.parent.requestId, plan: state.plan, text, identity: liveContext ? identity(liveContext) : undefined }, { audience: "capable" });
|
||||
} catch { send(nativeMessages.reportUnavailable, false); }
|
||||
};
|
||||
pi.on("session_start", (_e, ctx) => {
|
||||
restore(ctx); registerChannel();
|
||||
restore(ctx); registerChannel(ctx);
|
||||
});
|
||||
pi.on("session_tree", (_e, ctx) => restore(ctx));
|
||||
pi.on("session_shutdown", () => { reportStop(nativeMessages.shuttingDown); channel = undefined; generation++; finalReviewTurnDigest = undefined; planWatcher?.close(); planWatcher = undefined; clearTimeout(planEditTimer); planEditTimer = undefined; });
|
||||
pi.on("session_shutdown", (event) => { if (!replacing || event?.reason !== replacing) reportStop(nativeMessages.shuttingDown); channel = undefined; liveContext = undefined; ownIntercomId = undefined; generation++; finalReviewTurnDigest = undefined; planWatcher?.close(); planWatcher = undefined; clearTimeout(planEditTimer); planEditTimer = undefined; });
|
||||
// Only successful compaction needs resync; failed/cancelled attempts leave pending context alone.
|
||||
// Defer to prompt preparation: same-run continuation retains Pi's current role/context.
|
||||
pi.on("session_compact", () => { notice = true; fullPlanContextDue = true; });
|
||||
@@ -351,7 +483,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
return { systemPrompt: `${event.systemPrompt}\n\n${state.child ? childPlanRole : ""}\n${snapshot.error}` };
|
||||
}
|
||||
clearChangedFinalReview(snapshot.text);
|
||||
const role = state.child ? childPlanRole : state.mode === "supervising"
|
||||
const role = state.child ? childPlanRole + (state.mode === "paused" ? "\n" + pausedRole : "") : state.mode === "supervising"
|
||||
? supervisor(WORKER, state.plan!, ctx.sessionManager.getSessionId())
|
||||
: state.mode === "planning" ? planning(state.plan!) : state.mode === "paused" ? pausedRole : soloRole;
|
||||
fullPlanContextDue ||= requirements(snapshot.text) !== requirements(lastWorkingSet);
|
||||
@@ -384,7 +516,10 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
getArgumentCompletions: (prefix) => ["new", "attach", "edit", "discuss", "review", "ready", "status", "stop", "resume", "solo", "model", "help", "exit", "clear", "quit"].filter((verb) => verb.startsWith(prefix)).map((verb) => ({ value: verb, label: verb })),
|
||||
handler: async (args, ctx) => {
|
||||
try {
|
||||
if (state.child) { ctx.ui.notify("This is the delegated worker. Goal approval belongs to its parent.", "info"); return; }
|
||||
if (state.child) {
|
||||
if (["stop", "resume"].includes(args.trim())) { cancelControl(); state.mode = args.trim() === "stop" ? "paused" : "solo"; generation++; save(); ctx.ui.notify(nativeMessages.workerPause(state.mode === "paused"), "info"); return; }
|
||||
ctx.ui.notify("This is the delegated worker. Goal approval belongs to its parent.", "info"); return;
|
||||
}
|
||||
let command = args.trim();
|
||||
if (!command) {
|
||||
const actions = [
|
||||
@@ -472,6 +607,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
return;
|
||||
}
|
||||
if (command === "exit") {
|
||||
cancelControl();
|
||||
const storage = new CronStorage(ctx.cwd);
|
||||
const session = ctx.sessionManager.getSessionId();
|
||||
const matching = storage.getAllJobs().filter(j => j.name === `goals-${session}`);
|
||||
@@ -489,6 +625,7 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
if (command === "stop") {
|
||||
if (state.mode === "planning") { ctx.ui.notify("A draft cannot pause; use /goals quit to clear goal state and preserve the draft.", "warning"); return; }
|
||||
if (state.mode !== "solo" && state.mode !== "supervising") return;
|
||||
cancelControl();
|
||||
state.pausedFrom = state.mode;
|
||||
state.mode = "paused"; generation++; notice = true; fullPlanContextDue = true; save(); refresh(ctx); watchPlan(ctx);
|
||||
const pause = pauseExitNotice(state.worker, false);
|
||||
@@ -533,23 +670,39 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
});
|
||||
pi.registerTool({
|
||||
name: "OpenGoalWorker", label: "Open native goal worker", description: nativeMessages.openDescription,
|
||||
parameters: Type.Object({ task: Type.String({ minLength: 1 }) }),
|
||||
parameters: Type.Object({ task: Type.Optional(Type.String()), action: Type.Optional(Type.Union([Type.Literal("start"), Type.Literal("fresh"), Type.Literal("recover")])), model: Type.Optional(Type.String()), reviewedThrough: Type.Optional(Type.String()), writersStopped: Type.Optional(Type.Boolean()), sessionFile: Type.Optional(Type.String()) }),
|
||||
async execute(_id, params, signal, _update, ctx) {
|
||||
if (state.child || state.mode !== "supervising" || !state.plan) return result(goalToolBlocked(state.mode));
|
||||
if (opening || state.worker) return result(nativeMessages.alreadyRecorded);
|
||||
if (!channel?.snapshot().connected) return result(nativeMessages.intercomNotReady);
|
||||
const stamp = generation;
|
||||
const plan = state.plan;
|
||||
const action = params.action ?? "start";
|
||||
if (opening || state.worker?.pending || action === "start" && state.worker) return result(nativeMessages.alreadyRecorded);
|
||||
const preference = action === "recover" ? null : notedPlanValue("preferred worker model");
|
||||
if (params.model || preference && (preference.includes("/") || !/^(?:none|\(none|default|inherit|not stated)\b/i.test(preference))) return result(nativeMessages.modelRaceBoundary);
|
||||
if (action !== "recover" && !params.task) return result(nativeMessages.taskRequired);
|
||||
if (action === "fresh" && (!state.worker?.identity || !params.reviewedThrough)) return result(nativeMessages.reviewRequired);
|
||||
if (!channel?.snapshot().connected || !channel.snapshot().supported) return result(nativeMessages.intercomNotReady);
|
||||
const stamp = generation, plan = state.plan;
|
||||
const peers = await channel.listSessions().catch(() => undefined);
|
||||
if (!peers) return result(nativeMessages.intercomNotReady);
|
||||
const self = peers?.filter(peer => peer.pid === process.pid);
|
||||
if (self?.length !== 1) return result(nativeMessages.noIdentity);
|
||||
if (stamp !== generation || opening || state.worker || signal?.aborted) return result(messages.cancelled);
|
||||
const self = peers.filter(peer => peer.pid === process.pid);
|
||||
if (self.length !== 1) return result(nativeMessages.noIdentity);
|
||||
if (stamp !== generation || opening || signal?.aborted) return result(messages.cancelled);
|
||||
const requestId = randomUUID();
|
||||
state.worker = { requestId, parentId: self[0].id }; state.workerStopped = false; workerRevision++; opening = true; save();
|
||||
const request: WorkerRequest = { id: requestId, action, task: params.task, model: params.model, reviewedThrough: params.reviewedThrough, writersStopped: params.writersStopped, phase: "probe", previous: state.worker?.identity };
|
||||
if (action === "recover") {
|
||||
if (!params.writersStopped || params.model || params.task) return result(nativeMessages.stopRequired);
|
||||
try {
|
||||
request.sessionFile = params.sessionFile || state.worker?.sessionFile;
|
||||
if (!request.sessionFile || !isAbsolute(request.sessionFile)) return result(nativeMessages.notDurable);
|
||||
const saved = savedWorker(request.sessionFile);
|
||||
request.savedId = saved.header.id; request.savedDigest = saved.digest;
|
||||
request.savedIntercom = saved.state.parent?.selfId ?? (request.sessionFile === state.worker?.sessionFile ? state.worker.intercomId : undefined);
|
||||
if (saved.header.cwd !== ctx.cwd || saved.state.plan !== plan || saved.state.parent?.intercomId !== self[0].id || !request.savedIntercom) return result(nativeMessages.notOwned);
|
||||
} catch (error) { return result(String(error)); }
|
||||
}
|
||||
state.worker = { ...state.worker, requestId: state.worker?.requestId ?? requestId, parentId: self[0].id, pending: request }; state.workerStopped = false; workerRevision++; opening = true; save();
|
||||
try {
|
||||
const pane = await openProjectPane({ cwd: ctx.cwd, focus: false, signal, message: workerAssignment(plan, self[0].id, requestId, params.task) });
|
||||
if (pane.ok && state.plan === plan && state.worker?.requestId === requestId) { state.worker.paneId = pane.data.binding.paneId; save(); }
|
||||
const pane = await openProjectPane({ cwd: ctx.cwd, focus: false, signal }); // No prompt before verified capability/model selection.
|
||||
if (pane.ok && state.plan === plan && state.worker?.pending?.id === requestId) { state.worker.paneId = pane.data.binding.paneId; save(); probe(); }
|
||||
return result(pane.ok ? JSON.stringify({ disposition: pane.data.disposition, paneId: pane.data.binding.paneId, projectRoot: pane.data.binding.projectRoot, bindingPath: pane.data.bindingPath }) + nativeMessages.openReceipt : JSON.stringify(pane));
|
||||
} catch (error) { return result(nativeMessages.openFailed + String(error)); }
|
||||
finally { opening = false; }
|
||||
@@ -573,10 +726,10 @@ export default function mainSupervisor(pi: ExtensionAPI) {
|
||||
if (!peers) return result(nativeMessages.intercomNotReady);
|
||||
if (stamp !== generation) return result(messages.cancelled);
|
||||
if (!peers?.some(peer => peer.id === params.parent && peer.pid !== process.pid)) return result(nativeMessages.parentUnavailable);
|
||||
state = { ...initial(), child: true, mode: "solo", parent: { intercomId: params.parent!, requestId: params.requestId! } };
|
||||
state = { ...initial(), child: true, mode: "solo", parent: { intercomId: params.parent!, requestId: params.requestId!, selfId: peers.find(peer => peer.pid === process.pid)?.id, started: true } };
|
||||
}
|
||||
state.plan = params.path; generation++; notice = true; fullPlanContextDue = true; save(); refresh(ctx);
|
||||
if (state.parent) channel?.publish({ type: "attached", to: state.parent.intercomId, requestId: state.parent.requestId, plan: state.plan, sessionFile: ctx.sessionManager.getSessionFile() }, { audience: "capable" });
|
||||
if (state.parent) channel?.publish({ type: "attached", to: state.parent.intercomId, requestId: state.parent.requestId, plan: state.plan, sessionFile: ctx.sessionManager.getSessionFile(), identity: identity(ctx) }, { audience: "capable" });
|
||||
return result(childPlanAttached(params.path));
|
||||
},
|
||||
});
|
||||
|
||||
+19
-7
@@ -156,13 +156,13 @@ export const attachGoalPlanDescription = "Attach the absolute plan path explicit
|
||||
export const childPlanRole = "You are the delegated implementation worker. Save evidence and report progress for your delegated work; leave plan maintenance to the parent. Preserve agreed goals, requirements and discriminators; the supervisor owns goal-status changes and completion approval. Do not launch a second writer. Call AttachGoalPlan with the explicit plan path in your task before implementation (also after reconnect if unbound). Immediately report your actual Intercom UUID, saved-session path and current provider/model to the supplied supervisor ID. Identify unavailable fields as unknown; do not equate runtime IDs, session filenames and Intercom IDs. Send progress, completion and blocker reports there with artifact paths, then stay open for live messages. Do not exit or use caller_ping; unsent editor drafts are not visible in model context.";
|
||||
export function readyApproved(workerName: string, planPath: string, notedWorker: string | undefined, plan: string, supervisorId: string): string {
|
||||
const launch = notedWorker
|
||||
? `Inspect the recorded worker session ${notedWorker}; if live, steer that exact Intercom session. Do not open or replace its conversation. Saved-session recovery remains manual and requires confirmation no writer is active.`
|
||||
? `Inspect the recorded worker session ${notedWorker}; if live, steer that exact Intercom session. Do not open or replace its conversation. For stopped-writer recovery use OpenGoalWorker action recover with writersStopped=true after inspecting the actual old runtime and saved history. It restores context without replaying a task.`
|
||||
: `Use OpenGoalWorker with a bounded first task for '${workerName}'. It uses Nico project.open, not subagent.`;
|
||||
return `[pi-goals: approval — Ready]\nReady approved this plan: ${planPath}. Stay here as supervisor. ${launch} Confirm your actual Intercom UUID with status/list; your Pi session ID ${supervisorId} is a distinct field. Await explicit worker attachment and a report with actual Intercom UUID, saved-session path and resolved model. Inspect results and steer corrections in that same open session. A receipt or idle pane is not completion. Never reset or close the review conversation to start independent work.\n\n${quotedPlan(planPath, foldPlan(plan), "working set before Log")}`;
|
||||
return `[pi-goals: approval — Ready]\nReady approved this plan: ${planPath}. Stay here as supervisor. ${launch} Confirm your actual Intercom UUID with status/list; your Pi session ID ${supervisorId} is a distinct field. Await explicit worker attachment and a report with actual Intercom UUID, saved-session path and resolved model. Inspect results and steer corrections in that same open session. A receipt or idle pane is not completion. Only after reviewing the exact saved session may OpenGoalWorker action fresh start independent context in that pane, retaining its saved history; supply reviewedThrough from the inspected latest entry.\n\n${quotedPlan(planPath, foldPlan(plan), "working set before Log")}`;
|
||||
}
|
||||
|
||||
export function workerAssignment(plan: string, parent: string, requestId: string, task: string): string {
|
||||
return `You are the delegated goals-worker in a native Nico project pane. First call intercom status/list to establish the live connection and verify the parent. Before any implementation call AttachGoalPlan with ${JSON.stringify({ path: plan, parent, requestId })}. This grants worker context only, never supervisor approval authority. Read the complete supplied plan, applicable AGENTS.md and skills. Confirm the exact parent Intercom UUID ${parent} in the live roster; send it your initial actual Intercom UUID, saved-session path, resolved provider/model and thinking level. Do not infer one identity from another. Use normal tools; no model switch was requested by this launch. Implement only this assignment:\n\n${task}\n\nSave actual artifacts and verification output. Report blocked, error and result evidence through Intercom to that exact parent. The parent independently inspects and may send a concrete correction here. Do not approve goals or launch another writer. Respect human pauses and intervention. Keep this conversation open with the final review visible; do not exit, reset, switch session or close the pane.`;
|
||||
return `You are already attached as the delegated goals-worker in this native Nico project pane for plan ${plan}; assignment ${requestId}. This grants worker context only, never supervisor approval authority. Read the complete supplied plan, applicable AGENTS.md and skills. Confirm the exact parent Intercom UUID ${parent} in the live roster; send it your initial actual Intercom UUID, saved-session path, resolved provider/model and thinking level. Do not infer one identity from another. Use normal tools; no model switch was requested by this launch. Implement only this assignment:\n\n${task}\n\nSave actual artifacts and verification output. Report blocked, error and result evidence through Intercom to that exact parent. The parent independently inspects and may send a concrete correction here. Do not approve goals or launch another writer. Respect human pauses and intervention. Keep this conversation open with the final review visible; do not exit, reset, switch session or close the pane.`;
|
||||
}
|
||||
// Supervision and turn-event upkeep (not a scheduled wake-up).
|
||||
const supervisorJob = "Your job is to be an autonomous research partner and supervisor with responsibility for the user's goals. Keep perspective, bring diligence, and use research taste and wisdom to sustain work overnight and keep it on track. Resolve routine implementation decisions yourself; ask the user only when their judgment or authorization is needed. Let each check-in follow what changed or needs attention, rather than repeat the previous recap.";
|
||||
@@ -173,7 +173,7 @@ You can speculate and brainstorm around uncertainty or unexpected results. Label
|
||||
(b •_•)b -- wassname
|
||||
Take uncertainty as an invitation to investigate, not something to hide. Have room to play with ideas, question yourself and the worker, and appreciate a good surprise. Investigate surprising results, find mistaken assumptions, make complicated ideas simpler, and disagree usefully rather than agree politely. Keep the work moving without turning supervision into paperwork. A little affectionate teasing is welcome when it fits, and workers can push back too. Keep the humor friendly and the criticism specific. -- Pi/Astra
|
||||
Use OpenGoalWorker for the first native project pane and stock Intercom for exact-session assignment/report/steering. Do not use subagent as a second backend. A stored binding is not proof of liveness; missing runtime state is not proof of stop. Verify actual Intercom identities with list/status; your Pi session ID is ${supervisorId}, a distinct field. Require artifact paths, saved verification and blocker/error reports. When the worker stops for any reason, inspect actual artifacts and saved messages before approving or correcting it in the same open session. A recap or receipt alone sends no instruction and proves no action. Record actual pane identity, '- worker session:' and '- worker intercom session:' with provenance. CompleteGoal belongs only to this parent or explicitly confirmed solo self-verification.
|
||||
Keep normal tools and honor human model changes. project.open has no model override: inspect the native worker's resolved model; if a requested model is unavailable, report it rather than silently substituting. After compaction reread the plan. Lost connection or exhausted credits does not erase work. Preserve drafts and saved sessions; confirm other writers stopped before solo takeover. Automatic saved-session recovery and later fresh-session replacement are not implemented in this first-session port. Do not reset, close, or replace an earlier review conversation or start a duplicate writer.`;
|
||||
Keep normal tools and honor human model changes. project.open has no model override: inspect the native worker's resolved model; if a requested model is unavailable, report it rather than silently substituting. After compaction reread the plan. Lost connection or exhausted credits does not erase work. Preserve drafts and saved sessions; confirm other writers stopped before solo takeover. OpenGoalWorker action recover restores an owned durable session only after confirmed stopped writers, without replaying work or changing its model. Action fresh requires the exact last reviewed entry and an idle draft-free peer, and retains the old history. Revisions use ordinary Intercom in the same context. Requested-model automation currently fails closed at the public setter race; no preference means keep the native default. Never replace an unreviewed conversation or start a duplicate writer.`;
|
||||
}
|
||||
// Routine notices quote only selected goal lines; full context stops at Log.
|
||||
const goalLines = (text: string) => foldPlan(text).split("\n").filter(line => GOAL_LINE.test(line)).join("\n");
|
||||
@@ -238,7 +238,7 @@ export function pauseExitNotice(worker: { intercomId?: string; sessionFile?: str
|
||||
return `Goals ${exited ? "exited to ordinary chat" : "paused locally"}; plan and evidence retained. ${worker ? `Locate the recorded native pane ${worker.paneId ?? "unknown"}, Intercom session ${worker.intercomId ?? "unknown"}, saved session ${worker.sessionFile ?? "unknown"}. Send an explicit pause there; inspect and confirm actual stop without closing the review conversation.` : "No worker recorded: inspect Intercom and native panes; absence is not proof of stop."} Remote stop is NOT yet confirmed. Resume only after explicit authorization.`;
|
||||
}
|
||||
export function resumeNotice(workerName: string, planPath: string, worker: { sessionFile?: string; intercomId?: string } | undefined): string {
|
||||
return `User authorized continuation of ${planPath}. Inspect worker state before any launch. ${worker ? `Use the existing session ${worker.sessionFile ?? "unknown"} and exact Intercom UUID ${worker.intercomId ?? "unknown"}; if live, inspect/message it. Do not open a replacement. Saved-session recovery requires separate confirmation that all previous writers stopped.` : `Use OpenGoalWorker for '${workerName}' only after confirming no prior writer exists.`} Continue only unfinished goals; retain saved progress and scheduler edits.`;
|
||||
return `User authorized continuation of ${planPath}. Inspect worker state before any launch. ${worker ? `Use the existing session ${worker.sessionFile ?? "unknown"} and exact Intercom UUID ${worker.intercomId ?? "unknown"}; if live, inspect/message it. Do not open a replacement. For saved-session recovery use OpenGoalWorker action recover only after confirming all other writers stopped; do not pass a task or model to replay.` : `Use OpenGoalWorker for '${workerName}' only after confirming no prior writer exists.`} Continue only unfinished goals; retain saved progress and scheduler edits.`;
|
||||
}
|
||||
export const soloRole = "Solo mode: implement the approved plan directly; do not delegate a concurrent writer. Verify artifacts before CompleteGoal; completion is self-verification, not independent supervisor review. Continue only unfinished goals and keep plan/evidence current.";
|
||||
export function soloNotice(planPath: string): string {
|
||||
@@ -249,13 +249,25 @@ export function attachNotice(planPath: string, solo: boolean, notedWorker: strin
|
||||
}
|
||||
|
||||
export const nativeMessages = {
|
||||
openDescription: "After Ready, open the first native Nico project pane for this plan. Supply a bounded task. An open receipt is not attachment or completion. Existing bindings are preserved; never replace a prior conversation.",
|
||||
actionApplied: (action: string, peer: unknown) => `Native ${action} confirmed in the exact peer: ${JSON.stringify(peer)}. Recovery submits no task and does not change the model. Inspect the observed identity and durable history before further steering.`,
|
||||
workerPause: (paused: boolean) => `Worker ${paused ? "paused" : "unpaused"} locally; no new task submitted and no approval authority granted.`,
|
||||
notDurable: "No durable owned worker transcript is available; a prospective saved path is not recovery evidence.",
|
||||
notOwned: "The exact session/plan/parent ownership does not match; no worker action taken.",
|
||||
controlDescription: "Verified native worker control only; arbitrary command text never starts work.",
|
||||
controlRejected: "Native worker control rejected; inspect actual peer state before retrying.",
|
||||
controlChanged: "Worker identity, history, input, pause or writer state changed; no replacement/task authorized.",
|
||||
controlCancelled: "Native session replacement was cancelled; prior history remains active.",
|
||||
reviewRequired: "Fresh work requires the exact last reviewed session entry ID; preserve the current conversation until reviewed.",
|
||||
stopRequired: "Recovery requires confirmed stopped writers and a durable owned saved session. No task replay or model override is accepted on recovery.",
|
||||
taskRequired: "Supply an explicit bounded task for a new or fresh worker context.",
|
||||
modelRaceBoundary: "Requested-model startup is held: public setModel cannot atomically preserve a concurrent human selection. No task was submitted under a fallback.",
|
||||
openDescription: "After Ready, use action start for a blank native peer and bounded task. Use fresh only after inspecting the recorded session and supplying its exact reviewedThrough leaf ID; old history is retained. Use recover with writersStopped=true and the owned saved session, never a task replay. Requested models currently fail closed at an unresolved public setter race. A pane receipt is not actual attachment or completion.",
|
||||
disconnected: "Intercom disconnected; liveness and stop are unconfirmed. Inspect the saved session and pane; do not launch a replacement.",
|
||||
shuttingDown: "Worker session shutting down; inspect its last saved messages. No goal sign-off inferred.",
|
||||
noAssistant: "Worker run ended without an assistant result; inspect saved messages.",
|
||||
alreadyRecorded: "A worker is already recorded or opening. Inspect its native pane and exact Intercom session; do not create a duplicate or replace its conversation.",
|
||||
noIdentity: "Intercom identity unavailable; no worker opened.",
|
||||
openReceipt: "\nAwait actual attachment and Intercom report. If already-open, no assignment was delivered; inspect that conversation, do not reset it.",
|
||||
openReceipt: "\nAwait verified peer capability and actual action confirmation. An existing binding is only a surface; draft, history and ownership checks still apply. Do not resend or infer work from this receipt.",
|
||||
openFailed: "Native open failed; inspect binding and possible live writer before retry or solo takeover: ",
|
||||
parentUnavailable: "Parent Intercom identity is not live; no worker attachment changed.",
|
||||
intercomNotReady: "Intercom is still connecting. Call intercom status/list, verify the live parent identity, then retry this operation in the same session. No attachment or launch changed.",
|
||||
|
||||
+69
-3
@@ -25,8 +25,8 @@ function fixture(child = false) {
|
||||
const cwd = mkdtempSync(join(tmpdir(), "goals-main-test-")); roots.push(cwd);
|
||||
const entries: any[] = child ? [{ type: "custom", customType: "pi-goals-main-supervisor-v1", data: { mode: "solo", child: true, signoffs: {} } }] : []; const hooks = new Map<string, any>(); const commands = new Map<string, any>(); const tools = new Map<string, any>();
|
||||
const messages: any[] = [];
|
||||
const ctx = { cwd, sessionManager: { getBranch: () => entries, getSessionId: () => "copy-only", getSessionFile: () => join(cwd, "session.jsonl") }, hasUI: true, hasPendingMessages: vi.fn(() => false), ui: {
|
||||
theme: { fg: (_color: string, text: string) => text }, notify: vi.fn(), setStatus: vi.fn(), setWidget: vi.fn(), select: vi.fn(async (_title: string, _options: string[]) => "Ready"), editor: vi.fn(),
|
||||
const ctx = { cwd, sessionManager: { getBranch: () => entries, getSessionId: () => "copy-only", getSessionFile: () => join(cwd, "session.jsonl"), getLeafId: () => "reviewed-leaf", getHeader: () => ({ id: "copy-only", cwd }) }, hasUI: true, waitForIdle: vi.fn(async () => {}), newSession: vi.fn(async (_options: any) => ({ cancelled: false })), switchSession: vi.fn(async (_path: string) => ({ cancelled: false })), hasPendingMessages: vi.fn(() => false), ui: {
|
||||
getEditorText: vi.fn(() => ""), theme: { fg: (_color: string, text: string) => text }, notify: vi.fn(), setStatus: vi.fn(), setWidget: vi.fn(), select: vi.fn(async (_title: string, _options: string[]) => "Ready"), editor: vi.fn(),
|
||||
} };
|
||||
let registration: any;
|
||||
const channel = { snapshot: vi.fn(() => ({ connected: true, supported: true })), listSessions: vi.fn(async () => [{ id: "parent-intercom", pid: process.pid }, { id: "live-parent", pid: process.pid + 1 }]), publish: vi.fn() };
|
||||
@@ -41,6 +41,7 @@ function fixture(child = false) {
|
||||
sendUserMessage: (content: string, options: any) => messages.push({ message: { content }, options, savedPrompt: true }),
|
||||
events: { emit: vi.fn((name, data) => { if (name === "intercom:extension-register") { registration = data; data.onReady(channel); } }) },
|
||||
getAllTools: vi.fn((): any[] => []),
|
||||
getCommands: () => [...commands.keys()].map(name => ({name})),
|
||||
};
|
||||
goalsExtension(pi as unknown as ExtensionAPI);
|
||||
hooks.get("session_start")({}, ctx);
|
||||
@@ -1040,7 +1041,8 @@ it("opens no-focus, records explicit attachment only, and wakes review only for
|
||||
const waiting = await f.tools.get("OpenGoalWorker").execute("open", { task: "first" }, undefined, undefined, f.ctx);
|
||||
expect(waiting.content[0].text).toContain("still connecting"); expect(openProjectPane).not.toHaveBeenCalled();
|
||||
await f.launch({ id: "worker-id", sessionFile: "/tmp/native-worker.jsonl" });
|
||||
expect(openProjectPane).toHaveBeenCalledWith(expect.objectContaining({ cwd: f.ctx.cwd, focus: false, message: expect.stringContaining("AttachGoalPlan") }));
|
||||
expect(openProjectPane).toHaveBeenCalledWith(expect.objectContaining({ cwd: f.ctx.cwd, focus: false }));
|
||||
expect(vi.mocked(openProjectPane).mock.calls[0][0]).not.toHaveProperty("message");
|
||||
const worker = f.entries.at(-1).data.worker;
|
||||
expect(worker).toMatchObject({ paneId: "native-pane", intercomId: "worker-id", sessionFile: "/tmp/native-worker.jsonl" });
|
||||
const notice = { type: "stopped", to: worker.parentId, requestId: worker.requestId, plan: f.path, text: "Blocked: input missing" };
|
||||
@@ -1049,6 +1051,15 @@ it("opens no-focus, records explicit attachment only, and wakes review only for
|
||||
f.event({ type: "message", fromSessionId: "worker-id", payload: { ...notice, plan: "/foreign.md" } });
|
||||
f.event({ type: "message", fromSessionId: "worker-id", payload: { ...notice, requestId: "stale" } });
|
||||
expect(f.messages).toHaveLength(count);
|
||||
f.event({ type: "message", fromSessionId: "worker-id", payload: { ...notice, identity: { sessionId: "worker-session", sessionFile: "/tmp/native-worker.jsonl", paneId: "native-pane", leafId: "reviewed-leaf", requestId: worker.requestId, durable: true } } });
|
||||
await f.tools.get("OpenGoalWorker").execute("fresh", { action: "fresh", task: "later work", reviewedThrough: "reviewed-leaf" }, undefined, undefined, f.ctx);
|
||||
f.event({ type: "message", fromSessionId: "worker-id", payload: notice });
|
||||
expect(f.entries.at(-1).data.worker.pending).toBeDefined();
|
||||
const pendingId = f.entries.at(-1).data.worker.pending.id;
|
||||
f.event({ type: "message", fromSessionId: "worker-id", payload: { ...notice, type: "attached", sessionFile: "/tmp/native-worker.jsonl" } });
|
||||
expect(f.entries.at(-1).data.worker.pending.id).toBe(pendingId);
|
||||
expect(f.messages.at(-1).options).toEqual({ deliverAs: "nextTurn" });
|
||||
expect(f.messages.at(-1).message.content).not.toContain("Native fresh confirmed");
|
||||
f.event({ type: "message", fromSessionId: "worker-id", payload: notice });
|
||||
expect(f.messages.at(-1)).toMatchObject({ savedPrompt: true, message: { content: expect.stringContaining("Blocked: input missing") } });
|
||||
expect(f.entries.at(-1).data.signoffs).toEqual({});
|
||||
@@ -1082,6 +1093,9 @@ it("ordinary project peer explicitly attaches as worker, never gaining approval
|
||||
|
||||
it("pending or failed native opening never permits an unconfirmed second writer", async () => {
|
||||
const f = fixture(); await f.draft(); await f.command("ready");
|
||||
const tool = f.tools.get("OpenGoalWorker");
|
||||
expect(tool.parameters.properties.task.minLength ?? 0).toBe(0);
|
||||
expect((await tool.execute("empty", { task: "" }, undefined, undefined, f.ctx)).content[0].text).toContain("task");
|
||||
let release!: () => void;
|
||||
vi.mocked(openProjectPane).mockImplementationOnce(() => new Promise((_resolve, reject) => { release = () => reject(new Error("connection lost after open")); }));
|
||||
const opening = f.tools.get("OpenGoalWorker").execute("open", { task: "first" }, undefined, undefined, f.ctx);
|
||||
@@ -1092,4 +1106,56 @@ it("pending or failed native opening never permits an unconfirmed second writer"
|
||||
const again = await f.tools.get("OpenGoalWorker").execute("open", { task: "again" }, undefined, undefined, f.ctx);
|
||||
expect(again.content[0].text).toContain("already recorded");
|
||||
expect(openProjectPane).toHaveBeenCalledTimes(1);
|
||||
const worker = f.entries.at(-1).data.worker, saved = join(f.ctx.cwd, "owned-worker.jsonl");
|
||||
writeFileSync(saved, [{ type: "session", id: "old-worker", cwd: f.ctx.cwd }, { type: "custom", customType: "pi-goals-main-supervisor-v1", data: { child: true, mode: "solo", plan: f.path, parent: { intercomId: worker.parentId, selfId: "old-worker" } } }, { type: "message", message: { role: "assistant", content: [{ type: "text", text: "prior work" }] } }].map(entry => JSON.stringify(entry)).join("\n"));
|
||||
f.event({ type: "message", fromSessionId: "old-worker", payload: { type: "attached", to: worker.parentId, requestId: worker.requestId, plan: f.path, sessionFile: saved } });
|
||||
expect((await tool.execute("replay", { action: "recover", task: "repeat work", writersStopped: true }, undefined, undefined, f.ctx)).content[0].text).toContain("No task replay");
|
||||
await tool.execute("recover", { action: "recover", task: "", model: "", sessionFile: "", reviewedThrough: "", writersStopped: true }, undefined, undefined, f.ctx);
|
||||
expect(f.entries.at(-1).data.worker.pending).toMatchObject({ action: "recover", sessionFile: saved, task: "" });
|
||||
expect(openProjectPane).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("verified native control preserves drafts/history and fresh context uses only the public replacement API", async () => {
|
||||
const f = fixture(); const path = join(f.ctx.cwd, "owned.md"); writeFileSync(path, f.plan);
|
||||
await f.tools.get("AttachGoalPlan").execute("attach", { path, parent: "live-parent", requestId: "previous" }, undefined, undefined, f.ctx);
|
||||
const file = f.ctx.sessionManager.getSessionFile();
|
||||
writeFileSync(file, [{ type: "session", id: "copy-only", cwd: f.ctx.cwd }, ...f.entries, { type: "message", id: "reviewed-leaf", message: { role: "assistant", content: [{ type: "text", text: "reviewed output" }] } }].map(entry => JSON.stringify(entry)).join("\n"));
|
||||
const before = readFileSync(file, "utf8");
|
||||
f.event({ type: "message", fromSessionId: "live-parent", payload: { type: "inspect", paneId: process.env.HERDR_PANE_ID, requestId: "next" } });
|
||||
const expected = f.channel.publish.mock.calls.at(-1)![0].identity;
|
||||
const request = { id: "next", action: "fresh", phase: "control", reviewedThrough: "reviewed-leaf", previous: expected, task: "independent work" };
|
||||
const dispatch = async (from = "live-parent") => {
|
||||
f.event({ type: "message", fromSessionId: from, payload: { type: "control", to: "parent-intercom", plan: path, expected, request } });
|
||||
await f.commands.get("goals-worker-control").handler("", f.ctx);
|
||||
};
|
||||
f.ctx.ui.getEditorText.mockReturnValue("unsent draft"); await dispatch();
|
||||
expect(f.ctx.newSession).not.toHaveBeenCalled(); expect(f.ctx.ui.getEditorText()).toBe("unsent draft"); expect(readFileSync(file, "utf8")).toBe(before);
|
||||
f.ctx.ui.getEditorText.mockReturnValue(""); f.ctx.hasPendingMessages.mockReturnValueOnce(true); await dispatch();
|
||||
expect(f.ctx.newSession).not.toHaveBeenCalled();
|
||||
await dispatch("foreign"); expect(f.ctx.newSession).not.toHaveBeenCalled();
|
||||
request.id = "reviewed-next"; await dispatch();
|
||||
expect(f.ctx.newSession).toHaveBeenCalledWith(expect.objectContaining({ parentSession: file, setup: expect.any(Function) }));
|
||||
expect(f.ctx.switchSession).not.toHaveBeenCalled(); expect(readFileSync(file, "utf8")).toBe(before);
|
||||
const appendCustomEntry = vi.fn(); await f.ctx.newSession.mock.calls[0][0].setup({ appendCustomEntry });
|
||||
expect(appendCustomEntry).toHaveBeenCalledWith("pi-goals-main-supervisor-v1", expect.objectContaining({ child: true, mode: "solo", plan: path, parent: { intercomId: "live-parent", requestId: "reviewed-next" } }));
|
||||
expect(f.messages.filter(message => message.savedPrompt).every(message => message.message.content === "/goals-worker-control")).toBe(true);
|
||||
request.id = "replacement-fails"; f.ctx.newSession.mockRejectedValueOnce(new Error("replacement rejected")); await dispatch();
|
||||
expect(f.ctx.newSession).toHaveBeenCalledTimes(2);
|
||||
expect(f.channel.publish).toHaveBeenLastCalledWith(expect.objectContaining({ type: "rejected", text: expect.stringContaining("replacement rejected") }), { audience: "capable" });
|
||||
f.hooks.get("session_shutdown")({ reason: "new" }, f.ctx);
|
||||
expect(f.channel.publish).toHaveBeenLastCalledWith(expect.objectContaining({ type: "stopped", text: expect.stringContaining("shutting down") }), { audience: "capable" });
|
||||
});
|
||||
|
||||
it("model requests fail closed and an unavailable control command never falls through to inference", async () => {
|
||||
const f = fixture(); await f.draft(); await f.command("ready");
|
||||
const before = f.entries.length;
|
||||
for (const model of ["offline/requested", "missing/unavailable"]) {
|
||||
const reply = await f.tools.get("OpenGoalWorker").execute("open", { task: "must not run", model }, undefined, undefined, f.ctx);
|
||||
expect(reply.content[0].text).toContain("public setModel");
|
||||
}
|
||||
expect(f.entries).toHaveLength(before); expect(openProjectPane).not.toHaveBeenCalled();
|
||||
f.commands.delete("goals-worker-control");
|
||||
const count = f.messages.length;
|
||||
f.event({ type: "message", fromSessionId: "live-parent", payload: { type: "control", to: "parent-intercom", plan: f.path, expected: { paneId: process.env.HERDR_PANE_ID }, request: { id: "unknown-command", action: "start" } } });
|
||||
expect(f.messages).toHaveLength(count);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user