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

fix(server): preserve legacy prompt pagination

Browse Source
This commit is contained in:
Mohamed Boudra
2026-08-01 05:14:44 +02:00
parent 64be90766b
commit 1a4e510ef5
5 changed files with 159 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/runtime/host-runtime.test.ts
+25 -28
View File
@@ -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<Record<string, unknown>>);
vi.mock("@getpaseo/client/internal/daemon-client", async (importOriginal) => {
const actual = await importOriginal<typeof import("@getpaseo/client/internal/daemon-client")>();
return {
...actual,
DaemonClient: function DaemonClientDouble(options: Record<string, unknown>) {
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 () => {
packages/server/src/server/agent/agent-manager.ts
+35 -39
View File
@@ -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,
packages/server/src/server/agent/submitted-prompt-delivery.ts
+34 -15
View File
@@ -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,
};
}
packages/server/src/server/session.ts
+8 -8
View File
@@ -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,
packages/server/src/server/wire-compat.test.ts
+57 -26
View File
@@ -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", () => {