diff --git a/products/desktop/packages/api-client/src/posthog-client.ts b/products/desktop/packages/api-client/src/posthog-client.ts index 2c5df489f998..4b85b48c437f 100644 --- a/products/desktop/packages/api-client/src/posthog-client.ts +++ b/products/desktop/packages/api-client/src/posthog-client.ts @@ -3431,6 +3431,34 @@ 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) { + 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}`); + } + } + 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..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, @@ -135,6 +136,31 @@ export function createUserMessageEvent(text: string, ts: number): AcpMessage { return createUserPromptEvent([{ type: "text", text }], ts); } +/** + * 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 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(ts: number): AcpMessage[] { + return [ + createUserMessageEvent("/clear", ts), + { + 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..548fc646e262 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,17 @@ const SESSION_EVENT_EVICT_GRACE_MS = 20_000; */ const OPEN_TAIL_BYTES = 1_500_000; +/** + * 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 ?? ""); +} + class GitHubAuthorizationRequiredForCloudHandoffError extends Error { constructor( message = "Connect GitHub before continuing this task in cloud.", @@ -4119,6 +4131,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 +4445,26 @@ 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"); + } + 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(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..5d4e355c5fca --- /dev/null +++ b/products/desktop/packages/core/src/sessions/sessionServiceCloudClear.test.ts @@ -0,0 +1,123 @@ +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. + // 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/prompt", 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..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( @@ -2516,6 +2529,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..4f603b7fbca5 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 } + setConversationClearSupported: (supported: boolean) => { + supported: boolean + } 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 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 }), @@ -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,17 @@ export const runStreamLogic = kea([ reset: () => false, }, ], + // 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, + { + setConversationClearSupported: (_, { supported }) => supported, + }, + ], turnComplete: [ false, { @@ -3030,6 +3071,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 +3161,9 @@ export const runStreamLogic = kea([ cold_start: true, }) } + actions.setConversationClearSupported( + (notification.params as { conversationClear?: unknown } | undefined)?.conversationClear === true + ) 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..9c2a280a0c5b 100644 --- a/products/tasks/backend/facade/api.py +++ b/products/tasks/backend/facade/api.py @@ -2436,6 +2436,31 @@ 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 + 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) + + 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..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 @@ -2288,6 +2289,78 @@ 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. 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 = [ + { + "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"}, + "_meta": {"importedUserPrompt": True}, + }, + }, + }, + }, + { + "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 _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/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..93f865e37901 100644 --- a/products/tasks/backend/tests/test_api.py +++ b/products/tasks/backend/tests/test_api.py @@ -5448,6 +5448,47 @@ 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") + # 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 + # 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/` } 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