Skip to content
This repository was archived by the owner on Aug 6, 2026. It is now read-only.

Commit f4fb079

Browse files
richardsolomourichardsolomou
authored
fix(sessions): preserve conversation across reconnects (#3767)
1 parent 83828b9 commit f4fb079

5 files changed

Lines changed: 297 additions & 26 deletions

File tree

‎packages/agent/src/adapters/claude/session/jsonl-hydration.test.ts‎

Lines changed: 33 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1595,6 +1595,37 @@ describe("hydrateSessionJsonl", () => {
15951595
};
15961596
}
15971597

1598+
it("returns the selected conversation when hydrating from logs", async () => {
1599+
const posthogAPI = {
1600+
getTaskRun: vi.fn().mockResolvedValue({ log_url: "https://logs.test" }),
1601+
fetchTaskRunLogs: vi.fn().mockResolvedValue([
1602+
entry("user_message", {
1603+
content: { type: "text", text: "previous request" },
1604+
}),
1605+
]),
1606+
} as unknown as PostHogAPIClient;
1607+
const log = { info: vi.fn(), warn: vi.fn() };
1608+
1609+
const result = await hydrateSessionJsonl({
1610+
sessionId,
1611+
cwd,
1612+
taskId: "t1",
1613+
runId: "r1",
1614+
posthogAPI,
1615+
log,
1616+
});
1617+
1618+
expect(result).toEqual({
1619+
hasSession: true,
1620+
conversation: [
1621+
{
1622+
role: "user",
1623+
content: [{ type: "text", text: "previous request" }],
1624+
},
1625+
],
1626+
});
1627+
});
1628+
15981629
it("sanitizes an existing file and skips S3 hydration", async () => {
15991630
const file = await writeSessionFile();
16001631
const { posthogAPI, log } = makeDeps();
@@ -1608,7 +1639,7 @@ describe("hydrateSessionJsonl", () => {
16081639
log,
16091640
});
16101641

1611-
expect(result).toBe(true);
1642+
expect(result).toEqual({ hasSession: true });
16121643
expect(
16131644
(posthogAPI as unknown as { getTaskRun: ReturnType<typeof vi.fn> })
16141645
.getTaskRun,
@@ -1635,7 +1666,7 @@ describe("hydrateSessionJsonl", () => {
16351666
log,
16361667
});
16371668

1638-
expect(result).toBe(true);
1669+
expect(result).toEqual({ hasSession: true });
16391670
expect(log.warn).toHaveBeenCalledWith(
16401671
"Failed to sanitize existing session JSONL",
16411672
expect.anything(),

‎packages/agent/src/adapters/claude/session/jsonl-hydration.ts‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -10,12 +10,17 @@ import { isEmptyContentBlock } from "../../../utils/acp-content";
1010
import { neutralizeUnprocessableImages } from "../image-sanitization";
1111
import { supports1MContext } from "./models";
1212

13-
interface ConversationTurn {
13+
export interface ConversationTurn {
1414
role: "user" | "assistant";
1515
content: ContentBlock[];
1616
toolCalls?: ToolCallInfo[];
1717
}
1818

19+
export interface HydrateSessionJsonlResult {
20+
hasSession: boolean;
21+
conversation?: ConversationTurn[];
22+
}
23+
1924
interface ToolCallInfo {
2025
toolCallId: string;
2126
toolName: string;
@@ -681,7 +686,7 @@ export async function hydrateSessionJsonl(params: {
681686
permissionMode?: string;
682687
posthogAPI: PostHogAPIClient;
683688
log: HydrationLog;
684-
}): Promise<boolean> {
689+
}): Promise<HydrateSessionJsonlResult> {
685690
const { posthogAPI, log } = params;
686691

687692
try {
@@ -702,21 +707,21 @@ export async function hydrateSessionJsonl(params: {
702707
error: err instanceof Error ? err.message : String(err),
703708
});
704709
}
705-
return true;
710+
return { hasSession: true };
706711
} catch {
707712
// File doesn't exist, proceed with hydration
708713
}
709714

710715
const taskRun = await posthogAPI.getTaskRun(params.taskId, params.runId);
711716
if (!taskRun.log_url) {
712717
log.info("No log URL, skipping JSONL hydration");
713-
return false;
718+
return { hasSession: false };
714719
}
715720

716721
const entries = await posthogAPI.fetchTaskRunLogs(taskRun);
717722
if (entries.length === 0) {
718723
log.info("No S3 log entries, skipping JSONL hydration");
719-
return false;
724+
return { hasSession: false };
720725
}
721726

722727
const entryCounts: Record<string, number> = {};
@@ -742,7 +747,7 @@ export async function hydrateSessionJsonl(params: {
742747

743748
if (allTurns.length === 0) {
744749
log.info("No conversation to hydrate, skipping JSONL hydration");
745-
return false;
750+
return { hasSession: false };
746751
}
747752

748753
const maxTokens = supports1MContext(params.model ?? "")
@@ -774,12 +779,12 @@ export async function hydrateSessionJsonl(params: {
774779
turns: conversation.length,
775780
lines: jsonlLines.length,
776781
});
777-
return true;
782+
return { hasSession: true, conversation };
778783
} catch (err) {
779784
log.warn("Failed to hydrate session JSONL, continuing", {
780785
sessionId: params.sessionId,
781786
error: err instanceof Error ? err.message : String(err),
782787
});
783-
return false;
788+
return { hasSession: false };
784789
}
785790
}

‎packages/agent/src/server/agent-server.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -885,7 +885,7 @@ export class AgentServer {
885885
}
886886

887887
try {
888-
const hasSession = await hydrateSessionJsonl({
888+
const { hasSession } = await hydrateSessionJsonl({
889889
sessionId: priorSessionId,
890890
cwd,
891891
taskId: payload.task_id,

‎packages/workspace-server/src/services/agent/agent.test.ts‎

Lines changed: 162 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,12 @@ const mockNewSession = vi.hoisted(() =>
1515
configOptions: [],
1616
}),
1717
);
18+
const mockResumeSession = vi.hoisted(() =>
19+
vi.fn().mockResolvedValue({ configOptions: [] }),
20+
);
21+
const mockPrompt = vi.hoisted(() =>
22+
vi.fn().mockResolvedValue({ stopReason: "end_turn" }),
23+
);
1824

1925
const mockAcpClient = vi.hoisted(() => ({
2026
current: undefined as
@@ -40,7 +46,8 @@ const mockClientSideConnection = vi.hoisted(() =>
4046
this.initialize = vi.fn().mockResolvedValue({});
4147
this.newSession = mockNewSession;
4248
this.loadSession = vi.fn().mockResolvedValue({ configOptions: [] });
43-
this.resumeSession = vi.fn().mockResolvedValue({ configOptions: [] });
49+
this.resumeSession = mockResumeSession;
50+
this.prompt = mockPrompt;
4451
this.setSessionConfigOption = vi.fn(
4552
async ({ value }: { value: string }) => ({
4653
configOptions: [
@@ -70,6 +77,12 @@ const mockAgentRun = vi.hoisted(() =>
7077
),
7178
);
7279

80+
const mockResumeFromLog = vi.hoisted(() => vi.fn());
81+
const mockFormatConversationForResume = vi.hoisted(() => vi.fn());
82+
const mockHydrateSessionJsonl = vi.hoisted(() =>
83+
vi.fn().mockResolvedValue({ hasSession: false }),
84+
);
85+
7386
const mockAgentConstructor = vi.hoisted(() =>
7487
vi.fn().mockImplementation(function (this: Record<string, unknown>) {
7588
this.run = mockAgentRun;
@@ -103,6 +116,11 @@ vi.mock("@posthog/agent/posthog-api", () => ({
103116
getLlmGatewayUrl: vi.fn(() => "https://gateway.example.com"),
104117
}));
105118

119+
vi.mock("@posthog/agent/resume", () => ({
120+
resumeFromLog: mockResumeFromLog,
121+
formatConversationForResume: mockFormatConversationForResume,
122+
}));
123+
106124
vi.mock("@posthog/agent/gateway-models", () => ({
107125
DEFAULT_GATEWAY_MODEL: "claude-opus-4-8",
108126
DEFAULT_CODEX_MODEL: "gpt-5.5",
@@ -113,7 +131,7 @@ vi.mock("@posthog/agent/gateway-models", () => ({
113131
}));
114132

115133
vi.mock("@posthog/agent/adapters/claude/session/jsonl-hydration", () => ({
116-
hydrateSessionJsonl: vi.fn().mockResolvedValue(undefined),
134+
hydrateSessionJsonl: mockHydrateSessionJsonl,
117135
}));
118136

119137
vi.mock("node:fs", async (importOriginal) => {
@@ -334,6 +352,148 @@ describe("AgentService", () => {
334352
});
335353
});
336354

355+
describe("reconnect", () => {
356+
it("preserves conversation context when native reconnect fails", async () => {
357+
const apiClient = {};
358+
mockAgentConstructor.mockImplementationOnce(function (
359+
this: Record<string, unknown>,
360+
) {
361+
this.run = mockAgentRun;
362+
this.cleanup = vi.fn().mockResolvedValue(undefined);
363+
this.getPosthogAPI = vi.fn(() => apiClient);
364+
this.flushAllLogs = vi.fn().mockResolvedValue(undefined);
365+
});
366+
mockResumeFromLog.mockResolvedValue({ conversation: [{ role: "user" }] });
367+
mockFormatConversationForResume.mockReturnValue("User: previous request");
368+
mockResumeSession.mockRejectedValueOnce(new Error("not found"));
369+
await service.reconnectSession({
370+
...baseSessionParams,
371+
adapter: "codex",
372+
sessionId: "old-session",
373+
});
374+
375+
await service.prompt("run-1", [{ type: "text", text: "next request" }]);
376+
await service.prompt("run-1", [{ type: "text", text: "later request" }]);
377+
378+
expect(mockPrompt.mock.calls[0][0].prompt).toEqual([
379+
expect.objectContaining({
380+
type: "text",
381+
text: expect.stringContaining("previous request"),
382+
}),
383+
{ type: "text", text: "next request" },
384+
]);
385+
expect(mockPrompt.mock.calls[1][0].prompt).toEqual([
386+
{ type: "text", text: "later request" },
387+
]);
388+
});
389+
390+
it("preserves conversation context when reconnect has no session ID", async () => {
391+
mockAgentConstructor.mockImplementationOnce(function (
392+
this: Record<string, unknown>,
393+
) {
394+
this.run = mockAgentRun;
395+
this.cleanup = vi.fn().mockResolvedValue(undefined);
396+
this.getPosthogAPI = vi.fn(() => ({}));
397+
this.flushAllLogs = vi.fn().mockResolvedValue(undefined);
398+
});
399+
mockResumeFromLog.mockResolvedValue({ conversation: [{ role: "user" }] });
400+
mockFormatConversationForResume.mockReturnValue("User: previous request");
401+
402+
await service.reconnectSession({
403+
...baseSessionParams,
404+
adapter: "codex",
405+
});
406+
await service.prompt("run-1", [{ type: "text", text: "next request" }]);
407+
408+
expect(mockPrompt.mock.calls[0][0].prompt[0].text).toContain(
409+
"previous request",
410+
);
411+
});
412+
413+
it("reuses hydrated conversation when Claude resume fails", async () => {
414+
mockAgentConstructor.mockImplementationOnce(function (
415+
this: Record<string, unknown>,
416+
) {
417+
this.run = mockAgentRun;
418+
this.cleanup = vi.fn().mockResolvedValue(undefined);
419+
this.getPosthogAPI = vi.fn(() => ({}));
420+
this.flushAllLogs = vi.fn().mockResolvedValue(undefined);
421+
});
422+
mockHydrateSessionJsonl.mockResolvedValueOnce({
423+
hasSession: true,
424+
conversation: [{ role: "user", content: [] }],
425+
});
426+
mockFormatConversationForResume.mockReturnValue("User: hydrated request");
427+
mockResumeSession.mockRejectedValueOnce(new Error("not found"));
428+
429+
await service.reconnectSession({
430+
...baseSessionParams,
431+
adapter: "claude",
432+
sessionId: "old-session",
433+
});
434+
await service.prompt("run-1", [{ type: "text", text: "next request" }]);
435+
436+
expect(mockResumeFromLog).not.toHaveBeenCalled();
437+
expect(mockPrompt.mock.calls[0][0].prompt[0].text).toContain(
438+
"hydrated request",
439+
);
440+
});
441+
442+
it("does not resend hydrated conversation after native resume succeeds", async () => {
443+
mockAgentConstructor.mockImplementationOnce(function (
444+
this: Record<string, unknown>,
445+
) {
446+
this.run = mockAgentRun;
447+
this.cleanup = vi.fn().mockResolvedValue(undefined);
448+
this.getPosthogAPI = vi.fn(() => ({}));
449+
this.flushAllLogs = vi.fn().mockResolvedValue(undefined);
450+
});
451+
mockHydrateSessionJsonl.mockResolvedValueOnce({
452+
hasSession: true,
453+
conversation: [{ role: "user", content: [] }],
454+
});
455+
mockFormatConversationForResume.mockReturnValue("User: hydrated request");
456+
457+
await service.reconnectSession({
458+
...baseSessionParams,
459+
adapter: "claude",
460+
sessionId: "old-session",
461+
});
462+
await service.prompt("run-1", [{ type: "text", text: "next request" }]);
463+
464+
expect(mockPrompt.mock.calls[0][0].prompt).toEqual([
465+
{ type: "text", text: "next request" },
466+
]);
467+
});
468+
469+
it("retries recovered context after prompt failure", async () => {
470+
mockAgentConstructor.mockImplementationOnce(function (
471+
this: Record<string, unknown>,
472+
) {
473+
this.run = mockAgentRun;
474+
this.cleanup = vi.fn().mockResolvedValue(undefined);
475+
this.getPosthogAPI = vi.fn(() => ({}));
476+
this.flushAllLogs = vi.fn().mockResolvedValue(undefined);
477+
});
478+
mockResumeFromLog.mockResolvedValue({ conversation: [{ role: "user" }] });
479+
mockFormatConversationForResume.mockReturnValue("User: previous request");
480+
mockPrompt.mockRejectedValueOnce(new Error("connection lost"));
481+
482+
await service.reconnectSession({
483+
...baseSessionParams,
484+
adapter: "codex",
485+
});
486+
await expect(
487+
service.prompt("run-1", [{ type: "text", text: "first attempt" }]),
488+
).rejects.toThrow("connection lost");
489+
await service.prompt("run-1", [{ type: "text", text: "retry" }]);
490+
491+
expect(mockPrompt.mock.calls[1][0].prompt[0].text).toContain(
492+
"previous request",
493+
);
494+
});
495+
});
496+
337497
describe("MCP servers", () => {
338498
it("marks desktop sessions as local even though they have a taskRunId", async () => {
339499
await service.startSession({

0 commit comments

Comments
 (0)