Model-neutral Pi extension for live decode TPS, TTFT, server-side prefill TPS, and whole-request elapsed time (v2)

live-throughput-status.test.ts

/**
 * Test suite for live-throughput-status.ts
 *
 * Run:  bun run ~/.pi/agent/extensions/tests/live-throughput-status.test.ts
 *
 * Lives in extensions/tests/ (no index.ts) so pi's auto-discovery
 * (`extensions/*.ts`, `extensions/*\/index.ts`) does not load it.
 */

import assert from "node:assert/strict";
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import mod, { formatDuration, parsePrefillMetrics } from "./live-throughput-status.ts";

type Status = string | undefined;

/**
 * Every harness gets its own temp state dir by default, so calibration from one
 * test cannot leak into another. Pass `dir` to share state (persistence test).
 */
function harness(mode: "tui" | "rpc" | "print" = "tui", dirOverride?: string) {
	const dir = dirOverride ?? mkdtempSync(join(tmpdir(), "ltps-"));
	const prev = process.env.PI_LIVE_THROUGHPUT_DIR;
	process.env.PI_LIVE_THROUGHPUT_DIR = dir;

	const handlers: Record<string, Function[]> = {};
	const statuses: Status[] = [];
	const pi = { on: (n: string, f: Function) => (handlers[n] ??= []).push(f) } as any;
	const ctx = {
		mode,
		model: { provider: "test", id: "m1" },
		ui: {
			setStatus: (_k: string, t?: Status) => statuses.push(t),
			theme: { fg: (_c: string, s: string) => s },
		},
	} as any;
	mod(pi);

	// Factory already resolved paths; restore env so other code is unaffected.
	if (prev === undefined) delete process.env.PI_LIVE_THROUGHPUT_DIR;
	else process.env.PI_LIVE_THROUGHPUT_DIR = prev;

	const fire = async (n: string, e: any) => {
		for (const f of handlers[n] ?? []) await f(e, ctx);
	};
	return { fire, statuses, dir, statePath: join(dir, "live-throughput-state.json") };
}

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

function textDelta(chars: number, tokens?: number) {
	return {
		message: { role: "assistant" },
		assistantMessageEvent: {
			type: "text_delta",
			delta: "x".repeat(chars),
			...(tokens !== undefined ? { partial: { usage: { output: tokens } } } : {}),
		},
	};
}
function thinkDelta(chars: number, tokens?: number) {
	return {
		message: { role: "assistant" },
		assistantMessageEvent: {
			type: "thinking_delta",
			delta: "y".repeat(chars),
			...(tokens !== undefined ? { partial: { usage: { output: tokens } } } : {}),
		},
	};
}
function assistantStart(model = "m1") {
	return { message: { role: "assistant", provider: "test", model } };
}
function end(usage: any, stopReason = "stop") {
	return { message: { role: "assistant", stopReason, usage } };
}

async function startTurn(fire: (n: string, e: any) => Promise<void>, model = "m1") {
	await fire("session_start", {});
	await fire("before_provider_request", {});
	await fire("message_start", assistantStart(model));
}

const lastTokStatus = (statuses: Status[]) =>
	[...statuses].reverse().find((s) => s?.includes("tok/s")) ?? "n/a";

let passed = 0;
function ok(name: string, fn: () => void) {
	try {
		fn();
		passed++;
		console.log(`  ✓ ${name}`);
	} catch (err) {
		console.error(`  ✗ ${name}`);
		console.error(err instanceof Error ? err.message : err);
		process.exitCode = 1;
	}
}

async function testA() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await sleep(250);
	for (let i = 0; i < 50; i++) {
		await fire("message_update", textDelta(5));
		await sleep(20);
	}
	await fire("message_end", end({ input: 1000, output: 100, cacheRead: 0, cacheWrite: 0 }));

	ok("A first frame is 'warming up'", () => {
		assert.match(statuses[2]!, /warming up/);
		assert.doesNotMatch(statuses[2]!, /tok\/s/);
	});
	ok("A final decode rate is plausible (~90-110 tok/s)", () => {
		const live = lastTokStatus(statuses);
		const m = /([\d.]+) tok\/s TTFT: ([\d.]+)s\s+(\d+) tok/.exec(live);
		assert.ok(m, `unexpected format: ${live}`);
		const v = Number(m![1]);
		assert.ok(v > 80 && v < 120, `rate ${v}`);
	});
	ok("A shows Prompt count, not Input/TTFT", () => {
		const last = statuses[statuses.length - 1]!;
		assert.match(last, /Prompt: 1\.0k tok/);
		assert.doesNotMatch(last, /Input\/TTFT/);
	});
}

async function testB() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await sleep(300);
	await fire("message_update", textDelta(396));
	await sleep(500);
	await fire("message_update", textDelta(4));
	await fire("message_end", end({ input: 1000, output: 100, cacheRead: 0, cacheWrite: 0 }));

	ok("B batched final ~2 tok/s (not 198)", () => {
		const m = /([\d.]+) tok\/s TTFT:/.exec(lastTokStatus(statuses));
		assert.ok(m, `no rate: ${lastTokStatus(statuses)}`);
		const v = Number(m![1]);
		assert.ok(v < 5, `rate ${v}`);
	});
}

async function testC() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await fire("message_update", textDelta(4));
	await fire("message_end", end({ input: 10, output: 5, cacheRead: 0, cacheWrite: 0 }));
	ok("C single-chunk output -> rate unavailable", () => {
		assert.match(statuses[statuses.length - 1]!, /rate unavailable/);
	});
}

async function testD() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await fire("message_update", textDelta(1));
	await fire("message_end", end({ input: 10, output: 0, cacheRead: 0, cacheWrite: 0 }));
	ok("D explicit zero output -> '0 tok'", () => {
		assert.match(statuses[statuses.length - 1]!, /  0 tok/);
	});
}

async function testE() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await fire("message_update", textDelta(1));
	await fire("message_end", end({ input: 10, output: 5, cacheRead: 0, cacheWrite: 0 }, "aborted"));
	ok("E aborted -> 'aborted' status", () => {
		assert.match(statuses[statuses.length - 1]!, /aborted/);
	});
}

async function testF() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await fire("message_update", textDelta(1));
	await fire("message_end", end({ input: 10, output: 5, cacheRead: 0, cacheWrite: 0 }, "error"));
	ok("F error -> 'error' status", () => {
		assert.match(statuses[statuses.length - 1]!, /error/);
	});
}

async function testG() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await sleep(250);
	for (let i = 1; i <= 60; i++) {
		await fire("message_update", textDelta(2, i));
		await sleep(15);
	}
	await fire("message_end", end({ input: 500, output: 60, cacheRead: 0, cacheWrite: 0 }));
	ok("G exact-usage provider live has no '~'", () => {
		const live = lastTokStatus(statuses);
		assert.match(live, /  60 tok/);
		assert.doesNotMatch(live, /~/);
	});
}

async function testH() {
	const dir = mkdtempSync(join(tmpdir(), "ltps-"));
	try {
		const first = harness("tui", dir);
		await startTurn(first.fire);
		await sleep(250);
		for (let i = 0; i < 60; i++) {
			await first.fire("message_update", textDelta(2));
			await sleep(15);
		}
		await first.fire("message_end", end({ input: 500, output: 60, cacheRead: 0, cacheWrite: 0 }));
		ok("H calibrates to observed ratio 2 (2 chars/token)", () => {
			const state = JSON.parse(readFileSync(first.statePath, "utf8"));
			const bucket = state.models["test/m1"].other;
			assert.ok(Math.abs(bucket.ratio - 2) < 0.01, `ratio ${bucket.ratio}`);
			assert.ok(bucket.samples >= 1);
		});

		// A fresh instance in the same dir must load the learned ratio.
		const fresh = harness("tui", dir);
		await startTurn(fresh.fire);
		await sleep(250);
		for (let i = 0; i < 60; i++) {
			await fresh.fire("message_update", textDelta(2));
			await sleep(15);
		}
		const freshLive = lastTokStatus(fresh.statuses);
		ok("H reloaded instance uses persisted ratio (~60 tok)", () => {
			const m = /\s{2,}~?([\d.]+) tok/.exec(freshLive);
			assert.ok(m, `live: ${freshLive}`);
			const v = Number(m![1]);
			assert.ok(v > 40 && v <= 60, `tokens ${v} in ${freshLive}`);
		});
	} finally {
		rmSync(dir, { recursive: true, force: true });
	}
}

