给桌面版加了个常驻实时缓存监控挂件

[lisuyuan13]

🤖 声明：本扩展的开发、调试以及本文的撰写均通过 AI Agent（pi）协作完成，行号与数据均经过实测核对。

---

图里这个挂件叫 *

pi-cache-live*，是我（在 AI Agent 辅助下）写的一个常驻实时缓存监控扩展，效果如上图：输入框下方固定栏常驻，随每条回复实时刷新，往上滚会话也不丢。

它显示什么

• Token 四项：输入 / 输出 / 缓存读 / 缓存写（按量级自动 M/k 定档，3 位有效数字，零值恒显 0）
• 命中率：当前累计命中率 + 平均命中率 + 历史轮次 sparkline 曲线
• 打开历史会话秒级重建完整曲线；费用段在模型有 cost 配置时自动出现

安装（30 秒）

把 pi-cache-live.ts 扔进 ~/.pi/agent/extensions/，新开一个会话即可，零配置（user scope 自动发现）。开关：/cache-live off / /cache-live on

几个关键实现点（写给想抄作业的人）

1. 数据同源：只读 assistant 消息上的 usage 字段（input/output/cacheRead/cacheWrite/cost），provider API 真实计数，无任何估算
2. 常驻靠 belowEditor：桌面端只有 placement: "belowEditor" 会进输入框下方的固定栏；aboveEditor 渲染在消息流顶部，会被滚走（开始还以为写反了，翻了 .next-desktop 编译产物才对上）
3. 实时刷新：订阅 message_end——注意它触发时本条消息还没落盘，要从事件参数里补进去再统计；另订阅 session_start / session_compact / session_tree 做全量重算。O(n) 纯数字求和，微秒级
4. 中文布局：widget 是 <pre> + pre-wrap + mono 字体栈（含 CJK 回退），中文可行；但 CJK 占两格，跨行空格对齐会错位，所以做成「标签 + · 分隔」的单行式
5. 命中率公式：cacheRead / (input + cacheRead + cacheWrite)，OpenAI 风格 cacheWrite=0 时自动退化

