diff --git a/src/herdr.ts b/src/herdr.ts index 22443a5..bc80bfc 100644 --- a/src/herdr.ts +++ b/src/herdr.ts @@ -7,6 +7,7 @@ interface LaunchSupervisorInput { cwd: string; sourceSessionFile: string; workerSessionId: string; + workerIntercomId: string; planPath: string; approvalId: string; extensionPath: string; @@ -30,16 +31,24 @@ function findPaneId(value: unknown): string | null { return null; } -async function herdr(args: string[]): Promise { +async function herdr(args: string[], json = true): Promise { const bin = process.env.HERDR_BIN_PATH ?? "herdr"; const { stdout } = await execFileAsync(bin, args, { encoding: "utf8", timeout: 15_000 }); + if (!json) return stdout.trim(); return stdout.trim() ? JSON.parse(stdout) : {}; } +function stalePaneError(error: unknown): boolean { + const record = error as { stdout?: unknown; stderr?: unknown; message?: unknown }; + const text = [record.stdout, record.stderr, record.message].filter((value): value is string => typeof value === "string").join("\n"); + return /\b(?:NOT_FOUND|PANE_GONE|PANE_NOT_FOUND)\b/i.test(text); +} + export function supervisorCommand(input: LaunchSupervisorInput): string { const env = [ "PI_GOALS_ROLE=supervisor", `PI_GOALS_WORKER_ID=${input.workerSessionId}`, + `PI_GOALS_WORKER_INTERCOM_ID=${input.workerIntercomId}`, `PI_GOALS_PLAN_PATH=${input.planPath}`, `PI_GOALS_APPROVAL_ID=${input.approvalId}`, `PI_GOALS_OWNER_SESSION_ID=${input.workerSessionId}`, @@ -49,7 +58,7 @@ export function supervisorCommand(input: LaunchSupervisorInput): string { "--no-extensions", "-e", input.extensionPath, "-e", "npm:pi-intercom", - "-e", "npm:@wassname2/pi-supervise", + "-e", "npm:@wassname2/pi-supervise@0.0.4", "--fork", input.sourceSessionFile, "--name", `goals-supervisor-${input.workerSessionId.slice(0, 8)}`, ]; @@ -59,7 +68,7 @@ export function supervisorCommand(input: LaunchSupervisorInput): string { export async function openSupervisorPane(input: LaunchSupervisorInput): Promise { if (process.env.HERDR_ENV !== "1") throw new Error("Ready needs a Herdr session so pi-goals can open the supervisor session."); - await herdr(["--version"]); + await herdr(["--version"], false); const split = await herdr(["pane", "split", "--current", "--direction", "right", "--cwd", input.cwd, "--no-focus"]); const paneId = findPaneId(split); if (!paneId) throw new Error("Herdr did not return the new supervisor pane ID."); @@ -73,5 +82,10 @@ export async function openSupervisorPane(input: LaunchSupervisorInput): Promise< } export async function closeSupervisorPane(paneId: string): Promise { - await herdr(["pane", "close", paneId]); + try { + await herdr(["pane", "close", paneId]); + } catch (error) { + if (stalePaneError(error)) return; + throw error; + } } diff --git a/src/index.ts b/src/index.ts index 77ba679..abdfbc6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -22,6 +22,7 @@ import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-a import { Type } from "typebox"; import { approvalMatches, approvalPath, goalBlock, hashGoalBlock, readApproval, repositoryState } from "./approval.js"; import { closeSupervisorPane, openSupervisorPane } from "./herdr.js"; +import { registerGoalsIntercom } from "./intercom.js"; import { completeGoalDescription, completeGoalParamDescription, planDrafting, planningState, resync } from "./prompts.js"; import { isVisibleSupervisor, registerVisibleSupervisor } from "./supervisor-session.js"; @@ -99,10 +100,11 @@ interface PlanState { export default function piGoalsExtension(pi: ExtensionAPI): void { if (isVisibleSupervisor()) { - registerVisibleSupervisor(pi); + registerVisibleSupervisor(pi, registerGoalsIntercom(pi)); return; } if (!isMainSession()) return; + const intercom = registerGoalsIntercom(pi); let state: PlanState = { phase: null, supervisorModel: null, @@ -146,19 +148,26 @@ export default function piGoalsExtension(pi: ExtensionAPI): void { repositoryRoot(ctx.cwd); const sourceSessionFile = ctx.sessionManager.getSessionFile(); if (!sourceSessionFile) throw new Error("The current session is not persisted, so it cannot be forked."); + const workerIntercomId = await intercom.workerIntercomId(); beginReview(ctx); - state = { - ...state, - supervisorPaneId: await openSupervisorPane({ + let paneId: string | null = null; + try { + paneId = await openSupervisorPane({ cwd: ctx.cwd, sourceSessionFile, workerSessionId: ctx.sessionManager.getSessionId(), + workerIntercomId, planPath: planPath(ctx), approvalId: state.approvalId!, extensionPath: fileURLToPath(import.meta.url), model: state.supervisorModel, - }), - }; + }); + await intercom.waitForSupervisorReady(state.approvalId!); + } catch (error) { + if (paneId) await closeSupervisorPane(paneId).catch(() => {}); + throw error; + } + state = { ...state, supervisorPaneId: paneId }; persist(); } diff --git a/src/intercom.ts b/src/intercom.ts new file mode 100644 index 0000000..4c314c5 --- /dev/null +++ b/src/intercom.ts @@ -0,0 +1,100 @@ +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; + +const INTERCOM_REGISTER_EVENT = "intercom:extension-register"; +const NAMESPACE = "pi-goals/visible-supervisor/v1"; +const READY_TIMEOUT_MS = 15_000; + +type Channel = { + snapshot(): { connected: boolean }; + publish(payload: unknown, options?: { audience?: "owner" | "capable"; ownerOnly?: boolean }): void; + listSessions(): Promise>; +}; + +type IntercomEvent = { type: string; connected?: boolean; fromSessionId?: string; payload?: unknown }; + +type ReadyMessage = { type: "supervisor-ready"; to: string; approvalId: string }; + +function timeout(promise: Promise, message: string): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(message)), READY_TIMEOUT_MS); + promise.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (error) => { + clearTimeout(timer); + reject(error); + }, + ); + }); +} + +function isReadyMessage(value: unknown): value is ReadyMessage { + if (!value || typeof value !== "object") return false; + const record = value as Record; + return record.type === "supervisor-ready" && typeof record.to === "string" && typeof record.approvalId === "string"; +} + +export interface GoalsIntercom { + workerIntercomId(): Promise; + waitForSupervisorReady(approvalId: string): Promise; + announceSupervisorReady(workerIntercomId: string, approvalId: string): Promise; +} + +export function registerGoalsIntercom(pi: ExtensionAPI): GoalsIntercom { + let channel: Channel | undefined; + let resolveIntercomConnected!: () => void; + const intercomConnected = new Promise((resolve) => { + resolveIntercomConnected = resolve; + }); + const ready = new Map void>(); + const announced = new Set(); + let ownId = ""; + + const currentId = async (): Promise => { + await intercomConnected; + if (ownId) return ownId; + const sessions = await channel!.listSessions(); + const session = sessions.find((item) => item.pid === process.pid); + if (!session) throw new Error("pi-goals could not find this Pi session in pi-intercom."); + ownId = session.id; + return ownId; + }; + + (pi as unknown as { events: { emit(name: string, value: unknown): void } }).events.emit(INTERCOM_REGISTER_EVENT, { + namespace: NAMESPACE, + ownerEligible: false, + onReady(value: Channel) { + channel = value; + if (value.snapshot().connected) resolveIntercomConnected(); + }, + onEvent(event: IntercomEvent) { + if (event.type === "connection" && event.connected) { + resolveIntercomConnected(); + return; + } + if (event.type !== "message" || !isReadyMessage(event.payload)) return; + if (event.payload.to !== ownId) return; + const resolve = ready.get(event.payload.approvalId); + if (!resolve) { + announced.add(event.payload.approvalId); + return; + } + ready.delete(event.payload.approvalId); + resolve(); + }, + }); + + return { + workerIntercomId: () => timeout(currentId(), "pi-goals needs pi-intercom before it can start a visible supervisor."), + waitForSupervisorReady(approvalId) { + if (announced.delete(approvalId)) return Promise.resolve(); + return timeout(new Promise((resolve) => ready.set(approvalId, resolve)), "The visible supervisor did not acknowledge pairing with this worker."); + }, + async announceSupervisorReady(workerIntercomId, approvalId) { + await timeout(intercomConnected, "pi-goals needs pi-intercom before it can confirm visible-supervisor pairing."); + channel!.publish({ type: "supervisor-ready", to: workerIntercomId, approvalId }, { audience: "capable" }); + }, + }; +} diff --git a/src/supervise.ts b/src/supervise.ts new file mode 100644 index 0000000..bd73ebc --- /dev/null +++ b/src/supervise.ts @@ -0,0 +1,21 @@ +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; + +const PAIR_EVENT = "pi-supervise:pair:v1"; +const PAIR_TIMEOUT_MS = 15_000; + +export function pairWithPiSupervise(pi: ExtensionAPI, workerIntercomId: string, goal: string): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("pi-supervise did not accept the visible-supervisor pairing request.")), PAIR_TIMEOUT_MS); + const settle = (callback: () => void) => { + clearTimeout(timer); + callback(); + }; + (pi as unknown as { events: { emit(name: string, value: unknown): void } }).events.emit(PAIR_EVENT, { + version: 1, + workerIntercomId, + goal, + resolve: () => settle(resolve), + reject: (error: Error) => settle(() => reject(error)), + }); + }); +} diff --git a/src/supervisor-session.ts b/src/supervisor-session.ts index 0e297a5..738b817 100644 --- a/src/supervisor-session.ts +++ b/src/supervisor-session.ts @@ -3,12 +3,15 @@ import { resolve } from "node:path"; import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent"; import { Type } from "typebox"; import { approvalPath, goalBlock, hashGoalBlock, repositoryState, writeApproval } from "./approval.js"; +import type { GoalsIntercom } from "./intercom.js"; +import { pairWithPiSupervise } from "./supervise.js"; const BOOTSTRAPPED = "pi-goals-visible-supervisor-v1"; const COMPACT_AT_TOKENS = 100_000; interface SupervisorConfig { workerSessionId: string; + workerIntercomId: string; ownerSessionId: string; planPath: string; approvalId: string; @@ -27,12 +30,30 @@ function requiredEnv(name: string): string { function config(): SupervisorConfig { return { workerSessionId: requiredEnv("PI_GOALS_WORKER_ID"), + workerIntercomId: requiredEnv("PI_GOALS_WORKER_INTERCOM_ID"), ownerSessionId: requiredEnv("PI_GOALS_OWNER_SESSION_ID"), planPath: resolve(requiredEnv("PI_GOALS_PLAN_PATH")), approvalId: requiredEnv("PI_GOALS_APPROVAL_ID"), }; } +function hasEvidenceEntry(block: string): boolean { + const lines = block.split("\n"); + for (let index = 0; index < lines.length; index++) { + const evidence = /^\s*[-*]\s+evidence:\s*(.*)$/i.exec(lines[index]); + if (!evidence) continue; + if (evidence[1].trim() && !/^\(empty until sign-off\)$/i.test(evidence[1].trim())) return true; + const indent = lines[index].match(/^\s*/)?.[0].length ?? 0; + for (let child = index + 1; child < lines.length; child++) { + const childIndent = lines[child].match(/^\s*/)?.[0].length ?? 0; + if (lines[child].trim() && childIndent <= indent) break; + const entry = /^\s+[-*]\s+(.+?)\s*$/.exec(lines[child]); + if (entry?.[1].trim()) return true; + } + } + return false; +} + function latestWorkerView(ctx: ExtensionContext): string | null { for (const entry of [...ctx.sessionManager.getBranch()].reverse()) { const message = (entry as { type?: string; message?: { role?: string; content?: unknown[] } }).message; @@ -55,7 +76,7 @@ export function isVisibleSupervisor(): boolean { return process.env.PI_GOALS_ROLE === "supervisor"; } -export function registerVisibleSupervisor(pi: ExtensionAPI): void { +export function registerVisibleSupervisor(pi: ExtensionAPI, intercom: GoalsIntercom): void { const settings = config(); let compacting = false; @@ -66,24 +87,23 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void { pi.on("session_start", async (_event, ctx) => { const entries = ctx.sessionManager.getEntries(); if (entries.some((entry: { type?: string; customType?: string }) => entry.type === "custom" && entry.customType === BOOTSTRAPPED)) return; - if (!pi.getCommands().some((command) => command.name === "supervise" && command.source === "extension")) { - ctx.ui.notify("pi-goals supervisor needs the @wassname2/pi-supervise extension.", "error"); - return; - } compacting = true; - ctx.compact({ - customInstructions: `Preserve the user's decisions, preferences, and high-level objective from planning. Preserve unresolved risks and the plan path ${settings.planPath}. Remove implementation chatter. This summary is for a read-only supervisor that will judge and steer another Pi session.`, - onComplete: () => { - compacting = false; - pi.appendEntry(BOOTSTRAPPED, { version: 1, workerSessionId: settings.workerSessionId, planPath: settings.planPath }); - const sendCommand = pi.sendUserMessage as (content: string, options: { expandPromptTemplates: boolean }) => void; - sendCommand(`/supervise @${settings.workerSessionId} ${settings.planPath}`, { expandPromptTemplates: true }); - }, - onError: (error) => { - compacting = false; - ctx.ui.notify(`Supervisor compaction failed: ${error.message}`, "error"); - }, - }); + try { + await new Promise((resolve, reject) => { + ctx.compact({ + customInstructions: `Preserve the user's decisions, preferences, and high-level objective from planning. Preserve unresolved risks and the plan path ${settings.planPath}. Remove implementation chatter. This summary is for a read-only supervisor that will judge and steer another Pi session.`, + onComplete: () => resolve(), + onError: reject, + }); + }); + await pairWithPiSupervise(pi, settings.workerIntercomId, settings.planPath); + await intercom.announceSupervisorReady(settings.workerIntercomId, settings.approvalId); + pi.appendEntry(BOOTSTRAPPED, { version: 1, workerSessionId: settings.workerSessionId, planPath: settings.planPath }); + } catch (error) { + ctx.ui.notify(`Supervisor startup failed: ${error instanceof Error ? error.message : String(error)}`, "error"); + } finally { + compacting = false; + } }); pi.on("agent_settled", async (_event, ctx) => { @@ -131,7 +151,7 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void { if (!repository.cleanWorktree) return result("Cannot approve with a dirty worktree. Commit the worker changes first.", true); const block = goalBlock(plan, params.goal); if (!block) return result(`Cannot approve: no unique open goal matches "${params.goal}".`, true); - if (/evidence:\s*\(empty until sign-off\)/i.test(block)) return result("Cannot approve while the goal evidence is empty.", true); + if (!hasEvidenceEntry(block)) return result("Cannot approve without a nonblank evidence entry in the goal block.", true); const path = approvalPath(ctx.cwd, settings.ownerSessionId, params.goal); writeApproval(path, { version: 2, diff --git a/test/goals-flow.test.ts b/test/goals-flow.test.ts index 5741357..8ca1876 100644 --- a/test/goals-flow.test.ts +++ b/test/goals-flow.test.ts @@ -9,6 +9,13 @@ import { approvalPath, goalBlock, hashGoalBlock, repositoryState, writeApproval const openSupervisorPane = vi.fn(async () => "pane-2"); const closeSupervisorPane = vi.fn(async () => undefined); vi.mock("../src/herdr.js", () => ({ openSupervisorPane, closeSupervisorPane })); +vi.mock("../src/intercom.js", () => ({ + registerGoalsIntercom: () => ({ + workerIntercomId: async () => "worker-intercom", + waitForSupervisorReady: async () => {}, + announceSupervisorReady: async () => {}, + }), +})); const { default: piGoalsExtension, isMainSession } = await import("../src/index.js"); @@ -101,6 +108,7 @@ describe("/goals flow", () => { cwd: flow.cwd, sourceSessionFile: join(flow.cwd, "session.jsonl"), workerSessionId: "session-a", + workerIntercomId: "worker-intercom", planPath, })); expect(flow.entries.at(-1)?.data).toMatchObject({ phase: "working", supervisorPaneId: "pane-2" }); @@ -128,7 +136,7 @@ describe("/goals flow", () => { } }); - it("signs off only an approval for the exact clean commit and goal block", async () => { + it("accepts only an approval for the exact clean commit and goal block", async () => { const flow = setup(["Ready"]); try { await flow.commands.get("goals").handler("make the file", flow.ctx); diff --git a/test/herdr.test.ts b/test/herdr.test.ts index 36f0cf8..db5ef40 100644 --- a/test/herdr.test.ts +++ b/test/herdr.test.ts @@ -1,22 +1,54 @@ -import { describe, expect, it } from "vitest"; -import { supervisorCommand } from "../src/herdr.js"; +import { chmodSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { closeSupervisorPane, openSupervisorPane, supervisorCommand } from "../src/herdr.js"; + +function input() { + return { + cwd: "/repo", + sourceSessionFile: "/sessions/worker.jsonl", + workerSessionId: "worker-12345678", + workerIntercomId: "intercom-12345678", + planPath: "/repo/.pi/plan/worker-v1.md", + approvalId: "approval-1", + extensionPath: "/repo/src/index.ts", + model: "provider/supervisor", + }; +} + +afterEach(() => vi.unstubAllEnvs()); describe("supervisor pane command", () => { it("forks the planning session with an explicit supervisor role and model", () => { - const command = supervisorCommand({ - cwd: "/repo", - sourceSessionFile: "/sessions/worker.jsonl", - workerSessionId: "worker-12345678", - planPath: "/repo/.pi/plan/worker-v1.md", - approvalId: "approval-1", - extensionPath: "/repo/src/index.ts", - model: "provider/supervisor", - }); + const command = supervisorCommand(input()); expect(command).toContain("'PI_GOALS_ROLE=supervisor'"); + expect(command).toContain("'PI_GOALS_WORKER_INTERCOM_ID=intercom-12345678'"); expect(command).toContain("'pi' '--no-extensions' '-e' '/repo/src/index.ts'"); - expect(command).toContain("'-e' 'npm:pi-intercom' '-e' 'npm:@wassname2/pi-supervise'"); + expect(command).toContain("'-e' 'npm:pi-intercom' '-e' 'npm:@wassname2/pi-supervise@0.0.4'"); expect(command).toContain("'--fork' '/sessions/worker.jsonl'"); expect(command).toContain("'--model' 'provider/supervisor'"); expect(command).not.toContain("pi-subagents"); }); + + it("accepts Herdr's text version output and stale pane cleanup", async () => { + const cwd = mkdtempSync(join(tmpdir(), "pi-goals-herdr-")); + const bin = join(cwd, "herdr"); + writeFileSync(bin, `#!/bin/sh +if [ "$1" = "--version" ]; then echo "herdr 0.8.2"; exit 0; fi +if [ "$1" = "pane" ] && [ "$2" = "split" ]; then echo '{"pane_id":"new-pane"}'; exit 0; fi +if [ "$1" = "pane" ] && [ "$2" = "run" ]; then echo '{}'; exit 0; fi +if [ "$1" = "pane" ] && [ "$2" = "close" ]; then echo '{"error":{"code":"PANE_GONE"}}' >&2; exit 1; fi +exit 2 +`); + chmodSync(bin, 0o755); + vi.stubEnv("HERDR_ENV", "1"); + vi.stubEnv("HERDR_BIN_PATH", bin); + try { + await expect(openSupervisorPane(input())).resolves.toBe("new-pane"); + await expect(closeSupervisorPane("new-pane")).resolves.toBeUndefined(); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); }); diff --git a/test/supervisor-session.test.ts b/test/supervisor-session.test.ts index 11b809c..ed650bf 100644 --- a/test/supervisor-session.test.ts +++ b/test/supervisor-session.test.ts @@ -9,13 +9,15 @@ import { registerVisibleSupervisor } from "../src/supervisor-session.js"; function setup(cwd: string, planPath: string) { vi.stubEnv("PI_GOALS_WORKER_ID", "worker-session"); + vi.stubEnv("PI_GOALS_WORKER_INTERCOM_ID", "worker-intercom"); vi.stubEnv("PI_GOALS_OWNER_SESSION_ID", "worker-session"); vi.stubEnv("PI_GOALS_PLAN_PATH", planPath); vi.stubEnv("PI_GOALS_APPROVAL_ID", "approval-1"); const hooks = new Map(); const tools = new Map(); const entries: any[] = []; - const sent: Array<{ content: string; options?: unknown }> = []; + const paired: Array<{ workerIntercomId: string; goal: string }> = []; + const announced: Array<{ workerIntercomId: string; approvalId: string }> = []; let branch: any[] = []; const ctx = { cwd, @@ -30,14 +32,24 @@ function setup(cwd: string, planPath: string) { ui: { notify: vi.fn() }, }; const pi = { + events: { + on() {}, + emit(name: string, request: any) { + if (name !== "pi-supervise:pair:v1") return; + paired.push({ workerIntercomId: request.workerIntercomId, goal: request.goal }); + request.resolve(); + }, + }, on: (name: string, handler: any) => hooks.set(name, handler), registerTool: (tool: any) => tools.set(tool.name, tool), appendEntry: (customType: string, data: unknown) => entries.push({ type: "custom", customType, data }), - getCommands: () => [{ name: "supervise", source: "extension" }], - sendUserMessage: (content: string, options?: unknown) => sent.push({ content, options }), }; - registerVisibleSupervisor(pi as unknown as ExtensionAPI); - return { branch: (value: any[]) => { branch = value; }, ctx, entries, hooks, sent, tools }; + registerVisibleSupervisor(pi as unknown as ExtensionAPI, { + workerIntercomId: async () => "worker-intercom", + waitForSupervisorReady: async () => {}, + announceSupervisorReady: async (workerIntercomId, approvalId) => { announced.push({ workerIntercomId, approvalId }); }, + }); + return { announced, branch: (value: any[]) => { branch = value; }, ctx, entries, hooks, paired, tools }; } afterEach(() => vi.unstubAllEnvs()); @@ -50,10 +62,8 @@ describe("visible supervisor session", () => { await runtime.hooks.get("session_start")({}, runtime.ctx); expect(runtime.ctx.compact).toHaveBeenCalledOnce(); expect(runtime.entries.at(-1)).toMatchObject({ customType: "pi-goals-visible-supervisor-v1" }); - expect(runtime.sent).toEqual([{ - content: `/supervise @worker-session ${join(cwd, ".pi/plan/worker-v1.md")}`, - options: { expandPromptTemplates: true }, - }]); + expect(runtime.paired).toEqual([{ workerIntercomId: "worker-intercom", goal: join(cwd, ".pi/plan/worker-v1.md") }]); + expect(runtime.announced).toEqual([{ workerIntercomId: "worker-intercom", approvalId: "approval-1" }]); } finally { rmSync(cwd, { recursive: true, force: true }); } @@ -83,6 +93,12 @@ describe("visible supervisor session", () => { }, undefined, undefined, runtime.ctx); expect(approved.isError).toBe(false); expect(existsSync(approvalPath(cwd, "worker-session", "make the file"))).toBe(true); + writeFileSync(planPath, "# Plan\n\n## Goals\n\n1. [ ] goal: make the file\n - evidence:\n - \n - tasks:\n - write result.txt\n"); + const missingEvidence = await runtime.tools.get("ApproveGoal").execute("id", { + goal: "make the file", inspectedPlan: true, inspectedRepository: true, inspectedEvidence: true, inspectedVerifyOutput: true, + }, undefined, undefined, runtime.ctx); + expect(missingEvidence.isError).toBe(true); + expect(missingEvidence.content[0].text).toContain("nonblank evidence entry"); } finally { rmSync(cwd, { recursive: true, force: true }); }