async function testI() {
	const dir = mkdtempSync(join(tmpdir(), "ltps-"));
	try {
		const { fire } = harness("tui", dir);
		await startTurn(fire);
		await sleep(250);
		for (let i = 0; i < 30; i++) {
			await fire("message_update", thinkDelta(1)); // 1 char/token
			await sleep(10);
		}
		for (let i = 0; i < 30; i++) {
			await fire("message_update", textDelta(4)); // 4 chars/token
			await sleep(10);
		}
		await fire("message_end", end({ input: 10, output: 60, reasoning: 30, cacheRead: 0, cacheWrite: 0 }));
		ok("I per-content calibration stores distinct thinking/other ratios", () => {
			const state = JSON.parse(readFileSync(join(dir, "live-throughput-state.json"), "utf8"));
			const bucket = state.models["test/m1"];
			assert.ok(Math.abs(bucket.thinking.ratio - 1) < 0.01, `thinking ${bucket.thinking.ratio}`);
			assert.ok(Math.abs(bucket.other.ratio - 4) < 0.01, `other ${bucket.other.ratio}`);
		});
	} finally {
		rmSync(dir, { recursive: true, force: true });
	}
}

async function testJ() {
	const { fire, statuses } = harness();
	await fire("message_start", assistantStart()); // no before_provider_request
	await fire("message_update", textDelta(4));
	await fire("message_end", end({ input: 10, output: 5, cacheRead: 0, cacheWrite: 0 }));
	ok("J missing request hook -> TTFT flagged approximate", () => {
		assert.match(statuses[statuses.length - 1]!, /~TTFT:/);
	});
}

async function testK() {
	const dir = mkdtempSync(join(tmpdir(), "ltps-"));
	writeFileSync(
		join(dir, "live-throughput-config.json"),
		JSON.stringify({ idleClearMs: 50, updateIntervalMs: 0, minLiveWindowMs: 0 }),
	);
	try {
		const { fire, statuses } = harness("tui", dir);
		await startTurn(fire);
		await fire("message_update", textDelta(4));
		await fire("message_end", end({ input: 10, output: 5, cacheRead: 0, cacheWrite: 0 }));
		await sleep(120);
		ok("K idleClearMs config clears status after idle", () => {
			assert.equal(statuses[statuses.length - 1], undefined);
		});
	} finally {
		rmSync(dir, { recursive: true, force: true });
	}
}

async function testL() {
	const dir = mkdtempSync(join(tmpdir(), "ltps-"));
	writeFileSync(
		join(dir, "live-throughput-config.json"),
		JSON.stringify({ statusModes: ["tui"] }),
	);
	try {
		const { fire, statuses } = harness("rpc", dir);
		await startTurn(fire);
		await fire("message_update", textDelta(4));
		ok("L statusModes config suppresses RPC", () => {
			assert.equal(statuses.length, 0);
		});
	} finally {
		rmSync(dir, { recursive: true, force: true });
	}
}

async function testM() {
	const { fire, statuses } = harness("rpc");
	await startTurn(fire);
	await fire("message_update", textDelta(4));
	ok("M RPC mode receives status by default", () => {
		assert.ok(
			statuses.length > 0 &&
				statuses.some((s) => s?.includes("tok/s") || s?.includes("waiting")),
		);
	});
}

async function testN() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await fire("message_end", end({ input: 12345, output: 5, cacheRead: 0, cacheWrite: 0 }));
	ok("N large prompt count is formatted (12.3k)", () => {
		assert.match(statuses[statuses.length - 1]!, /Prompt: 12\.3k tok/);
	});
}

async function testO() {
	ok("O formatDuration scales s/m/h/d", () => {
		const cases: Array<[number, string]> = [
			[0, "0s"],
			[30_000, "30s"],
			[59_000, "59s"],
			[60_000, "1m 0s"],
			[90_000, "1m 30s"],
			[59 * 60_000, "59m 0s"],
			[3_600_000, "1h 0m"],
			[3_661_000, "1h 1m"],
			[86_399_000, "23h 59m"],
			[86_400_000, "1d 0h"],
			[90_000_000, "1d 1h"],
		];
		for (const [ms, expected] of cases) {
			assert.equal(formatDuration(ms), expected, `${ms}ms -> ${formatDuration(ms)}`);
		}
	});
}

async function testP() {
	const { fire, statuses } = harness();
	await fire("agent_start", {});
	await startTurn(fire);
	await sleep(250);
	for (let i = 0; i < 20; i++) {
		await fire("message_update", textDelta(4));
		await sleep(15);
	}
	await fire("message_end", end({ input: 500, output: 20, cacheRead: 0, cacheWrite: 0 }));
	await fire("agent_settled", {});
	ok("P agent_settled appends 'took Xs' to the final status", () => {
		const last = statuses[statuses.length - 1]!;
		assert.match(last, /took \d+s$/);
		assert.match(last, /tok\/s/);
	});
}

async function testQ() {
	const dir = mkdtempSync(join(tmpdir(), "ltps-"));
	writeFileSync(join(dir, "live-throughput-config.json"), JSON.stringify({ showElapsed: false }));
	try {
		const { fire, statuses } = harness("tui", dir);
		await fire("agent_start", {});
		await startTurn(fire);
		await fire("message_update", textDelta(4));
		await fire("message_end", end({ input: 10, output: 5, cacheRead: 0, cacheWrite: 0 }));
		await fire("agent_settled", {});
		ok("Q showElapsed:false suppresses 'took'", () => {
			assert.doesNotMatch(statuses[statuses.length - 1]!, /took/);
		});
	} finally {
		rmSync(dir, { recursive: true, force: true });
	}
}

async function testR() {
	ok("R parsePrefillMetrics sums local_compute + ttft across engines", () => {
		const text = [
			"# HELP vllm:prompt_tokens_by_source_total ...",
			'vllm:prompt_tokens_by_source_total{engine="0",source="local_compute"} 1000',
			'vllm:prompt_tokens_by_source_total{engine="0",source="local_cache_hit"} 999999',
			'vllm:prompt_tokens_by_source_total{engine="1",source="local_compute"} 200',
			'vllm:time_to_first_token_seconds_sum{engine="0"} 10.5',
			'vllm:time_to_first_token_seconds_sum{engine="1"} 12.5',
			'vllm:time_to_first_token_seconds_count{engine="0"} 1',
			'vllm:time_to_first_token_seconds_count{engine="1"} 2',
		].join("\n");
		const snap = parsePrefillMetrics(text);
		assert.equal(snap.localCompute, 1200);
		assert.equal(snap.ttftSum, 23);
		assert.equal(snap.ttftCount, 3);
	});
}

