From 5ebae2f2303635907dda08b70ff3ca054ae29fde Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Fri, 17 Jul 2026 13:42:34 +0100 Subject: [PATCH] fix(cloud): hydrate resumed transcripts atomically --- packages/core/src/sessions/sessionService.ts | 538 ++++++-- .../sessions/sessionServiceHost.test.ts | 1078 +++++++++++++++-- 2 files changed, 1393 insertions(+), 223 deletions(-) diff --git a/packages/core/src/sessions/sessionService.ts b/packages/core/src/sessions/sessionService.ts index a3ce8c66c0..fc34739647 100644 --- a/packages/core/src/sessions/sessionService.ts +++ b/packages/core/src/sessions/sessionService.ts @@ -152,6 +152,26 @@ type TrpcSubscription = { ) => { unsubscribe: () => void }; }; +interface CloudHydrationResult { + historyEntryCount: number; + liveStreamLineCount: number; +} + +interface CloudTaskWatcher { + runId: string; + apiHost: string; + teamId: number; + startToken: number; + resumeFromEntryCount?: number; + resumeHistoryCountOffset?: number; + resumeHydrationToken: number; + bufferResumeUpdates: boolean; + bufferedResumeUpdates: CloudTaskUpdatePayload[]; + processCloudUpdate: (update: CloudTaskUpdatePayload) => void; + subscription: { unsubscribe: () => void }; + onStatusChange?: () => void; +} + export interface SessionTrpc { agent: { start: TrpcMutation; @@ -529,6 +549,32 @@ function entriesScopedToTaskRun( }); } +function suffixPrefixOverlap(left: string[], right: string[]): number { + if (left.length === 0 || right.length === 0) return 0; + + const separator = Symbol("resume-chain-separator"); + const patternAndTail: (string | symbol)[] = [ + ...right, + separator, + ...left.slice(-right.length), + ]; + const prefixLengths = new Array(patternAndTail.length).fill(0); + for (let index = 1; index < patternAndTail.length; index += 1) { + let prefixLength = prefixLengths[index - 1]; + while ( + prefixLength > 0 && + patternAndTail[index] !== patternAndTail[prefixLength] + ) { + prefixLength = prefixLengths[prefixLength - 1]; + } + if (patternAndTail[index] === patternAndTail[prefixLength]) { + prefixLength += 1; + } + prefixLengths[index] = prefixLength; + } + return prefixLengths[prefixLengths.length - 1]; +} + function appendHydrationHash(hash: number, value: string): number { let nextHash = hash; for (let index = 0; index < value.length; index += 1) { @@ -1528,17 +1574,7 @@ export class SessionService { } >(); /** Active cloud task watchers, keyed by taskId */ - private cloudTaskWatchers = new Map< - string, - { - runId: string; - apiHost: string; - teamId: number; - startToken: number; - subscription: { unsubscribe: () => void }; - onStatusChange?: () => void; - } - >(); + private cloudTaskWatchers = new Map(); private cloudLogGapReconciler: CloudLogGapReconciler; /** Maps toolCallId → cloud requestId for routing permission responses */ private cloudPermissionRequestIds = new Map(); @@ -1554,6 +1590,10 @@ export class SessionService { { startedAtTs: number; agentTextChunks: number; agentOutputEvents: number } >(); private pendingPermissionHydratedRuns = new Set(); + private cloudHydrationPromises = new Map< + string, + Promise + >(); private idleKilledSubscription: { unsubscribe: () => void } | null = null; /** * Cached preview-config-options responses keyed by `${apiHost}::${adapter}`. @@ -4324,8 +4364,10 @@ export class SessionService { newSession.optimisticItems = ( this.getSessionByRunId(session.taskRunId)?.optimisticItems ?? [] ).filter((item) => item.type === "user_message" && item.pinToTop === false); - const resumeFromEntryCount = session.processedLineCount ?? 0; - newSession.processedLineCount = resumeFromEntryCount; + const resumeFromEntryCount = + session.cloudTranscriptEntryCount ?? session.processedLineCount ?? 0; + newSession.cloudTranscriptEntryCount = resumeFromEntryCount; + newSession.processedLineCount = 0; this.d.store.setSession(newSession); // Start the watcher immediately so we don't miss status updates. @@ -5465,6 +5507,19 @@ export class SessionService { initialReasoningEffort, ); } + if ( + typeof runState?.resume_from_run_id === "string" && + !this.pendingPermissionHydratedRuns.has(taskRunId) + ) { + void this.hydrateResumeCloudTaskSessionFromLogs( + taskId, + taskRunId, + logUrl, + taskDescription, + runStatus, + runState, + ); + } return () => {}; } @@ -5523,12 +5578,16 @@ export class SessionService { !this.pendingPermissionHydratedRuns.has(taskRunId) && (isTerminalStatus(existing.cloudStatus) || (runStatus !== undefined && isTerminalStatus(runStatus))); + const shouldHydrateResumeChain = + Boolean(runState?.resume_from_run_id) && + !this.pendingPermissionHydratedRuns.has(taskRunId); const shouldHydrateSession = !existing || existing.taskRunId !== taskRunId || shouldResetExistingSession || existing.events.length === 0 || - shouldHydratePersistedPermissions; + shouldHydratePersistedPermissions || + shouldHydrateResumeChain; if ( !existing || @@ -5589,47 +5648,97 @@ export class SessionService { initialReasoningEffort, ); - if (shouldHydrateSession) { - this.hydrateCloudTaskSessionFromLogs( - taskId, - taskRunId, - logUrl, - taskDescription, - runStatus, - runState, - ); - } + const processCloudUpdate = (update: CloudTaskUpdatePayload): void => { + if (update.kind === "logs" || update.kind === "snapshot") { + this.d.store.updateSession(taskRunId, { + cloudTranscriptEntryCount: update.totalEntryCount, + }); + } + const watcher = this.cloudTaskWatchers.get(taskId); + const resumeHistoryCountOffset = + watcher?.runId === runId ? (watcher.resumeHistoryCountOffset ?? 0) : 0; + const normalizedUpdate: CloudTaskUpdatePayload = + resumeHistoryCountOffset > 0 && + (update.kind === "logs" || update.kind === "snapshot") + ? { + ...update, + totalEntryCount: Math.max( + 0, + update.totalEntryCount - resumeHistoryCountOffset, + ), + } + : update; + this.handleCloudTaskUpdate(taskRunId, normalizedUpdate); + if ( + (update.kind === "status" || + update.kind === "snapshot" || + update.kind === "error") && + watcher?.onStatusChange + ) { + watcher.onStatusChange(); + } + }; + + const watcher: CloudTaskWatcher = { + runId, + apiHost, + teamId, + startToken, + resumeFromEntryCount, + resumeHistoryCountOffset: shouldHydrateResumeChain + ? resumeFromEntryCount + : 0, + resumeHydrationToken: 0, + bufferResumeUpdates: false, + bufferedResumeUpdates: [], + processCloudUpdate, + subscription: { unsubscribe: () => undefined }, + onStatusChange, + }; + this.cloudTaskWatchers.set(taskId, watcher); // Subscribe before starting the main-process watcher so the first replayed // SSE/log burst cannot race ahead of the renderer subscription. - const subscription = this.d.trpc.cloudTask.onUpdate.subscribe( + watcher.subscription = this.d.trpc.cloudTask.onUpdate.subscribe( { taskId, runId }, { onData: (update: CloudTaskUpdatePayload) => { - this.handleCloudTaskUpdate(taskRunId, update); - const watcher = this.cloudTaskWatchers.get(taskId); - if ( - (update.kind === "status" || - update.kind === "snapshot" || - update.kind === "error") && - watcher?.onStatusChange - ) { - watcher.onStatusChange(); + const activeWatcher = this.cloudTaskWatchers.get(taskId); + if (!activeWatcher || activeWatcher.runId !== runId) { + return; } + if (activeWatcher.bufferResumeUpdates) { + activeWatcher.bufferedResumeUpdates.push(update); + return; + } + activeWatcher.processCloudUpdate(update); }, onError: (err: unknown) => this.d.log.error("Cloud task subscription error", { taskId, err }), }, ); - this.cloudTaskWatchers.set(taskId, { - runId, - apiHost, - teamId, - startToken, - subscription, - onStatusChange, - }); + if (shouldHydrateSession) { + if (shouldHydrateResumeChain) { + void this.hydrateResumeCloudTaskSessionFromLogs( + taskId, + taskRunId, + logUrl, + taskDescription, + runStatus, + runState, + ); + } else { + void this.hydrateCloudTaskSessionFromLogs( + taskId, + taskRunId, + logUrl, + taskDescription, + runStatus, + runState, + ); + } + } // Start main-process watcher after the subscription is attached. void (async () => { @@ -5683,108 +5792,277 @@ export class SessionService { taskDescription?: string, runStatus?: TaskRunStatus, runState?: Record, + ): Promise { + const existing = this.cloudHydrationPromises.get(taskRunId); + if (existing) { + return existing; + } + const hydration = this.performCloudTaskSessionHydration( + taskId, + taskRunId, + logUrl, + taskDescription, + runStatus, + runState, + ).catch((err: unknown) => { + this.d.log.warn("Failed to hydrate cloud task session from logs", { + taskId, + taskRunId, + err, + }); + return undefined; + }); + this.cloudHydrationPromises.set(taskRunId, hydration); + void hydration.finally(() => { + if (this.cloudHydrationPromises.get(taskRunId) === hydration) { + this.cloudHydrationPromises.delete(taskRunId); + } + }); + return hydration; + } + + private async hydrateResumeCloudTaskSessionFromLogs( + taskId: string, + taskRunId: string, + logUrl?: string, + taskDescription?: string, + runStatus?: TaskRunStatus, + runState?: Record, + ): Promise { + const watcher = this.cloudTaskWatchers.get(taskId); + if (!watcher || watcher.runId !== taskRunId) return; + const hydrationToken = ++watcher.resumeHydrationToken; + watcher.bufferResumeUpdates = true; + + const result = await this.hydrateCloudTaskSessionFromLogs( + taskId, + taskRunId, + logUrl, + taskDescription, + runStatus, + runState, + ); + const activeWatcher = this.cloudTaskWatchers.get(taskId); + if ( + !activeWatcher || + activeWatcher.runId !== taskRunId || + activeWatcher.resumeHydrationToken !== hydrationToken + ) { + return; + } + + this.applyResumeHydrationOffset(taskId, taskRunId, result); + activeWatcher.bufferResumeUpdates = false; + const bufferedUpdates = activeWatcher.bufferedResumeUpdates.splice(0); + for (const update of bufferedUpdates) { + activeWatcher.processCloudUpdate(update); + } + } + + private applyResumeHydrationOffset( + taskId: string, + taskRunId: string, + result: CloudHydrationResult | undefined, ): void { - void (async () => { - let rawEntries: StoredLogEntry[]; - let totalLineCount: number; - const isResumeRun = Boolean(runState?.resume_from_run_id); - if (isTerminalStatus(runStatus) || isResumeRun) { - // Resume chains need the full history even while the leaf run is still - // active; otherwise a renderer restart hydrates only the final run. - // Non-resume in-progress runs keep using the single-run log so hydrate - // cannot race the live stream and double the active turn. - const authStatus = await this.getAuthCredentialsStatus(); - if (authStatus.kind !== "ready") { - return; - } - try { - rawEntries = await authStatus.auth.client.getTaskRunSessionLogs( + if (!result) return; + const watcher = this.cloudTaskWatchers.get(taskId); + if (!watcher || watcher.runId !== taskRunId) return; + watcher.resumeHistoryCountOffset = Math.max( + 0, + result.historyEntryCount - result.liveStreamLineCount, + ); + } + + private async performCloudTaskSessionHydration( + taskId: string, + taskRunId: string, + logUrl?: string, + taskDescription?: string, + runStatus?: TaskRunStatus, + runState?: Record, + ): Promise { + let rawEntries: StoredLogEntry[]; + let liveStreamLineCount: number; + let resumeLeafEntryStartIndex: number | undefined; + const resumeFromRunId = + typeof runState?.resume_from_run_id === "string" + ? runState.resume_from_run_id + : undefined; + const isResumeRun = Boolean(resumeFromRunId); + if (isTerminalStatus(runStatus) || isResumeRun) { + // Resume chains need the full history even while the leaf run is still + // active; otherwise a renderer restart hydrates only the final run. + // Non-resume in-progress runs keep using the single-run log so hydrate + // cannot race the live stream and double the active turn. + const authStatus = await this.getAuthCredentialsStatus(); + if (authStatus.kind !== "ready") { + return; + } + if (resumeFromRunId) { + const [ancestorResult, currentRunResult] = await Promise.all([ + authStatus.auth.client.getTaskRunSessionLogsResult( + taskId, + resumeFromRunId, + { limit: 100000 }, + ), + authStatus.auth.client.getTaskRunSessionLogsResult( taskId, taskRunId, { limit: 100000 }, - ); - } catch (err) { - this.d.log.warn("Failed to fetch session-log chain for hydrate", { + ), + ]); + if (!ancestorResult.complete || !currentRunResult.complete) { + this.d.log.warn("Resume session log hydration was incomplete", { taskId, taskRunId, - err, + resumeFromRunId, + ancestorComplete: ancestorResult.complete, + currentRunComplete: currentRunResult.complete, }); return; } - totalLineCount = rawEntries.length; + const ancestorEntries: StoredLogEntry[] = ancestorResult.entries; + const currentRunEntries: StoredLogEntry[] = currentRunResult.entries; + + const ancestorKeys = ancestorEntries.map((entry) => + JSON.stringify(entry), + ); + const currentKeys = currentRunEntries.map((entry) => + JSON.stringify(entry), + ); + const overlap = suffixPrefixOverlap(ancestorKeys, currentKeys); + const persistedLeafEntries = currentRunEntries.slice(overlap); + const leafLogs = await this.fetchSessionLogs(logUrl, taskRunId); + const leafKeys = new Set( + persistedLeafEntries.map((entry) => JSON.stringify(entry)), + ); + rawEntries = [ + ...ancestorEntries, + ...persistedLeafEntries, + ...leafLogs.rawEntries.filter( + (entry) => !leafKeys.has(JSON.stringify(entry)), + ), + ]; + resumeLeafEntryStartIndex = ancestorEntries.length; + liveStreamLineCount = Math.max( + leafLogs.totalLineCount, + persistedLeafEntries.length, + ); } else { - const parsed = await this.fetchSessionLogs(logUrl, taskRunId); - rawEntries = parsed.rawEntries; - totalLineCount = parsed.totalLineCount; + const result = await authStatus.auth.client.getTaskRunSessionLogsResult( + taskId, + taskRunId, + { limit: 100000 }, + ); + if (!result.complete) { + this.d.log.warn("Session log hydration was incomplete", { + taskId, + taskRunId, + }); + return; + } + rawEntries = result.entries; + liveStreamLineCount = rawEntries.length; } + } else { + const parsed = await this.fetchSessionLogs(logUrl, taskRunId); + rawEntries = parsed.rawEntries; + liveStreamLineCount = parsed.totalLineCount; + } - const session = this.d.store.getSessionByTaskId(taskId); - if (!session || session.taskRunId !== taskRunId) { - return; - } + const session = this.d.store.getSessionByTaskId(taskId); + if (!session || session.taskRunId !== taskRunId) { + return; + } - const events = convertStoredEntriesToEvents(rawEntries); - const hasUserPrompt = events.some( - (e: AcpMessage) => - isJsonRpcRequest(e.message) && e.message.method === "session/prompt", + let events = convertStoredEntriesToEvents(rawEntries, undefined, { + taskRunId, + startEntryIndex: 0, + firstPositionedEntryIndex: resumeLeafEntryStartIndex, + }); + if (isResumeRun && session.events.length > 0) { + const inheritedEvents = reconcileLiveEventsWithHydratedEvents( + session.events, + events, ); - - // Seed the optimistic user-message bubble whenever the agent has - // not yet recorded an initial `session/prompt` request — covers the - // brand-new task case as well as "agent has emitted lifecycle - // notifications but hasn't received its first prompt yet". Prefer the - // stashed initial prompt (which carries the channel CONTEXT.md block, so - // its chip renders right away) over the bare task description. - const seedContent = - this.initialCloudOptimisticPrompt.get(taskId) ?? taskDescription; - if (!hasUserPrompt && seedContent?.trim()) { - this.d.store.appendOptimisticItem(taskRunId, { - type: "user_message", - content: seedContent, - timestamp: Date.now(), - }); - } - if (hasUserPrompt) { - // The real prompt has landed; the stash is no longer needed. - this.initialCloudOptimisticPrompt.delete(taskId); - this.d.store.clearTailOptimisticItems(taskRunId); + events = [...events, ...inheritedEvents]; + const watcher = this.cloudTaskWatchers.get(taskId); + const hasLeafLocalWatcherCursor = + watcher?.runId === taskRunId && + watcher.resumeHistoryCountOffset !== undefined; + if (hasLeafLocalWatcherCursor) { + liveStreamLineCount = Math.max( + liveStreamLineCount, + session.processedLineCount ?? 0, + ); } + } + const hasUserPrompt = events.some( + (e: AcpMessage) => + isJsonRpcRequest(e.message) && e.message.method === "session/prompt", + ); - if (rawEntries.length === 0) { - this.pendingPermissionHydratedRuns.add(taskRunId); - return; - } + // Seed the optimistic user-message bubble whenever the agent has + // not yet recorded an initial `session/prompt` request — covers the + // brand-new task case as well as "agent has emitted lifecycle + // notifications but hasn't received its first prompt yet". Prefer the + // stashed initial prompt (which carries the channel CONTEXT.md block, so + // its chip renders right away) over the bare task description. + const seedContent = + this.initialCloudOptimisticPrompt.get(taskId) ?? taskDescription; + if (!hasUserPrompt && seedContent?.trim()) { + this.d.store.appendOptimisticItem(taskRunId, { + type: "user_message", + content: seedContent, + timestamp: Date.now(), + }); + } + if (hasUserPrompt) { + // The real prompt has landed; the stash is no longer needed. + this.initialCloudOptimisticPrompt.delete(taskId); + this.d.store.clearTailOptimisticItems(taskRunId); + } - // If live updates already populated a processed count, don't overwrite - // that newer state with the persisted baseline fetched during startup. - if ( - session.processedLineCount !== undefined && - session.processedLineCount > 0 - ) { - this.surfacePersistedPendingPermissions(taskRunId, rawEntries); - this.pendingPermissionHydratedRuns.add(taskRunId); - return; - } + if (rawEntries.length === 0) { + this.pendingPermissionHydratedRuns.add(taskRunId); + return { + historyEntryCount: 0, + liveStreamLineCount, + }; + } - this.d.store.updateSession(taskRunId, { - events, - isCloud: true, - logUrl: logUrl ?? session.logUrl, - processedLineCount: totalLineCount, - }); + // If live updates already populated a processed count, don't overwrite + // that newer state with the persisted baseline fetched during startup. + if ( + session.processedLineCount !== undefined && + session.processedLineCount > 0 && + !isResumeRun + ) { this.surfacePersistedPendingPermissions(taskRunId, rawEntries); this.pendingPermissionHydratedRuns.add(taskRunId); - // Without this the "Galumphing…" indicator stays hidden when the hydrated - // baseline already contains an in-flight session/prompt — the live delta - // path otherwise sees delta <= 0 and never re-evaluates the tail. - this.updatePromptStateFromEvents(taskRunId, events); - })().catch((err: unknown) => { - this.d.log.warn("Failed to hydrate cloud task session from logs", { - taskId, - taskRunId, - err, - }); + return { + historyEntryCount: rawEntries.length, + liveStreamLineCount: session.processedLineCount, + }; + } + + this.d.store.updateSession(taskRunId, { + events, + isCloud: true, + logUrl: logUrl ?? session.logUrl, + cloudTranscriptEntryCount: rawEntries.length, + processedLineCount: liveStreamLineCount, }); + this.surfacePersistedPendingPermissions(taskRunId, rawEntries); + this.pendingPermissionHydratedRuns.add(taskRunId); + // Without this the "Galumphing…" indicator stays hidden when the hydrated + // baseline already contains an in-flight session/prompt — the live delta + // path otherwise sees delta <= 0 and never re-evaluates the tail. + this.updatePromptStateFromEvents(taskRunId, events); + return { + historyEntryCount: rawEntries.length, + liveStreamLineCount, + }; } private isCurrentCloudTaskWatcher( @@ -6683,7 +6961,14 @@ export class SessionService { // Already caught up — skip duplicate entries } else if (plan.kind === "append-tail") { const entriesToAppend = update.newEntries.slice(-plan.tailCount); - const newEvents = convertStoredEntriesToEvents(entriesToAppend); + const newEvents = convertStoredEntriesToEvents( + entriesToAppend, + undefined, + { + taskRunId, + startEntryIndex: expectedCount - entriesToAppend.length, + }, + ); if (hasSessionPromptEvent(newEvents)) { this.d.store.clearTailOptimisticItems(taskRunId); } @@ -6962,7 +7247,10 @@ export class SessionService { logUrl: string | undefined, processedLineCount: number, ): void { - const events = convertStoredEntriesToEvents(rawEntries); + const events = convertStoredEntriesToEvents(rawEntries, undefined, { + taskRunId, + startEntryIndex: 0, + }); if (hasSessionPromptEvent(events)) { this.d.store.clearTailOptimisticItems(taskRunId); } diff --git a/packages/ui/src/features/sessions/sessionServiceHost.test.ts b/packages/ui/src/features/sessions/sessionServiceHost.test.ts index 0c84840a5e..791269d1ac 100644 --- a/packages/ui/src/features/sessions/sessionServiceHost.test.ts +++ b/packages/ui/src/features/sessions/sessionServiceHost.test.ts @@ -135,6 +135,7 @@ const mockAuthenticatedClient = vi.hoisted(() => ({ finalizeTaskStagedArtifactUploads: vi.fn(), startGithubUserIntegrationConnect: vi.fn(), getTaskRunSessionLogs: vi.fn(), + getTaskRunSessionLogsResult: vi.fn(), })); type MockAuthenticatedClient = typeof mockAuthenticatedClient; @@ -346,7 +347,13 @@ vi.mock("@posthog/shared", async (importOriginal) => ({ ), })); const mockConvertStoredEntriesToEvents = vi.hoisted(() => - vi.fn<(entries: unknown[]) => unknown[]>(() => []), + vi.fn< + ( + entries: unknown[], + taskDescription?: string, + positionOptions?: unknown, + ) => unknown[] + >(() => []), ); vi.mock("@posthog/core/sessions/sessionEvents", async () => { @@ -376,6 +383,7 @@ vi.mock("@posthog/core/sessions/sessionEvents", async () => { message: {}, })), extractPromptText: vi.fn((p) => (typeof p === "string" ? p : "text")), + getStoredLogEventPosition: actual.getStoredLogEventPosition, getUserShellExecutesSinceLastPrompt: vi.fn(() => []), hasSessionPromptEvent: actual.hasSessionPromptEvent, isAbsoluteFolderPath: actual.isAbsoluteFolderPath, @@ -449,6 +457,10 @@ describe("SessionService", () => { mockAuthenticatedClient.getTaskRunSessionLogs.mockResolvedValue([]); mockSessionConfigStore.getPersistedConfigOptions.mockReturnValue(undefined); mockAdapterFns.getAdapter.mockReturnValue(undefined); + mockAuthenticatedClient.getTaskRunSessionLogsResult.mockResolvedValue({ + entries: [], + complete: true, + }); mockSessionStoreSetters.getSessionByTaskId.mockReturnValue(undefined); mockSessionStoreSetters.getSessions.mockReturnValue({}); mockAuth.fetchAuthState.mockResolvedValue({ @@ -3463,60 +3475,63 @@ describe("SessionService", () => { mockSessionStoreSetters.getSessions.mockReturnValue({ "run-123": completedSession, }); - mockAuthenticatedClient.getTaskRunSessionLogs.mockResolvedValue([ - { - type: "notification", - notification: { - method: "_posthog/sdk_session", - params: { - taskRunId: "run-123", - sessionId: "acp-session-1", - adapter: "claude", + mockAuthenticatedClient.getTaskRunSessionLogsResult.mockResolvedValue({ + complete: true, + entries: [ + { + type: "notification", + notification: { + method: "_posthog/sdk_session", + params: { + taskRunId: "run-123", + sessionId: "acp-session-1", + adapter: "claude", + }, }, }, - }, - { - type: "notification", - notification: { - method: "_posthog/run_started", - params: { - sessionId: "acp-session-1", - runId: "run-123", - taskId: "task-123", + { + type: "notification", + notification: { + method: "_posthog/run_started", + params: { + sessionId: "acp-session-1", + runId: "run-123", + taskId: "task-123", + }, }, }, - }, - { - type: "notification", - notification: { - method: "_posthog/permission_request", - params: { - requestId: "request-1", - toolCall: { - toolCallId: "tool-1", - title: "What animal do you prefer?", - kind: "other", - _meta: { - codeToolKind: "question", - questions: [ - { - question: "What animal do you prefer?", - options: [ - { label: "cats", description: "Cats" }, - { label: "dogs", description: "Dogs" }, - ], - }, - ], + { + type: "notification", + notification: { + method: "_posthog/permission_request", + params: { + requestId: "request-1", + toolCall: { + toolCallId: "tool-1", + title: "What animal do you prefer?", + kind: "other", + _meta: { + codeToolKind: "question", + questions: [ + { + question: "What animal do you prefer?", + options: [ + { label: "cats", description: "Cats" }, + { label: "dogs", description: "Dogs" }, + ], + }, + ], + }, }, + options: [ + { optionId: "option_0", name: "cats", kind: "allow_once" }, + { optionId: "option_1", name: "dogs", kind: "allow_once" }, + ], }, - options: [ - { optionId: "option_0", name: "cats", kind: "allow_once" }, - { optionId: "option_1", name: "dogs", kind: "allow_once" }, - ], }, }, - }, - ]); + ], + }); service.watchCloudTask( "task-123", @@ -3626,23 +3641,258 @@ describe("SessionService", () => { ).not.toHaveBeenCalled(); }); - it("hydrates an in-progress resumed run from the full session-log chain", async () => { + it.each([ + { name: "leaf-only response", responseShape: "leaf" }, + { name: "full-chain response", responseShape: "full" }, + { name: "overlapping chain window", responseShape: "overlap" }, + ])( + "hydrates an in-progress resumed run from a $name", + async ({ responseShape }) => { + const service = getSessionService(); + const priorPrompt = { + type: "acp_message" as const, + ts: 1700000000, + message: { + jsonrpc: "2.0" as const, + id: 1, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "first request" }] }, + }, + }; + const resumePrompt = { + type: "acp_message" as const, + ts: 1700000060, + message: { + jsonrpc: "2.0" as const, + id: 2, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "continue" }] }, + }, + }; + const resumeCompletion = { + type: "acp_message" as const, + ts: 1700000120, + message: { + jsonrpc: "2.0" as const, + method: "_posthog/turn_complete", + params: { sessionId: "session-1", stopReason: "end_turn" }, + }, + }; + const resumedSession = createMockSession({ + taskRunId: "run-456", + taskId: "task-123", + status: "disconnected", + isCloud: true, + events: [resumePrompt], + processedLineCount: 1, + optimisticItems: [ + { + id: "optimistic-follow-up", + type: "user_message", + content: "continue", + timestamp: 1700000001, + pinToTop: false, + }, + ], + }); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue( + resumedSession, + ); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-456": resumedSession, + }); + const parentEntries = [ + { timestamp: "2024-01-01T00:00:00Z", notification: {} }, + { timestamp: "2024-01-01T00:00:30Z", notification: {} }, + ]; + const leafEntries = [ + { timestamp: "2024-01-01T00:01:00Z", notification: {} }, + ]; + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockResolvedValueOnce({ entries: parentEntries, complete: true }) + .mockResolvedValueOnce({ + entries: + responseShape === "full" + ? [...parentEntries, ...leafEntries] + : responseShape === "overlap" + ? [parentEntries[1], ...leafEntries] + : leafEntries, + complete: true, + }); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue( + JSON.stringify(leafEntries[0]), + ); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); + mockConvertStoredEntriesToEvents.mockReturnValueOnce([ + priorPrompt, + resumePrompt, + resumeCompletion, + ]); + + service.watchCloudTask( + "task-123", + "run-456", + "https://api.anthropic.com", + 123, + undefined, + "https://logs.example.com/run-456", + undefined, + "claude", + undefined, + "first request", + undefined, + "in_progress", + undefined, + { resume_from_run_id: "run-123" }, + ); + + const subscribeOptions = mockTrpcCloudTask.onUpdate.subscribe.mock + .calls[0][1] as { onData: (update: unknown) => void }; + subscribeOptions.onData({ + kind: "snapshot", + taskId: "task-123", + runId: "run-456", + totalEntryCount: 3, + newEntries: [...parentEntries, ...leafEntries], + status: "in_progress", + }); + expect(mockSessionStoreSetters.appendEvents).not.toHaveBeenCalled(); + + expect(mockTrpcCloudTask.watch.mutate).toHaveBeenCalledWith({ + taskId: "task-123", + runId: "run-456", + apiHost: "https://api.anthropic.com", + teamId: 123, + resumeFromEntryCount: undefined, + }); + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledWith("task-123", "run-123", { limit: 100000 }); + }); + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledWith("task-123", "run-456", { limit: 100000 }); + expect(mockConvertStoredEntriesToEvents).toHaveBeenCalledWith( + [...parentEntries, ...leafEntries], + undefined, + { + taskRunId: "run-456", + startEntryIndex: 0, + firstPositionedEntryIndex: parentEntries.length, + }, + ); + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-456", + expect.objectContaining({ + events: [priorPrompt, resumePrompt, resumeCompletion], + processedLineCount: 1, + }), + ); + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-456", + expect.objectContaining({ + isPromptPending: false, + promptStartedAt: null, + currentPromptId: null, + }), + ); + expect( + mockSessionStoreSetters.clearTailOptimisticItems, + ).toHaveBeenCalledWith("run-456"); + expect( + mockSessionStoreSetters.appendOptimisticItem, + ).not.toHaveBeenCalled(); + expect(mockSessionStoreSetters.appendEvents).not.toHaveBeenCalled(); + }, + ); + + it("reconciles repeated prompt occurrences and promptless live tails", async () => { const service = getSessionService(); + const ancestorPrompt = { + type: "acp_message" as const, + ts: 1700000010, + message: { + jsonrpc: "2.0" as const, + id: 1, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "repeat request" }] }, + }, + }; + const currentLivePrompt = { + type: "acp_message" as const, + ts: 1700000040, + message: ancestorPrompt.message, + }; + const persistedCurrentPrompt = { + ...currentLivePrompt, + ts: 1700000041, + }; + const persistedAncestorMessage = { + type: "acp_message" as const, + ts: 1700000020, + message: { + jsonrpc: "2.0" as const, + method: "session/update", + params: { + update: { + sessionUpdate: "agent_message", + content: { type: "text", text: "ancestor complete" }, + }, + }, + }, + }; + const currentLiveChunk = { + type: "acp_message" as const, + ts: 1700000050, + message: { + jsonrpc: "2.0" as const, + method: "session/update", + params: { + sessionId: "current-session", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "current partial" }, + }, + }, + }, + }; + const ancestorCompletion = { + type: "acp_message" as const, + ts: 1700000030, + message: { + jsonrpc: "2.0" as const, + method: "_posthog/turn_complete", + params: { stopReason: "end_turn" }, + }, + }; + const currentCompletion = { + ...ancestorCompletion, + ts: 1700000060, + }; + const promptlessLiveOnlyEvent = { + type: "acp_message" as const, + ts: 1700000035, + message: { + jsonrpc: "2.0" as const, + method: "_posthog/usage_update", + params: { used: 42 }, + }, + }; const resumedSession = createMockSession({ taskRunId: "run-456", taskId: "task-123", - status: "disconnected", + status: "connected", isCloud: true, - events: [], - optimisticItems: [ - { - id: "optimistic-follow-up", - type: "user_message", - content: "continue", - timestamp: 1700000001, - pinToTop: false, - }, + events: [ + promptlessLiveOnlyEvent, + persistedAncestorMessage, + ancestorCompletion, + currentLivePrompt, + currentLiveChunk, + currentCompletion, ], + processedLineCount: 1, }); mockSessionStoreSetters.getSessionByTaskId.mockReturnValue( resumedSession, @@ -3650,43 +3900,119 @@ describe("SessionService", () => { mockSessionStoreSetters.getSessions.mockReturnValue({ "run-456": resumedSession, }); - const chainedEntries = [ - { timestamp: "2024-01-01T00:00:00Z", notification: {} }, - { timestamp: "2024-01-01T00:01:00Z", notification: {} }, - ]; - mockAuthenticatedClient.getTaskRunSessionLogs.mockResolvedValue( - chainedEntries, - ); - mockTrpcLogs.readLocalLogs.query.mockResolvedValue( - "leaf log should not be used", - ); - mockTrpcLogs.fetchS3Logs.query.mockResolvedValue( - "leaf s3 log should not be used", + const parentEntry = { + timestamp: "2024-01-01T00:01:00Z", + notification: {}, + }; + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockResolvedValueOnce({ entries: [parentEntry], complete: true }) + .mockResolvedValueOnce({ entries: [parentEntry], complete: true }); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue(""); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); + mockConvertStoredEntriesToEvents.mockReturnValueOnce([ + ancestorPrompt, + persistedAncestorMessage, + ancestorCompletion, + persistedCurrentPrompt, + ]); + + service.watchCloudTask( + "task-123", + "run-456", + "https://api.anthropic.com", + 123, + undefined, + "https://logs.example.com/run-456", + undefined, + "claude", + undefined, + undefined, + undefined, + "in_progress", + undefined, + { resume_from_run_id: "run-123" }, ); - const priorPrompt = { + await vi.waitFor(() => { + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-456", + expect.objectContaining({ + events: [ + ancestorPrompt, + persistedAncestorMessage, + ancestorCompletion, + persistedCurrentPrompt, + promptlessLiveOnlyEvent, + currentLiveChunk, + currentCompletion, + ], + }), + ); + }); + }); + + it("preserves a promptless current completion that only matches an ancestor turn", async () => { + const service = getSessionService(); + const ancestorPrompt = { type: "acp_message" as const, - ts: 1700000000, + ts: 1700000010, message: { jsonrpc: "2.0" as const, id: 1, method: "session/prompt", - params: { prompt: [{ type: "text", text: "first request" }] }, + params: { prompt: [{ type: "text", text: "ancestor request" }] }, }, }; - const resumePrompt = { + const currentPrompt = { type: "acp_message" as const, - ts: 1700000060, + ts: 1700000040, message: { jsonrpc: "2.0" as const, id: 2, method: "session/prompt", - params: { prompt: [{ type: "text", text: "continue" }] }, + params: { prompt: [{ type: "text", text: "current request" }] }, }, }; + const ancestorCompletion = { + type: "acp_message" as const, + ts: 1700000030, + message: { + jsonrpc: "2.0" as const, + method: "_posthog/turn_complete", + params: { stopReason: "end_turn" }, + }, + }; + const currentCompletion = { + ...ancestorCompletion, + ts: 1700000060, + }; + const resumedSession = createMockSession({ + taskRunId: "run-456", + taskId: "task-123", + status: "connected", + isCloud: true, + events: [currentCompletion], + processedLineCount: 1, + }); + mockSessionStoreSetters.getSessionByTaskId.mockReturnValue( + resumedSession, + ); + mockSessionStoreSetters.getSessions.mockReturnValue({ + "run-456": resumedSession, + }); + const parentEntry = { + timestamp: "2024-01-01T00:01:00Z", + notification: {}, + }; + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockResolvedValueOnce({ entries: [parentEntry], complete: true }) + .mockResolvedValueOnce({ entries: [parentEntry], complete: true }); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue(""); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); mockConvertStoredEntriesToEvents.mockReturnValueOnce([ - priorPrompt, - resumePrompt, + ancestorPrompt, + ancestorCompletion, + currentPrompt, ]); service.watchCloudTask( @@ -3699,7 +4025,7 @@ describe("SessionService", () => { undefined, "claude", undefined, - "first request", + undefined, undefined, "in_progress", undefined, @@ -3707,26 +4033,582 @@ describe("SessionService", () => { ); await vi.waitFor(() => { - expect( - mockAuthenticatedClient.getTaskRunSessionLogs, - ).toHaveBeenCalledWith("task-123", "run-456", { limit: 100000 }); + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-456", + expect.objectContaining({ + events: [ + ancestorPrompt, + ancestorCompletion, + currentPrompt, + currentCompletion, + ], + }), + ); }); - expect(mockConvertStoredEntriesToEvents).toHaveBeenCalledWith( - chainedEntries, - ); - expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( - "run-456", - expect.objectContaining({ - events: [priorPrompt, resumePrompt], - processedLineCount: chainedEntries.length, - }), - ); - expect( - mockSessionStoreSetters.clearTailOptimisticItems, - ).toHaveBeenCalledWith("run-456"); - expect( - mockSessionStoreSetters.appendOptimisticItem, - ).not.toHaveBeenCalled(); + }); + + it("keeps immediate-resume watcher counts leaf-local while flushing buffered updates", async () => { + const service = getSessionService(); + const ancestorEvent = { + type: "acp_message" as const, + ts: 1700000000, + message: { + jsonrpc: "2.0" as const, + id: 1, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "first request" }] }, + }, + }; + const leafEvent = { + type: "acp_message" as const, + ts: 1700000060, + message: { + jsonrpc: "2.0" as const, + id: 2, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "continue" }] }, + }, + }; + const liveEvent = { + type: "acp_message" as const, + ts: 1700000120, + message: { + jsonrpc: "2.0" as const, + method: "session/update", + params: { update: { sessionUpdate: "agent_message_chunk" } }, + }, + }; + const resumedSession = createMockSession({ + taskRunId: "run-456", + taskId: "task-123", + status: "connected", + isCloud: true, + events: [ancestorEvent], + cloudTranscriptEntryCount: 3, + processedLineCount: 0, + }); + mockSessionStoreSetters.getSessionByTaskId.mockImplementation( + () => resumedSession, + ); + mockSessionStoreSetters.getSessions.mockImplementation(() => ({ + "run-456": resumedSession, + })); + mockSessionStoreSetters.updateSession.mockImplementation( + (_runId, updates) => Object.assign(resumedSession, updates), + ); + mockSessionStoreSetters.appendEvents.mockImplementation( + (_runId, events, processedLineCount) => { + resumedSession.events.push(...events); + if (processedLineCount !== undefined) { + resumedSession.processedLineCount = processedLineCount; + } + }, + ); + + const ancestorEntries = [ + { timestamp: "2024-01-01T00:00:00Z", notification: {} }, + { timestamp: "2024-01-01T00:00:01Z", notification: {} }, + { timestamp: "2024-01-01T00:00:02Z", notification: {} }, + ]; + const leafEntry = { + timestamp: "2024-01-01T00:01:00Z", + notification: {}, + }; + const liveEntry = { + timestamp: "2024-01-01T00:02:00Z", + notification: { method: "session/update" }, + }; + let resolveAncestor!: (result: { + entries: typeof ancestorEntries; + complete: boolean; + }) => void; + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveAncestor = resolve; + }), + ) + .mockResolvedValueOnce({ + entries: [...ancestorEntries, leafEntry], + complete: true, + }); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue( + JSON.stringify(leafEntry), + ); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); + mockConvertStoredEntriesToEvents + .mockReturnValueOnce([ancestorEvent, leafEvent]) + .mockReturnValueOnce([liveEvent]); + + service.watchCloudTask( + "task-123", + "run-456", + "https://api.anthropic.com", + 123, + undefined, + "https://logs.example.com/run-456", + undefined, + "claude", + undefined, + "first request", + 3, + "in_progress", + undefined, + { resume_from_run_id: "run-123" }, + ); + + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledTimes(2); + }); + + const subscribeOptions = mockTrpcCloudTask.onUpdate.subscribe.mock + .calls[0][1] as { onData: (update: unknown) => void }; + subscribeOptions.onData({ + kind: "logs", + taskId: "task-123", + runId: "run-456", + totalEntryCount: 5, + newEntries: [leafEntry, liveEntry], + }); + expect(mockSessionStoreSetters.appendEvents).not.toHaveBeenCalled(); + + resolveAncestor({ entries: ancestorEntries, complete: true }); + await vi.waitFor(() => { + expect(mockSessionStoreSetters.appendEvents).toHaveBeenCalledWith( + "run-456", + [liveEvent], + 2, + ); + }); + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-456", + expect.objectContaining({ + events: expect.arrayContaining([ancestorEvent, leafEvent]), + processedLineCount: 1, + }), + ); + expect(resumedSession.events).toEqual([ + ancestorEvent, + leafEvent, + liveEvent, + ]); + expect(resumedSession.processedLineCount).toBe(2); + expect(mockTrpcCloudTask.watch.mutate).toHaveBeenCalledWith({ + taskId: "task-123", + runId: "run-456", + apiHost: "https://api.anthropic.com", + teamId: 123, + resumeFromEntryCount: 3, + }); + }); + + it("uses the full A→B transcript count when B resumes into C", async () => { + const service = getSessionService(); + const aEvent = { + type: "acp_message" as const, + ts: 1700000000, + message: { + jsonrpc: "2.0" as const, + id: 1, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "start A" }] }, + }, + }; + const bEvent = { + type: "acp_message" as const, + ts: 1700000060, + message: { + jsonrpc: "2.0" as const, + id: 2, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "resume B" }] }, + }, + }; + const cEvent = { + type: "acp_message" as const, + ts: 1700000120, + message: { + jsonrpc: "2.0" as const, + id: 3, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "resume C" }] }, + }, + }; + const liveCEvent = { + type: "acp_message" as const, + ts: 1700000180, + message: { + jsonrpc: "2.0" as const, + method: "session/update", + params: { update: { sessionUpdate: "agent_message_chunk" } }, + }, + }; + let activeSession = createMockSession({ + taskRunId: "run-b", + taskId: "task-123", + status: "connected", + isCloud: true, + cloudStatus: "completed", + cloudBranch: "feature/resume-chain", + events: [aEvent, bEvent], + cloudTranscriptEntryCount: 7, + processedLineCount: 2, + }); + mockSessionStoreSetters.getSessionByTaskId.mockImplementation( + () => activeSession, + ); + mockSessionStoreSetters.getSessions.mockImplementation(() => ({ + [activeSession.taskRunId]: activeSession, + })); + mockSessionStoreSetters.setSession.mockImplementation((session) => { + activeSession = session; + }); + mockSessionStoreSetters.updateSession.mockImplementation( + (_runId, updates) => Object.assign(activeSession, updates), + ); + mockSessionStoreSetters.appendEvents.mockImplementation( + (_runId, events, processedLineCount) => { + activeSession.events.push(...events); + if (processedLineCount !== undefined) { + activeSession.processedLineCount = processedLineCount; + } + }, + ); + + mockAuthenticatedClient.getTaskRun.mockResolvedValue({ + id: "run-b", + task: "task-123", + team: 123, + branch: "feature/resume-chain", + runtime_adapter: "claude", + model: "claude-sonnet-4-20250514", + reasoning_effort: null, + environment: "cloud", + status: "completed", + log_url: "https://example.com/logs/run-b", + error_message: null, + output: {}, + state: { resume_from_run_id: "run-a" }, + created_at: "2026-04-14T00:00:00Z", + updated_at: "2026-04-14T00:05:00Z", + completed_at: "2026-04-14T00:05:00Z", + }); + mockAuthenticatedClient.getTask.mockResolvedValue(createMockTask()); + mockAuthenticatedClient.runTaskInCloud.mockResolvedValue( + createMockTask({ + latest_run: { + id: "run-c", + task: "task-123", + team: 123, + branch: "feature/resume-chain", + runtime_adapter: "claude", + model: "claude-sonnet-4-20250514", + reasoning_effort: null, + environment: "cloud", + status: "queued", + log_url: "https://example.com/logs/run-c", + error_message: null, + output: {}, + state: { resume_from_run_id: "run-b" }, + created_at: "2026-04-14T00:06:00Z", + updated_at: "2026-04-14T00:06:00Z", + completed_at: null, + }, + }), + ); + + const inheritedEntries = Array.from({ length: 7 }, (_, index) => ({ + timestamp: `2024-01-01T00:00:0${index}Z`, + notification: {}, + })); + const cEntry = { + timestamp: "2024-01-01T00:02:00Z", + notification: {}, + }; + const liveCEntry = { + timestamp: "2024-01-01T00:03:00Z", + notification: { method: "session/update" }, + }; + let resolveInherited!: (result: { + entries: typeof inheritedEntries; + complete: boolean; + }) => void; + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveInherited = resolve; + }), + ) + .mockResolvedValueOnce({ + entries: [...inheritedEntries, cEntry], + complete: true, + }); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue( + JSON.stringify(cEntry), + ); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); + mockConvertStoredEntriesToEvents + .mockReturnValueOnce([aEvent, bEvent, cEvent]) + .mockReturnValueOnce([liveCEvent]); + + const result = await service.sendPrompt("task-123", "resume C"); + expect(result.stopReason).toBe("queued"); + expect(activeSession).toEqual( + expect.objectContaining({ + taskRunId: "run-c", + cloudTranscriptEntryCount: 7, + processedLineCount: 0, + }), + ); + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledTimes(2); + }); + + const subscribeOptions = mockTrpcCloudTask.onUpdate.subscribe.mock + .calls[0][1] as { onData: (update: unknown) => void }; + subscribeOptions.onData({ + kind: "logs", + taskId: "task-123", + runId: "run-c", + totalEntryCount: 9, + newEntries: [cEntry, liveCEntry], + }); + expect(mockSessionStoreSetters.appendEvents).not.toHaveBeenCalled(); + + resolveInherited({ entries: inheritedEntries, complete: true }); + await vi.waitFor(() => { + expect(mockSessionStoreSetters.appendEvents).toHaveBeenCalledWith( + "run-c", + [liveCEvent], + 2, + ); + }); + expect(activeSession.events).toEqual([ + aEvent, + bEvent, + cEvent, + liveCEvent, + ]); + expect(activeSession.processedLineCount).toBe(2); + expect(activeSession.cloudTranscriptEntryCount).toBe(9); + expect(mockTrpcCloudTask.watch.mutate).toHaveBeenCalledWith({ + taskId: "task-123", + runId: "run-c", + apiHost: "https://api.anthropic.com", + teamId: 123, + resumeFromEntryCount: 7, + }); + }); + + it("switches a cold-reload watcher to leaf-local counts after hydration recovers", async () => { + const service = getSessionService(); + const resumePrompt = { + type: "acp_message" as const, + ts: 1700000060, + message: { + jsonrpc: "2.0" as const, + id: 2, + method: "session/prompt", + params: { prompt: [{ type: "text", text: "continue" }] }, + }, + }; + const resumedSession = createMockSession({ + taskRunId: "run-456", + taskId: "task-123", + status: "connected", + isCloud: true, + events: [], + processedLineCount: 0, + }); + mockSessionStoreSetters.getSessionByTaskId.mockImplementation( + () => resumedSession, + ); + mockSessionStoreSetters.getSessions.mockImplementation(() => ({ + "run-456": resumedSession, + })); + mockSessionStoreSetters.updateSession.mockImplementation( + (_runId, updates) => + Object.assign(resumedSession, { + ...updates, + ...(updates.events ? { events: [...updates.events] } : {}), + }), + ); + mockSessionStoreSetters.appendEvents.mockImplementation( + (_runId, events, processedLineCount) => { + resumedSession.events.push(...events); + if (processedLineCount !== undefined) { + resumedSession.processedLineCount = processedLineCount; + } + }, + ); + const parentEntries = Array.from({ length: 7 }, (_, index) => ({ + timestamp: `2024-01-01T00:00:0${index}Z`, + notification: {}, + })); + const leafEntry = { + timestamp: "2024-01-01T00:01:00Z", + notification: {}, + }; + const liveEntry = { + timestamp: "2024-01-01T00:02:00Z", + notification: { method: "session/update" }, + }; + const liveEvent = { + type: "acp_message" as const, + ts: 1700000120, + message: { + jsonrpc: "2.0" as const, + method: "session/update", + params: { update: { sessionUpdate: "agent_message_chunk" } }, + }, + }; + let resolveAncestor!: (result: { + entries: typeof parentEntries; + complete: boolean; + }) => void; + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveAncestor = resolve; + }), + ) + .mockResolvedValueOnce({ + entries: [...parentEntries, leafEntry], + complete: true, + }); + mockConvertStoredEntriesToEvents.mockImplementation((entries) => + entries.some( + (entry) => + (entry as { timestamp?: string }).timestamp === liveEntry.timestamp, + ) + ? [liveEvent] + : [resumePrompt], + ); + + const watch = (): void => { + service.watchCloudTask( + "task-123", + "run-456", + "https://api.anthropic.com", + 123, + undefined, + "https://logs.example.com/run-456", + undefined, + "claude", + undefined, + "first request", + undefined, + "in_progress", + undefined, + { resume_from_run_id: "run-123" }, + ); + }; + + watch(); + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledTimes(2); + }); + expect(mockTrpcCloudTask.watch.mutate).toHaveBeenCalledWith({ + taskId: "task-123", + runId: "run-456", + apiHost: "https://api.anthropic.com", + teamId: 123, + resumeFromEntryCount: undefined, + }); + watch(); + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledTimes(2); + + const subscribeOptions = mockTrpcCloudTask.onUpdate.subscribe.mock + .calls[0][1] as { onData: (update: unknown) => void }; + subscribeOptions.onData({ + kind: "logs", + taskId: "task-123", + runId: "run-456", + totalEntryCount: 8, + newEntries: [...parentEntries, leafEntry], + }); + expect(mockSessionStoreSetters.appendEvents).not.toHaveBeenCalled(); + + resolveAncestor({ entries: parentEntries, complete: false }); + await vi.waitFor(() => { + expect(mockSessionStoreSetters.appendEvents).toHaveBeenCalledWith( + "run-456", + [resumePrompt], + 8, + ); + }); + expect(resumedSession.processedLineCount).toBe(8); + + let resolveRetryAncestor!: (result: { + entries: typeof parentEntries; + complete: boolean; + }) => void; + mockAuthenticatedClient.getTaskRunSessionLogsResult + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveRetryAncestor = resolve; + }), + ) + .mockResolvedValueOnce({ + entries: [...parentEntries, leafEntry], + complete: true, + }); + mockTrpcLogs.readLocalLogs.query.mockResolvedValue( + JSON.stringify(leafEntry), + ); + mockTrpcLogs.fetchS3Logs.query.mockResolvedValue(""); + + watch(); + await vi.waitFor(() => { + expect( + mockAuthenticatedClient.getTaskRunSessionLogsResult, + ).toHaveBeenCalledTimes(4); + }); + const appendCountBeforeRetryUpdate = + mockSessionStoreSetters.appendEvents.mock.calls.length; + subscribeOptions.onData({ + kind: "logs", + taskId: "task-123", + runId: "run-456", + totalEntryCount: 9, + newEntries: [liveEntry], + }); + expect(mockSessionStoreSetters.appendEvents).toHaveBeenCalledTimes( + appendCountBeforeRetryUpdate, + ); + + resolveRetryAncestor({ entries: parentEntries, complete: true }); + await vi.waitFor(() => { + expect(mockSessionStoreSetters.updateSession).toHaveBeenCalledWith( + "run-456", + expect.objectContaining({ + events: [resumePrompt], + processedLineCount: 1, + }), + ); + }); + await vi.waitFor(() => { + expect(mockSessionStoreSetters.appendEvents).toHaveBeenLastCalledWith( + "run-456", + [liveEvent], + 2, + ); + }); + expect(resumedSession.events).toEqual([resumePrompt, liveEvent]); + expect(resumedSession.processedLineCount).toBe(2); + expect(resumedSession.cloudTranscriptEntryCount).toBe(9); }); it("ignores stale async starts when the same watcher is replaced", async () => {