以下内容复制到本地另存为:pi-cache-live.ts 扔进 `~/.pi/agent/extensions/ 新开一个会话**即可

/**
 * pi-cache-live —— 常驻实时缓存监控挂件
 *
 * 规格来源：《Pi Agent 实时缓存监控扩展开发记录》（思源笔记导出/溯源笔记/随手记/）
 *   - 数据同源：只读 assistant 消息上的 usage（provider API 真实计数，无估算）
 *   - 常驻：setWidget(..., { placement: "belowEditor" }) → 桌面版底部固定栏，滚动不丢
 *     ⚠️ 桌面版反转了 TUI 的 placement 语义，只有 belowEditor 才进固定栏
 *   - 实时：message_end 全量重算活动分支（O(n) 微秒级，天然容忍 fork/retry/剪枝）
 *   - 开关：/cache-live on|off|toggle
 *   - 单位：按四项最大值动态定档（≥1M 全用 M / ≥1k 全用 k / 否则裸数字），3 位有效数字
 *
 * 命中率公式（复用 pi-cache-graph）：
 *   cacheHit% = cacheRead / (input + cacheRead + cacheWrite) × 100
 *   分母 = 本轮实际发送的完整 prompt 大小
 *   Anthropic 风格 input 不含新写缓存；OpenAI 风格 cacheWrite=0 时公式自动退化
 */

import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";

/** widget key 会被渲染成标题栏，直接用作显示标题 */
export const WIDGET_KEY = "⚡ cache-live 缓存监控";

const SPARK = "▁▂▃▄▅▆▇█";
const SPARK_WIDTH = 9;

export interface Totals {
	input: number;
	output: number;
	cacheRead: number;
	cacheWrite: number;
	cost: number;
}

export interface Usage {
	input?: number;
	output?: number;
	cacheRead?: number;
	cacheWrite?: number;
	cost?: { total?: number } | undefined;
}

const emptyTotals = (): Totals => ({ input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0 });

/** 累计命中率（%）。分母为 0 时返回 0。 */
export function hitRate(t: Pick<Totals, "input" | "cacheRead" | "cacheWrite">): number {
	const denom = (t.input || 0) + (t.cacheRead || 0) + (t.cacheWrite || 0);
	if (denom <= 0) return 0;
	return (100 * (t.cacheRead || 0)) / denom;
}

/** 动态定档：四项总量最大值决定单位 */
export type Scale = "raw" | "k" | "M";

export function pickScale(values: number[]): Scale {
	const max = values.reduce((a, b) => Math.max(a, b || 0), 0);
	if (max >= 1_000_000) return "M";
	if (max >= 1_000) return "k";
	return "raw";
}

/** 3 位有效数字，去掉无意义的尾随 0（22.0 → 22，6.370 → 6.37） */
function sig3(n: number): string {
	if (!isFinite(n)) return "0";
	if (n === 0) return "0";
	let s = n.toPrecision(3);
	if (s.includes(".")) s = s.replace(/0+$/, "").replace(/\.$/, "");
	if (s.includes("e")) s = n.toFixed(2); // 极小值兜底，避免 3.00e-5
	return s;
}

/** 统一单位下的 token 显示。零值恒为裸 0。 */
export function fmtTokens(n: number, scale: Scale): string {
	const v = n || 0;
	if (v === 0) return "0";
	if (scale === "M") return `${sig3(v / 1_000_000)}M`;
	if (scale === "k") return `${sig3(v / 1_000)}k`;
	return String(Math.round(v));
}

export function fmtCost(c: number): string {
	if (!c || c <= 0) return "$0";
	if (c >= 100) return `$${Math.round(c)}`;
	if (c >= 1) return `$${c.toFixed(2)}`;
	return `$${c.toPrecision(2)}`;
}

/** 把一组命中率映射成 sparkline；全相等时退化为中间档，避免整条空柱 */
export function sparkline(values: number[], width = SPARK_WIDTH): string {
	const tail = values.slice(-width);
	if (tail.length === 0) return "";
	const min = Math.min(...tail);
	const max = Math.max(...tail);
	if (max - min < 1e-9) {
		const mid = max >= 90 ? SPARK.length - 1 : Math.floor(SPARK.length / 2);
		return SPARK[mid]!.repeat(tail.length);
	}
	return tail
		.map((v) => {
			const idx = Math.round(((v - min) / (max - min)) * (SPARK.length - 1));
			return SPARK[Math.max(0, Math.min(SPARK.length - 1, idx))]!;
		})
		.join("");
}

/** 只认 role==="assistant" 且带 usage 的消息 */
export function collect(messages: Array<{ role?: string; usage?: Usage } | undefined>): {
	totals: Totals;
	turns: number[];
} {
	const totals = emptyTotals();
	const turns: number[] = [];
	for (const m of messages) {
		if (!m || m.role !== "assistant" || !m.usage) continue;
		const u = m.usage;
		const t: Totals = {
			input: u.input || 0,
			output: u.output || 0,
			cacheRead: u.cacheRead || 0,
			cacheWrite: u.cacheWrite || 0,
			cost: u.cost?.total || 0,
		};
		totals.input += t.input;
		totals.output += t.output;
		totals.cacheRead += t.cacheRead;
		totals.cacheWrite += t.cacheWrite;
		totals.cost += t.cost;
		turns.push(hitRate(t));
	}
	return { totals, turns };
}

/** 渲染成 widget 文本行（CJK 占两格 → 标签前缀式 + `·` 分隔，不做跨行空格对齐） */
export function renderLines(totals: Totals, turns: number[]): string[] {
	if (turns.length === 0) return ["暂无数据 · 等待第一条 AI 回复"];
	const scale = pickScale([totals.input, totals.output, totals.cacheRead, totals.cacheWrite]);
	const line1Parts = [
		`输入 ${fmtTokens(totals.input, scale)}`,
		`输出 ${fmtTokens(totals.output, scale)}`,
		`缓存读 ${fmtTokens(totals.cacheRead, scale)}`,
		`缓存写 ${fmtTokens(totals.cacheWrite, scale)}`,
	];
	if (totals.cost > 0) line1Parts.push(`费用 ${fmtCost(totals.cost)}`);

	const overall = hitRate(totals);
	const avg = turns.reduce((a, b) => a + b, 0) / turns.length;
	// sparkline 属 ambiguous width，放行尾不接后续文本，避免错位
	const line2 = `命中率 ${Math.round(overall)}% 平均 ${Math.round(avg)}% · 第${turns.length}轮 ${sparkline(turns)}`;
	return [line1Parts.join(" · "), line2];
}

export default function (pi: ExtensionAPI) {
	let enabled = true;

	/**
	 * 取活动分支消息。
	 * ⚠️ message_end 在落盘前触发，事件里的 assistant 消息可能还不在分支里 → 手动补上。
	 */
	function gather(ctx: ExtensionContext, pending?: { role?: string; usage?: Usage }): Array<{ role?: string; usage?: Usage }> {
		const branch: any[] = ctx.sessionManager?.getBranch?.() ?? [];
		const msgs = branch.filter((e) => e?.type === "message").map((e) => e?.message).filter(Boolean);
		if (pending && pending.role === "assistant" && pending.usage && !msgs.includes(pending)) msgs.push(pending);
		return msgs;
	}

	function paint(ctx: ExtensionContext, pending?: { role?: string; usage?: Usage }) {
		if (!enabled) return;
		if (typeof ctx.ui?.setWidget !== "function") return;
		const { totals, turns } = collect(gather(ctx, pending));
		ctx.ui.setWidget(WIDGET_KEY, renderLines(totals, turns), { placement: "belowEditor" });
	}

	function clear(ctx: ExtensionContext) {
		if (typeof ctx.ui?.setWidget !== "function") return;
		ctx.ui.setWidget(WIDGET_KEY, undefined);
	}

	// 每条 AI 回复实时刷新
	pi.on("message_end", async (event, ctx) => paint(ctx, event.message as any));

	// 新会话 / 恢复 / fork / 切换 / 重载：全量重建，打开历史会话即刻有完整曲线
	pi.on("session_start", async (_event, ctx) => paint(ctx));
	pi.on("session_compact", async (_event, ctx) => paint(ctx));
	pi.on("session_info_changed", async (_event, ctx) => paint(ctx));
	pi.on("session_tree", async (_event, ctx) => paint(ctx)); // 分支跳转/剪枝后重算
	pi.on("model_select", async (_event, ctx) => paint(ctx));
	pi.on("agent_start", async (_event, ctx) => paint(ctx));

	pi.registerCommand("cache-live", {
		description: "常驻缓存命中率监控挂件（on|off|toggle）",
		handler: async (args, ctx) => {
			const arg = (args || "").trim().toLowerCase();
			if (arg === "off") {
				enabled = false;
				clear(ctx);
				ctx.ui.notify?.("cache-live 已关闭", "info");
				return;
			}
			if (arg === "on" || arg === "toggle") enabled = true;
			paint(ctx);
			ctx.ui.notify?.(enabled ? "cache-live 已开启" : "cache-live 状态异常", "info");
		},
	});
}