diff --git a/README.md b/README.md index 30cb7ba..a18b92f 100644 --- a/README.md +++ b/README.md @@ -50,13 +50,15 @@ If a required model or supervisor is unavailable, the widget says **goals paused - `/goals restart` explicitly closes only the tracked supervisor pane and starts a replacement for a working plan, preserving its file/version but invalidating old approvals. During planning it clears the failed pane so Ready can launch again. - In the supervisor pane, use `/model` then `/goals reconnect` to recover an unavailable supervisor model. +Both sessions must load the updated transport for the request/reply reconnect fix; mixed-version peers are not a supported recovery configuration. Ready announces worker readiness only after its model is restored, and clearing a plan cancels its pending readiness wait. + A new supervisor may still need up to five minutes for initial compaction. Recovery does not terminate background jobs. Planning/diagnostic command checks are guardrails, not an OS sandbox; loaded extensions and repository Git configuration must be trusted. Model choices are remembered per project and role in `.pi/pi-goals/models/`. Use `/model` in planning, worker, or supervisor sessions to change that role's choice. Ready restores the worker choice after the planning fork is ready. An unavailable saved model stops the transition instead of substituting another. `/goals model ` explicitly overrides the supervisor choice for launch. -- Pi/OpenAI ## Plan format -A goal is a checkbox line whose text starts with `goal:`: +Current goals belong above `## Log`; goal-shaped historical checklists below it are ignored by the widget, approval matching and sign-off. A goal is a checkbox line whose text starts with `goal:`: ```md 1. [ ] goal: Produce the report @@ -68,6 +70,8 @@ A goal is a checkbox line whose text starts with `goal:`: The worker saves verification output in a nonempty repository file, adds that path to evidence, and commits it. The supervisor calls `ApproveGoal` with the inspected path; the worker then calls `CompleteGoal` with the exact goal text. +If context usage is unavailable, the supervisor warns once that its custom 100k compaction trigger cannot be checked. Pi's normal post-compaction `tokens: null` sample does not produce that warning; default auto-compaction is unchanged. + ## Development ```bash diff --git a/src/approval.ts b/src/approval.ts index 6d4ca0a..a15ae74 100644 --- a/src/approval.ts +++ b/src/approval.ts @@ -3,7 +3,7 @@ import { createHash } from "node:crypto"; import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, statSync, writeFileSync } from "node:fs"; import { dirname, join, relative, resolve } from "node:path"; -const GOAL_LINE = /^\s*(?:\d+\.|[-*])\s*\[([ xX/-])\]\s*goal:\s*(.*)$/i; +import { foldPlan, GOAL_LINE } from "./plan.js"; export interface ApprovalRecord { version: 3; @@ -42,7 +42,7 @@ export function repositoryState(cwd: string): { repoRoot: string; head: string; } export function goalBlock(plan: string, goal: string): string | null { - const lines = plan.split(/^##\s+Log\s*$/im, 1)[0].split("\n"); + const lines = foldPlan(plan).split("\n"); const wanted = goal.trim().toLowerCase(); const hits = lines.flatMap((line, index) => { const match = GOAL_LINE.exec(line); diff --git a/src/index.ts b/src/index.ts index 70f1d5b..2366e55 100644 --- a/src/index.ts +++ b/src/index.ts @@ -24,11 +24,14 @@ import { approvalMatches, approvalPath, goalBlock, hashGoalBlock, readApproval, import { backgroundState } from "./background.js"; import { closeSupervisorPane, openSupervisorPane } from "./herdr.js"; import { GoalIntercom } from "./intercom.js"; +import { FOLD_LINE, foldPlan, GOAL_LINE } from "./plan.js"; import { completeGoalDescription, completeGoalParamDescription, planDrafting, planningState, resync } from "./prompts.js"; import { RoleModels } from "./role-models.js"; import { isVisibleSupervisor, registerVisibleSupervisor } from "./supervisor-session.js"; import { workerView } from "./worker-view.js"; +export { foldPlan } from "./plan.js"; + const STATE = "pi-goals-state"; const STATUS_KEY = "pi-goals"; const WIDGET_KEY = "pi-goals-widget"; @@ -39,35 +42,24 @@ const PLAN_SHAPE = `${PLAN_DIR}/-vN.md`; // Plan mode blocks edit/write except for its plan file. bash remains available for read-only inspection. -- Pi/Codex const PLAN_MODE_BLOCKED_TOOLS = ["edit", "write"]; -// A checkbox line beginning "goal:", used by the widget and supervisor scheduling. -// Everything else reads the file as prose. -const GOAL_LINE = /^\s*(?:\d+\.|[-*])\s*\[([ xX/-])\]\s*goal:\s*(.*)$/i; // An indented checkbox line that isn't a goal: a subtask. Only the widget reads these, so the human // sees the next action and not just the goal -- this file IS the task list. const SUBTASK_LINE = /^\s+(?:\d+\.|[-*])\s*\[([ xX/-])\]\s*(.*)$/; -// The fold separates current goals from the longer research record. -const FOLD_LINE = /^##\s+Log\s*$/im; type GoalStatus = "open" | "active" | "done" | "cancelled"; const CHAR_TO_STATUS: Record = { " ": "open", "/": "active", x: "done", "-": "cancelled" }; function scanGoals(plan: string): Array<{ status: GoalStatus; subject: string; line: number }> { const goals: Array<{ status: GoalStatus; subject: string; line: number }> = []; - plan.split("\n").forEach((line, i) => { + foldPlan(plan).split("\n").forEach((line, i) => { const m = GOAL_LINE.exec(line); if (m) goals.push({ status: CHAR_TO_STATUS[m[1].toLowerCase()] ?? "open", subject: m[2].trim(), line: i }); }); return goals; } -/** Return the short current-goal section above "## Log". Exported for the unit test. */ -export function foldPlan(plan: string): string { - const m = FOLD_LINE.exec(plan); - return (m ? plan.slice(0, m.index) : plan).trimEnd(); -} - /** Open subtasks under the goal on line `goalLine`, up to the next goal line. */ export function openSubtasks(plan: string, goalLine: number): string[] { - const lines = plan.split("\n"); + const lines = foldPlan(plan).split("\n"); const out: string[] = []; for (let i = goalLine + 1; i < lines.length; i++) { if (GOAL_LINE.test(lines[i])) break; @@ -123,6 +115,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { latestDirection: "", }; let modelError: string | null = null; + let readyAttempt: object | undefined; intercom.onConnectionChange = (ctx) => updateWidget(ctx); let planningContextPending = false; let resyncReason: string | null = "New session."; @@ -147,12 +140,15 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { function pauseReason(): string | null { if (!state.phase) return null; if (modelError) return `${modelError} Select /model, then run /goals reconnect.`; - if (state.phase === "working" && !intercom.connected) return "Supervisor disconnected. Run /goals reconnect, or /goals restart to replace its tracked pane without discarding the plan."; + if (state.phase === "working" && !intercom.connected) return intercom.peerPresent + ? "Supervisor is present but not ready. Inspect its pane for startup/compaction or model errors; recover with /model then /goals reconnect in the supervisor pane if needed." + : "Supervisor disconnected. Run /goals reconnect, or /goals restart to replace its tracked pane without discarding the plan."; return null; } async function restoreModel(role: "planning" | "worker", ctx: ExtensionContext): Promise { modelError = `${role} model restoration is pending.`; + intercom.markNotReady(); try { await models.enter(role, ctx); modelError = null; @@ -168,7 +164,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { } const approvalId = randomUUID(); state = { ...state, approvalId }; - intercom.configure(approvalId, "worker", ctx); + intercom.configure(approvalId, "worker", ctx, false); persist(); } @@ -176,17 +172,19 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { return execFileSync("git", ["rev-parse", "--show-toplevel"], { cwd, encoding: "utf8" }).trim(); } - async function startSupervisor(ctx: ExtensionContext): Promise { + async function startSupervisor(ctx: ExtensionContext, isCurrent = () => !intercom.ended): Promise { if (intercom.ended) throw new Error("Session ended before supervisor startup."); repositoryRoot(ctx.cwd); const sourceSessionFile = ctx.sessionManager.getSessionFile(); if (!sourceSessionFile) throw new Error("The current session is not persisted, so it cannot be forked."); if (state.supervisorPaneId && state.approvalId) { - intercom.configure(state.approvalId, "worker", ctx); - await intercom.waitReady(5000); + intercom.configure(state.approvalId, "worker", ctx, false); + await intercom.waitReady(5000, { peerOnly: true }); return; } beginReview(ctx); + const binding = state.approvalId; + const current = () => isCurrent() && !intercom.ended && state.approvalId === binding; let paneId: string | null = null; try { paneId = await openSupervisorPane({ @@ -198,7 +196,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { extensionPath: fileURLToPath(import.meta.url), model: state.supervisorModel, }, (opened) => { - if (intercom.ended) throw new Error("Session ended during supervisor startup."); + if (!current()) throw new Error("Supervisor startup was cancelled."); paneId = opened; state = { ...state, supervisorPaneId: opened }; persist(); @@ -207,10 +205,10 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { if (paneId) throw new Error(`Supervisor startup failed in Herdr pane ${paneId}; it remains open for inspection. ${error instanceof Error ? error.message : String(error)}`); throw error; } - if (intercom.ended) throw new Error("Session ended during supervisor startup."); + if (!current()) throw new Error("Supervisor startup was cancelled."); state = { ...state, supervisorPaneId: paneId }; persist(); - await intercom.waitReady(); + await intercom.waitReady(undefined, { peerOnly: true }); } let workerTurns = 0; @@ -252,6 +250,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { } async function stopSupervisor(): Promise { + readyAttempt = undefined; if (!state.supervisorPaneId) { stopWorkerTimers(); intercom.detach(); return true; } try { await closeSupervisorPane(state.supervisorPaneId); @@ -309,6 +308,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { if (arg === "reconnect" || arg === "restart") { if (!state.phase) { ctx.ui.notify("No active plan to recover.", "info"); return; } if (!ctx.isIdle()) { ctx.ui.notify("Stop the current turn before recovering goal supervision.", "warning"); return; } + readyAttempt = undefined; try { await restoreModel(state.phase === "planning" ? "planning" : "worker", ctx); if (arg === "restart") { @@ -319,12 +319,13 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { if (state.phase === "working" || state.supervisorPaneId) { if (arg === "reconnect") { if (!state.approvalId) throw new Error("No saved supervision binding. Use /goals restart."); - intercom.configure(state.approvalId, "worker", ctx); - await intercom.waitReady(5000); + intercom.configure(state.approvalId, "worker", ctx, false); + await intercom.waitReady(5000, { peerOnly: true }); } else await startSupervisor(ctx); } if (intercom.ended) return; if (state.phase === "working") { + intercom.markReady(); startWorkerTimers(ctx); await publishWorkerView(ctx, "settled"); } @@ -480,9 +481,11 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { return; } if (state.phase !== "planning" || modelError || !ctx.hasUI) return; + const version = state.planVersion; + const planning = () => !intercom.ended && state.phase === "planning" && state.planVersion === version; let printed = ""; while (true) { - if (intercom.ended) return; + if (!planning()) return; const plan = readPlan(ctx); if (scanGoals(plan).length === 0) { if (plan.trim()) ctx.ui.notify(`The plan has no goal line. Revise ${planRel(ctx)} to add one.`, "warning"); @@ -493,8 +496,10 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { pi.sendMessage({ customType: "plan", content: plan, display: true }); } const choice = await ctx.ui.select(`Plan drafted in ${planRel(ctx)}.`, ["Ready", "Refine", "Edit", "Cancel"]); + if (!planning()) return; if (choice === "Refine") { const notes = await ctx.ui.editor("What should change about the plan?", ""); + if (!planning()) return; if (!notes?.trim()) continue; state = { ...state, latestDirection: notes }; persist(); @@ -505,6 +510,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { } if (choice === "Edit") { const edited = await ctx.ui.editor("Edit the plan", plan); + if (!planning()) return; if (edited !== undefined && edited !== plan) writePlan(ctx, edited); continue; } @@ -519,20 +525,27 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { return; } if (choice !== "Ready") return; + const attempt = {}; + readyAttempt = attempt; + const current = () => !intercom.ended && readyAttempt === attempt && state.planVersion === version; try { - await startSupervisor(ctx); - if (intercom.ended) return; + await startSupervisor(ctx, current); + if (!current()) return; await restoreModel("worker", ctx); + if (!current()) return; state = { ...state, phase: "working" }; resyncReason = "The plan was approved."; persist(); + intercom.markReady(); startWorkerTimers(ctx); await publishWorkerView(ctx, "ready"); + if (!current()) return; updateWidget(ctx); ctx.ui.notify(`Visible supervisor opened in Herdr pane ${state.supervisorPaneId}.`, "info"); pi.sendUserMessage("The plan is approved. Begin implementation as the worker."); } catch (error) { - if (intercom.ended) return; + if (!current()) return; + intercom.markNotReady(); ctx.ui.notify(`Goal supervisor could not start: ${error instanceof Error ? error.message : String(error)} Use /goals reconnect to retry, or /goals restart to replace the tracked pane.`, "warning"); state = { ...state, phase: "planning" }; persist(); @@ -658,7 +671,7 @@ function stamp(): string { export function tickGoal(plan: string, goal: string): string | null { const lines = plan.split("\n"); const want = goal.trim().toLowerCase(); - const hits = lines.flatMap((l, i) => (GOAL_LINE.exec(l)?.[2].trim().toLowerCase() === want ? [i] : [])); + const hits = scanGoals(plan).filter(g => g.subject.toLowerCase() === want).map(g => g.line); if (hits.length !== 1) return null; lines[hits[0]] = lines[hits[0]].replace(/\[[ xX/-]\]/, "[x]"); return lines.join("\n"); @@ -668,7 +681,7 @@ export function tickGoal(plan: string, goal: string): string | null { export function appendLog(text: string, entry: string): string { const lines = text.split("\n"); const line = `- ${entry}`; - const header = lines.findIndex((l) => /^##\s+Log\s*$/i.test(l)); + const header = lines.findIndex((l) => FOLD_LINE.test(l)); if (header === -1) return `${text.replace(/\n+$/, "")}\n\n## Log\n${line}\n`; let insertAt = header + 1; for (let i = header + 1; i < lines.length; i++) { diff --git a/src/intercom.ts b/src/intercom.ts index ac4f6f8..6bbce29 100644 --- a/src/intercom.ts +++ b/src/intercom.ts @@ -4,7 +4,7 @@ import type { IntercomExtensionChannel, IntercomExtensionEvent } from "pi-interc export type Role = "worker" | "supervisor"; export interface View { id: string; text: string; reason: string; through?: string; backgroundQuiet: boolean } -interface Message { binding: string; role: Role; kind: "hello" | "view" | "steer" | "received"; id: string; text?: string; reason?: string; ready?: boolean; through?: string; backgroundQuiet?: boolean } +interface Message { binding: string; role: Role; kind: "hello" | "view" | "steer" | "received"; id: string; text?: string; reason?: string; ready?: boolean; reply?: boolean; through?: string; backgroundQuiet?: boolean } const STATE = "pi-goals-intercom"; export class GoalIntercom { @@ -19,7 +19,7 @@ export class GoalIntercom { private peerReady = false; private pending = new Map(); private received = new Set(); - private waiters = new Set<() => void>(); + private waiters = new Set<(error?: Error) => void>(); latestView?: View; acknowledgedEntry?: string; onView: (view: View) => void = () => {}; @@ -41,6 +41,7 @@ export class GoalIntercom { } configure(binding: string, role: Role, ctx: ExtensionContext, ready = role === "worker"): void { + for (const wake of this.waiters) wake(new Error("Supervision readiness wait cancelled by reconfiguration.")); this.binding = binding; this.role = role; this.ctx = ctx; @@ -76,21 +77,33 @@ export class GoalIntercom { this.peerReady = false; this.latestView = undefined; this.pending.clear(); + for (const wake of this.waiters) wake(new Error("Supervision readiness wait cancelled: plan detached.")); if (this.ctx) this.onConnectionChange(this.ctx); } - markReady(): void { if (!this.stopped) { this.ready = true; this.hello(); } } + markReady(): void { this.setReady(true); } + markNotReady(): void { this.setReady(false); } + private setReady(ready: boolean): void { + if (this.stopped) return; + this.ready = ready; + this.hello(); + if (this.ctx) this.onConnectionChange(this.ctx); + } get ended(): boolean { return this.stopped; } get bound(): boolean { return !this.stopped && Boolean(this.binding); } - get connected(): boolean { return Boolean(!this.stopped && this.ready && this.binding && this.channel?.snapshot().connected && this.peerReady); } + get peerPresent(): boolean { return Boolean(this.bound && this.peer && this.channel?.snapshot().connected); } + get connected(): boolean { return this.ready && this.peerPresent && this.peerReady; } - async waitReady(timeoutMs = 300_000): Promise { - if (this.connected) return; + // Startup can wait for the supervisor while the worker is still in planning/model recovery. + async waitReady(timeoutMs = 300_000, { peerOnly = false } = {}): Promise { + const ready = () => this.connected || (peerOnly && this.peerPresent && this.peerReady); + if (ready()) return; await new Promise((resolve, reject) => { - const finish = () => { - if (!this.connected && !this.stopped) return; + const finish = (error?: Error) => { + if (!error && !ready() && !this.stopped) return; clearTimeout(timer); this.waiters.delete(finish); - if (this.stopped) reject(new Error("Session ended while waiting for Intercom readiness.")); + if (error) reject(error); + else if (this.stopped) reject(new Error("Session ended while waiting for Intercom readiness.")); else resolve(); }; const timer = setTimeout(() => { this.waiters.delete(finish); reject(new Error("Supervisor did not become ready through pi-intercom; inspect its pane.")); }, timeoutMs); @@ -125,8 +138,8 @@ export class GoalIntercom { if (!this.channel?.snapshot().supported) throw new Error("pi-intercom broker does not support extension channels."); this.channel.publish(message, { audience: "capable" }); } - private hello(): void { - if (!this.stopped && this.binding && this.channel?.snapshot().connected) this.publish({ binding: this.binding, role: this.role, kind: "hello", id: "hello", ready: this.ready }); + private hello(reply = false): void { + if (!this.stopped && this.binding && this.channel?.snapshot().connected) this.publish({ binding: this.binding, role: this.role, kind: "hello", id: "hello", ready: this.ready, reply }); } private receive(event: IntercomExtensionEvent): void { if (this.stopped) return; @@ -154,12 +167,12 @@ export class GoalIntercom { const changed = !this.peer || this.peerReady !== Boolean(message.ready); this.peer = event.fromSessionId; this.peerReady = Boolean(message.ready); - if (changed) { - this.hello(); - if (this.peerReady && this.ready) { - if (this.role === "worker" && this.latestView) this.publish({ binding: this.binding, role: this.role, kind: "view", ...this.latestView }); - for (const pending of this.pending.values()) this.publish(pending); - } + // Every request gets one reply, even if only the sender forgot its peer. + // Replies never elicit hellos; own-ready transitions also trigger replay here. + if (!message.reply) this.hello(true); + if (this.peerReady && this.ready) { + if (this.role === "worker" && this.latestView) this.publish({ binding: this.binding, role: this.role, kind: "view", ...this.latestView }); + for (const pending of this.pending.values()) this.publish(pending); } if (changed && this.ctx) this.onConnectionChange(this.ctx); for (const wake of this.waiters) wake(); diff --git a/src/plan.ts b/src/plan.ts new file mode 100644 index 0000000..35d873c --- /dev/null +++ b/src/plan.ts @@ -0,0 +1,8 @@ +// Shared plan syntax: only the section above the Log contains current goals. +export const GOAL_LINE = /^\s*(?:\d+\.|[-*])\s*\[([ xX/-])\]\s*goal:\s*(.*)$/i; +export const FOLD_LINE = /^##\s+Log\s*$/im; + +export function foldPlan(plan: string): string { + const match = FOLD_LINE.exec(plan); + return (match ? plan.slice(0, match.index) : plan).trimEnd(); +} diff --git a/src/supervisor-session.ts b/src/supervisor-session.ts index 2a6b7b3..5610f65 100644 --- a/src/supervisor-session.ts +++ b/src/supervisor-session.ts @@ -93,6 +93,7 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void { const settings = config(); let compacting = false; let bootstrapping = false; + let warnedUnknownUsage = false; let modelError: string | null = null; const intercom = new GoalIntercom(pi); const models = new RoleModels(pi); @@ -167,7 +168,14 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void { }); pi.on("before_agent_start", async (_event, ctx) => ({ systemPrompt: `${ctx.getSystemPrompt()}\n\n${supervisorPrompt(settings)}` })); pi.on("agent_settled", async (_event, ctx) => { - if (compacting || (ctx.getContextUsage()?.tokens ?? 0) < COMPACT_AT_TOKENS) return; + if (compacting) return; + const usage = ctx.getContextUsage(); + if (!usage && !warnedUnknownUsage) { + warnedUnknownUsage = true; + ctx.ui.notify("Supervisor context usage unavailable; the custom 100k compaction trigger cannot be checked. Pi's default auto-compaction is unchanged.", "warning"); + } + // Pi reports tokens:null after compaction until a fresh assistant usage sample. + if (typeof usage?.tokens !== "number" || usage.tokens < COMPACT_AT_TOKENS) return; compacting = true; ctx.compact({ customInstructions: `Keep the user's high-level intent, current plan state, unresolved risks, approval decisions, and the supervisor's own concise findings. Remove old worker views and implementation detail. The canonical plan remains ${settings.planPath}.`, diff --git a/test/fold.test.ts b/test/fold.test.ts index 3394297..f6b3eef 100644 --- a/test/fold.test.ts +++ b/test/fold.test.ts @@ -61,3 +61,8 @@ describe("openSubtasks (the widget shows the next action, so the plan IS the tas expect(openSubtasks(plan, active)).not.toContain("write the readme"); }); }); + +it("does not show historical Log subtasks under the last active goal", () => { + const plan = "1. [/] goal: current\n - [ ] current task\n\n## Log\n - [ ] historical task\n"; + expect(openSubtasks(plan, 0)).toEqual(["current task"]); +}); diff --git a/test/goals-flow.test.ts b/test/goals-flow.test.ts index ac78504..b6ac07b 100644 --- a/test/goals-flow.test.ts +++ b/test/goals-flow.test.ts @@ -5,7 +5,9 @@ import { join } from "node:path"; import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { afterEach, describe, expect, it, vi } from "vitest"; import { approvalPath, goalBlock, hashGoalBlock, repositoryState, writeApproval } from "../src/approval.js"; +import { GoalIntercom } from "../src/intercom.js"; import { intercomFixture } from "./intercom-fixture.js"; +import { pairedIntercomFixture } from "./paired-intercom-fixture.js"; const openSupervisorPane = vi.fn(async () => "pane-2"); const closeSupervisorPane = vi.fn(async () => undefined); @@ -13,7 +15,7 @@ const shutdowns: Array<() => Promise> = []; vi.mock("../src/herdr.js", () => ({ openSupervisorPane, closeSupervisorPane })); const { default: piGoalsExtension, isMainSession } = await import("../src/index.js"); -function setup(selectChoices: Array, editorChoices: Array = []) { +function setup(selectChoices: Array, editorChoices: Array = [], events?: ExtensionAPI["events"]) { const cwd = mkdtempSync(join(tmpdir(), "pi-goals-flow-")); writeFileSync(join(cwd, ".gitignore"), ".pi/\n"); writeFileSync(join(cwd, "verify.txt"), "PASS\n"); @@ -51,7 +53,7 @@ function setup(selectChoices: Array, editorChoices: Array "pane-2"); const pi = { - events: transport.events, + events: events ?? transport.events, registerCommand: (name: string, command: any) => commands.set(name, command), on: (name: string, handler: any) => { const prior = hooks.get(name); @@ -239,10 +241,14 @@ describe("/goals flow", () => { verifyOutputPath: "verify.txt", supervisor: { sessionId: "supervisor", runId: null }, timestamp: new Date().toISOString(), }); - writeFileSync(planPath, `${plan}- Appended manual log after approval.\n`); + writeFileSync(planPath, `${plan}- Appended manual log after approval.\n1. [ ] goal: make the file\n2. [ ] goal: historical only\n`); const signed = await flow.tools.get("CompleteGoal").execute("id", { goal }, undefined, undefined, flow.ctx); expect(signed.isError).toBe(false); expect(readFileSync(planPath, "utf8")).toContain("1. [x] goal: make the file"); + expect(readFileSync(planPath, "utf8")).toContain("1. [ ] goal: make the file"); + await flow.hooks.get("agent_settled")({}, flow.ctx); + expect(flow.entries.at(-1)?.data).toMatchObject({ phase: null }); + expect(flow.ctx.ui.setWidget).toHaveBeenLastCalledWith("pi-goals-widget", ["✔ complete"]); } finally { rmSync(flow.cwd, { recursive: true, force: true }); } @@ -374,3 +380,99 @@ it("does not persist startup results or launch work after session shutdown", asy expect(flow.messages).toHaveLength(messages); } finally { rmSync(flow.cwd, { recursive: true, force: true }); } }); + +it("keeps a failed Ready model not-ready and recovers the same real supervisor binding", async () => { + const wire = pairedIntercomFixture(); + const flow = setup(["Ready", "Ready"], [], wire.worker.events as ExtensionAPI["events"]); + const supervisorEntries: any[] = []; + const supervisor = new GoalIntercom({ events: wire.supervisor.events, on: () => {}, appendEntry: (customType: string, data: unknown) => supervisorEntries.push({ type: "custom", customType, data }) } as unknown as ExtensionAPI); + const supervisorCtx = { sessionManager: { getEntries: () => supervisorEntries }, ui: { notify: vi.fn() } }; + try { + await flow.commands.get("goals").handler("make the file", flow.ctx); + approvedPlan(flow.cwd); + mkdirSync(join(flow.cwd, ".pi/pi-goals/models"), { recursive: true }); + writeFileSync(join(flow.cwd, ".pi/pi-goals/models/worker.json"), JSON.stringify({ provider: "gone", id: "expired" })); + flow.ctx.modelRegistry.find = () => undefined as any; + openSupervisorPane.mockImplementationOnce(async (input: any) => { + supervisor.configure(input.approvalId, "supervisor", supervisorCtx as any, true); + return "pane-2"; + }); + await flow.hooks.get("agent_settled")({}, flow.ctx); + await new Promise(resolve => setImmediate(resolve)); + const binding = (flow.entries.at(-1)?.data as any).approvalId; + expect(flow.entries.at(-1)?.data).toMatchObject({ phase: "planning", planVersion: 1 }); + expect(wire.worker.sent.filter(message => message.kind === "hello" && message.ready)).toHaveLength(0); + expect(supervisor.connected).toBe(false); + expect(() => supervisor.steer("Must wait.")).toThrow("disconnected"); + await flow.hooks.get("model_select")({ source: "set", model: { provider: "test", id: "chosen" } }, flow.ctx); + flow.ctx.modelRegistry.find = (provider, id) => ({ provider, id }); + await flow.commands.get("goals").handler("reconnect", flow.ctx); + expect(supervisor.connected).toBe(false); // Planning is not implementation readiness. + await flow.hooks.get("agent_settled")({}, flow.ctx); + await new Promise(resolve => setImmediate(resolve)); + expect(supervisor.connected).toBe(true); + expect(flow.entries.at(-1)?.data).toMatchObject({ phase: "working", approvalId: binding, planVersion: 1 }); + expect(openSupervisorPane).toHaveBeenCalledTimes(1); + expect(closeSupervisorPane).not.toHaveBeenCalled(); + await flow.commands.get("goals").handler("reconnect", flow.ctx); + supervisor.steer("Recovered instruction."); + await new Promise(resolve => setImmediate(resolve)); + expect(flow.messages.filter(message => message.content === "[supervisor] Recovered instruction.")).toHaveLength(1); + expect(supervisorCtx.ui.notify).not.toHaveBeenCalled(); + } finally { rmSync(flow.cwd, { recursive: true, force: true }); } +}); + +it("points a present-but-paused peer recovery at the supervisor pane", async () => { + const flow = setup([]); + try { + restoredPlan(flow); + await flow.hooks.get("session_start")({}, flow.ctx); + flow.transport.replyToHello(false); + flow.transport.receive({ binding: "restored-binding", role: "supervisor", kind: "hello", id: "hello", ready: false }); + expect(flow.ctx.ui.setWidget).toHaveBeenLastCalledWith("pi-goals-widget", [expect.stringContaining("Supervisor is present but not ready")]); + const prompt = await flow.hooks.get("before_agent_start")({}, flow.ctx); + expect(prompt.systemPrompt).toContain("/goals reconnect in the supervisor pane"); + } finally { rmSync(flow.cwd, { recursive: true, force: true }); } +}); + +it("clear during the initial Ready wait cancels immediately and cannot resurrect the plan", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "setInterval", "clearInterval"] }); + const flow = setup(["Ready"]); + try { + await flow.commands.get("goals").handler("make the file", flow.ctx); + const path = approvedPlan(flow.cwd); + flow.transport.replyToHello(false); + const starting = flow.hooks.get("agent_settled")({}, flow.ctx); + await new Promise(resolve => setImmediate(resolve)); + expect(flow.entries.at(-1)?.data).toMatchObject({ supervisorPaneId: "pane-2" }); + await flow.commands.get("goals").handler("clear", flow.ctx); + await starting; // No timer advancement: detach must cancel the five-minute wait. + const entryCount = flow.entries.length; + await vi.advanceTimersByTimeAsync(300_000); + expect(flow.entries).toHaveLength(entryCount); + expect(flow.entries.at(-1)?.data).toMatchObject({ phase: null, planVersion: null, approvalId: null }); + expect(flow.messages.some(message => message.content.includes("Begin implementation"))).toBe(false); + expect(flow.ctx.ui.setWidget).toHaveBeenLastCalledWith("pi-goals-widget", undefined); + expect(readFileSync(path, "utf8")).toContain("make the file"); + } finally { rmSync(flow.cwd, { recursive: true, force: true }); } +}); + +it("clear before the launcher resolves rejects late pane callbacks without restoring the binding", async () => { + const flow = setup(["Ready"]); + try { + await flow.commands.get("goals").handler("make the file", flow.ctx); + approvedPlan(flow.cwd); + let finish!: () => void; + openSupervisorPane.mockImplementationOnce((_input: any, opened: any) => new Promise((resolve, reject) => { + finish = () => { try { opened("late-pane"); resolve("late-pane"); } catch (error) { reject(error); } }; + })); + const starting = flow.hooks.get("agent_settled")({}, flow.ctx); + await new Promise(resolve => setImmediate(resolve)); + await flow.commands.get("goals").handler("clear", flow.ctx); + const entryCount = flow.entries.length; + finish(); await starting; + expect(flow.entries).toHaveLength(entryCount); + expect(flow.entries.at(-1)?.data).toMatchObject({ phase: null, planVersion: null, approvalId: null }); + expect(flow.messages.some(message => message.content.includes("Begin implementation"))).toBe(false); + } finally { rmSync(flow.cwd, { recursive: true, force: true }); } +}); diff --git a/test/intercom-fixture.ts b/test/intercom-fixture.ts index 7ab78e7..3be85aa 100644 --- a/test/intercom-fixture.ts +++ b/test/intercom-fixture.ts @@ -18,7 +18,7 @@ export function intercomFixture() { snapshot: () => ({ connected, supported: true }), publish: (message: any) => { sent.push(message); - if (message.kind === "hello" && autoHello) queueMicrotask(() => receive({ ...message, role: message.role === "worker" ? "supervisor" : "worker", ready: true })); + if (message.kind === "hello" && !message.reply && autoHello) queueMicrotask(() => receive({ ...message, role: message.role === "worker" ? "supervisor" : "worker", ready: true, reply: true })); }, }); return true; diff --git a/test/intercom-handshake.test.ts b/test/intercom-handshake.test.ts new file mode 100644 index 0000000..32f404f --- /dev/null +++ b/test/intercom-handshake.test.ts @@ -0,0 +1,81 @@ +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; +import { expect, it, vi } from "vitest"; +import { GoalIntercom } from "../src/intercom.js"; +import { pairedIntercomFixture } from "./paired-intercom-fixture.js"; + +function endpoint(transport: ReturnType["worker"]) { + const entries: any[] = []; + const ctx = { sessionManager: { getEntries: () => entries }, ui: { notify: vi.fn() } }; + const link = new GoalIntercom({ events: transport.events, on: () => {}, appendEntry: (customType: string, data: unknown) => entries.push({ type: "custom", customType, data }) } as unknown as ExtensionAPI); + return { link, ctx, entries }; +} +const settle = () => new Promise(resolve => setImmediate(resolve)); + +it("re-handshakes unchanged peers in either direction without hello ping-pong or lost advice", async () => { + const wire = pairedIntercomFixture(); + const worker = endpoint(wire.worker); + const supervisor = endpoint(wire.supervisor); + worker.link.configure("binding", "worker", worker.ctx as any); + supervisor.link.configure("binding", "supervisor", supervisor.ctx as any); + supervisor.link.markReady(); + await settle(); + expect(worker.link.connected && supervisor.link.connected).toBe(true); + const deliver = vi.fn(); worker.link.onSteer = deliver; + for (const side of [worker, supervisor, worker]) { + const before = wire.worker.sent.length + wire.supervisor.sent.length; + side.link.configure("binding", side === worker ? "worker" : "supervisor", side.ctx as any, true); + await side.link.waitReady(100); + await settle(); + expect(worker.link.connected && supervisor.link.connected).toBe(true); + expect(wire.worker.sent.length + wire.supervisor.sent.length - before).toBe(2); + } + const before = wire.worker.sent.length + wire.supervisor.sent.length; + worker.link.configure("binding", "worker", worker.ctx as any, true); + supervisor.link.configure("binding", "supervisor", supervisor.ctx as any, true); + await settle(); + expect(worker.link.connected && supervisor.link.connected).toBe(true); + expect(wire.worker.sent.length + wire.supervisor.sent.length - before).toBe(4); + supervisor.link.steer("Read actual output."); + await settle(); + expect(deliver).toHaveBeenCalledExactlyOnceWith("Read actual output."); + expect(worker.ctx.ui.notify).not.toHaveBeenCalled(); + expect(supervisor.ctx.ui.notify).not.toHaveBeenCalled(); +}); + +it("replays pending advice and views across either role's own readiness transition", async () => { + const wire = pairedIntercomFixture(); + const worker = endpoint(wire.worker), supervisor = endpoint(wire.supervisor); + worker.link.configure("binding", "worker", worker.ctx as any); + supervisor.link.configure("binding", "supervisor", supervisor.ctx as any, true); + await settle(); + wire.supervisor.drop = message => message.kind === "steer"; + supervisor.link.steer("Pending advice."); + wire.supervisor.drop = () => false; + const deliver = vi.fn(); worker.link.onSteer = deliver; + supervisor.link.markNotReady(); await settle(); + expect(worker.link.connected).toBe(false); + supervisor.link.markReady(); await settle(); + expect(deliver).toHaveBeenCalledExactlyOnceWith("Pending advice."); + worker.link.markNotReady(); await settle(); + const onView = vi.fn(); supervisor.link.onView = onView; + worker.link.view("Fresh view.", "settled"); + expect(onView).not.toHaveBeenCalled(); + worker.link.markReady(); await settle(); + expect(onView).toHaveBeenCalledTimes(1); + expect(worker.link.connected && supervisor.link.connected).toBe(true); + wire.worker.connect(false); wire.worker.connect(true); await settle(); + expect(worker.link.connected && supervisor.link.connected).toBe(true); + expect(deliver).toHaveBeenCalledTimes(1); + expect(onView).toHaveBeenCalledTimes(1); +}); + +it("cancels pending waits immediately on detach or reconfiguration", async () => { + const wire = pairedIntercomFixture(); + const worker = endpoint(wire.worker); + worker.link.configure("binding", "worker", worker.ctx as any); + const cancelled = expect(worker.link.waitReady()).rejects.toThrow("plan detached"); + worker.link.detach(); await cancelled; + worker.link.configure("next", "worker", worker.ctx as any); + const replaced = expect(worker.link.waitReady()).rejects.toThrow("reconfiguration"); + worker.link.configure("third", "worker", worker.ctx as any); await replaced; +}); diff --git a/test/intercom.test.ts b/test/intercom.test.ts index ee5ebcf..32387cc 100644 --- a/test/intercom.test.ts +++ b/test/intercom.test.ts @@ -42,14 +42,16 @@ describe("pi-intercom transport", () => { const resumed = setup("supervisor", [...first.entries]); resumed.link.markReady(); await resumed.link.waitReady(); - expect(resumed.fixture.sent.filter(message => message.kind === "steer")).toMatchObject([{ id, text: "Read the full output." }]); + const retries = resumed.fixture.sent.filter(message => message.kind === "steer"); + expect(retries.length).toBeGreaterThan(0); + for (const retry of retries) expect(retry).toMatchObject({ id, text: "Read the full output." }); resumed.fixture.receive({ binding: "binding", role: "worker", kind: "received", id }); resumed.fixture.connect(false); expect(resumed.link.connected).toBe(false); expect(() => resumed.link.steer("Must not send.")).toThrow("disconnected"); resumed.fixture.connect(true); await resumed.link.waitReady(); - expect(resumed.fixture.sent.filter(message => message.kind === "steer")).toHaveLength(1); + expect(resumed.fixture.sent.filter(message => message.kind === "steer")).toHaveLength(retries.length); }); it("advances the incremental overview only after acknowledgment", async () => { diff --git a/test/paired-intercom-fixture.ts b/test/paired-intercom-fixture.ts new file mode 100644 index 0000000..5778cd9 --- /dev/null +++ b/test/paired-intercom-fixture.ts @@ -0,0 +1,31 @@ +// Wire two real GoalIntercom adapters. This router never invents hello replies. +export function pairedIntercomFixture() { + const receivers = new Map void>(); + const endpoint = (id: string, peer: string) => { + const sent: any[] = []; + let connected = true; + const transport = { + sent, + drop: (_message: any) => false, + connect: (value: boolean) => { connected = value; receivers.get(id)?.({ type: "connection", connected: value, supported: true }); }, + events: { + on: () => () => {}, + emit: (name: string, registration: any) => { + if (name !== "intercom:extension-register") return false; + receivers.set(id, registration.onEvent); + registration.onReady({ + snapshot: () => ({ connected, supported: true }), + publish: (message: any) => { + sent.push(message); + if (sent.length > 200) throw new Error("Handshake did not settle; possible hello loop."); + if (connected && !transport.drop(message)) queueMicrotask(() => receivers.get(peer)?.({ type: "message", fromSessionId: id, payload: message })); + }, + }); + return true; + }, + }, + }; + return transport; + }; + return { worker: endpoint("worker", "supervisor"), supervisor: endpoint("supervisor", "worker") }; +} diff --git a/test/supervisor-session.test.ts b/test/supervisor-session.test.ts index c9c49d1..75b4226 100644 --- a/test/supervisor-session.test.ts +++ b/test/supervisor-session.test.ts @@ -271,3 +271,20 @@ it("keeps a supervisor unready after model restoration failure, then recovers ex expect(runtime.ready()).toBe(true); } finally { rmSync(cwd, { recursive: true, force: true }); } }); + +it("warns once on unavailable usage but stays quiet for Pi's post-compaction null token sample", async () => { + const cwd = mkdtempSync(join(tmpdir(), "pi-goals-usage-")); + try { + const runtime = setup(cwd, join(cwd, "plan.md")); + await runtime.start(); + runtime.ctx.ui.notify.mockClear(); + runtime.ctx.getContextUsage = () => ({ tokens: null }) as any; + await runtime.hooks.get("agent_settled")({}, runtime.ctx); + expect(runtime.ctx.ui.notify).not.toHaveBeenCalled(); + runtime.ctx.getContextUsage = () => undefined; + await runtime.hooks.get("agent_settled")({}, runtime.ctx); + await runtime.hooks.get("agent_settled")({}, runtime.ctx); + expect(runtime.ctx.ui.notify).toHaveBeenCalledExactlyOnceWith(expect.stringContaining("custom 100k compaction trigger cannot be checked"), "warning"); + expect(runtime.ctx.compact).not.toHaveBeenCalled(); + } finally { rmSync(cwd, { recursive: true, force: true }); } +}); diff --git a/test/tick-goal.test.ts b/test/tick-goal.test.ts index f5655c8..66edc7c 100644 --- a/test/tick-goal.test.ts +++ b/test/tick-goal.test.ts @@ -26,7 +26,14 @@ describe("tickGoal (sign-off ticks the goal; agent only ticks on wording drift)" }); it("returns null when the subject matches more than one goal line", () => { - const dup = `${plan}3. [ ] goal: Ship the docs\n`; + const dup = plan.replace("## Log", "3. [ ] goal: Ship the docs\n\n## Log"); expect(tickGoal(dup, "Ship the docs")).toBeNull(); }); + + it("ignores a historical duplicate below the Log and leaves it unchanged", () => { + const historical = `${plan}3. [ ] goal: Ship the docs\n`; + const result = tickGoal(historical, "Ship the docs"); + expect(result).toContain("2. [x] goal: Ship the docs"); + expect(result).toContain("## Log\n3. [ ] goal: Ship the docs\n"); + }); });