wassname/pi-plan
1
0
Fork 0
mirror of https://github.com/wassname/pi-plan.git synced 2026-09-26 14:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

Send incremental worker views with direction and tracked job state

Browse Source 
Borrow the provider tracker queries from cecb1e9. Keep unavailable state unknown, bind incremental views to acknowledged source entries, reset after compaction, bound serialized payloads, and recheck tracked work at sign-off.
This commit is contained in:
wassname
2026-09-08 16:41:47 +08:00
parent 489298d58b
commit 47cc054582
13 changed files with 305 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/background.ts
+33
View File
@@ -0,0 +1,33 @@
import { randomUUID } from "node:crypto";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
// Pi/OpenAI: Adapted from this repo's feature/simple-visible-supervision at cecb1e9.
export async function backgroundState(pi: ExtensionAPI): Promise<{ quiet: boolean; description: string }> {
const tools = pi.getAllTools();
const hasProcesses = tools.some(tool => tool.name === "process");
const hasSubagents = tools.some(tool => tool.name === "subagent");
let processes: unknown;
pi.events.emit("processes:request:list", { reply: (value: unknown) => { processes = value; } });
const rows = Array.isArray(processes) ? processes : !hasProcesses && processes === undefined ? [] : null;
const known = rows?.every(p => p && ["running", "terminating", "terminate_timeout", "exited", "killed"].includes(p.status));
const activeProcesses = known ? rows!.filter(p => !["exited", "killed"].includes(p.status)) : null;
let subagents: number | null = hasSubagents ? null : 0;
if (hasSubagents) {
const requestId = randomUUID();
subagents = await new Promise<number | null>(resolve => {
let unsubscribe: (() => void) | undefined;
const finish = (value: number | null) => { clearTimeout(timer); unsubscribe?.(); resolve(value); };
const timer = setTimeout(() => finish(null), 2000);
unsubscribe = pi.events.on(`subagents:rpc:v1:reply:${requestId}`, (reply: any) => {
if (reply?.requestId !== requestId) return;
const count = reply?.success && reply?.data?.fleet?.version === 1 ? reply.data.fleet.totalActive : undefined;
finish(Number.isSafeInteger(count) && count >= 0 ? count : null);
});
pi.events.emit("subagents:rpc:v1:request", { version: 1, requestId, method: "status", params: {}, source: { extension: "pi-goals" } });
});
}
return {
quiet: activeProcesses?.length === 0 && subagents === 0,
description: `processes: ${activeProcesses?.length ?? "unknown"}${activeProcesses?.length ? ` (${activeProcesses.map(p => p.name || p.id).join(", ")})` : ""}; subagents: ${subagents ?? "unknown"}; unregistered detached work is not tracked`,
};
}
src/index.ts
+34 -10
View File
@@ -21,6 +21,7 @@ import { fileURLToPath } from "node:url";
import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
import { Type } from "typebox";
import { approvalMatches, approvalPath, goalBlock, hashGoalBlock, readApproval, repositoryState } from "./approval.js";
import { backgroundState } from "./background.js";
import { closeSupervisorPane, openSupervisorPane } from "./herdr.js";
import { GoalIntercom } from "./intercom.js";
import { completeGoalDescription, completeGoalParamDescription, planDrafting, planningState, resync } from "./prompts.js";
@@ -97,6 +98,7 @@ interface PlanState {
supervisorPaneId: string | null;
approvalId: string | null;
planVersion: number | null;
latestDirection: string;
}
export default function piGoalsExtension(pi: ExtensionAPI): void {
@@ -116,6 +118,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
supervisorPaneId: null,
approvalId: null,
planVersion: null,
latestDirection: "",
};
let planningContextPending = false;
let resyncReason: string | null = "New session.";
@@ -185,11 +188,22 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
}
let workerTurns = 0;
let viewGeneration = 0;
let viewTimer: ReturnType<typeof setInterval> | undefined;
function publishWorkerView(ctx: ExtensionContext, reason: "ready" | "settled" | "turns" | "interval" | "started"): void {
if (state.phase !== "working") return;
intercom.view(workerView(ctx.sessionManager.getBranch(), reason, reason !== "started" && ctx.isIdle()), reason);
async function publishWorkerView(ctx: ExtensionContext, reason: "ready" | "settled" | "turns" | "interval" | "started"): Promise<void> {
if (state.phase !== "working" || intercom.ended) return;
const generation = ++viewGeneration;
const binding = state.approvalId;
const background = reason === "started" ? { quiet: false, description: "agent starting; background state not queried" } : await backgroundState(pi);
if (intercom.ended || generation !== viewGeneration || binding !== state.approvalId || state.phase !== "working") return;
const entries = ctx.sessionManager.getBranch();
const view = workerView(entries, reason, reason !== "started" && ctx.isIdle(), {
sourceSession: ctx.sessionManager.getSessionFile()!, latestDirection: state.latestDirection,
model: ctx.model ? `${ctx.model.provider}/${ctx.model.id}` : "not selected",
since: intercom.acknowledgedEntry, background: background.description,
});
intercom.view(view, reason, entries.at(-1)?.id, background.quiet);
const goals = scanGoals(readPlan(ctx));
if (goals.length > 0 && goals.every((goal) => goal.status === "done" || goal.status === "cancelled")) {
stopWorkerTimers();
@@ -199,7 +213,9 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
}
function startWorkerTimers(ctx: ExtensionContext): void {
if (!viewTimer) viewTimer = setInterval(() => publishWorkerView(ctx, "interval"), 60 * 60_000);
if (!viewTimer) viewTimer = setInterval(() => {
void publishWorkerView(ctx, "interval").catch(error => { if (!intercom.ended) ctx.ui.notify(`Worker view failed: ${String(error)}`, "error"); });
}, 60 * 60_000);
}
function stopWorkerTimers(): void {
@@ -290,7 +306,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, planVersion: nextVersion(ctx) };
state = { ...state, phase: "planning", supervisorPaneId: null, approvalId: null, planVersion: nextVersion(ctx), latestDirection: arg };
planningContextPending = true;
resyncReason = null;
writePlan(ctx, "");
@@ -344,11 +360,14 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
// PI: Human plan-mode replies are durable evidence of the interview, not model summaries.
pi.on("input", async (event, ctx) => {
if (state.phase === "planning" && event.source !== "extension") writePlan(ctx, appendInterview(readPlan(ctx), event.text));
if (event.source === "extension") return;
state = { ...state, latestDirection: event.text };
persist();
if (state.phase === "planning") writePlan(ctx, appendInterview(readPlan(ctx), event.text));
});
pi.on("agent_start", async (_event, ctx) => {
publishWorkerView(ctx, "started");
await publishWorkerView(ctx, "started");
});
pi.on("turn_end", async (_event, ctx) => {
@@ -357,7 +376,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
workerTurns++;
if (workerTurns < 50) return;
workerTurns = 0;
publishWorkerView(ctx, "turns");
await publishWorkerView(ctx, "turns");
});
pi.on("tool_call", async (event, ctx) => {
@@ -383,7 +402,7 @@ 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") {
publishWorkerView(ctx, "settled");
await publishWorkerView(ctx, "settled");
return;
}
if (state.phase !== "planning" || !ctx.hasUI) return;
@@ -402,6 +421,8 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
if (choice === "Refine") {
const notes = await ctx.ui.editor("What should change about the plan?", "");
if (!notes?.trim()) continue;
state = { ...state, latestDirection: notes };
persist();
writePlan(ctx, appendInterview(plan, notes));
planningContextPending = true;
pi.sendUserMessage(`Revise the plan at ${planPath(ctx)} using these human notes:\n\n${notes}\n\nKeep the same goal structure.`, { deliverAs: "followUp" });
@@ -427,7 +448,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
resyncReason = "The plan was approved.";
persist();
startWorkerTimers(ctx);
publishWorkerView(ctx, "ready");
await publishWorkerView(ctx, "ready");
updateWidget(ctx);
ctx.ui.notify(`Visible supervisor opened in Herdr pane ${state.supervisorPaneId}.`, "info");
pi.sendUserMessage("The plan is approved. Begin implementation as the worker.");
@@ -452,6 +473,7 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
supervisorPaneId: last?.data?.supervisorPaneId ?? null,
approvalId: last?.data?.approvalId ?? null,
planVersion: last?.data?.planVersion ?? null,
latestDirection: last?.data?.latestDirection ?? "",
};
planningContextPending = state.phase === "planning";
resyncReason = state.phase === "working" ? "New session." : null;
@@ -476,6 +498,8 @@ export default function piGoalsExtension(pi: ExtensionAPI): void {
async execute(_id, params, _signal, _onUpdate, ctx) {
if (state.phase !== "working") return result("Planning is not approved. Choose Ready before signing off a goal.", true);
if (!state.approvalId) return result("Goal sign-off blocked: no current supervisor review.", true);
const background = await backgroundState(pi);
if (intercom.ended || !background.quiet) return result(`Goal sign-off blocked: ${background.description}`, true);
const plan = readPlan(ctx);
if (!plan.trim()) return result(`No plan file at ${planRel(ctx)}. Run /goals to draft one.`, true);
const block = goalBlock(plan, params.goal);
src/intercom.ts
+25 -11
View File
@@ -3,8 +3,8 @@ import type { ExtensionAPI, ExtensionContext, SessionStartEvent } from "@earendi
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 }
export interface View { id: string; text: string; reason: string; through?: string; backgroundQuiet: boolean }
interface Message { binding: string; role: Role; kind: "hello" | "view" | "steer" | "received"; id: string; text?: string; reason?: string; ready?: boolean; through?: string; backgroundQuiet?: boolean }
const STATE = "pi-goals-intercom";
export class GoalIntercom {
@@ -21,6 +21,7 @@ export class GoalIntercom {
private received = new Set<string>();
private waiters = new Set<() => void>();
latestView?: View;
acknowledgedEntry?: string;
onView: (view: View) => void = () => {};
onSteer: (text: string) => void = () => {};
@@ -48,15 +49,19 @@ export class GoalIntercom {
this.pending.clear();
this.received.clear();
this.latestView = undefined;
this.acknowledgedEntry = 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 === "ack") {
this.pending.delete(message.id);
if (message.through) this.acknowledgedEntry = message.through;
}
if (record.direction === "in") this.received.add(message.id);
if (message.kind === "view") this.latestView = { id: message.id, text: message.text!, reason: message.reason! };
if (message.kind === "view") this.latestView = { id: message.id, text: message.text!, reason: message.reason!, through: message.through, backgroundQuiet: message.backgroundQuiet === true };
}
this.hello();
}
@@ -80,11 +85,11 @@ export class GoalIntercom {
});
}
view(text: string, reason: string): View {
view(text: string, reason: string, through?: string, backgroundQuiet = false): View {
const id = randomUUID();
const message: Message = { binding: this.binding, role: this.role, kind: "view", id, text: `${text}\n\nworker view id: ${id}`, reason };
const message: Message = { binding: this.binding, role: this.role, kind: "view", id, text: `${text}\n\nworker view id: ${id}`, reason, through, backgroundQuiet };
this.record("out", message);
this.latestView = { id, text: message.text!, reason };
this.latestView = { id, text: message.text!, reason, through, backgroundQuiet };
if (this.connected) this.publish(message);
return this.latestView;
}
@@ -143,14 +148,23 @@ export class GoalIntercom {
return;
}
if (event.fromSessionId !== this.peer || !this.ready) return;
if (message.kind === "received") { this.pending.delete(message.id); this.record("ack", message); return; }
if (message.kind === "received") {
this.pending.delete(message.id);
const through = message.id === this.latestView?.id ? this.latestView.through : undefined;
if (through) this.acknowledgedEntry = through;
this.record("ack", { ...message, through });
return;
}
if (this.received.has(message.id)) {
if (message.kind === "steer") this.publish({ ...message, role: this.role, kind: "received" });
if (message.kind === "steer" || (message.kind === "view" && message.reason !== "started")) this.publish({ binding: this.binding, role: this.role, kind: "received", id: message.id });
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);
this.latestView = { id: message.id, text: message.text!, reason: message.reason!, through: message.through, backgroundQuiet: message.backgroundQuiet === true };
if (message.reason !== "started") {
this.onView(this.latestView);
this.publish({ binding: this.binding, role: this.role, kind: "received", id: message.id });
}
} else if (message.kind === "steer" && this.role === "worker") {
this.onSteer(message.text!);
this.received.add(message.id);
src/supervisor-session.ts
+1
View File
@@ -192,6 +192,7 @@ export function registerVisibleSupervisor(pi: ExtensionAPI): void {
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);
if (!newest.backgroundQuiet) return result("Cannot approve while tracked background work is active or unknown.", true);
const pendingTool = view.match(/^tool calls with no result: (?!none$)(.+)$/m);
const pendingChild = view.match(/^child pi processes still running: (?!none$)(.+)$/m);
if (pendingTool || pendingChild) return result(`Cannot approve while work is active: ${(pendingTool ?? pendingChild)![1]}`, true);
src/worker-view.ts
+26 -5
View File
@@ -12,6 +12,7 @@ export interface SessionMessage {
}
export interface SessionEntry {
id?: string;
type?: string;
summary?: string;
message?: SessionMessage;
@@ -38,10 +39,30 @@ function outstandingTools(entries: SessionEntry[]): string[] {
return [...calls].filter(([id]) => !results.has(id)).map(([, name]) => name);
}
export function workerView(entries: SessionEntry[], reason: "ready" | "settled" | "turns" | "interval" | "started", idle: boolean): string {
const summary = [...entries].reverse().find((entry) => entry.type === "compaction" && entry.summary)?.summary;
const recent = entries.flatMap((entry) => entry.type === "message" && entry.message ? [text(entry.message)] : []).filter(Boolean).slice(-12).join("\n\n").slice(-12_000);
const outstanding = outstandingTools(entries);
function bounded(value: string, bytes: number, tail = false): string {
if (Buffer.byteLength(JSON.stringify(value)) <= bytes) return value;
let size = Math.min(value.length, bytes - 100);
while (Buffer.byteLength(JSON.stringify(tail ? value.slice(-size) : value.slice(0, size))) > bytes - 100) size = Math.floor(size * 0.8);
const notice = "[truncated; inspect source session]";
return tail ? `${notice}\n${value.slice(-size)}` : `${value.slice(0, size)}\n${notice}`;
}
export interface ViewContext {
sourceSession: string;
latestDirection: string;
model: string;
since?: string;
background: string;
}
export function workerView(entries: SessionEntry[], reason: "ready" | "settled" | "turns" | "interval" | "started", idle: boolean, context: ViewContext): string {
const compactAt = entries.map(entry => entry.type).lastIndexOf("compaction");
const since = context.since ? entries.findIndex(entry => entry.id === context.since) : -1;
const from = since >= compactAt ? since + 1 : compactAt + 1;
const fresh = entries.slice(from);
const recent = fresh.flatMap(entry => entry.type === "message" && entry.message ? [text(entry.message)] : []).filter(Boolean).join("\n\n");
const summary = since < compactAt ? entries[compactAt]?.summary : undefined;
const outstanding = outstandingTools(entries.slice(compactAt + 1));
const state = reason === "ready" ? "is ready to begin" : idle ? "stopped" : "is still working";
return `The worker ${state}.\n\nreview trigger: ${reason}\ntool calls with no result: ${outstanding.join(", ") || "none"}\nbackground job state: not measured; inspect job evidence before approval\n\n${summary ? `last compaction summary:\n${summary}\n\n` : ""}recent worker transcript:\n${recent || "none"}`;
return `The worker ${state}.\n\nreview trigger: ${reason}\nsource session: ${bounded(context.sourceSession, 800)}\nworker model: ${bounded(context.model, 300)}\nlatest human direction:\n${bounded(context.latestDirection || "not recorded", 1800)}\ntool calls with no result: ${bounded(outstanding.join(", ") || "none", 500)}\ntracked background work: ${bounded(context.background, 800)}\n\n${summary ? `compaction summary (worker account, not independent evidence):\n${bounded(summary, 2500)}\n\n` : ""}new worker transcript${since === -1 ? " (initial or reset view)" : " since the last acknowledged view"}:\n${bounded(recent || "No new messages.", 7000, true)}`;
}