async function testS() {
	const dir = mkdtempSync(join(tmpdir(), "ltps-"));
	writeFileSync(
		join(dir, "live-throughput-config.json"),
		JSON.stringify({ metricsUrls: { test: "http://metrics.local" }, metricsTimeoutMs: 200 }),
	);
	const origFetch = globalThis.fetch;
	const snapshots = [
		'vllm:prompt_tokens_by_source_total{source="local_compute"} 1000\nvllm:time_to_first_token_seconds_sum 10\nvllm:time_to_first_token_seconds_count 1\n',
		'vllm:prompt_tokens_by_source_total{source="local_compute"} 2600\nvllm:time_to_first_token_seconds_sum 12\nvllm:time_to_first_token_seconds_count 2\n',
	];
	let call = 0;
	(globalThis as any).fetch = async () => ({
		ok: true,
		text: async () => snapshots[Math.min(call++, snapshots.length - 1)],
	});
	try {
		const { fire, statuses } = harness("tui", dir);
		await startTurn(fire); // before_provider_request snapshots #0
		await fire("message_update", textDelta(4));
		await fire("message_end", end({ input: 100, output: 5, cacheRead: 0, cacheWrite: 0 })); // samples #1
		ok("S derives prefill TPS from server metrics (1600 tok / 2s = 800)", () => {
			assert.match(statuses[statuses.length - 1]!, /800 t\/s/);
		});
	} finally {
		globalThis.fetch = origFetch;
		rmSync(dir, { recursive: true, force: true });
	}
}

async function testT() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	// Big batched first chunk: ~100 estimated tokens at seed ratio 4.
	await fire("message_update", textDelta(400));
	await sleep(300); // pass minLiveWindowMs
	// Only ~1 token actually decoded inside the window.
	await fire("message_update", textDelta(4));
	ok("T live fencepost excludes chunk 1 (no inflation)", () => {
		const live = lastTokStatus(statuses);
		const m = /([\d.]+) tok\/s/.exec(live);
		assert.ok(m, `no rate: ${live}`);
		const v = Number(m![1]);
		assert.ok(v < 20, `rate ${v} (chunk 1 must be excluded)`);
	});
}

async function testU() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	await fire("message_update", textDelta(4));
	await fire("message_end", end({ input: 10, output: 5, cacheRead: 0, cacheWrite: 0 }));
	await fire("session_shutdown", {});
	ok("U session_shutdown clears the status line", () => {
		assert.equal(statuses[statuses.length - 1], undefined);
	});
}

async function testV() {
	const { fire, statuses } = harness();
	await fire("agent_start", {});
	await startTurn(fire);
	await fire("message_update", textDelta(4));
	await sleep(250);
	await fire("message_update", textDelta(4));
	await fire("message_end", end({ input: 100, output: 5, cacheRead: 0, cacheWrite: 0 }));
	await fire("turn_end", {});
	const afterTurnEnd = statuses[statuses.length - 1];
	await fire("agent_settled", {});
	const finalStatus = statuses[statuses.length - 1]!;
	ok("V turn_end clears footer; agent_settled restores final line", () => {
		assert.equal(afterTurnEnd, undefined);
		assert.match(finalStatus, /tok\/s/);
		assert.match(finalStatus, /took/);
	});
}

async function testW() {
	const dir = mkdtempSync(join(tmpdir(), "ltps-"));
	writeFileSync(
		join(dir, "live-throughput-config.json"),
		JSON.stringify({ showGenerationTps: true }),
	);
	try {
		const { fire, statuses } = harness("tui", dir);
		await startTurn(fire);
		await fire("message_update", textDelta(4));
		await fire("message_end", end({ input: 100, output: 50, cacheRead: 0, cacheWrite: 0 }));
		ok("W generation throughput = output / (request -> message_end)", () => {
			assert.match(statuses[statuses.length - 1]!, /gen: [\d.]+ tok\/s/);
		});
	} finally {
		rmSync(dir, { recursive: true, force: true });
	}
}

async function testX() {
	const { fire, statuses } = harness();
	await startTurn(fire);
	// Big first chunk (~100 est. tokens), then ~1 est. token inside the window.
	await fire("message_update", textDelta(400));
	await sleep(300);
	await fire("message_update", textDelta(4));
	// No provider output usage -> final uses the estimated fallback path.
	await fire("message_end", end({ input: 10, cacheRead: 0, cacheWrite: 0 }));
	ok("X estimated final rate also excludes chunk 1", () => {
		const final = lastTokStatus(statuses);
		const m = /([\d.]+) tok\/s/.exec(final);
		assert.ok(m, `no rate: ${final}`);
		const v = Number(m![1]);
		assert.ok(v < 20, `rate ${v} (chunk 1 must be excluded)`);
	});
}

async function main() {
	console.log("live-throughput-status tests\n");
	await testA();
	await testB();
	await testC();
	await testD();
	await testE();
	await testF();
	await testG();
	await testH();
	await testI();
	await testJ();
	await testK();
	await testL();
	await testM();
	await testN();
	await testO();
	await testP();
	await testQ();
	await testR();
	await testS();
	await testT();
	await testU();
	await testV();
	await testW();
	await testX();
	console.log(`\n${passed} passed${process.exitCode ? ", some failed" : ""}`);
}

main();

live-throughput-status.ts

import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { homedir } from "node:os";
import { dirname, join } from "node:path";

/**
 * live-throughput-status — model-neutral TTFT / decode-TPS footer.
 *
 * Live decode TPS is exact when a provider reports cumulative `usage.output`
 * on intermediate chunks (Gemini-style / non-standard OpenAI servers). Otherwise
 * it falls back to a chars/token estimate calibrated per model and per content
 * kind (thinking vs text/tool-call), learned across messages via EWMA.
 *
 * Provider matrix (live display):
 *   - OpenAI-completions (standard) : estimated (usage arrives in final chunk)
 *   - Anthropic-messages            : estimated
 *   - OpenAI Responses              : estimated
 *   - Gemini / cumulative-usage OAI : exact
 *
 * The final decode rate uses the provider's reported output-token count and the
 * client-observed interval from the first streamed chunk to the last, excluding
 * the tokens delivered in the first chunk (which define the window start).
 */

const STATUS_KEY = "live-throughput";

interface RatioState {
	ratio: number;
	samples: number;
}

interface ModelState {
	thinking: RatioState;
	other: RatioState;
}

interface PersistedState {
	version: number;
	models: Record<string, ModelState>;
}

interface Config {
	charsPerTokenSeed: number;
	ewmaAlpha: number;
	ratioMin: number;
	ratioMax: number;
	updateIntervalMs: number;
	minLiveWindowMs: number;
	idleClearMs: number;
	showElapsed: boolean;
	clearOnTurnEnd: boolean;
	showGenerationTps: boolean;
	metricsUrls: Record<string, string>;
	metricsTimeoutMs: number;
	statusModes: string[];
}

const DEFAULT_CONFIG: Config = {
	charsPerTokenSeed: 4,
	ewmaAlpha: 0.3,
	ratioMin: 1,
	ratioMax: 16,
	updateIntervalMs: 200,
	minLiveWindowMs: 200,
	idleClearMs: 0, // 0 = keep the final status until the next turn
	showElapsed: true, // append 'took X' at agent_settled
	clearOnTurnEnd: true, // clear the footer while idle between turns
	showGenerationTps: false, // show output/(request->message_end): gen throughput
	// provider id -> Prometheus /metrics URL. Empty by default: prefill TPS is
	// only shown when the backend exposes server-side prefill metrics.
	metricsUrls: {},
	metricsTimeoutMs: 500,
	statusModes: ["tui", "rpc"],
};

function isFiniteNumber(value: unknown): value is number {
	return typeof value === "number" && Number.isFinite(value);
}

function clampNumber(value: number, min: number, max: number): number {
	return Math.min(max, Math.max(min, value));
}

function positiveNumber(value: unknown): number | undefined {
	return isFiniteNumber(value) && value > 0 ? value : undefined;
}

