Sync TUI metadata and make Stop responsive

extensions/web-sessions.ts

@@ -22,6 +22,7 @@ import type {
 	AgentCommand,
 	AgentEventMessage,
 	AgentHelloMessage,
+	AgentHistoryMessage,
 	AgentResponseMessage,
 	AgentSessionReplacedMessage,
 	AgentSubagentsMessage,
@@ -35,6 +36,8 @@ import type {
 import { WEB_STATE_VERSION } from "../web/protocol.js";
 import { expandSlashCommand, isSkillSlashCommand } from "../web/slash-commands.js";
 import { formatWorktreeCreateCommandArgs } from "../web/worktree-command.js";
+import { boundedWebHistory } from "../web/history.js";
+import { managedWorktreeFromEntries } from "../web/server/worktrees.js";
 import { readWebTailscaleSetting, writeWebTailscaleSetting } from "./web-settings.js";
 import {
 	consumeWorktreeReplacement,
@@ -89,7 +92,7 @@ export function splitWebWorktreeCommandArgs(args: string): { token: string; work
 }
 
 /** Abort the main session and wait for subagent abort operations registered through waitUntil. */
-export async function abortSessionAndSubagents(options: {
+export function abortSessionAndSubagents(options: {
 	sessionId: string;
 	abortMain(): void;
 	emit(request: SubagentAbortRequest): void;
@@ -107,7 +110,7 @@ export async function abortSessionAndSubagents(options: {
 		// A broken optional listener must never prevent the main Stop request.
 	}
 	options.abortMain();
-	await Promise.allSettled(operations);
+	return Promise.allSettled(operations).then(() => undefined);
 }
 
 /** Apply a route change and roll it back if the matching settings write fails. */
@@ -287,6 +290,11 @@ function addWebUsage(target: NonNullable<WebSession["usage"]>, value: unknown):
 	}
 }
 
+function contextUsage(ctx: ExtensionContext): WebSession["contextUsage"] {
+	const usage = ctx.getContextUsage();
+	return usage ? { ...usage } : ctx.model ? { tokens: null, contextWindow: ctx.model.contextWindow, percent: null } : undefined;
+}
+
 function sessionMetrics(ctx: ExtensionContext): Pick<WebSession, "usage" | "contextUsage"> {
 	const usage = zeroWebUsage();
 	for (const entry of ctx.sessionManager.getEntries()) {
@@ -296,11 +304,14 @@ function sessionMetrics(ctx: ExtensionContext): Pick<WebSession, "usage" | "cont
 			addWebUsage(usage, entry.usage);
 		}
 	}
-	const contextUsage = ctx.getContextUsage();
-	return {
-		usage,
-		contextUsage: contextUsage ? { ...contextUsage } : ctx.model ? { tokens: null, contextWindow: ctx.model.contextWindow, percent: null } : undefined,
-	};
+	return { usage, contextUsage: contextUsage(ctx) };
+}
+
+function refreshIncrementalMetrics(state: BridgeState, usageValue: unknown): void {
+	const current = state.metrics.usage ?? zeroWebUsage();
+	const usage = { ...current, cost: { ...current.cost } };
+	addWebUsage(usage, usageValue);
+	state.metrics = { usage, contextUsage: contextUsage(state.ctx) };
 }
 
 function safeClone(value: unknown): Record<string, unknown> {
@@ -364,6 +375,12 @@ function flush(state: BridgeState): void {
 	for (const message of state.pending.splice(0)) state.socket.send(JSON.stringify(message));
 }
 
+function discardPendingCoveredByHello(state: BridgeState): void {
+	const latestSubagents = [...state.pending].reverse().find((message) => message.type === "agent.subagents");
+	state.pending = state.pending.filter((message) => message.type === "agent.response");
+	if (latestSubagents) state.pending.push(latestSubagents);
+}
+
 function statusForContext(ctx: ExtensionContext): WebSession["status"] {
 	return ctx.isIdle() ? "idle" : "working";
 }
@@ -394,8 +411,7 @@ async function refreshGitMetadata(pi: ExtensionAPI, state: BridgeState): Promise
 	if (!state.closed) updateSession(state, { branch, pullRequest });
 }
 
-function updateSession(state: BridgeState, patch: Partial<WebSession> = {}, refreshMetrics = false): void {
-	if (refreshMetrics) state.metrics = sessionMetrics(state.ctx);
+function updateSession(state: BridgeState, patch: Partial<WebSession> = {}): void {
 	state.session = {
 		...state.session,
 		...state.metrics,
@@ -509,14 +525,21 @@ async function executeAgentCommand(
 				respond(state, requestId, true);
 				return;
 			}
-			case "abort":
-				await abortSessionAndSubagents({
+			case "abort": {
+				// Invoke the main abort before acknowledging, then let subagent teardown
+				// settle in the background. Compaction can delay that settlement well
+				// past the browser's command bound even though Stop has taken effect.
+				const settlement = abortSessionAndSubagents({
 					sessionId: state.session.id,
 					abortMain: () => state.ctx.abort(),
 					emit: (request) => pi.events.emit(SUBAGENT_ABORT_EVENT, request),
 				});
-				respond(state, requestId, true);
+				respond(state, requestId, true, { accepted: true });
+				void settlement.catch((error) => {
+					console.error(`Pi web Stop failed after acknowledgement: ${error instanceof Error ? error.message : String(error)}`);
+				});
 				return;
+			}
 			case "replace_queue":
 				respond(state, requestId, true);
 				return;
@@ -730,13 +753,14 @@ async function connect(pi: ExtensionAPI, state: BridgeState): Promise<void> {
 				return;
 			}
 			state.reconnectAttempt = 0;
+			discardPendingCoveredByHello(state);
 			const hello: AgentHelloMessage = {
 				type: "agent.hello",
 				session: state.session,
-				// The server can read JSONL history from session.file. Sending every
-				// entry here makes large sessions exceed WebSocket frame limits and
-				// prevents the native bridge from registering after a daemon restart.
-				entries: [],
+				historyMode: "replace",
+				// Send only active, compaction-aware history and bound its encoded size.
+				// The append-only JSONL can be hundreds of MB after old context is gone.
+				entries: boundedWebHistory(state.ctx.sessionManager.buildContextEntries()),
 			};
 			socket.send(JSON.stringify(hello));
 			if (state.sourceReplacement) {
@@ -755,15 +779,26 @@ async function connect(pi: ExtensionAPI, state: BridgeState): Promise<void> {
 	};
 }
 
+function previewFromMessage(message: unknown): string | undefined {
+	if (!isRecord(message) || (message.role !== "user" && message.role !== "assistant")) return undefined;
+	if (typeof message.content === "string") return message.content.slice(0, 180);
+	if (!Array.isArray(message.content)) return undefined;
+	const preview = message.content
+		.filter((item): item is Record<string, unknown> => isRecord(item) && item.type === "text" && typeof item.text === "string")
+		.map((item) => item.text as string)
+		.join("");
+	return preview ? preview.slice(0, 180) : undefined;
+}
+
 function makeSession(ctx: ExtensionContext, branch: string | undefined): WebSession {
 	const entries = ctx.sessionManager.getEntries();
 	const header = ctx.sessionManager.getHeader();
-	const firstUser = entries.find((entry) => entry.type === "message" && entry.message.role === "user");
 	let preview: string | undefined;
-	if (firstUser?.type === "message" && firstUser.message.role === "user") {
-		preview = typeof firstUser.message.content === "string"
-			? firstUser.message.content.slice(0, 180)
-			: firstUser.message.content.filter((item) => item.type === "text").map((item) => item.text).join("").slice(0, 180);
+	for (let index = entries.length - 1; index >= 0; index -= 1) {
+		const entry = entries[index];
+		if (entry?.type !== "message" || (entry.message.role !== "user" && entry.message.role !== "assistant")) continue;
+		preview = previewFromMessage(entry.message);
+		if (preview) break;
 	}
 	return {
 		id: ctx.sessionManager.getSessionId(),
@@ -780,6 +815,7 @@ function makeSession(ctx: ExtensionContext, branch: string | undefined): WebSess
 		messageCount: entries.filter((entry) => entry.type === "message").length,
 		preview,
 		parentSession: header?.parentSession,
+		managedWorktree: managedWorktreeFromEntries(entries),
 		...sessionMetrics(ctx),
 	};
 }
@@ -816,17 +852,21 @@ export default function webSessions(pi: ExtensionAPI): void {
 
 	const forward = (event: unknown, ctx: ExtensionContext, status?: WebSession["status"], refreshMetrics = false): void => {
 		if (!bridge || bridge.closed || ctx.sessionManager.getSessionId() !== bridge.session.id) return;
+		if (refreshMetrics) {
+			const message = isRecord(event) && event.type === "message_end" ? event.message : undefined;
+			refreshIncrementalMetrics(bridge, isRecord(message) ? message.usage : undefined);
+		}
 		send(bridge, {
 			type: "agent.event",
 			sessionId: bridge.session.id,
 			event: safeClone(event),
 		} satisfies AgentEventMessage);
+		const latestPreview = isRecord(event) && event.type === "message_end" ? previewFromMessage(event.message) : undefined;
 		updateSession(bridge, {
 			status,
-			messageCount: refreshMetrics
-				? ctx.sessionManager.getEntries().filter((entry) => entry.type === "message").length
-				: bridge.session.messageCount,
-		}, refreshMetrics);
+			preview: latestPreview ?? bridge.session.preview,
+			messageCount: refreshMetrics ? bridge.session.messageCount + 1 : bridge.session.messageCount,
+		});
 	};
 
 	pi.registerCommand("web", {
@@ -1088,8 +1128,14 @@ export default function webSessions(pi: ExtensionAPI): void {
 		}, { once: true });
 	});
 	pi.on("session_compact", (event, ctx) => {
-		if (bridge) updateSession(bridge, {}, true);
 		if (bridge && ctx.sessionManager.getSessionId() === bridge.session.id) {
+			refreshIncrementalMetrics(bridge, event.compactionEntry.usage);
+			updateSession(bridge);
+			send(bridge, {
+				type: "agent.history",
+				sessionId: bridge.session.id,
+				entries: boundedWebHistory(ctx.sessionManager.buildContextEntries()),
+			} satisfies AgentHistoryMessage);
 			endBridgeCompaction(bridge, {
 				aborted: false,
 				willRetry: event.willRetry,

tests/web-client-api.test.ts

@@ -16,7 +16,7 @@ test("clone and fork outlive the server's 30-second operation bound", () => {
 	expect(sessionCommandTimeout({ type: "create_worktree", existing: "/repo/worktree" })).toBeGreaterThanOrEqual(10 * 60_000);
 	expect(sessionCommandTimeout({ type: "create_worktree_v2", repository: "/repo", name: "pr-30", branch: "owner/topic", startPoint: "origin/owner/topic" })).toBeGreaterThanOrEqual(10 * 60_000);
 	expect(sessionCommandTimeout({ type: "reload" })).toBeGreaterThanOrEqual(10 * 60_000);
-	expect(sessionCommandTimeout({ type: "abort" })).toBe(15_000);
+	expect(sessionCommandTimeout({ type: "abort" })).toBe(35_000);
 });
 
 test("worktree capability is refreshed for each create request", async () => {

tests/web-history.test.ts

@@ -0,0 +1,63 @@
+import { expect, test } from "bun:test";
+import { boundedWebHistory, messagesToWebHistory } from "../web/history.ts";
+
+function message(id: string, text: string) {
+	return {
+		type: "message",
+		id,
+		parentId: null,
+		timestamp: "2026-01-01T00:00:00.000Z",
+		message: { role: "assistant", content: text, timestamp: 1 },
+	};
+}
+
+test("web history drops entries before the latest compaction boundary", () => {
+	const history = boundedWebHistory([
+		message("old", "old transcript"),
+		{ type: "compaction", id: "compact", parentId: "old", timestamp: "2026-01-01T00:01:00.000Z", summary: "summary" },
+		message("new", "new transcript"),
+	]);
+	expect(history).toHaveLength(2);
+	expect(history[0]).toMatchObject({ id: "web-compaction-compact", message: { role: "assistant" } });
+	expect(JSON.stringify(history)).toContain("summary");
+	expect(JSON.stringify(history)).not.toContain("old transcript");
+	expect(JSON.stringify(history)).toContain("new transcript");
+});
+
+test("web history reserves space for the compaction summary", () => {
+	const history = boundedWebHistory([
+		{ type: "compaction", id: "compact", timestamp: "2026-01-01T00:01:00.000Z", summary: "required summary" },
+		message("one", "one"),
+		message("two", "two"),
+	], { maxEntries: 2 });
+	expect(history.map((entry) => (entry as { id?: string }).id)).toEqual(["web-compaction-compact", "two"]);
+});
+
+test("web history remains byte bounded and truncates oversized content", () => {
+	const history = boundedWebHistory([message("large", "x".repeat(400_000))], { maxBytes: 400_000 });
+	expect(history).toHaveLength(1);
+	const serialized = JSON.stringify(history);
+	expect(new TextEncoder().encode(serialized).byteLength).toBeLessThanOrEqual(400_000);
+	expect(serialized).toContain("Pi Web truncated");
+});
+
+test("oversized history images become explicit omission markers", () => {
+	const history = boundedWebHistory([{
+		type: "message",
+		id: "image",
+		message: { role: "user", content: [{ type: "image", mimeType: "image/png", data: "x".repeat(4 * 1024 * 1024 + 1) }] },
+	}]);
+	expect(JSON.stringify(history)).toContain("omitted an oversized image/png attachment");
+});
+
+test("managed context messages become bounded semantic history", () => {
+	const history = messagesToWebHistory([
+		{ role: "compactionSummary", summary: "prior work", tokensBefore: 100, timestamp: 1 },
+		{ role: "custom", customType: "hidden", display: false, content: "not visible", timestamp: 2 },
+		{ role: "branchSummary", summary: "not previously rendered", timestamp: 3 },
+		{ role: "user", content: "continue", timestamp: 4 },
+	]);
+	expect(history).toHaveLength(2);
+	expect(history[0]).toMatchObject({ message: { role: "assistant" } });
+	expect(history[1]).toMatchObject({ message: { role: "user", content: "continue" } });
+});

tests/web-serialized-writer.test.ts

@@ -0,0 +1,55 @@
+import { expect, test } from "bun:test";
+import { SerializedWriter } from "../web/server/serialized-writer.ts";
+
+test("serialized writes acknowledge completion and preserve order", async () => {
+	const releases: Array<() => void> = [];
+	const started: string[] = [];
+	const writer = new SerializedWriter<string>(async (value) => {
+		started.push(value);
+		await new Promise<void>((resolve) => releases.push(resolve));
+	});
+	let firstDelivered = false;
+	let secondDelivered = false;
+	const first = writer.write("first").then(() => { firstDelivered = true; });
+	const second = writer.write("second").then(() => { secondDelivered = true; });
+	await Bun.sleep(0);
+	expect(started).toEqual(["first"]);
+	expect(firstDelivered).toBe(false);
+	expect(secondDelivered).toBe(false);
+	releases.shift()?.();
+	await first;
+	await Bun.sleep(0);
+	expect(started).toEqual(["first", "second"]);
+	expect(secondDelivered).toBe(false);
+	releases.shift()?.();
+	await second;
+	expect(secondDelivered).toBe(true);
+});
+
+test("serialized writes skip queued work that expires before delivery", async () => {
+	let releaseFirst!: () => void;
+	const started: string[] = [];
+	const writer = new SerializedWriter<string>(async (value) => {
+		started.push(value);
+		if (value === "first") await new Promise<void>((resolve) => { releaseFirst = resolve; });
+	});
+	let secondActive = true;
+	const first = writer.write("first");
+	const second = writer.write("expired", () => secondActive);
+	await Bun.sleep(0);
+	secondActive = false;
+	releaseFirst();
+	await Promise.all([first, second]);
+	expect(started).toEqual(["first"]);
+});
+
+test("serialized writes report a delivery failure and recover for later writes", async () => {
+	let attempts = 0;
+	const writer = new SerializedWriter<string>(async () => {
+		attempts += 1;
+		if (attempts === 1) throw new Error("stdin write failed");
+	});
+	await expect(writer.write("abort")).rejects.toThrow("stdin write failed");
+	await expect(writer.write("later command")).resolves.toBeUndefined();
+	expect(attempts).toBe(2);
+});