diff --git a/packages/agent/src/acp-extensions.ts b/packages/agent/src/acp-extensions.ts index c700c32bb7..6bb9da61fa 100644 --- a/packages/agent/src/acp-extensions.ts +++ b/packages/agent/src/acp-extensions.ts @@ -69,6 +69,10 @@ export const POSTHOG_NOTIFICATIONS = { /** Marks a boundary for log compaction */ COMPACT_BOUNDARY: "_posthog/compact_boundary", + /** Conversation history was cleared via /clear. Carries the fresh SDK + * session id; rehydration treats the entry as a conversation boundary. */ + CONVERSATION_CLEARED: "_posthog/conversation_cleared", + /** Token usage update for a session turn */ USAGE_UPDATE: "_posthog/usage_update", diff --git a/packages/agent/src/adapters/base-acp-agent.ts b/packages/agent/src/adapters/base-acp-agent.ts index 243228a65a..6e804964c8 100644 --- a/packages/agent/src/adapters/base-acp-agent.ts +++ b/packages/agent/src/adapters/base-acp-agent.ts @@ -72,7 +72,7 @@ export abstract class BaseAcpAgent implements Agent { protected abstract interrupt(): Promise; async cancel(params: CancelNotification): Promise { - if (this.sessionId !== params.sessionId) { + if (!this.hasSession(params.sessionId)) { throw new Error("Session ID mismatch"); } this.session.cancelled = true; @@ -100,6 +100,8 @@ export abstract class BaseAcpAgent implements Agent { } } + /** Adapters may widen this to accept alternate ids for the live session + * (e.g. the Claude adapter's post-/clear SDK session id). */ hasSession(sessionId: string): boolean { return this.sessionId === sessionId; } @@ -108,7 +110,7 @@ export abstract class BaseAcpAgent implements Agent { sessionId: string, notification: SessionNotification, ): void { - if (this.sessionId === sessionId) { + if (this.hasSession(sessionId)) { this.session.notificationHistory.push(notification); } } diff --git a/packages/agent/src/adapters/claude/UPSTREAM.md b/packages/agent/src/adapters/claude/UPSTREAM.md index 06a4299843..1c066cc0c9 100644 --- a/packages/agent/src/adapters/claude/UPSTREAM.md +++ b/packages/agent/src/adapters/claude/UPSTREAM.md @@ -365,7 +365,11 @@ Fork of `@anthropic-ai/claude-agent-acp`. Upstream repo: https://github.com/anth - **Model alias version match** (#702, e1e1c69): Refuse cross-version alias matches in `resolveModelPreference` so `claude-opus-4-6` doesn't get copied onto the `opus` alias when it resolves to 4.7. - **Hide /clear** (#705, cfce130): `/clear` removed from advertised commands; clients should use - `session/new` for the same effect. + `session/new` for the same effect. Superseded: PostHog Code now implements `/clear` itself in + `clearConversation` (prompt() intercepts it and swaps in a fresh SDK session; a + `_posthog/conversation_cleared` log marker bounds rehydration), still never forwarding it to the SDK. + A build that predates this marker skips it as an unrecognized notification and keeps rendering + pre-clear history on rehydration, so the history-drop only renders correctly on builds that ship it. - **No-op ping events** (#698, 694221a): `streamEventToAcpNotifications` no-ops `ping` keep-alive events instead of falling through to `unreachable` and spamming stderr. diff --git a/packages/agent/src/adapters/claude/claude-agent.clear.test.ts b/packages/agent/src/adapters/claude/claude-agent.clear.test.ts new file mode 100644 index 0000000000..7c07e323c7 --- /dev/null +++ b/packages/agent/src/adapters/claude/claude-agent.clear.test.ts @@ -0,0 +1,643 @@ +import * as fs from "node:fs"; +import type { AgentSideConnection } from "@agentclientprotocol/sdk"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { POSTHOG_METHODS, POSTHOG_NOTIFICATIONS } from "../../acp-extensions"; +import { Pushable } from "../../utils/streams"; +import { getSessionJsonlPath } from "./session/jsonl-hydration"; + +type InitResult = { + result: "success"; + commands?: unknown[]; + models?: unknown[]; +}; + +type SdkQueryHandle = { + interrupt: ReturnType; + setModel: ReturnType; + setMcpServers: ReturnType; + mcpServerStatus: ReturnType; + supportedCommands: ReturnType; + initializationResult: ReturnType; + close: ReturnType; + [Symbol.asyncIterator]: () => AsyncIterator; +}; + +let nextInitPromise: Promise = Promise.resolve({ + result: "success", + commands: [], + models: [], +}); + +function makeQueryHandle(): SdkQueryHandle { + return { + interrupt: vi.fn().mockResolvedValue(undefined), + setModel: vi.fn().mockResolvedValue(undefined), + setMcpServers: vi.fn().mockResolvedValue(undefined), + mcpServerStatus: vi.fn().mockResolvedValue([]), + supportedCommands: vi.fn().mockResolvedValue([]), + initializationResult: vi.fn().mockImplementation(() => nextInitPromise), + close: vi.fn(), + [Symbol.asyncIterator]: async function* () { + /* never yields */ + } as never, + }; +} + +/** Points nextInitPromise at a deferred the test settles once the clear has + * reached its init await (after `vi.waitFor` on createdQueries). */ +function deferInit() { + let resolve!: (result: InitResult) => void; + let reject!: (error: Error) => void; + nextInitPromise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { resolve, reject }; +} + +const lastQueryCall: { options?: Record } = {}; +const createdQueries: SdkQueryHandle[] = []; + +vi.mock("@anthropic-ai/claude-agent-sdk", () => ({ + query: vi.fn((params: { options: Record }) => { + lastQueryCall.options = params.options; + const handle = makeQueryHandle(); + createdQueries.push(handle); + return handle; + }), +})); + +vi.mock("./mcp/tool-metadata", () => ({ + fetchMcpToolMetadata: vi.fn().mockResolvedValue(undefined), + getConnectedMcpServerNames: vi.fn().mockReturnValue([]), + getCachedMcpTools: vi.fn().mockReturnValue([]), + clearMcpToolMetadataCache: vi.fn(), +})); + +// Import after the mocks so ClaudeAcpAgent resolves the mocked SDK +const { ClaudeAcpAgent } = await import("./claude-agent"); +type Agent = InstanceType; + +interface ClientMocks { + sessionUpdate: ReturnType; + extNotification: ReturnType; +} + +function makeAgent(): { agent: Agent; client: ClientMocks } { + const client: ClientMocks = { + sessionUpdate: vi.fn().mockResolvedValue(undefined), + extNotification: vi.fn().mockResolvedValue(undefined), + }; + const agent = new ClaudeAcpAgent(client as unknown as AgentSideConnection); + return { agent, client }; +} + +function installFakeSession(agent: Agent, sessionId: string) { + const oldQuery = makeQueryHandle(); + const input = new Pushable(); + const endSpy = vi.spyOn(input, "end"); + const abortController = new AbortController(); + + const session = { + query: oldQuery, + sdkSessionId: sessionId, + queryOptions: { + sessionId, + cwd: "/tmp/repo", + model: "claude-sonnet-4-6", + mcpServers: { + posthog: { type: "http", url: "https://posthog" }, + "posthog-code-tools": { + type: "sdk", + name: "posthog-code-tools", + instance: { stale: true }, + }, + }, + abortController, + }, + buildInProcessMcpServers: vi.fn(() => ({ + "posthog-code-tools": { + type: "sdk" as const, + name: "posthog-code-tools", + instance: { fresh: true }, + }, + })), + localToolsServerNames: ["posthog-code-tools"], + input, + cancelled: false, + settingsManager: { dispose: vi.fn() }, + permissionMode: "default", + abortController, + accumulatedUsage: { + inputTokens: 0, + outputTokens: 0, + cachedReadTokens: 0, + cachedWriteTokens: 0, + }, + sessionResources: new Set(), + configOptions: [], + turnQueue: [] as unknown[], + activeTurn: null as unknown, + pendingOrphanResults: 0, + queryGeneration: 0, + cwd: "/tmp/repo", + notificationHistory: [] as unknown[], + taskRunId: "run-1", + lastContextWindowSize: 200_000, + modelId: "claude-sonnet-4-6", + taskState: new Map(), + }; + + (agent as unknown as { session: typeof session }).session = session; + (agent as unknown as { sessionId: string }).sessionId = sessionId; + + return { session, oldQuery, endSpy, abortController }; +} + +function findUpdate( + client: ClientMocks, + sessionUpdate: string, +): Record | undefined { + const match = client.sessionUpdate.mock.calls.find( + ([call]) => + (call as { update?: { sessionUpdate?: string } }).update + ?.sessionUpdate === sessionUpdate, + ); + return (match?.[0] as { update: Record } | undefined) + ?.update; +} + +function findExtNotification( + client: ClientMocks, + method: string, +): Record | undefined { + const match = client.extNotification.mock.calls.find( + ([calledMethod]) => calledMethod === method, + ); + return match?.[1] as Record | undefined; +} + +function findAllExtNotifications( + client: ClientMocks, + method: string, +): Record[] { + return client.extNotification.mock.calls + .filter(([calledMethod]) => calledMethod === method) + .map(([, params]) => params as Record); +} + +describe("ClaudeAcpAgent /clear", () => { + beforeEach(() => { + vi.clearAllMocks(); + lastQueryCall.options = undefined; + createdQueries.length = 0; + nextInitPromise = Promise.resolve({ + result: "success", + commands: [], + models: [], + }); + }); + + it("swaps in a fresh SDK session and emits the clear marker", async () => { + const { agent, client } = makeAgent(); + const { session, oldQuery, endSpy } = installFakeSession(agent, "s-1"); + session.taskState.set("task-1", { title: "old task" }); + + const result = await agent.prompt({ + sessionId: "s-1", + prompt: [{ type: "text", text: "/clear" }], + }); + + expect(result.stopReason).toBe("end_turn"); + + // Old query retired, new query started fresh (no resume, new id). + expect(oldQuery.interrupt).toHaveBeenCalledTimes(1); + expect(endSpy).toHaveBeenCalledTimes(1); + expect(createdQueries).toHaveLength(1); + expect(lastQueryCall.options?.resume).toBeUndefined(); + const newSessionId = lastQueryCall.options?.sessionId as string; + expect(newSessionId).toBeDefined(); + expect(newSessionId).not.toBe("s-1"); + + // The in-process local-tools server is rebuilt fresh. + const servers = lastQueryCall.options?.mcpServers as Record< + string, + { instance?: unknown } + >; + expect(servers["posthog-code-tools"].instance).toEqual({ fresh: true }); + expect(servers.posthog).toMatchObject({ type: "http" }); + + // ACP identity is stable; the SDK session id diverges underneath. + expect((agent as unknown as { sessionId: string }).sessionId).toBe("s-1"); + expect(session.sdkSessionId).toBe(newSessionId); + expect(agent.hasSession("s-1")).toBe(true); + expect(agent.hasSession(newSessionId)).toBe(true); + + // Repoints stored session ids and marks the boundary in the log. + expect( + findExtNotification(client, POSTHOG_NOTIFICATIONS.SDK_SESSION), + ).toMatchObject({ + taskRunId: "run-1", + sessionId: newSessionId, + adapter: "claude", + }); + expect( + findExtNotification(client, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED), + ).toMatchObject({ sessionId: newSessionId }); + + // A "clearing" status opens immediately and closes on success, so the + // user sees feedback for the whole swap even if it's slow. + const statusNotifications = findAllExtNotifications( + client, + POSTHOG_NOTIFICATIONS.STATUS, + ); + expect(statusNotifications).toEqual([ + { sessionId: "s-1", status: "clearing" }, + { sessionId: "s-1", status: "clearing", isComplete: true }, + ]); + + // The /clear prompt is echoed to the transcript, the plan panel resets, + // and the context indicator drops to zero. + expect(findUpdate(client, "user_message_chunk")).toMatchObject({ + content: { type: "text", text: "/clear" }, + }); + expect(session.taskState.size).toBe(0); + expect(findUpdate(client, "plan")).toMatchObject({ entries: [] }); + expect(findUpdate(client, "usage_update")).toMatchObject({ + used: 0, + size: 200_000, + }); + }); + + it("deletes the stale local jsonl for the stable ACP id after a successful clear", async () => { + // A cold reconnect hydrates by the stable ACP id (clients never learn the + // internal SDK id). If the SDK's original file under that id survived a + // /clear, a future hydration would find it, skip re-fetching the + // authoritative log, and resume the pre-clear conversation. + const unlinkSpy = vi + .spyOn(fs.promises, "unlink") + .mockResolvedValue(undefined); + const { agent } = makeAgent(); + installFakeSession(agent, "s-stale"); + + await agent.prompt({ + sessionId: "s-stale", + prompt: [{ type: "text", text: "/clear" }], + }); + + expect(unlinkSpy).toHaveBeenCalledWith( + getSessionJsonlPath("s-stale", "/tmp/repo"), + ); + unlinkSpy.mockRestore(); + }); + + it("still completes the clear if removing the stale jsonl fails for a reason other than a missing file", async () => { + const unlinkSpy = vi + .spyOn(fs.promises, "unlink") + .mockRejectedValue( + Object.assign(new Error("EACCES"), { code: "EACCES" }), + ); + const { agent } = makeAgent(); + installFakeSession(agent, "s-unlink-fails"); + + const result = await agent.prompt({ + sessionId: "s-unlink-fails", + prompt: [{ type: "text", text: "/clear" }], + }); + + expect(result.stopReason).toBe("end_turn"); + unlinkSpy.mockRestore(); + }); + + it("emits the marker after the user message so /clear sits before the boundary", async () => { + const { agent, client } = makeAgent(); + installFakeSession(agent, "s-order"); + + let clearedAt = -1; + let userMessageAt = -1; + let order = 0; + client.extNotification.mockImplementation(async (method: string) => { + if (method === POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED) { + clearedAt = order++; + } + }); + client.sessionUpdate.mockImplementation( + async (call: { update?: { sessionUpdate?: string } }) => { + if (call.update?.sessionUpdate === "user_message_chunk") { + userMessageAt = order++; + } + }, + ); + + await agent.prompt({ + sessionId: "s-order", + prompt: [{ type: "text", text: "/clear" }], + }); + + expect(userMessageAt).toBeGreaterThanOrEqual(0); + expect(clearedAt).toBeGreaterThan(userMessageAt); + }); + + it.each([ + { + name: "an active turn", + setup: (session: ReturnType["session"]) => { + session.activeTurn = { promptUuid: "u-1", settled: false }; + }, + }, + { + name: "a queued turn", + setup: (session: ReturnType["session"]) => { + session.turnQueue.push({ promptUuid: "u-2" }); + }, + }, + ])("refuses to clear while $name is in flight", async ({ setup }) => { + const { agent, client } = makeAgent(); + const { session, oldQuery } = installFakeSession(agent, "s-busy"); + setup(session); + + const result = await agent.prompt({ + sessionId: "s-busy", + prompt: [{ type: "text", text: "/clear" }], + }); + + expect(result.stopReason).toBe("end_turn"); + expect(oldQuery.interrupt).not.toHaveBeenCalled(); + expect(createdQueries).toHaveLength(0); + const chunk = findUpdate(client, "agent_message_chunk"); + expect((chunk?.content as { text?: string })?.text).toMatch( + /Cannot clear the conversation/, + ); + expect( + findExtNotification(client, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED), + ).toBeUndefined(); + }); + + it("refuses a second /clear while one is already in progress", async () => { + // ACP handlers are not serialized, so a second /clear can arrive at any + // await point of the first. Racing two swaps against the same session + // would orphan a live SDK query; the second must be refused. + const { agent, client } = makeAgent(); + installFakeSession(agent, "s-concurrent"); + const init = deferInit(); + + const first = agent.prompt({ + sessionId: "s-concurrent", + prompt: [{ type: "text", text: "/clear" }], + }); + // Let the first clear reach its init await (one replacement query live). + await vi.waitFor(() => expect(createdQueries).toHaveLength(1)); + + const second = await agent.prompt({ + sessionId: "s-concurrent", + prompt: [{ type: "text", text: "/clear" }], + }); + + expect(second.stopReason).toBe("end_turn"); + const chunk = findUpdate(client, "agent_message_chunk"); + expect((chunk?.content as { text?: string })?.text).toMatch( + /already in progress/, + ); + // The refused clear started no second swap. + expect(createdQueries).toHaveLength(1); + + init.resolve({ result: "success", commands: [], models: [] }); + await expect(first).resolves.toMatchObject({ stopReason: "end_turn" }); + expect( + findAllExtNotifications( + client, + POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED, + ), + ).toHaveLength(1); + }); + + it("ignores a cancel that arrives while a clear is in progress", async () => { + // cancel() → interrupt() targets session.query, which mid-clear is the + // half-initialized replacement; interrupting it would corrupt the swap. + const { agent, client } = makeAgent(); + installFakeSession(agent, "s-cancel-mid-clear"); + const init = deferInit(); + + const clearPromise = agent.prompt({ + sessionId: "s-cancel-mid-clear", + prompt: [{ type: "text", text: "/clear" }], + }); + await vi.waitFor(() => expect(createdQueries).toHaveLength(1)); + + await agent.cancel({ sessionId: "s-cancel-mid-clear" }); + expect(createdQueries[0].interrupt).not.toHaveBeenCalled(); + + init.resolve({ result: "success", commands: [], models: [] }); + await expect(clearPromise).resolves.toMatchObject({ + stopReason: "end_turn", + }); + expect( + findExtNotification(client, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED), + ).toBeDefined(); + }); + + it("closes the session and reports clearing_failed when the fresh session fails to initialize", async () => { + // A non-timeout failure (SDK subprocess crash) must get the same + // treatment as a timeout: terminate the unproven replacement, close the + // session, and resolve the "Clearing…" spinner as failed — never leave + // it spinning with the session half-swapped. + const { agent, client } = makeAgent(); + const { session } = installFakeSession(agent, "s-init-crash"); + const init = deferInit(); + + const promptPromise = agent.prompt({ + sessionId: "s-init-crash", + prompt: [{ type: "text", text: "/clear" }], + }); + const rejection = expect(promptPromise).rejects.toThrow( + /SDK subprocess crashed/, + ); + await vi.waitFor(() => expect(createdQueries).toHaveLength(1)); + init.reject(new Error("SDK subprocess crashed")); + await rejection; + + expect((session as unknown as { queryClosed: boolean }).queryClosed).toBe( + true, + ); + // The failed replacement query is torn down, not leaked. + expect(createdQueries).toHaveLength(1); + expect(createdQueries[0].close).toHaveBeenCalled(); + // No trace of the /clear in the log, and the spinner resolves as failed. + expect(findUpdate(client, "user_message_chunk")).toBeUndefined(); + expect( + findExtNotification(client, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED), + ).toBeUndefined(); + expect( + findAllExtNotifications(client, POSTHOG_NOTIFICATIONS.STATUS), + ).toEqual([ + { sessionId: "s-init-crash", status: "clearing" }, + { + sessionId: "s-init-crash", + status: "clearing_failed", + error: "SDK subprocess crashed", + }, + ]); + }); + + it("rejects a prompt that arrives mid-clear once the clear fails", async () => { + // A prompt racing the swap waits for the clear to settle instead of + // pushing into the retired input stream; a failed clear then surfaces + // as the usual session-ended rejection. + const { agent } = makeAgent(); + installFakeSession(agent, "s-prompt-mid-clear"); + const init = deferInit(); + + const clearPromise = agent.prompt({ + sessionId: "s-prompt-mid-clear", + prompt: [{ type: "text", text: "/clear" }], + }); + await vi.waitFor(() => expect(createdQueries).toHaveLength(1)); + + const followUp = agent.prompt({ + sessionId: "s-prompt-mid-clear", + prompt: [{ type: "text", text: "hello" }], + }); + + const clearRejection = expect(clearPromise).rejects.toThrow( + /SDK subprocess crashed/, + ); + const followUpRejection = + expect(followUp).rejects.toThrow(/session has ended/); + init.reject(new Error("SDK subprocess crashed")); + await clearRejection; + await followUpRejection; + }); + + it("rejects /clear after the session has ended", async () => { + const { agent } = makeAgent(); + const { session } = installFakeSession(agent, "s-ended"); + (session as unknown as { queryClosed: boolean }).queryClosed = true; + + await expect( + agent.prompt({ + sessionId: "s-ended", + prompt: [{ type: "text", text: "/clear" }], + }), + ).rejects.toThrow(/session has ended/); + expect(createdQueries).toHaveLength(0); + }); + + it("refreshSession resumes the post-clear SDK session", async () => { + const { agent } = makeAgent(); + installFakeSession(agent, "s-refresh"); + + await agent.prompt({ + sessionId: "s-refresh", + prompt: [{ type: "text", text: "/clear" }], + }); + const newSessionId = lastQueryCall.options?.sessionId as string; + + await agent.extMethod(POSTHOG_METHODS.REFRESH_SESSION, { + mcpServers: [ + { name: "posthog", type: "http" as const, url: "https://fresh" }, + ], + }); + + expect(lastQueryCall.options?.resume).toBe(newSessionId); + expect(lastQueryCall.options?.sessionId).toBeUndefined(); + }); + + it("times out, closes the query, and never logs the /clear prompt if the fresh session never finishes initializing", async () => { + vi.useFakeTimers(); + try { + const { agent, client } = makeAgent(); + const { session } = installFakeSession(agent, "s-timeout"); + nextInitPromise = new Promise(() => { + // Never resolves, forcing the initializationResult() race to time out. + }); + + const promptPromise = agent.prompt({ + sessionId: "s-timeout", + prompt: [{ type: "text", text: "/clear" }], + }); + const rejection = expect(promptPromise).rejects.toThrow(/timed out/); + + // Matches the module-private SESSION_VALIDATION_TIMEOUT_MS in claude-agent.ts. + await vi.advanceTimersByTimeAsync(30_000); + await rejection; + + expect((session as unknown as { queryClosed: boolean }).queryClosed).toBe( + true, + ); + // The /clear prompt is only broadcast (and thus logged) once the new + // session is confirmed live, so a timeout must leave no trace of it. + expect(findUpdate(client, "user_message_chunk")).toBeUndefined(); + expect( + findExtNotification(client, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED), + ).toBeUndefined(); + // The "clearing" spinner opened, then the failure closes it out. + expect( + findAllExtNotifications(client, POSTHOG_NOTIFICATIONS.STATUS), + ).toEqual([ + { sessionId: "s-timeout", status: "clearing" }, + { + sessionId: "s-timeout", + status: "clearing_failed", + error: "Conversation clear timed out after 30000ms", + }, + ]); + } finally { + vi.useRealTimers(); + } + }); + + it("resumeSession after /clear echoes the canonical ACP id when matched via the new SDK id", async () => { + const { agent } = makeAgent(); + installFakeSession(agent, "s-reconnect"); + + await agent.prompt({ + sessionId: "s-reconnect", + prompt: [{ type: "text", text: "/clear" }], + }); + const newSessionId = lastQueryCall.options?.sessionId as string; + + const response = await agent.resumeSession({ + sessionId: newSessionId, + cwd: "/tmp/repo", + }); + + expect((response as unknown as { sessionId: string }).sessionId).toBe( + "s-reconnect", + ); + }); + + it("resets pre-clear plan and notification state so it can't resurface after /clear", async () => { + // ExitPlanMode falls back to lastPlanContent/fileContentCache/ + // notificationHistory when its tool input omits an explicit plan; left + // untouched, a plan written before /clear (possibly carrying + // repo-injected content) could resurface in the fresh session. + const { agent } = makeAgent(); + const { session } = installFakeSession(agent, "s-plan"); + (session as unknown as { lastPlanContent?: string }).lastPlanContent = + "stale pre-clear plan"; + (session as unknown as { lastPlanFilePath?: string }).lastPlanFilePath = + "/tmp/repo/.claude/plans/old.md"; + session.notificationHistory.push({ type: "assistant", text: "old" }); + agent.fileContentCache["/tmp/repo/.claude/plans/old.md"] = "stale content"; + + await agent.prompt({ + sessionId: "s-plan", + prompt: [{ type: "text", text: "/clear" }], + }); + + expect( + (session as unknown as { lastPlanContent?: string }).lastPlanContent, + ).toBeUndefined(); + expect( + (session as unknown as { lastPlanFilePath?: string }).lastPlanFilePath, + ).toBeUndefined(); + // The reset happens before broadcastUserMessage, which legitimately logs + // the "/clear" command itself afterward — assert the stale pre-clear + // entry is gone rather than the history being empty. + expect(session.notificationHistory).not.toContainEqual({ + type: "assistant", + text: "old", + }); + expect(agent.fileContentCache).toEqual({}); + }); +}); diff --git a/packages/agent/src/adapters/claude/claude-agent.refresh.test.ts b/packages/agent/src/adapters/claude/claude-agent.refresh.test.ts index 4222b68dd5..a85e685237 100644 --- a/packages/agent/src/adapters/claude/claude-agent.refresh.test.ts +++ b/packages/agent/src/adapters/claude/claude-agent.refresh.test.ts @@ -99,6 +99,7 @@ function installFakeSession( const session = { query: oldQuery, + sdkSessionId: sessionId, queryOptions: { sessionId, cwd: "/tmp/repo", diff --git a/packages/agent/src/adapters/claude/claude-agent.slash-command.test.ts b/packages/agent/src/adapters/claude/claude-agent.slash-command.test.ts index 0e9f422a08..a8fff01c3a 100644 --- a/packages/agent/src/adapters/claude/claude-agent.slash-command.test.ts +++ b/packages/agent/src/adapters/claude/claude-agent.slash-command.test.ts @@ -45,6 +45,7 @@ function installFakeSession( const session = { query, + sdkSessionId: sessionId, queryOptions: { sessionId, cwd: "/tmp/repo", abortController }, buildInProcessMcpServers: () => ({}), localToolsServerNames: [] as string[], diff --git a/packages/agent/src/adapters/claude/claude-agent.ts b/packages/agent/src/adapters/claude/claude-agent.ts index 14974dd3c6..5435fd0cd1 100644 --- a/packages/agent/src/adapters/claude/claude-agent.ts +++ b/packages/agent/src/adapters/claude/claude-agent.ts @@ -66,6 +66,7 @@ import { type PostHogProductId, } from "../../posthog-products"; import type { PostHogAPIConfig } from "../../types"; +import { text } from "../../utils/acp-content"; import { isCloudRun, unreachable, @@ -109,6 +110,7 @@ import { } from "./mcp/tool-metadata"; import { canUseTool } from "./permissions/permission-handlers"; import { getAvailableSlashCommands } from "./session/commands"; +import { getSessionJsonlPath } from "./session/jsonl-hydration"; import { parseMcpServers } from "./session/mcp-config"; import { applyAvailableModelsAllowlist, @@ -473,6 +475,19 @@ export class ClaudeAcpAgent extends BaseAcpAgent { isLocalOnlyCommand = true; } + if (commandMatch?.[1] === "/clear") { + // Handled by the adapter, never forwarded to the SDK (whose own /clear + // is unreliable in this embedding — see UPSTREAM.md "Hide /clear"). + return this.clearConversation(params); + } + + if (this.session.clearing) { + // A /clear is swapping the SDK query underneath. Wait for it to settle + // so this prompt lands on the fresh input stream, not the retired one + // (a failed clear sets queryClosed, which the check below rejects). + await this.session.clearing; + } + if (commandMatch && !isLocalOnlyCommand) { await this.refreshSlashCommandsForPrompt(commandMatch[1]); } @@ -1430,6 +1445,15 @@ export class ClaudeAcpAgent extends BaseAcpAgent { if (session.queryClosed) { return; } + if (session.clearing) { + // A /clear is swapping the SDK query: there is no turn to cancel, and + // interrupting the half-initialized replacement would corrupt the swap. + // A wedged clear self-limits via SESSION_VALIDATION_TIMEOUT_MS. + this.logger.debug("Ignoring cancel while a /clear is in progress", { + sessionId: this.sessionId, + }); + return; + } session.cancelled = true; // Settle not-yet-echoed turns immediately; the SDK still runs their @@ -1515,10 +1539,304 @@ export class ClaudeAcpAgent extends BaseAcpAgent { return { refreshed: true }; } + /** Retire the current consumer and SDK query so a replacement Query can be + * swapped in place (refreshSession, clearConversation). The generation bump + * makes the retired consumer exit quietly. */ + private async retireQuery(session: Session): Promise { + session.queryGeneration += 1; + const oldConsumer = session.consumer; + session.consumer = undefined; + session.cancelController?.abort(); + session.cancelController = undefined; + + // Abort FIRST so any stuck in-flight HTTP request unblocks — otherwise + // interrupt() can deadlock waiting on an API call that never returns. + // Callers allocate a fresh controller for the new Query so aborting + // the old one doesn't poison it. + session.abortController.abort(); + try { + await session.query.interrupt(); + } catch (error) { + this.logger.debug("Ignoring interrupt error while retiring query", { + sessionId: this.sessionId, + error: error instanceof Error ? error.message : String(error), + }); + } + session.input.end(); + if (oldConsumer) { + // Bounded so a wedged old query can't block the swap. + await withTimeout(oldConsumer, 5_000); + } + } + + /** + * `/clear` — drop the conversation and start over in place. + * + * The SDK's own /clear is not forwarded (see UPSTREAM.md "Hide /clear"); + * instead the current Query is retired and a brand-new SDK session (fresh + * session id, no resume) is swapped in under the same ACP session. The + * session log stays append-only: a `conversation_cleared` marker records + * the boundary (rehydration rebuilds only post-clear turns) and an updated + * `sdk_session` mapping points future resumes at the fresh SDK session. + */ + private async clearConversation( + params: PromptRequest, + ): Promise { + const session = this.session; + if (session.queryClosed) { + throw RequestError.internalError(undefined, SESSION_ENDED_MESSAGE); + } + // A second /clear mid-swap would race the same session fields + // (query/input/abortController) and orphan a live SDK query; a clear + // mid-turn would rip the query out from under the active prompt. + const refusal = session.clearing + ? "A conversation clear is already in progress." + : session.activeTurn !== null || session.turnQueue.length > 0 + ? "Cannot clear the conversation while a turn is in progress. Wait for it to finish (or cancel it) and try again." + : null; + if (refusal) { + await this.client.sessionUpdate({ + sessionId: params.sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: text(refusal), + }, + }); + return { stopReason: "end_turn" }; + } + + // Claim the session synchronously, before the first await: ACP handlers + // are not serialized, so a prompt/cancel/second clear can arrive at any + // await point of the swap. They key off this flag (see Session.clearing). + // performClear runs synchronously up to its first await, so the claim is + // visible before any other handler can interleave. Waiters only need + // settlement; a failure still surfaces through the returned promise. + const clear = this.performClear(params, session); + session.clearing = clear.then( + () => undefined, + () => undefined, + ); + try { + return await clear; + } finally { + session.clearing = undefined; + } + } + + /** Body of {@link clearConversation}; only runs holding `session.clearing`. */ + private async performClear( + params: PromptRequest, + session: Session, + ): Promise { + this.logger.info("Clearing conversation", { sessionId: params.sessionId }); + + // Signal the in-progress state immediately (mirrors the "compacting" + // status row): the swap below is normally sub-second, but on a slow or + // timed-out clear the user would otherwise see no feedback at all + // between typing `/clear` and either the divider or an error appearing. + await this.client.extNotification(POSTHOG_NOTIFICATIONS.STATUS, { + sessionId: params.sessionId, + status: "clearing", + }); + + let newQuery: Query | undefined; + let newAbortController: AbortController | undefined; + try { + await this.retireQuery(session); + + const newSdkSessionId = uuidv7(); + newAbortController = new AbortController(); + const { + resume: _dropResume, + forkSession: _dropFork, + ...rest + } = session.queryOptions; + + // Rebuild the in-process ("sdk") server fresh; reusing the prior instance + // throws "Already connected to a transport". + const freshInProcess = session.buildInProcessMcpServers(); + + const newOptions: Options = { + ...rest, + mcpServers: { + ...externalMcpServers(rest.mcpServers), + ...freshInProcess, + }, + sessionId: newSdkSessionId, + abortController: newAbortController, + // `rest.model` is the creation-time value; the user may have switched + // models since, so re-root the new Query on the live session model. + ...(session.modelId && { model: toSdkModelId(session.modelId) }), + }; + + const newInput = new Pushable(); + newQuery = query({ prompt: newInput, options: newOptions }); + + session.query = newQuery; + session.input = newInput; + session.queryOptions = newOptions; + session.abortController = newAbortController; + + const result = await withTimeout( + newQuery.initializationResult(), + SESSION_VALIDATION_TIMEOUT_MS, + ); + if (result.result === "timeout") { + throw new Error( + `Conversation clear timed out after ${SESSION_VALIDATION_TIMEOUT_MS}ms`, + ); + } + return await this.finishClear( + params, + session, + newQuery, + newSdkSessionId, + result.value, + ); + } catch (error) { + // The old query is already retired and the new one is unproven, so any + // failure here — timeout, SDK init rejection, a consumer that died while + // being retired — leaves the session unusable. Close it out and report + // the outcome (same as a failed compaction) rather than leaving the + // "Clearing…" spinner unresolved and the session half-swapped. + if (newQuery && newAbortController) { + this.terminateQuery(newQuery, newAbortController); + } + session.queryClosed = true; + const message = error instanceof Error ? error.message : String(error); + try { + await this.client.extNotification(POSTHOG_NOTIFICATIONS.STATUS, { + sessionId: params.sessionId, + status: "clearing_failed", + error: message, + }); + } catch { + // The client transport itself is failing; don't mask the cause. + } + throw new RequestError(-32603, message, { sessionId: params.sessionId }); + } + } + + /** Post-swap bookkeeping and notifications once the fresh SDK session is + * confirmed live. Failures still propagate to performClear's catch: the + * log may already show the /clear, but the session state is authoritative + * only once everything (marker included) has been persisted. */ + private async finishClear( + params: PromptRequest, + session: Session, + newQuery: Query, + newSdkSessionId: string, + initResult: Awaited>, + ): Promise { + session.knownSlashCommands = collectKnownSlashCommands(initResult.commands); + session.fastModeEnabled = fastModeStateEnabled(initResult.fast_mode_state); + + // Future resumes (refreshSession, desktop reconnect, cloud rehydration) + // must target the fresh SDK session. `this.sessionId` (the ACP-visible + // id) stays stable — clients keep addressing the session with it. + session.sdkSessionId = newSdkSessionId; + + // Invalidate the local jsonl the SDK wrote under the stable ACP id + // (non-empty only before a session's first /clear — later clears never + // write there again). Left in place, a cold reconnect that hydrates by + // this id would find it, skip re-fetching the authoritative log, and + // resume the pre-clear conversation instead of the cleared one. + try { + await fs.promises.unlink( + getSessionJsonlPath(this.sessionId, session.cwd), + ); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + this.logger.warn("Failed to remove stale session jsonl after /clear", { + sessionId: this.sessionId, + error: error instanceof Error ? error.message : String(error), + }); + } + } + + const hadTasks = session.taskState.size > 0; + session.taskState.clear(); + this.toolUseStreamCache.clear(); + this.emittedToolCalls.clear(); + // Nothing from before the boundary should be able to reach the fresh + // session: reset the plan/notification state ExitPlanMode falls back + // to when its tool input omits an explicit plan, so a stale (possibly + // repo-injected) pre-clear plan can't resurface after approval. + session.notificationHistory.length = 0; + session.lastPlanFilePath = undefined; + session.lastPlanContent = undefined; + this.fileContentCache = {}; + + // Only broadcast (and thus persist) the "/clear" prompt once the new + // session is confirmed live — the log must never show a "/clear" whose + // clear never actually happened. Broadcast before the marker so it lands + // on the pre-clear side of the rehydration boundary and gets dropped + // rather than replayed as a turn after resume. + await this.broadcastUserMessage(params); + + // These notifications are independent of one another (only the + // user-message broadcast above must precede them); issue them + // concurrently rather than paying sequential round trips. + const postClearNotifications: Promise[] = [ + // Clear the "Clearing…" spinner. `conversation_cleared` normally + // supersedes it visually, but signal completion explicitly (same + // rationale as the compacting spinner) rather than relying on that. + this.client.extNotification(POSTHOG_NOTIFICATIONS.STATUS, { + sessionId: params.sessionId, + status: "clearing", + isComplete: true, + }), + ]; + if (session.taskRunId) { + postClearNotifications.push( + this.client.extNotification(POSTHOG_NOTIFICATIONS.SDK_SESSION, { + taskRunId: session.taskRunId, + sessionId: newSdkSessionId, + adapter: "claude", + }), + ); + } + postClearNotifications.push( + this.client.extNotification(POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED, { + sessionId: newSdkSessionId, + }), + ); + if (hadTasks) { + postClearNotifications.push( + this.client.sessionUpdate({ + sessionId: params.sessionId, + update: { sessionUpdate: "plan", entries: [] }, + }), + ); + } + postClearNotifications.push( + this.client.sessionUpdate({ + sessionId: params.sessionId, + update: { + sessionUpdate: "usage_update", + used: 0, + size: + session.lastContextWindowSize ?? + this.getContextWindowForModel(session.modelId ?? ""), + }, + }), + ); + await Promise.all(postClearNotifications); + + this.refreshMcpMetadata(newQuery); + return { stopReason: "end_turn" }; + } + private async refreshSession( mcpServers: Record, ): Promise { const prev = this.session; + if (prev.clearing) { + throw new RequestError( + -32002, + "Cannot refresh session while a conversation clear is in progress", + ); + } if (prev.activeTurn !== null || prev.turnQueue.length > 0) { throw new RequestError( -32002, @@ -1537,31 +1855,7 @@ export class ClaudeAcpAgent extends BaseAcpAgent { sessionId: this.sessionId, }); - // Retire the old consumer: the generation bump makes it exit quietly. - prev.queryGeneration += 1; - const oldConsumer = prev.consumer; - prev.consumer = undefined; - prev.cancelController?.abort(); - prev.cancelController = undefined; - - // Abort FIRST so any stuck in-flight HTTP request unblocks — otherwise - // interrupt() can deadlock waiting on an API call that never returns. - // We allocate a fresh controller for the new Query below so aborting - // the old one doesn't poison it. - prev.abortController.abort(); - try { - await prev.query.interrupt(); - } catch (error) { - this.logger.debug("Ignoring interrupt error during session refresh", { - sessionId: this.sessionId, - error: error instanceof Error ? error.message : String(error), - }); - } - prev.input.end(); - if (oldConsumer) { - // Bounded so a wedged old query can't block the refresh. - await withTimeout(oldConsumer, 5_000); - } + await this.retireQuery(prev); // Reuse every option from the running session; swap mcpServers, re-root // identity on `resume` instead of `sessionId`, and give the new Query a @@ -1582,7 +1876,7 @@ export class ClaudeAcpAgent extends BaseAcpAgent { const newOptions: Options = { ...rest, mcpServers: { ...mcpServers, ...freshInProcess }, - resume: this.sessionId, + resume: prev.sdkSessionId, forkSession: false, abortController: newAbortController, // `rest.model` is the creation-time value; the user may have switched @@ -2090,6 +2384,7 @@ export class ClaudeAcpAgent extends BaseAcpAgent { const session: Session = { query: q, + sdkSessionId: sessionId, queryOptions: options, buildInProcessMcpServers, localToolsServerNames, @@ -2387,10 +2682,18 @@ export class ClaudeAcpAgent extends BaseAcpAgent { }); } + /** Matches the ACP session id, or the underlying SDK session id after a + * /clear (desktop hosts re-key on the sdk_session notification). */ + hasSession(sessionId: string): boolean { + return ( + super.hasSession(sessionId) || this.session?.sdkSessionId === sessionId + ); + } + private getExistingSessionState( sessionId: string, ): NewSessionResponse | null { - if (this.sessionId !== sessionId || !this.session) return null; + if (!this.hasSession(sessionId) || !this.session) return null; const availableModes = getAvailableModes(); const modes: SessionModeState = { @@ -2403,7 +2706,10 @@ export class ClaudeAcpAgent extends BaseAcpAgent { }; return { - sessionId, + // Echo the canonical ACP session id even if the caller matched via the + // post-/clear SDK session id (see hasSession) — clients must keep + // addressing the session with the stable id. + sessionId: this.sessionId, modes, configOptions: this.session.configOptions, }; @@ -2520,7 +2826,11 @@ export class ClaudeAcpAgent extends BaseAcpAgent { ): Promise { let info: Awaited>; try { - info = await getSessionInfo(sessionId, { dir: session.cwd }); + // The SDK stores session info under the SDK session id, which diverges + // from the client-addressed id after a /clear. + info = await getSessionInfo(session.sdkSessionId, { + dir: session.cwd, + }); } catch (error) { this.logger.warn("Failed to read session info for title update", { sessionId, diff --git a/packages/agent/src/adapters/claude/session/commands.test.ts b/packages/agent/src/adapters/claude/session/commands.test.ts new file mode 100644 index 0000000000..ccda4d7793 --- /dev/null +++ b/packages/agent/src/adapters/claude/session/commands.test.ts @@ -0,0 +1,46 @@ +import type { SlashCommand } from "@anthropic-ai/claude-agent-sdk"; +import { describe, expect, it } from "vitest"; +import { getAvailableSlashCommands } from "./commands"; + +function sdkCommand(name: string, description = ""): SlashCommand { + return { name, description, argumentHint: null } as unknown as SlashCommand; +} + +describe("getAvailableSlashCommands", () => { + it("filters unsupported commands", () => { + const available = getAvailableSlashCommands([ + sdkCommand("compact"), + sdkCommand("context"), + sdkCommand("cost"), + sdkCommand("login"), + ]); + const names = available.map((c) => c.name); + expect(names).toContain("compact"); + expect(names).not.toContain("context"); + expect(names).not.toContain("cost"); + expect(names).not.toContain("login"); + }); + + it("passes the SDK's /clear entry through", () => { + const available = getAvailableSlashCommands([ + sdkCommand("clear", "Clear conversation history"), + ]); + const clear = available.filter((c) => c.name === "clear"); + expect(clear).toHaveLength(1); + expect(clear[0].description).toBe("Clear conversation history"); + }); + + it("injects /clear when the SDK does not advertise it", () => { + const available = getAvailableSlashCommands([sdkCommand("compact")]); + const clear = available.find((c) => c.name === "clear"); + expect(clear).toBeDefined(); + expect(clear?.description).toMatch(/clear conversation history/i); + }); + + it("renames MCP commands to the mcp: prefix", () => { + const available = getAvailableSlashCommands([ + sdkCommand("linear (MCP)", "Linear tools"), + ]); + expect(available.map((c) => c.name)).toContain("mcp:linear"); + }); +}); diff --git a/packages/agent/src/adapters/claude/session/commands.ts b/packages/agent/src/adapters/claude/session/commands.ts index 47037142cf..7bfc2f95c8 100644 --- a/packages/agent/src/adapters/claude/session/commands.ts +++ b/packages/agent/src/adapters/claude/session/commands.ts @@ -2,7 +2,6 @@ import type { AvailableCommand } from "@agentclientprotocol/sdk"; import type { SlashCommand } from "@anthropic-ai/claude-agent-sdk"; const UNSUPPORTED_COMMANDS = [ - "clear", "context", "cost", "keybindings-help", @@ -16,7 +15,7 @@ const UNSUPPORTED_COMMANDS = [ export function getAvailableSlashCommands( commands: SlashCommand[], ): AvailableCommand[] { - return commands + const available = commands .map((command) => { const input = command.argumentHint != null @@ -40,4 +39,14 @@ export function getAvailableSlashCommands( (command: AvailableCommand) => !UNSUPPORTED_COMMANDS.includes(command.name), ); + // /clear is implemented by the adapter (clearConversation), not forwarded + // to the SDK — advertise it even when the SDK doesn't list it. + if (!available.some((command) => command.name === "clear")) { + available.push({ + name: "clear", + description: "Clear conversation history and free up context", + input: null, + }); + } + return available; } diff --git a/packages/agent/src/adapters/claude/session/jsonl-hydration.test.ts b/packages/agent/src/adapters/claude/session/jsonl-hydration.test.ts index 2bb4f3d38e..a99d544b93 100644 --- a/packages/agent/src/adapters/claude/session/jsonl-hydration.test.ts +++ b/packages/agent/src/adapters/claude/session/jsonl-hydration.test.ts @@ -3,6 +3,7 @@ import * as os from "node:os"; import * as path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { PostHogAPIClient } from "../../../posthog-api"; +import { createNotification } from "../../../sagas/test-fixtures"; import type { StoredEntry } from "../../../types"; import { conversationTurnsToJsonlEntries, @@ -117,6 +118,47 @@ describe("rebuildConversation", () => { expect(rebuildConversation(entries)).toEqual([]); }); + it.each([ + { method: "_posthog/conversation_cleared" }, + { method: "__posthog/conversation_cleared" }, + ])("drops turns before a $method marker (/clear boundary)", ({ method }) => { + const turns = rebuildConversation([ + entry("user_message", { content: { type: "text", text: "old" } }), + entry("agent_message", { + content: { type: "text", text: "old reply" }, + }), + createNotification(method, { sessionId: "sdk-new" }), + entry("user_message", { content: { type: "text", text: "new" } }), + entry("agent_message", { + content: { type: "text", text: "new reply" }, + }), + ]); + + expect(turns).toHaveLength(2); + expect(turns[0]).toMatchObject({ + role: "user", + content: [{ type: "text", text: "new" }], + }); + expect(turns[1]).toMatchObject({ + role: "assistant", + content: [{ type: "text", text: "new reply" }], + }); + }); + + it("drops an in-progress assistant turn at a clear marker", () => { + const turns = rebuildConversation([ + entry("user_message", { content: { type: "text", text: "old" } }), + entry("agent_message_chunk", { + content: { type: "text", text: "partial" }, + }), + createNotification("_posthog/conversation_cleared", { + sessionId: "sdk-new", + }), + ]); + + expect(turns).toEqual([]); + }); + it("produces a single user turn from user_message", () => { const turns = rebuildConversation([ entry("user_message", { diff --git a/packages/agent/src/adapters/claude/session/jsonl-hydration.ts b/packages/agent/src/adapters/claude/session/jsonl-hydration.ts index 65cf55fe65..0c07f0389b 100644 --- a/packages/agent/src/adapters/claude/session/jsonl-hydration.ts +++ b/packages/agent/src/adapters/claude/session/jsonl-hydration.ts @@ -3,6 +3,7 @@ import * as fs from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; import type { ContentBlock } from "@agentclientprotocol/sdk"; +import { isNotification, POSTHOG_NOTIFICATIONS } from "../../../acp-extensions"; import { DEFAULT_GATEWAY_MODEL } from "../../../gateway-models"; import type { PostHogAPIClient } from "../../../posthog-api"; import type { StoredEntry } from "../../../types"; @@ -110,7 +111,7 @@ export function getSessionJsonlPath(sessionId: string, cwd: string): string { export function rebuildConversation( entries: StoredEntry[], ): ConversationTurn[] { - const turns: ConversationTurn[] = []; + let turns: ConversationTurn[] = []; let currentAssistantContent: ContentBlock[] = []; let currentToolCalls: ToolCallInfo[] = []; @@ -118,6 +119,15 @@ export function rebuildConversation( const method = entry.notification?.method; const params = entry.notification?.params as Record; + // /clear starts an empty conversation: everything before the marker is + // gone from the model's context and must not be rehydrated. + if (isNotification(method, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED)) { + turns = []; + currentAssistantContent = []; + currentToolCalls = []; + continue; + } + if (method === "session/update" && params?.update) { const update = params.update as SessionUpdate; diff --git a/packages/agent/src/adapters/claude/types.ts b/packages/agent/src/adapters/claude/types.ts index 82317f536a..9ae775cccc 100644 --- a/packages/agent/src/adapters/claude/types.ts +++ b/packages/agent/src/adapters/claude/types.ts @@ -55,6 +55,9 @@ export type Turn = { export type Session = BaseSession & { query: Query; + /** Id of the underlying SDK session. Equal to the ACP session id until a + * /clear swaps in a fresh SDK session; resume/refresh must target this id. */ + sdkSessionId: string; /** The Options object passed to query() — mutating it affects subsequent prompts */ queryOptions: Options; /** Rebuilds the in-process ("sdk") signed-commit server with a fresh instance @@ -106,6 +109,10 @@ export type Session = BaseSession & { queryGeneration: number; /** The query iterator ended and can't be revived; new prompts reject. */ queryClosed?: boolean; + /** Set while a /clear is swapping the SDK query; resolves when it settles + * (success or failure). Prompts await it, cancel/refresh refuse during it, + * and a second /clear is rejected — the swap must never be raced. */ + clearing?: Promise; cancelController?: AbortController; forceCancelTimer?: ReturnType; emitRawSDKMessages: boolean | SDKMessageFilter[]; diff --git a/packages/agent/src/sagas/resume-saga.test.ts b/packages/agent/src/sagas/resume-saga.test.ts index a3919c6de7..1f4885fa3a 100644 --- a/packages/agent/src/sagas/resume-saga.test.ts +++ b/packages/agent/src/sagas/resume-saga.test.ts @@ -112,6 +112,41 @@ describe("ResumeSaga", () => { expect(result.data.conversation[3].role).toBe("assistant"); }); + it("drops turns before a conversation_cleared marker (/clear boundary)", async () => { + (mockApiClient.getTaskRun as ReturnType).mockResolvedValue( + createTaskRun(), + ); + ( + mockApiClient.fetchTaskRunLogs as ReturnType + ).mockResolvedValue([ + createUserMessage("Old prompt"), + createAgentChunk("Old reply"), + createNotification(POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED, { + sessionId: "session-cleared", + }), + createUserMessage("New prompt"), + createAgentChunk("New reply"), + ]); + + const saga = new ResumeSaga(mockLogger); + const result = await saga.run({ + taskId: "task-1", + runId: "run-1", + repositoryPath: repo.path, + apiClient: mockApiClient, + }); + + expect(result.success).toBe(true); + if (!result.success) return; + + expect(result.data.conversation).toHaveLength(2); + expect(result.data.conversation[0]).toMatchObject({ + role: "user", + content: [{ type: "text", text: "New prompt" }], + }); + expect(result.data.conversation[1].role).toBe("assistant"); + }); + it("merges consecutive text chunks", async () => { (mockApiClient.getTaskRun as ReturnType).mockResolvedValue( createTaskRun(), @@ -595,6 +630,27 @@ describe("ResumeSaga", () => { entries: () => [createUserMessage("Hello"), createAgentChunk("Hi")], expected: null, }, + { + name: "a conversation_cleared marker supersedes an earlier run_started", + entries: () => [ + runStarted("session-old"), + createUserMessage("Hello"), + createNotification(POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED, { + sessionId: "session-cleared", + }), + ], + expected: "session-cleared", + }, + { + name: "a run_started after a clear wins (later run restarted)", + entries: () => [ + createNotification(`_${POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED}`, { + sessionId: "session-cleared", + }), + runStarted("session-restarted"), + ], + expected: "session-restarted", + }, ])("$name", async ({ entries, expected }) => { (mockApiClient.getTaskRun as ReturnType).mockResolvedValue( createTaskRun(), diff --git a/packages/agent/src/sagas/resume-saga.ts b/packages/agent/src/sagas/resume-saga.ts index d18309675f..799bbdf96c 100644 --- a/packages/agent/src/sagas/resume-saga.ts +++ b/packages/agent/src/sagas/resume-saga.ts @@ -1,6 +1,10 @@ import type { ContentBlock } from "@agentclientprotocol/sdk"; import { Saga } from "@posthog/shared"; -import { type NativeGoalState, POSTHOG_NOTIFICATIONS } from "../acp-extensions"; +import { + isNotification, + type NativeGoalState, + POSTHOG_NOTIFICATIONS, +} from "../acp-extensions"; import type { PostHogAPIClient } from "../posthog-api"; import type { DeviceInfo, @@ -163,10 +167,15 @@ export class ResumeSaga extends Saga { } private findSessionId(entries: StoredNotification[]): string | null { - const runStarted = POSTHOG_NOTIFICATIONS.RUN_STARTED; + // RUN_STARTED carries the session id the run booted with; a later + // CONVERSATION_CLEARED (/clear) supersedes it with the fresh SDK session + // id it swapped in. Latest entry of either kind wins. for (let i = entries.length - 1; i >= 0; i--) { const method = entries[i].notification?.method; - if (method === runStarted || method === `_${runStarted}`) { + if ( + isNotification(method, POSTHOG_NOTIFICATIONS.RUN_STARTED) || + isNotification(method, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED) + ) { const params = entries[i].notification?.params as | { sessionId?: string } | undefined; @@ -219,7 +228,7 @@ export class ResumeSaga extends Saga { private rebuildConversation( entries: StoredNotification[], ): ConversationTurn[] { - const turns: ConversationTurn[] = []; + let turns: ConversationTurn[] = []; let currentAssistantContent: ContentBlock[] = []; let currentToolCalls: ToolCallInfo[] = []; @@ -227,6 +236,15 @@ export class ResumeSaga extends Saga { const method = entry.notification?.method; const params = entry.notification?.params as Record; + // /clear starts an empty conversation: everything before the marker is + // gone from the model's context and must not be rehydrated. + if (isNotification(method, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED)) { + turns = []; + currentAssistantContent = []; + currentToolCalls = []; + continue; + } + if (method === "session/update" && params?.update) { const update = params.update as Record; const sessionUpdate = update.sessionUpdate as string; diff --git a/packages/core/src/sessions/acpNotifications.ts b/packages/core/src/sessions/acpNotifications.ts index 242e69ad14..80642db79b 100644 --- a/packages/core/src/sessions/acpNotifications.ts +++ b/packages/core/src/sessions/acpNotifications.ts @@ -17,6 +17,7 @@ export const POSTHOG_NOTIFICATIONS = { PROGRESS: "_posthog/progress", TASK_NOTIFICATION: "_posthog/task_notification", COMPACT_BOUNDARY: "_posthog/compact_boundary", + CONVERSATION_CLEARED: "_posthog/conversation_cleared", USAGE_UPDATE: "_posthog/usage_update", PERMISSION_RESPONSE: "_posthog/permission_response", PERMISSION_REQUEST: "_posthog/permission_request", diff --git a/packages/ui/src/features/sessions/components/ConversationView.tsx b/packages/ui/src/features/sessions/components/ConversationView.tsx index 5a049e930a..8c734caaa6 100644 --- a/packages/ui/src/features/sessions/components/ConversationView.tsx +++ b/packages/ui/src/features/sessions/components/ConversationView.tsx @@ -156,6 +156,7 @@ export function ConversationView({ items: conversationItems, lastTurnInfo, isCompacting, + isClearing, completedToolCallCount, } = useConversationItems(events, isPromptPending, { showDebugLogs, @@ -466,6 +467,7 @@ export function ConversationView({ hasPendingPermission={pendingPermissionsCount > 0} pausedDurationMs={pausedDurationMs} isCompacting={isCompacting} + isClearing={isClearing} usage={contextUsage} completedToolCallCount={completedToolCallCount} /> diff --git a/packages/ui/src/features/sessions/components/SessionFooter.tsx b/packages/ui/src/features/sessions/components/SessionFooter.tsx index 82548518d7..12979be4d3 100644 --- a/packages/ui/src/features/sessions/components/SessionFooter.tsx +++ b/packages/ui/src/features/sessions/components/SessionFooter.tsx @@ -21,6 +21,9 @@ interface SessionFooterProps { hasPendingPermission?: boolean; pausedDurationMs?: number; isCompacting?: boolean; + /** A /clear is in flight; its dedicated "Clearing…" row replaces the + * generic generating indicator, same as compaction. */ + isClearing?: boolean; usage?: ContextUsage | null; /** Number of tool calls finished so far; the generating indicator advances * its status word each time this changes. */ @@ -37,6 +40,7 @@ export function SessionFooter({ hasPendingPermission = false, pausedDurationMs, isCompacting = false, + isClearing = false, usage, completedToolCallCount, }: SessionFooterProps) { @@ -49,7 +53,7 @@ export function SessionFooter({ ); - if (isPromptPending && !isCompacting) { + if (isPromptPending && !isCompacting && !isClearing) { if (hasPendingPermission) { return ( diff --git a/packages/ui/src/features/sessions/components/buildConversationItems.test.ts b/packages/ui/src/features/sessions/components/buildConversationItems.test.ts index 50dba8b580..e40060d09b 100644 --- a/packages/ui/src/features/sessions/components/buildConversationItems.test.ts +++ b/packages/ui/src/features/sessions/components/buildConversationItems.test.ts @@ -272,6 +272,104 @@ describe("buildConversationItems", () => { expect(result.isCompacting).toBe(false); }); + it("clears the clearing spinner on a successful completion status, without duplicating the row", () => { + // A successful /clear sends a terminal `status: clearing, isComplete: + // true`. It must flip the existing status row, not append a second one. + const result = buildConversationItems( + [ + userPromptMsg(1, 1, "/clear"), + statusMsg(2, "clearing"), + statusMsg(3, "clearing", true), + ], + null, + ); + + const statusItems = result.items.filter( + (i): i is Extract => + i.type === "session_update" && i.update.sessionUpdate === "status", + ); + expect(statusItems).toHaveLength(1); + expect((statusItems[0].update as { isComplete?: boolean }).isComplete).toBe( + true, + ); + // The completion resets the flag that suppressed the generic + // "Generating…" footer while the clear ran. + expect(result.isClearing).toBe(false); + expect(result.isCompacting).toBe(false); + }); + + it("flags isClearing while a clear is in flight so the generic generating footer is suppressed", () => { + // The /clear prompt RPC keeps isPromptPending true for the entire swap, + // so the footer needs this flag to avoid rendering "Generating…" next to + // the dedicated "Clearing…" row (mirrors isCompacting). + const result = buildConversationItems( + [userPromptMsg(1, 1, "/clear"), statusMsg(2, "clearing")], + true, + ); + expect(result.isClearing).toBe(true); + expect(result.isCompacting).toBe(false); + }); + + it("renders a timed-out clear as a clearing_failed status row and clears the spinner", () => { + // A timed-out clear emits no conversation_cleared marker, so the adapter + // sends a structured `clearing_failed` status: it clears the spinner (the + // original clearing row goes complete) and adds the outcome row. + const result = buildConversationItems( + [ + userPromptMsg(1, 1, "/clear"), + statusMsg(2, "clearing"), + statusMsg(3, "clearing_failed", undefined, "Timed out after 30000ms."), + ], + null, + ); + + const statusItems = result.items.filter( + (i): i is Extract => + i.type === "session_update" && i.update.sessionUpdate === "status", + ); + // Spinner row (now complete) + the failure row. + expect(statusItems.map((i) => i.update)).toEqual([ + { + sessionUpdate: "status", + status: "clearing", + isComplete: true, + startedAt: 2, + }, + { + sessionUpdate: "status", + status: "clearing_failed", + error: "Timed out after 30000ms.", + }, + ]); + // The failure also releases the generating-footer suppression. + expect(result.isClearing).toBe(false); + }); + + it("renders a conversation_cleared divider after a /clear", () => { + const result = buildConversationItems( + [ + userPromptMsg(1, 1, "/clear"), + { + type: "acp_message", + ts: 2, + message: { + jsonrpc: "2.0", + method: "_posthog/conversation_cleared", + params: { sessionId: "sdk-new" }, + }, + }, + ], + null, + ); + + const clearedItems = result.items.filter( + (i): i is Extract => + i.type === "session_update" && + i.update.sessionUpdate === "conversation_cleared", + ); + expect(clearedItems).toHaveLength(1); + }); + it("renders a terminal refusal as a status row carrying the explanation", () => { const result = buildConversationItems( [ diff --git a/packages/ui/src/features/sessions/components/buildConversationItems.ts b/packages/ui/src/features/sessions/components/buildConversationItems.ts index 17abfe38e2..2986a20a3a 100644 --- a/packages/ui/src/features/sessions/components/buildConversationItems.ts +++ b/packages/ui/src/features/sessions/components/buildConversationItems.ts @@ -78,6 +78,9 @@ export interface BuildResult { items: ConversationItem[]; lastTurnInfo: LastTurnInfo | null; isCompacting: boolean; + /** A `/clear` is in flight (its status row shows the dedicated spinner), so + * the generic "Generating…" footer must stay hidden — same as compaction. */ + isClearing: boolean; /** Number of tool calls settled into a terminal status so far. Monotonic * within a thread; consumers treat a change as "a tool/MCP call finished". */ completedToolCallCount: number; @@ -121,6 +124,7 @@ export interface ItemBuilder { pendingPrompts: Map; shellExecutes: Map; isCompacting: boolean; + isClearing: boolean; nextId: () => number; /** Progress cards keyed by the backend-supplied `group` id. The first event * for a group opens the card inline where it arrived; every subsequent @@ -151,6 +155,7 @@ export function createItemBuilder(): ItemBuilder { pendingPrompts: new Map(), shellExecutes: new Map(), isCompacting: false, + isClearing: false, nextId: () => idCounter++, progressCards: new Map(), lowestTouchedProgressIndex: Number.POSITIVE_INFINITY, @@ -252,6 +257,7 @@ export function buildConversationItems( items: b.items, lastTurnInfo, isCompacting: b.isCompacting, + isClearing: b.isClearing, completedToolCallCount: b.completedToolCallCount, }; } @@ -308,6 +314,7 @@ export function buildAgentConversationItems( items: b.items, lastTurnInfo: readLastTurnInfo(b), isCompacting: b.isCompacting, + isClearing: b.isClearing, completedToolCallCount: b.completedToolCallCount, }; } @@ -695,6 +702,12 @@ function handleNotification( return; } + if (isNotification(msg.method, POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED)) { + ensureImplicitTurn(b, ts); + pushItem(b, { sessionUpdate: "conversation_cleared" }); + return; + } + if (isNotification(msg.method, POSTHOG_NOTIFICATIONS.STATUS)) { ensureImplicitTurn(b, ts); const params = msg.params as { @@ -756,6 +769,26 @@ function handleRuntimeStatus( } else if (status.status === "retrying" && status.isComplete) { markRuntimeStatusComplete(b, "retrying"); return; + } else if (status.status === "clearing") { + if (status.isComplete) { + markRuntimeStatusComplete(b, "clearing"); + return; + } + // The /clear prompt RPC keeps isPromptPending true for the whole swap, + // so without this flag the generic "Generating…" footer would render + // alongside the dedicated "Clearing…" row (compaction has the same + // gate via isCompacting). + b.isClearing = true; + } else if (status.status === "clearing_failed") { + // A timed-out clear emits no `conversation_cleared` marker, so clear + // the spinner and render the outcome as its own status row. + markRuntimeStatusComplete(b, "clearing"); + pushItem(b, { + sessionUpdate: "status", + status: "clearing_failed", + error: status.error, + }); + return; } pushItem(b, { @@ -852,6 +885,9 @@ function markRuntimeStatusComplete(b: ItemBuilder, status: string) { if (status === "compacting") { b.isCompacting = false; } + if (status === "clearing") { + b.isClearing = false; + } for (let i = b.items.length - 1; i >= 0; i--) { const item = b.items[i]; if ( diff --git a/packages/ui/src/features/sessions/components/chat-thread/ChatThreadFooter.tsx b/packages/ui/src/features/sessions/components/chat-thread/ChatThreadFooter.tsx index cabb5777e3..2ddc08091b 100644 --- a/packages/ui/src/features/sessions/components/chat-thread/ChatThreadFooter.tsx +++ b/packages/ui/src/features/sessions/components/chat-thread/ChatThreadFooter.tsx @@ -37,7 +37,7 @@ export function ChatThreadFooter({ }: ChatThreadFooterProps) { const showDebugLogs = useSettingsStore((s) => s.debugLogsCloudRuns); const contextUsage = useContextUsage(events); - const { lastTurnInfo, isCompacting, completedToolCallCount } = + const { lastTurnInfo, isCompacting, isClearing, completedToolCallCount } = useConversationItems(events, isPromptPending, { showDebugLogs }); const pendingPermissions = usePendingPermissionsForTask(taskId ?? ""); const queuedCount = useQueuedMessagesForTask(taskId).length; @@ -60,6 +60,7 @@ export function ChatThreadFooter({ hasPendingPermission={pendingPermissions.size > 0} pausedDurationMs={pausedDurationMs} isCompacting={isCompacting} + isClearing={isClearing} usage={contextUsage} completedToolCallCount={completedToolCallCount} /> diff --git a/packages/ui/src/features/sessions/components/incrementalConversationItems.ts b/packages/ui/src/features/sessions/components/incrementalConversationItems.ts index b4bb784120..8d4631fed3 100644 --- a/packages/ui/src/features/sessions/components/incrementalConversationItems.ts +++ b/packages/ui/src/features/sessions/components/incrementalConversationItems.ts @@ -82,6 +82,7 @@ export function createIncrementalConversationBuilder() { items: builder.items, lastTurnInfo: readLastTurnInfo(builder), isCompacting: builder.isCompacting, + isClearing: builder.isClearing, completedToolCallCount: builder.completedToolCallCount, }; // A finalized builder can't be safely continued; the next streaming @@ -147,6 +148,7 @@ export function createIncrementalConversationBuilder() { items: assembleItems(builder, activeStart), lastTurnInfo: readLastTurnInfoForOutput(builder), isCompacting: builder.isCompacting, + isClearing: builder.isClearing, completedToolCallCount: builder.completedToolCallCount, }; } diff --git a/packages/ui/src/features/sessions/components/session-update/ConversationClearedView.tsx b/packages/ui/src/features/sessions/components/session-update/ConversationClearedView.tsx new file mode 100644 index 0000000000..569660f447 --- /dev/null +++ b/packages/ui/src/features/sessions/components/session-update/ConversationClearedView.tsx @@ -0,0 +1,31 @@ +import { Eraser } from "@phosphor-icons/react"; +import { ChatMarker, ChatMarkerContent } from "@posthog/quill"; +import { Box, Flex, Text } from "@radix-ui/themes"; +import { useChatThreadChrome } from "../chat-thread/chatThreadChrome"; + +// New thread renders the boundary as a centered separator marker; the legacy +// thread keeps a bordered row so ConversationView is unchanged when the chat +// thread is off (mirrors CompactBoundaryView). +export function ConversationClearedView() { + const chatChrome = useChatThreadChrome(); + + if (chatChrome) { + return ( + + Conversation cleared + + ); + } + + return ( + + + + Conversation cleared + + (earlier messages are no longer in the agent's context) + + + + ); +} diff --git a/packages/ui/src/features/sessions/components/session-update/SessionUpdateView.tsx b/packages/ui/src/features/sessions/components/session-update/SessionUpdateView.tsx index dc49fd9abe..0249b6f30d 100644 --- a/packages/ui/src/features/sessions/components/session-update/SessionUpdateView.tsx +++ b/packages/ui/src/features/sessions/components/session-update/SessionUpdateView.tsx @@ -1,6 +1,7 @@ import { AgentMessage } from "@posthog/ui/features/sessions/components/session-update/AgentMessage"; import { CompactBoundaryView } from "@posthog/ui/features/sessions/components/session-update/CompactBoundaryView"; import { ConsoleMessage } from "@posthog/ui/features/sessions/components/session-update/ConsoleMessage"; +import { ConversationClearedView } from "@posthog/ui/features/sessions/components/session-update/ConversationClearedView"; import { ErrorNotificationView } from "@posthog/ui/features/sessions/components/session-update/ErrorNotificationView"; import { ProgressGroupView } from "@posthog/ui/features/sessions/components/session-update/ProgressGroupView"; import { StatusNotificationView } from "@posthog/ui/features/sessions/components/session-update/StatusNotificationView"; @@ -29,6 +30,9 @@ export type RenderItem = preTokens: number; contextSize?: number; } + | { + sessionUpdate: "conversation_cleared"; + } | { sessionUpdate: "status"; status: string; @@ -131,6 +135,8 @@ export const SessionUpdateView = memo(function SessionUpdateView({ contextSize={item.contextSize} /> ); + case "conversation_cleared": + return ; case "status": return ( + {message} + + ); + } + return ( + + + + {message} + + + ); + } + + if (status === "clearing") { + if (isComplete) { + return null; + } + return ( + + ); + } + // Generic status display for other statuses return ( @@ -204,20 +239,27 @@ function RetryingStatusView({ } /** - * In-flight compaction row. Compaction is a single streaming summarization call - * with no measurable percentage, so we pair the spinner with an indeterminate - * progress bar (constant motion, so it never reads as frozen) and a live - * elapsed-time counter, which is the one honest progress signal we have. + * In-flight row for a single streaming operation with no measurable + * percentage (compaction, `/clear`), so we pair the spinner with an + * indeterminate progress bar (constant motion, so it never reads as frozen) + * and a live elapsed-time counter, which is the one honest progress signal + * we have. */ -function CompactingStatusView({ startedAt }: { startedAt?: number }) { +function CompactingStatusView({ + startedAt, + label = "Compacting conversation history...", +}: { + startedAt?: number; + label?: string; +}) { const [elapsed, setElapsed] = useState(() => startedAt ? Date.now() - startedAt : 0, ); useEffect(() => { - // Anchor to the persisted compaction start time so remounting this row - // (e.g. scrolling it out of and back into the virtualized list while - // compaction runs) keeps counting from when compaction began rather than + // Anchor to the persisted start time so remounting this row (e.g. + // scrolling it out of and back into the virtualized list while the + // operation runs) keeps counting from when it began rather than // resetting to zero. Fall back to mount time only if it's missing. const start = startedAt ?? Date.now(); const tick = () => setElapsed(Date.now() - start); @@ -230,9 +272,7 @@ function CompactingStatusView({ startedAt }: { startedAt?: number }) { - - Compacting conversation history... - + {label} {formatDuration(elapsed, 1)}