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(timeline): gate prompt revisions by client capability

Browse Source
This commit is contained in:
Mohamed Boudra
2026-08-01 04:23:24 +02:00
parent 2b639f1f44
commit 7e757468a1
13 changed files with 661 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/timeline-sync.md
+8
View File
@@ -178,6 +178,14 @@ provider accepts the turn. A correlated provider echo enriches that same sequenc
`messageId`; it never appends a second source row. The row's identity and pagination position are
therefore independent of provider notification timing. A late echo is streamed as a revision at the
original sequence; clients apply it only to the already-covered row with the same client identity.
Clients that understand these same-sequence updates advertise the
`canonical_submitted_prompt_revisions` client capability. For older destinations, the daemon keeps
the canonical row globally but withholds a rewind-capable foreground prompt until provider identity
arrives, then delivers the enriched row once at its original sequence. Timeline fetches apply the
same destination projection: a trailing identity-pending suffix is temporarily outside the visible
window, while daemon-handled prompts and providers without the rewind identity contract remain
immediately visible.
An immediately interrupted turn remains
canonical even when the provider never emits its echo.
Content matching is limited to the dated compatibility path for daemon timelines created before
packages/app/src/runtime/host-runtime.test.ts
+8
View File
@@ -8,12 +8,14 @@ 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";
import { isAgentArchiving, setAgentArchiving } from "@/hooks/use-archive-agent";
import { queryClient } from "@/data/query-client";
import {
APP_CLIENT_CAPABILITIES,
HostRuntimeController,
HostRuntimeStore,
readInitialDaemonConnectionHint,
@@ -22,6 +24,12 @@ import {
} from "./host-runtime";
import { ReplicaCache } from "./replica-cache";
it("advertises canonical submitted prompt revision support", () => {
expect(APP_CLIENT_CAPABILITIES).toMatchObject({
[CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true,
});
});
class FakeDaemonClient {
private state: ConnectionState = { status: "idle" };
private listeners = new Set<(status: ConnectionState) => void>();
packages/app/src/runtime/host-runtime.ts
+6 -1
View File
@@ -456,6 +456,11 @@ function probeIntervalForConnection(
return PROBE_MAX_BACKOFF_MS;
}
export const APP_CLIENT_CAPABILITIES = {
[CLIENT_CAPS.selectiveAgentTimeline]: true,
[CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true,
} as const;
function createDefaultDeps(): HostRuntimeControllerDeps {
const browserHostAvailable =
typeof getDesktopHost()?.browser?.executeAutomationCommand === "function";
@@ -468,7 +473,7 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
}
: undefined;
const appCapabilities = {
[CLIENT_CAPS.selectiveAgentTimeline]: true,
...APP_CLIENT_CAPABILITIES,
...browserAutomationCapabilities,
};
packages/protocol/src/client-capabilities.ts
+3
View File
@@ -20,6 +20,9 @@ export const CLIENT_CAPS = {
providerSubagents: "provider_subagents",
// COMPAT(projectUpdates): added in v0.1.109, remove gate after 2027-01-15.
projectUpdates: "project_updates",
// COMPAT(canonicalSubmittedPromptRevisions): added in v0.2.6, remove gate after 2027-02-01.
// Capable clients reconcile same-sequence provider-identity revisions in canonical prompt rows.
canonicalSubmittedPromptRevisions: "canonical_submitted_prompt_revisions",
browserHost: "browser_host",
} as const;
packages/protocol/src/messages.client-capabilities.test.ts
+19
View File
@@ -0,0 +1,19 @@
import { describe, expect, it } from "vitest";
import { CLIENT_CAPS } from "./client-capabilities.js";
import { WSHelloMessageSchema } from "./messages.js";
describe("client capabilities", () => {
it("accepts canonical submitted prompt revision support in hello", () => {
const hello = WSHelloMessageSchema.parse({
type: "hello",
clientId: "client-capability-test",
clientType: "mobile",
protocolVersion: 1,
capabilities: {
[CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true,
},
});
expect(hello.capabilities?.[CLIENT_CAPS.canonicalSubmittedPromptRevisions]).toBe(true);
});
});
packages/protocol/src/messages.ts
+1
View File
@@ -5886,6 +5886,7 @@ export const WSHelloMessageSchema = z.object({
[CLIENT_CAPS.terminalReflowableSnapshot]: z.boolean().optional(),
[CLIENT_CAPS.providerSubagents]: z.boolean().optional(),
[CLIENT_CAPS.projectUpdates]: z.boolean().optional(),
[CLIENT_CAPS.canonicalSubmittedPromptRevisions]: z.boolean().optional(),
[CLIENT_CAPS.browserHost]: BrowserAutomationHostCapabilitySchema.optional(),
})
.passthrough()
packages/server/src/server/agent/agent-manager.test.ts
+87 -26
View File
@@ -8076,23 +8076,21 @@ test("provider user_message is recorded from the live stream", async () => {
class UnexpectedUserMessageSession extends TestAgentSession {
override async startTurn(): Promise<{ turnId: string }> {
const turnId = "turn-unexpected-1";
setTimeout(() => {
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
// Provider yields user_message (e.g., system continuation)
this.pushEvent({
type: "timeline",
provider: this.provider,
item: { type: "user_message", text: "continuation prompt" },
turnId,
});
this.pushEvent({
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "continuation reply" },
turnId,
});
this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
}, 0);
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
// Provider yields a staged user_message without client identity (e.g., system continuation).
this.pushEvent({
type: "timeline",
provider: this.provider,
item: { type: "user_message", text: "continuation prompt" },
turnId,
});
this.pushEvent({
type: "timeline",
provider: this.provider,
item: { type: "assistant_message", text: "continuation reply" },
turnId,
});
this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
return { turnId };
}
}
@@ -8130,6 +8128,9 @@ test("provider user_message is recorded from the live stream", async () => {
// Provider's user_message should be recorded (no canonical to dedup against)
expect(userMessages).toHaveLength(1);
expect(userMessages[0].text).toBe("continuation prompt");
expect(
manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 }).rows[0]?.delivery,
).toBeUndefined();
});
test("authoritative timeline includes provider-emitted submitted user prompt", async () => {
@@ -8138,13 +8139,20 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
const storage = new AgentStorage(storagePath, logger);
class SubmittedUserMessageSession extends TestAgentSession {
override readonly capabilities = {
...TEST_CAPABILITIES,
supportsRewindConversation: true,
};
private turnCount = 0;
override async startTurn(
prompt: AgentPromptInput,
options?: AgentRunOptions,
): Promise<{ turnId: string }> {
const turnId = "turn-submitted-user-message";
this.turnCount += 1;
const turnId = `turn-submitted-user-message-${this.turnCount}`;
const text = typeof prompt === "string" ? prompt : "";
setTimeout(() => {
const publishProviderIdentity = () => {
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
this.pushEvent({
type: "timeline",
@@ -8153,12 +8161,22 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
item: {
type: "user_message",
text,
messageId: "provider-message-1",
messageId: `provider-message-${this.turnCount}`,
clientMessageId: options?.clientMessageId,
},
});
if (this.turnCount === 1) {
this.pushEvent({
type: "timeline",
provider: this.provider,
turnId,
item: { type: "assistant_message", text: "provider output after identity" },
});
}
this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
}, 0);
};
if (this.turnCount === 1) setTimeout(publishProviderIdentity, 0);
else publishProviderIdentity();
return { turnId };
}
}
@@ -8209,6 +8227,7 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
type: event.event.type,
item: event.event.item,
seq: event.seq,
delivery: event.delivery,
}
: { type: event.event.type },
),
@@ -8222,6 +8241,7 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
clientMessageId: "msg-client-1",
},
seq: 1,
delivery: "awaiting_provider_identity",
},
{
type: "timeline",
@@ -8232,17 +8252,57 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
clientMessageId: "msg-client-1",
},
seq: 1,
delivery: "awaiting_provider_identity",
},
{
type: "timeline",
item: { type: "assistant_message", text: "provider output after identity" },
seq: 2,
delivery: undefined,
},
{ type: "turn_completed" },
]);
const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows;
expect(timeline.map((row) => row.item)).toEqual([
expect(timeline).toEqual([
{
type: "user_message",
text: "hello from composer",
messageId: "provider-message-1",
clientMessageId: "msg-client-1",
seq: 1,
timestamp: expect.any(String),
delivery: "awaiting_provider_identity",
item: {
type: "user_message",
text: "hello from composer",
messageId: "provider-message-1",
clientMessageId: "msg-client-1",
},
},
{
seq: 2,
timestamp: expect.any(String),
item: { type: "assistant_message", text: "provider output after identity" },
},
]);
events.length = 0;
await manager.runAgent(snapshot.id, "staged identity", {
clientMessageId: "msg-client-2",
});
expect(
events.flatMap((event) =>
event.type === "agent_stream" && event.event.type === "timeline"
? [{ seq: event.seq, delivery: event.delivery, item: event.event.item }]
: [],
),
).toEqual([
{
seq: 3,
delivery: undefined,
item: {
type: "user_message",
text: "staged identity",
messageId: "provider-message-2",
clientMessageId: "msg-client-2",
},
},
]);
} finally {
@@ -8324,6 +8384,7 @@ test("authoritative timeline records a daemon-handled submitted prompt before it
},
{ type: "assistant_message", text: "Handled by the daemon" },
]);
expect(timeline.every((row) => row.delivery === undefined)).toBe(true);
} finally {
await manager.flush().catch(() => undefined);
await storage.flush().catch(() => undefined);
packages/server/src/server/agent/agent-manager.ts
+39 -14
View File
@@ -61,6 +61,7 @@ import {
AgentStreamCoalescer,
} from "./agent-stream-coalescer.js";
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
import { AWAITING_PROVIDER_IDENTITY_DELIVERY } from "./submitted-prompt-delivery.js";
import { AgentRunState, type ForegroundTurnWaiter } from "./agent-run-state.js";
import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest";
import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js";
@@ -185,6 +186,7 @@ export type AgentManagerEvent =
seq?: number;
epoch?: string;
timestamp?: string;
delivery?: AgentTimelineRow["delivery"];
};
export type AgentSubscriber = (event: AgentManagerEvent) => void;
@@ -2070,12 +2072,14 @@ export class AgentManager {
{ type: "turn_started", provider: agent.provider, turnId },
{ timestamp: turnStartedAt.toISOString() },
);
const stagedSubmittedPromptEcho = pendingRun.stagedEvents.find(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" &&
event.item.type === "user_message" &&
event.item.clientMessageId === options?.clientMessageId,
);
const stagedSubmittedPromptEcho = options?.clientMessageId
? pendingRun.stagedEvents.find(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" &&
event.item.type === "user_message" &&
event.item.clientMessageId === options.clientMessageId,
)
: undefined;
if (options?.clientMessageId) {
this.recordSubmittedPrompt(
agent,
@@ -3932,15 +3936,30 @@ export class AgentManager {
}
this.touchUpdatedAt(agent);
agent.lastUserMessageAt = new Date();
this.recordAndDispatchTimelineItem(
const item: AgentTimelineItem = {
type: "user_message",
text: submittedPromptText(prompt),
clientMessageId,
...(messageId ? { messageId } : {}),
};
const capabilities = agent.session.capabilities;
const delivery =
messageId === undefined &&
(capabilities.supportsRewindConversation ||
capabilities.supportsRewindFiles ||
capabilities.supportsRewindBoth)
? AWAITING_PROVIDER_IDENTITY_DELIVERY
: undefined;
const row = this.recordTimeline(agent.id, item, delivery ? { delivery } : undefined);
this.dispatchStream(
agent.id,
{ type: "timeline", item, provider: agent.provider },
{
type: "user_message",
text: submittedPromptText(prompt),
clientMessageId,
...(messageId ? { messageId } : {}),
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
...(row.delivery ? { delivery: row.delivery } : {}),
},
agent.provider,
);
}
@@ -3972,6 +3991,7 @@ export class AgentManager {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
...(row.delivery ? { delivery: row.delivery } : {}),
},
);
}
@@ -4035,7 +4055,7 @@ export class AgentManager {
private recordTimeline(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string },
options?: { timestamp?: string; delivery?: AgentTimelineRow["delivery"] },
): AgentTimelineRow {
item = limitAgentTimelineItemContent(item);
const row = this.timelineStore.append(agentId, item, options);
@@ -4236,7 +4256,12 @@ export class AgentManager {
private dispatchStream(
agentId: string,
event: AgentStreamEvent,
metadata?: { seq?: number; epoch?: string; timestamp?: string },
metadata?: {
seq?: number;
epoch?: string;
timestamp?: string;
delivery?: AgentTimelineRow["delivery"];
},
): void {
if (event.type === "timeline") {
event = {
packages/server/src/server/agent/agent-timeline-store-types.ts
+3
View File
@@ -1,9 +1,12 @@
import type { AgentTimelineItem } from "./agent-sdk-types.js";
export type AgentTimelineRowDelivery = "awaiting_provider_identity";
export interface AgentTimelineRow {
seq: number;
timestamp: string;
item: AgentTimelineItem;
readonly delivery?: AgentTimelineRowDelivery;
}
export interface AgentTimelineCursor {
packages/server/src/server/agent/agent-timeline-store.ts
+2 -1
View File
@@ -260,13 +260,14 @@ export class InMemoryAgentTimelineStore {
append(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string },
options?: { timestamp?: string; delivery?: AgentTimelineRow["delivery"] },
): AgentTimelineRow {
const state = this.requireState(agentId);
const row: AgentTimelineRow = {
seq: state.nextSeq,
timestamp: options?.timestamp ?? new Date().toISOString(),
item,
...(options?.delivery ? { delivery: options.delivery } : {}),
};
state.nextSeq += 1;
state.rows.push(row);
packages/server/src/server/agent/submitted-prompt-delivery.ts
+63
View File
@@ -0,0 +1,63 @@
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import type {
AgentTimelineFetchResult,
AgentTimelineRow,
AgentTimelineRowDelivery,
} from "./agent-timeline-store-types.js";
export const AWAITING_PROVIDER_IDENTITY_DELIVERY = "awaiting_provider_identity" as const;
export function shouldDeliverSubmittedPromptRow(input: {
delivery: AgentTimelineRowDelivery | undefined;
item: AgentTimelineItem;
supportsRevisions: boolean;
}): boolean {
return (
input.supportsRevisions ||
input.delivery !== AWAITING_PROVIDER_IDENTITY_DELIVERY ||
input.item.type !== "user_message" ||
input.item.messageId !== undefined
);
}
export function findSubmittedPromptVisibleMaxSeq(
rows: readonly AgentTimelineRow[],
maxSeq: number,
): number {
let visibleMaxSeq = maxSeq;
for (let index = rows.length - 1; index >= 0; index -= 1) {
const row = rows[index];
if (
!row ||
row.seq !== visibleMaxSeq ||
shouldDeliverSubmittedPromptRow({
delivery: row.delivery,
item: row.item,
supportsRevisions: false,
})
) {
break;
}
visibleMaxSeq -= 1;
}
return visibleMaxSeq;
}
export function projectSubmittedPromptTimeline(input: {
timeline: AgentTimelineFetchResult;
visibleMaxSeq: number;
cursorSeq?: number;
}): AgentTimelineFetchResult {
const rows = input.timeline.rows.filter((row) => row.seq <= input.visibleMaxSeq);
const selectedEndSeq = rows.at(-1)?.seq ?? input.cursorSeq ?? 0;
return {
...input.timeline,
window: {
...input.timeline.window,
maxSeq: input.visibleMaxSeq,
nextSeq: input.visibleMaxSeq + 1,
},
hasNewer: input.timeline.hasNewer && selectedEndSeq < input.visibleMaxSeq,
rows,
};
}
packages/server/src/server/session.test.ts
+305 -1
View File
@@ -22,7 +22,11 @@ import { isSessionRpcAllowed, Session } from "./session.js";
import { DownloadTokenStore } from "./file-download/token-store.js";
import { StructuredAgentFallbackError } from "./agent/agent-response-loop.js";
import type { StoredAgentRecord } from "./agent/agent-storage.js";
import type { AgentManagerEvent } from "./agent/agent-manager.js";
import type {
AgentManagerEvent,
AgentTimelineFetchResult,
ManagedAgent,
} from "./agent/agent-manager.js";
import type { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
import { createPersistedProjectRecord } from "./workspace-registry.js";
import { deriveProjectKey } from "./project-key.js";
@@ -4880,6 +4884,306 @@ test("keeps selective delivery scoped per socket when a retained session also ha
]);
});
test("projects canonical submitted prompt revisions per retained socket capability", async () => {
const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = [];
const agentEventListeners: Array<(event: AgentManagerEvent) => void> = [];
const rewind = vi.fn().mockResolvedValue(undefined);
const session = createSessionForTest({
targetedMessages,
agentManager: {
rewind,
subscribe: vi.fn((listener: (event: AgentManagerEvent) => void) => {
agentEventListeners.push(listener);
return () => {};
}),
},
});
const legacySocket = {};
const capableSocket = {};
session.updateClientCapabilities(null, legacySocket);
session.updateClientCapabilities(
{ [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true },
capableSocket,
);
const listener = agentEventListeners[0];
if (!listener) throw new Error("Agent event listener was not installed");
listener({
type: "agent_stream",
agentId: "agent-revisions",
event: {
type: "timeline",
provider: "codex",
item: {
type: "user_message",
text: "submitted prompt",
clientMessageId: "client-message-1",
},
},
seq: 5,
epoch: "epoch-1",
delivery: "awaiting_provider_identity",
});
listener({
type: "agent_stream",
agentId: "agent-revisions",
event: {
type: "timeline",
provider: "codex",
item: {
type: "user_message",
text: "submitted prompt",
clientMessageId: "client-message-1",
messageId: "provider-message-1",
},
},
seq: 5,
epoch: "epoch-1",
delivery: "awaiting_provider_identity",
});
listener({
type: "agent_stream",
agentId: "agent-revisions",
event: {
type: "timeline",
provider: "codex",
item: {
type: "user_message",
text: "staged identity",
clientMessageId: "client-message-2",
messageId: "provider-message-2",
},
},
seq: 6,
epoch: "epoch-1",
delivery: "awaiting_provider_identity",
});
listener({
type: "agent_stream",
agentId: "agent-revisions",
event: {
type: "timeline",
provider: "codex",
item: {
type: "user_message",
text: "provider message without revision contract",
},
},
seq: 7,
epoch: "epoch-1",
});
const timelineFor = (source: object) =>
targetedMessages.flatMap(({ source: destination, message }) =>
destination === source &&
message.type === "agent_stream" &&
message.payload.event.type === "timeline"
? [
{
seq: message.payload.seq,
clientMessageId:
message.payload.event.item.type === "user_message"
? message.payload.event.item.clientMessageId
: undefined,
messageId:
message.payload.event.item.type === "user_message"
? message.payload.event.item.messageId
: undefined,
},
]
: [],
);
expect(timelineFor(capableSocket)).toEqual([
{ seq: 5, clientMessageId: "client-message-1", messageId: undefined },
{ seq: 5, clientMessageId: "client-message-1", messageId: "provider-message-1" },
{ seq: 6, clientMessageId: "client-message-2", messageId: "provider-message-2" },
{ seq: 7, clientMessageId: undefined, messageId: undefined },
]);
expect(timelineFor(legacySocket)).toEqual([
{ seq: 5, clientMessageId: "client-message-1", messageId: "provider-message-1" },
{ seq: 6, clientMessageId: "client-message-2", messageId: "provider-message-2" },
{ seq: 7, clientMessageId: undefined, messageId: undefined },
]);
await session.handleMessage(
{
type: "agent.rewind.request",
requestId: "rewind-provider-identity",
agentId: "agent-revisions",
messageId: "provider-message-1",
mode: "conversation",
},
legacySocket,
);
expect(rewind).toHaveBeenCalledWith("agent-revisions", "provider-message-1", "conversation");
});
test("projects a coherent legacy fetch window around pending provider identity", async () => {
const source = {};
const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = [];
const timestamp = "2026-08-01T00:00:00.000Z";
const agent = {
id: "agent-fetch-revisions",
provider: "codex",
cwd: "/tmp/agent-fetch-revisions",
capabilities: {
supportsStreaming: true,
supportsSessionPersistence: true,
supportsDynamicModes: false,
supportsMcpServers: false,
supportsReasoningStream: false,
supportsToolInvocations: false,
supportsRewindConversation: true,
},
config: { provider: "codex", cwd: "/tmp/agent-fetch-revisions" },
runtimeInfo: { provider: "codex", sessionId: "session-fetch-revisions" },
createdAt: new Date(timestamp),
updatedAt: new Date(timestamp),
availableModes: [],
currentModeId: null,
pendingPermissions: new Map(),
persistence: null,
historyPrimed: true,
lastUserMessageAt: new Date(timestamp),
activeTurnId: null,
activeTurnStartedAt: null,
lastUsage: undefined,
attention: { requiresAttention: false },
labels: {},
lifecycle: "idle",
activeForegroundTurnId: null,
} as unknown as ManagedAgent;
let timeline: AgentTimelineFetchResult = {
epoch: "epoch-fetch-revisions",
direction: "tail",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 1, maxSeq: 5, nextSeq: 6 },
hasOlder: true,
hasNewer: false,
rows: [
{
seq: 4,
timestamp,
item: { type: "assistant_message", text: "before submitted prompt" },
},
{
seq: 5,
timestamp,
delivery: "awaiting_provider_identity",
item: {
type: "user_message",
text: "submitted prompt",
clientMessageId: "client-message-fetch",
},
},
],
};
const session = createSessionForTest({
targetedMessages,
agentManager: {
waitForAgentClose: vi.fn().mockResolvedValue(undefined),
getAgent: vi.fn(() => agent),
fetchTimeline: vi.fn(() => timeline),
},
});
session.updateClientCapabilities(null, source);
await session.handleMessage(
{
type: "fetch_agent_timeline_request",
requestId: "fetch-before-provider-identity",
agentId: agent.id,
direction: "tail",
projection: "canonical",
limit: 20,
},
source,
);
const before = targetedMessages.find(
({ message }) =>
message.type === "fetch_agent_timeline_response" &&
message.payload.requestId === "fetch-before-provider-identity",
)?.message;
expect(before).toMatchObject({
type: "fetch_agent_timeline_response",
payload: {
requestId: "fetch-before-provider-identity",
error: null,
window: { minSeq: 1, maxSeq: 4, nextSeq: 5 },
endCursor: { epoch: "epoch-fetch-revisions", seq: 4 },
hasNewer: false,
entries: [
{
seqStart: 4,
seqEnd: 4,
item: { type: "assistant_message", text: "before submitted prompt" },
},
],
},
});
timeline = {
...timeline,
rows: [
timeline.rows[0]!,
{
...timeline.rows[1]!,
item: {
type: "user_message",
text: "submitted prompt",
clientMessageId: "client-message-fetch",
messageId: "provider-message-fetch",
},
},
],
};
targetedMessages.length = 0;
await session.handleMessage(
{
type: "fetch_agent_timeline_request",
requestId: "fetch-after-provider-identity",
agentId: agent.id,
direction: "tail",
projection: "canonical",
limit: 20,
},
source,
);
const after = targetedMessages.find(
({ message }) =>
message.type === "fetch_agent_timeline_response" &&
message.payload.requestId === "fetch-after-provider-identity",
)?.message;
expect(after).toMatchObject({
type: "fetch_agent_timeline_response",
payload: {
requestId: "fetch-after-provider-identity",
error: null,
window: { minSeq: 1, maxSeq: 5, nextSeq: 6 },
endCursor: { epoch: "epoch-fetch-revisions", seq: 5 },
hasNewer: false,
entries: [
{ seqStart: 4, seqEnd: 4 },
{
seqStart: 5,
seqEnd: 5,
item: {
type: "user_message",
text: "submitted prompt",
clientMessageId: "client-message-fetch",
messageId: "provider-message-fetch",
},
},
],
},
});
});
test("sends project updates only to capable sockets in a retained session", () => {
const messages: SessionOutboundMessage[] = [];
const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = [];
packages/server/src/server/session.ts
+117 -53
View File
@@ -76,6 +76,11 @@ import type {
AgentTimelineFetchResult,
ManagedAgent,
} from "./agent/agent-manager.js";
import {
findSubmittedPromptVisibleMaxSeq,
projectSubmittedPromptTimeline,
shouldDeliverSubmittedPromptRow,
} from "./agent/submitted-prompt-delivery.js";
import { createAgentCommand } from "./agent/create-agent/create.js";
import { resolveCreateAgentIntent, type CreateAgentIntent } from "./agent/create-agent/intent.js";
import {
@@ -1063,6 +1068,16 @@ export class Session {
serializedEvent: Extract<SessionOutboundMessage, { type: "agent_stream" }>["payload"]["event"],
): void {
if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) {
if (
serializedEvent.type === "timeline" &&
!shouldDeliverSubmittedPromptRow({
delivery: event.delivery,
item: serializedEvent.item,
supportsRevisions: this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions),
})
) {
return;
}
if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") {
this.emit({
type: "agent_attention_required",
@@ -1088,6 +1103,16 @@ export class Session {
for (const [source, capabilities] of this.clientCapabilitiesBySource) {
const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline);
if (
serializedEvent.type === "timeline" &&
!shouldDeliverSubmittedPromptRow({
delivery: event.delivery,
item: serializedEvent.item,
supportsRevisions: capabilities.has(CLIENT_CAPS.canonicalSubmittedPromptRevisions),
})
) {
continue;
}
if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") {
this.onMessageToSource(source, {
type: "agent_attention_required",
@@ -1881,7 +1906,7 @@ export class Session {
): Promise<void> | undefined {
switch (msg.type) {
case "fetch_agent_timeline_request":
return this.handleFetchAgentTimelineRequest(msg);
return this.handleFetchAgentTimelineRequest(msg, source);
case "agent.provider_subagents.list.request":
return this.handleProviderSubagentListRequest(msg);
case "agent.provider_subagents.timeline.get.request":
@@ -6095,13 +6120,22 @@ export class Session {
direction: AgentTimelineFetchDirection;
cursor?: AgentTimelineCursor;
pageLimit: number;
visibleMaxSeq?: number;
}): AgentTimelineProjectionSelection {
const timeline = this.shouldUseFullTimelineForProjectedPage({
const selectedTimeline = this.shouldUseFullTimelineForProjectedPage({
timeline: input.controlTimeline,
pageLimit: input.pageLimit,
})
? this.agentManager.fetchTimeline(input.agentId, { direction: "tail", limit: 0 })
: input.controlTimeline;
const timeline =
input.visibleMaxSeq === undefined
? selectedTimeline
: projectSubmittedPromptTimeline({
timeline: selectedTimeline,
visibleMaxSeq: input.visibleMaxSeq,
...(input.cursor ? { cursorSeq: input.cursor.seq } : {}),
});
const page = selectProjectedTimelinePage({
rows: timeline.rows,
bounds: timeline.window,
@@ -6127,6 +6161,7 @@ export class Session {
direction: AgentTimelineFetchDirection;
cursor?: AgentTimelineCursor;
pageLimit: number;
visibleMaxSeq?: number;
}): AgentTimelineProjectionSelection {
if (input.projection === "canonical") {
return this.selectCanonicalTimelineProjection({ timeline: input.controlTimeline });
@@ -6137,6 +6172,7 @@ export class Session {
private async handleFetchAgentTimelineRequest(
msg: Extract<SessionInboundMessage, { type: "fetch_agent_timeline_request" }>,
source?: object,
): Promise<void> {
const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail");
const projection: TimelineProjectionMode = msg.projection ?? "projected";
@@ -6157,11 +6193,28 @@ export class Session {
});
const agentPayload = await this.buildAgentPayload(snapshot);
const controlTimeline = this.agentManager.fetchTimeline(msg.agentId, {
const fetchedControlTimeline = this.agentManager.fetchTimeline(msg.agentId, {
direction,
cursor,
limit: pageLimit,
});
const supportsSubmittedPromptRevisions = source
? this.supportsForSource(CLIENT_CAPS.canonicalSubmittedPromptRevisions, source)
: this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions);
const visibleMaxSeq = supportsSubmittedPromptRevisions
? undefined
: findSubmittedPromptVisibleMaxSeq(
this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0 }).rows,
fetchedControlTimeline.window.maxSeq,
);
const controlTimeline =
visibleMaxSeq === undefined
? fetchedControlTimeline
: projectSubmittedPromptTimeline({
timeline: fetchedControlTimeline,
visibleMaxSeq,
...(cursor ? { cursorSeq: cursor.seq } : {}),
});
const selectedTimeline = this.selectTimelineProjection({
agentId: msg.agentId,
projection,
@@ -6169,6 +6222,7 @@ export class Session {
direction,
...(cursor ? { cursor } : {}),
pageLimit,
...(visibleMaxSeq !== undefined ? { visibleMaxSeq } : {}),
});
const startCursor =
selectedTimeline.startSeq !== null
@@ -6179,63 +6233,73 @@ export class Session {
? { epoch: selectedTimeline.timeline.epoch, seq: selectedTimeline.endSeq }
: null;
this.emit({
type: "fetch_agent_timeline_response",
payload: {
requestId: msg.requestId,
agentId: msg.agentId,
agent: agentPayload,
direction,
projection,
epoch: selectedTimeline.timeline.epoch,
reset: controlTimeline.reset,
staleCursor: controlTimeline.staleCursor,
gap: controlTimeline.gap,
window: selectedTimeline.timeline.window,
startCursor,
endCursor,
hasOlder: selectedTimeline.hasOlder,
hasNewer: selectedTimeline.hasNewer,
entries: selectedTimeline.entries.map((entry) => ({
provider: snapshot.provider,
item: entry.item,
timestamp: entry.timestamp,
seqStart: entry.seqStart,
seqEnd: entry.seqEnd,
sourceSeqRanges: entry.sourceSeqRanges,
collapsed: this.supports(CLIENT_CAPS.reasoningMergeEnum)
? entry.collapsed
: entry.collapsed.filter((value) => value !== "reasoning_merge"),
})),
error: null,
this.emitForSource(
{
type: "fetch_agent_timeline_response",
payload: {
requestId: msg.requestId,
agentId: msg.agentId,
agent: agentPayload,
direction,
projection,
epoch: selectedTimeline.timeline.epoch,
reset: controlTimeline.reset,
staleCursor: controlTimeline.staleCursor,
gap: controlTimeline.gap,
window: selectedTimeline.timeline.window,
startCursor,
endCursor,
hasOlder: selectedTimeline.hasOlder,
hasNewer: selectedTimeline.hasNewer,
entries: selectedTimeline.entries.map((entry) => ({
provider: snapshot.provider,
item: entry.item,
timestamp: entry.timestamp,
seqStart: entry.seqStart,
seqEnd: entry.seqEnd,
sourceSeqRanges: entry.sourceSeqRanges,
collapsed: (
source
? this.supportsForSource(CLIENT_CAPS.reasoningMergeEnum, source)
: this.supports(CLIENT_CAPS.reasoningMergeEnum)
)
? entry.collapsed
: entry.collapsed.filter((value) => value !== "reasoning_merge"),
})),
error: null,
},
},
});
source,
);
} catch (error) {
this.sessionLogger.error(
{ err: error, agentId: msg.agentId },
"Failed to handle fetch_agent_timeline_request",
);
this.emit({
type: "fetch_agent_timeline_response",
payload: {
requestId: msg.requestId,
agentId: msg.agentId,
agent: null,
direction,
projection,
epoch: "",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 0, maxSeq: 0, nextSeq: 0 },
startCursor: null,
endCursor: null,
hasOlder: false,
hasNewer: false,
entries: [],
error: error instanceof Error ? error.message : String(error),
this.emitForSource(
{
type: "fetch_agent_timeline_response",
payload: {
requestId: msg.requestId,
agentId: msg.agentId,
agent: null,
direction,
projection,
epoch: "",
reset: false,
staleCursor: false,
gap: false,
window: { minSeq: 0, maxSeq: 0, nextSeq: 0 },
startCursor: null,
endCursor: null,
hasOlder: false,
hasNewer: false,
entries: [],
error: error instanceof Error ? error.message : String(error),
},
},
});
source,
);
}
}