diff --git a/README.md b/README.md index 5d6a060..c16eccc 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,10 @@ Managed ownership records the directory name and whether Pi created the local br The LLM-callable `worktree` tool provides the same `name`, `repository`, `branch`, `startPoint`, and `existing` flows. For a pull request URL, agents must resolve the PR's real head branch and fetched remote-tracking ref and pass them explicitly rather than deriving a branch from a directory such as `pr-30`. The tool queues a correlated `/worktree` follow-up, ends the old run, verifies the replacement, and resumes its continuation there. Create-only requests that should not enter the checkout remain ordinary Git operations. +## Context clearing + +Run `/clear` to exclude the conversation so far from subsequent model requests without removing it from the session transcript. Pi and Pi Web both show `Context cleared.` when the new context boundary is active. + ## Web sessions `extensions/web-sessions.ts` connects every running Pi session to a local Bun server. The first Pi process starts the server on `127.0.0.1:31415`; later processes discover it through `~/.pi/agent/web/server.json` and attach their own live event streams. @@ -234,5 +238,5 @@ bun install --frozen-lockfile bun run check bun test bun run webBuild -pi -e ./extensions/session-footer.ts -e ./extensions/pr-footer.ts -e ./extensions/subagents.ts -e ./extensions/worktree.ts -e ./extensions/web-sessions.ts -e ./extensions/auto-router.ts +pi -e ./extensions/session-footer.ts -e ./extensions/pr-footer.ts -e ./extensions/subagents.ts -e ./extensions/worktree.ts -e ./extensions/web-sessions.ts -e ./extensions/auto-router.ts -e ./extensions/clear-context.ts ``` diff --git a/extensions/clear-context.ts b/extensions/clear-context.ts new file mode 100644 index 0000000..138a412 --- /dev/null +++ b/extensions/clear-context.ts @@ -0,0 +1,184 @@ +import { + type ContextEvent, + type ExtensionAPI, + estimateTokens, + findCutPoint, + generateSummaryWithUsage, + type SessionEntry, + sessionEntryToContextMessages, +} from "@earendil-works/pi-coding-agent"; +import { WEB_CLEAR_CONTEXT_ENTRY } from "../web/clear-command.js"; + +export const CLEAR_CONTEXT_ENTRY = WEB_CLEAR_CONTEXT_ENTRY; +export const CLEAR_CONTEXT_MATERIALIZED_ENTRY = + "vessup:clear-context-materialized"; +export const CLEAR_CONTEXT_COMPLETE_MESSAGE = "Context cleared."; +const CLEAR_COMPACTION_DETAIL = "clearContextBoundary"; + +function latestActiveClear( + entries: readonly SessionEntry[], +): SessionEntry | undefined { + for (let index = entries.length - 1; index >= 0; index -= 1) { + const entry = entries[index]; + if (entry?.type !== "custom") continue; + if (entry.customType === CLEAR_CONTEXT_MATERIALIZED_ENTRY) return undefined; + if (entry.customType === CLEAR_CONTEXT_ENTRY) return entry; + } + return undefined; +} + +/** Add a durable boundary after which earlier conversation is excluded from LLM context. */ +export function clearSessionContext( + pi: Pick, +): void { + pi.appendEntry(CLEAR_CONTEXT_ENTRY); +} + +/** Preserve the transcript while returning only context messages after the latest clear boundary. */ +export function contextAfterLatestClear( + entries: readonly SessionEntry[], +): ContextEvent["messages"] | undefined { + const clear = latestActiveClear(entries); + if (!clear) return undefined; + const clearIndex = entries.indexOf(clear); + return entries + .slice(clearIndex + 1) + .flatMap((entry) => sessionEntryToContextMessages(entry)); +} + +function preparePostClearCompaction( + entries: SessionEntry[], + settings: { keepRecentTokens: number; reserveTokens: number }, +): + | { + firstKeptEntryId: string; + messagesToSummarize: ContextEvent["messages"]; + turnPrefixMessages: ContextEvent["messages"]; + tokensBefore: number; + settings: typeof settings; + } + | undefined { + const cutPoint = findCutPoint( + entries, + 0, + entries.length, + settings.keepRecentTokens, + ); + const firstKeptEntry = entries[cutPoint.firstKeptEntryIndex]; + if (!firstKeptEntry?.id) return undefined; + const historyEnd = cutPoint.isSplitTurn + ? cutPoint.turnStartIndex + : cutPoint.firstKeptEntryIndex; + const messagesToSummarize = entries + .slice(0, historyEnd) + .flatMap((entry) => sessionEntryToContextMessages(entry)); + const turnPrefixMessages = cutPoint.isSplitTurn + ? entries + .slice(cutPoint.turnStartIndex, cutPoint.firstKeptEntryIndex) + .flatMap((entry) => sessionEntryToContextMessages(entry)) + : []; + if (messagesToSummarize.length === 0 && turnPrefixMessages.length === 0) + return undefined; + return { + firstKeptEntryId: firstKeptEntry.id, + messagesToSummarize, + turnPrefixMessages, + tokensBefore: entries + .flatMap((entry) => sessionEntryToContextMessages(entry)) + .reduce((total, message) => total + estimateTokens(message), 0), + settings, + }; +} + +export default function clearContextExtension(pi: ExtensionAPI): void { + pi.registerCommand("clear", { + description: "Clear conversation context while keeping the transcript", + handler: async (args, ctx) => { + if (args.trim()) { + ctx.ui.notify("/clear does not accept arguments", "error"); + return; + } + await ctx.waitForIdle(); + clearSessionContext(pi); + ctx.ui.notify(CLEAR_CONTEXT_COMPLETE_MESSAGE, "info"); + }, + }); + + pi.on("context", (_event, ctx) => { + const messages = contextAfterLatestClear( + ctx.sessionManager.buildContextEntries(), + ); + return messages === undefined ? undefined : { messages }; + }); + + // Pi prepares compaction from the raw branch rather than the context hook's + // filtered messages. Rebuild preparation from the post-clear branch so neither + // automatic nor explicit compaction can summarize pre-clear text. + pi.on("session_before_compact", async (event, ctx) => { + const clear = latestActiveClear(event.branchEntries); + if (!clear) return undefined; + + const clearIndex = event.branchEntries.indexOf(clear); + const postClearEntries = event.branchEntries + .slice(clearIndex) + .map((entry, index) => + index === 0 ? { ...entry, parentId: null } : entry, + ); + const preparation = preparePostClearCompaction( + postClearEntries, + event.preparation.settings, + ); + if (!preparation) { + return { + compaction: { + summary: "", + firstKeptEntryId: clear.id, + tokensBefore: event.preparation.tokensBefore, + details: { [CLEAR_COMPACTION_DETAIL]: true }, + }, + }; + } + + const model = ctx.model; + if (!model) return { cancel: true }; + const auth = await ctx.modelRegistry.getApiKeyAndHeaders(model); + if (!auth.ok) throw new Error(auth.error); + const response = await generateSummaryWithUsage( + [...preparation.messagesToSummarize, ...preparation.turnPrefixMessages], + model, + preparation.settings.reserveTokens, + auth.apiKey, + auth.headers + ? Object.fromEntries( + Object.entries(auth.headers).flatMap(([key, value]) => + typeof value === "string" ? [[key, value]] : [], + ), + ) + : undefined, + event.signal, + event.customInstructions, + undefined, + ctx.thinkingLevel, + ); + return { + compaction: { + summary: response.text, + firstKeptEntryId: preparation.firstKeptEntryId, + tokensBefore: preparation.tokensBefore, + usage: response.usage, + details: { [CLEAR_COMPACTION_DETAIL]: true }, + }, + }; + }); + + pi.on("session_compact", (event) => { + const details = event.compactionEntry.details; + if ( + !details || + typeof details !== "object" || + !(CLEAR_COMPACTION_DETAIL in details) + ) + return; + pi.appendEntry(CLEAR_CONTEXT_MATERIALIZED_ENTRY); + }); +} diff --git a/extensions/web-sessions.ts b/extensions/web-sessions.ts index 023917a..56bca3d 100644 --- a/extensions/web-sessions.ts +++ b/extensions/web-sessions.ts @@ -10,6 +10,7 @@ import { type Theme, } from "@earendil-works/pi-coding-agent"; import { agentEndTerminalNotice } from "../web/assistant-message.js"; +import { isWebClearContextCommand } from "../web/clear-command.js"; import { WEB_COMPACT_COMMAND, WEB_COMPACT_EXTENSION_COMMAND, @@ -52,6 +53,7 @@ import { AUTO_ROUTER_COMPACTION_EVENT, AUTO_ROUTER_MODEL_ROUTING_EVENT, } from "./auto-router.js"; +import { clearSessionContext } from "./clear-context.js"; import { FOOTER_CONTRIBUTION_EVENT, type FooterContribution, @@ -140,11 +142,12 @@ export function isScopedModelAllowed( function bridgeCommandList(pi: ExtensionAPI) { const commands = pi .getCommands() - .filter( - (command) => - command.source === "prompt" || - command.source === "skill" || - command.name === "worktree", + .filter((command) => + command.name === "clear" + ? isWebClearContextCommand(command) + : command.source === "prompt" || + command.source === "skill" || + command.name === "worktree", ) .map((command) => ({ name: command.name, @@ -286,6 +289,8 @@ type BridgeState = { autoTurnRouting: boolean; /** True while Auto itself is applying a runtime model swap. */ autoRuntimeRouting: boolean; + /** A user Stop must settle as idle even when Pi reports abort as an error. */ + abortRequested: boolean; /** Latest browser model choice waiting for the active turn to settle. */ pendingModelSelection?: { provider: string; modelId: string }; applyingModelSelection?: boolean; @@ -1012,6 +1017,10 @@ async function executeAgentCommand( return; } case "abort": { + // Record the user intent before invoking Pi. Some providers/runtime + // paths surface cancellation as an error-shaped assistant message; + // the explicit Stop command is the authoritative classification. + state.abortRequested = true; // 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. @@ -1083,6 +1092,14 @@ async function executeAgentCommand( throw new Error("Wait for Pi to become idle before reloading"); pi.sendUserMessage(`/web-reload ${requestId}`); return; + case "clear": + if (!pi.getCommands().some(isWebClearContextCommand)) + throw new Error("Pi clear context support is unavailable"); + if (!state.ctx.isIdle()) + throw new Error("Wait for Pi to become idle before clearing context"); + clearSessionContext(pi); + respond(state, requestId, true, { cleared: true }); + return; case "create_worktree": case "create_worktree_v2": { if (!state.ctx.isIdle()) @@ -1309,11 +1326,12 @@ async function connect(pi: ExtensionAPI, state: BridgeState): Promise { type: "agent.hello", session: state.session, 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(), - ), + // Send the raw active branch and bound its encoded size. The append-only + // JSONL can be hundreds of MB after old context is gone. Keep raw entries + // here: buildContextEntries() is already + // compaction-projected and would discard the pre-clear transcript before + // boundedWebHistory can recognize a clear-boundary compaction. + entries: boundedWebHistory(state.ctx.sessionManager.getBranch()), // Forward the session's --models scope so the daemon's model picker // shows the same list the TUI would. scopedModels: state.ctx.scopedModels.map((item) => ({ @@ -1426,9 +1444,7 @@ export default function webSessions(pi: ExtensionAPI): void { updateSession(bridge, { model: selectedModel, lastModel: - typeof value.restoreRoute === "string" - ? value.restoreRoute - : null, + typeof value.restoreRoute === "string" ? value.restoreRoute : null, }); } } @@ -1510,9 +1526,7 @@ export default function webSessions(pi: ExtensionAPI): void { }); }; - const activeBridgeFor = ( - ctx: ExtensionContext, - ): BridgeState | undefined => { + const activeBridgeFor = (ctx: ExtensionContext): BridgeState | undefined => { const state = bridge; return state && !state.closed && @@ -1785,6 +1799,7 @@ export default function webSessions(pi: ExtensionAPI): void { pending: [], autoTurnRouting: false, autoRuntimeRouting: false, + abortRequested: false, metrics: { usage: session.usage, contextUsage: session.contextUsage }, sourceReplacement, }; @@ -1877,15 +1892,25 @@ export default function webSessions(pi: ExtensionAPI): void { }); pi.on("thinking_level_select", (event, ctx) => { const activeBridge = activeBridgeFor(ctx); - if (activeBridge) updateSession(activeBridge, { thinkingLevel: event.level }); + if (activeBridge) + updateSession(activeBridge, { thinkingLevel: event.level }); forward(event, ctx); }); - pi.on("agent_start", (event, ctx) => forward(event, ctx, "working")); - // The visible run is complete at agent_end. Surface provider/runtime failures - // instead of making an unfinished transcript look successfully idle. + pi.on("agent_start", (event, ctx) => { + const activeBridge = activeBridgeFor(ctx); + if (activeBridge) activeBridge.abortRequested = false; + forward(event, ctx, "working"); + }); + // The visible run is complete at agent_end. An explicit Stop is authoritative: + // Pi may encode cancellation as stopReason "error", but user cancellation is + // not a failed session and must settle the sidebar back to idle. pi.on("agent_end", (event, ctx) => { + const activeBridge = activeBridgeFor(ctx); const status = - agentEndTerminalNotice(event)?.kind === "error" ? "error" : "idle"; + activeBridge?.abortRequested || + agentEndTerminalNotice(event)?.kind !== "error" + ? "idle" + : "error"; forward(event, ctx, status); }); pi.on("agent_settled", async (event, ctx) => { @@ -1910,8 +1935,11 @@ export default function webSessions(pi: ExtensionAPI): void { forward( event, ctx, - activeBridge?.session.status === "error" ? "error" : "idle", + activeBridge?.abortRequested || activeBridge?.session.status !== "error" + ? "idle" + : "error", ); + if (activeBridge) activeBridge.abortRequested = false; // Settlement must remain observable even if provider credential resolution // for the deferred model is slow. The server-side pending-model gate keeps // queued prompts blocked until the following model update succeeds. @@ -1954,7 +1982,9 @@ export default function webSessions(pi: ExtensionAPI): void { send(bridge, { type: "agent.history", sessionId: bridge.session.id, - entries: boundedWebHistory(ctx.sessionManager.buildContextEntries()), + // Send raw active-branch entries so clear-boundary compactions retain + // the transcript before the boundary in the web projection. + entries: boundedWebHistory(ctx.sessionManager.getBranch()), } satisfies AgentHistoryMessage); endBridgeCompaction(bridge, { aborted: false, diff --git a/package.json b/package.json index 90f464a..7c04263 100644 --- a/package.json +++ b/package.json @@ -40,6 +40,7 @@ }, "pi": { "extensions": [ + "./extensions/clear-context.ts", "./extensions/model-order.ts", "./extensions/session-footer.ts", "./extensions/pr-footer.ts", diff --git a/tests/clear-context-extension.test.ts b/tests/clear-context-extension.test.ts new file mode 100644 index 0000000..2b01abc --- /dev/null +++ b/tests/clear-context-extension.test.ts @@ -0,0 +1,153 @@ +import { expect, test } from "bun:test"; +import type { + ExtensionAPI, + ExtensionCommandContext, + SessionEntry, +} from "@earendil-works/pi-coding-agent"; +import clearContextExtension, { + CLEAR_CONTEXT_COMPLETE_MESSAGE, + CLEAR_CONTEXT_ENTRY, + CLEAR_CONTEXT_MATERIALIZED_ENTRY, + contextAfterLatestClear, +} from "../extensions/clear-context.ts"; + +function entry(value: Record): SessionEntry { + return value as unknown as SessionEntry; +} + +test("clear context retains only messages after the latest durable boundary", () => { + const before = { + role: "user" as const, + content: "old context", + timestamp: 1, + }; + const after = { + role: "user" as const, + content: "new context", + timestamp: 2, + }; + const entries = [ + entry({ type: "message", message: before }), + entry({ type: "custom", customType: CLEAR_CONTEXT_ENTRY }), + entry({ type: "model_change", provider: "test", modelId: "model" }), + entry({ type: "message", message: after }), + ]; + + expect(contextAfterLatestClear(entries)).toEqual([after]); + expect(contextAfterLatestClear(entries.slice(0, 1))).toBeUndefined(); + expect(contextAfterLatestClear(entries.slice(0, 2))).toEqual([]); +}); + +test("a clear boundary is materialized without summarizing cleared context", async () => { + const handlers = new Map unknown>(); + const appended: string[] = []; + clearContextExtension({ + appendEntry: (customType: string) => appended.push(customType), + registerCommand: () => undefined, + on: (name: string, handler: (event: never) => unknown) => { + handlers.set(name, handler); + }, + } as unknown as ExtensionAPI); + + const clear = entry({ + type: "custom", + id: "clear-id", + customType: CLEAR_CONTEXT_ENTRY, + }); + const compaction = (await handlers.get("session_before_compact")?.({ + branchEntries: [ + entry({ type: "message", message: { role: "user" } }), + clear, + ], + preparation: { + tokensBefore: 123, + settings: { keepRecentTokens: 20_000, reserveTokens: 16_384 }, + }, + } as never)) as { + compaction?: { + summary?: string; + firstKeptEntryId?: string; + tokensBefore?: number; + }; + }; + expect(compaction.compaction).toMatchObject({ + summary: "", + firstKeptEntryId: "clear-id", + tokensBefore: 123, + }); + + handlers.get("session_compact")?.({ + compactionEntry: { details: { clearContextBoundary: true } }, + } as never); + expect(appended).toEqual([CLEAR_CONTEXT_MATERIALIZED_ENTRY]); + expect( + contextAfterLatestClear([ + clear, + entry({ + type: "custom", + customType: CLEAR_CONTEXT_MATERIALIZED_ENTRY, + }), + ]), + ).toBeUndefined(); +}); + +test("the clear command appends a boundary and reports completion", async () => { + let handler: + | ((args: string, ctx: ExtensionCommandContext) => Promise) + | undefined; + const appended: string[] = []; + const notifications: Array<{ message: string; level?: string }> = []; + + clearContextExtension({ + appendEntry: (customType: string) => appended.push(customType), + registerCommand: (_name: string, command: { handler: typeof handler }) => { + handler = command.handler; + }, + on: () => undefined, + } as unknown as ExtensionAPI); + if (!handler) throw new Error("clear command was not registered"); + + await handler("", { + waitForIdle: async () => undefined, + ui: { + notify: (message: string, level?: string) => + notifications.push({ message, level }), + }, + } as unknown as ExtensionCommandContext); + + expect(appended).toEqual([CLEAR_CONTEXT_ENTRY]); + expect(notifications).toEqual([ + { message: CLEAR_CONTEXT_COMPLETE_MESSAGE, level: "info" }, + ]); +}); + +test("the clear command rejects arguments without changing context", async () => { + let handler: + | ((args: string, ctx: ExtensionCommandContext) => Promise) + | undefined; + let appended = false; + let notification = ""; + + clearContextExtension({ + appendEntry: () => { + appended = true; + }, + registerCommand: (_name: string, command: { handler: typeof handler }) => { + handler = command.handler; + }, + on: () => undefined, + } as unknown as ExtensionAPI); + if (!handler) throw new Error("clear command was not registered"); + + await handler("unexpected", { + waitForIdle: async () => undefined, + ui: { + notify: (message: string) => { + notification = message; + }, + }, + } as unknown as ExtensionCommandContext); + + expect(appended).toBe(false); + expect(notification).toBe("/clear does not accept arguments"); +}); diff --git a/tests/web-assistant-message.test.ts b/tests/web-assistant-message.test.ts index b3aad00..7260ba0 100644 --- a/tests/web-assistant-message.test.ts +++ b/tests/web-assistant-message.test.ts @@ -34,6 +34,66 @@ test("a Stop that surfaces as an error with an abort message is not a failure", title: "Stopped", detail: "This operation was aborted", }); + expect( + assistantTerminalNotice({ + role: "assistant", + content: [], + stopReason: "error", + errorMessage: "Request was aborted", + }), + ).toEqual({ + kind: "stopped", + title: "Stopped", + detail: "Request was aborted", + }); + expect( + assistantTerminalNotice({ + role: "assistant", + content: [], + stopReason: "error", + errorMessage: "Request was aborted.", + }), + ).toEqual({ + kind: "stopped", + title: "Stopped", + detail: "Request was aborted.", + }); + expect( + assistantTerminalNotice({ + role: "assistant", + content: [], + stopReason: "error", + errorMessage: "Request was aborted by the user", + }), + ).toEqual({ + kind: "stopped", + title: "Stopped", + detail: "Request was aborted by the user", + }); + expect( + assistantTerminalNotice({ + role: "assistant", + content: [], + stopReason: "error", + errorMessage: "Request aborted", + }), + ).toEqual({ + kind: "stopped", + title: "Stopped", + detail: "Request aborted", + }); + expect( + assistantTerminalNotice({ + role: "assistant", + content: [], + stopReason: "error", + errorMessage: "The operation was aborted before Pi could finish.", + }), + ).toEqual({ + kind: "stopped", + title: "Stopped", + detail: "The operation was aborted before Pi could finish.", + }); expect( assistantTerminalNotice({ role: "assistant", diff --git a/tests/web-history.test.ts b/tests/web-history.test.ts index 7c6bed8..1eb6750 100644 --- a/tests/web-history.test.ts +++ b/tests/web-history.test.ts @@ -1,5 +1,14 @@ import { expect, test } from "bun:test"; -import { boundedWebHistory, messagesToWebHistory } from "../web/history.ts"; +import { + boundedWebHistory, + messagesToWebHistory, + WEB_HISTORY_MAX_BYTES, + WEB_HISTORY_MAX_ENTRIES, +} from "../web/history.ts"; +import { + CommandRejectedError, + ManagedRpcSession, +} from "../web/server/managed-rpc-session.ts"; function message(id: string, text: string) { return { @@ -33,6 +42,421 @@ test("web history drops entries before the latest compaction boundary", () => { expect(JSON.stringify(history)).toContain("new transcript"); }); +test("web history keeps the transcript across a clear-boundary compaction", () => { + const input = [ + message("old", "old transcript"), + { + type: "compaction", + id: "compact", + summary: "post-clear summary", + details: { clearContextBoundary: true }, + }, + message("new", "new transcript"), + ]; + const history = boundedWebHistory(input); + expect(boundedWebHistory(history)).toEqual(history); + + expect(history.map((item) => (item as { id?: string }).id)).toEqual([ + "old", + "web-compaction-compact", + "new", + ]); + expect(JSON.stringify(history)).toContain("old transcript"); + expect(JSON.stringify(history)).toContain("post-clear summary"); + expect(JSON.stringify(history)).toContain("new transcript"); +}); + +test("web history reserves the newest summary after a clear boundary", () => { + const history = boundedWebHistory( + [ + message("old", "old transcript"), + { + type: "compaction", + id: "clear-compaction", + summary: "obsolete clear summary", + details: { clearContextBoundary: true }, + }, + message("middle", "middle transcript"), + { + type: "compaction", + id: "latest-compaction", + summary: "latest summary", + }, + message("new", "new transcript"), + ], + { maxEntries: 2 }, + ); + + expect(history.map((item) => (item as { id?: string }).id)).toEqual([ + "web-compaction-latest-compaction", + "new", + ]); + expect(JSON.stringify(history)).toContain("latest summary"); + expect(JSON.stringify(history)).not.toContain("obsolete clear summary"); +}); + +test("web history remains idempotent after bounding a clear transcript", () => { + const input = [ + message("old", "old transcript"), + { + type: "compaction", + id: "clear-compaction", + summary: "clear summary", + details: { clearContextBoundary: true }, + }, + message("middle", "middle transcript"), + { + type: "compaction", + id: "latest-compaction", + summary: "latest summary", + }, + message("new", "new transcript"), + ]; + const history = boundedWebHistory(input, { maxEntries: 3 }); + expect(history.map((item) => (item as { id?: string }).id)).toEqual([ + "middle", + "web-compaction-latest-compaction", + "new", + ]); + expect(boundedWebHistory(history, { maxEntries: 3 })).toEqual(history); +}); + +test("web history keeps transcript messages across a clear-context marker", () => { + const history = boundedWebHistory([ + message("old", "old transcript"), + { + type: "custom", + id: "clear", + customType: "vessup:clear-context", + timestamp: "2026-01-01T00:01:00.000Z", + }, + message("new", "new transcript"), + ]); + + expect(history.map((item) => (item as { id?: string }).id)).toEqual([ + "old", + "new", + ]); +}); + +test("managed web history follows the active branch and fetches later entries incrementally", async () => { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + const calls: Array = []; + session.getEntries = async (since?: string) => { + calls.push(since); + if (since === "active") { + return { + entries: [ + { + type: "message", + id: "next", + parentId: "active", + message: { role: "assistant", content: "continued" }, + }, + ], + leafId: "next", + }; + } + return { + entries: [ + { + type: "message", + id: "root", + parentId: null, + message: { role: "user", content: "root" }, + }, + { + type: "message", + id: "abandoned", + parentId: "root", + message: { role: "assistant", content: "abandoned" }, + }, + { + type: "message", + id: "active", + parentId: "root", + message: { role: "assistant", content: "active" }, + }, + ], + leafId: "active", + }; + }; + + expect(JSON.stringify(await session.getWebHistory())).not.toContain( + "abandoned", + ); + expect(JSON.stringify(await session.getWebHistory())).toContain("continued"); + expect(calls).toEqual([undefined, "active"]); +}); + +test("managed initial raw history cache stays bounded", async () => { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + const entries = Array.from( + { length: WEB_HISTORY_MAX_ENTRIES + 100 }, + (_, index) => ({ + id: `entry-${index}`, + type: "message", + parentId: index > 0 ? `entry-${index - 1}` : null, + message: { role: "assistant", content: `message ${index}` }, + }), + ); + session.getEntries = async () => ({ + entries, + leafId: `entry-${entries.length - 1}`, + }); + + await session.getWebHistory(); + + const cache = (session as unknown as { webHistoryEntries: unknown[] }) + .webHistoryEntries; + expect(cache).toHaveLength(WEB_HISTORY_MAX_ENTRIES); + expect(cache[0]).toMatchObject({ id: "entry-100" }); +}); + +test("managed history does not rehydrate an intentionally truncated ancestry", async () => { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + const entries = Array.from( + { length: WEB_HISTORY_MAX_ENTRIES + 1 }, + (_, index) => ({ + id: `entry-${index}`, + type: "message", + parentId: index > 0 ? `entry-${index - 1}` : null, + message: { role: "assistant", content: `message ${index}` }, + }), + ); + const calls: Array = []; + session.getEntries = async (since?: string) => { + calls.push(since); + return since + ? { entries: [], leafId: entries.at(-1)?.id ?? null } + : { entries, leafId: entries.at(-1)?.id ?? null }; + }; + + await session.getWebHistory(); + await session.getWebHistory(); + + expect(calls).toEqual([undefined, "entry-600"]); +}); + +test("managed history rehydrates when a new branch crosses an omitted ancestor", async () => { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + const entries = Array.from( + { length: WEB_HISTORY_MAX_ENTRIES + 1 }, + (_, index) => ({ + id: `entry-${index}`, + type: "message", + parentId: index > 0 ? `entry-${index - 1}` : null, + message: { role: "assistant", content: `message ${index}` }, + }), + ); + const calls: Array = []; + session.getEntries = async (since?: string) => { + calls.push(since); + if (calls.length === 1) return { entries, leafId: "entry-600" }; + if (since === "entry-600") + return { + entries: [ + { + id: "branch", + type: "message", + parentId: "entry-50", + message: { role: "assistant", content: "branch" }, + }, + ], + leafId: "branch", + }; + return { + entries: [ + ...entries, + { + id: "branch", + type: "message", + parentId: "entry-50", + message: { role: "assistant", content: "branch" }, + }, + ], + leafId: "branch", + }; + }; + + await session.getWebHistory(); + await session.getWebHistory(); + + expect(calls).toEqual([undefined, "entry-600", undefined]); +}); + +test("managed raw history cache drops an oversized entry at the byte bound", async () => { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + const entries = [ + { + id: "small", + type: "message", + parentId: null, + message: { role: "assistant", content: "small" }, + }, + { + id: "oversized", + type: "message", + parentId: "small", + message: { + role: "assistant", + content: "x".repeat(WEB_HISTORY_MAX_BYTES), + }, + }, + { + id: "latest", + type: "message", + parentId: "oversized", + message: { role: "assistant", content: "latest" }, + }, + ]; + session.getEntries = async () => ({ entries, leafId: "latest" }); + + await session.getWebHistory(); + + const cache = (session as unknown as { webHistoryEntries: unknown[] }) + .webHistoryEntries; + expect(cache.map((entry) => (entry as { id?: string }).id)).toEqual([ + "latest", + ]); + expect(cache.map((entry) => (entry as { id?: string }).id)).not.toContain( + "oversized", + ); + expect( + new TextEncoder().encode(JSON.stringify(cache)).byteLength, + ).toBeLessThanOrEqual(WEB_HISTORY_MAX_BYTES); +}); + +test("managed clear waits for a newly appended context boundary", async () => { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + let entriesCall = 0; + const sinceCalls: Array = []; + session.getCommands = async () => ({ + commands: [ + { + name: "clear", + source: "extension", + sourceInfo: { path: "/repo/extensions/clear-context.ts" }, + }, + ], + }); + session.getEntries = async (since?: string) => { + sinceCalls.push(since); + entriesCall += 1; + return { + entries: + entriesCall === 1 + ? [{ id: "before", type: "message" }] + : [ + { id: "before", type: "message" }, + { + id: "clear", + type: "custom", + customType: "vessup:clear-context", + }, + ], + leafId: "clear", + }; + }; + session.send = async () => undefined; + + await expect(session.clear()).resolves.toEqual({ cleared: true }); + expect(entriesCall).toBe(2); + expect(sinceCalls).toEqual([undefined, "before"]); +}); + +test("managed clear treats post-dispatch history failures as uncertain", async () => { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + let entriesCall = 0; + session.getCommands = async () => ({ + commands: [ + { + name: "clear", + source: "extension", + sourceInfo: { path: "/repo/extensions/clear-context.ts" }, + }, + ], + }); + session.getEntries = async () => { + entriesCall += 1; + if (entriesCall === 1) return { entries: [{ id: "before" }] }; + throw new Error("history transport timed out"); + }; + session.send = async () => undefined; + + await expect(session.clear()).rejects.toMatchObject({ + name: "CommandDeliveryUncertainError", + }); + expect(entriesCall).toBe(2); +}); + +test("managed history falls back only for an unsupported entries command", async () => { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + session.getEntries = async () => { + throw new CommandRejectedError("Unknown command: get_entries"); + }; + session.getMessages = async () => ({ + messages: [{ role: "user", content: "fallback" }], + }); + expect(JSON.stringify(await session.getWebHistory())).toContain("fallback"); +}); + +test("managed history propagates other entries failures without falling back", async () => { + for (const failure of [ + new CommandRejectedError("Unknown command: get_entries_v2"), + new Error("entries transport failed"), + ]) { + const session = new ManagedRpcSession({ + cwd: "/repo", + onEvent: () => undefined, + onExit: () => undefined, + }); + session.getEntries = async () => { + throw failure; + }; + let messagesCalled = false; + session.getMessages = async () => { + messagesCalled = true; + return { messages: [{ role: "user", content: "fallback" }] }; + }; + + await expect(session.getWebHistory()).rejects.toThrow(failure.message); + expect(messagesCalled).toBe(false); + } +}); + test("web history reserves space for the compaction summary", () => { const history = boundedWebHistory( [ diff --git a/tests/web-prompts.test.ts b/tests/web-prompts.test.ts index b1649a6..ab575f4 100644 --- a/tests/web-prompts.test.ts +++ b/tests/web-prompts.test.ts @@ -2,6 +2,11 @@ import { afterEach, expect, test } from "bun:test"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { + isWebClearCommand, + isWebClearContextCommand, + isWebClearInvocation, +} from "../web/clear-command.ts"; import { includeWebCompactCommand, parseWebCompactCommand, @@ -76,6 +81,37 @@ test("prompt arguments preserve empty and escaped quoted values", async () => { ).rejects.toThrow('Unterminated " quote'); }); +test("web clear routing only accepts the exact command", () => { + expect( + isWebClearContextCommand({ + name: "clear", + source: "extension", + sourceInfo: { path: "/repo/extensions/clear-context.ts" }, + }), + ).toBe(true); + expect( + isWebClearContextCommand({ + name: "clear", + source: "extension", + sourceInfo: { path: "/repo/extensions/other.ts" }, + }), + ).toBe(false); + expect( + isWebClearContextCommand({ + name: "clear", + source: "prompt", + sourceInfo: { path: "/repo/prompts/clear.md" }, + }), + ).toBe(false); + expect(isWebClearCommand("/clear")).toBe(true); + expect(isWebClearCommand("/clear ")).toBe(true); + expect(isWebClearCommand("/clear now")).toBe(false); + expect(isWebClearCommand("please /clear")).toBe(false); + expect(isWebClearInvocation("/clear now")).toBe(true); + expect(isWebClearInvocation("/clear")).toBe(true); + expect(isWebClearInvocation("/clearance")).toBe(false); +}); + test("web reload routing only accepts the exact built-in command", () => { expect(isWebReloadCommand("/reload")).toBe(true); expect(isWebReloadCommand("/reload ")).toBe(true); @@ -92,7 +128,7 @@ test("web compact routing accepts optional instructions without matching prose", expect(parseWebCompactCommand("/compacted")).toBeUndefined(); }); -test("the web slash menu exposes control commands across stale native metadata", () => { +test("the web slash menu does not invent clear on stale native metadata", () => { const commands = includeWebCompactCommand( includeWebReloadCommand([ { @@ -124,6 +160,7 @@ test("the web slash menu exposes control commands across stale native metadata", (command) => command.name === "compact", ), ).toHaveLength(1); + expect(commands.some((command) => command.name === "clear")).toBe(false); }); test("web skill commands stay intact with their arguments", async () => { diff --git a/tests/web-server-catalog.test.ts b/tests/web-server-catalog.test.ts index ec3f36c..445006c 100644 --- a/tests/web-server-catalog.test.ts +++ b/tests/web-server-catalog.test.ts @@ -8,13 +8,8 @@ import { writeFile, } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { - join, -} from "node:path"; -import { - browserSocket, - waitForState, -} from "./web-server-test-helpers.ts"; +import { join } from "node:path"; +import { browserSocket, waitForState } from "./web-server-test-helpers.ts"; let child: Bun.Subprocess | undefined; let evictedDaemon: Bun.Subprocess | undefined; @@ -517,4 +512,3 @@ for await (const line of lines) { client.close(); } }, 10_000); - diff --git a/tests/web-server-cleanup-worktrees.test.ts b/tests/web-server-cleanup-worktrees.test.ts index 6901da8..367b3a5 100644 --- a/tests/web-server-cleanup-worktrees.test.ts +++ b/tests/web-server-cleanup-worktrees.test.ts @@ -11,19 +11,13 @@ import { writeFile, } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { - dirname, - join, -} from "node:path"; +import { dirname, join } from "node:path"; import type { WebSession } from "../web/protocol.ts"; import { createWebWorktree, WORKTREE_SESSION_ENTRY, } from "../web/server/worktrees.ts"; -import { - browserSocket, - waitForState, -} from "./web-server-test-helpers.ts"; +import { browserSocket, waitForState } from "./web-server-test-helpers.ts"; let child: Bun.Subprocess | undefined; let evictedDaemon: Bun.Subprocess | undefined; @@ -927,4 +921,3 @@ for await (const line of lines) { base.session.id, ]); }, 15_000); - diff --git a/tests/web-server-managed.test.ts b/tests/web-server-managed.test.ts index 8809af3..2a21339 100644 --- a/tests/web-server-managed.test.ts +++ b/tests/web-server-managed.test.ts @@ -8,9 +8,7 @@ import { writeFile, } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { - join, -} from "node:path"; +import { join } from "node:path"; import { browserSocket, sessionCommand, @@ -360,4 +358,3 @@ test("managed RPC requests fail within the configured bound when Pi wedges", asy "RPC command get_state timed out after 50ms", ); }, 10_000); - diff --git a/tests/web-server-native-compaction.test.ts b/tests/web-server-native-compaction.test.ts index 497790e..60fb17f 100644 --- a/tests/web-server-native-compaction.test.ts +++ b/tests/web-server-native-compaction.test.ts @@ -1,12 +1,7 @@ import { afterEach, expect, test } from "bun:test"; -import { - mkdtemp, - rm, -} from "node:fs/promises"; +import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { - join, -} from "node:path"; +import { join } from "node:path"; import { browserSocket, semanticHistory, @@ -376,6 +371,188 @@ test("failed native compactions do not announce completion", async () => { agent.close(); }, 10_000); +test("native sessions route web clear and announce completion", async () => { + tempDir = await mkdtemp(join(tmpdir(), "pi-kit-native-clear-test-")); + const statePath = join(tempDir, "server.json"); + child = Bun.spawn({ + cmd: ["bun", "run", "web/server/index.ts"], + cwd: process.cwd(), + env: { + ...process.env, + PI_WEB_PORT: "0", + PI_WEB_ROOT: process.cwd(), + PI_WEB_STATE_FILE: statePath, + PI_CODING_AGENT_DIR: join(tempDir, "pi-agent"), + }, + stdout: "ignore", + stderr: "ignore", + }); + const { port } = await waitForState(statePath, child); + const sessionId = `clear-${crypto.randomUUID()}`; + const socketUrl = `ws://127.0.0.1:${port}`; + const agent = new WebSocket(`${socketUrl}/ws/agent`); + await new Promise((resolve, reject) => { + agent.onopen = () => { + agent.send( + JSON.stringify({ + type: "agent.hello", + session: { + id: sessionId, + cwd: tempDir, + status: "idle", + source: "tui", + createdAt: Date.now(), + updatedAt: Date.now(), + messageCount: 0, + }, + entries: [], + }), + ); + resolve(); + }; + agent.onerror = () => reject(new Error("clear agent websocket failed")); + }); + + const routed = new Promise((resolve, reject) => { + const timeout = setTimeout( + () => reject(new Error("clear command was not routed to native Pi")), + 3_000, + ); + agent.onmessage = ({ data }) => { + const message = JSON.parse(String(data)) as { + type?: string; + requestId?: string; + command?: { type?: string }; + }; + if ( + message.type !== "agent.command" || + !message.requestId || + message.command?.type !== "clear" + ) + return; + clearTimeout(timeout); + resolve(message.requestId); + }; + }); + const result = new Promise<{ + response: unknown; + completion: string | undefined; + }>((resolve, reject) => { + const client = browserSocket(`${socketUrl}/ws/client`); + const requestId = crypto.randomUUID(); + let sent = false; + let response: unknown; + let completion: string | undefined; + const timeout = setTimeout(() => { + client.close(); + reject(new Error("native clear response timed out")); + }, 5_000); + const finish = () => { + if (response === undefined || completion === undefined) return; + clearTimeout(timeout); + client.close(); + resolve({ response, completion }); + }; + client.onopen = () => client.send(JSON.stringify({ type: "client.hello" })); + client.onmessage = ({ data }) => { + const message = JSON.parse(String(data)) as { + type?: string; + requestId?: string; + success?: boolean; + data?: unknown; + error?: string; + event?: { + type?: string; + message?: { content?: Array<{ type?: string; text?: string }> }; + }; + }; + if (message.type === "server.snapshot") + client.send(JSON.stringify({ type: "client.subscribe", sessionId })); + if (message.type === "server.history" && !sent) { + sent = true; + client.send( + JSON.stringify({ + type: "client.prompt", + requestId, + sessionId, + message: "/clear", + images: [], + }), + ); + } + if ( + message.type === "server.event" && + message.event?.type === "message_end" + ) { + completion = message.event.message?.content?.find( + (part) => part.type === "text", + )?.text; + finish(); + } + if (message.type !== "server.response" || message.requestId !== requestId) + return; + if (!message.success) { + clearTimeout(timeout); + client.close(); + reject(new Error(message.error ?? "native clear failed")); + return; + } + response = message.data; + finish(); + }; + }); + const requestId = await routed; + agent.send( + JSON.stringify({ + type: "agent.response", + requestId, + success: true, + data: { cleared: true }, + }), + ); + expect(await result).toEqual({ + response: { cleared: true }, + completion: "Context cleared.", + }); + const invalidClear = new Promise((resolve, reject) => { + const client = browserSocket(`${socketUrl}/ws/client`); + const requestId = crypto.randomUUID(); + const timeout = setTimeout(() => { + client.close(); + reject(new Error("invalid clear response timed out")); + }, 3_000); + client.onopen = () => client.send(JSON.stringify({ type: "client.hello" })); + client.onmessage = ({ data }) => { + const message = JSON.parse(String(data)) as { + type?: string; + requestId?: string; + success?: boolean; + error?: string; + }; + if (message.type === "server.snapshot") { + client.send( + JSON.stringify({ + type: "client.prompt", + requestId, + sessionId, + message: "/clear unexpected", + images: [], + }), + ); + return; + } + if (message.type !== "server.response" || message.requestId !== requestId) + return; + clearTimeout(timeout); + client.close(); + if (message.success) reject(new Error("invalid clear was accepted")); + else resolve(message.error ?? ""); + }; + }); + await expect(invalidClear).resolves.toBe("/clear does not accept arguments"); + agent.close(); +}, 10_000); + test("web reload survives a native bridge reconnect", async () => { tempDir = await mkdtemp(join(tmpdir(), "pi-kit-native-reload-test-")); const statePath = join(tempDir, "server.json"); diff --git a/tests/web-server-native-daemon.test.ts b/tests/web-server-native-daemon.test.ts index 41f3c4e..ce0c77b 100644 --- a/tests/web-server-native-daemon.test.ts +++ b/tests/web-server-native-daemon.test.ts @@ -8,9 +8,7 @@ import { writeFile, } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { - join, -} from "node:path"; +import { join } from "node:path"; import type { ServerStateFile } from "../web/protocol.ts"; import { sessionCommand, diff --git a/tests/web-server-native-queue.test.ts b/tests/web-server-native-queue.test.ts index 0f4dd67..f560afb 100644 --- a/tests/web-server-native-queue.test.ts +++ b/tests/web-server-native-queue.test.ts @@ -1,12 +1,7 @@ import { afterEach, expect, test } from "bun:test"; -import { - mkdtemp, - rm, -} from "node:fs/promises"; +import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { - join, -} from "node:path"; +import { join } from "node:path"; import { compactionLifecycle, idleQueueReplacementStartsAutomatically, diff --git a/tests/web-server-native-worktrees.test.ts b/tests/web-server-native-worktrees.test.ts index 435e955..519e47f 100644 --- a/tests/web-server-native-worktrees.test.ts +++ b/tests/web-server-native-worktrees.test.ts @@ -1,19 +1,8 @@ import { afterEach, expect, test } from "bun:test"; -import { - mkdir, - mkdtemp, - rm, - writeFile, -} from "node:fs/promises"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { - dirname, - join, -} from "node:path"; -import { - browserSocket, - waitForState, -} from "./web-server-test-helpers.ts"; +import { dirname, join } from "node:path"; +import { browserSocket, waitForState } from "./web-server-test-helpers.ts"; let child: Bun.Subprocess | undefined; let evictedDaemon: Bun.Subprocess | undefined; diff --git a/tests/web-server-projects.test.ts b/tests/web-server-projects.test.ts index 603837c..eb07300 100644 --- a/tests/web-server-projects.test.ts +++ b/tests/web-server-projects.test.ts @@ -151,4 +151,3 @@ test("Git projects expose checkout roots for submodules and bare repositories", await realpath(bareRepository), ); }); - diff --git a/tests/web-server-queue-recovery.test.ts b/tests/web-server-queue-recovery.test.ts index 2f1ba77..f8de42d 100644 --- a/tests/web-server-queue-recovery.test.ts +++ b/tests/web-server-queue-recovery.test.ts @@ -1,14 +1,7 @@ import { afterEach, expect, test } from "bun:test"; -import { - mkdir, - mkdtemp, - rm, - writeFile, -} from "node:fs/promises"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { - join, -} from "node:path"; +import { join } from "node:path"; import { browserSocket, sessionCommand, @@ -538,4 +531,3 @@ test("restored uncertain delivery is never automatic and requires explicit recon expect(prompts).toBe(0); agent.close(); }, 10_000); - diff --git a/tests/web-server-security.test.ts b/tests/web-server-security.test.ts index 29520a5..58cdfd1 100644 --- a/tests/web-server-security.test.ts +++ b/tests/web-server-security.test.ts @@ -218,9 +218,9 @@ test("Bun web server keeps tokenless clients inside localhost and same-origin tr const registrationDeadline = Date.now() + 3_000; let registered = false; while (Date.now() < registrationDeadline) { - const catalog = (await fetch( - `http://127.0.0.1:${port}/api/sessions`, - ).then((response) => response.json())) as { + const catalog = (await fetch(`http://127.0.0.1:${port}/api/sessions`).then( + (response) => response.json(), + )) as { sessions: Array<{ id: string }>; }; if (catalog.sessions.some((session) => session.id === sessionId)) { @@ -255,4 +255,3 @@ test("Bun web server keeps tokenless clients inside localhost and same-origin tr ).toEqual(["working", "idle", "idle"]); agent.close(); }, 15_000); - diff --git a/tests/web-server-test-helpers.ts b/tests/web-server-test-helpers.ts index baa65ca..c8cdb41 100644 --- a/tests/web-server-test-helpers.ts +++ b/tests/web-server-test-helpers.ts @@ -1356,4 +1356,3 @@ export async function waitForMirroredSession(port: number): Promise { } throw new Error("native Pi session did not register with the web server"); } - diff --git a/tests/web-session-queue-coordinator.test.ts b/tests/web-session-queue-coordinator.test.ts index 33f9233..184540a 100644 --- a/tests/web-session-queue-coordinator.test.ts +++ b/tests/web-session-queue-coordinator.test.ts @@ -213,6 +213,41 @@ test("queued prompts cannot steer before their required model is active", async expect(target.queue[0]?.deliveryState).toBeUndefined(); }); +test("queued clear executes as a control command and announces completion", async () => { + const target = record([{ id: "clear", message: "/clear" }]); + target.agentRunning = false; + const { coordinator, deliveries, broadcasts } = setup(target); + + await coordinator.flushWebQueue(target); + + expect(deliveries).toEqual([{ type: "clear" }]); + expect( + broadcasts.some( + (message) => + message.type === "server.event" && + message.event.type === "web_queue_delivery" && + message.event.phase === "completed", + ), + ).toBe(true); + expect(JSON.stringify(broadcasts)).toContain("Context cleared."); + expect(target.queue).toEqual([]); + coordinator.cancelWebQueueWork(target); +}); + +test("queue edits reject clear arguments", async () => { + const target = record([{ id: "clear", message: "/clear" }]); + const { coordinator, deliveries } = setup(target); + + await expect( + coordinator.routeQueueCommand(target, { + type: "replace_queue", + queue: [{ id: "clear", message: "/clear unexpected" }], + }), + ).rejects.toThrow("/clear does not accept arguments"); + expect(target.queue).toEqual([{ id: "clear", message: "/clear" }]); + expect(deliveries).toEqual([]); +}); + test("queued control commands cannot be converted into steering prompts", async () => { const target = record( [{ id: "compact", message: "/compact preserve names" }], diff --git a/tests/web-slash-command-service.test.ts b/tests/web-slash-command-service.test.ts index 106dc3a..195ba60 100644 --- a/tests/web-slash-command-service.test.ts +++ b/tests/web-slash-command-service.test.ts @@ -106,7 +106,10 @@ test("slash command projection reuses fallbacks and hides private transports", ( (path) => path, () => ({}) as ManagedRpcSession, ); - const sourceInfo = { path: "web-sessions.ts", scope: "temporary" as const }; + const sourceInfo = { + path: "/repo/extensions/clear-context.ts", + scope: "temporary" as const, + }; expect( service .toWeb( @@ -117,9 +120,10 @@ test("slash command projection reuses fallbacks and hides private transports", ( sourceInfo, }, { name: "web-reload", source: "extension", sourceInfo }, + { name: "clear", source: "extension", sourceInfo }, ], true, ) .map((command) => command.name), - ).toEqual(["compact", "reload"]); + ).toEqual(["compact", "reload", "clear"]); }); diff --git a/web/assistant-message.ts b/web/assistant-message.ts index a5e753b..987ecf0 100644 --- a/web/assistant-message.ts +++ b/web/assistant-message.ts @@ -19,19 +19,38 @@ export function assistantTerminalNotice( typeof message.errorMessage === "string" && message.errorMessage.trim() ? message.errorMessage.trim() : undefined; - // Pi's user-initiated Stop can surface as stopReason "error" with one of - // these exact runtime messages. Do not infer cancellation from arbitrary - // provider prose containing "abort", which could hide a genuine failure. - const aborted = + const stopDetail = + rawDetail + ?.trim() + .replace(/[.!?]+$/u, "") + .toLowerCase() ?? ""; + + const hasUserAbortContext = (tail: string): boolean => + /\b(before\s+pi\s+could\s+finish|user|stop|button|manual|cancel(?:led)?|by\s+the\s+user)\b/u.test( + tail, + ); + + const startsWithUserAbortPhrase = (prefix: string): boolean => { + if (!stopDetail.startsWith(prefix)) return false; + const tail = stopDetail.slice(prefix.length).trim(); + return tail === "" || hasUserAbortContext(tail); + }; + + const isUserAbortMessage = stopReason === "aborted" || - rawDetail === "This operation was aborted" || - rawDetail === "Request was aborted"; + startsWithUserAbortPhrase("request was aborted") || + startsWithUserAbortPhrase("request aborted") || + startsWithUserAbortPhrase("this operation was aborted") || + startsWithUserAbortPhrase("this operation aborted") || + startsWithUserAbortPhrase("the operation was aborted") || + startsWithUserAbortPhrase("the operation aborted"); + const detail = rawDetail ?? - (aborted + (isUserAbortMessage ? "The operation was aborted before Pi could finish." : "Pi stopped before completing the response."); - return aborted + return isUserAbortMessage ? { kind: "stopped", title: "Stopped", detail } : { kind: "error", title: "Run failed", detail }; } diff --git a/web/clear-command.ts b/web/clear-command.ts new file mode 100644 index 0000000..4e1cbeb --- /dev/null +++ b/web/clear-command.ts @@ -0,0 +1,32 @@ +export const WEB_CLEAR_CONTEXT_ENTRY = "vessup:clear-context"; + +/** + * Identify Pi's clear-context extension command rather than an unrelated + * extension/prompt/skill that happens to use the reserved name. + */ +export function isWebClearContextCommand(value: unknown): boolean { + if (!value || typeof value !== "object") return false; + const command = value as { + name?: unknown; + source?: unknown; + sourceInfo?: { path?: unknown }; + }; + return ( + command.name === "clear" && + command.source === "extension" && + typeof command.sourceInfo?.path === "string" && + /(?:^|[\\/])extensions[\\/]clear-context\.ts$/u.test( + command.sourceInfo.path, + ) + ); +} + +/** Match a clear command invocation, including one with rejected arguments. */ +export function isWebClearInvocation(text: string): boolean { + return /^\/clear(?:\s|$)/.test(text); +} + +/** Match only the argument-free clear command. */ +export function isWebClearCommand(text: string): boolean { + return /^\/clear\s*$/.test(text); +} diff --git a/web/client/app.tsx b/web/client/app.tsx index 6cc508f..7d855c2 100644 --- a/web/client/app.tsx +++ b/web/client/app.tsx @@ -14,6 +14,7 @@ import { PanelLeftClose, PanelLeftOpen, Plus } from "lucide-react"; import * as React from "react"; import { createPortal } from "react-dom"; import { agentEndTerminalNotice } from "../assistant-message"; +import { isWebClearCommand } from "../clear-command"; import { includeWebCompactCommand, parseWebCompactCommand, @@ -65,10 +66,6 @@ import { setHashSessionId, } from "./app-preferences"; import { NewSessionDialog } from "./components/new-session-dialog"; -import { - isQueuedFollowUpResponse, - shouldShowOptimisticPrompt, -} from "./composer-send"; import { DeleteSessionDialog, ForkSessionDialog, @@ -80,6 +77,10 @@ import { SidebarFilterButton, } from "./components/session-sidebar"; import { Button } from "./components/ui/button"; +import { + isQueuedFollowUpResponse, + shouldShowOptimisticPrompt, +} from "./composer-send"; import { assertClientPromptPayloadFits } from "./image-payload"; import { cn } from "./lib/utils"; import { @@ -611,6 +612,11 @@ export function App() { ) ? immediateOptimisticId : `optimistic-queued-${item.id}`; + const queuedOptimisticIds = new Set([ + optimisticId, + immediateOptimisticId, + `optimistic-queued-${item.id}`, + ]); if (event.phase === "started") { // Atomically move the follow-up out of the editable queue and into the // normal transcript before the server asks Pi to begin its turn. @@ -643,9 +649,12 @@ export function App() { entriesRef.current = next; setEntries(next); } - } else if (event.phase === "failed") { + } else if (event.phase === "completed" || event.phase === "failed") { + // Control commands do not produce an authoritative user message. + // Remove their optimistic queue bubble when delivery completes (or + // when a normal rejection rolls it back). const next = entriesRef.current.filter( - (entry) => entry.id !== optimisticId, + (entry) => !entry.id || !queuedOptimisticIds.has(entry.id), ); entriesRef.current = next; setEntries(next); @@ -1055,10 +1064,14 @@ export function App() { selectedSession?.status, ); const queuedFollowUp = !showOptimisticPrompt; - const worktreeCommand = /^\/worktree(?:\s|$)/.test(message.trim()); + const normalizedMessage = message.trim(); + const clearCommand = isWebClearCommand(normalizedMessage); + const reloadCommand = isWebReloadCommand(normalizedMessage); + const worktreeCommand = /^\/worktree(?:\s|$)/.test(normalizedMessage); const compactCommand = parseWebCompactCommand(message); const controlCommand = - isWebReloadCommand(message) || + clearCommand || + reloadCommand || compactCommand !== undefined || worktreeCommand; const optimisticallyWorking = @@ -1195,7 +1208,7 @@ export function App() { // exclusively in the queue until web_queue_delivery starts. return; } - if (isWebReloadCommand(message)) { + if (clearCommand || reloadCommand) { const next = entriesRef.current.filter( (entry) => entry.id !== optimisticId, ); diff --git a/web/history.ts b/web/history.ts index 7fc3b9a..a0e1c47 100644 --- a/web/history.ts +++ b/web/history.ts @@ -81,6 +81,9 @@ export function compactionSummaryHistoryEntry( id: `web-compaction-${typeof entry.id === "string" ? entry.id : fallbackIndex}`, parentId: null, timestamp, + ...(isClearBoundaryCompaction(entry) + ? { details: { clearContextBoundary: true } } + : {}), message: { role: "assistant", content: [ @@ -91,18 +94,34 @@ export function compactionSummaryHistoryEntry( }; } +function isClearBoundaryCompaction(entry: unknown): boolean { + if (!isRecord(entry)) return false; + if (entry.type === "compaction") + return ( + isRecord(entry.details) && entry.details.clearContextBoundary === true + ); + return ( + entry.type === "message" && + isRecord(entry.details) && + entry.details.clearContextBoundary === true + ); +} + export function boundedWebHistory( entries: readonly unknown[], options: { maxBytes?: number; maxEntries?: number } = {}, ): unknown[] { const maxBytes = options.maxBytes ?? WEB_HISTORY_MAX_BYTES; const maxEntries = options.maxEntries ?? WEB_HISTORY_MAX_ENTRIES; + const preserveTranscript = entries.some(isClearBoundaryCompaction); let source = entries; - for (let index = entries.length - 1; index >= 0; index -= 1) { - const entry = entries[index]; - if (isRecord(entry) && entry.type === "compaction") { - source = [entry, ...entries.slice(index + 1)]; - break; + if (!preserveTranscript) { + for (let index = entries.length - 1; index >= 0; index -= 1) { + const entry = entries[index]; + if (isRecord(entry) && entry.type === "compaction") { + source = [entry, ...entries.slice(index + 1)]; + break; + } } } const visible = source.flatMap((entry, index) => { @@ -112,24 +131,28 @@ export function boundedWebHistory( } return isRecord(entry) && entry.type === "message" ? [entry] : []; }); - const summary = visible.find( - (entry) => + let summary: RecordValue | undefined; + for (let index = visible.length - 1; index >= 0; index -= 1) { + const entry = visible[index]; + if ( isRecord(entry) && typeof entry.id === "string" && - entry.id.startsWith("web-compaction-"), - ); + entry.id.startsWith("web-compaction-") + ) { + summary = entry; + break; + } + } const sanitizedSummary = summary ? sanitizedEntry(summary, maxBytes) : undefined; - const selected: unknown[] = []; - let bytes = - 2 + - (sanitizedSummary && sanitizedSummary.bytes + 2 <= maxBytes - ? sanitizedSummary.bytes + 1 - : 0); - const availableEntries = - maxEntries - - (sanitizedSummary && sanitizedSummary.bytes + 2 <= maxBytes ? 1 : 0); + const selected: Array<{ entry: unknown; index: number }> = []; + const summaryForOutput = + sanitizedSummary && sanitizedSummary.bytes + 2 <= maxBytes + ? sanitizedSummary + : undefined; + let bytes = 2 + (summaryForOutput ? summaryForOutput.bytes + 1 : 0); + const availableEntries = maxEntries - (summaryForOutput ? 1 : 0); for ( let index = visible.length - 1; index >= 0 && selected.length < availableEntries; @@ -139,13 +162,21 @@ export function boundedWebHistory( if (entry === summary) continue; const sanitized = sanitizedEntry(entry, maxBytes); if (!sanitized || sanitized.bytes + bytes > maxBytes) continue; - selected.push(sanitized.entry); + selected.push({ entry: sanitized.entry, index }); bytes += sanitized.bytes + 1; } selected.reverse(); - if (sanitizedSummary && sanitizedSummary.bytes + 2 <= maxBytes) - selected.unshift(sanitizedSummary.entry); - return selected; + if (summaryForOutput) { + const summaryIndex = visible.indexOf(summary as RecordValue); + const insertionIndex = selected.findIndex( + ({ index }) => index > summaryIndex, + ); + selected.splice(insertionIndex < 0 ? selected.length : insertionIndex, 0, { + entry: summaryForOutput.entry, + index: summaryIndex, + }); + } + return selected.map(({ entry }) => entry); } export function webHistoryByteLength(entries: readonly unknown[]): number { diff --git a/web/protocol.ts b/web/protocol.ts index 31ed5c1..b77f319 100644 --- a/web/protocol.ts +++ b/web/protocol.ts @@ -410,6 +410,7 @@ export type AgentCommand = | { type: "set_thinking_level"; level: string } | { type: "shutdown" } | { type: "reload" } + | { type: "clear" } | { type: "create_worktree"; repository: string; diff --git a/web/server/agentMessages.ts b/web/server/agentMessages.ts index 7ea5007..acbe0f8 100644 --- a/web/server/agentMessages.ts +++ b/web/server/agentMessages.ts @@ -118,7 +118,7 @@ export function createAgentMessages(options: { async function handleAgentMessage( socket: Bun.ServerWebSocket, message: AgentToServerMessage, - ): Promise { + ): Promise { // Trust the /ws/agent upgrade (local-only and rejected when forwarded by // Tailscale Serve) plus per-session socket ownership. agent.hello is the // session-binding handshake: only mark the socket authenticated after the @@ -237,6 +237,7 @@ export function createAgentMessages(options: { event.event.type === "agent_start" || event.event.type === "turn_start" ) { + record.abortRequested = false; record.agentStartGeneration = (record.agentStartGeneration ?? 0) + 1; markAgentActivity(record); cancelQueueSettleFallback(record); @@ -247,9 +248,10 @@ export function createAgentMessages(options: { if (event.event.type === "agent_end" && !record.compaction) { markAgentSettling(record); record.status = - agentEndTerminalNotice(event.event)?.kind === "error" - ? "error" - : "idle"; + record.abortRequested || + agentEndTerminalNotice(event.event)?.kind !== "error" + ? "idle" + : "error"; record.agentRunning = false; scheduleQueueSettleFallback(record); lifecycleChanged = true; @@ -291,8 +293,10 @@ export function createAgentMessages(options: { // Pi emits agent_settled only when no retry, compaction, or internal // follow-up remains. It is authoritative even when an interrupted // overflow compaction last advertised willRetry=true. - if (record.status !== "error") record.status = "idle"; + if (record.abortRequested || record.status !== "error") + record.status = "idle"; record.agentRunning = false; + record.abortRequested = false; void flushWebQueue(record); } const subagentsChanged = updateSubagentsFromToolEvent( @@ -313,10 +317,7 @@ export function createAgentMessages(options: { : undefined; sessionMetadataChanged = true; } - if ( - event.event.type === "message_end" && - isRecord(event.event.message) - ) { + if (event.event.type === "message_end" && isRecord(event.event.message)) { appendRecordHistory(record, { type: "message", id: randomUUID(), @@ -341,10 +342,7 @@ export function createAgentMessages(options: { if (preview) record.preview = preview.slice(0, 180); else if (terminalNotice) record.preview = - `${terminalNotice.title}: ${terminalNotice.detail}`.slice( - 0, - 180, - ); + `${terminalNotice.title}: ${terminalNotice.detail}`.slice(0, 180); } sessionMetadataChanged = true; } diff --git a/web/server/clientMessages.ts b/web/server/clientMessages.ts index ae57504..1fa4606 100644 --- a/web/server/clientMessages.ts +++ b/web/server/clientMessages.ts @@ -1,3 +1,4 @@ +import { isWebClearCommand, isWebClearInvocation } from "../clear-command.js"; import { parseWebCompactCommand } from "../compact-command.js"; import type { ClientToServerMessage, @@ -48,6 +49,7 @@ export function createClientMessages(options: { const { enqueueWebFollowUp, webQueueEvent, + broadcastClearComplete, broadcastReloadComplete, sendSessionState, } = queue; @@ -123,11 +125,33 @@ export function createClientMessages(options: { try { if (!record) throw new Error(`Unknown session: ${message.sessionId}`); const normalizedPrompt = message.message.trim(); + const clear = isWebClearCommand(normalizedPrompt); + if (isWebClearInvocation(normalizedPrompt) && !clear) + throw new Error("/clear does not accept arguments"); const reload = isWebReloadCommand(normalizedPrompt); const compact = parseWebCompactCommand(normalizedPrompt); const worktree = parseWorktreeInvocation(message.message); let responseData: unknown; - if (reload) { + if (clear) { + if (message.images?.length) + throw new Error("/clear does not accept image attachments"); + if ( + message.streamingBehavior === "followUp" && + (record.status === "working" || + hasActiveWebSubagents(record.subagents)) + ) { + await enqueueWebFollowUp(record, { + id: message.requestId, + message: normalizedPrompt, + }); + responseData = { queued: true, reason: "followUp" }; + } else { + if (message.streamingBehavior === "steer") + throw new Error("/clear must be queued or run while Pi is idle"); + responseData = await routeCommand(record, { type: "clear" }); + broadcastClearComplete(record); + } + } else if (reload) { if (message.images?.length) throw new Error("/reload does not accept image attachments"); if ( @@ -254,7 +278,7 @@ export function createClientMessages(options: { const previousSessionId = record.id; const data = await routeCommand(record, message.command); if (message.command.type !== "abort") - await refreshManagedSession(record); + await refreshManagedSession(record, message.command.type === "clear"); const responseData = record.id !== previousSessionId && (message.command.type === "clone" || diff --git a/web/server/commandRouter.ts b/web/server/commandRouter.ts index 3782823..79b187f 100644 --- a/web/server/commandRouter.ts +++ b/web/server/commandRouter.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import { existsSync } from "node:fs"; import type { Api, Model } from "@earendil-works/pi-ai"; import { getSupportedThinkingLevels } from "@earendil-works/pi-ai"; +import { isWebClearContextCommand } from "../clear-command.js"; import { isPrivateWebSessionCommand } from "../compact-command.js"; import type { ClientCommandMessage, @@ -333,8 +334,15 @@ export function createCommandRouter(options: { record.managedIdentityOperation = undefined; } } - if (command.type !== "prompt") - return await routeCommandCore(record, command); + if (command.type !== "prompt") { + if (command.type === "abort") record.abortRequested = true; + try { + return await routeCommandCore(record, command); + } catch (error) { + if (command.type === "abort") record.abortRequested = false; + throw error; + } + } const shouldMarkWorking = record.status !== "working" || record.agentRunning !== true; @@ -426,6 +434,15 @@ export function createCommandRouter(options: { ) { return await routeQueueCommand(record, command); } + if (command.type === "clear") { + const settled = + (record.status === "idle" || record.status === "error") && + record.agentRunning !== true; + if (!settled || hasActiveWebSubagents(record.subagents)) + throw new Error( + "Wait for Pi and its subagents to become idle before clearing context", + ); + } if (command.type === "reload" && record.managed) { const settled = (record.status === "idle" || record.status === "error") && @@ -566,7 +583,8 @@ export function createCommandRouter(options: { isPrivateWebSessionCommand(command.name) || (command.source !== "extension" && command.source !== "prompt" && - command.source !== "skill") + command.source !== "skill") || + (command.name === "clear" && !isWebClearContextCommand(command)) ) return []; return [ @@ -703,6 +721,14 @@ export function createCommandRouter(options: { return undefined; case "compact": return await record.managed.compact(command.customInstructions); + case "clear": { + const result = await record.managed.clear(); + // The boundary has already been verified by ManagedRpcSession.clear; + // a metadata/history refresh is useful but must not turn a successful + // user action into a retryable clear failure. + await refreshManagedSession(record, true); + return result; + } case "extension_ui_response": return await record.managed.respondToExtensionUi(command); } diff --git a/web/server/managed-rpc-session.ts b/web/server/managed-rpc-session.ts index a2db705..d7e5705 100644 --- a/web/server/managed-rpc-session.ts +++ b/web/server/managed-rpc-session.ts @@ -1,7 +1,18 @@ import { randomUUID } from "node:crypto"; import { mkdirSync } from "node:fs"; import { fileURLToPath } from "node:url"; +import type { SessionEntry } from "@earendil-works/pi-coding-agent"; +import { + isWebClearContextCommand, + WEB_CLEAR_CONTEXT_ENTRY, +} from "../clear-command.js"; import { WEB_COMPACT_EXTENSION_COMMAND } from "../compact-command.js"; +import { + boundedWebHistory, + messagesToWebHistory, + WEB_HISTORY_MAX_BYTES, + WEB_HISTORY_MAX_ENTRIES, +} from "../history.js"; import type { RpcSessionCommand } from "../protocol.js"; import { SerializedWriter } from "./serialized-writer.js"; @@ -58,7 +69,7 @@ export class CommandDeliveryUncertainError extends Error { } export function isUncertainRpcDeliveryCommand(command: string): boolean { - return command === "prompt" || command === "compact"; + return command === "prompt" || command === "compact" || command === "clear"; } export function rpcDeliveryError(command: string, message: string): Error { @@ -67,6 +78,89 @@ export function rpcDeliveryError(command: string, message: string): Error { : new Error(message); } +function isUnsupportedGetEntriesError(error: unknown): boolean { + return ( + error instanceof CommandRejectedError && + error.message.trim() === "Unknown command: get_entries" + ); +} + +function boundedRawHistorySuffix( + entries: readonly SessionEntry[], +): SessionEntry[] { + const retained: SessionEntry[] = []; + let bytes = 2; + for (let index = entries.length - 1; index >= 0; index -= 1) { + if (retained.length >= WEB_HISTORY_MAX_ENTRIES) break; + const entry = entries[index]; + const serialized = JSON.stringify(entry); + if (serialized === undefined) continue; + const entryBytes = encoder.encode(serialized).byteLength; + if ( + entryBytes + 2 > WEB_HISTORY_MAX_BYTES || + bytes + entryBytes + 1 > WEB_HISTORY_MAX_BYTES + ) + break; + retained.push(entry); + bytes += entryBytes + 1; + } + retained.reverse(); + return retained; +} + +function hasCompleteActiveBranch( + entries: readonly SessionEntry[], + leafId: string | null, + allowedMissingParents: ReadonlySet = new Set(), +): boolean { + if (leafId === null) return true; + const byId = new Map(entries.map((entry) => [entry.id, entry])); + let current = byId.get(leafId); + if (!current) return false; + while (current.parentId) { + const parent = byId.get(current.parentId); + if (!parent) return allowedMissingParents.has(current.parentId); + current = parent; + } + return true; +} + +function intentionallyOmittedParents( + entries: readonly SessionEntry[], + retained: readonly SessionEntry[], + previous: ReadonlySet, +): Set { + const sourceIds = new Set(entries.map((entry) => entry.id)); + const retainedIds = new Set(retained.map((entry) => entry.id)); + const omitted = new Set(); + for (const entry of retained) { + const parentId = entry.parentId; + if ( + parentId && + !retainedIds.has(parentId) && + (sourceIds.has(parentId) || previous.has(parentId)) + ) + omitted.add(parentId); + } + return omitted; +} + +function activeBranchEntries( + entries: readonly SessionEntry[], + leafId: string | null, +): SessionEntry[] { + if (leafId === null) return []; + const byId = new Map(entries.map((entry) => [entry.id, entry])); + let current = (leafId ? byId.get(leafId) : undefined) ?? entries.at(-1); + const branch: SessionEntry[] = []; + while (current) { + branch.push(current); + current = current.parentId ? byId.get(current.parentId) : undefined; + } + branch.reverse(); + return branch; +} + let cachedRpcCommand: string[] | undefined; /** @@ -134,6 +228,11 @@ export class ManagedRpcSession { } | undefined; private reloadInFlight: Promise | undefined; + private webHistoryEntries: SessionEntry[] = []; + private webHistoryCursor: string | undefined; + /** Parent IDs intentionally omitted by the bounded raw-history cache. */ + private webHistoryAncestryGaps = new Set(); + private webHistorySupportsEntries: boolean | undefined; private readonly lineWriter = new SerializedWriter((line) => this.writeLineNow(line), ); @@ -595,6 +694,61 @@ export class ManagedRpcSession { return await this.send({ type: "get_messages" }); } + /** Prefer branch-aware raw entries so context filters such as /clear never erase transcript history. */ + async getWebHistory(): Promise { + if (this.webHistorySupportsEntries === false) + return messagesToWebHistory((await this.getMessages()).messages); + + let snapshot: { entries: unknown[]; leafId: string | null }; + try { + snapshot = await this.getEntries(this.webHistoryCursor); + } catch (error) { + if (isUnsupportedGetEntriesError(error)) { + this.webHistorySupportsEntries = false; + return messagesToWebHistory((await this.getMessages()).messages); + } + if (!this.webHistoryCursor) throw error; + this.resetWebHistory(); + snapshot = await this.getEntries(); + } + if (!Array.isArray(snapshot.entries)) + throw new Error("Pi returned an invalid get_entries response"); + this.webHistorySupportsEntries = true; + let entries = [ + ...this.webHistoryEntries, + ...(snapshot.entries as SessionEntry[]), + ]; + let ancestryGaps = this.webHistoryAncestryGaps; + if ( + snapshot.leafId && + !entries.some((entry) => entry.id === snapshot.leafId) + ) { + const full = await this.getEntries(); + if (!Array.isArray(full.entries)) + throw new Error("Pi returned an invalid get_entries response"); + snapshot = full; + entries = full.entries as SessionEntry[]; + ancestryGaps = new Set(); + } + if (!hasCompleteActiveBranch(entries, snapshot.leafId, ancestryGaps)) { + const full = await this.getEntries(); + if (!Array.isArray(full.entries)) + throw new Error("Pi returned an invalid get_entries response"); + snapshot = full; + entries = full.entries as SessionEntry[]; + ancestryGaps = new Set(); + } + this.webHistoryEntries = boundedRawHistorySuffix(entries); + this.webHistoryAncestryGaps = intentionallyOmittedParents( + entries, + this.webHistoryEntries, + ancestryGaps, + ); + const last = snapshot.entries.at(-1) as { id?: unknown } | undefined; + if (typeof last?.id === "string") this.webHistoryCursor = last.id; + return boundedWebHistory(activeBranchEntries(entries, snapshot.leafId)); + } + async getSessionStats(): Promise> { return await this.send({ type: "get_session_stats" }); } @@ -606,11 +760,18 @@ export class ManagedRpcSession { } async fork(entryId: string): Promise<{ text: string; cancelled: boolean }> { - return await this.send({ type: "fork", entryId }); + const result = await this.send<{ text: string; cancelled: boolean }>({ + type: "fork", + entryId, + }); + if (!result.cancelled) this.resetWebHistory(); + return result; } async clone(): Promise<{ cancelled: boolean }> { - return await this.send({ type: "clone" }); + const result = await this.send<{ cancelled: boolean }>({ type: "clone" }); + if (!result.cancelled) this.resetWebHistory(); + return result; } async compact(customInstructions?: string): Promise { @@ -635,10 +796,7 @@ export class ManagedRpcSession { : `/${WEB_COMPACT_EXTENSION_COMMAND}`; try { const [, result] = await Promise.all([ - this.send( - { type: "prompt", message }, - LONG_RUNNING_COMMAND_TIMEOUT_MS, - ), + this.send({ type: "prompt", message }, LONG_RUNNING_COMMAND_TIMEOUT_MS), completion, ]); return result; @@ -706,6 +864,58 @@ export class ManagedRpcSession { await this.send({ type: "prompt", message, images, streamingBehavior }); } + async clear(): Promise<{ cleared: true }> { + const commands = await this.getCommands(); + if (!commands.commands.some(isWebClearContextCommand)) + throw new Error("Pi clear context support is unavailable"); + const before = await this.getEntries(); + const knownEntryIds = new Set( + before.entries.flatMap((entry) => + typeof entry === "object" && entry !== null && "id" in entry + ? [String((entry as { id: unknown }).id)] + : [], + ), + ); + const lastBefore = before.entries.at(-1) as { id?: unknown } | undefined; + let since = typeof lastBefore?.id === "string" ? lastBefore.id : undefined; + await this.send( + { type: "prompt", message: "/clear" }, + LONG_RUNNING_COMMAND_TIMEOUT_MS, + ); + try { + const deadline = Date.now() + LONG_RUNNING_COMMAND_TIMEOUT_MS; + while (Date.now() < deadline) { + const snapshot = await this.getEntries(since); + const last = snapshot.entries.at(-1) as { id?: unknown } | undefined; + if (typeof last?.id === "string") since = last.id; + const marker = snapshot.entries.find( + (entry) => + typeof entry === "object" && + entry !== null && + (entry as { type?: unknown }).type === "custom" && + (entry as { customType?: unknown }).customType === + WEB_CLEAR_CONTEXT_ENTRY && + typeof (entry as { id?: unknown }).id === "string" && + !knownEntryIds.has((entry as { id: string }).id), + ); + if (marker) { + this.resetWebHistory(); + return { cleared: true }; + } + await Bun.sleep(25); + } + } catch (error) { + if (error instanceof CommandDeliveryUncertainError) throw error; + const message = error instanceof Error ? error.message : String(error); + throw new CommandDeliveryUncertainError( + `Could not verify /clear after dispatch: ${message}`, + ); + } + throw new CommandDeliveryUncertainError( + "Timed out waiting for /clear to append its context boundary", + ); + } + async worktree(message: string): Promise { const before = (await this.getState()) as { sessionId?: unknown }; const previousId = @@ -736,8 +946,10 @@ export class ManagedRpcSession { replacement && replacement.previousSessionId === previousId && replacement.replacementSessionId === state.sessionId - ) + ) { + this.resetWebHistory(); return; + } } await Bun.sleep(25); } @@ -771,6 +983,13 @@ export class ManagedRpcSession { async switchSession(sessionPath: string): Promise { await this.send({ type: "switch_session", sessionPath }); + this.resetWebHistory(); + } + + private resetWebHistory(): void { + this.webHistoryEntries = []; + this.webHistoryCursor = undefined; + this.webHistoryAncestryGaps = new Set(); } async shutdown(): Promise { diff --git a/web/server/managedSessionCreate.ts b/web/server/managedSessionCreate.ts index 96ae3d6..ebc2253 100644 --- a/web/server/managedSessionCreate.ts +++ b/web/server/managedSessionCreate.ts @@ -1,10 +1,9 @@ import { randomUUID } from "node:crypto"; -import { isAutoModelReference } from "../model-status.js"; import { agentEndTerminalNotice, assistantTerminalNotice, } from "../assistant-message.js"; -import { messagesToWebHistory } from "../history.js"; +import { isAutoModelReference } from "../model-status.js"; import type { ServerEventMessage, ServerHistoryMessage, @@ -211,6 +210,7 @@ export function createManagedSessionLauncher(options: { record.updatedAt = Date.now(); updateSubagentsFromToolEvent(record, event); if (event.type === "agent_start" || event.type === "turn_start") { + record.abortRequested = false; record.agentStartGeneration = (record.agentStartGeneration ?? 0) + 1; markAgentActivity(record); cancelQueueSettleFallback(record); @@ -268,7 +268,10 @@ export function createManagedSessionLauncher(options: { if (event.type === "agent_end" && !record.compaction) { markAgentSettling(record); record.status = - agentEndTerminalNotice(event)?.kind === "error" ? "error" : "idle"; + record.abortRequested || + agentEndTerminalNotice(event)?.kind !== "error" + ? "idle" + : "error"; record.agentRunning = false; scheduleQueueSettleFallback(record); } @@ -305,8 +308,10 @@ export function createManagedSessionLauncher(options: { // Pi emits agent_settled only when no retry, compaction, or internal // follow-up remains. It is authoritative even when an interrupted // overflow compaction last advertised willRetry=true. - if (record.status !== "error") record.status = "idle"; + if (record.abortRequested || record.status !== "error") + record.status = "idle"; record.agentRunning = false; + record.abortRequested = false; const deferredModelSelection = router.flushPendingModelSelection(record); if (deferredModelSelection) { @@ -364,14 +369,14 @@ export function createManagedSessionLauncher(options: { // notice can wait for it before broadcasting; otherwise the notice // would be wiped by this history replacement on subscribed clients. const refresh: Promise = runtimeSession - .getMessages() - .then(({ messages }) => { + .getWebHistory() + .then((history) => { if ( record.managed !== runtimeSession || runtime.sessions.get(record.id) !== record ) return; - replaceRecordHistory(record, messagesToWebHistory(messages)); + replaceRecordHistory(record, history); broadcastToSessionClients(record.id, { type: "server.history", sessionId: record.id, @@ -437,10 +442,7 @@ export function createManagedSessionLauncher(options: { runtime.sessions.set(record.id, record); } try { - replaceRecordHistory( - record, - messagesToWebHistory((await managed.getMessages()).messages), - ); + replaceRecordHistory(record, await managed.getWebHistory()); } catch { // A failed context request must not publish a blank active resume. Read only // a bounded suffix instead of hydrating the append-only session archive. diff --git a/web/server/managedSessionRefresh.ts b/web/server/managedSessionRefresh.ts index a8debba..e4b6e1a 100644 --- a/web/server/managedSessionRefresh.ts +++ b/web/server/managedSessionRefresh.ts @@ -1,7 +1,6 @@ import { randomUUID } from "node:crypto"; import { existsSync, readdirSync, renameSync, rmSync } from "node:fs"; import { join } from "node:path"; -import { messagesToWebHistory } from "../history.js"; import { replacementFromEntries } from "../worktree-replacement.js"; import type { ClientBroadcast } from "./clientBroadcast.js"; import { preserveRetryAroundQuiescence } from "./queue-mutation.js"; @@ -279,10 +278,7 @@ export function createManagedSessionRefresh(options: { activityGeneration, ); try { - replaceRecordHistory( - record, - messagesToWebHistory((await managed.getMessages()).messages), - ); + replaceRecordHistory(record, await managed.getWebHistory()); } catch { // Keep the last complete bounded history snapshot. } @@ -359,9 +355,8 @@ export function createManagedSessionRefresh(options: { for (const entry of entries) { if (!entry.isFile()) continue; const basePath = - entry.parentPath ?? (typeof entry.path === "string" - ? entry.path - : sessionsDir); + entry.parentPath ?? + (typeof entry.path === "string" ? entry.path : sessionsDir); const path = join(basePath, entry.name); const match = path.match(/^(.*\.jsonl)\.replaced-[0-9a-f-]+\.tmp$/i); if (match) tombstones.push({ tombstone: path, source: match[1] }); diff --git a/web/server/server-types.ts b/web/server/server-types.ts index fe4b7b3..f3d64d9 100644 --- a/web/server/server-types.ts +++ b/web/server/server-types.ts @@ -66,6 +66,8 @@ export type SessionRecord = { historyBytes?: number; active: boolean; agentRunning?: boolean; + /** A browser Stop explicitly cancels the current run, even if Pi reports an error-shaped abort. */ + abortRequested?: boolean; /** Internal generation used to ignore stale get_state responses from prior turns. */ modelTurnGeneration?: number; /** True while an Auto selection is being resolved for the current turn. */ diff --git a/web/server/session-queue-coordinator.ts b/web/server/session-queue-coordinator.ts index 6c6df3a..0ae028f 100644 --- a/web/server/session-queue-coordinator.ts +++ b/web/server/session-queue-coordinator.ts @@ -1,3 +1,4 @@ +import { isWebClearCommand, isWebClearInvocation } from "../clear-command.js"; import { parseWebCompactCommand } from "../compact-command.js"; import type { ClientCommandMessage, @@ -62,8 +63,8 @@ export function createSessionQueueCoordinator( sessionId: record.id, event: { type: "web_queue_update", - queue: record.queue.map(({ requiredModel: _requiredModel, ...item }) => - item, + queue: record.queue.map( + ({ requiredModel: _requiredModel, ...item }) => item, ), }, }; @@ -73,9 +74,7 @@ export function createSessionQueueCoordinator( return queue.map((item) => ({ ...item, images: item.images?.map((image) => ({ ...image })), - requiredModel: item.requiredModel - ? { ...item.requiredModel } - : undefined, + requiredModel: item.requiredModel ? { ...item.requiredModel } : undefined, })); } @@ -205,7 +204,7 @@ export function createSessionQueueCoordinator( function broadcastQueueDelivery( record: SessionRecord, item: WebQueuedMessage, - phase: "started" | "failed" | "uncertain", + phase: "started" | "completed" | "failed" | "uncertain", error?: string, ): void { broadcast(record.id, { @@ -234,6 +233,10 @@ export function createSessionQueueCoordinator( broadcastControlComplete(record, "Reload complete."); } + function broadcastClearComplete(record: SessionRecord): void { + broadcastControlComplete(record, "Context cleared."); + } + function broadcastCompactionComplete(record: SessionRecord): void { broadcastControlComplete(record, "Compaction complete."); } @@ -403,7 +406,18 @@ export function createSessionQueueCoordinator( broadcastQueueDelivery(record, item, "started"); try { const compact = parseWebCompactCommand(item.message); - if (isWebReloadCommand(item.message)) { + if ( + isWebClearInvocation(item.message) && + !isWebClearCommand(item.message) + ) + throw new Error("/clear does not accept arguments"); + if (isWebClearCommand(item.message)) { + if (item.images?.length) + throw new Error("/clear does not accept image attachments"); + await deliverCommand(record, { type: "clear" }); + broadcastQueueDelivery(record, item, "completed"); + broadcastClearComplete(record); + } else if (isWebReloadCommand(item.message)) { if (item.images?.length) throw new Error("/reload does not accept image attachments"); // Queued control commands execute through their dedicated route only after @@ -566,9 +580,15 @@ export function createSessionQueueCoordinator( ); } if ( + isWebClearInvocation(queued.message) || isWebReloadCommand(queued.message) || parseWebCompactCommand(queued.message) ) { + if ( + isWebClearInvocation(queued.message) && + !isWebClearCommand(queued.message) + ) + throw new Error("/clear does not accept arguments"); throw new Error( `${queued.message.split(/\s/, 1)[0]} must remain queued until the current run settles`, ); @@ -689,6 +709,16 @@ export function createSessionQueueCoordinator( for (const replacement of command.queue) { if (seenIds.has(replacement.id)) throw new Error(`Duplicate queue item ${replacement.id}`); + if ( + isWebClearInvocation(replacement.message) && + !isWebClearCommand(replacement.message) + ) + throw new Error("/clear does not accept arguments"); + if ( + isWebClearCommand(replacement.message) && + replacement.images?.length + ) + throw new Error("/clear does not accept image attachments"); if ( isWebReloadCommand(replacement.message) && replacement.images?.length @@ -812,6 +842,7 @@ export function createSessionQueueCoordinator( cancelWebQueueWork, broadcastWebQueue, broadcastQueueDelivery, + broadcastClearComplete, broadcastReloadComplete, broadcastCompactionComplete, sendSessionState, diff --git a/web/server/sessionHistory.ts b/web/server/sessionHistory.ts index df0f668..4448e7b 100644 --- a/web/server/sessionHistory.ts +++ b/web/server/sessionHistory.ts @@ -1,7 +1,3 @@ -import { - buildContextEntries, - type SessionEntry, -} from "@earendil-works/pi-coding-agent"; import { boundedWebHistory, WEB_HISTORY_MAX_BYTES, @@ -16,6 +12,34 @@ import type { SessionFileCatalog, SessionRecord } from "./server-types.js"; * record.history, record.historyReady, and record.historyBytes through * replaceRecordHistory. */ +function activeBranchEntries(entries: readonly unknown[]): unknown[] { + const byId = new Map( + entries.flatMap((entry) => { + if ( + typeof entry !== "object" || + entry === null || + typeof (entry as { id?: unknown }).id !== "string" + ) + return []; + return [[(entry as { id: string }).id, entry] as const]; + }), + ); + let current = entries.at(-1) as + | { id?: unknown; parentId?: unknown } + | undefined; + const branch: unknown[] = []; + const seen = new Set(); + while (current && typeof current.id === "string" && !seen.has(current.id)) { + seen.add(current.id); + branch.push(current); + if (typeof current.parentId !== "string") break; + current = byId.get(current.parentId) as + | { id?: unknown; parentId?: unknown } + | undefined; + } + return branch.reverse(); +} + export function createSessionHistory(options: { catalog: SessionFileCatalog }) { const { parseSessionFile, isRecord } = options.catalog; @@ -61,10 +85,7 @@ export function createSessionHistory(options: { catalog: SessionFileCatalog }) { if (record.file) { const scan = parseSessionFile(record.file); if (scan) { - replaceRecordHistory( - record, - buildContextEntries(scan.history as SessionEntry[]), - ); + replaceRecordHistory(record, activeBranchEntries(scan.history)); return [...record.history]; } } diff --git a/web/server/slash-command-service.ts b/web/server/slash-command-service.ts index 9e9eca9..6d819c2 100644 --- a/web/server/slash-command-service.ts +++ b/web/server/slash-command-service.ts @@ -1,3 +1,4 @@ +import { isWebClearContextCommand } from "../clear-command.js"; import { includeWebCompactCommand, isPrivateWebSessionCommand, @@ -138,10 +139,12 @@ export class SlashCommandService { .filter( (command) => !isPrivateWebSessionCommand(command.name) && - (includeExtensions || - command.source === "prompt" || - command.source === "skill" || - command.name === "worktree"), + (command.name === "clear" + ? isWebClearContextCommand(command) + : includeExtensions || + command.source === "prompt" || + command.source === "skill" || + command.name === "worktree"), ) .map((command) => ({ name: command.name,