Skip to content

feat(bin): serve quota dashboard on the tailnet #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bingb0t5 merged 7 commits into from
Aug 24, 2026
Merged

feat(bin): serve quota dashboard on the tailnet #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
8fa0505
feat(pi): default branch supervision and route heartbeats (#2939)
kunchenguid Aug 24, 2026
1fa9463
feat(bin): serve the fleet's quota dashboard on the tailnet
bingb0t5 Aug 24, 2026
2f10cde
no-mistakes(review): fix status classification, escaping, and test te…
bingb0t5 Aug 24, 2026
158bfae
no-mistakes(review): fix pace colors, NaN countdown, systemd PATH, pa…
bingb0t5 Aug 24, 2026
647d42a
no-mistakes(review): round percent display and mark stale providers a…
bingb0t5 Aug 24, 2026
469c200
fix(bin): outlive quota-axi's per-provider timeout on the dashboard
bingb0t5 Aug 24, 2026
b96295f
no-mistakes(review): Make quota dashboard tests executable
bingb0t5 Aug 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 52 additions & 59 deletions .pi/extensions/fm-branch-supervision.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,10 @@
// merges an append-only note to main's tail. Main's captain/assistant dialog
// is mirrored into the branch as read-only fm-main-mirror context at main's
// turn_end. Pi-only by construction: this file lives in .pi/extensions, so no
// other harness ever loads it, and a home that has not explicitly granted the
// wake's project in config/pi-supervision-branch (or runs away mode) keeps
// today's wake-to-main behavior untouched.
// other harness ever loads it. Supervision is default-on for every task once
// this Pi session owns the fleet lock: no captain grant file is required.
// Away mode (or a broken branch) keeps today's wake-to-main behavior
// untouched regardless.
//
// Prefix stability (the cache contract, owner: bin/fm-branch-prompt.sh
// header): the branch's system prompt is the generator's byte-stable output,
Expand Down Expand Up @@ -58,6 +59,7 @@ import { Text } from "@earendil-works/pi-tui";
import { Type } from "typebox";
import {
FM_BRANCH_DISPATCH_EVENT,
scopeForUnreadWake,
type BranchDispatchOffer,
} from "./lib/fm-branch-dispatch.ts";
import { encodeFirstmateOperationalInput } from "./lib/fm-operational-input.ts";
Expand All @@ -69,7 +71,6 @@ const fmHome = process.env.FM_HOME || process.env.FM_ROOT_OVERRIDE || root;
const fmRoot = process.env.FM_ROOT_OVERRIDE || root;
const state = process.env.FM_STATE_OVERRIDE || `${fmHome}/state`;
const config = process.env.FM_CONFIG_OVERRIDE || `${fmHome}/config`;
const configFile = join(config, "pi-supervision-branch");
const afkFlag = join(state, ".afk");
const sessionsDir = join(state, "branch-session");
const sessionPointer = join(state, ".branch-session");
Expand All @@ -90,7 +91,6 @@ const branchCacheKey = `fm-branch-${createHash("sha256").update(fmHome).digest("

const MIRROR_MESSAGE_CAP = 4000;
const MERGE_NOTE_BOAT = "⛵";
const MERGE_NOTE_ANCHOR = "⚓";
type MirrorItem = { tag: "captain" | "main"; text: string };
type MirrorCursor = { file: string; index: number };
type Verdict = "routine" | "captain";
Expand All @@ -104,32 +104,8 @@ const scriptEnv = {
FM_CONFIG_OVERRIDE: config,
};

function grantedProjects(): Set<string> {
try {
// Standing autonomy is project-specific. Each line must be an exact
// `project=<value>` matching task metadata; comments and blank lines are
// allowed, while malformed content fails the whole grant closed.
const grants = new Set<string>();
for (const raw of readFileSync(configFile, "utf8").split(/\r?\n/)) {
const line = raw.trim();
if (!line || line.startsWith("#")) continue;
if (!line.startsWith("project=") || line.length === 8) return new Set();
grants.add(line.slice(8));
}
return grants;
} catch {
return new Set();
}
}

function branchConfigured(): boolean {
return grantedProjects().size > 0;
}

function offerIsGranted(offer: BranchDispatchOffer): boolean {
if (!Array.isArray(offer.projects) || offer.projects.length === 0) return false;
const grants = grantedProjects();
return grants.size > 0 && offer.projects.every((project) => grants.has(project));
function offerEligible(offer: BranchDispatchOffer): boolean {
return offer.eligible === true;
}

function afkActive(): boolean {
Expand Down Expand Up @@ -343,28 +319,34 @@ export default function (pi: ExtensionAPI) {
// Append-only merge into main. The store row is already durable when this
// runs; the note is a cache of it at main's tail. Delivery modes per the
// design: routine+idle appends now with no turn, routine+busy appends after
// the captain's next prompt, captain-relevant appends and triggers exactly
// one turn (queued as a follow-up while main is busy). The read cursor
// advances once the note is handed to Pi; a crash inside Pi's own delivery
// window leaves the outcome durable in the store, where main's
// fm_branch_outcomes tool still reads it on demand.
// the captain's next prompt, captain-relevant triggers exactly one turn
// (queued as a follow-up while main is busy) - that follow-up turn is
// itself the captain-visible outcome, so the captain-facing note is
// delivered silently (display: false) rather than printed or rendered a
// second time; routine notes stay rendered except an explicitly silent
// no-change heartbeat. The read cursor advances once the note is handed to
// Pi; a crash inside Pi's
// own delivery window leaves the outcome durable in the store, where
// main's fm_branch_outcomes tool still reads it on demand.
function mergeIntoMain(
expectedGeneration: number,
seq: string,
task: string,
verdict: Verdict,
summary: string,
silent: boolean,
): boolean {
if (!actingAsOwner(expectedGeneration)) return false;
const glyph = verdict === "captain" ? MERGE_NOTE_ANCHOR : MERGE_NOTE_BOAT;
const note = `${glyph} ${task}: ${summary}`;
const message = { customType: "fm-branch-merge", content: note, display: true };
if (verdict === "captain") {
const message = { customType: "fm-branch-merge", content: `${task}: ${summary}`, display: false };
pi.sendMessage(message, { triggerTurn: true, deliverAs: "followUp" });
} else if (mainStreaming) {
pi.sendMessage(message, { deliverAs: "nextTurn" });
} else {
pi.sendMessage(message, {});
const message = { customType: "fm-branch-merge", content: `${MERGE_NOTE_BOAT} ${task}: ${summary}`, display: !(task === "fleet" && silent) };
if (mainStreaming) {
pi.sendMessage(message, { deliverAs: "nextTurn" });
} else {
pi.sendMessage(message, {});
}
}
if (/^[0-9]+$/.test(seq)) {
if (!actingAsOwner(expectedGeneration)) return false;
Expand All @@ -378,7 +360,7 @@ export default function (pi: ExtensionAPI) {
name: "fm_branch_report",
label: "Report supervision outcome",
description:
"Record the outcome of one handled fleet event: write it durably to the outcome store, then merge an append-only note into the captain-facing main conversation. verdict captain surfaces it to the captain in one turn; verdict routine merges silently.",
"Record the outcome of one handled fleet event: write it durably to the outcome store, then merge an append-only note into the captain-facing main conversation. verdict captain surfaces it to the captain in one turn; routine notes render unless silent marks a no-change heartbeat.",
parameters: Type.Object({
task: Type.String({ description: "The task id the event belongs to (or 'fleet' for fleet-wide events)" }),
verdict: Type.Union([Type.Literal("routine"), Type.Literal("captain")], {
Expand All @@ -389,21 +371,25 @@ export default function (pi: ExtensionAPI) {
"One or two sentences in captain outcome language; include the full https:// PR URL when a PR is involved",
}),
wake: Type.Optional(Type.String({ description: "The wake reason line this outcome answers" })),
silent: Type.Optional(Type.Boolean({
description: "True only when a fleet-wide heartbeat review found literally nothing worth reporting; omit or use false whenever any action was taken or any routine result is worth a note",
})),
}),
execute: async (_toolCallId, params) => {
const task = String((params as { task: unknown }).task || "").trim();
const verdictRaw = String((params as { verdict: unknown }).verdict || "");
const summary = String((params as { summary: unknown }).summary || "").trim();
const wake = String((params as { wake?: unknown }).wake ?? "").trim();
if (!task || !summary || (verdictRaw !== "routine" && verdictRaw !== "captain")) {
const silent = (params as { silent?: unknown }).silent === true;
if (!task || !summary || (verdictRaw !== "routine" && verdictRaw !== "captain") || (silent && (task !== "fleet" || verdictRaw !== "routine"))) {
return {
content: [{ type: "text", text: "invalid report: task, verdict (routine|captain), and summary are required" }],
details: undefined,
isError: true,
};
}
const verdict = verdictRaw as Verdict;
const appendArgs = ["append", "--task", task, "--verdict", verdict, "--summary", summary];
const appendArgs = ["append", "--task", task, "--verdict", verdict, "--summary", summary, "--silent", String(silent)];
if (wake) appendArgs.push("--wake", wake);
if (!actingAsOwner(toolGeneration)) {
return {
Expand All @@ -420,7 +406,7 @@ export default function (pi: ExtensionAPI) {
isError: true,
};
}
if (!mergeIntoMain(toolGeneration, appended.stdout, task, verdict, summary)) {
if (!mergeIntoMain(toolGeneration, appended.stdout, task, verdict, summary, silent)) {
return {
content: [{ type: "text", text: `recorded seq ${appended.stdout}, but merge refused after supervision replacement or lock loss` }],
details: undefined,
Expand Down Expand Up @@ -604,6 +590,14 @@ ${context.command}
const session = await ensureBranch(acceptedGeneration);
await flushMirror(session, acceptedGeneration);
if (!actingAsOwner(acceptedGeneration)) throw new Error("supervision session no longer owns the fleet lock");
const heartbeat = /^heartbeat($|:)/.test(message);
const scope = scopeForUnreadWake(state, heartbeat);
if (scope.status === "empty") return;
if (scope.status === "unsafe") {
throw new Error("unread wake queue now contains a main-owned row or could not be read safely");
}
// A row can still arrive between this re-check and the model starting
// the drain; that residual is accepted by the confused-agent-grade boundary.
await session.prompt(
`FIRSTMATE SUPERVISION WAKE: ${message}\n\nHandle this per your operating procedure and finish with fm_branch_report.`,
);
Expand Down Expand Up @@ -635,10 +629,10 @@ ${context.command}
pi.events?.on?.(FM_BRANCH_DISPATCH_EVENT, (data) => {
const offer = data as BranchDispatchOffer;
if (!offer || typeof offer.accept !== "function") return;
// Check project-specific consent before ownership activation so an
// unconfigured or out-of-scope wake gets neither branch routing nor
// branch-owned state/lease cleanup side effects.
if (!offerIsGranted(offer)) return;
// Check eligibility before ownership activation so an out-of-scope wake
// gets neither branch routing nor branch-owned state/lease cleanup side
// effects.
if (!offerEligible(offer)) return;
if (!actingAsOwner()) return; // cold start pre-lock, secondary session, or shutdown
if (afkActive()) return; // the away daemon owns supervision while afk
if (branchBroken) return; // fail back to today's wake-to-main path
Expand All @@ -661,7 +655,7 @@ ${context.command}
// lands before any later wake. The durable cursor advances only in
// flushMirror after the complete pending batch reaches the branch.
pi.on?.("turn_end", (_event, ctx) => {
if (!branchConfigured() || !actingAsOwner()) return;
if (!actingAsOwner()) return;
try {
pendingMirror.push(...collectMainDialog(ctx.sessionManager, mirrorCollection));
} catch {
Expand All @@ -681,7 +675,7 @@ ${context.command}
shuttingDown = false;
branchBroken = "";
generation += 1;
if (branchConfigured()) actingAsOwner(generation);
actingAsOwner(generation);
});

pi.on?.("session_shutdown", () => {
Expand Down Expand Up @@ -727,17 +721,16 @@ ${context.command}
},
});

// Pi only calls this renderer for a message with display: true, which
// mergeIntoMain sets for every routine note except an explicitly silent
// fleet heartbeat; captain-facing notes are never printed or rendered here.
pi.registerMessageRenderer?.("fm-branch-merge", (message, _options, theme) => {
const note = textOfContent(message.content);
const glyph = note.startsWith(MERGE_NOTE_ANCHOR)
? MERGE_NOTE_ANCHOR
: note.startsWith(MERGE_NOTE_BOAT)
? MERGE_NOTE_BOAT
: "";
const rest = glyph ? note.slice(glyph.length) : note;
const hasGlyph = note.startsWith(MERGE_NOTE_BOAT);
const rest = hasGlyph ? note.slice(MERGE_NOTE_BOAT.length) : note;
const outputPad = 1;
return new Text(
`${glyph ? theme.fg("customMessageText", glyph) : ""}${theme.fg("dim", rest)}`,
`${hasGlyph ? theme.fg("customMessageText", MERGE_NOTE_BOAT) : ""}${theme.fg("dim", rest)}`,
outputPad,
0,
);
Expand Down
69 changes: 8 additions & 61 deletions .pi/extensions/fm-primary-pi-watch.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
// callbacks from a prior generation are no-ops against the active replacement.
import { spawn, spawnSync, type ChildProcess } from "node:child_process";
import { createHash } from "node:crypto";
import { mkdirSync, readFileSync, readdirSync, writeFileSync } from "node:fs";
import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { dirname, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import type { ExtensionAPI, Theme } from "@earendil-works/pi-coding-agent";
Expand All @@ -19,6 +19,7 @@ import { Type } from "typebox";
import {
createBranchDispatchOffer,
FM_BRANCH_DISPATCH_EVENT,
scopeForUnreadWake,
} from "./lib/fm-branch-dispatch.ts";
import {
type CalmPresentationState,
Expand Down Expand Up @@ -296,72 +297,18 @@ export default function (pi: ExtensionAPI) {
return confirmHandlingDelivery(snapshot());
}

// Resolve every unread row that the branch's mandatory fm-wake-drain call
// would present. Only task-local signal/stale rows can be delegated: a
// fleet-wide check/heartbeat or an unresolvable task stays with main. This
// prevents one project's autonomy grant from authorizing work on another
// project merely because both projects share a firstmate home.
function projectsForUnreadWake(): string[] {
let queue = "";
try {
queue = readFileSync(`${state}/.wake-queue`, "utf8");
} catch {
return [];
}

const projects = new Set<string>();
const metadata = new Map<string, string>();
try {
for (const name of readdirSync(state)) {
if (!name.endsWith(".meta")) continue;
const task = name.slice(0, -5);
const fields = readFileSync(`${state}/${name}`, "utf8").split(/\r?\n/);
const project = fields.find((line) => line.startsWith("project="))?.slice(8) ?? "";
const window = fields.find((line) => line.startsWith("window="))?.slice(7) ?? "";
if (project) {
metadata.set(task, project);
if (window) metadata.set(window, project);
}
}
} catch {
return [];
}

let found = false;
for (const line of queue.split(/\r?\n/)) {
if (!line) continue;
const fields = line.split("\t");
if (fields.length < 4 || !/^[0-9]+$/.test(fields[1])) continue;
const kind = fields[2];
const key = fields[3];
let project = "";
if (kind === "signal") {
const task = key.replace(/\.(?:status|turn-ended)$/, "");
project = metadata.get(task) ?? "";
} else if (kind === "stale") {
project = metadata.get(key) ?? metadata.get(key.replace(/^fm-/, "")) ?? "";
} else {
return [];
}
if (!project) return [];
found = true;
projects.add(project);
}
return found ? [...projects] : [];
}

// Offer an ordinary, project-scoped actionable wake to the supervision
// branch. A synchronous accept means the branch now owns delivery and
// handling; every unsafe or unaccepted offer keeps today's main path.
function offerWakeToBranch(message: string): boolean {
const offer = createBranchDispatchOffer(message, projectsForUnreadWake());
const heartbeat = /^heartbeat($|:)/.test(message);
const scope = scopeForUnreadWake(state, heartbeat);
const offer = createBranchDispatchOffer(message, scope.projects, heartbeat, scope.eligible);
pi.events?.emit?.(FM_BRANCH_DISPATCH_EVENT, offer);
return offer.accepted;
}

async function deliverActionableWake(
owner: SessionGeneration,
message: string,
repairFailed: boolean,
recovery?: { generation: string; watcherPid: string },
): Promise<void> {
if (!generationIsLive(owner)) return;
Expand All @@ -376,7 +323,7 @@ export default function (pi: ExtensionAPI) {
return;
}
}
if (offerWakeToBranch(message)) return;
if (!repairFailed && offerWakeToBranch(message)) return;
await sendWake(owner, message);
}

Expand Down Expand Up @@ -574,7 +521,7 @@ export default function (pi: ExtensionAPI) {
const restoration = await restoreAfterActionableClose(owner, predecessor);
if (!generationIsLive(owner)) return;
const message = restoration.failure ? `${classification.message}\n\n${restoration.failure}` : classification.message;
await deliverActionableWake(owner, message, restoration.recovery);
await deliverActionableWake(owner, message, Boolean(restoration.failure), restoration.recovery);
} catch (error) {
const detail = error instanceof Error ? error.message : String(error);
surfaceFailure(owner, `watcher: FAILED - Pi extension could not deliver an actionable wake\n${detail}`);
Expand Down
Loading
Loading