function loadConfig(path: string): Config {
	const config: Config = { ...DEFAULT_CONFIG };
	if (!existsSync(path)) return config;
	try {
		const raw = JSON.parse(readFileSync(path, "utf8")) as Record<string, unknown>;
		if (isFiniteNumber(raw.charsPerTokenSeed) && raw.charsPerTokenSeed > 0) {
			config.charsPerTokenSeed = raw.charsPerTokenSeed;
		}
		if (isFiniteNumber(raw.ewmaAlpha)) {
			config.ewmaAlpha = clampNumber(raw.ewmaAlpha, 0, 1);
		}
		if (isFiniteNumber(raw.ratioMin) && raw.ratioMin > 0) config.ratioMin = raw.ratioMin;
		if (isFiniteNumber(raw.ratioMax) && raw.ratioMax > config.ratioMin) {
			config.ratioMax = raw.ratioMax;
		}
		if (isFiniteNumber(raw.updateIntervalMs) && raw.updateIntervalMs >= 0) {
			config.updateIntervalMs = raw.updateIntervalMs;
		}
		if (isFiniteNumber(raw.minLiveWindowMs) && raw.minLiveWindowMs >= 0) {
			config.minLiveWindowMs = raw.minLiveWindowMs;
		}
		if (isFiniteNumber(raw.idleClearMs) && raw.idleClearMs >= 0) {
			config.idleClearMs = raw.idleClearMs;
		}
		if (typeof raw.showElapsed === "boolean") {
			config.showElapsed = raw.showElapsed;
		}
		if (typeof raw.clearOnTurnEnd === "boolean") {
			config.clearOnTurnEnd = raw.clearOnTurnEnd;
		}
		if (typeof raw.showGenerationTps === "boolean") {
			config.showGenerationTps = raw.showGenerationTps;
		}
		if (raw.metricsUrls && typeof raw.metricsUrls === "object") {
			const urls: Record<string, string> = {};
			for (const [key, value] of Object.entries(raw.metricsUrls as Record<string, unknown>)) {
				if (typeof value === "string" && value.length > 0) urls[key] = value;
			}
			config.metricsUrls = urls;
		}
		if (isFiniteNumber(raw.metricsTimeoutMs) && raw.metricsTimeoutMs > 0) {
			config.metricsTimeoutMs = raw.metricsTimeoutMs;
		}
		if (Array.isArray(raw.statusModes)) {
			const modes = raw.statusModes.filter((m): m is string => typeof m === "string");
			if (modes.length > 0) config.statusModes = modes;
		}
	} catch {
		// Ignore malformed config and keep defaults.
	}
	return config;
}

function loadState(path: string): PersistedState {
	const empty: PersistedState = { version: 1, models: {} };
	if (!existsSync(path)) return empty;
	try {
		const raw = JSON.parse(readFileSync(path, "utf8")) as Partial<PersistedState>;
		const models: Record<string, ModelState> = {};
		if (raw.models && typeof raw.models === "object") {
			for (const [key, value] of Object.entries(raw.models)) {
				const candidate = value as Partial<ModelState>;
				const thinking = candidate?.thinking;
				const other = candidate?.other;
				if (
					isFiniteNumber(thinking?.ratio) &&
					isFiniteNumber(thinking?.samples) &&
					isFiniteNumber(other?.ratio) &&
					isFiniteNumber(other?.samples)
				) {
					models[key] = {
						thinking: { ratio: thinking.ratio, samples: thinking.samples },
						other: { ratio: other.ratio, samples: other.samples },
					};
				}
			}
		}
		return { version: 1, models };
	} catch {
		return empty;
	}
}

function saveState(path: string, state: PersistedState): void {
	try {
		mkdirSync(dirname(path), { recursive: true });
		writeFileSync(path, JSON.stringify(state, null, 2), "utf8");
	} catch {
		// Persistence is best-effort; never break the session over it.
	}
}

function deltaInfo(event: unknown): { kind: "thinking" | "other"; chars: number } | undefined {
	if (!event || typeof event !== "object") return undefined;
	const streamEvent = event as { type?: string; delta?: unknown };
	if (typeof streamEvent.delta !== "string" || streamEvent.delta.length === 0) {
		return undefined;
	}
	if (streamEvent.type === "thinking_delta") {
		return { kind: "thinking", chars: streamEvent.delta.length };
	}
	if (streamEvent.type === "text_delta" || streamEvent.type === "toolcall_delta") {
		return { kind: "other", chars: streamEvent.delta.length };
	}
	return undefined;
}

// Some providers (Gemini-style, non-standard OpenAI servers) send a cumulative
// usage object on intermediate chunks. When present, `partial.usage.output`
// gives the exact token count so far, which beats any chars/token estimate.
function exactOutputTokens(event: unknown): number {
	if (!event || typeof event !== "object") return 0;
	const partial = (event as { partial?: { usage?: { output?: unknown } } }).partial;
	const output = partial?.usage?.output;
	return isFiniteNumber(output) && output > 0 ? output : 0;
}

function seconds(milliseconds: number): number {
	return Math.max(0, milliseconds) / 1000;
}

function rate(value: number, durationSeconds: number): string {
	return (value / Math.max(0.001, durationSeconds)).toFixed(1);
}

function formatTokens(value: number): string {
	if (value >= 1_000_000) return `${(value / 1_000_000).toFixed(1)}M`;
	if (value >= 1_000) return `${(value / 1_000).toFixed(1)}k`;
	return String(value);
}

/**
 * Scale a duration for display: seconds below a minute, then minutes, hours,
 * and days. Examples: 30s, 1m 30s, 2h 5m, 1d 3h.
 */
export function formatDuration(milliseconds: number): string {
	const totalSeconds = Math.max(0, Math.round(milliseconds / 1000));
	if (totalSeconds < 60) return `${totalSeconds}s`;
	const totalMinutes = Math.floor(totalSeconds / 60);
	if (totalMinutes < 60) return `${totalMinutes}m ${totalSeconds % 60}s`;
	const totalHours = Math.floor(totalMinutes / 60);
	if (totalHours < 24) return `${totalHours}h ${totalMinutes % 60}m`;
	const days = Math.floor(totalHours / 24);
	return `${days}d ${totalHours % 24}h`;
}

interface PrefillSnapshot {
	/** vLLM tokens actually prefilled (excludes prefix-cache hits). */
	localCompute: number;
	/** Cumulative server time-to-first-token (seconds). */
	ttftSum: number;
	ttftCount: number;
}

/**
 * Parse the handful of vLLM Prometheus counters needed to derive server-side
 * prefill throughput. Prefix caching means `usage.input` (full prompt) hugely
 * overstates real prefill work, so we use `local_compute` tokens divided by the
 * server's TTFT, sampled before/after a request.
 */
export function parsePrefillMetrics(text: string): PrefillSnapshot {
	let localCompute = 0;
	let ttftSum = 0;
	let ttftCount = 0;
	for (const rawLine of text.split("\n")) {
		const line = rawLine.trim();
		if (!line || line.startsWith("#")) continue;
		const match =
			/^([a-zA-Z_:][a-zA-Z0-9_:]*)(\{[^}]*\})?\s+([0-9eE+.\-]+)$/.exec(line);
		if (!match) continue;
		const name = match[1];
		const labels = match[2] ?? "";
		const value = Number(match[3]);
		if (!Number.isFinite(value)) continue;
		if (name === "vllm:prompt_tokens_by_source_total") {
			if (/source="local_compute"/.test(labels)) localCompute += value;
		} else if (name === "vllm:time_to_first_token_seconds_sum") {
			ttftSum += value;
		} else if (name === "vllm:time_to_first_token_seconds_count") {
			ttftCount += value;
		}
	}
	return { localCompute, ttftSum, ttftCount };
}

