puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-31 05:32:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Keep one copy of the timeline when an agent is refreshed (#5286)

Browse Source 
* Clear the committed timeline when an agent refresh rehydrates from disk

Refreshing a loaded agent reloads it with `rehydrateFromDisk`, which wipes
the in-memory timeline so provider history can be re-streamed. It left the
committed timeline in place, and that is what `getTimelineRows` reads, so
every refresh appended a second full copy of the history: N refreshes, N
copies of every message.

Wipe both timelines on that path, the way the force-hydrate path already
does, so the re-streamed history replaces the old rows instead of stacking
on them.

* Replace the committed timeline from the replay instead of wiping it first

Deleting the committed rows on the reload path meant a history replay that
failed halfway left the agent with no transcript, and a retry appended a
fresh replay to whatever partial rows it had written.

Collect the whole replay before touching either store, then drop the rows a
previous hydration committed and write the new ones. A failed stream now
leaves the committed timeline exactly as it was, and this covers the replay
that runs when an agent is loaded after a daemon restart, which duplicated
the same way a refresh did.

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
This commit is contained in:
paseo-bot[bot]
2026-09-23 18:46:20 +02:00
committed by GitHub
parent c4771ca467
commit 897a0abb1a
2 changed files with 173 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-manager.test.ts
+131
View File
@@ -692,6 +692,137 @@ test("uses an injected timeline store without making it a production requirement
}
});
test("refreshing an agent replaces the injected timeline store instead of appending a second copy", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-rehydrate-"));
const store = new RecordingTimelineStore();
class HistorySession extends TestAgentSession {
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
yield {
type: "timeline",
provider: "codex",
item: { type: "user_message", text: "continue" },
};
yield {
type: "timeline",
provider: "codex",
item: { type: "assistant_message", text: "done" },
};
}
}
class HistoryClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new HistorySession(config);
}
override async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
return new HistorySession({ provider: "codex", cwd: config?.cwd ?? workdir });
}
}
const manager = new AgentManager({
clients: { codex: new HistoryClient() },
durableTimelineStore: store,
logger,
});
let agentId: string | null = null;
try {
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
agentId = agent.id;
// What session.handleRefreshAgentRequest does for a loaded agent, twice over.
for (let refresh = 0; refresh < 2; refresh += 1) {
await manager.reloadAgentSession(agent.id, undefined, { rehydrateFromDisk: true });
await manager.hydrateTimelineFromProvider(agent.id, { broadcast: true });
await manager.flush();
}
const rows = await manager.getTimelineRows(agent.id);
expect(rows.map((row) => row.item)).toEqual([
{ type: "user_message", text: "continue" },
{ type: "assistant_message", text: "done" },
]);
} finally {
if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
rmSync(workdir, { recursive: true, force: true });
}
});
test("a failed history replay leaves the committed timeline intact", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-replay-fail-"));
const store = new RecordingTimelineStore();
let failReplay = false;
class FlakyHistorySession extends TestAgentSession {
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
yield {
type: "timeline",
provider: "codex",
item: { type: "user_message", text: "continue" },
};
if (failReplay) {
throw new Error("history stream closed");
}
yield {
type: "timeline",
provider: "codex",
item: { type: "assistant_message", text: "done" },
};
}
}
class FlakyHistoryClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new FlakyHistorySession(config);
}
override async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
return new FlakyHistorySession({ provider: "codex", cwd: config?.cwd ?? workdir });
}
}
const manager = new AgentManager({
clients: { codex: new FlakyHistoryClient() },
durableTimelineStore: store,
logger,
});
let agentId: string | null = null;
try {
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
agentId = agent.id;
await manager.reloadAgentSession(agent.id, undefined, { rehydrateFromDisk: true });
await manager.hydrateTimelineFromProvider(agent.id, { broadcast: true });
await manager.flush();
const committed = await manager.getTimelineRows(agent.id);
failReplay = true;
await manager.reloadAgentSession(agent.id, undefined, { rehydrateFromDisk: true });
await expect(
manager.hydrateTimelineFromProvider(agent.id, { broadcast: true }),
).rejects.toThrow("history stream closed");
await manager.flush();
expect(await manager.getTimelineRows(agent.id)).toEqual(committed);
// A retry once the provider recovers replaces that history rather than stacking on it.
failReplay = false;
await manager.hydrateTimelineFromProvider(agent.id, { broadcast: true });
await manager.flush();
expect((await manager.getTimelineRows(agent.id)).map((row) => row.item)).toEqual([
{ type: "user_message", text: "continue" },
{ type: "assistant_message", text: "done" },
]);
} finally {
if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
rmSync(workdir, { recursive: true, force: true });
}
});
test("retries provider history hydration after a stream failure", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-retry-"));
let attempts = 0;
packages/server/src/server/agent/agent-manager.ts
+42 -26
View File
@@ -4025,23 +4025,17 @@ export class AgentManager {
broadcast: boolean | (() => boolean),
): Promise<void> {
const deferredBroadcast = typeof broadcast === "function";
const timelineEvents: Array<{
event: Extract<AgentStreamEvent, { type: "timeline" }>;
row: AgentTimelineRow;
}> = [];
const providerSubagentEvents: AgentManagerEvent[] = [];
const historyEvents: Extract<AgentStreamEvent, { type: "timeline" }>[] = [];
const historySubagentEvents: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [];
agent.historyPrimed = false;
try {
// Collect the whole replay before touching either store. A stream that fails
// halfway then leaves the committed timeline as it was, instead of a partial
// copy the next attempt would append to.
for await (const rawEvent of agent.session.streamHistory()) {
const event = limitAgentStreamEventContent(rawEvent);
if (event.type === "provider_subagent") {
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
const managerEvent: AgentManagerEvent = { type: "provider_subagent", event: update };
if (deferredBroadcast) {
providerSubagentEvents.push(managerEvent);
} else if (broadcast) {
this.dispatch(managerEvent);
}
historySubagentEvents.push(event);
continue;
}
if (event.type !== "timeline") {
@@ -4050,25 +4044,47 @@ export class AgentManager {
if (event.item.type === "user_message" && isSystemInjectedEnvelope(event.item.text)) {
continue;
}
const row = this.recordTimeline(
agent.id,
event.item,
event.timestamp ? { timestamp: event.timestamp } : undefined,
);
if (deferredBroadcast) {
timelineEvents.push({ event, row });
} else if (broadcast) {
this.dispatchStream(agent.id, event, {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
});
}
historyEvents.push(event);
}
} catch (error) {
this.logger.warn({ err: error, agentId: agent.id }, "Failed to hydrate provider history");
throw error;
}
// The replay is the timeline, so drop the rows a previous hydration committed.
// Keeping them would leave getTimelineRows reading one copy per hydration.
await this.deleteCommittedTimeline(agent.id);
const timelineEvents: Array<{
event: Extract<AgentStreamEvent, { type: "timeline" }>;
row: AgentTimelineRow;
}> = [];
const providerSubagentEvents: AgentManagerEvent[] = [];
for (const event of historySubagentEvents) {
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
const managerEvent: AgentManagerEvent = { type: "provider_subagent", event: update };
if (deferredBroadcast) {
providerSubagentEvents.push(managerEvent);
} else if (broadcast) {
this.dispatch(managerEvent);
}
}
for (const event of historyEvents) {
const row = this.recordTimeline(
agent.id,
event.item,
event.timestamp ? { timestamp: event.timestamp } : undefined,
);
if (deferredBroadcast) {
timelineEvents.push({ event, row });
} else if (broadcast) {
this.dispatchStream(agent.id, event, {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
});
}
}
agent.historyPrimed = true;
if (typeof broadcast !== "function" || !broadcast()) {