puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-07 17:04:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(server): check durable timeline prefix and preserve version-switch notices

Browse Source
This commit is contained in:
fdkevin
2026-11-07 18:50:47 +08:00
parent d765000f44
commit 6394cc2d0c
2 changed files with 94 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-manager.test.ts
+45 -3
View File
@@ -12650,8 +12650,10 @@ test("commits startup notices once on create and after restored history", async
},
};
const limitedToolOutput = limitAgentTimelineItemContent(toolOutput);
let activeNotice = notice;
let lastSession: NotifyingSession | null = null;
class NotifyingSession extends TestAgentSession {
readonly initialTimeline = [{ item: notice, timestamp: "2026-09-22T00:00:00.000Z" }];
readonly initialTimeline = [{ item: activeNotice, timestamp: "2026-09-22T00:00:00.000Z" }];
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
yield {
type: "timeline",
@@ -12670,10 +12672,14 @@ test("commits startup notices once on create and after restored history", async
}
const client = new (class extends TestAgentClient {
override async createSession(config: AgentSessionConfig) {
return new NotifyingSession(config);
const session = new NotifyingSession(config);
lastSession = session;
return session;
}
override async resumeSession() {
return new NotifyingSession({ provider: "codex", cwd: workdir });
const session = new NotifyingSession({ provider: "codex", cwd: workdir });
lastSession = session;
return session;
}
})();
const store = new RecordingTimelineStore();
@@ -12700,6 +12706,42 @@ test("commits startup notices once on create and after restored history", async
created.id,
);
expect((await manager.getTimelineRows(created.id)).map((row) => row.item)).toEqual([notice]);
const followUp: AgentTimelineItem = {
type: "user_message",
text: "after cold restore",
messageId: "msg-after-cold-restore",
};
lastSession!.pushEvent({ type: "timeline", provider: "codex", item: followUp });
await manager.reloadAgentSession(created.id);
expect((await manager.getTimelineRows(created.id)).map((row) => row.item)).toEqual([
notice,
followUp,
]);
const v1Notice: AgentTimelineItem = {
type: "notification",
level: "info",
message: "This chat uses OpenCode v1.",
};
const v2Notice: AgentTimelineItem = {
type: "notification",
level: "info",
message: "This chat uses OpenCode v2.",
};
activeNotice = v1Notice;
await manager.reloadAgentSession(created.id);
activeNotice = v2Notice;
await manager.reloadAgentSession(created.id);
activeNotice = v1Notice;
await manager.reloadAgentSession(created.id);
await manager.reloadAgentSession(created.id);
expect((await manager.getTimelineRows(created.id)).map((row) => row.item)).toEqual([
notice,
followUp,
v1Notice,
v2Notice,
v1Notice,
]);
activeNotice = notice;
const resumed = await manager.resumeAgentFromPersistence({
provider: "codex",
sessionId: "existing",
packages/server/src/server/agent/agent-manager.ts
+49 -20
View File
@@ -4029,11 +4029,11 @@ export class AgentManager {
// Legacy/imported chats need their existing history before startup rows.
await this.primeTimelineFromLegacyProviderHistory(managed, false, startupHistory);
} else {
await this.recordInitialTimelineIfMissing(
managed,
session.initialTimeline,
await this.recordInitialTimelineIfMissing({
agent: managed,
initialTimeline: session.initialTimeline,
durableTimelineHasRows,
);
});
}
this.refreshSessionPersistence(managed);
}
@@ -4226,26 +4226,30 @@ export class AgentManager {
};
}
private async recordInitialTimelineIfMissing(
agent: ActiveManagedAgent,
initialTimeline: NonNullable<AgentSession["initialTimeline"]>,
durableTimelineHasRows: boolean,
): Promise<void> {
const existingItems: AgentTimelineItem[] = [...this.timelineStore.getItems(agent.id)];
if (existingItems.length === 0 && durableTimelineHasRows && this.durableTimelineStore) {
private async recordInitialTimelineIfMissing(options: {
agent: ActiveManagedAgent;
initialTimeline: NonNullable<AgentSession["initialTimeline"]>;
durableTimelineHasRows: boolean;
}): Promise<void> {
const { agent, initialTimeline, durableTimelineHasRows } = options;
const inMemoryRows = this.timelineStore.getRows(agent.id);
const firstInMemorySeq = inMemoryRows[0]?.seq ?? Number.POSITIVE_INFINITY;
const existingItems: AgentTimelineItem[] = [];
if (durableTimelineHasRows && this.durableTimelineStore && firstInMemorySeq > 1) {
for (const row of await this.durableTimelineStore.getCommittedRows(agent.id)) {
existingItems.push(row.item);
if (row.seq < firstInMemorySeq) {
existingItems.push(row.item);
}
}
}
for (const row of inMemoryRows) {
existingItems.push(row.item);
}
for (const entry of buildImportedTimelineRows(initialTimeline)) {
const alreadyRecorded = existingItems.some(
(item) =>
item.type === "notification" &&
entry.item.type === "notification" &&
item.level === entry.item.level &&
item.message === entry.item.message,
);
if (alreadyRecorded) {
if (
entry.item.type === "notification" &&
isDuplicateInitialTimelineNotification({ existingItems, incoming: entry.item })
) {
continue;
}
this.recordTimeline(agent.id, entry.item, { timestamp: entry.timestamp });
@@ -5995,6 +5999,31 @@ export class AgentManager {
}
}
function isDuplicateInitialTimelineNotification(input: {
existingItems: readonly AgentTimelineItem[];
incoming: Extract<AgentTimelineItem, { type: "notification" }>;
}): boolean {
const { existingItems, incoming } = input;
const isOpenCodeRuntimeVersionNotice = /^This chat uses OpenCode v\d+\.$/.test(incoming.message);
if (isOpenCodeRuntimeVersionNotice) {
const lastVersionNotice = existingItems.findLast(
(item): item is Extract<AgentTimelineItem, { type: "notification" }> =>
item.type === "notification" && /^This chat uses OpenCode v\d+\.$/.test(item.message),
);
return (
lastVersionNotice !== undefined &&
lastVersionNotice.level === incoming.level &&
lastVersionNotice.message === incoming.message
);
}
return existingItems.some(
(item) =>
item.type === "notification" &&
item.level === incoming.level &&
item.message === incoming.message,
);
}
function matchesImportableSessionQuery(
session: ImportableProviderSession,
rawQuery: string | undefined,