async function fetchPrefillMetrics(
	url: string,
	timeoutMs: number,
): Promise<PrefillSnapshot | undefined> {
	if (typeof fetch !== "function") return undefined;
	const controller = new AbortController();
	const timer = setTimeout(() => controller.abort(), timeoutMs);
	try {
		const response = await fetch(url, { signal: controller.signal });
		if (!response.ok) return undefined;
		return parsePrefillMetrics(await response.text());
	} catch {
		return undefined;
	} finally {
		clearTimeout(timer);
	}
}

export default function (pi: ExtensionAPI) {
	const agentDir =
		process.env.PI_LIVE_THROUGHPUT_DIR || join(homedir(), ".pi", "agent");
	const config = loadConfig(join(agentDir, "live-throughput-config.json"));
	const state = loadState(join(agentDir, "live-throughput-state.json"));
	const statePath = join(agentDir, "live-throughput-state.json");

	// ---- calibration ------------------------------------------------------
	function getModelState(key: string): ModelState {
		let model = state.models[key];
		if (!model) {
			model = {
				thinking: { ratio: config.charsPerTokenSeed, samples: 0 },
				other: { ratio: config.charsPerTokenSeed, samples: 0 },
			};
			state.models[key] = model;
		}
		return model;
	}

	function ratioFor(key: string, kind: "thinking" | "other"): number {
		const bucket = getModelState(key)[kind];
		return bucket.samples > 0 ? bucket.ratio : config.charsPerTokenSeed;
	}

	function calibrateObserved(key: string, kind: "thinking" | "other", observed: number): void {
		if (!isFiniteNumber(observed) || observed < config.ratioMin || observed > config.ratioMax) {
			return;
		}
		const bucket = getModelState(key)[kind];
		bucket.ratio =
			bucket.samples === 0
				? observed
				: bucket.ratio * (1 - config.ewmaAlpha) + observed * config.ewmaAlpha;
		bucket.samples++;
	}

	function calibrate(key: string, kind: "thinking" | "other", chars: number, tokens: number): void {
		if (tokens <= 0 || chars <= 0) return;
		calibrateObserved(key, kind, chars / tokens);
	}

	function calibrateFromUsage(
		key: string,
		outputTokens: number,
		reasoning: unknown,
	): void {
		const reasoningTokens = positiveNumber(reasoning);
		if (reasoningTokens !== undefined && streamedThinkingChars > 0) {
			calibrate(key, "thinking", streamedThinkingChars, reasoningTokens);
		}
		const otherTokens = outputTokens - (reasoningTokens ?? 0);
		if (otherTokens > 0 && streamedOtherChars > 0) {
			calibrate(key, "other", streamedOtherChars, otherTokens);
		}
		// No reasoning split reported: fall back to a blended sample so both
		// live buckets still learn from the content that actually streamed.
		if (reasoningTokens === undefined && outputTokens > 0) {
			const totalChars = streamedThinkingChars + streamedOtherChars;
			if (totalChars > 0) {
				const blended = totalChars / outputTokens;
				if (streamedThinkingChars > 0) calibrateObserved(key, "thinking", blended);
				if (streamedOtherChars > 0) calibrateObserved(key, "other", blended);
			}
		}
	}

	function estimateTokens(key: string): number {
		return (
			streamedThinkingChars / ratioFor(key, "thinking") +
			streamedOtherChars / ratioFor(key, "other")
		);
	}

	// ---- status / lifecycle ----------------------------------------------
	function setStatus(ctx: ExtensionContext, text?: string): void {
		if (!config.statusModes.includes(ctx.mode)) return;
		lastStatusText = text ?? "";
		ctx.ui.setStatus(
			STATUS_KEY,
			text === undefined ? undefined : ctx.ui.theme.fg("accent", `⚡ ${text}`),
		);
	}

	let idleTimer: ReturnType<typeof setTimeout> | undefined;
	function clearIdleTimer(): void {
		if (idleTimer !== undefined) {
			clearTimeout(idleTimer);
			idleTimer = undefined;
		}
	}
	function scheduleIdleClear(ctx: ExtensionContext): void {
		clearIdleTimer();
		if (config.idleClearMs > 0) {
			idleTimer = setTimeout(() => {
				idleTimer = undefined;
				setStatus(ctx, undefined);
			}, config.idleClearMs);
		}
	}

	// ---- per-turn state ---------------------------------------------------
	let modelKey = "default";
	let currentProvider = "";
	let lastStatusText = "";
	let agentStartedAt: number | undefined;
	let prefillBefore: Promise<PrefillSnapshot | undefined> | undefined;
	let requestStartedAt: number | undefined;
	let requestHookSeen = false;
	let ttftEstimated = false;
	let firstOutputAt: number | undefined;
	let lastOutputAt: number | undefined;
	let ttftSeconds: number | undefined;
	let streamedThinkingChars = 0;
	let streamedOtherChars = 0;
	let firstChunkTokens = 0;
	let lastExactOutput = 0;
	let sawExactStream = false;
	let lastDisplayAt = 0;

	function resetCounters(): void {
		ttftEstimated = false;
		firstOutputAt = undefined;
		lastOutputAt = undefined;
		ttftSeconds = undefined;
		streamedThinkingChars = 0;
		streamedOtherChars = 0;
		firstChunkTokens = 0;
		lastExactOutput = 0;
		sawExactStream = false;
		lastDisplayAt = 0;
	}

	function resetTurn(): void {
		clearIdleTimer();
		resetCounters();
		prefillBefore = undefined;
		requestStartedAt = undefined;
		requestHookSeen = false;
	}

	function ttftPart(): string | undefined {
		if (ttftSeconds === undefined) return undefined;
		return `${ttftEstimated ? "~" : ""}TTFT: ${ttftSeconds.toFixed(2)}s`;
	}

	function warmingStatus(): string {
		const ttft = ttftPart();
		return ttft ? `${ttft}   warming up…` : "TTFT: waiting";
	}

	// ---- events -----------------------------------------------------------
	pi.on("session_start", async (_event, ctx) => {
		resetTurn();
		setStatus(ctx, "TTFT: waiting");
	});

	pi.on("model_select", async (_event, ctx) => {
		resetTurn();
		setStatus(ctx, undefined);
	});

	pi.on("session_shutdown", async (_event, ctx) => {
		clearIdleTimer();
		saveState(statePath, state);
		// Drop the status so it does not linger after the session ends.
		setStatus(ctx, undefined);
	});

	// Whole-request wall clock: one agent run per user prompt, including every
	// tool-calling turn and auto-retry, until Pi settles. Displayed once, at
	// agent_settled — never at intermediate turns.
	pi.on("agent_start", async (_event, ctx) => {
		// Keep the earliest start so auto-retries stay inside one settle window.
		if (agentStartedAt === undefined) agentStartedAt = performance.now();
		lastStatusText = "";
		setStatus(ctx, undefined);
	});

	pi.on("agent_settled", async (_event, ctx) => {
		if (agentStartedAt === undefined) return;
		const elapsed = performance.now() - agentStartedAt;
		agentStartedAt = undefined;
		if (!config.showElapsed) return;
		const base = lastStatusText;
		setStatus(
			ctx,
			base ? `${base}   took ${formatDuration(elapsed)}` : `took ${formatDuration(elapsed)}`,
		);
		scheduleIdleClear(ctx);
	});

	// Clear the footer while the agent is idle between turns/tool calls instead
	// of leaving a stale status frozen on screen. lastStatusText is intentionally
	// kept so agent_settled can rebuild the complete final line (with 'took X').
	pi.on("turn_end", async (_event, ctx) => {
		if (!config.clearOnTurnEnd) return;
		// Only clear while an agent run is still active. A late turn_end after
		// agent_settled must not wipe the final line.
		if (agentStartedAt === undefined) return;
		clearIdleTimer();
		// setStatus() rewrites lastStatusText, so preserve it across the clear.
		const preserved = lastStatusText;
		setStatus(ctx, undefined);
		lastStatusText = preserved;
	});

	// Fired right before the provider request is sent. When a metrics URL is
	// configured for this provider, snapshot prefill counters BEFORE the request
	// so we can diff them against a post-request sample.
	pi.on("before_provider_request", async (_event, ctx) => {
		const provider = ctx.model?.provider ?? currentProvider;
		const url = provider ? config.metricsUrls[provider] : undefined;
		if (url) {
			prefillBefore = fetchPrefillMetrics(url, config.metricsTimeoutMs);
			await prefillBefore;
		} else {
			prefillBefore = undefined;
		}
		requestStartedAt = performance.now();
		requestHookSeen = true;
	});

	pi.on("message_start", async (event, ctx) => {
		if (event.message.role !== "assistant") return;
		// Timestamps from the just-fired before_provider_request are retained;
		// only per-output counters are reset here.
		resetCounters();
		currentProvider = event.message.provider;
		modelKey = `${event.message.provider}/${event.message.model}`;
		if (requestStartedAt === undefined) {
			// Fallback for a provider that does not emit the request hook.
			requestStartedAt = performance.now();
			requestHookSeen = false;
		}
		setStatus(ctx, "TTFT: waiting");
	});

	pi.on("message_update", async (event, ctx) => {
		if (event.message.role !== "assistant") return;
		const info = deltaInfo(event.assistantMessageEvent);
		if (!info) return;

		const now = performance.now();
		const exact = exactOutputTokens(event.assistantMessageEvent);

		if (firstOutputAt === undefined) {
			firstOutputAt = now;
			ttftSeconds = seconds(now - (requestStartedAt ?? now));
			ttftEstimated = !requestHookSeen || requestStartedAt === undefined;
			firstChunkTokens =
				exact > 0 ? exact : Math.max(1, info.chars / ratioFor(modelKey, info.kind));
			lastDisplayAt = now;
			setStatus(ctx, warmingStatus());
		}

		lastOutputAt = now;
		if (info.kind === "thinking") streamedThinkingChars += info.chars;
		else streamedOtherChars += info.chars;

		if (exact > lastExactOutput) {
			const deltaTokens = exact - lastExactOutput;
			lastExactOutput = exact;
			if (deltaTokens > 0) {
				calibrate(modelKey, info.kind, info.chars, deltaTokens);
				sawExactStream = true;
			}
		}

		if (now - lastDisplayAt < config.updateIntervalMs) return;
		lastDisplayAt = now;

		const decodeSeconds = seconds(now - firstOutputAt);
		if (decodeSeconds * 1000 < config.minLiveWindowMs) {
			setStatus(ctx, warmingStatus());
			return;
		}

		const exactLive = lastExactOutput > 0;
		// Fencepost consistency with message_end: the live window starts at the
		// arrival of the first chunk, so the first chunk's tokens must be excluded
		// from the numerator (otherwise the initial rate is inflated, badly so when
		// a gateway batches a large first chunk).
		const grossTokens = exactLive ? lastExactOutput : estimateTokens(modelKey);
		const liveTokens = Math.max(0, grossTokens - firstChunkTokens);
		if (liveTokens <= 0) {
			setStatus(ctx, warmingStatus());
			return;
		}
		const approx = exactLive ? "" : "~";
		const ttft = ttftPart();
		const rateText = `${rate(liveTokens, decodeSeconds)} tok/s`;
		const tokenText = `${approx}${formatTokens(Math.round(liveTokens))} tok`;
		setStatus(ctx, ttft ? `${rateText} ${ttft}   ${tokenText}` : `${rateText}   ${tokenText}`);
	});

	pi.on("message_end", async (event, ctx) => {
		if (event.message.role !== "assistant") return;

		const stopReason = event.message.stopReason;
		if (stopReason === "aborted" || stopReason === "error") {
			setStatus(ctx, stopReason === "aborted" ? "aborted" : "error");
			resetTurn();
			return;
		}

		const usage = event.message.usage;
		const rawOutput = isFiniteNumber(usage?.output) ? usage.output : undefined;
		const outputTokens = positiveNumber(rawOutput);
		const rawInput = isFiniteNumber(usage?.input) ? usage.input : 0;
		const rawCacheWrite = isFiniteNumber(usage?.cacheWrite) ? usage.cacheWrite : 0;

		// Learn ratios from the completed message unless the provider already
		// gave us exact mid-stream samples (which were calibrated incrementally).
		if (!sawExactStream && outputTokens !== undefined) {
			calibrateFromUsage(modelKey, outputTokens, usage?.reasoning);
		}
		// Never attribute more first-chunk tokens than were produced.
		if (outputTokens !== undefined) {
			firstChunkTokens = Math.min(firstChunkTokens, outputTokens);
		}
		saveState(statePath, state);

		// Server-side prefill TPS (vLLM): diff local_compute tokens and TTFT
		// counters sampled before and after this request. Prefix caching makes
		// usage.input (the full prompt) useless for this, so the server counters
		// are the only honest per-request prefill rate available.
		let prefillTps: number | undefined;
		if (prefillBefore) {
			const url = config.metricsUrls[currentProvider];
			const before = await prefillBefore;
			prefillBefore = undefined;
			if (before && url) {
				const after = await fetchPrefillMetrics(url, config.metricsTimeoutMs);
				if (after) {
					const prefilledTokens = after.localCompute - before.localCompute;
					const prefillSeconds = after.ttftSum - before.ttftSum;
					if (prefilledTokens > 0 && prefillSeconds > 0) {
						prefillTps = prefilledTokens / prefillSeconds;
					}
				}
			}
		}
		const prefillPart =
			prefillTps !== undefined ? `   ${formatTokens(Math.round(prefillTps))} t/s` : "";

		// Generation throughput (arhen): output / (request sent -> message_end).
		// Spans queue + prefill + TTFT + generation, but excludes tool execution
		// (unlike 'took X', which is agent_start -> agent_settled).
		let generationPart = "";
		if (
			config.showGenerationTps &&
			outputTokens !== undefined &&
			requestStartedAt !== undefined
		) {
			const genSeconds = seconds(performance.now() - requestStartedAt);
			if (genSeconds > 0) {
				generationPart = `   gen: ${rate(outputTokens, genSeconds)} tok/s`;
			}
		}

		const decodeSeconds =
			firstOutputAt !== undefined && lastOutputAt !== undefined
				? seconds(lastOutputAt - firstOutputAt)
				: undefined;

		const ttft = ttftPart();
		// Prompt token count is factual; a tokens/TTFT rate is NOT server prefill
		// throughput (TTFT includes network, queue, scheduling, first-token decode),
		// so we deliberately show the raw count instead of a misleading ratio.
		const promptTokens = rawInput + rawCacheWrite;
		const promptPart = promptTokens > 0 ? `   Prompt: ${formatTokens(promptTokens)} tok` : "";

		const decodeTokens =
			outputTokens !== undefined ? Math.max(0, outputTokens - firstChunkTokens) : undefined;

		let rateText: string | undefined;
		let tokenText: string;
		let rateUnavailable = false;

		if (rawOutput === 0) {
			tokenText = "0 tok";
		} else if (
			decodeTokens !== undefined &&
			decodeTokens > 0 &&
			decodeSeconds !== undefined &&
			decodeSeconds > 0
		) {
			rateText = `${rate(decodeTokens, decodeSeconds)} tok/s`;
			tokenText = `${formatTokens(outputTokens!)} tok`;
		} else if (outputTokens !== undefined) {
			tokenText = `${formatTokens(outputTokens)} tok`;
			rateUnavailable = true;
		} else if (
			streamedThinkingChars + streamedOtherChars > 0 &&
			decodeSeconds !== undefined &&
			decodeSeconds > 0
		) {
			// Same fencepost rule as the exact path: the final rate numerator must
			// exclude the first chunk, while the token count still reports the total.
			const estimatedTotal = estimateTokens(modelKey);
			const estimatedDecode = Math.max(0, estimatedTotal - firstChunkTokens);
			tokenText = `~${formatTokens(Math.round(estimatedTotal))} tok`;
			if (estimatedDecode > 0) {
				rateText = `${rate(estimatedDecode, decodeSeconds)} tok/s`;
			}
		} else {
			tokenText = "no output tokens";
		}

		let statusText: string;
		if (rateText) {
			statusText = ttft ? `${rateText} ${ttft}   ${tokenText}` : `${rateText}   ${tokenText}`;
		} else {
			statusText = `${ttft ? `${ttft}   ` : ""}${tokenText}`;
			if (rateUnavailable) statusText += "   rate unavailable";
		}
		if (prefillPart) statusText += prefillPart;
		if (generationPart) statusText += generationPart;
		statusText += promptPart;

		setStatus(ctx, statusText);
		scheduleIdleClear(ctx);
		requestStartedAt = undefined;
		requestHookSeen = false;
	});
}

