From 2c7e954ed85509da2824f8effd272b30c51de728 Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Mon, 3 Aug 2026 13:24:12 -0700 Subject: [PATCH 1/4] feat(tasks): clear a finished cloud run's conversation without a sandbox MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A live run's /clear goes to its agent. A finished one has no agent, and resuming into a whole new run just to clear a conversation that run would rebuild from the log anyway is wasteful, so the boundary is recorded against the finished run instead: POST runs/{id}/clear_conversation appends the typed message plus a _posthog/conversation_cleared marker to its log. The logs endpoint serves a resume chain concatenated, so both rehydration paths already stop at the marker and the next run continues with an empty conversation, its checkpoints and visible history intact. Only for a finished run — an active one has an agent that owns the clear, and a live writer on the same log object this read-modify-write append would race, so it returns 409. Both clients gate on the agent's conversationClear capability and fall back to an ordinary new run when it is absent. An agent that predates the marker ignores it on resume, so without the gate a user would see the boundary rendered and believe a clear happened that never did. The cloud thread renders the boundary and a "Clearing…" spinner, and /clear is sent unwrapped: a context block would hide the command behind it, since the agent reads the command off the front of the message. Claude-Session: https://claude.ai/code/session_01HJQHhq27qXnKGj98x7UrXZ --- .../packages/api-client/src/posthog-client.ts | 18 +++ .../core/src/sessions/sessionEvents.ts | 41 ++++++ .../core/src/sessions/sessionService.ts | 31 +++++ .../sessions/sessionServiceCloudClear.test.ts | 121 ++++++++++++++++++ .../frontend/components/ThreadItems.tsx | 53 ++++++-- .../frontend/components/ThreadRow.tsx | 5 +- .../frontend/components/ThreadView.tsx | 1 + .../logics/runInteractionLogic.test.ts | 75 ++++++++++- .../frontend/logics/runInteractionLogic.ts | 93 ++++++++++++-- .../frontend/logics/runStreamLogic.test.ts | 33 +++++ .../frontend/logics/runStreamLogic.ts | 65 +++++++++- .../posthog_ai/frontend/types/streamTypes.ts | 3 +- .../frontend/types/wireTypes.test.ts | 5 + .../posthog_ai/frontend/types/wireTypes.ts | 10 ++ products/tasks/backend/facade/api.py | 18 +++ products/tasks/backend/models.py | 48 +++++++ .../tasks/backend/presentation/views/api.py | 37 ++++++ products/tasks/backend/tests/test_api.py | 31 +++++ products/tasks/frontend/generated/api.ts | 20 +++ 19 files changed, 677 insertions(+), 31 deletions(-) create mode 100644 products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts diff --git a/products/desktop/packages/api-client/src/posthog-client.ts b/products/desktop/packages/api-client/src/posthog-client.ts index 2c5df489f998..a01fe53a6ec8 100644 --- a/products/desktop/packages/api-client/src/posthog-client.ts +++ b/products/desktop/packages/api-client/src/posthog-client.ts @@ -3431,6 +3431,24 @@ export class PostHogAPIClient { } } + /** + * Record a `/clear` boundary in a finished run's log, so the next run in the + * chain resumes past it with an empty conversation. Only valid for a finished + * run — an active one has an agent that owns the clear (409 otherwise). + */ + async clearTaskRunConversation(taskId: string, runId: string): Promise { + const teamId = await this.getTeamId(); + const path = `/api/projects/${teamId}/tasks/${taskId}/runs/${runId}/clear_conversation/`; + const response = await this.api.fetcher.fetch({ + method: "post", + url: new URL(`${this.api.baseUrl}${path}`), + path, + }); + if (!response.ok) { + throw new Error(`Failed to clear conversation: ${response.statusText}`); + } + } + async getTaskRunSessionLogs( taskId: string, runId: string, diff --git a/products/desktop/packages/core/src/sessions/sessionEvents.ts b/products/desktop/packages/core/src/sessions/sessionEvents.ts index 8c5e906fe4a9..7e1058a12d7f 100644 --- a/products/desktop/packages/core/src/sessions/sessionEvents.ts +++ b/products/desktop/packages/core/src/sessions/sessionEvents.ts @@ -135,6 +135,47 @@ export function createUserMessageEvent(text: string, ts: number): AcpMessage { return createUserPromptEvent([{ type: "text", text }], ts); } +/** + * The two frames a `/clear` on a finished cloud run produces, in log order: the + * message the user typed, then the boundary rehydration stops at. + * + * The backend writes these into the run log (there is no sandbox to emit them), + * and mirrors this shape — a finished run has no live stream to echo them back, + * so the client paints them from here and a later reload folds the persisted + * copies into the same thread. + */ +export function createConversationClearedEvents( + sessionId: string, + ts: number, +): AcpMessage[] { + return [ + { + type: "acp_message", + ts, + message: { + jsonrpc: "2.0", + method: "session/update", + params: { + sessionId, + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "/clear" }, + }, + }, + }, + }, + { + type: "acp_message", + ts, + message: { + jsonrpc: "2.0", + method: POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED, + params: {}, + }, + }, + ]; +} + /** * Create a user shell execute event. * When id is provided, it's used to track async execution (start/complete). diff --git a/products/desktop/packages/core/src/sessions/sessionService.ts b/products/desktop/packages/core/src/sessions/sessionService.ts index 34818d7e8881..5996a869da97 100644 --- a/products/desktop/packages/core/src/sessions/sessionService.ts +++ b/products/desktop/packages/core/src/sessions/sessionService.ts @@ -91,6 +91,7 @@ import { } from "./permissionResponse"; import { convertStoredEntriesToEvents, + createConversationClearedEvents, createUserShellExecuteEvent, extractPromptText, getStoredLogEventPosition, @@ -144,6 +145,11 @@ const SESSION_EVENT_EVICT_GRACE_MS = 20_000; */ const OPEN_TAIL_BYTES = 1_500_000; +/** Matches a bare `/clear` invocation, not a longer command that starts with it. */ +function isClearCommand(text: string | undefined): boolean { + return /^\/clear(?:\s|$)/.test(text ?? ""); +} + class GitHubAuthorizationRequiredForCloudHandoffError extends Error { constructor( message = "Connect GitHub before continuing this task in cloud.", @@ -4119,6 +4125,15 @@ export class SessionService { } if (isTerminalStatus(session.cloudStatus)) { + // `/clear` is handled by the agent, not the model, so resuming would spin a + // whole sandbox to clear a conversation the next run rebuilds from the log + // anyway. The backend records the boundary against this run instead — but only + // when the agent understands it. An older one ignores the marker and resumes the + // conversation it was meant to retire, so an ordinary resume is the honest + // degradation: the clear doesn't happen, and nothing claims it did. + if (isClearCommand(transport.messageText) && session.conversationClear) { + return this.clearCloudConversation(session); + } // If the agent never booted (no `run_started`), resuming spins another // sandbox that hits the same provisioning failure — surface the error // instead of looping. @@ -4424,6 +4439,22 @@ export class SessionService { } } + /** Records the `/clear` boundary against a finished run and paints it locally. */ + private async clearCloudConversation( + session: AgentSession, + ): Promise<{ stopReason: string }> { + const client = await this.d.getAuthenticatedClient(); + if (!client) { + throw new Error("Authentication required for cloud commands"); + } + await client.clearTaskRunConversation(session.taskId, session.taskRunId); + this.d.store.appendEvents( + session.taskRunId, + createConversationClearedEvents(session.taskRunId, Date.now()), + ); + return { stopReason: "end_turn" }; + } + private async resumeCloudRun( session: AgentSession, prompt: string | ContentBlock[], diff --git a/products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts b/products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts new file mode 100644 index 000000000000..09141bb3fe7b --- /dev/null +++ b/products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts @@ -0,0 +1,121 @@ +import type { AgentSession } from "@posthog/shared"; +import { describe, expect, it, vi } from "vitest"; +import { POSTHOG_NOTIFICATIONS } from "./acpNotifications"; +import { SessionService, type SessionServiceDeps } from "./sessionService"; + +const TASK_ID = "task-1"; +const TASK_RUN_ID = `run-${TASK_ID}`; + +function createHarness({ + conversationClear = true, +}: { + conversationClear?: boolean; +} = {}) { + const sessions: Record = { + [TASK_RUN_ID]: { + taskRunId: TASK_RUN_ID, + taskId: TASK_ID, + taskTitle: "Test task", + channel: "", + events: [], + startedAt: 1, + status: "connected", + isCloud: true, + cloudStatus: "completed", + conversationClear, + isPromptPending: false, + isCompacting: false, + promptStartedAt: null, + pendingPermissions: new Map(), + pausedDurationMs: 0, + messageQueue: [], + optimisticItems: [], + } as unknown as AgentSession, + }; + + const appendEvents = vi.fn(); + const clearTaskRunConversation = vi.fn().mockResolvedValue(undefined); + const runTaskInCloud = vi.fn(); + + const deps = { + store: { + getSessions: () => sessions, + getSessionByTaskId: (taskId: string) => + Object.values(sessions).find((s) => s.taskId === taskId), + appendEvents, + updateSession: vi.fn(), + appendOptimisticItem: vi.fn(), + clearTailOptimisticItems: vi.fn(), + }, + h: { + getCloudPromptTransport: (prompt: string) => ({ + promptText: prompt, + messageText: prompt, + filePaths: [], + skillBundles: [], + }), + }, + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + toast: { error: vi.fn(), info: vi.fn() }, + track: vi.fn(), + getIsOnline: () => true, + addDirectoryDialog: { open: false }, + getAuthenticatedClient: async () => ({ + clearTaskRunConversation, + runTaskInCloud, + }), + trpc: { + agent: { + onSessionIdleKilled: { subscribe: () => ({ unsubscribe: vi.fn() }) }, + }, + }, + } as unknown as SessionServiceDeps; + + return { + service: new SessionService(deps), + appendEvents, + clearTaskRunConversation, + runTaskInCloud, + }; +} + +describe("SessionService /clear on a finished cloud run", () => { + it("records the boundary and renders it without resuming into a new run", async () => { + const { service, appendEvents, clearTaskRunConversation, runTaskInCloud } = + createHarness(); + + const result = await service.sendPrompt(TASK_ID, "/clear"); + + expect(result).toEqual({ stopReason: "end_turn" }); + expect(clearTaskRunConversation).toHaveBeenCalledWith(TASK_ID, TASK_RUN_ID); + // Resuming would spin a whole sandbox to clear a conversation the next run + // rebuilds from the log anyway. + expect(runTaskInCloud).not.toHaveBeenCalled(); + + // A finished run streams nothing back, so the thread is painted from here. + const [, events] = appendEvents.mock.calls[0]; + expect( + events.map((e: { message: { method: string } }) => e.message.method), + ).toEqual(["session/update", POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED]); + }); + + it("resumes into a new run when the agent cannot honour the boundary", async () => { + // An older agent ignores the marker and resumes the conversation it was meant to + // retire, so recording one would claim a clear that never happens. + const { service, clearTaskRunConversation } = createHarness({ + conversationClear: false, + }); + + await service.sendPrompt(TASK_ID, "/clear").catch(() => undefined); + + expect(clearTaskRunConversation).not.toHaveBeenCalled(); + }); + + it("still resumes into a new run for an ordinary message", async () => { + const { service, clearTaskRunConversation } = createHarness(); + + await service.sendPrompt(TASK_ID, "keep going").catch(() => undefined); + + expect(clearTaskRunConversation).not.toHaveBeenCalled(); + }); +}); diff --git a/products/posthog_ai/frontend/components/ThreadItems.tsx b/products/posthog_ai/frontend/components/ThreadItems.tsx index 6db7dd6f047c..9b4b93d2f6f6 100644 --- a/products/posthog_ai/frontend/components/ThreadItems.tsx +++ b/products/posthog_ai/frontend/components/ThreadItems.tsx @@ -1,4 +1,4 @@ -import { IconCheck, IconCollapse, IconWarning, IconX } from '@posthog/icons' +import { IconCheck, IconCircleDashed, IconCollapse, IconWarning, IconX } from '@posthog/icons' import { Spinner } from '@posthog/lemon-ui' import { humanFriendlyNumber } from 'lib/utils/numbers' @@ -7,23 +7,52 @@ import type { ThreadItem } from '../types/streamTypes' import { Activity } from './ActivityPrimitives' import type { ActivityStatus } from './ActivityPrimitives' -/** Inline `_posthog/status` item — a spinner while compacting, a generic status line otherwise. */ -export function StatusItem({ item }: { item: ThreadItem }): JSX.Element { - const isCompacting = item.status === 'compacting' && !item.isComplete +/** Statuses that run for a while and get a spinner with their own label, keyed by wire status. */ +const IN_PROGRESS_STATUS_LABELS: Record = { + compacting: 'Compacting conversation history…', + clearing: 'Clearing conversation…', +} + +function StatusLine({ icon, children }: { icon?: JSX.Element; children: React.ReactNode }): JSX.Element { return (
- {isCompacting ? ( - <> - - Compacting conversation history… - - ) : ( - Status: {item.status} - )} + {icon} + {children}
) } +/** Inline `_posthog/status` item — a spinner while an operation runs, a status line otherwise. */ +export function StatusItem({ item }: { item: ThreadItem }): JSX.Element { + const inProgressLabel = item.isComplete ? undefined : IN_PROGRESS_STATUS_LABELS[item.status ?? ''] + if (inProgressLabel) { + return }>{inProgressLabel} + } + // A failed clear leaves the agent session closed, so the way forward is a new run, not a retry. + if (item.status === 'clearing_failed') { + const reason = item.errorMessage + ? `Couldn't clear the conversation: ${item.errorMessage}` + : "Couldn't clear the conversation" + return }>{reason}. Start a new run to keep going. + } + return Status: {item.status} +} + +/** Inline `_posthog/conversation_cleared` item — the `/clear` boundary card. */ +export function ConversationClearedItem({ item }: { item: ThreadItem }): JSX.Element { + return ( + } + animate={false} + showCompletionIcon={false} + /> + ) +} + /** Inline `_posthog/compact_boundary` item — the post-compaction card. */ export function CompactBoundaryItem({ item }: { item: ThreadItem }): JSX.Element { const parts = [ diff --git a/products/posthog_ai/frontend/components/ThreadRow.tsx b/products/posthog_ai/frontend/components/ThreadRow.tsx index e9b58e899f2f..8f99e0a9b29f 100644 --- a/products/posthog_ai/frontend/components/ThreadRow.tsx +++ b/products/posthog_ai/frontend/components/ThreadRow.tsx @@ -15,7 +15,7 @@ import type { ProgressStep, ThreadItem } from '../types/streamTypes' import { resolveToolCall } from '../utils/toolResolver' import { RunActivity } from './RunActivity' import { RunAlertActivity } from './RunAlertActivity' -import { CompactBoundaryItem, StatusItem, TaskNotificationItem } from './ThreadItems' +import { CompactBoundaryItem, ConversationClearedItem, StatusItem, TaskNotificationItem } from './ThreadItems' import { ToolCallCard } from './tool/ToolCallCard' type ToolInvocations = typeof runStreamLogic.values.toolInvocations @@ -161,6 +161,9 @@ export const ThreadRow = memo(function ThreadRow({ if (item.type === 'compact_boundary') { return } + if (item.type === 'conversation_cleared') { + return + } if (item.type === 'task_notification') { return } diff --git a/products/posthog_ai/frontend/components/ThreadView.tsx b/products/posthog_ai/frontend/components/ThreadView.tsx index 011053a32656..98eb26a90faa 100644 --- a/products/posthog_ai/frontend/components/ThreadView.tsx +++ b/products/posthog_ai/frontend/components/ThreadView.tsx @@ -34,6 +34,7 @@ const THREAD_ITEM_HEIGHT_ESTIMATES: Partial> error: 42, status: 42, compact_boundary: 42, + conversation_cleared: 42, task_notification: 26, progress: 42, debug: 30, diff --git a/products/posthog_ai/frontend/logics/runInteractionLogic.test.ts b/products/posthog_ai/frontend/logics/runInteractionLogic.test.ts index 52b8fdbf332c..eb28ef28256a 100644 --- a/products/posthog_ai/frontend/logics/runInteractionLogic.test.ts +++ b/products/posthog_ai/frontend/logics/runInteractionLogic.test.ts @@ -6,7 +6,11 @@ import { aiConsentLogic } from 'scenes/settings/organization/aiConsentLogic' import { initKeaTests } from '~/test/init' -import { tasksRunCreate, tasksRunsCommandCreate } from 'products/tasks/frontend/generated/api' +import { + tasksRunCreate, + tasksRunsClearConversationCreate, + tasksRunsCommandCreate, +} from 'products/tasks/frontend/generated/api' import { contextItemLine } from '../utils/posthogContextBlock' import { attachedContextLogic } from './attachedContextLogic' @@ -25,12 +29,14 @@ jest.mock('./runStreamLogic', () => { key((p: { streamKey: string }) => p.streamKey), actions({ pushHumanMessage: (content: string) => ({ content }), + pushConversationCleared: true, respondToPermission: (payload: unknown) => ({ payload }), cancelRun: (run?: unknown) => ({ run }), markTurnComplete: true, setCurrentMode: (mode: string) => ({ mode }), setStubStatus: (status: string | null) => ({ status }), setStubThinking: (thinking: boolean) => ({ thinking }), + setStubClearSupported: (supported: boolean) => ({ supported }), }), reducers({ currentRunStatus: [ @@ -46,6 +52,12 @@ jest.mock('./runStreamLogic', () => { }, ], pendingPermissionRequest: [null, {}], + conversationClearSupported: [ + true, + { + setStubClearSupported: (_: boolean, { supported }: { supported: boolean }) => supported, + }, + ], respondingToPermission: [false, {}], currentMode: [ null, @@ -82,6 +94,7 @@ jest.mock('scenes/projectLogic', () => { jest.mock('products/tasks/frontend/generated/api', () => ({ tasksRunsCommandCreate: jest.fn(), tasksRunCreate: jest.fn(), + tasksRunsClearConversationCreate: jest.fn(), })) jest.mock('lib/lemon-ui/LemonToast', () => ({ @@ -122,6 +135,7 @@ describe('runInteractionLogic', () => { jest.clearAllMocks() ;(tasksRunsCommandCreate as jest.Mock).mockResolvedValue({}) ;(tasksRunCreate as jest.Mock).mockResolvedValue({ latest_run: { id: 'run-2' } }) + ;(tasksRunsClearConversationCreate as jest.Mock).mockResolvedValue({}) initKeaTests() project = projectLogic() project.mount() @@ -365,6 +379,40 @@ describe('runInteractionLogic', () => { expect(logic.values.composerForm.draft).toBe('') }) + it('records the boundary instead of starting a run when /clear is sent to a terminal run', async () => { + setStatus('completed') + logic.actions.setComposerFormValues({ draft: '/clear' }) + + await expectLogic(logic, () => { + logic.actions.submitComposerForm() + }).toFinishAllListeners() + + // Booting a sandbox would clear a conversation the next run rebuilds from the log anyway. + expect(tasksRunCreate).not.toHaveBeenCalled() + expect(tasksRunsClearConversationCreate).toHaveBeenCalledWith('997', TASK_ID, RUN_ID) + expect(logic.values.composerForm.draft).toBe('') + // Nothing streams back on a finished run, so the boundary is echoed from here. + await expectLogic(stream).toDispatchActions([ + (action) => action.type === stream.actionTypes.pushHumanMessage && action.payload.content === '/clear', + (action) => action.type === stream.actionTypes.pushConversationCleared, + ]) + }) + + it('falls back to a new run when the chain agent cannot honour the clear boundary', async () => { + // An older agent ignores the marker and resumes the conversation it was meant to retire, + // so a divider here would claim a clear that never happens. + ;(stream.actions as unknown as { setStubClearSupported: (s: boolean) => void }).setStubClearSupported(false) + setStatus('completed') + logic.actions.setComposerFormValues({ draft: '/clear' }) + + await expectLogic(logic, () => { + logic.actions.submitComposerForm() + }).toFinishAllListeners() + + expect(tasksRunsClearConversationCreate).not.toHaveBeenCalled() + expect(tasksRunCreate).toHaveBeenCalled() + }) + it('keeps the draft and toasts when starting a new run fails', async () => { ;(tasksRunCreate as jest.Mock).mockRejectedValue(new Error('boom')) setStatus('completed') @@ -439,6 +487,31 @@ describe('runInteractionLogic', () => { expect(send.params.content).toBe('follow up') }) + it('sends /clear unwrapped so the agent still sees the command at the front, and keeps the context pending', async () => { + const item = { type: 'insight', key: 'sig', label: 'Signups' } + attachedContextLogic().actions.registerContext('scene', [item]) + setThinking(false) + + logic.actions.setComposerFormValues({ draft: '/clear' }) + await expectLogic(logic, () => { + logic.actions.submitComposerForm() + }).toFinishAllListeners() + + const send = (tasksRunsCommandCreate as jest.Mock).mock.calls[0][3] as { params: { content: string } } + expect(send.params.content).toBe('/clear') + + // The agent drops the message rather than reading it, so the ref was never really delivered: + // the next real send must still carry it. + ;(tasksRunsCommandCreate as jest.Mock).mockClear() + logic.actions.setComposerFormValues({ draft: 'why the drop?' }) + await expectLogic(logic, () => { + logic.actions.submitComposerForm() + }).toFinishAllListeners() + + const next = (tasksRunsCommandCreate as jest.Mock).mock.calls[0][3] as { params: { content: string } } + expect(next.params.content).toContain('- insight sig ("Signups")') + }) + it('keeps pruning context sent by a terminal-run send after re-pointing to the fresh run', async () => { attachedContextLogic().actions.registerContext('scene', [{ type: 'insight', key: 'sig', label: 'Signups' }]) diff --git a/products/posthog_ai/frontend/logics/runInteractionLogic.ts b/products/posthog_ai/frontend/logics/runInteractionLogic.ts index 10118bc260e7..2c2d6ff7d01f 100644 --- a/products/posthog_ai/frontend/logics/runInteractionLogic.ts +++ b/products/posthog_ai/frontend/logics/runInteractionLogic.ts @@ -16,7 +16,11 @@ import { getModeOption, type PermissionMode, } from 'products/posthog_ai/frontend/utils/composerModes' -import { tasksRunCreate, tasksRunsCommandCreate } from 'products/tasks/frontend/generated/api' +import { + tasksRunCreate, + tasksRunsClearConversationCreate, + tasksRunsCommandCreate, +} from 'products/tasks/frontend/generated/api' import { ClaudeRuntimeAdapterEnumApi, type ClaudeTaskRunCreateSchemaApi, @@ -68,6 +72,11 @@ const EFFORT_CONFIG_ID = 'effort' // `set_config_option { configId: 'mode' }` is how `/code` applies a live shift+tab mode change. const MODE_CONFIG_ID = 'mode' +/** Matches a bare `/clear` invocation, not a longer command that starts with it. */ +function isClearCommand(content: string): boolean { + return /^\/clear(?:\s|$)/.test(content) +} + // Generated by kea-typegen. Update if you're an agent, ignore if you're human. export interface runInteractionLogicValues { dataProcessingAccepted: boolean // aiConsentLogic @@ -75,12 +84,14 @@ export interface runInteractionLogicValues { seenContextLinesByTask: Record // attachedContextLogic sentContextKeysByTask: Record // attachedContextLogic currentProjectId: number | null // projectLogic + conversationClearSupported: boolean // runStreamLogic currentMode: string | null // runStreamLogic currentRunStatus: RunStatus | null // runStreamLogic isThinking: boolean // runStreamLogic pendingPermissionRequest: PermissionRequestRecord | null // runStreamLogic respondingToPermission: boolean // runStreamLogic canSend: boolean + clearing: boolean composerForm: { draft: string } @@ -151,6 +162,9 @@ export interface runInteractionLogicActions { markTurnComplete: () => { value: true } // runStreamLogic + pushConversationCleared: () => { + value: true + } // runStreamLogic pushHumanMessage: (content: string) => { content: string } // runStreamLogic @@ -187,6 +201,9 @@ export interface runInteractionLogicActions { clearConsentBlock: () => { value: true } + clearConversation: () => { + value: true + } clearQueue: () => { value: true } @@ -214,6 +231,9 @@ export interface runInteractionLogicActions { content: string source: 'draft' | 'queue' } + setClearing: (clearing: boolean) => { + clearing: boolean + } setComposerFormManualErrors: (errors: Record) => { errors: Record } @@ -302,7 +322,7 @@ export interface runInteractionLogicMeta { selectedMode: (modeOverride: PermissionMode | null, currentMode: string | null, arg: any) => PermissionMode isBusy: (isThinking: boolean) => boolean canSend: (sending: boolean, isTerminal: boolean, currentProjectId: number | null) => boolean - isSubmitting: (sending: boolean, startingRun: boolean) => boolean + isSubmitting: (sending: boolean, startingRun: boolean, clearing: boolean) => boolean pendingContextItems: ( contextItems: AttachedContextItem[], sentContextKeysByTask: Record, @@ -342,7 +362,14 @@ export const runInteractionLogic = kea([ projectLogic, ['currentProjectId'], runStreamLogic({ streamKey: props.streamKey ?? props.runId }), - ['currentRunStatus', 'pendingPermissionRequest', 'respondingToPermission', 'isThinking', 'currentMode'], + [ + 'currentRunStatus', + 'pendingPermissionRequest', + 'respondingToPermission', + 'isThinking', + 'currentMode', + 'conversationClearSupported', + ], attachedContextLogic, ['contextItems', 'sentContextKeysByTask', 'seenContextLinesByTask'], aiConsentLogic, @@ -350,7 +377,14 @@ export const runInteractionLogic = kea([ ], actions: [ runStreamLogic({ streamKey: props.streamKey ?? props.runId }), - ['pushHumanMessage', 'respondToPermission', 'cancelRun', 'markTurnComplete', 'setCurrentMode'], + [ + 'pushHumanMessage', + 'pushConversationCleared', + 'respondToPermission', + 'cancelRun', + 'markTurnComplete', + 'setCurrentMode', + ], attachedContextLogic, ['markContextSent'], toolStreamEventsLogic, @@ -365,6 +399,8 @@ export const runInteractionLogic = kea([ // Start a fresh run on the task, seeded with this message and chained from the finished run. startNewRun: (content: string) => ({ content }), setStartingRun: (starting: boolean) => ({ starting }), + clearConversation: true, + setClearing: (clearing: boolean) => ({ clearing }), // Internal: POST one `user_message` now. `source` says where the content lives so a successful send // clears the right place and a failed send preserves it for retry ('draft' → composer, 'queue' → // the staged buffer combined into this send). @@ -414,6 +450,12 @@ export const runInteractionLogic = kea([ setStartingRun: (_, { starting }) => starting, }, ], + clearing: [ + false, + { + setClearing: (_, { clearing }) => clearing, + }, + ], queuedMessages: [ [] as QueuedMessage[], { @@ -487,7 +529,7 @@ export const runInteractionLogic = kea([ // The composer draft lives here so the input region is a real
. `submit` is the single entry // point the composer's `onSubmit` calls — it decides send-now vs enqueue vs new-run. It dispatches // synchronously (no await), so `isComposerFormSubmitting` isn't the UI loading state — `isSubmitting` - // (sending || startingRun) is. `errors` gates programmatic `submitComposerForm()`; the UI's own + // covers all three in-flight paths. `errors` gates programmatic `submitComposerForm()`; the UI's own // `Composer.Root` disabled-reason is the parallel guard. composerForm: { defaults: { draft: '' as string }, @@ -504,8 +546,18 @@ export const runInteractionLogic = kea([ return } // A finished run can't take a follow-up signal — send starts a fresh run instead, seeded with - // this message and chained from the run just viewed. + // this message and chained from the run just viewed. Except `/clear`, which the agent + // handles rather than the model: a whole run to clear a conversation the next run would + // rebuild from the log anyway, so the boundary is recorded against this run instead. + // + // Only when the chain's agent understands the boundary. An older one ignores the marker + // and resumes the conversation it was meant to retire, so falling back to an ordinary + // new run is the honest degradation — the clear doesn't happen, and nothing claims it did. if (values.isTerminal) { + if (isClearCommand(content) && values.conversationClearSupported) { + actions.clearConversation() + return + } actions.startNewRun(content) return } @@ -556,10 +608,10 @@ export const runInteractionLogic = kea([ (sending: boolean, isTerminal: boolean, currentProjectId: number | null): boolean => !sending && !isTerminal && currentProjectId != null, ], - // In-flight indicator for the composer's send button — a live send or a new-run start. + // In-flight indicator for the composer's send button — a live send, a new-run start, or a clear. isSubmitting: [ - (s) => [s.sending, s.startingRun], - (sending: boolean, startingRun: boolean): boolean => sending || startingRun, + (s) => [s.sending, s.startingRun, s.clearing], + (sending: boolean, startingRun: boolean, clearing: boolean): boolean => sending || startingRun || clearing, ], // Attached context not yet wrapped into a message for this task, the snapshot the next send wraps. // Two dedupe layers, both task-scoped (not run-scoped, so the dedupe survives a terminal-run send @@ -627,7 +679,9 @@ export const runInteractionLogic = kea([ } actions.setSending(true) const streamKey = props.streamKey ?? props.runId - const pendingContext = values.pendingContextItems + // `/clear` goes unwrapped: a context block would hide the command behind it (the + // agent reads the command off the front) and mark refs sent that nothing ever read. + const pendingContext = isClearCommand(content) ? [] : values.pendingContextItems actions.claimApplyBackTargets(streamKey) // Clear the draft synchronously before the await so text the user types while the send is in // flight isn't clobbered when the request resolves; a failed send restores it ahead of anything @@ -757,6 +811,25 @@ export const runInteractionLogic = kea([ actions.setStartingRun(false) } }, + + clearConversation: async () => { + if (values.clearing || values.currentProjectId == null) { + return + } + actions.setClearing(true) + try { + await tasksRunsClearConversationCreate(String(values.currentProjectId), props.taskId, props.runId) + actions.resetComposerForm() + // A finished run has no live stream to echo these back, so paint them from here. + // They match what the backend persisted, so a later replay folds the same thread. + actions.pushHumanMessage('/clear') + actions.pushConversationCleared() + } catch { + lemonToast.error('Failed to clear the conversation. Please try again.') + } finally { + actions.setClearing(false) + } + }, } }), ]) diff --git a/products/posthog_ai/frontend/logics/runStreamLogic.test.ts b/products/posthog_ai/frontend/logics/runStreamLogic.test.ts index f5af95514494..15d660efce3d 100644 --- a/products/posthog_ai/frontend/logics/runStreamLogic.test.ts +++ b/products/posthog_ai/frontend/logics/runStreamLogic.test.ts @@ -2516,6 +2516,39 @@ describe('runStreamLogic', () => { }) }) + describe('/clear inline items', () => { + it('replaces the in-progress clearing spinner with the conversation_cleared divider', async () => { + await expectLogic(logic, () => { + logic.actions.ingestAcpFrame(notification('_posthog/status', { status: 'clearing' })) + logic.actions.ingestAcpFrame(notification('_posthog/conversation_cleared', { sessionId: 'sess_new' })) + logic.actions.ingestAcpFrame(notification('_posthog/status', { status: 'clearing', isComplete: true })) + }).toFinishAllListeners() + + expect(logic.values.threadItems).toEqual([expect.objectContaining({ type: 'conversation_cleared' })]) + }) + + it('reports a failed clear in place of the spinner, since no boundary follows it', async () => { + await expectLogic(logic, () => { + logic.actions.ingestAcpFrame(notification('_posthog/status', { status: 'clearing' })) + logic.actions.ingestAcpFrame( + notification('_posthog/status', { + status: 'clearing_failed', + error: 'Conversation clear timed out after 30000ms', + }) + ) + }).toFinishAllListeners() + + expect(logic.values.threadItems).toEqual([ + expect.objectContaining({ + type: 'status', + status: 'clearing_failed', + isComplete: true, + errorMessage: 'Conversation clear timed out after 30000ms', + }), + ]) + }) + }) + describe('_posthog/task_notification inline item', () => { it('pushes a task_notification item carrying status + summary', async () => { await expectLogic(logic, () => { diff --git a/products/posthog_ai/frontend/logics/runStreamLogic.ts b/products/posthog_ai/frontend/logics/runStreamLogic.ts index 69bcdb668b21..72456f2a04b7 100644 --- a/products/posthog_ai/frontend/logics/runStreamLogic.ts +++ b/products/posthog_ai/frontend/logics/runStreamLogic.ts @@ -556,9 +556,9 @@ function findLastBufferIndex(state: ThreadItem[], id: string, type: ThreadItemTy return -1 } -/** The in-progress compaction spinner item — cleared when compaction completes or a boundary lands. */ -function isPendingCompactingStatus(item: ThreadItem): boolean { - return item.type === 'status' && item.status === 'compacting' && item.isComplete !== true +/** The in-progress spinner for a long-running status — retired when it completes, fails, or its boundary lands. */ +function isPendingStatus(item: ThreadItem, status: string): boolean { + return item.type === 'status' && item.status === status && item.isComplete !== true } function insertHumanMessageAtTurnStart(state: ThreadItem[], item: ThreadItem): ThreadItem[] { @@ -1016,6 +1016,7 @@ export function foldLogToThread(entries: StoredEntry[], options: { isResumeRun: let errorSeq = 0 let statusSeq = 0 let compactSeq = 0 + let clearedSeq = 0 let taskSeq = 0 let consoleSeq = 0 let contextSeq = 0 @@ -1214,15 +1215,26 @@ export function foldLogToThread(entries: StoredEntry[], options: { isResumeRun: if (method === '_posthog/status') { const status = String(params.status ?? '') const isComplete = params.isComplete === true - if (status === 'compacting' && isComplete) { - items = items.filter((item) => !isPendingCompactingStatus(item)) + if (isComplete && (status === 'compacting' || status === 'clearing')) { + items = items.filter((item) => !isPendingStatus(item, status)) + } else if (status === 'clearing_failed') { + // A failed clear emits no `conversation_cleared` marker, so retire the spinner + // here and report the outcome in its place. + items = items.filter((item) => !isPendingStatus(item, 'clearing')) + items.push({ + id: `status-${statusSeq++}`, + type: 'status', + status, + isComplete: true, + errorMessage: stringifyOptional(params.error), + }) } else { items.push({ id: `status-${statusSeq++}`, type: 'status', status, isComplete }) } continue } if (method === '_posthog/compact_boundary') { - items = items.filter((item) => !isPendingCompactingStatus(item)) + items = items.filter((item) => !isPendingStatus(item, 'compacting')) items.push({ id: `compact-${compactSeq++}`, type: 'compact_boundary', @@ -1232,6 +1244,13 @@ export function foldLogToThread(entries: StoredEntry[], options: { isResumeRun: }) continue } + if (method === '_posthog/conversation_cleared') { + // The divider supersedes the spinner visually, but the completing `_posthog/status` + // frame is a separate notification that may not have landed yet. + items = items.filter((item) => !isPendingStatus(item, 'clearing')) + items.push({ id: `cleared-${clearedSeq++}`, type: 'conversation_cleared' }) + continue + } if (method === '_posthog/task_notification') { items.push({ id: `task-${taskSeq++}`, @@ -1378,6 +1397,7 @@ export interface runStreamLogicValues { runArtifacts: RunArtifacts runConnectionState: RunConnectionState | null runOpening: boolean + conversationClearSupported: boolean runStarted: boolean sdkSession: SdkSession | null seenPermissionRequestIds: Set @@ -1478,6 +1498,9 @@ export interface runStreamLogicActions { markPermissionRequestSeen: (requestId: string) => { requestId: string } + markConversationClearSupported: () => { + value: true + } markRunStarted: () => { value: true } @@ -1507,6 +1530,9 @@ export interface runStreamLogicActions { permissionResponseFailed: () => { value: true } + pushConversationCleared: () => { + value: true + } pushErrorItem: ( errorMessage: string, variant?: 'crash' | 'error' @@ -1778,6 +1804,8 @@ export const runStreamLogic = kea([ /** Optional `task_run_state.stage` — wired for a future richer status surface (G6). */ setCurrentStage: (stage: string | null) => ({ stage }), markRunStarted: true, + /** Records the agent's `/clear` capability, read off the `_posthog/run_started` frame. */ + markConversationClearSupported: true, markTurnComplete: true, /** Echoes the user's own message into the thread as a `client`-sourced log entry (the wire never replays a live turn). */ pushHumanMessage: (content: string) => ({ content }), @@ -1792,6 +1820,8 @@ export const runStreamLogic = kea([ startOptimisticRun: (message?: string) => ({ message }), /** Injects a client-side error (terminal failure / stream disconnect) into the log as a `client`-sourced entry. */ pushErrorItem: (errorMessage: string, variant: 'error' | 'crash' = 'error') => ({ errorMessage, variant }), + /** Echoes a `/clear` boundary the backend just recorded against a finished run, which has no stream to send it back. */ + pushConversationCleared: true, /** Union the products an answer was grounded in — accumulates across the whole session. */ mergeResourcesUsed: (products: { id?: string; label?: string }[]) => ({ products }), /** Latest-wins merge of git artifacts (PR url / branch / base / repo) a run exposes. */ @@ -2051,6 +2081,15 @@ export const runStreamLogic = kea([ reset: () => false, }, ], + // Sticky across the resume chain's replay: any run in the chain having been served by a + // capable agent is enough, since the next run boots from the same (latest) agent image. + // `reset` is deliberately not handled — a re-bootstrap replays run_started and re-derives it. + conversationClearSupported: [ + false, + { + markConversationClearSupported: () => true, + }, + ], turnComplete: [ false, { @@ -3030,6 +3069,17 @@ export const runStreamLogic = kea([ }, ]) }, + pushConversationCleared: () => { + actions.appendEntries([ + { + entry: { + type: 'notification', + notification: { method: '_posthog/conversation_cleared', params: {} }, + }, + source: 'client', + }, + ]) + }, pushErrorItem: ({ errorMessage, variant }) => { // Client-side errors (terminal failure, stream disconnect) aren't wire frames — append // them as `client`-sourced log entries so the projection renders them in thread order. @@ -3109,6 +3159,9 @@ export const runStreamLogic = kea([ cold_start: true, }) } + if ((notification.params as { conversationClear?: unknown } | undefined)?.conversationClear === true) { + actions.markConversationClearSupported() + } cache.isBootstrapping = false actions.markRunStarted() return diff --git a/products/posthog_ai/frontend/types/streamTypes.ts b/products/posthog_ai/frontend/types/streamTypes.ts index 4bededf24c65..0f59831a4c65 100644 --- a/products/posthog_ai/frontend/types/streamTypes.ts +++ b/products/posthog_ai/frontend/types/streamTypes.ts @@ -120,6 +120,7 @@ export type ThreadItemType = | 'error' | 'status' | 'compact_boundary' + | 'conversation_cleared' | 'task_notification' | 'progress' | 'debug' @@ -140,7 +141,7 @@ export interface ThreadItem { complete?: boolean /** For `tool_invocation` items — the keyed tool call id (look up in `toolInvocations`). */ toolCallId?: string - /** For `error` items. */ + /** For `error` items, and for `status` items whose status is a `*_failed` phase. */ errorMessage?: string /** * For `error` items — distinguishes a friendlier agent-crash affordance (`crash`) from a diff --git a/products/posthog_ai/frontend/types/wireTypes.test.ts b/products/posthog_ai/frontend/types/wireTypes.test.ts index 8730c9435a8d..f130db648bc5 100644 --- a/products/posthog_ai/frontend/types/wireTypes.test.ts +++ b/products/posthog_ai/frontend/types/wireTypes.test.ts @@ -81,8 +81,13 @@ const NOTIFICATION_PARAMS_BY_METHOD: { [M in keyof PosthogNotificationParamsByMe '_posthog/status': [ { sessionId: 'sess_a1b2c3', status: 'compacting' }, { sessionId: 'sess_a1b2c3', status: 'compacting', isComplete: true }, + { sessionId: 'sess_a1b2c3', status: 'clearing' }, + { sessionId: 'sess_a1b2c3', status: 'clearing', isComplete: true }, + { sessionId: 'sess_a1b2c3', status: 'clearing_failed', error: 'Conversation clear timed out after 30000ms' }, ], '_posthog/compact_boundary': [{ sessionId: 'sess_a1b2c3', trigger: 'auto', preTokens: 168000, contextSize: 54000 }], + // The fresh agent session the `/clear` swapped in, not the one the run booted with. + '_posthog/conversation_cleared': [{ sessionId: 'sess_d4e5f6' }], '_posthog/task_notification': [ { sessionId: 'sess_a1b2c3', diff --git a/products/posthog_ai/frontend/types/wireTypes.ts b/products/posthog_ai/frontend/types/wireTypes.ts index 988d208a040c..78fa712b569f 100644 --- a/products/posthog_ai/frontend/types/wireTypes.ts +++ b/products/posthog_ai/frontend/types/wireTypes.ts @@ -331,6 +331,13 @@ export interface PosthogStatusParams { sessionId?: string status?: string isComplete?: boolean + /** Failure reason, set on a `*_failed` status (e.g. `clearing_failed`). */ + error?: string +} + +/** `/clear` boundary — `sessionId` is the fresh agent session swapped in behind it. */ +export interface PosthogConversationClearedParams { + sessionId?: string } export interface PosthogCompactBoundaryParams { @@ -383,6 +390,8 @@ export interface PosthogRunStartedParams { runId?: string taskId?: string agentVersion?: string + /** The agent implements `/clear` and honours the conversation-cleared boundary. Absent on older agents. */ + conversationClear?: boolean } export interface PosthogTurnCompleteParams { @@ -398,6 +407,7 @@ export interface PosthogNotificationParamsByMethod { '_posthog/usage_update': PosthogUsageUpdateParams '_posthog/status': PosthogStatusParams '_posthog/compact_boundary': PosthogCompactBoundaryParams + '_posthog/conversation_cleared': PosthogConversationClearedParams '_posthog/task_notification': PosthogTaskNotificationParams '_posthog/error': PosthogErrorParams '_posthog/sdk_session': PosthogSdkSessionParams diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py index f85089cd471a..af61a6944b61 100644 --- a/products/tasks/backend/facade/api.py +++ b/products/tasks/backend/facade/api.py @@ -2436,6 +2436,24 @@ def append_task_run_log( return _task_run_detail_to_dto(run) +def clear_task_run_conversation( + run_id: str | UUID, task_id: str | UUID, team_id: int +) -> tuple[Literal["cleared", "not_found", "not_terminal"], contracts.TaskRunDetailDTO | None]: + """Write a `/clear` boundary into a finished run's log, for the next run to resume from. + + Only for a finished run: a live one has a sandbox that owns the clear (and a writer + streaming into the same log object, which this read-modify-write append would race), + so the caller sends `/clear` to it as an ordinary message instead. + """ + run = _get_visible_run(run_id, task_id, team_id) + if run is None: + return "not_found", None + if not run.is_terminal: + return "not_terminal", None + run.emit_conversation_cleared() + return "cleared", _task_run_detail_to_dto(run) + + def ensure_task_run_session(run_id: str | UUID) -> UUID: with transaction.atomic(): run = TaskRun.objects.select_for_update(of=("self",)).select_related("task__team").get(id=run_id) diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index 70849c524975..320053bc236e 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -2288,6 +2288,54 @@ def emit_console_event(self, level: LogLevel, message: str) -> None: self.append_log([event]) self.publish_stream_event(event) + def emit_conversation_cleared(self) -> None: + """Record a `/clear` that had no sandbox to run it. + + A live run clears through the agent, which swaps in a fresh agent session and + emits this marker itself. A finished run has no sandbox, and booting one just to + clear it would cost a whole run, so the marker is written straight to the log. + Resume reads a chain's logs concatenated and rebuilds only the turns after the + marker, so the next run continues the task with an empty conversation while its + checkpoints, artifacts, and visible history stay intact. + + The `/clear` message is recorded ahead of the marker, matching the agent, so the + transcript shows what the user typed and rehydration drops it with everything + else on the pre-clear side. + + The marker carries no `sessionId`: there is no agent session behind it, and + resume reads that field to decide which session to continue. + """ + timestamp = django_timezone.now().isoformat() + events = [ + { + "type": "notification", + "timestamp": timestamp, + "notification": { + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": str(self.id), + "update": { + "sessionUpdate": "user_message_chunk", + "content": {"type": "text", "text": "/clear"}, + }, + }, + }, + }, + { + "type": "notification", + "timestamp": timestamp, + "notification": { + "jsonrpc": "2.0", + "method": "_posthog/conversation_cleared", + "params": {}, + }, + }, + ] + self.append_log(events) + for event in events: + self.publish_stream_event(event) + def emit_progress_event( self, step: str, diff --git a/products/tasks/backend/presentation/views/api.py b/products/tasks/backend/presentation/views/api.py index 2bc956ad173c..7de6f4ccf5e5 100644 --- a/products/tasks/backend/presentation/views/api.py +++ b/products/tasks/backend/presentation/views/api.py @@ -1301,6 +1301,43 @@ def append_log(self, request, pk=None, **kwargs): response["Server-Timing"] = timer.to_header_string() return response + @extend_schema( + request=None, + responses={ + 200: OpenApiResponse(response=TaskRunDetailSerializer, description="Run with the boundary recorded"), + 404: OpenApiResponse(description="Run not found"), + 409: OpenApiResponse( + response=TaskRunErrorResponseSerializer, description="Run is still active; send /clear to its agent" + ), + }, + summary="Clear conversation history", + description=( + "Record a `/clear` boundary in a finished run's log so the next run in the chain " + "starts with an empty conversation. Its checkpoints, artifacts, and visible history " + "are unaffected. Only for a finished run: an active one has an agent that owns the " + "clear, so send `/clear` to it as an ordinary message instead." + ), + ) + @action( + detail=True, + methods=["post"], + url_path="clear_conversation", + required_scopes=["task:write"], + ) + def clear_conversation(self, request, pk=None, **kwargs): + task_id = self._ensure_task_accessible() + outcome, run = tasks_facade.clear_task_run_conversation(pk, task_id, self.team_id) + if outcome == "not_found": + raise NotFound() + if outcome == "not_terminal": + return Response( + TaskRunErrorResponseSerializer({"error": "Run is still active; send /clear to its agent instead"}).data, + status=status.HTTP_409_CONFLICT, + ) + if run is None: + raise NotFound() + return Response(TaskRunDetailSerializer(run).data) + @extend_schema( responses={ 200: TaskSessionResponseSerializer, diff --git a/products/tasks/backend/tests/test_api.py b/products/tasks/backend/tests/test_api.py index 88674ecd1041..dbdda1662828 100644 --- a/products/tasks/backend/tests/test_api.py +++ b/products/tasks/backend/tests/test_api.py @@ -5448,6 +5448,37 @@ def test_append_log_entries(self): self.assertEqual(log_entries[1]["type"], "progress") self.assertEqual(log_entries[1]["message"], "Step 1 complete") + def test_clear_conversation_records_the_boundary(self): + task = self.create_task() + run = TaskRun.objects.create(task=task, team=self.team, status=TaskRun.Status.COMPLETED) + + response = self.client.post(f"/api/projects/@current/tasks/{task.id}/runs/{run.id}/clear_conversation/") + self.assertEqual(response.status_code, status.HTTP_200_OK) + + log_content = object_storage.read(run.log_url) + assert log_content is not None + entries = [json.loads(line)["notification"] for line in log_content.strip().split("\n")] + + # The typed message first, so rehydration drops it with the rest of the pre-clear side. + self.assertEqual(entries[0]["method"], "session/update") + self.assertEqual(entries[0]["params"]["update"]["content"]["text"], "/clear") + self.assertEqual(entries[1]["method"], "_posthog/conversation_cleared") + # No agent session stands behind this marker, and resume reads sessionId to pick the + # session it continues — carrying one would resume the conversation just cleared. + self.assertNotIn("sessionId", entries[1]["params"]) + + @parameterized.expand([("queued", TaskRun.Status.QUEUED), ("in_progress", TaskRun.Status.IN_PROGRESS)]) + def test_clear_conversation_rejects_an_active_run(self, _name, run_status): + # An active run's agent owns the clear, and its log has a live writer this + # read-modify-write append would race. + task = self.create_task() + run = TaskRun.objects.create(task=task, team=self.team, status=run_status) + + response = self.client.post(f"/api/projects/@current/tasks/{task.id}/runs/{run.id}/clear_conversation/") + + self.assertEqual(response.status_code, status.HTTP_409_CONFLICT) + self.assertIsNone(object_storage.read(run.log_url, missing_ok=True)) + @patch("products.tasks.backend.temporal.process_task.activities.post_slack_update.post_slack_update") def test_set_output_with_pr_url_posts_slack_update_when_mapping_exists(self, mock_post_slack_update): from posthog.models.integration import Integration diff --git a/products/tasks/frontend/generated/api.ts b/products/tasks/frontend/generated/api.ts index 08f12f31816e..e8a1c70407c6 100644 --- a/products/tasks/frontend/generated/api.ts +++ b/products/tasks/frontend/generated/api.ts @@ -1717,6 +1717,26 @@ export const tasksRunsCancelCreate = async ( }) } +export const getTasksRunsClearConversationCreateUrl = (projectId: string, taskId: string, id: string) => { + return `/api/projects/${projectId}/tasks/${taskId}/runs/${id}/clear_conversation/` +} + +/** + * Record a `/clear` boundary in a finished run's log so the next run in the chain starts with an empty conversation. Its checkpoints, artifacts, and visible history are unaffected. Only for a finished run: an active one has an agent that owns the clear, so send `/clear` to it as an ordinary message instead. + * @summary Clear conversation history + */ +export const tasksRunsClearConversationCreate = async ( + projectId: string, + taskId: string, + id: string, + options?: RequestInit +): Promise => { + return apiMutator(getTasksRunsClearConversationCreateUrl(projectId, taskId, id), { + ...options, + method: 'POST', + }) +} + export const getTasksRunsCommandCreateUrl = (projectId: string, taskId: string, id: string) => { return `/api/projects/${projectId}/tasks/${taskId}/runs/${id}/command/` } From 90a387c5afaf2a1ac0ec5ce070fea3b55fbf7cba Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Mon, 3 Aug 2026 13:58:45 -0700 Subject: [PATCH 2/4] chore(tasks): register clear_conversation as a (disabled) MCP tool Generated registration for the new endpoint, carried over from the original branch. Disabled like its siblings; the entry keeps the MCP codegen from drifting against the OpenAPI spec. --- products/tasks/mcp/tools.yaml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/products/tasks/mcp/tools.yaml b/products/tasks/mcp/tools.yaml index c3a496c78254..9045c20a7fc9 100644 --- a/products/tasks/mcp/tools.yaml +++ b/products/tasks/mcp/tools.yaml @@ -450,6 +450,9 @@ tools: tasks-runs-cancel-create: operation: tasks_runs_cancel_create enabled: false + tasks-runs-clear-conversation-create: + operation: tasks_runs_clear_conversation_create + enabled: false tasks-runs-command-create: operation: tasks_runs_command_create enabled: false From 503f3d8db7949d21a911838151faa30e3b4202a0 Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Thu, 6 Aug 2026 13:56:14 -0700 Subject: [PATCH 3/4] fix(tasks): address review findings on the finished-run /clear - Paint the /clear as a session/prompt request and tag the logged chunk so desktop renders the bubble live and after reload - Skip the append when the log already ends at the boundary, so repeats don't stack duplicate markers - Hold the run's row lock across the terminal check and append, so a concurrent resume can't flip the run active mid-write - Surface the 409 body's error detail instead of bare statusText - Log before the clear call, matching sibling cloud operations Generated-By: PostHog Code Task-Id: 2578f561-b94e-46ac-8546-d3368f5098bb --- .../packages/api-client/src/posthog-client.ts | 12 +++++- .../core/src/sessions/sessionEvents.ts | 37 ++++++------------- .../core/src/sessions/sessionService.ts | 14 ++++++- .../sessions/sessionServiceCloudClear.test.ts | 4 +- products/tasks/backend/facade/api.py | 13 +++++-- products/tasks/backend/models.py | 27 +++++++++++++- products/tasks/backend/tests/test_api.py | 10 +++++ 7 files changed, 83 insertions(+), 34 deletions(-) diff --git a/products/desktop/packages/api-client/src/posthog-client.ts b/products/desktop/packages/api-client/src/posthog-client.ts index a01fe53a6ec8..4b85b48c437f 100644 --- a/products/desktop/packages/api-client/src/posthog-client.ts +++ b/products/desktop/packages/api-client/src/posthog-client.ts @@ -3445,7 +3445,17 @@ export class PostHogAPIClient { path, }); if (!response.ok) { - throw new Error(`Failed to clear conversation: ${response.statusText}`); + const err = (await response.json().catch(() => ({}))) as { + error?: unknown; + detail?: unknown; + }; + const reason = + typeof err.error === "string" + ? err.error + : typeof err.detail === "string" + ? err.detail + : response.statusText; + throw new Error(`Failed to clear conversation: ${reason}`); } } diff --git a/products/desktop/packages/core/src/sessions/sessionEvents.ts b/products/desktop/packages/core/src/sessions/sessionEvents.ts index 7e1058a12d7f..34c44e900828 100644 --- a/products/desktop/packages/core/src/sessions/sessionEvents.ts +++ b/products/desktop/packages/core/src/sessions/sessionEvents.ts @@ -83,8 +83,9 @@ function storedEntryToAcpMessage( * A typed user prompt replayed from an imported Claude Code session arrives as * a `user_message_chunk` tagged with `_meta.importedUserPrompt`. The renderer * ignores raw user_message_chunks (live, user turns render from session/prompt - * requests), so promote the tagged ones into a session/prompt user event. Only - * affects imported sessions; normal logs carry no such marker. + * requests), so promote the tagged ones into a session/prompt user event. + * Imported sessions and the backend-recorded `/clear` on a finished cloud run + * carry the tag; normal logs don't. */ function promoteImportedUserPrompt( entry: StoredLogEntry, @@ -136,34 +137,18 @@ export function createUserMessageEvent(text: string, ts: number): AcpMessage { } /** - * The two frames a `/clear` on a finished cloud run produces, in log order: the + * The two frames a `/clear` on a finished cloud run paints, in thread order: the * message the user typed, then the boundary rehydration stops at. * - * The backend writes these into the run log (there is no sandbox to emit them), - * and mirrors this shape — a finished run has no live stream to echo them back, - * so the client paints them from here and a later reload folds the persisted - * copies into the same thread. + * The backend writes the same pair into the run log (there is no sandbox to emit + * them). The painted user message is a `session/prompt` request because that is + * the shape the renderer displays; the persisted copy is a `user_message_chunk` + * tagged `importedUserPrompt`, which log replay promotes back into this same + * request shape (see {@link promoteImportedUserPrompt}). */ -export function createConversationClearedEvents( - sessionId: string, - ts: number, -): AcpMessage[] { +export function createConversationClearedEvents(ts: number): AcpMessage[] { return [ - { - type: "acp_message", - ts, - message: { - jsonrpc: "2.0", - method: "session/update", - params: { - sessionId, - update: { - sessionUpdate: "user_message_chunk", - content: { type: "text", text: "/clear" }, - }, - }, - }, - }, + createUserMessageEvent("/clear", ts), { type: "acp_message", ts, diff --git a/products/desktop/packages/core/src/sessions/sessionService.ts b/products/desktop/packages/core/src/sessions/sessionService.ts index 5996a869da97..548fc646e262 100644 --- a/products/desktop/packages/core/src/sessions/sessionService.ts +++ b/products/desktop/packages/core/src/sessions/sessionService.ts @@ -145,7 +145,13 @@ const SESSION_EVENT_EVICT_GRACE_MS = 20_000; */ const OPEN_TAIL_BYTES = 1_500_000; -/** Matches a bare `/clear` invocation, not a longer command that starts with it. */ +/** + * Matches a leading `/clear` command the way the agent's own detection does + * (`leadingSlashCommand` in the claude adapter): a longer command such as + * `/clearcache` doesn't match, and trailing text doesn't change the command, + * so the finished-run shortcut and a live agent treat the same message the + * same way. + */ function isClearCommand(text: string | undefined): boolean { return /^\/clear(?:\s|$)/.test(text ?? ""); } @@ -4447,10 +4453,14 @@ export class SessionService { if (!client) { throw new Error("Authentication required for cloud commands"); } + this.d.log.info("Clearing cloud conversation", { + taskId: session.taskId, + taskRunId: session.taskRunId, + }); await client.clearTaskRunConversation(session.taskId, session.taskRunId); this.d.store.appendEvents( session.taskRunId, - createConversationClearedEvents(session.taskRunId, Date.now()), + createConversationClearedEvents(Date.now()), ); return { stopReason: "end_turn" }; } diff --git a/products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts b/products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts index 09141bb3fe7b..5d4e355c5fca 100644 --- a/products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts +++ b/products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts @@ -93,10 +93,12 @@ describe("SessionService /clear on a finished cloud run", () => { expect(runTaskInCloud).not.toHaveBeenCalled(); // A finished run streams nothing back, so the thread is painted from here. + // The user message must be a session/prompt request: the renderer drops raw + // user_message_chunks, so painting one would show only the divider. const [, events] = appendEvents.mock.calls[0]; expect( events.map((e: { message: { method: string } }) => e.message.method), - ).toEqual(["session/update", POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED]); + ).toEqual(["session/prompt", POSTHOG_NOTIFICATIONS.CONVERSATION_CLEARED]); }); it("resumes into a new run when the agent cannot honour the boundary", async () => { diff --git a/products/tasks/backend/facade/api.py b/products/tasks/backend/facade/api.py index af61a6944b61..9c2a280a0c5b 100644 --- a/products/tasks/backend/facade/api.py +++ b/products/tasks/backend/facade/api.py @@ -2448,9 +2448,16 @@ def clear_task_run_conversation( run = _get_visible_run(run_id, task_id, team_id) if run is None: return "not_found", None - if not run.is_terminal: - return "not_terminal", None - run.emit_conversation_cleared() + with transaction.atomic(): + # Hold the row lock across the append: resume_task_run_in_cloud locks this same + # row to flip a finished run back to QUEUED, so locking here keeps the terminal + # check true while the boundary is written, and serializes concurrent clears so + # the dedup in emit_conversation_cleared holds. The block writes nothing to + # Postgres; the lock is mutual exclusion only. + run = _task_run_queryset().select_for_update(of=("self",)).get(pk=run.pk) + if not run.is_terminal: + return "not_terminal", None + run.emit_conversation_cleared() return "cleared", _task_run_detail_to_dto(run) diff --git a/products/tasks/backend/models.py b/products/tasks/backend/models.py index 320053bc236e..911ecfdc7e29 100644 --- a/products/tasks/backend/models.py +++ b/products/tasks/backend/models.py @@ -1,5 +1,6 @@ import os import re +import json import uuid import string import secrets @@ -2300,11 +2301,18 @@ def emit_conversation_cleared(self) -> None: The `/clear` message is recorded ahead of the marker, matching the agent, so the transcript shows what the user typed and rehydration drops it with everything - else on the pre-clear side. + else on the pre-clear side. It carries the `importedUserPrompt` tag because the + desktop client renders user turns from `session/prompt` requests and drops raw + `user_message_chunk`s; the tag tells its log replay to promote the chunk into one. The marker carries no `sessionId`: there is no agent session behind it, and resume reads that field to decide which session to continue. + + A repeat call while the log already ends at the boundary appends nothing, so a + double-submitted or retried clear doesn't stack duplicate markers. """ + if self._log_tail_is_conversation_cleared(): + return timestamp = django_timezone.now().isoformat() events = [ { @@ -2318,6 +2326,7 @@ def emit_conversation_cleared(self) -> None: "update": { "sessionUpdate": "user_message_chunk", "content": {"type": "text", "text": "/clear"}, + "_meta": {"importedUserPrompt": True}, }, }, }, @@ -2336,6 +2345,22 @@ def emit_conversation_cleared(self) -> None: for event in events: self.publish_stream_event(event) + def _log_tail_is_conversation_cleared(self) -> bool: + # Reads the whole object because S3 offers no cheap tail read; clears are rare + # and the subsequent append re-reads it anyway. + content = object_storage.read(self.log_url, missing_ok=True) or "" + last_line = content.strip().rsplit("\n", 1)[-1] + if not last_line: + return False + try: + entry = json.loads(last_line) + except json.JSONDecodeError: + return False + if not isinstance(entry, dict): + return False + notification = entry.get("notification") + return isinstance(notification, dict) and notification.get("method") == "_posthog/conversation_cleared" + def emit_progress_event( self, step: str, diff --git a/products/tasks/backend/tests/test_api.py b/products/tasks/backend/tests/test_api.py index dbdda1662828..93f865e37901 100644 --- a/products/tasks/backend/tests/test_api.py +++ b/products/tasks/backend/tests/test_api.py @@ -5462,11 +5462,21 @@ def test_clear_conversation_records_the_boundary(self): # The typed message first, so rehydration drops it with the rest of the pre-clear side. self.assertEqual(entries[0]["method"], "session/update") self.assertEqual(entries[0]["params"]["update"]["content"]["text"], "/clear") + # The desktop client renders user turns from session/prompt requests and drops raw + # user_message_chunks; this tag tells its log replay to promote the chunk into one. + self.assertEqual(entries[0]["params"]["update"]["_meta"], {"importedUserPrompt": True}) self.assertEqual(entries[1]["method"], "_posthog/conversation_cleared") # No agent session stands behind this marker, and resume reads sessionId to pick the # session it continues — carrying one would resume the conversation just cleared. self.assertNotIn("sessionId", entries[1]["params"]) + # A repeat clear with nothing recorded since must not stack another boundary. + response = self.client.post(f"/api/projects/@current/tasks/{task.id}/runs/{run.id}/clear_conversation/") + self.assertEqual(response.status_code, status.HTTP_200_OK) + log_content = object_storage.read(run.log_url) + assert log_content is not None + self.assertEqual(len(log_content.strip().split("\n")), 2) + @parameterized.expand([("queued", TaskRun.Status.QUEUED), ("in_progress", TaskRun.Status.IN_PROGRESS)]) def test_clear_conversation_rejects_an_active_run(self, _name, run_status): # An active run's agent owns the clear, and its log has a live writer this From cca74dab71d313b4ea0f2a9bc36910ab4949a18f Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Thu, 6 Aug 2026 16:20:21 -0700 Subject: [PATCH 4/4] fix(tasks): drop the web /clear gate when the latest agent lacks the capability The reducer ratcheted permanently true, so after an agent rollback an earlier capable run kept authorizing the terminal-run clear endpoint and the UI painted a clear the next agent would ignore on resume. Follow the latest run_started's advertisement instead, matching the desktop client. Generated-By: PostHog Code Task-Id: 2578f561-b94e-46ac-8546-d3368f5098bb --- .../frontend/logics/runStreamLogic.test.ts | 13 ++++++++++ .../frontend/logics/runStreamLogic.ts | 24 ++++++++++--------- 2 files changed, 26 insertions(+), 11 deletions(-) diff --git a/products/posthog_ai/frontend/logics/runStreamLogic.test.ts b/products/posthog_ai/frontend/logics/runStreamLogic.test.ts index 15d660efce3d..62ff44960afd 100644 --- a/products/posthog_ai/frontend/logics/runStreamLogic.test.ts +++ b/products/posthog_ai/frontend/logics/runStreamLogic.test.ts @@ -262,6 +262,19 @@ describe('runStreamLogic', () => { expect(logic.values.threadItems.some((item) => item.type === 'turn_separator')).toEqual(true) }) + it('follows the latest run_started conversationClear advertisement', async () => { + // A run served by a capable agent followed by one whose agent does not advertise + // the capability (an agent rollback): the gate must drop, or the client records a + // clear boundary the current agent ignores on resume. + await expectLogic(logic, () => { + logic.actions.ingestAcpFrame(notification('_posthog/run_started', { conversationClear: true })) + }).toMatchValues({ conversationClearSupported: true }) + + await expectLogic(logic, () => { + logic.actions.ingestAcpFrame(notification('_posthog/run_started', {})) + }).toMatchValues({ conversationClearSupported: false }) + }) + it('sets currentMode on a current_mode_update frame', async () => { await expectLogic(logic, () => { logic.actions.ingestAcpFrame( diff --git a/products/posthog_ai/frontend/logics/runStreamLogic.ts b/products/posthog_ai/frontend/logics/runStreamLogic.ts index 72456f2a04b7..4f603b7fbca5 100644 --- a/products/posthog_ai/frontend/logics/runStreamLogic.ts +++ b/products/posthog_ai/frontend/logics/runStreamLogic.ts @@ -1498,8 +1498,8 @@ export interface runStreamLogicActions { markPermissionRequestSeen: (requestId: string) => { requestId: string } - markConversationClearSupported: () => { - value: true + setConversationClearSupported: (supported: boolean) => { + supported: boolean } markRunStarted: () => { value: true @@ -1804,8 +1804,8 @@ export const runStreamLogic = kea([ /** Optional `task_run_state.stage` — wired for a future richer status surface (G6). */ setCurrentStage: (stage: string | null) => ({ stage }), markRunStarted: true, - /** Records the agent's `/clear` capability, read off the `_posthog/run_started` frame. */ - markConversationClearSupported: true, + /** Records the agent's `/clear` capability, read off each `_posthog/run_started` frame. */ + setConversationClearSupported: (supported: boolean) => ({ supported }), markTurnComplete: true, /** Echoes the user's own message into the thread as a `client`-sourced log entry (the wire never replays a live turn). */ pushHumanMessage: (content: string) => ({ content }), @@ -2081,13 +2081,15 @@ export const runStreamLogic = kea([ reset: () => false, }, ], - // Sticky across the resume chain's replay: any run in the chain having been served by a - // capable agent is enough, since the next run boots from the same (latest) agent image. - // `reset` is deliberately not handled — a re-bootstrap replays run_started and re-derives it. + // The latest run_started in the resume chain wins, matching the desktop client: the gate + // predicts the next run's agent, and after an agent rollback an earlier capable run must + // not authorize recording a boundary the current agent would ignore on resume (the UI + // would claim a clear that never happens). `reset` is deliberately not handled: a + // re-bootstrap replays the chain's run_started frames and re-derives it. conversationClearSupported: [ false, { - markConversationClearSupported: () => true, + setConversationClearSupported: (_, { supported }) => supported, }, ], turnComplete: [ @@ -3159,9 +3161,9 @@ export const runStreamLogic = kea([ cold_start: true, }) } - if ((notification.params as { conversationClear?: unknown } | undefined)?.conversationClear === true) { - actions.markConversationClearSupported() - } + actions.setConversationClearSupported( + (notification.params as { conversationClear?: unknown } | undefined)?.conversationClear === true + ) cache.isBootstrapping = false actions.markRunStarted() return