From 43f136deba843ea34c2399af5b6a3fc6a010c8e5 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 9 May 2026 19:52:40 +0700 Subject: [PATCH] fix: reload agent rehydrates timeline from disk under new epoch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reload was a no-op for imported agents whose on-disk session was mutated externally. The server preserved the in-memory timeline + historyPrimed flag, so hydrate short-circuited and the epoch never changed. The client also fetched without a cursor on reload, so the server returned reset:false and the reducer dropped the new-epoch rows against a stale cursor — the chat stayed stale until refocus. Server: reloadAgentSession gains rehydrateFromDisk (used by the refresh RPC). When set, durable + in-memory timeline are wiped and historyPrimed is reset, so register mints a new epoch and hydrateTimelineFromProvider re-streams the freshly read provider history. Voice-mode reloads keep the existing preserve-timeline semantics. Client: handleReloadAgent shows a sticky "Reloading agent…" toast, sends the stale cursor with the fetch so the server returns reset:true, and only flips to "Reloaded agent" after the response listener has replaced the visible tail. --- .../screens/workspace/workspace-screen.tsx | 16 +- .../server/src/server/agent/agent-manager.ts | 20 ++- ...nt-refresh-rehydrates-timeline.e2e.test.ts | 156 ++++++++++++++++++ packages/server/src/server/session.ts | 4 +- 4 files changed, 191 insertions(+), 5 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts diff --git a/packages/app/src/screens/workspace/workspace-screen.tsx b/packages/app/src/screens/workspace/workspace-screen.tsx index 272d5f395..66462500a 100644 --- a/packages/app/src/screens/workspace/workspace-screen.tsx +++ b/packages/app/src/screens/workspace/workspace-screen.tsx @@ -2441,14 +2441,28 @@ function WorkspaceScreenContent({ return; } + toast.show("Reloading agent…", { durationMs: null }); try { await client.refreshAgent(agentId); + // Send the existing cursor so the server detects the new epoch and + // returns reset:true. Without a cursor, the server returns reset:false + // and the client takes the incremental path, where new-epoch rows are + // dropped against the stale cursor. + const sessionState = useSessionStore.getState().sessions[normalizedServerId]; + const currentCursor = sessionState?.agentTimelineCursor.get(agentId); + await client.fetchAgentTimeline(agentId, { + direction: "tail", + projection: "canonical", + ...(currentCursor + ? { cursor: { epoch: currentCursor.epoch, seq: currentCursor.endSeq } } + : {}), + }); toast.show("Reloaded agent", { variant: "success" }); } catch (error) { toast.error(error instanceof Error ? error.message : "Failed to reload agent"); } }, - [client, isConnected, toast], + [client, isConnected, normalizedServerId, toast], ); const handleCopyWorkspacePath = useCallback(async () => { diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 199a4b71f..62afd20b7 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -879,17 +879,23 @@ export class AgentManager { return this.registerSession(session, normalizedConfig, resolvedAgentId, options); } - // Hot-reload an active agent session with config overrides while preserving - // in-memory timeline state. + // Hot-reload an active agent session with config overrides. By default the + // in-memory timeline is preserved (used for voice-mode toggles and similar + // config swaps). When `rehydrateFromDisk` is set, the timeline is wiped so a + // new epoch is minted and provider history is re-streamed — this is what the + // user-facing "Reload agent" action wants when the on-disk session was + // mutated outside Paseo. async reloadAgentSession( agentId: string, overrides?: Partial, + options?: { rehydrateFromDisk?: boolean }, ): Promise { let existing = this.requireSessionAgent(agentId); if (this.hasInFlightRun(agentId)) { await this.cancelAgentRun(agentId); existing = this.requireSessionAgent(agentId); } + const rehydrateFromDisk = options?.rehydrateFromDisk ?? false; const preservedHistoryPrimed = existing.historyPrimed; const preservedLastUsage = existing.lastUsage; const preservedLastError = existing.lastError; @@ -919,13 +925,21 @@ export class AgentManager { this.foregroundRuns.clearAgent(agentId, existing); await this.closeReloadedSession(existing.session, agentId); + if (rehydrateFromDisk) { + // Wipe both durable and in-memory timeline so registerSession mints a + // new epoch and hydrateTimelineFromProvider re-streams the freshly read + // provider history into an empty timeline. + await this.deleteCommittedTimeline(agentId); + this.timelineStore.delete(agentId); + } + // Preserve existing labels and timeline during reload. return this.registerSession(session, normalizedConfig, agentId, { labels: existing.labels, createdAt: existing.createdAt, updatedAt: existing.updatedAt, lastUserMessageAt: existing.lastUserMessageAt, - historyPrimed: preservedHistoryPrimed, + historyPrimed: rehydrateFromDisk ? false : preservedHistoryPrimed, lastUsage: preservedLastUsage, lastError: preservedLastError, attention: preservedAttention, diff --git a/packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts new file mode 100644 index 000000000..44e774d7a --- /dev/null +++ b/packages/server/src/server/daemon-e2e/agent-refresh-rehydrates-timeline.e2e.test.ts @@ -0,0 +1,156 @@ +import { afterEach, beforeEach, describe, expect, test } from "vitest"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync, appendFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; + +import { ClaudeAgentClient } from "../agent/providers/claude/agent.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; +import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js"; + +function sanitizeClaudeProjectPath(cwd: string): string { + return cwd.replace(/[\\/._:]/g, "-"); +} + +interface ClaudeJsonlEntry { + type: "user" | "assistant"; + uuid?: string; + sessionId: string; + cwd: string; + message: { role: "user" | "assistant"; content: string }; +} + +function userEntry( + sessionId: string, + cwd: string, + content: string, + uuid: string, +): ClaudeJsonlEntry { + return { + type: "user", + uuid, + sessionId, + cwd, + message: { role: "user", content }, + }; +} + +function assistantEntry(sessionId: string, cwd: string, content: string): ClaudeJsonlEntry { + return { + type: "assistant", + sessionId, + cwd, + message: { role: "assistant", content }, + }; +} + +function timelineText(entries: ReadonlyArray<{ item: { type: string; text?: string } }>): string { + return entries + .filter( + (entry): entry is { item: { type: "user_message" | "assistant_message"; text: string } } => + entry.item.type === "user_message" || entry.item.type === "assistant_message", + ) + .map((entry) => entry.item.text) + .join("\n"); +} + +describe("daemon E2E - refresh rehydrates timeline from on-disk session", () => { + let claudeConfigDir: string; + let prevClaudeConfigDir: string | undefined; + let cwd: string; + let sessionFile: string; + let daemon: TestPaseoDaemon | undefined; + let client: DaemonClient | undefined; + + const sessionId = "external-edits-session"; + + beforeEach(() => { + prevClaudeConfigDir = process.env.CLAUDE_CONFIG_DIR; + claudeConfigDir = mkdtempSync(path.join(tmpdir(), "claude-cfg-refresh-")); + process.env.CLAUDE_CONFIG_DIR = claudeConfigDir; + + cwd = mkdtempSync(path.join(tmpdir(), "claude-cwd-refresh-")); + const projectsDir = path.join(claudeConfigDir, "projects", sanitizeClaudeProjectPath(cwd)); + mkdirSync(projectsDir, { recursive: true }); + sessionFile = path.join(projectsDir, `${sessionId}.jsonl`); + + const initial: ClaudeJsonlEntry[] = [ + userEntry(sessionId, cwd, "first hello", "user-uuid-1"), + assistantEntry(sessionId, cwd, "first reply"), + ]; + writeFileSync( + sessionFile, + `${initial.map((entry) => JSON.stringify(entry)).join("\n")}\n`, + "utf8", + ); + }); + + afterEach(async () => { + await client?.close().catch(() => undefined); + await daemon?.close().catch(() => undefined); + client = undefined; + daemon = undefined; + rmSync(claudeConfigDir, { recursive: true, force: true }); + rmSync(cwd, { recursive: true, force: true }); + if (prevClaudeConfigDir === undefined) { + delete process.env.CLAUDE_CONFIG_DIR; + } else { + process.env.CLAUDE_CONFIG_DIR = prevClaudeConfigDir; + } + }, 60_000); + + test("refresh picks up entries appended externally and advances the epoch", async () => { + const logger = pino({ level: "silent" }); + daemon = await createTestPaseoDaemon({ + agentClients: { + claude: new ClaudeAgentClient({ + logger, + resolveBinary: async () => "/test/claude/bin", + }), + }, + logger, + }); + client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + await client.connect(); + await client.fetchAgents({ + subscribe: { subscriptionId: "refresh-rehydrate-test" }, + }); + + const imported = await client.importAgent({ provider: "claude", sessionId, cwd }); + expect(imported.id).toBeTruthy(); + + const before = await client.fetchAgentTimeline(imported.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + const beforeText = timelineText(before.entries); + expect(beforeText).toContain("first hello"); + expect(beforeText).toContain("first reply"); + const epochBefore = before.epoch; + const countBefore = before.entries.length; + + const additions: ClaudeJsonlEntry[] = [ + userEntry(sessionId, cwd, "second hello", "user-uuid-2"), + assistantEntry(sessionId, cwd, "second reply"), + ]; + appendFileSync( + sessionFile, + `${additions.map((entry) => JSON.stringify(entry)).join("\n")}\n`, + "utf8", + ); + + await client.refreshAgent(imported.id); + + const after = await client.fetchAgentTimeline(imported.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + const afterText = timelineText(after.entries); + expect(afterText).toContain("second hello"); + expect(afterText).toContain("second reply"); + expect(after.entries.length).toBeGreaterThan(countBefore); + expect(after.epoch).not.toBe(epochBefore); + }, 30_000); +}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 4e0a81784..7532ecbcf 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -3306,7 +3306,9 @@ export class Session { const existing = this.agentManager.getAgent(agentId); if (existing) { await this.interruptAgentIfRunning(agentId); - snapshot = await this.agentManager.reloadAgentSession(agentId); + snapshot = await this.agentManager.reloadAgentSession(agentId, undefined, { + rehydrateFromDisk: true, + }); } else { const record = await this.agentStorage.get(agentId); if (!record) {