From 86cda4cc828804117849324e0bb194da3621739e Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Wed, 24 Jun 2026 13:12:26 +0100 Subject: [PATCH 1/2] feat(agent): natively resume prior session on cloud resume Cloud task resumes restored conversation context by flattening the prior log into a summary blob and replaying it as a synthetic prompt. Instead, recover the prior Claude session id from the run_started notification, hydrate its JSONL (warm from a filesystem snapshot, or rebuilt from the log), and resumeSession so the SDK continues the real conversation. The new user message is then a plain continuation. The summary path remains as a last-resort fallback. Reuse the already-loaded resume conversation when hydrating the JSONL so the prior run's log is not fetched twice, and share the post-prompt turn handling between the native and summary resume paths. --- .../claude/session/jsonl-hydration.ts | 65 ++-- packages/agent/src/resume.ts | 2 + packages/agent/src/sagas/resume-saga.test.ts | 64 ++++ packages/agent/src/sagas/resume-saga.ts | 24 ++ packages/agent/src/server/agent-server.ts | 299 ++++++++++++++---- 5 files changed, 359 insertions(+), 95 deletions(-) diff --git a/packages/agent/src/adapters/claude/session/jsonl-hydration.ts b/packages/agent/src/adapters/claude/session/jsonl-hydration.ts index 14b39a71b6..a4548d5d66 100644 --- a/packages/agent/src/adapters/claude/session/jsonl-hydration.ts +++ b/packages/agent/src/adapters/claude/session/jsonl-hydration.ts @@ -497,6 +497,7 @@ export async function hydrateSessionJsonl(params: { permissionMode?: string; posthogAPI: PostHogAPIClient; log: HydrationLog; + conversation?: ConversationTurn[]; }): Promise { const { posthogAPI, log } = params; @@ -509,40 +510,46 @@ export async function hydrateSessionJsonl(params: { // File doesn't exist, proceed with hydration } - const taskRun = await posthogAPI.getTaskRun(params.taskId, params.runId); - if (!taskRun.log_url) { - log.info("No log URL, skipping JSONL hydration"); - return false; - } + let allTurns: ConversationTurn[]; + if (params.conversation && params.conversation.length > 0) { + allTurns = params.conversation; + } else { + const taskRun = await posthogAPI.getTaskRun(params.taskId, params.runId); + if (!taskRun.log_url) { + log.info("No log URL, skipping JSONL hydration"); + return false; + } - const entries = await posthogAPI.fetchTaskRunLogs(taskRun); - if (entries.length === 0) { - log.info("No S3 log entries, skipping JSONL hydration"); - return false; - } + const entries = await posthogAPI.fetchTaskRunLogs(taskRun); + if (entries.length === 0) { + log.info("No S3 log entries, skipping JSONL hydration"); + return false; + } - const entryCounts: Record = {}; - for (const entry of entries) { - const method = entry.notification?.method ?? "unknown"; - const entryParams = entry.notification?.params as - | Record - | undefined; - const update = entryParams?.update as - | { sessionUpdate?: string } - | undefined; - const key = update?.sessionUpdate - ? `${method}:${update.sessionUpdate}` - : method; - entryCounts[key] = (entryCounts[key] ?? 0) + 1; + const entryCounts: Record = {}; + for (const entry of entries) { + const method = entry.notification?.method ?? "unknown"; + const entryParams = entry.notification?.params as + | Record + | undefined; + const update = entryParams?.update as + | { sessionUpdate?: string } + | undefined; + const key = update?.sessionUpdate + ? `${method}:${update.sessionUpdate}` + : method; + entryCounts[key] = (entryCounts[key] ?? 0) + 1; + } + log.info("S3 log entry breakdown", { + totalEntries: entries.length, + types: entryCounts, + }); + + allTurns = rebuildConversation(entries); } - log.info("S3 log entry breakdown", { - totalEntries: entries.length, - types: entryCounts, - }); - const allTurns = rebuildConversation(entries); if (allTurns.length === 0) { - log.info("No conversation in S3 logs, skipping JSONL hydration"); + log.info("No conversation to hydrate, skipping JSONL hydration"); return false; } diff --git a/packages/agent/src/resume.ts b/packages/agent/src/resume.ts index 2f43181826..fe4c600f29 100644 --- a/packages/agent/src/resume.ts +++ b/packages/agent/src/resume.ts @@ -28,6 +28,7 @@ export interface ResumeState { interrupted: boolean; lastDevice?: DeviceInfo; logEntryCount: number; + sessionId: string | null; } export interface ConversationTurn { @@ -93,6 +94,7 @@ export async function resumeFromLog( interrupted: result.data.interrupted, lastDevice: result.data.lastDevice, logEntryCount: result.data.logEntryCount, + sessionId: result.data.sessionId, }; } diff --git a/packages/agent/src/sagas/resume-saga.test.ts b/packages/agent/src/sagas/resume-saga.test.ts index 3e8675e546..a26e6c7824 100644 --- a/packages/agent/src/sagas/resume-saga.test.ts +++ b/packages/agent/src/sagas/resume-saga.test.ts @@ -1,5 +1,6 @@ import type { SagaLogger } from "@posthog/shared"; import { afterEach, beforeEach, describe, expect, it, type vi } from "vitest"; +import { POSTHOG_NOTIFICATIONS } from "../acp-extensions"; import type { PostHogAPIClient } from "../posthog-api"; import { ResumeSaga } from "./resume-saga"; import { @@ -8,6 +9,7 @@ import { createGitCheckpointNotification, createMockApiClient, createMockLogger, + createNotification, createTaskRun, createTestRepo, createToolCall, @@ -553,6 +555,68 @@ describe("ResumeSaga", () => { }); }); + describe("session id", () => { + const runStarted = (sessionId: string) => + createNotification(POSTHOG_NOTIFICATIONS.RUN_STARTED, { sessionId }); + const sdkPrefixedRunStarted = (sessionId: string) => + createNotification(`_${POSTHOG_NOTIFICATIONS.RUN_STARTED}`, { + sessionId, + }); + + it.each([ + { + name: "extracts the session id from the run_started notification", + entries: () => [ + runStarted("session-abc"), + createUserMessage("Hello"), + createAgentChunk("Hi"), + ], + expected: "session-abc", + }, + { + name: "reads the sdk-prefixed run_started method too", + entries: () => [ + sdkPrefixedRunStarted("session-prefixed"), + createUserMessage("Hello"), + ], + expected: "session-prefixed", + }, + { + name: "returns the most recent session id when several are present", + entries: () => [ + runStarted("session-old"), + createUserMessage("Hello"), + runStarted("session-new"), + ], + expected: "session-new", + }, + { + name: "returns null when no run_started notification is present", + entries: () => [createUserMessage("Hello"), createAgentChunk("Hi")], + expected: null, + }, + ])("$name", async ({ entries, expected }) => { + (mockApiClient.getTaskRun as ReturnType).mockResolvedValue( + createTaskRun(), + ); + ( + mockApiClient.fetchTaskRunLogs as ReturnType + ).mockResolvedValue(entries()); + + const saga = new ResumeSaga(mockLogger); + const result = await saga.run({ + taskId: "task-1", + runId: "run-1", + repositoryPath: repo.path, + apiClient: mockApiClient, + }); + + expect(result.success).toBe(true); + if (!result.success) return; + expect(result.data.sessionId).toBe(expected); + }); + }); + describe("log entry count", () => { it("reports correct log entry count", async () => { (mockApiClient.getTaskRun as ReturnType).mockResolvedValue( diff --git a/packages/agent/src/sagas/resume-saga.ts b/packages/agent/src/sagas/resume-saga.ts index 6762effffb..09ad7e45e7 100644 --- a/packages/agent/src/sagas/resume-saga.ts +++ b/packages/agent/src/sagas/resume-saga.ts @@ -36,6 +36,7 @@ export interface ResumeOutput { interrupted: boolean; lastDevice?: DeviceInfo; logEntryCount: number; + sessionId: string | null; } export class ResumeSaga extends Saga { @@ -87,9 +88,14 @@ export class ResumeSaga extends Saga { Promise.resolve(this.findLastDeviceInfo(entries)), ); + const sessionId = await this.readOnlyStep("find_session_id", () => + Promise.resolve(this.findSessionId(entries)), + ); + this.log.info("Resume state rebuilt", { turns: conversation.length, hasGitCheckpoint: !!latestGitCheckpoint, + hasSessionId: !!sessionId, interrupted: false, }); @@ -99,6 +105,7 @@ export class ResumeSaga extends Saga { interrupted: false, lastDevice, logEntryCount: entries.length, + sessionId, }; } @@ -108,9 +115,26 @@ export class ResumeSaga extends Saga { latestGitCheckpoint: null, interrupted: false, logEntryCount: 0, + sessionId: null, }; } + private findSessionId(entries: StoredNotification[]): string | null { + const runStarted = POSTHOG_NOTIFICATIONS.RUN_STARTED; + for (let i = entries.length - 1; i >= 0; i--) { + const method = entries[i].notification?.method; + if (method === runStarted || method === `_${runStarted}`) { + const params = entries[i].notification?.params as + | { sessionId?: string } + | undefined; + if (typeof params?.sessionId === "string" && params.sessionId) { + return params.sessionId; + } + } + } + return null; + } + private findLatestGitCheckpoint( entries: StoredNotification[], ): GitCheckpointEvent | null { diff --git a/packages/agent/src/server/agent-server.ts b/packages/agent/src/server/agent-server.ts index cd05523174..0ebec260c3 100644 --- a/packages/agent/src/server/agent-server.ts +++ b/packages/agent/src/server/agent-server.ts @@ -1,4 +1,4 @@ -import { mkdir, writeFile } from "node:fs/promises"; +import { access, mkdir, writeFile } from "node:fs/promises"; import { basename, join } from "node:path"; import { pathToFileURL } from "node:url"; import type { @@ -22,6 +22,10 @@ import { createAcpConnection, type InProcessAcpConnection, } from "../adapters/acp-connection"; +import { + getSessionJsonlPath, + hydrateSessionJsonl, +} from "../adapters/claude/session/jsonl-hydration"; import type { GatewayEnv } from "../adapters/claude/session/options"; import { type AgentErrorClassification, @@ -259,6 +263,7 @@ export class AgentServer { private readonly evaluatedPrUrls = new Set(); private lastReportedBranch: string | null = null; private resumeState: ResumeState | null = null; + private nativeResume: { sessionId: string; warm: boolean } | null = null; // Guards against concurrent session initialization. autoInitializeSession() and // the GET /events SSE handler can both call initializeSession() — the SSE connection // often arrives while newSession() is still awaited (this.session is still null), @@ -590,6 +595,79 @@ export class AgentServer { } } + private async prepareNativeResume( + payload: JwtPayload, + posthogAPI: PostHogAPIClient, + preTaskRun: TaskRun | null, + runtimeAdapter: "claude" | "codex", + cwd: string, + permissionMode: PermissionMode, + ): Promise<{ sessionId: string; warm: boolean } | null> { + if (runtimeAdapter !== "claude") return null; + + const resumeRunId = this.getResumeRunId(preTaskRun); + if (!resumeRunId) return null; + + if (!this.resumeState) { + await this.loadResumeState(payload.task_id, resumeRunId, payload.run_id); + } + + const priorSessionId = this.resumeState?.sessionId ?? null; + if (!priorSessionId) { + this.logger.debug("No prior session id; using summary resume fallback", { + resumeRunId, + }); + return null; + } + + let warm = false; + try { + await access(getSessionJsonlPath(priorSessionId, cwd)); + warm = true; + } catch { + warm = false; + } + + try { + const hasSession = await hydrateSessionJsonl({ + sessionId: priorSessionId, + cwd, + taskId: payload.task_id, + runId: resumeRunId, + model: this.config.model, + permissionMode, + posthogAPI, + conversation: this.resumeState?.conversation, + log: { + info: (msg, data) => this.logger.debug(msg, data), + warn: (msg, data) => this.logger.warn(msg, data), + }, + }); + if (!hasSession) { + this.logger.debug( + "No session JSONL to resume; using summary fallback", + { + resumeRunId, + priorSessionId, + }, + ); + return null; + } + } catch (error) { + this.logger.warn( + "Session JSONL hydration failed; using summary fallback", + { + priorSessionId, + error: error instanceof Error ? error.message : String(error), + }, + ); + return null; + } + + this.logger.debug("Native resume prepared", { priorSessionId, warm }); + return { sessionId: priorSessionId, warm }; + } + async stop(): Promise { this.logger.debug("Stopping agent server..."); @@ -909,6 +987,9 @@ export class AgentServer { await this.cleanupSession(); } + this.resumeState = null; + this.nativeResume = null; + this.logger.debug("Initializing session", { runId: payload.run_id, taskId: payload.task_id, @@ -1059,29 +1140,57 @@ export class AgentServer { : runtimeAdapter === "codex" ? "auto" : "bypassPermissions"; - const sessionResponse = await clientConnection.newSession({ - cwd: this.config.repositoryPath ?? "/tmp/workspace", - mcpServers: this.config.mcpServers ?? [], - _meta: { - sessionId: payload.run_id, - taskRunId: payload.run_id, - taskId: payload.task_id, - environment: "cloud", - systemPrompt: sessionSystemPrompt, - ...(this.config.model && { model: this.config.model }), - allowedDomains: this.config.allowedDomains, - jsonSchema: preTask?.json_schema ?? null, - permissionMode: initialPermissionMode, - ...(this.config.baseBranch && { baseBranch: this.config.baseBranch }), - ...this.buildClaudeCodeSessionMeta(runtimeAdapter), - }, - }); + const sessionCwd = this.config.repositoryPath ?? "/tmp/workspace"; + const sessionMeta = { + sessionId: payload.run_id, + taskRunId: payload.run_id, + taskId: payload.task_id, + environment: "cloud", + systemPrompt: sessionSystemPrompt, + ...(this.config.model && { model: this.config.model }), + allowedDomains: this.config.allowedDomains, + jsonSchema: preTask?.json_schema ?? null, + permissionMode: initialPermissionMode, + ...(this.config.baseBranch && { baseBranch: this.config.baseBranch }), + ...this.buildClaudeCodeSessionMeta(runtimeAdapter), + }; - const acpSessionId = sessionResponse.sessionId; - this.logger.debug("ACP session created", { - acpSessionId, - runId: payload.run_id, - }); + const nativeResume = await this.prepareNativeResume( + payload, + posthogAPI, + preTaskRun, + runtimeAdapter, + sessionCwd, + initialPermissionMode, + ); + + let acpSessionId: string; + if (nativeResume) { + await clientConnection.resumeSession({ + sessionId: nativeResume.sessionId, + cwd: sessionCwd, + mcpServers: this.config.mcpServers ?? [], + _meta: { ...sessionMeta, sessionId: nativeResume.sessionId }, + }); + acpSessionId = nativeResume.sessionId; + this.nativeResume = nativeResume; + this.logger.debug("ACP session resumed", { + acpSessionId, + runId: payload.run_id, + warm: nativeResume.warm, + }); + } else { + const sessionResponse = await clientConnection.newSession({ + cwd: sessionCwd, + mcpServers: this.config.mcpServers ?? [], + _meta: sessionMeta, + }); + acpSessionId = sessionResponse.sessionId; + this.logger.debug("ACP session created", { + acpSessionId, + runId: payload.run_id, + }); + } this.prAttributed = false; this.evaluatedPrUrls.clear(); @@ -1207,6 +1316,11 @@ export class AgentServer { } } + if (this.nativeResume) { + await this.sendResumeContinuation(payload, taskRun); + return; + } + if (!this.resumeState) { const resumeRunId = this.getResumeRunId(taskRun); if (resumeRunId) { @@ -1218,7 +1332,6 @@ export class AgentServer { } } - // Resume flow: if we have resume state, format conversation history as context if (this.resumeState && this.resumeState.conversation.length > 0) { await this.sendResumeMessage(payload, taskRun); return; @@ -1288,44 +1401,14 @@ export class AgentServer { taskRun: TaskRun | null, ): Promise { if (!this.session || !this.resumeState) return; + const resumeState = this.resumeState; - try { + await this.runResumeTurn(payload, "Resume message", async () => { const conversationSummary = formatConversationForResume( - this.resumeState.conversation, + resumeState.conversation, ); - let checkpointApplied = false; - if ( - this.resumeState.latestGitCheckpoint && - this.config.repositoryPath && - this.posthogAPI - ) { - try { - const checkpointTracker = new HandoffCheckpointTracker({ - repositoryPath: this.config.repositoryPath, - taskId: payload.task_id, - runId: payload.run_id, - apiClient: this.posthogAPI, - logger: this.logger.child("HandoffCheckpoint"), - }); - const metrics = await checkpointTracker.applyFromHandoff( - this.resumeState.latestGitCheckpoint, - ); - checkpointApplied = true; - this.logger.debug("Git checkpoint applied", { - branch: this.resumeState.latestGitCheckpoint.branch, - head: this.resumeState.latestGitCheckpoint.head, - packBytes: metrics.packBytes, - indexBytes: metrics.indexBytes, - totalBytes: metrics.totalBytes, - }); - } catch (error) { - this.logger.warn("Failed to apply git checkpoint", { - error: error instanceof Error ? error.message : String(error), - branch: this.resumeState.latestGitCheckpoint.branch, - }); - } - } + const checkpointApplied = await this.applyResumeGitCheckpoint(payload); const pendingUserPrompt = await this.getPendingUserPrompt(taskRun); @@ -1365,26 +1448,72 @@ export class AgentServer { this.logger.debug("Sending resume message", { taskId: payload.task_id, - conversationTurns: this.resumeState.conversation.length, + conversationTurns: resumeState.conversation.length, promptLength: promptBlocksToText(resumePromptBlocks).length, hasPendingUserMessage: !!pendingUserPrompt?.length, checkpointApplied, - hasGitCheckpoint: !!this.resumeState.latestGitCheckpoint, - gitCheckpointBranch: - this.resumeState.latestGitCheckpoint?.branch ?? null, + hasGitCheckpoint: !!resumeState.latestGitCheckpoint, + gitCheckpointBranch: resumeState.latestGitCheckpoint?.branch ?? null, }); - // Clear resume state so it's not reused this.resumeState = null; + return resumePromptBlocks; + }); + } + + private async sendResumeContinuation( + payload: JwtPayload, + taskRun: TaskRun | null, + ): Promise { + if (!this.session) return; + + await this.runResumeTurn(payload, "Resume continuation", async () => { + const checkpointApplied = this.nativeResume?.warm + ? false + : await this.applyResumeGitCheckpoint(payload); + + const pendingUserPrompt = await this.getPendingUserPrompt(taskRun); + const prompt: ContentBlock[] = pendingUserPrompt?.length + ? pendingUserPrompt + : [ + { + type: "text", + text: "Continue from where you left off. The user is waiting for your response.", + }, + ]; + + this.logger.debug("Sending resume continuation", { + taskId: payload.task_id, + sessionId: this.nativeResume?.sessionId, + warm: this.nativeResume?.warm, + checkpointApplied, + hasPendingUserMessage: !!pendingUserPrompt?.length, + }); + + this.resumeState = null; + this.nativeResume = null; + return prompt; + }); + } + + private async runResumeTurn( + payload: JwtPayload, + logLabel: string, + buildPrompt: () => Promise, + ): Promise { + if (!this.session) return; + + try { + const prompt = await buildPrompt(); this.session.logWriter.resetTurnMessages(payload.run_id); const result = await this.session.clientConnection.prompt({ sessionId: this.session.acpSessionId, - prompt: resumePromptBlocks, + prompt, }); - this.logger.debug("Resume message completed", { + this.logger.debug(`${logLabel} completed`, { stopReason: result.stopReason, }); @@ -1398,7 +1527,7 @@ export class AgentServer { await this.relayAgentResponse(payload); } } catch (error) { - this.logger.error("Failed to send resume message", error); + this.logger.error(`Failed to send ${logLabel.toLowerCase()}`, error); if (this.session) { await this.session.logWriter.flushAll(); } @@ -1406,6 +1535,44 @@ export class AgentServer { } } + private async applyResumeGitCheckpoint( + payload: JwtPayload, + ): Promise { + if ( + !this.resumeState?.latestGitCheckpoint || + !this.config.repositoryPath || + !this.posthogAPI + ) { + return false; + } + try { + const checkpointTracker = new HandoffCheckpointTracker({ + repositoryPath: this.config.repositoryPath, + taskId: payload.task_id, + runId: payload.run_id, + apiClient: this.posthogAPI, + logger: this.logger.child("HandoffCheckpoint"), + }); + const metrics = await checkpointTracker.applyFromHandoff( + this.resumeState.latestGitCheckpoint, + ); + this.logger.debug("Git checkpoint applied", { + branch: this.resumeState.latestGitCheckpoint.branch, + head: this.resumeState.latestGitCheckpoint.head, + packBytes: metrics.packBytes, + indexBytes: metrics.indexBytes, + totalBytes: metrics.totalBytes, + }); + return true; + } catch (error) { + this.logger.warn("Failed to apply git checkpoint", { + error: error instanceof Error ? error.message : String(error), + branch: this.resumeState.latestGitCheckpoint.branch, + }); + return false; + } + } + private getInitialPromptOverride(taskRun: TaskRun): string | null { const state = taskRun.state as Record | undefined; const override = state?.initial_prompt_override; From 196ea73e4ade0da3674db45e40dc487957251642 Mon Sep 17 00:00:00 2001 From: Alessandro Pogliaghi Date: Thu, 25 Jun 2026 11:46:49 +0100 Subject: [PATCH 2/2] fix(agent): hydrate native resume from canonical logs --- .../claude/session/jsonl-hydration.ts | 64 ++++---- .../agent/src/server/agent-server.test.ts | 137 +++++++++++++++++- packages/agent/src/server/agent-server.ts | 1 - 3 files changed, 164 insertions(+), 38 deletions(-) diff --git a/packages/agent/src/adapters/claude/session/jsonl-hydration.ts b/packages/agent/src/adapters/claude/session/jsonl-hydration.ts index a4548d5d66..ec2836b3f0 100644 --- a/packages/agent/src/adapters/claude/session/jsonl-hydration.ts +++ b/packages/agent/src/adapters/claude/session/jsonl-hydration.ts @@ -497,7 +497,6 @@ export async function hydrateSessionJsonl(params: { permissionMode?: string; posthogAPI: PostHogAPIClient; log: HydrationLog; - conversation?: ConversationTurn[]; }): Promise { const { posthogAPI, log } = params; @@ -510,43 +509,38 @@ export async function hydrateSessionJsonl(params: { // File doesn't exist, proceed with hydration } - let allTurns: ConversationTurn[]; - if (params.conversation && params.conversation.length > 0) { - allTurns = params.conversation; - } else { - const taskRun = await posthogAPI.getTaskRun(params.taskId, params.runId); - if (!taskRun.log_url) { - log.info("No log URL, skipping JSONL hydration"); - return false; - } - - const entries = await posthogAPI.fetchTaskRunLogs(taskRun); - if (entries.length === 0) { - log.info("No S3 log entries, skipping JSONL hydration"); - return false; - } + const taskRun = await posthogAPI.getTaskRun(params.taskId, params.runId); + if (!taskRun.log_url) { + log.info("No log URL, skipping JSONL hydration"); + return false; + } - const entryCounts: Record = {}; - for (const entry of entries) { - const method = entry.notification?.method ?? "unknown"; - const entryParams = entry.notification?.params as - | Record - | undefined; - const update = entryParams?.update as - | { sessionUpdate?: string } - | undefined; - const key = update?.sessionUpdate - ? `${method}:${update.sessionUpdate}` - : method; - entryCounts[key] = (entryCounts[key] ?? 0) + 1; - } - log.info("S3 log entry breakdown", { - totalEntries: entries.length, - types: entryCounts, - }); + const entries = await posthogAPI.fetchTaskRunLogs(taskRun); + if (entries.length === 0) { + log.info("No S3 log entries, skipping JSONL hydration"); + return false; + } - allTurns = rebuildConversation(entries); + const entryCounts: Record = {}; + for (const entry of entries) { + const method = entry.notification?.method ?? "unknown"; + const entryParams = entry.notification?.params as + | Record + | undefined; + const update = entryParams?.update as + | { sessionUpdate?: string } + | undefined; + const key = update?.sessionUpdate + ? `${method}:${update.sessionUpdate}` + : method; + entryCounts[key] = (entryCounts[key] ?? 0) + 1; } + log.info("S3 log entry breakdown", { + totalEntries: entries.length, + types: entryCounts, + }); + + const allTurns = rebuildConversation(entries); if (allTurns.length === 0) { log.info("No conversation to hydrate, skipping JSONL hydration"); diff --git a/packages/agent/src/server/agent-server.test.ts b/packages/agent/src/server/agent-server.test.ts index e7030e7b18..919287a8b6 100644 --- a/packages/agent/src/server/agent-server.test.ts +++ b/packages/agent/src/server/agent-server.test.ts @@ -1,3 +1,5 @@ +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; import jwt from "jsonwebtoken"; import { type SetupServerApi, setupServer } from "msw/node"; import { @@ -10,9 +12,18 @@ import { it, vi, } from "vitest"; -import { createTestRepo, type TestRepo } from "../test/fixtures/api"; +import { getSessionJsonlPath } from "../adapters/claude/session/jsonl-hydration"; +import type { PermissionMode } from "../execution-mode"; +import type { PostHogAPIClient } from "../posthog-api"; +import type { ResumeState } from "../resume"; +import { + createMockApiClient, + createTaskRun, + createTestRepo, + type TestRepo, +} from "../test/fixtures/api"; import { createPostHogHandlers } from "../test/mocks/msw-handlers"; -import type { TaskRun } from "../types"; +import type { StoredEntry, TaskRun } from "../types"; import { AgentServer, isTurnCompleteNotification, @@ -217,6 +228,18 @@ interface TestableServer { ): { claudeCode: { options: Record } } | undefined; } +interface NativeResumeTestServer { + resumeState: ResumeState | null; + prepareNativeResume( + payload: JwtPayload, + posthogAPI: PostHogAPIClient, + preTaskRun: TaskRun | null, + runtimeAdapter: "claude" | "codex", + cwd: string, + permissionMode: PermissionMode, + ): Promise<{ sessionId: string; warm: boolean } | null>; +} + let nextTestPort = 20000; function getNextTestPort(): number { @@ -272,6 +295,21 @@ function createTestJwt( ); } +function sessionUpdateEntry( + sessionUpdate: string, + extra: Record = {}, +): StoredEntry { + return { + type: "notification", + timestamp: new Date().toISOString(), + notification: { + jsonrpc: "2.0", + method: "session/update", + params: { update: { sessionUpdate, ...extra } }, + }, + }; +} + // Test RSA key pair (2048-bit, for testing only) const TEST_PRIVATE_KEY = `-----BEGIN PRIVATE KEY----- MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDqh94SYMFsvG4C @@ -1204,6 +1242,101 @@ describe("AgentServer HTTP Mode", () => { }); }); + describe("native resume", () => { + it("hydrates cold sessions from S3 logs instead of cached resume conversation", async () => { + const originalConfigDir = process.env.CLAUDE_CONFIG_DIR; + process.env.CLAUDE_CONFIG_DIR = join(repo.path, ".claude-test"); + + try { + const s = createServer() as unknown as NativeResumeTestServer; + s.resumeState = { + conversation: [ + { + role: "user", + content: [{ type: "text", text: "continue" }], + }, + { + role: "assistant", + content: [{ type: "text", text: "visible answer only" }], + }, + ], + latestGitCheckpoint: null, + interrupted: false, + logEntryCount: 3, + sessionId: "prior-session", + }; + + const posthogAPI = createMockApiClient(); + (posthogAPI.getTaskRun as ReturnType).mockResolvedValue( + createTaskRun({ id: "previous-run", log_url: "s3://logs" }), + ); + ( + posthogAPI.fetchTaskRunLogs as ReturnType + ).mockResolvedValue([ + sessionUpdateEntry("user_message", { + content: { type: "text", text: "continue" }, + }), + sessionUpdateEntry("agent_thought_chunk", { + content: { + type: "thinking", + thinking: "preserve extended thinking", + }, + }), + sessionUpdateEntry("agent_message", { + content: { type: "text", text: "visible answer" }, + }), + ]); + + const result = await s.prepareNativeResume( + { + task_id: "test-task-id", + run_id: "test-run-id", + team_id: 1, + user_id: 1, + distinct_id: "test-distinct-id", + mode: "interactive", + }, + posthogAPI, + createTaskRun({ + id: "test-run-id", + state: { resume_from_run_id: "previous-run" }, + }), + "claude", + repo.path, + "bypassPermissions", + ); + + expect(result).toEqual({ sessionId: "prior-session", warm: false }); + expect(posthogAPI.fetchTaskRunLogs).toHaveBeenCalledTimes(1); + + const jsonl = await readFile( + getSessionJsonlPath("prior-session", repo.path), + "utf-8", + ); + const blocks = jsonl + .trim() + .split("\n") + .flatMap((line) => { + const parsed = JSON.parse(line) as { + message?: { content?: unknown[] }; + }; + return parsed.message?.content ?? []; + }); + + expect(blocks).toContainEqual({ + type: "thinking", + thinking: "preserve extended thinking", + }); + } finally { + if (originalConfigDir === undefined) { + delete process.env.CLAUDE_CONFIG_DIR; + } else { + process.env.CLAUDE_CONFIG_DIR = originalConfigDir; + } + } + }); + }); + describe("PR attribution", () => { const PR_URL = "https://github.com/PostHog/posthog.com/pull/17764"; const payload: JwtPayload = { diff --git a/packages/agent/src/server/agent-server.ts b/packages/agent/src/server/agent-server.ts index 0ebec260c3..fd0e30fd9f 100644 --- a/packages/agent/src/server/agent-server.ts +++ b/packages/agent/src/server/agent-server.ts @@ -637,7 +637,6 @@ export class AgentServer { model: this.config.model, permissionMode, posthogAPI, - conversation: this.resumeState?.conversation, log: { info: (msg, data) => this.logger.debug(msg, data), warn: (msg, data) => this.logger.warn(msg, data),