From 1a4e510ef59376610e1d54d5409026a60ce51e93 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 1 Aug 2026 05:14:44 +0200 Subject: [PATCH] fix(server): preserve legacy prompt pagination --- packages/app/src/runtime/host-runtime.test.ts | 53 ++++++------ .../server/src/server/agent/agent-manager.ts | 74 ++++++++--------- .../server/agent/submitted-prompt-delivery.ts | 49 +++++++---- packages/server/src/server/session.ts | 16 ++-- .../server/src/server/wire-compat.test.ts | 83 +++++++++++++------ 5 files changed, 159 insertions(+), 116 deletions(-) diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 9773e02e8e..74cdf85316 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -8,7 +8,6 @@ import type { import type { ConnectionOffer } from "@getpaseo/protocol/connection-offer"; import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; import type { AgentPermissionRequest } from "@getpaseo/protocol/agent-types"; -import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities"; import type { HostConnection, HostProfile } from "@/types/host-connection"; import { useSessionStore, type Agent } from "@/stores/session-store"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; @@ -23,19 +22,6 @@ import { } from "./host-runtime"; import { ReplicaCache } from "./replica-cache"; -const capturedDefaultClientOptions = vi.hoisted(() => [] as Array>); - -vi.mock("@getpaseo/client/internal/daemon-client", async (importOriginal) => { - const actual = await importOriginal(); - return { - ...actual, - DaemonClient: function DaemonClientDouble(options: Record) { - capturedDefaultClientOptions.push(options); - return new FakeDaemonClient(); - }, - }; -}); - class FakeDaemonClient { private state: ConnectionState = { status: "idle" }; private listeners = new Set<(status: ConnectionState) => void>(); @@ -572,25 +558,36 @@ describe("HostRuntimeController", () => { expect(controller.getSnapshot().agentDirectoryStatus).toBe("initial_loading"); }); - it("passes resolved client configuration into created active clients", async () => { - capturedDefaultClientOptions.length = 0; - const values = new Map([["@paseo:client-id-v1", "cid_runtime_stable"]]); - (globalThis as { window?: unknown }).window = { - localStorage: { - getItem: (key: string) => values.get(key) ?? null, - setItem: (key: string, value: string) => values.set(key, value), + it("passes resolved client id into created active clients", async () => { + const host = makeHost({ + connections: [ + { + id: "direct:lan:6767", + type: "directTcp", + endpoint: "lan:6767", + }, + ], + }); + const seenClientIds: string[] = []; + const fakeClient = new FakeDaemonClient(); + const controller = new HostRuntimeController({ + host, + deps: { + createClient: ({ clientId }) => { + seenClientIds.push(clientId); + return fakeClient as unknown as DaemonClient; + }, + connectToDaemon: async () => { + throw new Error("probe unavailable"); + }, + getClientId: async () => "cid_runtime_stable", }, - }; - const controller = new HostRuntimeController({ host: makeHost() }); + }); await controller.activateConnection({ connectionId: "direct:lan:6767" }); - expect(capturedDefaultClientOptions[0]).toMatchObject({ - clientId: "cid_runtime_stable", - capabilities: { [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true }, - }); + expect(seenClientIds).toEqual(["cid_runtime_stable"]); expect(controller.getSnapshot().connectionStatus).toBe("online"); - await controller.stop(); }); it("keeps browser client lifecycle tied to the active host runtime client", async () => { diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index f068757ece..ea28126d36 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2189,7 +2189,9 @@ export class AgentManager { private applyActiveTurnTerminal( agent: ActiveManagedAgent, turnId?: string, + fromHistory = false, ): ActiveTurnTerminalDisposition { + if (fromHistory) return "stale"; if (!agent.activeTurnId) return "untracked"; if (turnId && agent.activeTurnId !== turnId) return "stale"; agent.activeTurnId = null; @@ -3421,7 +3423,7 @@ export class AgentManager { const identified = attachManagedTurnIdentity(agent, event, options?.fromHistory === true); event = identified.event; const eventTurnId = identified.turnId; - const isForegroundEvent = Boolean(eventTurnId && agent.activeForegroundTurnId === eventTurnId); + const isForegroundEvent = agent.activeForegroundTurnId === eventTurnId; this.traceHandleStreamEventStart(agent, event, eventTurnId, isForegroundEvent); if ( eventTurnId && @@ -3448,13 +3450,17 @@ export class AgentManager { this.agentStreamCoalescer.flushFor(agent.id); } - const terminalDisposition = this.applyStreamTerminalDisposition( - agent, - event, - eventTurnId, - options?.fromHistory === true, - isForegroundEvent, - ); + let terminalDisposition: ActiveTurnTerminalDisposition = "untracked"; + if (isTurnTerminalEvent(event)) { + terminalDisposition = this.applyActiveTurnTerminal( + agent, + eventTurnId, + options?.fromHistory === true, + ); + if (isForegroundEvent && terminalDisposition !== "stale") { + this.finalizeSubmittedPrompt(agent, event); + } + } const flags: StreamEventFlags = { shouldDispatchEvent: true, shouldNotifyWaiters: true }; @@ -3471,15 +3477,17 @@ export class AgentManager { await dispatchPromise; } - if (!options?.fromHistory && isTurnTerminalEvent(event)) { - this.runs.settleTerminalRun(agent.id, eventTurnId); - if (isForegroundEvent) { - this.finalizeForegroundTurn(agent, eventTurnId); + if (!options?.fromHistory) { + if (isTurnTerminalEvent(event)) { + this.runs.settleTerminalRun(agent.id, eventTurnId); + if (isForegroundEvent) { + this.finalizeForegroundTurn(agent, eventTurnId); + } } - } - if (!options?.fromHistory && flags.shouldDispatchEvent) { - this.dispatchStream(agent.id, event, { timestamp: new Date().toISOString() }); + if (flags.shouldDispatchEvent) { + this.dispatchStream(agent.id, event, { timestamp: new Date().toISOString() }); + } } this.traceHandleStreamEventEnd(agent, event, eventTurnId, flags); @@ -3508,21 +3516,6 @@ export class AgentManager { ); } - private applyStreamTerminalDisposition( - agent: ActiveManagedAgent, - event: AgentStreamEvent, - turnId: string | undefined, - fromHistory: boolean, - isForegroundEvent: boolean, - ): ActiveTurnTerminalDisposition { - if (!isTurnTerminalEvent(event)) return "untracked"; - const disposition = fromHistory ? "stale" : this.applyActiveTurnTerminal(agent, turnId); - if (isForegroundEvent && disposition !== "stale") { - this.finalizeSubmittedPrompt(agent, event); - } - return disposition; - } - private traceCoalescerBuffered( agent: ActiveManagedAgent, event: AgentStreamEvent, @@ -4012,15 +4005,18 @@ export class AgentManager { correlatedClientMessageId?: string, messageId?: string, ): void { - const pending = this.timelineStore - .getRows(agent.id) - .findLast( - (row) => - row.delivery === AWAITING_PROVIDER_IDENTITY_DELIVERY && row.item.type === "user_message", - ); - const clientMessageId = - correlatedClientMessageId ?? - (pending?.item.type === "user_message" ? pending.item.clientMessageId : undefined); + let clientMessageId = correlatedClientMessageId; + if (!clientMessageId) { + const pending = this.timelineStore + .getRows(agent.id) + .findLast( + (row) => + row.delivery === AWAITING_PROVIDER_IDENTITY_DELIVERY && + row.item.type === "user_message", + ); + clientMessageId = + pending?.item.type === "user_message" ? pending.item.clientMessageId : undefined; + } if (!clientMessageId) return; const row = this.timelineStore.finalizeSubmittedUserMessage( agent.id, diff --git a/packages/server/src/server/agent/submitted-prompt-delivery.ts b/packages/server/src/server/agent/submitted-prompt-delivery.ts index 1372f91645..693e0c3678 100644 --- a/packages/server/src/server/agent/submitted-prompt-delivery.ts +++ b/packages/server/src/server/agent/submitted-prompt-delivery.ts @@ -1,8 +1,10 @@ import type { AgentTimelineItem } from "./agent-sdk-types.js"; import type { + AgentTimelineFetchOptions, AgentTimelineFetchResult, AgentTimelineRowDelivery, } from "./agent-timeline-store-types.js"; +import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js"; export const AWAITING_PROVIDER_IDENTITY_DELIVERY = "awaiting_provider_identity" as const; @@ -21,30 +23,47 @@ export function shouldDeliverSubmittedPromptRow( export function projectSubmittedPromptTimeline( timeline: AgentTimelineFetchResult, - fullTimeline = timeline, - cursorSeq = 0, ): AgentTimelineFetchResult { - const pending = fullTimeline.rows.at(-1); + const pending = timeline.rows.at(-1); if ( !pending || - pending.seq !== fullTimeline.window.maxSeq || + pending.seq !== timeline.window.maxSeq || shouldDeliverSubmittedPromptRow(pending.delivery, pending.item, false) ) { return timeline; } - const visibleMaxSeq = pending.seq - 1; - const rows = timeline.rows.filter((row) => row.seq <= visibleMaxSeq); - const selectedEndSeq = rows.at(-1)?.seq ?? cursorSeq; + const rows = timeline.rows.slice(0, -1); + const minSeq = rows[0]?.seq ?? 0; + const maxSeq = rows.at(-1)?.seq ?? 0; return { ...timeline, - window: { - ...timeline.window, - minSeq: - visibleMaxSeq === 0 || timeline.window.minSeq > visibleMaxSeq ? 0 : timeline.window.minSeq, - maxSeq: visibleMaxSeq, - nextSeq: visibleMaxSeq + 1, - }, - hasNewer: timeline.hasNewer && selectedEndSeq < visibleMaxSeq, + window: { minSeq, maxSeq, nextSeq: maxSeq + 1 }, + hasOlder: false, + hasNewer: false, rows, }; } + +export function selectSubmittedPromptTimelinePage( + fullTimeline: AgentTimelineFetchResult, + controlTimeline: AgentTimelineFetchResult, + options: AgentTimelineFetchOptions, +): AgentTimelineFetchResult { + const timeline = new InMemoryAgentTimelineStore(); + timeline.initialize("compatibility-projection", { + epoch: fullTimeline.epoch, + rows: fullTimeline.rows, + nextSeq: fullTimeline.window.nextSeq, + }); + const selected = timeline.fetch("compatibility-projection", { + ...options, + ...(controlTimeline.reset ? { direction: "tail", cursor: undefined } : {}), + }); + return { + ...selected, + direction: controlTimeline.direction, + reset: controlTimeline.reset, + staleCursor: controlTimeline.staleCursor, + gap: controlTimeline.gap, + }; +} diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 13a6026db6..b338dd9a7c 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -78,6 +78,7 @@ import type { } from "./agent/agent-manager.js"; import { projectSubmittedPromptTimeline, + selectSubmittedPromptTimelinePage, shouldDeliverSubmittedPromptRow, } from "./agent/submitted-prompt-delivery.js"; import { createAgentCommand } from "./agent/create-agent/create.js"; @@ -6201,17 +6202,16 @@ export class Session { ? fetchedControlTimeline : this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0 }); } - const controlTimeline = - fetchedFullTimeline === undefined - ? fetchedControlTimeline - : projectSubmittedPromptTimeline( - fetchedControlTimeline, - fetchedFullTimeline, - cursor?.seq, - ); const fullTimeline = fetchedFullTimeline ? projectSubmittedPromptTimeline(fetchedFullTimeline) : undefined; + const controlTimeline = fullTimeline + ? selectSubmittedPromptTimelinePage(fullTimeline, fetchedControlTimeline, { + direction, + cursor, + limit: pageLimit, + }) + : fetchedControlTimeline; const selectedTimeline = this.selectTimelineProjection({ agentId: msg.agentId, projection, diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index daa1061acb..ce5b4d952c 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -491,20 +491,23 @@ describe("wire compatibility", () => { test("projects identity-pending prompt rows coherently for legacy fetches", async () => { const timestamp = "2026-08-01T00:00:00.000Z"; - const prompt = (options?: { delivery?: AgentTimelineRow["delivery"]; messageId?: string }) => ({ - seq: 1, + const prompt = ( + seq: number, + options?: { delivery?: AgentTimelineRow["delivery"]; messageId?: string }, + ) => ({ + seq, timestamp, ...(options?.delivery ? { delivery: options.delivery } : {}), item: { type: "user_message" as const, - text: "hello", - clientMessageId: "client-message-1", + text: `prompt ${seq}`, + clientMessageId: `client-message-${seq}`, ...(options?.messageId ? { messageId: options.messageId } : {}), }, }); const pending = await emitTimelineResponse({ - rows: [prompt({ delivery: "awaiting_provider_identity" })], + rows: [prompt(1, { delivery: "awaiting_provider_identity" })], request: { projection: "canonical" }, }); const pageShape = (response: typeof pending) => [ @@ -531,7 +534,7 @@ describe("wire compatibility", () => { { projection: "projected" as const }, ]) { const finalized = await emitTimelineResponse({ - rows: [prompt({ messageId: "provider-message-1" })], + rows: [prompt(1, { messageId: "provider-message-1" })], request, }); expect(pageShape(finalized)).toEqual([[1, 1, 2], 1, 1, false, false, [1]]); @@ -542,32 +545,60 @@ describe("wire compatibility", () => { } const daemonHandled = await emitTimelineResponse({ - rows: [prompt()], + rows: [prompt(1)], request: { projection: "canonical" }, }); expect(daemonHandled.payload.entries).toHaveLength(1); - const pageRows: AgentTimelineRow[] = [1, 2, 3].map((seq) => ({ - seq, - timestamp, - item: { - type: "user_message", - text: `prompt ${seq}`, - clientMessageId: `client-message-${seq}`, - }, - })); + const rowsWithPendingSuffix = [ + prompt(1), + prompt(2, { delivery: "awaiting_provider_identity" }), + ]; for (const projection of ["canonical", "projected"] as const) { - const page = await emitTimelineResponse({ - rows: pageRows, - request: { - projection, - direction: "before", - cursor: { epoch: "epoch-1", seq: 3 }, - limit: 1, - }, - }); - expect(pageShape(page)).toEqual([[1, 3, 4], 2, 2, true, true, [2]]); + for (const request of [ + { direction: "tail" as const, limit: 1 }, + { direction: "after" as const, cursor: { epoch: "epoch-1", seq: 0 }, limit: 1 }, + { direction: "after" as const, cursor: { epoch: "epoch-1", seq: 0 }, limit: 0 }, + { direction: "before" as const, cursor: { epoch: "epoch-1", seq: 2 }, limit: 1 }, + { direction: "before" as const, cursor: { epoch: "epoch-1", seq: 2 }, limit: 0 }, + ]) { + const page = await emitTimelineResponse({ + rows: rowsWithPendingSuffix, + request: { projection, ...request }, + }); + expect(pageShape(page)).toEqual([[1, 1, 2], 1, 1, false, false, [1]]); + } } + + for (const { rows, cursor, flags } of [ + { + rows: rowsWithPendingSuffix, + cursor: { epoch: "stale-epoch", seq: 0 }, + flags: [true, true, false], + }, + { + rows: [prompt(5), prompt(6, { delivery: "awaiting_provider_identity" })], + cursor: { epoch: "epoch-1", seq: 1 }, + flags: [true, false, true], + }, + ] as const) { + const page = await emitTimelineResponse({ + rows, + request: { projection: "canonical", direction: "after", cursor, limit: 1 }, + }); + expect([page.payload.reset, page.payload.staleCursor, page.payload.gap]).toEqual(flags); + expect(page.payload.entries.map((entry) => entry.seqStart)).toEqual([rows[0].seq]); + } + + const finalizedSuffix = await emitTimelineResponse({ + rows: [prompt(1), prompt(2, { messageId: "provider-message-2" })], + request: { projection: "canonical", direction: "tail", limit: 1 }, + }); + expect(finalizedSuffix.payload.entries.map((entry) => entry.seqStart)).toEqual([2]); + expect(finalizedSuffix.payload.entries[0]?.item).toMatchObject({ + type: "user_message", + messageId: "provider-message-2", + }); }); test("sub_agent tool-call payload still parses against the v0.1.65-beta.3 schema", () => {