TPS_PI_EXTENSION.md

Live decode TPS, TTFT & prefill throughput for Pi (v2)

A model-neutral throughput footer for Pi, validated against @earendil-works/pi-coding-agent 0.85.1.

It observes Pi's standard assistant-stream events, so it works with local and hosted models. For real server-side prefill throughput it can optionally read Prometheus metrics (e.g. vLLM /metrics); prefix caching is handled correctly, which client-side usage.input cannot.

What the footer shows

Live, while streaming (estimated until the provider reports token counts):

⚡ 33.8 tok/s TTFT: 0.25s   ~7 tok

Final, with exact usage, optional server prefill, prompt size, and elapsed:

⚡ 62.8 tok/s TTFT: 0.25s   60 tok   800 t/s   Prompt: 1.0k tok   took 1s

Segment	Meaning
62.8 tok/s	Decode rate (client-observed; first chunk excluded)
TTFT: 0.25s	Time from request send to first output delta
~TTFT	TTFT is approximate (request hook did not fire)
60 tok / ~7 tok	Output tokens (exact vs. estimated)
800 t/s	Server-side prefill throughput (only if metricsUrls set)
gen: 12.4 tok/s	Optional: generation throughput incl. queue/prefill/TTFT
Prompt: 1.0k tok	Input + cache-write tokens
took 1s	Whole-request wall clock (agent_start → agent_settled)

