mirror of
https://github.com/wassname/pi-goals.git
synced 2026-09-12 12:50:58 +08:00
Replace supervision mailbox polling with pi-intercom
Scope messages to each pairing, wait for supervisor readiness, retain instructions in session history until acknowledgment, and rejoin after disconnect. Reuse installed Intercom or load the package dependency. Validate over an isolated real broker; existing user panes are untouched.
This commit is contained in:
wassname
@@ -8,18 +8,18 @@ Plan in one Pi session, then do the work there while a stronger visible Pi sessi
|
||||
2. Pi asks only material questions, writes the plan, and shows **Ready / Refine / Edit / Cancel**.
|
||||
3. **Ready** opens a second Herdr pane. The new Pi session explicitly forks the planning session and compacts that fork.
|
||||
4. The original session becomes the implementation worker. It keeps the full conversation and normal tools.
|
||||
5. The fork becomes a read-only supervisor. Worker views and supervisor instructions use a session-scoped mailbox under ignored `.pi/goals-supervision/`.
|
||||
6. Ready waits for the supervisor's durable readiness receipt; the worker does not begin before the fork has compacted and started.
|
||||
5. The fork becomes a read-only supervisor. Worker views and supervisor instructions travel over pi-intercom's extension channel, scoped to this plan pairing.
|
||||
6. Ready waits for the supervisor's Intercom readiness message; the worker does not begin before the fork has compacted and started.
|
||||
7. The supervisor compacts again when its context reaches 100k tokens.
|
||||
8. The supervisor records a private approval only after it sees a stopped worker, no active work, a clean commit, evidence, and saved verification output. `CompleteGoal` checks that approval against the exact plan block and Git tree before it ticks `[x]`.
|
||||
|
||||
The two Pi sessions are visible. You can switch to the supervisor pane and talk to it directly. Supervisor instructions are shown in full, including in collapsed tool rows; ordinary messages and emitted thinking use Pi's display settings. The supervisor is prompted to give brief progress assessments and use judgment about when to intervene.
|
||||
|
||||
On resume, monitoring and read-only tools are restored. Periodic views report whether Pi is idle; they do not measure background jobs. Readiness is a startup receipt, not a continuous health check. Reviews stop after all goals are completed or cancelled, and both panes remain available. These mechanics are tested; useful judgment and savings from a cheaper worker still require a representative two-model run. -- Pi/OpenAI
|
||||
On resume, monitoring and read-only tools are restored. Periodic views report whether Pi is idle; they do not measure background jobs. Intercom disconnects are reported; unsent current views and unacknowledged instructions are retained in Pi session history for reconnect. A receipt confirms transport handling, not execution. Reviews stop after all goals are completed or cancelled, and both panes remain available. These mechanics are tested; useful judgment and savings from a cheaper worker still require a representative two-model run. -- Pi/OpenAI
|
||||
|
||||
## Install
|
||||
|
||||
This branch requires Herdr 0.7.5 or newer and one Pi package:
|
||||
This branch requires Herdr 0.7.5 or newer and one Pi package. It reuses installed pi-intercom or loads its pi-intercom dependency when none is registered:
|
||||
|
||||
```bash
|
||||
pi install npm:@wassname2/pi-goals
|
||||
@@ -66,6 +66,6 @@ npm run typecheck
|
||||
npm run lint
|
||||
```
|
||||
|
||||
`test/rpc-review.test.ts` runs the planning review flow through Pi's real RPC protocol with a local deterministic model. The Herdr launcher and visible supervisor bootstrap have focused tests; use a real Herdr session for the final two-pane check.
|
||||
`test/intercom-broker.test.ts` checks readiness and exact message delivery through an isolated real Intercom broker. `test/rpc-review.test.ts` runs the planning review flow through Pi's real RPC protocol with a local deterministic model. The Herdr launcher and visible supervisor bootstrap have focused tests; use a real Herdr session for the final two-pane check.
|
||||
|
||||
-- PI[gpt-5.6-sol]
|
||||
|
||||
Generated
+1729
-154
File diff suppressed because it is too large
Load Diff
@@ -43,6 +43,9 @@
|
||||
"lint": "biome check src/ test/",
|
||||
"lint:fix": "biome check --fix src/ test/"
|
||||
},
|
||||
"dependencies": {
|
||||
"pi-intercom": "^0.13.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@biomejs/biome": "^2.4.8",
|
||||
"@earendil-works/pi-coding-agent": "^0.84.1",
|
||||
|
||||
@@ -5,11 +5,12 @@ User priority: pi-intercom is the session-to-session transport. Keep one pi-goal
|
||||
- [x] goal: supervisor investigates claims and keeps authorized work moving
|
||||
- evidence: commits `9410252` and `386305a`; prompt requires justified confidence, sourced observations, competing explanations, and verification of stopping/completion claims.
|
||||
- limitation: prompt checks do not prove behavioral improvement.
|
||||
- [ ] goal: worker and visible supervisor communicate through pi-intercom
|
||||
- [/] goal: worker and visible supervisor communicate through pi-intercom
|
||||
- Replace mailbox files and polling with the existing Intercom extension channel. No separate RPC transport or mailbox fallback.
|
||||
- Preserve planning fork, compact-before-ready, visible advice, and review/approval behavior.
|
||||
- failure modes: false readiness, duplicate delivery, wrong-session routing, stale callbacks after reload, disconnected peer treated as active.
|
||||
- deliverable: isolated two-session message transcript with exact instructions received, reconnect/reload checks, and saved validation output. Do not operate user panes.
|
||||
- evidence: [37 passing tests, typecheck and lint](../reviews/20260908-intercom-validation.txt). Two client sessions exchange readiness, view, and exact advice through a real isolated Intercom broker. Lifecycle mocks cover resume/deduplication. This is not yet a full live Pi-pair test. Mailbox source and polling were removed.
|
||||
- [ ] goal: supervisor receives a useful bounded worker overview
|
||||
- Borrow latest human direction, source-session path, and incremental progress from `origin/feature/simple-visible-supervision`.
|
||||
- failure modes: lost authorization, repeated summaries, truncated evidence treated as complete.
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
|
||||
> @wassname2/pi-goals@0.2.2 test
|
||||
> vitest run --reporter=verbose
|
||||
|
||||
|
||||
RUN v4.1.9 /home/code/.pi/agent/git/github.com/wassname/pi-goals
|
||||
|
||||
✓ test/package-agent.test.ts > package manifest > includes the extension without registering a packaged subagent 6ms
|
||||
✓ test/herdr.test.ts > supervisor pane command > forks the planning session with pi-goals owning its Intercom dependency 3ms
|
||||
✓ test/herdr.test.ts > supervisor pane command > accepts Herdr's text version output and stale pane cleanup 40ms
|
||||
✓ test/prompts.test.ts > planning prompt > requires fact finding or a focused question before a goal 2ms
|
||||
✓ test/prompts.test.ts > planning prompt > restores the same rule after compaction 0ms
|
||||
✓ test/prompts.test.ts > planning prompt > anchors work and sign-off to the user-visible result 0ms
|
||||
✓ test/intercom.test.ts > pi-intercom transport > receives exact advice once, acknowledges it and rejects unrelated peers 14ms
|
||||
✓ test/intercom.test.ts > pi-intercom transport > restores an unacknowledged steer on reconnect and stops replay after acknowledgment 7ms
|
||||
✓ test/intercom.test.ts > pi-intercom transport > cancels a readiness wait on shutdown 4ms
|
||||
✓ test/fold.test.ts > foldPlan (current goals are above ## Log; durable memory is below it) > keeps the title, user voice and goals 2ms
|
||||
✓ test/fold.test.ts > foldPlan (current goals are above ## Log; durable memory is below it) > drops the log, the learnings and the unlimited appendix 1ms
|
||||
✓ test/fold.test.ts > foldPlan (current goals are above ## Log; durable memory is below it) > returns the whole plan when there is no ## Log yet (a fresh draft) 0ms
|
||||
✓ test/fold.test.ts > openSubtasks (the widget shows the next action, so the plan IS the task list) > lists the active goal's open and in-progress subtasks, stopping at the next goal 1ms
|
||||
✓ test/fold.test.ts > openSubtasks (the widget shows the next action, so the plan IS the task list) > does not leak subtasks from the goal below 1ms
|
||||
✓ test/tick-goal.test.ts > tickGoal (sign-off ticks the goal; agent only ticks on wording drift) > ticks the exact-matching goal line, case-insensitive, leaving subtasks alone 2ms
|
||||
✓ test/tick-goal.test.ts > tickGoal (sign-off ticks the goal; agent only ticks on wording drift) > returns null on wording drift (fuzzy matching is the judge's job, not TypeScript's) 0ms
|
||||
✓ test/tick-goal.test.ts > tickGoal (sign-off ticks the goal; agent only ticks on wording drift) > returns null when the subject matches more than one goal line 0ms
|
||||
stdout | test/intercom-broker.test.ts > exchanges readiness, views and exact advice over a real isolated pi-intercom broker
|
||||
Intercom broker: readiness confirmed; exact worker view and supervisor advice received.
|
||||
|
||||
✓ test/append-log.test.ts > appendLog (the extension's only plan-file write) > creates ## Log at EOF when absent 2ms
|
||||
✓ test/append-log.test.ts > appendLog (the extension's only plan-file write) > appends after the last existing log line, before any following header 1ms
|
||||
✓ test/intercom-broker.test.ts > exchanges readiness, views and exact advice over a real isolated pi-intercom broker 221ms
|
||||
✓ test/goals-flow.test.ts > /goals flow > reports actual idle state, invalidates stopped views on start, and stops completed plans 38ms
|
||||
✓ test/goals-flow.test.ts > /goals flow > preserves drafts, records the interview, and keeps planning read-only 21ms
|
||||
✓ test/goals-flow.test.ts > /goals flow > forks a visible supervisor on Ready and keeps the main session as worker 24ms
|
||||
✓ test/goals-flow.test.ts > /goals flow > starts work only after the supervisor launcher resolves 20ms
|
||||
✓ test/goals-flow.test.ts > /goals flow > delivers an Intercom instruction to the worker 20ms
|
||||
✓ test/goals-flow.test.ts > /goals flow > closes the supervisor on clear but keeps the plan file 24ms
|
||||
✓ test/goals-flow.test.ts > /goals flow > accepts only an approval for the exact clean commit and goal block 49ms
|
||||
✓ test/goals-flow.test.ts > process role > keeps subagent children and visible supervisors out of the worker extension 0ms
|
||||
✓ test/supervisor-session.test.ts > visible supervisor session > restores monitoring and read-only tools without replaying persisted views 6ms
|
||||
✓ test/supervisor-session.test.ts > visible supervisor session > renders all advice in real Pi tool rows, including collapsed and restored rows 25ms
|
||||
✓ test/supervisor-session.test.ts > visible supervisor session > asks for judgment and useful recaps without inventing instructions 1ms
|
||||
✓ test/supervisor-session.test.ts > visible supervisor session > writes readiness only after removing writing tools 2ms
|
||||
✓ test/supervisor-session.test.ts > visible supervisor session > compacts a large planning fork before writing readiness 1ms
|
||||
✓ test/supervisor-session.test.ts > visible supervisor session > does not become ready when initial compaction fails 1ms
|
||||
✓ test/supervisor-session.test.ts > visible supervisor session > writes a durable worker instruction 1ms
|
||||
✓ test/supervisor-session.test.ts > visible supervisor session > records approval only from a stopped view with evidence and no active work 58ms
|
||||
✓ test/rpc-review.test.ts > RPC review flow > opens Refine's editor before it starts the revision turn 1384ms
|
||||
|
||||
Test Files 11 passed (11)
|
||||
Tests 37 passed (37)
|
||||
Start at 16:31:10
|
||||
Duration 1.67s (transform 1.31s, setup 0ms, import 3.59s, tests 2.00s, environment 1ms)
|
||||
|
||||
|
||||
> @wassname2/pi-goals@0.2.2 typecheck
|
||||
> tsc --noEmit
|
||||
|
||||
|
||||
> @wassname2/pi-goals@0.2.2 lint
|
||||
> biome check src/ test/
|
||||
|
||||
Checked 21 files in 67ms. No fixes applied.
|
||||
@@ -65,7 +65,7 @@ The transport rewrite is an implementation choice, not the user's goal.
|
||||
|
||||
## Reproduction
|
||||
|
||||
Run `node_modules/.bin/tsx slop/reviews/20260908_supervisor-intent-reproduction.ts`.
|
||||
The mailbox reproduction script is historical; retrieve it at commit `386305a`. The Intercom migration removes that implementation. Current transport checks are in `test/intercom.test.ts` and `test/intercom-broker.test.ts`.
|
||||
|
||||
[Saved output](20260908_supervisor-intent-reproduction.txt) records the exact observations quoted above. The harness uses only temporary mailbox files and mocked Pi lifecycle events; it neither launches Pi nor contacts another session. It asserts the currently observed failure, not desired behavior.
|
||||
|
||||
|
||||
@@ -1,66 +0,0 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { createMailbox, supervisorReady, writeWorkerView } from "../../src/mailbox.js";
|
||||
import { registerVisibleSupervisor } from "../../src/supervisor-session.js";
|
||||
import { workerView } from "../../src/worker-view.js";
|
||||
|
||||
const cwd = mkdtempSync(join(tmpdir(), "goals-intent-review-"));
|
||||
const plan = join(cwd, "plan.md");
|
||||
const mailbox = createMailbox(cwd, "worker", "approval", plan);
|
||||
Object.assign(process.env, {
|
||||
PI_GOALS_WORKER_ID: "worker", PI_GOALS_OWNER_SESSION_ID: "worker",
|
||||
PI_GOALS_PLAN_PATH: plan, PI_GOALS_APPROVAL_ID: "approval", PI_GOALS_MAILBOX_PATH: mailbox.path,
|
||||
});
|
||||
const entries: any[] = [];
|
||||
function runtime() {
|
||||
const hooks = new Map();
|
||||
const tools = new Map();
|
||||
const messages: string[] = [];
|
||||
let activeTools = ["read", "write", "bash"];
|
||||
const pi = {
|
||||
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 }),
|
||||
sendUserMessage: (text: string) => {
|
||||
messages.push(text);
|
||||
entries.push({ type: "message", message: { role: "user", content: [{ type: "text", text }] } });
|
||||
},
|
||||
getActiveTools: () => activeTools,
|
||||
setActiveTools: (tools: string[]) => { activeTools = tools; },
|
||||
};
|
||||
const ctx = {
|
||||
getContextUsage: () => ({ tokens: 10 }),
|
||||
sessionManager: { getEntries: () => entries },
|
||||
ui: { notify: (message: string) => { throw new Error(message); } },
|
||||
};
|
||||
registerVisibleSupervisor(pi as any);
|
||||
return { hooks, tools, messages, ctx, activeTools: () => activeTools };
|
||||
}
|
||||
try {
|
||||
writeWorkerView(mailbox, "settled", "The worker stopped.\n\nA fresh result awaits review.");
|
||||
for (const name of ["fresh", "resumed"]) {
|
||||
const run = runtime();
|
||||
try {
|
||||
await run.hooks.get("session_start")({}, run.ctx);
|
||||
await new Promise(setImmediate);
|
||||
console.log(`${name}: deliveredViews=${run.messages.length}, activeTools=${run.activeTools().join(",")}, readyReceipt=${supervisorReady(mailbox.path)}`);
|
||||
assert.equal(run.messages.length, name === "fresh" ? 1 : 0);
|
||||
assert.deepEqual(run.activeTools(), ["read"]);
|
||||
if (name === "fresh") {
|
||||
const tool = run.tools.get("SteerWorker");
|
||||
const result = await tool.execute("id", { instruction: "Compare the signs in the two saved outputs." });
|
||||
assert.equal(typeof tool.renderCall, "function");
|
||||
console.log(`steer: renderCall=${typeof tool.renderCall}, result=${result.content[0].text}`);
|
||||
}
|
||||
} finally {
|
||||
await run.hooks.get("session_shutdown")();
|
||||
}
|
||||
}
|
||||
const idleView = workerView([], "interval", true).split("\n")[0];
|
||||
assert.equal(idleView, "The worker stopped.");
|
||||
console.log(`interval view without any work: ${idleView}`);
|
||||
} finally {
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
+3
-19
@@ -1,9 +1,7 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { supervisorReady } from "./mailbox.js";
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
const STARTUP_TIMEOUT_MS = 5 * 60_000;
|
||||
|
||||
interface LaunchSupervisorInput {
|
||||
cwd: string;
|
||||
@@ -11,7 +9,6 @@ interface LaunchSupervisorInput {
|
||||
workerSessionId: string;
|
||||
planPath: string;
|
||||
approvalId: string;
|
||||
mailboxPath: string;
|
||||
extensionPath: string;
|
||||
model: string | null;
|
||||
}
|
||||
@@ -53,7 +50,6 @@ export function supervisorCommand(input: LaunchSupervisorInput): string {
|
||||
`PI_GOALS_PLAN_PATH=${input.planPath}`,
|
||||
`PI_GOALS_APPROVAL_ID=${input.approvalId}`,
|
||||
`PI_GOALS_OWNER_SESSION_ID=${input.workerSessionId}`,
|
||||
`PI_GOALS_MAILBOX_PATH=${input.mailboxPath}`,
|
||||
];
|
||||
const args = [
|
||||
"pi",
|
||||
@@ -66,27 +62,15 @@ export function supervisorCommand(input: LaunchSupervisorInput): string {
|
||||
return `env ${[...env, ...args].map(shellQuote).join(" ")}`;
|
||||
}
|
||||
|
||||
export async function waitForSupervisorReady(mailboxPath: string, timeoutMs = STARTUP_TIMEOUT_MS): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (!supervisorReady(mailboxPath)) {
|
||||
if (Date.now() >= deadline) throw new Error("The visible supervisor did not become ready.");
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
}
|
||||
}
|
||||
|
||||
export async function openSupervisorPane(input: LaunchSupervisorInput): Promise<string> {
|
||||
export async function openSupervisorPane(input: LaunchSupervisorInput, onOpened: (paneId: string) => void): Promise<string> {
|
||||
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"], 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.");
|
||||
onOpened(paneId);
|
||||
await herdr(["pane", "run", paneId, supervisorCommand(input)]);
|
||||
try {
|
||||
await waitForSupervisorReady(input.mailboxPath);
|
||||
return paneId;
|
||||
} catch (error) {
|
||||
throw new Error(`Supervisor startup incomplete in Herdr pane ${paneId}; inspect that pane. ${error instanceof Error ? error.message : String(error)}`);
|
||||
}
|
||||
return paneId;
|
||||
}
|
||||
|
||||
export async function closeSupervisorPane(paneId: string): Promise<void> {
|
||||
|
||||
+29
-38
@@ -1,6 +1,6 @@
|
||||
/**
|
||||
* PI: pi-goals owns one versioned plan per session. After Ready, the main session implements the
|
||||
* plan while a compacted, visible fork supervises it through a durable mailbox.
|
||||
* plan while a compacted, visible fork supervises it through pi-intercom.
|
||||
*
|
||||
* Each /goals call makes `.pi/plan/<session_id>-vN.md`. The selected version survives resume and
|
||||
* compaction. Old plans stay on disk but inactive. A session with no selected plan has no widget,
|
||||
@@ -22,7 +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 { createMailbox, workerSteersAfter, writeWorkerView } from "./mailbox.js";
|
||||
import { GoalIntercom } from "./intercom.js";
|
||||
import { completeGoalDescription, completeGoalParamDescription, planDrafting, planningState, resync } from "./prompts.js";
|
||||
import { isVisibleSupervisor, registerVisibleSupervisor } from "./supervisor-session.js";
|
||||
import { workerView } from "./worker-view.js";
|
||||
@@ -96,8 +96,6 @@ interface PlanState {
|
||||
supervisorModel: string | null;
|
||||
supervisorPaneId: string | null;
|
||||
approvalId: string | null;
|
||||
mailboxPath: string | null;
|
||||
lastSteer: number;
|
||||
planVersion: number | null;
|
||||
}
|
||||
|
||||
@@ -107,13 +105,16 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
return;
|
||||
}
|
||||
if (!isMainSession()) return;
|
||||
const intercom = new GoalIntercom(pi);
|
||||
intercom.onSteer = (instruction) => {
|
||||
if (state.phase !== "working") throw new Error("Worker plan is not active; instruction rejected.");
|
||||
pi.sendUserMessage(`[supervisor] ${instruction}`, { deliverAs: "steer" });
|
||||
};
|
||||
let state: PlanState = {
|
||||
phase: null,
|
||||
supervisorModel: null,
|
||||
supervisorPaneId: null,
|
||||
approvalId: null,
|
||||
mailboxPath: null,
|
||||
lastSteer: 0,
|
||||
planVersion: null,
|
||||
};
|
||||
let planningContextPending = false;
|
||||
@@ -141,8 +142,8 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
rmSync(approvalPath(ctx.cwd, ctx.sessionManager.getSessionId(), goal.subject), { force: true });
|
||||
}
|
||||
const approvalId = randomUUID();
|
||||
const mailbox = createMailbox(ctx.cwd, ctx.sessionManager.getSessionId(), approvalId, planPath(ctx));
|
||||
state = { ...state, approvalId, mailboxPath: mailbox.path, lastSteer: 0 };
|
||||
state = { ...state, approvalId };
|
||||
intercom.configure(approvalId, "worker", ctx);
|
||||
persist();
|
||||
}
|
||||
|
||||
@@ -154,6 +155,11 @@ 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.");
|
||||
if (state.supervisorPaneId && state.approvalId) {
|
||||
intercom.configure(state.approvalId, "worker", ctx);
|
||||
await intercom.waitReady();
|
||||
return;
|
||||
}
|
||||
beginReview(ctx);
|
||||
let paneId: string | null = null;
|
||||
try {
|
||||
@@ -163,9 +169,11 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
workerSessionId: ctx.sessionManager.getSessionId(),
|
||||
planPath: planPath(ctx),
|
||||
approvalId: state.approvalId!,
|
||||
mailboxPath: state.mailboxPath!,
|
||||
extensionPath: fileURLToPath(import.meta.url),
|
||||
model: state.supervisorModel,
|
||||
}, (opened) => {
|
||||
state = { ...state, supervisorPaneId: opened };
|
||||
persist();
|
||||
});
|
||||
} catch (error) {
|
||||
if (paneId) throw new Error(`Supervisor startup failed in Herdr pane ${paneId}; it remains open for inspection. ${error instanceof Error ? error.message : String(error)}`);
|
||||
@@ -173,20 +181,15 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
}
|
||||
state = { ...state, supervisorPaneId: paneId };
|
||||
persist();
|
||||
await intercom.waitReady();
|
||||
}
|
||||
|
||||
let workerTurns = 0;
|
||||
let viewTimer: ReturnType<typeof setInterval> | undefined;
|
||||
let steerTimer: ReturnType<typeof setInterval> | undefined;
|
||||
|
||||
function mailbox(ctx: ExtensionContext) {
|
||||
if (!state.approvalId || !state.mailboxPath) throw new Error("No active supervisor mailbox.");
|
||||
return createMailbox(ctx.cwd, ctx.sessionManager.getSessionId(), state.approvalId, planPath(ctx));
|
||||
}
|
||||
|
||||
function publishWorkerView(ctx: ExtensionContext, reason: "ready" | "settled" | "turns" | "interval" | "started"): void {
|
||||
if (state.phase !== "working") return;
|
||||
writeWorkerView(mailbox(ctx), reason, workerView(ctx.sessionManager.getBranch(), reason, reason !== "started" && ctx.isIdle()));
|
||||
intercom.view(workerView(ctx.sessionManager.getBranch(), reason, reason !== "started" && ctx.isIdle()), reason);
|
||||
const goals = scanGoals(readPlan(ctx));
|
||||
if (goals.length > 0 && goals.every((goal) => goal.status === "done" || goal.status === "cancelled")) {
|
||||
stopWorkerTimers();
|
||||
@@ -195,25 +198,13 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
}
|
||||
}
|
||||
|
||||
function deliverWorkerSteers(ctx: ExtensionContext): void {
|
||||
if (state.phase !== "working") return;
|
||||
for (const steer of workerSteersAfter(mailbox(ctx).path, state.lastSteer)) {
|
||||
state = { ...state, lastSteer: steer.sequence };
|
||||
persist();
|
||||
pi.sendUserMessage(`[supervisor] ${steer.instruction}`, { deliverAs: "followUp" });
|
||||
}
|
||||
}
|
||||
|
||||
function startWorkerTimers(ctx: ExtensionContext): void {
|
||||
if (!viewTimer) viewTimer = setInterval(() => publishWorkerView(ctx, "interval"), 60 * 60_000);
|
||||
if (!steerTimer) steerTimer = setInterval(() => deliverWorkerSteers(ctx), 1_000);
|
||||
}
|
||||
|
||||
function stopWorkerTimers(): void {
|
||||
if (viewTimer) clearInterval(viewTimer);
|
||||
if (steerTimer) clearInterval(steerTimer);
|
||||
viewTimer = undefined;
|
||||
steerTimer = undefined;
|
||||
}
|
||||
|
||||
async function stopSupervisor(): Promise<boolean> {
|
||||
@@ -274,7 +265,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
ctx.ui.notify("Could not close the visible supervisor; the plan remains connected.", "warning");
|
||||
return;
|
||||
}
|
||||
state = { ...state, phase: null, supervisorPaneId: null, approvalId: null, mailboxPath: null, lastSteer: 0, planVersion: null };
|
||||
state = { ...state, phase: null, supervisorPaneId: null, approvalId: null, planVersion: null };
|
||||
persist();
|
||||
updateWidget(ctx);
|
||||
ctx.ui.notify(`Disconnected from ${currentPlan}; the file remains on disk.`, "info");
|
||||
@@ -290,7 +281,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
return;
|
||||
}
|
||||
const ref = arg.slice("model".length).trim();
|
||||
state = { ...state, supervisorModel: ref || null, supervisorPaneId: null, approvalId: null, mailboxPath: null, lastSteer: 0 };
|
||||
state = { ...state, supervisorModel: ref || null, supervisorPaneId: null, approvalId: null };
|
||||
persist();
|
||||
ctx.ui.notify(`Goal-supervisor model ${ref ? `set to ${ref}` : "reset to the current Pi default"}.`, "info");
|
||||
return;
|
||||
@@ -299,7 +290,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
ctx.ui.notify("Could not close the visible supervisor; no new plan was started.", "warning");
|
||||
return;
|
||||
}
|
||||
state = { ...state, phase: "planning", supervisorPaneId: null, approvalId: null, mailboxPath: null, lastSteer: 0, planVersion: nextVersion(ctx) };
|
||||
state = { ...state, phase: "planning", supervisorPaneId: null, approvalId: null, planVersion: nextVersion(ctx) };
|
||||
planningContextPending = true;
|
||||
resyncReason = null;
|
||||
writePlan(ctx, "");
|
||||
@@ -329,7 +320,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
pi.on("before_agent_start", async (_event, ctx) => {
|
||||
if (state.phase === "working") {
|
||||
return {
|
||||
systemPrompt: `${ctx.getSystemPrompt()}\n\nYou are the implementation worker for ${planRel(ctx)}. Keep the full conversation and do the work directly. A stronger read-only supervisor watches this session through its durable mailbox and can steer you. Commit clean evidence before asking for sign-off. Stop when a goal appears complete so the supervisor can inspect a settled worker view. Call CompleteGoal only after the supervisor says it recorded approval. -- PI[Kimi K3]`,
|
||||
systemPrompt: `${ctx.getSystemPrompt()}\n\nYou are the implementation worker for ${planRel(ctx)}. Keep the full conversation and do the work directly. A stronger read-only supervisor watches this session through pi-intercom and can steer you. Commit clean evidence before asking for sign-off. Stop when a goal appears complete so the supervisor can inspect a settled worker view. Call CompleteGoal only after the supervisor says it recorded approval. -- PI[Kimi K3]`,
|
||||
};
|
||||
}
|
||||
if (!planningContextPending) return;
|
||||
@@ -392,7 +383,6 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
// PI: Print after Pi settles. agent_end is still streaming, so its message queues behind the menu.
|
||||
pi.on("agent_settled", async (_event, ctx) => {
|
||||
if (state.phase === "working") {
|
||||
deliverWorkerSteers(ctx);
|
||||
publishWorkerView(ctx, "settled");
|
||||
return;
|
||||
}
|
||||
@@ -424,7 +414,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
}
|
||||
if (choice === "Cancel") {
|
||||
rmSync(planPath(ctx), { force: true });
|
||||
state = { ...state, phase: null, supervisorPaneId: null, approvalId: null, mailboxPath: null, lastSteer: 0, planVersion: null };
|
||||
state = { ...state, phase: null, supervisorPaneId: null, approvalId: null, planVersion: null };
|
||||
persist();
|
||||
updateWidget(ctx);
|
||||
ctx.ui.notify("Plan discarded.", "info");
|
||||
@@ -443,7 +433,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
pi.sendUserMessage("The plan is approved. Begin implementation as the worker.");
|
||||
} catch (error) {
|
||||
ctx.ui.notify(`Goal supervisor could not start: ${error instanceof Error ? error.message : String(error)}`, "warning");
|
||||
state = { ...state, phase: "planning", supervisorPaneId: null, approvalId: null, mailboxPath: null, lastSteer: 0 };
|
||||
state = { ...state, phase: "planning" };
|
||||
persist();
|
||||
updateWidget(ctx);
|
||||
}
|
||||
@@ -461,13 +451,14 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
|
||||
supervisorModel: last?.data?.supervisorModel ?? null,
|
||||
supervisorPaneId: last?.data?.supervisorPaneId ?? null,
|
||||
approvalId: last?.data?.approvalId ?? null,
|
||||
mailboxPath: last?.data?.mailboxPath ?? null,
|
||||
lastSteer: last?.data?.lastSteer ?? 0,
|
||||
planVersion: last?.data?.planVersion ?? null,
|
||||
};
|
||||
planningContextPending = state.phase === "planning";
|
||||
resyncReason = state.phase === "working" ? "New session." : null;
|
||||
if (state.phase === "working") startWorkerTimers(ctx);
|
||||
if (state.phase === "working") {
|
||||
intercom.configure(state.approvalId!, "worker", ctx);
|
||||
startWorkerTimers(ctx);
|
||||
}
|
||||
updateWidget(ctx);
|
||||
});
|
||||
|
||||
|
||||
Vendored
+4
@@ -0,0 +1,4 @@
|
||||
declare module "pi-intercom" {
|
||||
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
|
||||
export default function intercom(pi: ExtensionAPI): void;
|
||||
}
|
||||
+187
@@ -0,0 +1,187 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { ExtensionAPI, ExtensionContext, SessionStartEvent } from "@earendil-works/pi-coding-agent";
|
||||
import type { IntercomExtensionChannel, IntercomExtensionEvent } from "pi-intercom/extension-api.ts";
|
||||
|
||||
export type Role = "worker" | "supervisor";
|
||||
export interface View { id: string; text: string; reason: string }
|
||||
interface Message { binding: string; role: Role; kind: "hello" | "view" | "steer" | "received"; id: string; text?: string; reason?: string; ready?: boolean }
|
||||
const STATE = "pi-goals-intercom";
|
||||
|
||||
export class GoalIntercom {
|
||||
private channel?: IntercomExtensionChannel;
|
||||
private ctx?: ExtensionContext;
|
||||
private stopped = false;
|
||||
private registered = false;
|
||||
private binding = "";
|
||||
private role: Role = "worker";
|
||||
private ready = false;
|
||||
private peer?: string;
|
||||
private peerReady = false;
|
||||
private pending = new Map<string, Message>();
|
||||
private received = new Set<string>();
|
||||
private waiters = new Set<() => void>();
|
||||
latestView?: View;
|
||||
onView: (view: View) => void = () => {};
|
||||
onSteer: (text: string) => void = () => {};
|
||||
|
||||
constructor(private pi: ExtensionAPI) {
|
||||
pi.events.on("intercom:extension-registry-ready", () => this.register());
|
||||
this.register();
|
||||
pi.on("session_start", async (event, ctx) => {
|
||||
this.ctx = ctx;
|
||||
if (!this.channel) await this.loadIntercom(event, ctx);
|
||||
});
|
||||
pi.on("session_shutdown", async () => {
|
||||
this.stopped = true;
|
||||
this.peerReady = false;
|
||||
for (const wake of this.waiters) wake();
|
||||
});
|
||||
}
|
||||
|
||||
configure(binding: string, role: Role, ctx: ExtensionContext): void {
|
||||
this.binding = binding;
|
||||
this.role = role;
|
||||
this.ctx = ctx;
|
||||
this.ready = role === "worker";
|
||||
this.peer = undefined;
|
||||
this.peerReady = false;
|
||||
this.pending.clear();
|
||||
this.received.clear();
|
||||
this.latestView = undefined;
|
||||
for (const entry of ctx.sessionManager.getEntries()) {
|
||||
if (entry.type !== "custom" || entry.customType !== STATE) continue;
|
||||
const record = entry.data as { direction: string; message: Message };
|
||||
const message = record.message;
|
||||
if (message.binding !== binding) continue;
|
||||
if (record.direction === "out" && message.kind === "steer") this.pending.set(message.id, message);
|
||||
if (record.direction === "ack") this.pending.delete(message.id);
|
||||
if (record.direction === "in") this.received.add(message.id);
|
||||
if (message.kind === "view") this.latestView = { id: message.id, text: message.text!, reason: message.reason! };
|
||||
}
|
||||
this.hello();
|
||||
}
|
||||
|
||||
markReady(): void { if (!this.stopped) { this.ready = true; this.hello(); } }
|
||||
get ended(): boolean { return this.stopped; }
|
||||
get connected(): boolean { return Boolean(this.channel?.snapshot().connected && this.peerReady); }
|
||||
|
||||
async waitReady(timeoutMs = 300_000): Promise<void> {
|
||||
if (this.connected) return;
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const finish = () => {
|
||||
if (!this.connected && !this.stopped) return;
|
||||
clearTimeout(timer); this.waiters.delete(finish);
|
||||
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);
|
||||
this.waiters.add(finish);
|
||||
finish();
|
||||
});
|
||||
}
|
||||
|
||||
view(text: string, reason: string): View {
|
||||
const id = randomUUID();
|
||||
const message: Message = { binding: this.binding, role: this.role, kind: "view", id, text: `${text}\n\nworker view id: ${id}`, reason };
|
||||
this.record("out", message);
|
||||
this.latestView = { id, text: message.text!, reason };
|
||||
if (this.connected) this.publish(message);
|
||||
return this.latestView;
|
||||
}
|
||||
|
||||
steer(text: string): string {
|
||||
if (!this.connected) throw new Error("Worker is disconnected; no instruction was sent.");
|
||||
const message: Message = { binding: this.binding, role: this.role, kind: "steer", id: randomUUID(), text };
|
||||
this.record("out", message);
|
||||
this.pending.set(message.id, message);
|
||||
this.publish(message);
|
||||
return message.id;
|
||||
}
|
||||
|
||||
private record(direction: string, message: Message): void { this.pi.appendEntry(STATE, { direction, message }); }
|
||||
private publish(message: Message): void {
|
||||
if (this.stopped) throw new Error("Intercom session ended.");
|
||||
if (Buffer.byteLength(JSON.stringify(message)) > 16_000) throw new Error("Supervisor message exceeds the Intercom payload limit.");
|
||||
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 receive(event: IntercomExtensionEvent): void {
|
||||
if (this.stopped) return;
|
||||
if (event.type === "connection") {
|
||||
if (!event.connected) {
|
||||
if (this.peerReady) this.ctx?.ui.notify("Goal supervision disconnected from pi-intercom.", "warning");
|
||||
this.peer = undefined; this.peerReady = false;
|
||||
}
|
||||
else this.hello();
|
||||
return;
|
||||
}
|
||||
if (event.type === "session_left" && event.sessionId === this.peer) {
|
||||
this.peer = undefined; this.peerReady = false;
|
||||
this.ctx?.ui.notify("Goal supervision peer disconnected; reconnect the existing session.", "warning");
|
||||
return;
|
||||
}
|
||||
if (event.type === "session_joined") { this.hello(); return; }
|
||||
if (event.type !== "message") return;
|
||||
const message = event.payload as Message;
|
||||
if (!message || message.binding !== this.binding || message.role !== (this.role === "worker" ? "supervisor" : "worker")) return;
|
||||
if (message.kind === "hello") {
|
||||
if (this.peer && this.peer !== event.fromSessionId) throw new Error("Two peers claim this supervision binding. Stop the duplicate session.");
|
||||
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) {
|
||||
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);
|
||||
}
|
||||
}
|
||||
for (const wake of this.waiters) wake();
|
||||
return;
|
||||
}
|
||||
if (event.fromSessionId !== this.peer || !this.ready) return;
|
||||
if (message.kind === "received") { this.pending.delete(message.id); this.record("ack", message); return; }
|
||||
if (this.received.has(message.id)) {
|
||||
if (message.kind === "steer") this.publish({ ...message, role: this.role, kind: "received" });
|
||||
return;
|
||||
}
|
||||
if (message.kind === "view" && this.role === "supervisor") {
|
||||
this.latestView = { id: message.id, text: message.text!, reason: message.reason! };
|
||||
if (message.reason !== "started") this.onView(this.latestView);
|
||||
} else if (message.kind === "steer" && this.role === "worker") {
|
||||
this.onSteer(message.text!);
|
||||
this.received.add(message.id);
|
||||
this.record("in", message);
|
||||
this.publish({ binding: this.binding, role: this.role, kind: "received", id: message.id });
|
||||
return;
|
||||
} else return;
|
||||
this.received.add(message.id);
|
||||
this.record("in", message);
|
||||
}
|
||||
private register(): void {
|
||||
if (this.stopped || this.registered) return;
|
||||
this.pi.events.emit("intercom:extension-register", {
|
||||
namespace: "pi-goals", ownerEligible: false,
|
||||
onReady: (channel: IntercomExtensionChannel) => { if (this.stopped) return; this.registered = true; this.channel = channel; this.hello(); },
|
||||
onEvent: (event: IntercomExtensionEvent) => {
|
||||
try { this.receive(event); }
|
||||
catch (error) { if (!this.stopped) this.ctx?.ui.notify(`Goal Intercom error: ${String(error)}`, "error"); }
|
||||
},
|
||||
});
|
||||
}
|
||||
private async loadIntercom(event: SessionStartEvent, ctx: ExtensionContext): Promise<void> {
|
||||
if (this.pi.getAllTools().some(tool => tool.name === "intercom")) throw new Error("Installed pi-intercom has no extension channel; update it before continuing.");
|
||||
const starts: Array<(event: SessionStartEvent, ctx: ExtensionContext) => unknown> = [];
|
||||
const api = { ...this.pi, on: (name: string, handler: (...args: any[]) => any) => {
|
||||
if (name === "session_start") starts.push(handler);
|
||||
else this.pi.on(name as Parameters<ExtensionAPI["on"]>[0], handler);
|
||||
} } as ExtensionAPI;
|
||||
const { default: intercom } = await import("pi-intercom");
|
||||
intercom(api);
|
||||
for (const start of starts) await start(event, ctx);
|
||||
this.register();
|
||||
}
|
||||
}
|
||||
-107
@@ -1,107 +0,0 @@
|
||||
import { mkdirSync, readdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
|
||||
import { join, resolve } from "node:path";
|
||||
|
||||
const VERSION = 1;
|
||||
const VIEWS = "views";
|
||||
const STEERS = "steers";
|
||||
|
||||
export interface SupervisorMailbox {
|
||||
path: string;
|
||||
workerSessionId: string;
|
||||
planPath: string;
|
||||
approvalId: string;
|
||||
}
|
||||
|
||||
export interface WorkerView {
|
||||
version: 1;
|
||||
sequence: number;
|
||||
reason: "ready" | "settled" | "turns" | "interval" | "started";
|
||||
text: string;
|
||||
timestamp: string;
|
||||
}
|
||||
|
||||
export interface WorkerSteer {
|
||||
version: 1;
|
||||
sequence: number;
|
||||
instruction: string;
|
||||
timestamp: string;
|
||||
}
|
||||
|
||||
function writeJson(path: string, value: object): void {
|
||||
const temporary = `${path}.${process.pid}.tmp`;
|
||||
writeFileSync(temporary, `${JSON.stringify(value)}\n`);
|
||||
renameSync(temporary, path);
|
||||
}
|
||||
|
||||
function readJson<T>(path: string): T | null {
|
||||
try {
|
||||
return JSON.parse(readFileSync(path, "utf8")) as T;
|
||||
} catch (error: unknown) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
function sequence(name: string, prefix: string): number | null {
|
||||
const match = new RegExp(`^${prefix}-(\\d+)\\.json$`).exec(name);
|
||||
return match ? Number(match[1]) : null;
|
||||
}
|
||||
|
||||
function nextSequence(directory: string, prefix: string): number {
|
||||
return Math.max(0, ...readdirSync(directory).flatMap((name) => {
|
||||
const value = sequence(name, prefix);
|
||||
return value === null ? [] : [value];
|
||||
})) + 1;
|
||||
}
|
||||
|
||||
export function createMailbox(cwd: string, workerSessionId: string, approvalId: string, planPath: string): SupervisorMailbox {
|
||||
const path = resolve(cwd, ".pi", "goals-supervision", workerSessionId, approvalId);
|
||||
mkdirSync(join(path, VIEWS), { recursive: true });
|
||||
mkdirSync(join(path, STEERS), { recursive: true });
|
||||
return { path, workerSessionId, approvalId, planPath };
|
||||
}
|
||||
|
||||
export function readyMailbox(path: string): void {
|
||||
writeJson(join(path, "ready.json"), { version: VERSION, timestamp: new Date().toISOString() });
|
||||
}
|
||||
|
||||
export function supervisorReady(path: string): boolean {
|
||||
return readJson<{ version?: number }>(join(path, "ready.json"))?.version === VERSION;
|
||||
}
|
||||
|
||||
export function writeWorkerView(mailbox: SupervisorMailbox, reason: WorkerView["reason"], text: string): WorkerView {
|
||||
const directory = join(mailbox.path, VIEWS);
|
||||
const sequence = nextSequence(directory, "view");
|
||||
const view: WorkerView = { version: 1, sequence, reason, text: `${text}\n\nworker view sequence: ${sequence}`, timestamp: new Date().toISOString() };
|
||||
writeJson(join(directory, `view-${view.sequence}.json`), view);
|
||||
return view;
|
||||
}
|
||||
|
||||
export function workerViewsAfter(path: string, after: number): WorkerView[] {
|
||||
const directory = join(path, VIEWS);
|
||||
return readdirSync(directory)
|
||||
.flatMap((name) => {
|
||||
const value = sequence(name, "view");
|
||||
return value !== null && value > after ? [readJson<WorkerView>(join(directory, name))] : [];
|
||||
})
|
||||
.filter((view): view is WorkerView => view !== null)
|
||||
.sort((a, b) => a.sequence - b.sequence);
|
||||
}
|
||||
|
||||
export function writeWorkerSteer(path: string, instruction: string): WorkerSteer {
|
||||
const directory = join(path, STEERS);
|
||||
const steer: WorkerSteer = { version: 1, sequence: nextSequence(directory, "steer"), instruction, timestamp: new Date().toISOString() };
|
||||
writeJson(join(directory, `steer-${steer.sequence}.json`), steer);
|
||||
return steer;
|
||||
}
|
||||
|
||||
export function workerSteersAfter(path: string, after: number): WorkerSteer[] {
|
||||
const directory = join(path, STEERS);
|
||||
return readdirSync(directory)
|
||||
.flatMap((name) => {
|
||||
const value = sequence(name, "steer");
|
||||
return value !== null && value > after ? [readJson<WorkerSteer>(join(directory, name))] : [];
|
||||
})
|
||||
.filter((steer): steer is WorkerSteer => steer !== null)
|
||||
.sort((a, b) => a.sequence - b.sequence);
|
||||
}
|
||||
+2
-2
@@ -3,7 +3,7 @@
|
||||
*
|
||||
* Design: the plan file is for LLMs and the human, not for TypeScript. No parser and no schema;
|
||||
* the skeleton below is a convention the drafting prompt teaches. The main session implements it,
|
||||
* while a visible forked Pi session supervises through a durable mailbox.
|
||||
* while a visible forked Pi session supervises through pi-intercom.
|
||||
*
|
||||
* THE FOLD: everything above "## Log" is the short current-goal section. Everything below it
|
||||
* (Log, Learnings, Appendix) is durable memory: unlimited, read on demand, and sent in full at
|
||||
@@ -164,7 +164,7 @@ export function resync(plan: string, planRel: string, why: string): string {
|
||||
<system-reminder>
|
||||
${why} This is the whole plan file (${planRel}), appendix included. You are the implementation worker.
|
||||
Keep the high-level goal and human intent stable and do the work directly. A visible read-only Pi
|
||||
session supervises you through a durable mailbox. The human's latest message outranks the plan: if it
|
||||
session supervises you through pi-intercom. The human's latest message outranks the plan: if it
|
||||
changes scope, amend the plan rather than preserving an obsolete decision.
|
||||
|
||||
${plan}
|
||||
|
||||
+16
-37
@@ -1,10 +1,10 @@
|
||||
import { readFileSync, rmSync } from "node:fs";
|
||||
import { join, resolve } from "node:path";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { resolve } from "node:path";
|
||||
import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
|
||||
import { Text } from "@earendil-works/pi-tui";
|
||||
import { Type } from "typebox";
|
||||
import { approvalPath, goalBlock, hashGoalBlock, repositoryState, verifyOutputPath, writeApproval } from "./approval.js";
|
||||
import { readyMailbox, workerViewsAfter, writeWorkerSteer } from "./mailbox.js";
|
||||
import { GoalIntercom } from "./intercom.js";
|
||||
|
||||
const BOOTSTRAPPED = "pi-goals-visible-supervisor-v2";
|
||||
const INITIAL_COMPACT_AT_TOKENS = 20_000;
|
||||
@@ -16,7 +16,6 @@ interface SupervisorConfig {
|
||||
ownerSessionId: string;
|
||||
planPath: string;
|
||||
approvalId: string;
|
||||
mailboxPath: string;
|
||||
}
|
||||
|
||||
function result(text: string, isError = false) {
|
||||
@@ -35,7 +34,6 @@ function config(): SupervisorConfig {
|
||||
ownerSessionId: requiredEnv("PI_GOALS_OWNER_SESSION_ID"),
|
||||
planPath: resolve(requiredEnv("PI_GOALS_PLAN_PATH")),
|
||||
approvalId: requiredEnv("PI_GOALS_APPROVAL_ID"),
|
||||
mailboxPath: resolve(requiredEnv("PI_GOALS_MAILBOX_PATH")),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -94,19 +92,11 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void {
|
||||
const settings = config();
|
||||
let compacting = false;
|
||||
let bootstrapping = false;
|
||||
let deliveredView = 0;
|
||||
let viewTimer: ReturnType<typeof setInterval> | undefined;
|
||||
|
||||
const deliverWorkerViews = (): void => {
|
||||
const view = workerViewsAfter(settings.mailboxPath, deliveredView).at(-1);
|
||||
if (view) {
|
||||
if (view.reason !== "started") pi.sendUserMessage(view.text, { deliverAs: "followUp" });
|
||||
deliveredView = view.sequence;
|
||||
}
|
||||
};
|
||||
const intercom = new GoalIntercom(pi);
|
||||
intercom.onView = (view) => pi.sendUserMessage(view.text, { deliverAs: "followUp" });
|
||||
|
||||
const bootstrap = async (ctx: ExtensionContext): Promise<void> => {
|
||||
if (bootstrapping) return;
|
||||
if (bootstrapping || intercom.ended) return;
|
||||
const entries = ctx.sessionManager.getEntries();
|
||||
bootstrapping = true;
|
||||
try {
|
||||
@@ -117,18 +107,7 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void {
|
||||
if (!entries.some((entry: { type?: string; customType?: string }) => entry.type === "custom" && entry.customType === BOOTSTRAPPED)) {
|
||||
pi.appendEntry(BOOTSTRAPPED, { version: 2, workerSessionId: settings.workerSessionId, planPath: settings.planPath });
|
||||
}
|
||||
for (const entry of entries) {
|
||||
const message = (entry as { message?: { role?: string; content?: unknown } }).message;
|
||||
if (message?.role !== "user" || !Array.isArray(message.content)) continue;
|
||||
for (const part of message.content) {
|
||||
if (part.type !== "text" || !part.text.startsWith("The worker ")) continue;
|
||||
const sequence = /^worker view sequence: (\d+)$/m.exec(part.text);
|
||||
if (sequence) deliveredView = Math.max(deliveredView, Number(sequence[1]));
|
||||
}
|
||||
}
|
||||
readyMailbox(settings.mailboxPath);
|
||||
viewTimer = setInterval(deliverWorkerViews, 1_000);
|
||||
deliverWorkerViews();
|
||||
intercom.markReady();
|
||||
} catch (error) {
|
||||
ctx.ui.notify(`Supervisor startup failed: ${error instanceof Error ? error.message : String(error)}`, "error");
|
||||
}
|
||||
@@ -146,25 +125,23 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void {
|
||||
customInstructions: `Preserve the user's high-level intent, decisions, unresolved risks, and the supervisor's remit. The canonical plan is ${settings.planPath}; it remains available directly and must not be replaced by this summary.`,
|
||||
onComplete: () => {
|
||||
compacting = false;
|
||||
if (intercom.ended) return;
|
||||
ctx.ui.notify("Supervisor planning context compacted before work started.", "info");
|
||||
void bootstrap(ctx);
|
||||
},
|
||||
onError: (error) => {
|
||||
compacting = false;
|
||||
if (intercom.ended) return;
|
||||
ctx.ui.notify(`Supervisor startup compaction failed: ${error.message}`, "error");
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
pi.on("session_start", async (_event, ctx) => {
|
||||
rmSync(join(settings.mailboxPath, "ready.json"), { force: true });
|
||||
intercom.configure(settings.approvalId, "supervisor", ctx);
|
||||
pi.setActiveTools(pi.getActiveTools().filter((tool) => !WRITER_TOOLS.has(tool.toLowerCase())));
|
||||
setImmediate(() => { bootstrapAfterInitialCompaction(ctx); });
|
||||
});
|
||||
pi.on("session_shutdown", async () => {
|
||||
if (viewTimer) clearInterval(viewTimer);
|
||||
rmSync(join(settings.mailboxPath, "ready.json"), { force: true });
|
||||
});
|
||||
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;
|
||||
@@ -173,10 +150,12 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void {
|
||||
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}.`,
|
||||
onComplete: () => {
|
||||
compacting = false;
|
||||
if (intercom.ended) return;
|
||||
ctx.ui.notify("Supervisor context compacted at 100k tokens.", "info");
|
||||
},
|
||||
onError: (error) => {
|
||||
compacting = false;
|
||||
if (intercom.ended) return;
|
||||
ctx.ui.notify(`Supervisor compaction failed: ${error.message}`, "error");
|
||||
},
|
||||
});
|
||||
@@ -194,8 +173,8 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void {
|
||||
async execute(_id, params) {
|
||||
const instruction = params.instruction.trim();
|
||||
if (!instruction) return result("A worker instruction cannot be empty.", true);
|
||||
const steer = writeWorkerSteer(settings.mailboxPath, instruction);
|
||||
return result(`Worker instruction ${steer.sequence} recorded. Worker receipt and execution are not confirmed.`);
|
||||
const id = intercom.steer(instruction);
|
||||
return result(`Worker instruction ${id} sent through pi-intercom. Receipt and execution are not confirmed by this result.`);
|
||||
},
|
||||
});
|
||||
|
||||
@@ -210,8 +189,8 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void {
|
||||
}),
|
||||
async execute(_id, params, _signal, _onUpdate, ctx) {
|
||||
const view = latestWorkerView(ctx);
|
||||
const newest = workerViewsAfter(settings.mailboxPath, 0).at(-1);
|
||||
if (!newest || view !== newest.text) return result("Cannot approve without inspecting the latest worker view.", true);
|
||||
const newest = intercom.latestView;
|
||||
if (!intercom.connected || !newest || view !== newest.text) return result("Cannot approve without inspecting the latest worker view.", true);
|
||||
if (!view?.startsWith("The worker stopped.")) return result("Cannot approve without a current stopped-worker view.", true);
|
||||
const pendingTool = view.match(/^tool calls with no result: (?!none$)(.+)$/m);
|
||||
const pendingChild = view.match(/^child pi processes still running: (?!none$)(.+)$/m);
|
||||
|
||||
+19
-15
@@ -5,7 +5,7 @@ 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 { workerViewsAfter, writeWorkerSteer } from "../src/mailbox.js";
|
||||
import { intercomFixture } from "./intercom-fixture.js";
|
||||
|
||||
const openSupervisorPane = vi.fn(async () => "pane-2");
|
||||
const closeSupervisorPane = vi.fn(async () => undefined);
|
||||
@@ -19,6 +19,7 @@ function setup(selectChoices: Array<string | undefined>, editorChoices: Array<st
|
||||
execFileSync("git", ["init", "-q"], { cwd });
|
||||
execFileSync("git", ["add", ".gitignore", "verify.txt"], { cwd });
|
||||
execFileSync("git", ["-c", "user.name=test", "-c", "user.email=test@example.com", "commit", "-qm", "initial"], { cwd });
|
||||
const transport = intercomFixture();
|
||||
const commands = new Map<string, any>();
|
||||
const hooks = new Map<string, any>();
|
||||
const tools = new Map<string, any>();
|
||||
@@ -47,16 +48,20 @@ function setup(selectChoices: Array<string | undefined>, editorChoices: Array<st
|
||||
};
|
||||
openSupervisorPane.mockImplementation(async () => "pane-2");
|
||||
const pi = {
|
||||
events: transport.events,
|
||||
registerCommand: (name: string, command: any) => commands.set(name, command),
|
||||
on: (name: string, handler: any) => hooks.set(name, handler),
|
||||
appendEntry: (customType: string, data: unknown) => entries.push({ type: "custom", customType, data }),
|
||||
on: (name: string, handler: any) => {
|
||||
const prior = hooks.get(name);
|
||||
hooks.set(name, async (...args: any[]) => { await prior?.(...args); return handler(...args); });
|
||||
},
|
||||
appendEntry: (customType: string, data: unknown) => { if (customType === "pi-goals-state") entries.push({ type: "custom", customType, data }); },
|
||||
registerTool: (tool: any) => tools.set(tool.name, tool),
|
||||
getAllTools: () => [],
|
||||
sendMessage: (message: { content: string; display?: boolean }) => messages.push(message),
|
||||
sendUserMessage: (content: string) => messages.push({ content }),
|
||||
};
|
||||
piGoalsExtension(pi as unknown as ExtensionAPI);
|
||||
return { commands, ctx, cwd, entries, hooks, messages, notifications, tools };
|
||||
return { commands, ctx, cwd, entries, hooks, messages, notifications, tools, transport };
|
||||
}
|
||||
|
||||
function writePlan(cwd: string, content: string): string {
|
||||
@@ -83,21 +88,21 @@ describe("/goals flow", () => {
|
||||
await flow.commands.get("goals").handler("make the file", flow.ctx);
|
||||
const path = approvedPlan(flow.cwd);
|
||||
await flow.hooks.get("agent_settled")({}, flow.ctx);
|
||||
const mailbox = (flow.entries.at(-1)?.data as { mailboxPath: string }).mailboxPath;
|
||||
const views = () => flow.transport.sent.filter(message => message.kind === "view");
|
||||
await vi.advanceTimersByTimeAsync(60 * 60_000);
|
||||
expect(workerViewsAfter(mailbox, 0).at(-1)?.text).toMatch(/^The worker stopped\./);
|
||||
expect(views().at(-1)?.text).toMatch(/^The worker stopped\./);
|
||||
flow.ctx.isIdle.mockReturnValue(false);
|
||||
await flow.hooks.get("agent_start")({}, flow.ctx);
|
||||
expect(workerViewsAfter(mailbox, 0).at(-1)?.text).toMatch(/^The worker is still working\./);
|
||||
expect(views().at(-1)?.text).toMatch(/^The worker is still working\./);
|
||||
await flow.hooks.get("agent_settled")({}, flow.ctx);
|
||||
expect(workerViewsAfter(mailbox, 0).at(-1)?.text).toMatch(/^The worker is still working\./);
|
||||
expect(views().at(-1)?.text).toMatch(/^The worker is still working\./);
|
||||
flow.ctx.isIdle.mockReturnValue(true);
|
||||
writeFileSync(path, readFileSync(path, "utf8").replace("[ ] goal:", "[x] goal:"));
|
||||
await flow.hooks.get("agent_settled")({}, flow.ctx);
|
||||
const count = workerViewsAfter(mailbox, 0).length;
|
||||
const count = views().length;
|
||||
expect(flow.entries.at(-1)?.data).toMatchObject({ phase: null });
|
||||
await vi.advanceTimersByTimeAsync(60 * 60_000);
|
||||
expect(workerViewsAfter(mailbox, 0)).toHaveLength(count);
|
||||
expect(views()).toHaveLength(count);
|
||||
} finally {
|
||||
await flow.hooks.get("session_shutdown")();
|
||||
vi.useRealTimers();
|
||||
@@ -135,8 +140,7 @@ describe("/goals flow", () => {
|
||||
sourceSessionFile: join(flow.cwd, "session.jsonl"),
|
||||
workerSessionId: "session-a",
|
||||
planPath,
|
||||
mailboxPath: expect.stringContaining(".pi/goals-supervision/session-a/"),
|
||||
}));
|
||||
}), expect.any(Function));
|
||||
expect(flow.entries.at(-1)?.data).toMatchObject({ phase: "working", supervisorPaneId: "pane-2" });
|
||||
expect(flow.messages.at(-1)?.content).toBe("The plan is approved. Begin implementation as the worker.");
|
||||
const prompt = await flow.hooks.get("before_agent_start")({}, flow.ctx);
|
||||
@@ -165,14 +169,14 @@ describe("/goals flow", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("delivers a mailbox instruction to the worker", async () => {
|
||||
it("delivers an Intercom instruction to the worker", async () => {
|
||||
const flow = setup(["Ready"]);
|
||||
try {
|
||||
await flow.commands.get("goals").handler("make the file", flow.ctx);
|
||||
approvedPlan(flow.cwd);
|
||||
await flow.hooks.get("agent_settled")({}, flow.ctx);
|
||||
const mailboxPath = (flow.entries.at(-1)?.data as { mailboxPath: string }).mailboxPath;
|
||||
writeWorkerSteer(mailboxPath, "Run the focused test.");
|
||||
const binding = (flow.entries.at(-1)?.data as { approvalId: string }).approvalId;
|
||||
flow.transport.receive({ binding, role: "supervisor", kind: "steer", id: "steer-1", text: "Run the focused test." });
|
||||
await flow.hooks.get("agent_settled")({}, flow.ctx);
|
||||
expect(flow.messages.some((message) => message.content === "[supervisor] Run the focused test.")).toBe(true);
|
||||
} finally {
|
||||
|
||||
+9
-33
@@ -2,46 +2,26 @@ 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, waitForSupervisorReady } from "../src/herdr.js";
|
||||
import { createMailbox, readyMailbox } from "../src/mailbox.js";
|
||||
import { closeSupervisorPane, openSupervisorPane, supervisorCommand } from "../src/herdr.js";
|
||||
|
||||
function input(mailboxPath = "/repo/.pi/goals-supervision/worker/approval") {
|
||||
function input() {
|
||||
return {
|
||||
cwd: "/repo",
|
||||
sourceSessionFile: "/sessions/worker.jsonl",
|
||||
workerSessionId: "worker-12345678",
|
||||
planPath: "/repo/.pi/plan/worker-v1.md",
|
||||
approvalId: "approval-1",
|
||||
mailboxPath,
|
||||
extensionPath: "/repo/src/index.ts",
|
||||
model: "provider/supervisor",
|
||||
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",
|
||||
};
|
||||
}
|
||||
|
||||
afterEach(() => vi.unstubAllEnvs());
|
||||
|
||||
describe("supervisor pane command", () => {
|
||||
it("forks the planning session with only pi-goals and its mailbox", () => {
|
||||
it("forks the planning session with pi-goals owning its Intercom dependency", () => {
|
||||
const command = supervisorCommand(input());
|
||||
expect(command).toContain("'PI_GOALS_ROLE=supervisor'");
|
||||
expect(command).toContain("'PI_GOALS_MAILBOX_PATH=/repo/.pi/goals-supervision/worker/approval'");
|
||||
expect(command).toContain("'PI_GOALS_APPROVAL_ID=approval-1'");
|
||||
expect(command).toContain("'pi' '--no-extensions' '-e' '/repo/src/index.ts'");
|
||||
expect(command).toContain("'--fork' '/sessions/worker.jsonl'");
|
||||
expect(command).toContain("'--model' 'provider/supervisor'");
|
||||
expect(command).not.toContain("pi-supervise");
|
||||
expect(command).not.toContain("pi-intercom");
|
||||
});
|
||||
|
||||
it("waits for an explicit mailbox readiness receipt", async () => {
|
||||
const cwd = mkdtempSync(join(tmpdir(), "pi-goals-herdr-"));
|
||||
try {
|
||||
const mailbox = createMailbox(cwd, "worker", "approval", join(cwd, "plan.md"));
|
||||
await expect(waitForSupervisorReady(mailbox.path, 10)).rejects.toThrow("did not become ready");
|
||||
readyMailbox(mailbox.path);
|
||||
await expect(waitForSupervisorReady(mailbox.path, 10)).resolves.toBeUndefined();
|
||||
} finally {
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("accepts Herdr's text version output and stale pane cleanup", async () => {
|
||||
@@ -58,14 +38,10 @@ exit 2
|
||||
vi.stubEnv("HERDR_ENV", "1");
|
||||
vi.stubEnv("HERDR_BIN_PATH", bin);
|
||||
try {
|
||||
const mailbox = createMailbox(cwd, "worker", "approval", join(cwd, "plan.md"));
|
||||
readyMailbox(mailbox.path);
|
||||
await expect(openSupervisorPane({ ...input(mailbox.path), cwd })).resolves.toBe("new-pane");
|
||||
await expect(openSupervisorPane({ ...input(), cwd }, vi.fn())).resolves.toBe("new-pane");
|
||||
await expect(closeSupervisorPane("new-pane")).resolves.toBeUndefined();
|
||||
vi.stubEnv("HERDR_SMOKE_RUN_FAIL", "1");
|
||||
await expect(openSupervisorPane({ ...input(mailbox.path), cwd })).rejects.toThrow("run failed");
|
||||
} finally {
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
await expect(openSupervisorPane({ ...input(), cwd }, vi.fn())).rejects.toThrow("run failed");
|
||||
} finally { rmSync(cwd, { recursive: true, force: true }); }
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { once } from "node:events";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
|
||||
import { IntercomClient } from "pi-intercom/broker/client.ts";
|
||||
import { expect, it, vi } from "vitest";
|
||||
import { GoalIntercom } from "../src/intercom.js";
|
||||
|
||||
it("exchanges readiness, views and exact advice over a real isolated pi-intercom broker", async () => {
|
||||
const directory = mkdtempSync(join(tmpdir(), "goals-intercom-"));
|
||||
vi.stubEnv("PI_CODING_AGENT_DIR", directory);
|
||||
const broker = spawn(process.execPath, [resolve("node_modules/tsx/dist/cli.mjs"), resolve("node_modules/pi-intercom/broker/broker.ts")], { env: { ...process.env }, stdio: ["ignore", "pipe", "pipe"] });
|
||||
const clients: IntercomClient[] = [];
|
||||
const shutdowns: Array<() => void> = [];
|
||||
try {
|
||||
await new Promise<void>((resolveReady, reject) => {
|
||||
const timer = setTimeout(() => reject(new Error("Isolated broker did not start.")), 5000);
|
||||
broker.stdout.on("data", chunk => { if (String(chunk).includes("Intercom broker started")) { clearTimeout(timer); resolveReady(); } });
|
||||
broker.once("exit", code => { clearTimeout(timer); reject(new Error(`Broker exited: ${code}`)); });
|
||||
});
|
||||
async function endpoint(role: "worker" | "supervisor") {
|
||||
const client = new IntercomClient();
|
||||
clients.push(client);
|
||||
await client.connect({ name: role, cwd: directory, model: "test", pid: process.pid, startedAt: Date.now(), lastActivity: Date.now(), extensions: [{ namespace: "pi-goals", ownerEligible: false }] });
|
||||
const entries: any[] = [];
|
||||
const api = {
|
||||
on: (name: string, callback: () => void) => { if (name === "session_shutdown") shutdowns.push(callback); },
|
||||
appendEntry: (customType: string, data: unknown) => entries.push({ type: "custom", customType, data }),
|
||||
events: {
|
||||
on() {},
|
||||
emit: (_name: string, registration: any) => {
|
||||
client.on("extension_message", message => registration.onEvent({ type: "message", fromSessionId: message.fromSessionId, payload: message.payload }));
|
||||
client.on("disconnected", () => registration.onEvent({ type: "connection", connected: false, supported: true }));
|
||||
client.onBrokerMessage(message => { if (message.type === "session_left") registration.onEvent(message); });
|
||||
registration.onReady({
|
||||
snapshot: () => ({ connected: client.isConnected(), supported: client.supportsFeature("extension-bus-v1") }),
|
||||
publish: (payload: unknown) => client.sendExtensionMessage({ type: "extension_publish", namespace: "pi-goals", audience: "capable", payload }),
|
||||
});
|
||||
},
|
||||
},
|
||||
};
|
||||
const link = new GoalIntercom(api as unknown as ExtensionAPI);
|
||||
link.configure("isolated-binding", role, { sessionManager: { getEntries: () => entries }, ui: { notify() {} } } as any);
|
||||
return { link, client };
|
||||
}
|
||||
const worker = await endpoint("worker");
|
||||
const supervisor = await endpoint("supervisor");
|
||||
expect(worker.link.connected).toBe(false);
|
||||
supervisor.link.markReady();
|
||||
await Promise.all([worker.link.waitReady(3000), supervisor.link.waitReady(3000)]);
|
||||
const viewed = new Promise<string>(resolveView => { supervisor.link.onView = view => resolveView(view.text); });
|
||||
const view = worker.link.view("The worker stopped.\nModal uses a remote GPU.", "settled");
|
||||
expect(await viewed).toBe(view.text);
|
||||
const advice = "Check the Modal dependency. Keep the local GPU queue paused.";
|
||||
const received = new Promise<string>(resolveAdvice => { worker.link.onSteer = resolveAdvice; });
|
||||
supervisor.link.steer(advice);
|
||||
expect(await received).toBe(advice);
|
||||
console.log("Intercom broker: readiness confirmed; exact worker view and supervisor advice received.");
|
||||
} finally {
|
||||
for (const shutdown of shutdowns) await shutdown();
|
||||
for (const client of clients) await client.disconnect();
|
||||
if (broker.exitCode === null) { broker.kill("SIGTERM"); await once(broker, "exit"); }
|
||||
vi.unstubAllEnvs();
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
}
|
||||
}, 15_000);
|
||||
@@ -0,0 +1,26 @@
|
||||
export function intercomFixture() {
|
||||
let registration: any;
|
||||
const sent: any[] = [];
|
||||
let connected = true;
|
||||
const receive = (payload: any, fromSessionId = "peer") => registration.onEvent({ type: "message", fromSessionId, payload });
|
||||
return {
|
||||
sent, receive,
|
||||
event: (event: any) => registration.onEvent(event),
|
||||
connect: (value: boolean) => { connected = value; registration.onEvent({ type: "connection", connected: value, supported: true }); },
|
||||
events: {
|
||||
on: () => () => {},
|
||||
emit: (name: string, value: any) => {
|
||||
if (name !== "intercom:extension-register") return false;
|
||||
registration = value;
|
||||
value.onReady({
|
||||
snapshot: () => ({ connected, supported: true }),
|
||||
publish: (message: any) => {
|
||||
sent.push(message);
|
||||
if (message.kind === "hello") queueMicrotask(() => receive({ ...message, role: message.role === "worker" ? "supervisor" : "worker", ready: true }));
|
||||
},
|
||||
});
|
||||
return true;
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { GoalIntercom } from "../src/intercom.js";
|
||||
import { intercomFixture } from "./intercom-fixture.js";
|
||||
|
||||
function setup(role: "worker" | "supervisor", entries: any[] = []) {
|
||||
const fixture = intercomFixture();
|
||||
const hooks = new Map<string, any>();
|
||||
const ctx = { sessionManager: { getEntries: () => entries }, ui: { notify: vi.fn() } };
|
||||
const api = { events: fixture.events, on: (name: string, hook: any) => hooks.set(name, hook), appendEntry: (customType: string, data: unknown) => entries.push({ type: "custom", customType, data }) };
|
||||
const link = new GoalIntercom(api as unknown as ExtensionAPI);
|
||||
link.configure("binding", role, ctx as any);
|
||||
return { link, fixture, entries, ctx, hooks };
|
||||
}
|
||||
|
||||
describe("pi-intercom transport", () => {
|
||||
it("receives exact advice once, acknowledges it and rejects unrelated peers", async () => {
|
||||
const runtime = setup("worker");
|
||||
await runtime.link.waitReady();
|
||||
const instruction = "Check the Modal dependency.\nKeep the local queue paused.";
|
||||
const delivered = vi.fn();
|
||||
runtime.link.onSteer = delivered;
|
||||
const message = { binding: "binding", role: "supervisor", kind: "steer", id: "instruction", text: instruction };
|
||||
runtime.fixture.receive({ ...message, binding: "other" });
|
||||
runtime.fixture.receive(message, "wrong-peer");
|
||||
expect(delivered).not.toHaveBeenCalled();
|
||||
runtime.fixture.receive(message);
|
||||
runtime.fixture.receive(message);
|
||||
expect(delivered).toHaveBeenCalledExactlyOnceWith(instruction);
|
||||
expect(runtime.fixture.sent.filter(message => message.kind === "received")).toHaveLength(2);
|
||||
await runtime.hooks.get("session_shutdown")();
|
||||
runtime.fixture.receive({ ...message, id: "late" });
|
||||
expect(delivered).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("restores an unacknowledged steer on reconnect and stops replay after acknowledgment", async () => {
|
||||
const first = setup("supervisor");
|
||||
first.link.markReady();
|
||||
await first.link.waitReady();
|
||||
const id = first.link.steer("Read the full output.");
|
||||
await first.hooks.get("session_shutdown")();
|
||||
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." }]);
|
||||
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);
|
||||
});
|
||||
|
||||
it("cancels a readiness wait on shutdown", async () => {
|
||||
const runtime = setup("worker");
|
||||
await runtime.link.waitReady();
|
||||
runtime.fixture.connect(false);
|
||||
const wait = runtime.link.waitReady();
|
||||
const rejection = expect(wait).rejects.toThrow("Session ended");
|
||||
await runtime.hooks.get("session_shutdown")();
|
||||
await rejection;
|
||||
});
|
||||
});
|
||||
@@ -6,18 +6,17 @@ import { stripVTControlCharacters } from "node:util";
|
||||
import { AssistantMessageComponent, type ExtensionAPI, initTheme, ToolExecutionComponent } from "@earendil-works/pi-coding-agent";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { approvalPath } from "../src/approval.js";
|
||||
import { createMailbox, supervisorReady, workerSteersAfter, writeWorkerView } from "../src/mailbox.js";
|
||||
import { registerVisibleSupervisor } from "../src/supervisor-session.js";
|
||||
import { intercomFixture } from "./intercom-fixture.js";
|
||||
|
||||
const shutdowns: Array<() => Promise<void>> = [];
|
||||
|
||||
function setup(cwd: string, planPath: string, tokens: number | null = 10, onCompact: (options: any) => void = (options) => options.onComplete()) {
|
||||
const mailbox = createMailbox(cwd, "worker-session", "approval-1", planPath);
|
||||
const transport = intercomFixture();
|
||||
vi.stubEnv("PI_GOALS_WORKER_ID", "worker-session");
|
||||
vi.stubEnv("PI_GOALS_OWNER_SESSION_ID", "worker-session");
|
||||
vi.stubEnv("PI_GOALS_PLAN_PATH", planPath);
|
||||
vi.stubEnv("PI_GOALS_APPROVAL_ID", "approval-1");
|
||||
vi.stubEnv("PI_GOALS_MAILBOX_PATH", mailbox.path);
|
||||
const hooks = new Map<string, any>();
|
||||
const tools = new Map<string, any>();
|
||||
const entries: any[] = [];
|
||||
@@ -33,7 +32,11 @@ function setup(cwd: string, planPath: string, tokens: number | null = 10, onComp
|
||||
ui: { notify: vi.fn() },
|
||||
};
|
||||
const pi = {
|
||||
on: (name: string, handler: any) => hooks.set(name, handler),
|
||||
events: transport.events,
|
||||
on: (name: string, handler: any) => {
|
||||
const prior = hooks.get(name);
|
||||
hooks.set(name, async (...args: any[]) => { await prior?.(...args); return handler(...args); });
|
||||
},
|
||||
registerTool: (tool: any) => tools.set(tool.name, tool),
|
||||
appendEntry: (customType: string, data: unknown) => entries.push({ type: "custom", customType, data }),
|
||||
sendUserMessage: (message: string) => messages.push(message),
|
||||
@@ -42,7 +45,15 @@ function setup(cwd: string, planPath: string, tokens: number | null = 10, onComp
|
||||
};
|
||||
registerVisibleSupervisor(pi as unknown as ExtensionAPI);
|
||||
shutdowns.push(() => hooks.get("session_shutdown")());
|
||||
return { activeTools: () => activeTools, branch: (value: any[]) => { branch = value; }, ctx, entries, hooks, mailbox, messages, tools };
|
||||
return {
|
||||
activeTools: () => activeTools, branch: (value: any[]) => { branch = value; }, ctx, entries, hooks, transport, messages, tools,
|
||||
ready: () => transport.sent.some(message => message.kind === "hello" && message.role === "supervisor" && message.ready),
|
||||
start: async () => { await hooks.get("session_start")({}, ctx); await new Promise(resolve => setImmediate(resolve)); },
|
||||
view: (id: string, text: string, reason = "settled") => {
|
||||
transport.receive({ binding: "approval-1", role: "worker", kind: "view", id, text, reason });
|
||||
return { text };
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
@@ -53,29 +64,23 @@ afterEach(async () => {
|
||||
|
||||
describe("visible supervisor session", () => {
|
||||
it("restores monitoring and read-only tools without replaying persisted views", async () => {
|
||||
vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
|
||||
const cwd = mkdtempSync(join(tmpdir(), "pi-goals-resume-"));
|
||||
try {
|
||||
const first = setup(cwd, join(cwd, "plan.md"));
|
||||
const view = writeWorkerView(first.mailbox, "settled", "The worker stopped.");
|
||||
await first.hooks.get("session_start")({}, first.ctx);
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
await first.start();
|
||||
const view = first.view("first", "The worker stopped.");
|
||||
expect(first.messages).toEqual([view.text]);
|
||||
await first.hooks.get("session_shutdown")();
|
||||
expect(supervisorReady(first.mailbox.path)).toBe(false);
|
||||
const resumed = setup(cwd, join(cwd, "plan.md"), 30_000);
|
||||
resumed.entries.push(...first.entries, { type: "message", message: { role: "user", content: [{ type: "text", text: view.text }] } });
|
||||
await resumed.hooks.get("session_start")({}, resumed.ctx);
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
resumed.entries.push(...first.entries);
|
||||
await resumed.start();
|
||||
expect(resumed.activeTools()).toEqual(["read", "grep"]);
|
||||
expect(resumed.ctx.compact).not.toHaveBeenCalled();
|
||||
resumed.view("first", view.text);
|
||||
expect(resumed.messages).toEqual([]);
|
||||
writeWorkerView(resumed.mailbox, "interval", "The worker stopped.\nOld view.");
|
||||
const latest = writeWorkerView(resumed.mailbox, "interval", "The worker stopped.\nCurrent view.");
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
const latest = resumed.view("second", "The worker stopped.\nCurrent view.");
|
||||
expect(resumed.messages).toEqual([latest.text]);
|
||||
writeWorkerView(resumed.mailbox, "started", "The worker is still working.");
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
resumed.view("third", "The worker is still working.", "started");
|
||||
expect(resumed.messages).toEqual([latest.text]);
|
||||
} finally { rmSync(cwd, { recursive: true, force: true }); }
|
||||
});
|
||||
@@ -147,7 +152,7 @@ describe("visible supervisor session", () => {
|
||||
await runtime.hooks.get("session_start")({}, runtime.ctx);
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
expect(runtime.ctx.compact).not.toHaveBeenCalled();
|
||||
expect(supervisorReady(runtime.mailbox.path)).toBe(true);
|
||||
expect(runtime.ready()).toBe(true);
|
||||
expect(runtime.activeTools()).toEqual(["read", "grep"]);
|
||||
expect(runtime.entries.at(-1)).toMatchObject({ customType: "pi-goals-visible-supervisor-v2" });
|
||||
} finally { rmSync(cwd, { recursive: true, force: true }); }
|
||||
@@ -161,9 +166,9 @@ describe("visible supervisor session", () => {
|
||||
await runtime.hooks.get("session_start")({}, runtime.ctx);
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
expect(runtime.ctx.compact).toHaveBeenCalledOnce();
|
||||
expect(supervisorReady(runtime.mailbox.path)).toBe(false);
|
||||
expect(runtime.ready()).toBe(false);
|
||||
complete!();
|
||||
expect(supervisorReady(runtime.mailbox.path)).toBe(true);
|
||||
expect(runtime.ready()).toBe(true);
|
||||
} finally { rmSync(cwd, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
@@ -173,7 +178,7 @@ describe("visible supervisor session", () => {
|
||||
const runtime = setup(cwd, join(cwd, ".pi/plan/worker-v1.md"), null, (options) => options.onError(new Error("offline")));
|
||||
await runtime.hooks.get("session_start")({}, runtime.ctx);
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
expect(supervisorReady(runtime.mailbox.path)).toBe(false);
|
||||
expect(runtime.ready()).toBe(false);
|
||||
expect(runtime.ctx.ui.notify).toHaveBeenCalledWith("Supervisor startup compaction failed: offline", "error");
|
||||
} finally { rmSync(cwd, { recursive: true, force: true }); }
|
||||
});
|
||||
@@ -182,9 +187,10 @@ describe("visible supervisor session", () => {
|
||||
const cwd = mkdtempSync(join(tmpdir(), "pi-goals-supervisor-"));
|
||||
try {
|
||||
const runtime = setup(cwd, join(cwd, "plan.md"));
|
||||
await runtime.start();
|
||||
const steered = await runtime.tools.get("SteerWorker").execute("id", { instruction: "Run the saved verification." });
|
||||
expect(steered.isError).toBe(false);
|
||||
expect(workerSteersAfter(runtime.mailbox.path, 0)).toMatchObject([{ sequence: 1, instruction: "Run the saved verification." }]);
|
||||
expect(runtime.transport.sent.filter(message => message.kind === "steer")).toMatchObject([{ text: "Run the saved verification." }]);
|
||||
} finally { rmSync(cwd, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
@@ -200,12 +206,13 @@ describe("visible supervisor session", () => {
|
||||
execFileSync("mkdir", ["-p", join(cwd, ".pi/plan")]);
|
||||
writeFileSync(planPath, "# Plan\n\n## Goals\n\n1. [ ] goal: make the file\n - discriminator: output exists\n - evidence:\n - `result.txt`: contains ok\n\n## Log\n");
|
||||
const runtime = setup(cwd, planPath);
|
||||
const view = writeWorkerView(runtime.mailbox, "settled", "The worker stopped.\n\ntool calls with no result: none");
|
||||
await runtime.start();
|
||||
const view = runtime.view("first", "The worker stopped.\n\ntool calls with no result: none");
|
||||
runtime.branch([{ type: "message", message: { role: "user", content: [{ type: "text", text: view.text }] } }]);
|
||||
const approved = await runtime.tools.get("ApproveGoal").execute("id", { goal: "make the file", verifyOutputPath: "verify.txt" }, undefined, undefined, runtime.ctx);
|
||||
expect(approved.isError).toBe(false);
|
||||
expect(existsSync(approvalPath(cwd, "worker-session", "make the file"))).toBe(true);
|
||||
writeWorkerView(runtime.mailbox, "started", "The worker is still working.");
|
||||
runtime.view("second", "The worker is still working.", "started");
|
||||
const stale = await runtime.tools.get("ApproveGoal").execute("id", { goal: "make the file", verifyOutputPath: "verify.txt" }, undefined, undefined, runtime.ctx);
|
||||
expect(stale.isError).toBe(true);
|
||||
expect(stale.content[0].text).toContain("latest worker view");
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
"target": "ES2022",
|
||||
"module": "ES2022",
|
||||
"moduleResolution": "bundler",
|
||||
"paths": { "pi-intercom": ["./src/intercom-package.d.ts"] },
|
||||
"strict": true,
|
||||
"esModuleInterop": true,
|
||||
"skipLibCheck": true,
|
||||
|
||||
Reference in New Issue
Block a user