fix: reload agent rehydrates timeline from disk under new epoch
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.
This commit is contained in:
@@ -2441,14 +2441,28 @@ function WorkspaceScreenContent({
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
toast.show("Reloading agent…", { durationMs: null });
|
||||||
try {
|
try {
|
||||||
await client.refreshAgent(agentId);
|
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" });
|
toast.show("Reloaded agent", { variant: "success" });
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
toast.error(error instanceof Error ? error.message : "Failed to reload agent");
|
toast.error(error instanceof Error ? error.message : "Failed to reload agent");
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
[client, isConnected, toast],
|
[client, isConnected, normalizedServerId, toast],
|
||||||
);
|
);
|
||||||
|
|
||||||
const handleCopyWorkspacePath = useCallback(async () => {
|
const handleCopyWorkspacePath = useCallback(async () => {
|
||||||
|
|||||||
@@ -879,17 +879,23 @@ export class AgentManager {
|
|||||||
return this.registerSession(session, normalizedConfig, resolvedAgentId, options);
|
return this.registerSession(session, normalizedConfig, resolvedAgentId, options);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Hot-reload an active agent session with config overrides while preserving
|
// Hot-reload an active agent session with config overrides. By default the
|
||||||
// in-memory timeline state.
|
// 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(
|
async reloadAgentSession(
|
||||||
agentId: string,
|
agentId: string,
|
||||||
overrides?: Partial<AgentSessionConfig>,
|
overrides?: Partial<AgentSessionConfig>,
|
||||||
|
options?: { rehydrateFromDisk?: boolean },
|
||||||
): Promise<ManagedAgent> {
|
): Promise<ManagedAgent> {
|
||||||
let existing = this.requireSessionAgent(agentId);
|
let existing = this.requireSessionAgent(agentId);
|
||||||
if (this.hasInFlightRun(agentId)) {
|
if (this.hasInFlightRun(agentId)) {
|
||||||
await this.cancelAgentRun(agentId);
|
await this.cancelAgentRun(agentId);
|
||||||
existing = this.requireSessionAgent(agentId);
|
existing = this.requireSessionAgent(agentId);
|
||||||
}
|
}
|
||||||
|
const rehydrateFromDisk = options?.rehydrateFromDisk ?? false;
|
||||||
const preservedHistoryPrimed = existing.historyPrimed;
|
const preservedHistoryPrimed = existing.historyPrimed;
|
||||||
const preservedLastUsage = existing.lastUsage;
|
const preservedLastUsage = existing.lastUsage;
|
||||||
const preservedLastError = existing.lastError;
|
const preservedLastError = existing.lastError;
|
||||||
@@ -919,13 +925,21 @@ export class AgentManager {
|
|||||||
this.foregroundRuns.clearAgent(agentId, existing);
|
this.foregroundRuns.clearAgent(agentId, existing);
|
||||||
await this.closeReloadedSession(existing.session, agentId);
|
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.
|
// Preserve existing labels and timeline during reload.
|
||||||
return this.registerSession(session, normalizedConfig, agentId, {
|
return this.registerSession(session, normalizedConfig, agentId, {
|
||||||
labels: existing.labels,
|
labels: existing.labels,
|
||||||
createdAt: existing.createdAt,
|
createdAt: existing.createdAt,
|
||||||
updatedAt: existing.updatedAt,
|
updatedAt: existing.updatedAt,
|
||||||
lastUserMessageAt: existing.lastUserMessageAt,
|
lastUserMessageAt: existing.lastUserMessageAt,
|
||||||
historyPrimed: preservedHistoryPrimed,
|
historyPrimed: rehydrateFromDisk ? false : preservedHistoryPrimed,
|
||||||
lastUsage: preservedLastUsage,
|
lastUsage: preservedLastUsage,
|
||||||
lastError: preservedLastError,
|
lastError: preservedLastError,
|
||||||
attention: preservedAttention,
|
attention: preservedAttention,
|
||||||
|
|||||||
@@ -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);
|
||||||
|
});
|
||||||
@@ -3306,7 +3306,9 @@ export class Session {
|
|||||||
const existing = this.agentManager.getAgent(agentId);
|
const existing = this.agentManager.getAgent(agentId);
|
||||||
if (existing) {
|
if (existing) {
|
||||||
await this.interruptAgentIfRunning(agentId);
|
await this.interruptAgentIfRunning(agentId);
|
||||||
snapshot = await this.agentManager.reloadAgentSession(agentId);
|
snapshot = await this.agentManager.reloadAgentSession(agentId, undefined, {
|
||||||
|
rehydrateFromDisk: true,
|
||||||
|
});
|
||||||
} else {
|
} else {
|
||||||
const record = await this.agentStorage.get(agentId);
|
const record = await this.agentStorage.get(agentId);
|
||||||
if (!record) {
|
if (!record) {
|
||||||
|
|||||||
Reference in New Issue
Block a user