Metric definitions

• Decode TPS — rate-first. Exact when a provider reports cumulative usage.output on intermediate chunks (Gemini-style / cumulative-usage OpenAI servers); otherwise a per-model, per-content (thinking vs. text/tool) chars/token ratio learned via EWMA and persisted across /reload. Both live and final rates exclude the first chunk from the numerator, since that chunk defines the window start (fencepost consistency). This is what keeps batched/buffered SSE streams from being overstated.
• TTFT — before_provider_request → first streamed delta.
• Prefill TPS — server-side only. Diffs Prometheus counters vllm:prompt_tokens_by_source_total{source="local_compute"} and vllm:time_to_first_token_seconds_sum across one request: Δlocal_compute ÷ ΔTTFT_sum. Prefix-cache hits are ignored, so this is the honest per-request prefill rate.
• Generation throughput (optional) — output ÷ (request → message_end); includes queue + prefill + TTFT + generation, excludes tool execution.
• took X — agent_start → agent_settled, the whole user request including tool time. Scaled: 30s, 1m 30s, 2h 5m, 1d 3h.
• Prompt — uncached input + cache-write tokens. A tokens/TTFT ratio is intentionally not shown: TTFT contains network, queue, scheduling, and first-token decode, so it is not prefill throughput.

Install

Copy the bundled live-throughput-status.ts into ~/.pi/agent/extensions/:

curl -fsSL https://gist.githubusercontent.com/Anemll/f95a14877862f289e19b12586850eded/raw/live-throughput-status.ts \
  -o ~/.pi/agent/extensions/live-throughput-status.ts

Then run /reload in Pi.

Tests

live-throughput-status.test.ts covers the calibration, fencepost, lifecycle, prefill-parsing, and generation-metric paths (27 assertions). Run it with Bun:

mkdir -p /tmp/ltps && cd /tmp/ltps
BASE=https://gist.githubusercontent.com/Anemll/f95a14877862f289e19b12586850eded/raw
curl -fsSL "$BASE/live-throughput-status.ts"      -o live-throughput-status.ts
curl -fsSL "$BASE/live-throughput-status.test.ts" -o live-throughput-status.test.ts
bun run live-throughput-status.test.ts
# 27 passed

Configuration

Optional ~/.pi/agent/live-throughput-config.json:

{
  "metricsUrls": {
    "gx10-vllm": "http://192.168.1.68:8888/metrics"
  },
  "showGenerationTps": false
}

Field	Default	Description
charsPerTokenSeed	4	Initial chars/token estimate
ewmaAlpha	0.3	Calibration learning rate
ratioMin / ratioMax	1 / 16	Sanity bounds for learned ratios
updateIntervalMs	200	Minimum live repaint interval
minLiveWindowMs	200	Minimum window before showing a live rate
idleClearMs	0	Auto-clear status after idle (0 = off)
showElapsed	true	Append took X at agent_settled
clearOnTurnEnd	true	Clear footer while idle between turns
showGenerationTps	false	Show the generation-throughput segment
metricsUrls	{}	Provider id → Prometheus /metrics URL
metricsTimeoutMs	500	Metrics fetch timeout
statusModes	["tui","rpc"]	Modes where the status is shown

Calibration is persisted to ~/.pi/agent/live-throughput-state.json.

Provider matrix

Provider kind	Live decode	Final decode	Prefill
Standard OpenAI-completions	estimated	exact	only if metricsUrls
Anthropic messages	estimated	exact	—
OpenAI Responses	estimated	exact	—
Gemini / cumulative-usage OAI	exact	exact	only if metricsUrls
vLLM (with /metrics)	estimated/exact	exact	server-side

For vLLM, the required series are:

vllm:prompt_tokens_by_source_total{source="local_compute"}
vllm:prompt_tokens_by_source_total{source="local_cache_hit"}   # ignored
vllm:time_to_first_token_seconds_sum
vllm:time_to_first_token_seconds_count

Limitations

• Client-observed decode TPS measures the stream window, not model decode time. Gateways that batch SSE chunks can still distort the live value; the first-chunk exclusion mitigates it. True decode requires server timing.
• Server-side decode TPS (vLLM vllm:inter_token_latency_seconds_*) is not wired in; only prefill is.
• Compact mode is not included in this revision.
• metricsUrls adds one short GET before each request (up to metricsTimeoutMs); keep the endpoint on the LAN.

Changelog

v2 (this revision)

• Rate-first footer; removed the Decode: label and the misleading Input/TTFT rate.
• Live and final decode numerators exclude the first chunk (fencepost fix).
• Adaptive per-model, per-content chars/token calibration via EWMA, persisted across /reload.
• Exact usage.output path for providers that stream cumulative usage.
• Server-side prefill TPS from Prometheus metrics (prefix-cache aware).
• Optional generation throughput segment (showGenerationTps).
• Whole-request wall clock (took X) with s/m/h/d scaling.
• Lifecycle cleanup on session_shutdown and turn_end (clearOnTurnEnd).
• Robust abort/error/zero-output handling; approximate TTFT flag.
• Large-number formatting (12.3k, 1.2M); configurable statusModes.

v1 (74d80a9)

• Original: chars/4 live estimate, output-1 final numerator, Input/TTFT prompt-rate estimate, fixed Decode: footer.

Proposed Improvements: Live stream fencepost fix, idle status cleanup, and compact mode

Nice work on the pure-decode interval calculation in message_end! A few suggestions and fixes from testing:

1. Live stream fencepost consistency: In message_update, dividing streamedChars by now - firstOutputAt inflates the initial live rate (+25% to +50%) because firstOutputAt marks chunk 1’s arrival, but chunk 1’s characters are counted in the numerator. Excluding chunk 1 from the live numerator matches the fencepost logic in message_end (outputTokens - 1).
2. Lifecycle cleanup (turn_end / session_shutdown): The extension currently lacks turn_end and session_shutdown listeners, leaving the 75-character string or "waiting" text frozen on the status line while the user is idle.
3. Compact mode option: In status bars with multiple segments (model, git branch, path, context %, cost), the 75-character output easily wraps or clips other segments. Adding a compact toggle prevents overflow on dense terminal displays.

@@ -6,2 +6,3 @@
 const UPDATE_INTERVAL_MS = 200;
+const COMPACT_MODE = false;

@@ -52,2 +53,3 @@
 	let streamedChars = 0;
+	let firstChunkChars = 0;
 	let lastDisplayAt = 0;
@@ -58,2 +60,3 @@
 		streamedChars = 0;
+		firstChunkChars = 0;
 		lastDisplayAt = 0;
@@ -82,3 +85,4 @@
 		if (firstOutputAt === undefined) {
 			firstOutputAt = now;
+			firstChunkChars = chars;
 			ttftSeconds = seconds(now - (requestStartedAt ?? now));
 		}
 		lastOutputAt = now;
 		streamedChars += chars;

 		if (now - lastDisplayAt < UPDATE_INTERVAL_MS) return;
 		lastDisplayAt = now;
 		const decodeSeconds = seconds(now - firstOutputAt);
-		const estimatedTokens = streamedChars / CHARS_PER_TOKEN;
+		const liveChars = Math.max(0, streamedChars - firstChunkChars);
+		const estimatedTokens = liveChars / CHARS_PER_TOKEN;

 		if (COMPACT_MODE) {
 			setStatus(ctx, `Decode: ~${rate(estimatedTokens, decodeSeconds)} tok/s`);
 			return;
 		}
 		setStatus(
 			ctx,
 			`TTFT: ${ttftSeconds!.toFixed(2)}s · Decode: ~${rate(estimatedTokens, decodeSeconds)} tok/s · ~${Math.round(estimatedTokens)} tok`,
 		);
 	});

@@ -127,2 +136,8 @@
+		if (COMPACT_MODE && outputTokens && decodeSeconds) {
+			const ttftPart = observedTtft !== undefined ? ` (${observedTtft.toFixed(2)}s TTFT)` : "";
+			setStatus(ctx, `Decode: ${rate(outputTokens - 1, decodeSeconds)} tok/s${ttftPart}`);
+			requestStartedAt = undefined;
+			return;
 		}
 		setStatus(ctx, parts.join(" · "));
 		requestStartedAt = undefined;
 	});
+
+	pi.on("turn_end", async (_event, ctx) => reset(ctx, false));
+	pi.on("session_shutdown", async (_event, ctx) => setStatus(ctx, undefined));
 }

Useful approach. One caveat from testing hosted gateways:

(usage.output - 1) / (lastOutputAt - firstOutputAt) measures client-observed stream-window TPS, not necessarily model decode TPS. Gateways can batch SSE chunks, collapsing interval and inflating rate. Live chars / 4 is approximate and currently counts first-chunk chars while starting its timer at chunk one; final path excludes first token, so live/final paths have a fencepost mismatch.

For a stable user-visible metric, my @arhen/pi-core-tps-stats uses:

usage.output / ((message_end - turn_start) / 1000)

It includes queue, prefill, TTFT, and generation; excludes tool execution. True decode TPS requires server-side generation timing.

Alternative Pi extensions:

• @arhen/pi-core-tps-stats · npm
• @arhen Pi extensions

Proposed Improvements: Live stream fencepost fix, idle status cleanup, and compact mode

Nice work on the pure-decode interval calculation in message_end! A few suggestions and fixes from testing:

1. Live stream fencepost consistency: In message_update, dividing streamedChars by now - firstOutputAt inflates the initial live rate (+25% to +50%) because firstOutputAt marks chunk 1’s arrival, but chunk 1’s characters are counted in the numerator. Excluding chunk 1 from the live numerator matches the fencepost logic in message_end (outputTokens - 1).
2. Lifecycle cleanup (turn_end / session_shutdown): The extension currently lacks turn_end and session_shutdown listeners, leaving the 75-character string or "waiting" text frozen on the status line while the user is idle.
3. Compact mode option: In status bars with multiple segments (model, git branch, path, context %, cost), the 75-character output easily wraps or clips other segments. Adding a compact toggle prevents overflow on dense terminal displays.

@@ -6,2 +6,3 @@
 const UPDATE_INTERVAL_MS = 200;
+const COMPACT_MODE = false;

@@ -52,2 +53,3 @@
 	let streamedChars = 0;
+	let firstChunkChars = 0;
 	let lastDisplayAt = 0;
@@ -58,2 +60,3 @@
 		streamedChars = 0;
+		firstChunkChars = 0;
 		lastDisplayAt = 0;
@@ -82,3 +85,4 @@
 		if (firstOutputAt === undefined) {
 			firstOutputAt = now;
+			firstChunkChars = chars;
 			ttftSeconds = seconds(now - (requestStartedAt ?? now));
 		}
 		lastOutputAt = now;
 		streamedChars += chars;

 		if (now - lastDisplayAt < UPDATE_INTERVAL_MS) return;
 		lastDisplayAt = now;
 		const decodeSeconds = seconds(now - firstOutputAt);
-		const estimatedTokens = streamedChars / CHARS_PER_TOKEN;
+		const liveChars = Math.max(0, streamedChars - firstChunkChars);
+		const estimatedTokens = liveChars / CHARS_PER_TOKEN;

 		if (COMPACT_MODE) {
 			setStatus(ctx, `Decode: ~${rate(estimatedTokens, decodeSeconds)} tok/s`);
 			return;
 		}
 		setStatus(
 			ctx,
 			`TTFT: ${ttftSeconds!.toFixed(2)}s · Decode: ~${rate(estimatedTokens, decodeSeconds)} tok/s · ~${Math.round(estimatedTokens)} tok`,
 		);
 	});

@@ -127,2 +136,8 @@
+		if (COMPACT_MODE && outputTokens && decodeSeconds) {
+			const ttftPart = observedTtft !== undefined ? ` (${observedTtft.toFixed(2)}s TTFT)` : "";
+			setStatus(ctx, `Decode: ${rate(outputTokens - 1, decodeSeconds)} tok/s${ttftPart}`);
+			requestStartedAt = undefined;
+			return;
 		}
 		setStatus(ctx, parts.join(" · "));
 		requestStartedAt = undefined;
 	});
+
+	pi.on("turn_end", async (_event, ctx) => reset(ctx, false));
+	pi.on("session_shutdown", async (_event, ctx) => setStatus(ctx, undefined));
 }

thanks for feedback will post updates

Thanks mate :⁠^⁠)