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

refactor(timeline): mint submitted prompt identity at acceptance

Browse Source 
Canonical prompt rows now keep one immutable wire identity while provider identity remains internal for rewind translation. This removes the unreleased same-sequence revision and legacy projection contract.
This commit is contained in:
Mohamed Boudra
2026-08-01 11:46:13 +02:00
parent 9b76c272b2
commit 20a542126e
12 changed files with 98 additions and 678 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/runtime/host-runtime.ts
-1
View File
@@ -469,7 +469,6 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
: undefined;
const appCapabilities = {
[CLIENT_CAPS.selectiveAgentTimeline]: true,
[CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true,
...browserAutomationCapabilities,
};
packages/protocol/src/client-capabilities.ts
-3
View File
@@ -20,9 +20,6 @@ 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.ts
-1
View File
@@ -5886,7 +5886,6 @@ 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
+56 -124
View File
@@ -4781,8 +4781,8 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the
(row) =>
row.item.type === "user_message" &&
row.item.clientMessageId === "replacement-client-message",
)?.delivery,
).toBe("awaiting_provider_identity");
)?.item,
).toMatchObject({ messageId: "replacement-client-message" });
allowSecondTurnToComplete.resolve();
await firstRunDrain;
await secondRunDrain;
@@ -8146,12 +8146,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 () => {
test("canonical submitted prompt keeps wire identity while rewind resolves provider identity", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-submitted-prompt-"));
const storagePath = join(workdir, "agents");
const storage = new AgentStorage(storagePath, logger);
@@ -8159,63 +8156,57 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
class SubmittedUserMessageSession extends TestAgentSession {
override readonly capabilities = {
...TEST_CAPABILITIES,
supportsRewindConversation: true,
supportsRewindFiles: true,
};
private turnCount = 0;
readonly rewindMessageIds: string[] = [];
override async startTurn(
prompt: AgentPromptInput,
options?: AgentRunOptions,
): Promise<{ turnId: string }> {
this.turnCount += 1;
const turnId = `turn-submitted-user-message-${this.turnCount}`;
const turnId = "turn-submitted-user-message";
const text = typeof prompt === "string" ? prompt : "";
const publishProviderIdentity = () => {
setTimeout(() => {
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
if (this.turnCount !== 3 && this.turnCount !== 5) {
this.pushEvent({
type: "timeline",
provider: this.provider,
turnId,
item: {
type: "user_message",
text,
...(this.turnCount === 4 ? {} : { 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" },
});
} else if (this.turnCount === 3) {
this.pushEvent({
type: "timeline",
provider: this.provider,
turnId,
item: { type: "assistant_message", text: "provider output without echo" },
});
}
this.pushEvent({
type: "timeline",
provider: this.provider,
turnId,
item: { type: "assistant_message", text: "output before provider echo" },
});
this.pushEvent({
type: "timeline",
provider: this.provider,
turnId,
item: {
type: "user_message",
text,
messageId: "provider-message-1",
clientMessageId: options?.clientMessageId,
},
});
this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
};
if (this.turnCount === 2) publishProviderIdentity();
else setTimeout(publishProviderIdentity, 0);
}, 0);
return { turnId };
}
override async revertFiles({ messageId }: { messageId: string }): Promise<void> {
this.rewindMessageIds.push(messageId);
}
}
class SubmittedUserMessageClient extends TestAgentClient {
session: SubmittedUserMessageSession | null = null;
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new SubmittedUserMessageSession(config);
this.session = new SubmittedUserMessageSession(config);
return this.session;
}
}
const client = new SubmittedUserMessageClient();
const manager = new AgentManager({
clients: { codex: new SubmittedUserMessageClient() },
clients: { codex: client },
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000402",
@@ -8231,7 +8222,6 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
{
type: item.type,
seq: event.seq,
delivery: event.delivery,
...(item.type === "user_message"
? {
text: item.text,
@@ -8248,106 +8238,47 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
workspaceId: undefined,
});
await manager.runAgent(
snapshot.id,
[
{
type: "text",
mimeType: "text/plain",
contextKind: "chat_history",
title: "Chat history",
text: "<chat-history-summary>hidden fork history</chat-history-summary>",
},
{ type: "text", text: "hello from composer" },
],
{ clientMessageId: "msg-client-1" },
);
await manager.runAgent(snapshot.id, "hello from composer", { clientMessageId: "msg-client-1" });
expect(streamEvents()).toEqual([
{ type: "turn_started" },
{
type: "user_message",
seq: 1,
delivery: "awaiting_provider_identity",
text: "hello from composer",
clientMessageId: "msg-client-1",
messageId: undefined,
messageId: "msg-client-1",
},
{
type: "user_message",
seq: 1,
delivery: undefined,
text: "hello from composer",
clientMessageId: "msg-client-1",
messageId: "provider-message-1",
},
{ type: "assistant_message", seq: 2, delivery: undefined },
{ type: "assistant_message", seq: 2 },
{ type: "turn_completed" },
]);
const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows;
expect(timeline.map((row) => [row.seq, row.delivery, row.item])).toEqual([
[
1,
undefined,
{
expect(timeline).toEqual([
{
seq: 1,
timestamp: expect.any(String),
providerMessageId: "provider-message-1",
item: {
type: "user_message",
text: "hello from composer",
messageId: "provider-message-1",
messageId: "msg-client-1",
clientMessageId: "msg-client-1",
},
],
[2, undefined, { type: "assistant_message", text: "provider output after identity" }],
]);
events.length = 0;
await manager.runAgent(snapshot.id, "staged identity", {
clientMessageId: "msg-client-2",
});
expect(streamEvents().filter((event) => event.type === "user_message")).toEqual([
},
{
type: "user_message",
seq: 3,
delivery: undefined,
text: "staged identity",
messageId: "provider-message-2",
clientMessageId: "msg-client-2",
seq: 2,
timestamp: expect.any(String),
item: { type: "assistant_message", text: "output before provider echo" },
},
]);
events.length = 0;
await manager.runAgent(snapshot.id, "output without echo", {
clientMessageId: "msg-client-3",
});
expect(streamEvents()).toEqual([
{ type: "turn_started" },
expect.objectContaining({
type: "user_message",
seq: 4,
delivery: "awaiting_provider_identity",
}),
expect.objectContaining({ type: "user_message", seq: 4, delivery: undefined }),
{ type: "assistant_message", seq: 5, delivery: undefined },
{ type: "turn_completed" },
await manager.rewind(snapshot.id, "msg-client-1", "files");
await manager.rewind(snapshot.id, "provider-native-message", "files");
expect(client.session?.rewindMessageIds).toEqual([
"provider-message-1",
"provider-native-message",
]);
for (const [prompt, clientMessageId, seq] of [
["echo without identity", "msg-client-4", 6],
["terminal without echo", "msg-client-5", 7],
] as const) {
events.length = 0;
await manager.runAgent(snapshot.id, prompt, { clientMessageId });
expect(streamEvents()).toEqual([
{ type: "turn_started" },
expect.objectContaining({
type: "user_message",
seq,
delivery: "awaiting_provider_identity",
}),
expect.objectContaining({ type: "user_message", seq, delivery: undefined }),
{ type: "turn_completed" },
]);
}
} finally {
await manager.flush().catch(() => undefined);
await storage.flush().catch(() => undefined);
@@ -8419,6 +8350,7 @@ test("authoritative timeline records a daemon-handled submitted prompt before it
type: "user_message",
text: "/handled",
clientMessageId: "msg-client-daemon-handled",
messageId: "msg-client-daemon-handled",
},
{ type: "assistant_message", text: "Handled by the daemon" },
]);
@@ -8429,10 +8361,10 @@ test("authoritative timeline records a daemon-handled submitted prompt before it
type: "user_message",
text: "/handled",
clientMessageId: "msg-client-daemon-handled",
messageId: "msg-client-daemon-handled",
},
{ 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
+29 -93
View File
@@ -61,7 +61,6 @@ 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";
@@ -186,7 +185,6 @@ export type AgentManagerEvent =
seq?: number;
epoch?: string;
timestamp?: string;
delivery?: AgentTimelineRow["delivery"];
};
export type AgentSubscriber = (event: AgentManagerEvent) => void;
@@ -2082,8 +2080,7 @@ export class AgentManager {
: undefined;
if (options?.clientMessageId) {
this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, {
awaitProviderIdentity: true,
messageId:
providerMessageId:
stagedSubmittedPromptEcho?.item.type === "user_message"
? stagedSubmittedPromptEcho.item.messageId
: undefined,
@@ -2493,11 +2490,23 @@ export class AgentManager {
const lock = this.runs.createPendingRun(agentId);
try {
const submittedRow = this.timelineStore
.getRows(agentId)
.find(
(row) =>
row.item.type === "user_message" &&
row.item.messageId === messageId &&
row.item.clientMessageId === messageId,
);
if (submittedRow && !submittedRow.providerMessageId) {
throw new Error("Cannot rewind before the provider acknowledges the submitted prompt");
}
const providerMessageId = submittedRow?.providerMessageId ?? messageId;
this.logger.info(
{ agentId, provider: agent.provider, messageId, mode },
"agent.rewind.start",
);
await invokeRewindCapability(agent.session, { messageId, mode });
await invokeRewindCapability(agent.session, { messageId: providerMessageId, mode });
if (mode !== "files") {
await this.hydrateTimelineFromProvider(agentId, { force: true, broadcast: true });
}
@@ -3433,13 +3442,6 @@ export class AgentManager {
return false;
}
this.finalizeSubmittedPromptBeforeTimeline(
agent,
event,
options?.fromHistory === true,
isForegroundEvent,
);
// Only update timestamp for live events, not history replay
if (!options?.fromHistory) {
this.touchUpdatedAt(agent);
@@ -3457,9 +3459,6 @@ export class AgentManager {
eventTurnId,
options?.fromHistory === true,
);
if (isForegroundEvent && terminalDisposition !== "stale") {
this.finalizeSubmittedPrompt(agent, event);
}
}
const flags: StreamEventFlags = { shouldDispatchEvent: true, shouldNotifyWaiters: true };
@@ -3680,7 +3679,7 @@ export class AgentManager {
if (
event.item.type === "user_message" &&
event.item.clientMessageId &&
this.reconcileSubmittedPromptEcho(agent, event, event.item)
this.reconcileSubmittedPromptEcho(agent, event.item)
) {
flags.shouldDispatchEvent = false;
flags.shouldNotifyWaiters = false;
@@ -3911,9 +3910,9 @@ export class AgentManager {
item: AgentTimelineItem,
provider: AgentProvider,
turnId?: string,
delivery?: AgentTimelineRow["delivery"],
options?: { providerMessageId?: string },
): AgentStreamEvent {
const row = this.recordTimeline(agentId, item, delivery ? { delivery } : undefined);
const row = this.recordTimeline(agentId, item, options);
const event: AgentStreamEvent = {
type: "timeline",
item,
@@ -3924,7 +3923,6 @@ export class AgentManager {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agentId),
timestamp: row.timestamp,
...(row.delivery ? { delivery: row.delivery } : {}),
});
if (
@@ -3946,7 +3944,7 @@ export class AgentManager {
agent: ActiveManagedAgent,
prompt: AgentPromptInput,
clientMessageId: string,
options?: { awaitProviderIdentity: boolean; messageId: string | undefined },
options?: { providerMessageId?: string },
): void {
if (this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId)) {
return;
@@ -3957,91 +3955,30 @@ export class AgentManager {
type: "user_message",
text: submittedPromptText(prompt),
clientMessageId,
...(options?.messageId ? { messageId: options.messageId } : {}),
messageId: clientMessageId,
};
const delivery =
options?.awaitProviderIdentity === true && options.messageId === undefined
? AWAITING_PROVIDER_IDENTITY_DELIVERY
: undefined;
this.recordAndDispatchTimelineItem(agent.id, item, agent.provider, undefined, delivery);
this.recordAndDispatchTimelineItem(agent.id, item, agent.provider, undefined, options);
}
private reconcileSubmittedPromptEcho(
agent: ActiveManagedAgent,
event: Extract<AgentStreamEvent, { type: "timeline" }>,
item: Extract<AgentTimelineItem, { type: "user_message" }>,
): AgentTimelineRow | null {
const { clientMessageId, messageId } = item;
if (!clientMessageId) return null;
const existing = this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId);
if (!existing || existing.item.type !== "user_message") return null;
this.finalizeSubmittedPrompt(agent, event, clientMessageId, messageId);
if (messageId) {
const enriched = this.timelineStore.enrichSubmittedUserMessage(
agent.id,
clientMessageId,
messageId,
);
if (enriched) this.enqueueDurableTimelineUpdate(agent.id, enriched);
}
return existing;
}
private finalizeSubmittedPromptBeforeTimeline(
agent: ActiveManagedAgent,
event: AgentStreamEvent,
fromHistory: boolean,
isForegroundEvent: boolean,
): void {
if (fromHistory || !isForegroundEvent || event.type !== "timeline") return;
const item = event.item;
if (item.type === "user_message") {
if (isSystemInjectedEnvelope(item.text)) return;
if (
item.clientMessageId &&
this.timelineStore.getSubmittedUserMessage(agent.id, item.clientMessageId)
) {
return;
}
}
this.finalizeSubmittedPrompt(agent, event);
}
private finalizeSubmittedPrompt(
agent: ActiveManagedAgent,
event: AgentStreamEvent,
correlatedClientMessageId?: string,
messageId?: string,
): void {
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,
clientMessageId,
messageId ? { messageId } : undefined,
);
if (!row) return;
this.enqueueDurableTimelineUpdate(agent.id, row);
const turnId = getAgentStreamEventTurnId(event);
this.dispatchStream(
agent.id,
{
type: "timeline",
item: row.item,
provider: event.provider,
...(turnId ? { turnId } : {}),
},
{
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
},
);
}
private async appendSystemErrorTimelineMessage(
agent: ActiveManagedAgent,
provider: AgentProvider,
@@ -4099,7 +4036,7 @@ export class AgentManager {
private recordTimeline(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string; delivery?: AgentTimelineRow["delivery"] },
options?: { timestamp?: string; providerMessageId?: string },
): AgentTimelineRow {
item = limitAgentTimelineItemContent(item);
const row = this.timelineStore.append(agentId, item, options);
@@ -4304,7 +4241,6 @@ export class AgentManager {
seq?: number;
epoch?: string;
timestamp?: string;
delivery?: AgentTimelineRow["delivery"];
},
): void {
if (event.type === "timeline") {
packages/server/src/server/agent/agent-timeline-store-types.ts
+1 -3
View File
@@ -1,12 +1,10 @@
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;
readonly providerMessageId?: string;
}
export interface AgentTimelineCursor {
packages/server/src/server/agent/agent-timeline-store.test.ts
-35
View File
@@ -2,41 +2,6 @@ import { describe, expect, it } from "vitest";
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
describe("InMemoryAgentTimelineStore", () => {
it("finalizes a submitted prompt without allocating a second sequence", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent-1", { epoch: "epoch-1" });
store.append(
"agent-1",
{
type: "user_message",
text: "Accepted prompt",
clientMessageId: "client-message-1",
},
{ delivery: "awaiting_provider_identity" },
);
const updated = store.finalizeSubmittedUserMessage("agent-1", "client-message-1", {
messageId: "provider-message-1",
});
expect(updated).toMatchObject({
seq: 1,
item: {
type: "user_message",
text: "Accepted prompt",
clientMessageId: "client-message-1",
messageId: "provider-message-1",
},
});
expect(updated?.delivery).toBeUndefined();
expect(store.getRows("agent-1")).toEqual([updated]);
expect(store.fetch("agent-1", { direction: "tail", limit: 1 }).window).toEqual({
minSeq: 1,
maxSeq: 1,
nextSeq: 2,
});
});
it("returns a bounded reset window when an after cursor is behind retained history", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent-1", {
packages/server/src/server/agent/agent-timeline-store.ts
+8 -12
View File
@@ -176,10 +176,10 @@ export class InMemoryAgentTimelineStore {
return row ? cloneRow(row) : null;
}
finalizeSubmittedUserMessage(
enrichSubmittedUserMessage(
agentId: string,
clientMessageId: string,
identity?: { messageId: string },
providerMessageId: string,
): AgentTimelineRow | null {
const state = this.requireState(agentId);
const index = state.rows.findIndex(
@@ -188,16 +188,12 @@ export class InMemoryAgentTimelineStore {
candidate.item.clientMessageId === clientMessageId,
);
const row = state.rows[index];
if (!row || row.item.type !== "user_message" || row.delivery !== "awaiting_provider_identity") {
if (!row || row.item.type !== "user_message") {
return null;
}
const finalized: AgentTimelineRow = {
seq: row.seq,
timestamp: row.timestamp,
item: identity ? { ...row.item, messageId: identity.messageId } : row.item,
};
state.rows[index] = finalized;
return cloneRow(finalized);
const enriched: AgentTimelineRow = { ...row, providerMessageId };
state.rows[index] = enriched;
return cloneRow(enriched);
}
getEpoch(agentId: string): string {
@@ -268,14 +264,14 @@ export class InMemoryAgentTimelineStore {
append(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string; delivery?: AgentTimelineRow["delivery"] },
options?: { timestamp?: string; providerMessageId?: string },
): AgentTimelineRow {
const state = this.requireState(agentId);
const row: AgentTimelineRow = {
seq: state.nextSeq,
timestamp: options?.timestamp ?? new Date().toISOString(),
item,
...(options?.delivery ? { delivery: options.delivery } : {}),
...(options?.providerMessageId ? { providerMessageId: options.providerMessageId } : {}),
};
state.nextSeq += 1;
state.rows.push(row);
packages/server/src/server/agent/submitted-prompt-delivery.ts
-69
View File
@@ -1,69 +0,0 @@
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;
export function shouldDeliverSubmittedPromptRow(
delivery: AgentTimelineRowDelivery | undefined,
item: AgentTimelineItem,
supportsRevisions: boolean,
): boolean {
return (
supportsRevisions ||
delivery !== AWAITING_PROVIDER_IDENTITY_DELIVERY ||
item.type !== "user_message" ||
item.messageId !== undefined
);
}
export function projectSubmittedPromptTimeline(
timeline: AgentTimelineFetchResult,
): AgentTimelineFetchResult {
const pending = timeline.rows.at(-1);
if (
!pending ||
pending.seq !== timeline.window.maxSeq ||
shouldDeliverSubmittedPromptRow(pending.delivery, pending.item, false)
) {
return timeline;
}
const rows = timeline.rows.slice(0, -1);
const minSeq = rows[0]?.seq ?? 0;
const maxSeq = rows.at(-1)?.seq ?? 0;
return {
...timeline,
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.test.ts
-173
View File
@@ -4880,179 +4880,6 @@ 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 session = createSessionForTest({
targetedMessages,
agentManager: {
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",
});
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",
});
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 },
]);
});
test.each([
[null, ["provider-message-1"]],
[{ [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true }, [undefined, "provider-message-1"]],
])("uses the negotiated fallback capability for prompt revisions", (capabilities, expectedIds) => {
const messages: SessionOutboundMessage[] = [];
let listener: ((event: AgentManagerEvent) => void) | undefined;
const session = createSessionForTest({
messages,
agentManager: {
subscribe: vi.fn((next: (event: AgentManagerEvent) => void) => {
listener = next;
return () => {};
}),
},
});
session.updateClientCapabilities(capabilities);
const item = {
type: "user_message" as const,
text: "submitted prompt",
clientMessageId: "client-message-1",
};
listener?.({
type: "agent_stream",
agentId: "agent-revisions",
event: { type: "timeline", provider: "codex", item },
seq: 1,
epoch: "epoch-1",
delivery: "awaiting_provider_identity",
});
listener?.({
type: "agent_stream",
agentId: "agent-revisions",
event: {
type: "timeline",
provider: "codex",
item: { ...item, messageId: "provider-message-1" },
},
seq: 1,
epoch: "epoch-1",
});
expect(
messages.flatMap((message) =>
message.type === "agent_stream" && message.payload.event.type === "timeline"
? [
message.payload.event.item.type === "user_message"
? message.payload.event.item.messageId
: undefined,
]
: [],
),
).toEqual(expectedIds);
});
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
+4 -50
View File
@@ -76,11 +76,6 @@ import type {
AgentTimelineFetchResult,
ManagedAgent,
} from "./agent/agent-manager.js";
import {
projectSubmittedPromptTimeline,
selectSubmittedPromptTimelinePage,
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 {
@@ -1064,16 +1059,6 @@ export class Session {
serializedEvent: Extract<SessionOutboundMessage, { type: "agent_stream" }>["payload"]["event"],
): void {
if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) {
if (
serializedEvent.type === "timeline" &&
!shouldDeliverSubmittedPromptRow(
event.delivery,
serializedEvent.item,
this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions),
)
) {
return;
}
if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") {
this.emit({
type: "agent_attention_required",
@@ -1099,16 +1084,6 @@ export class Session {
for (const [source, capabilities] of this.clientCapabilitiesBySource) {
const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline);
if (
serializedEvent.type === "timeline" &&
!shouldDeliverSubmittedPromptRow(
event.delivery,
serializedEvent.item,
capabilities.has(CLIENT_CAPS.canonicalSubmittedPromptRevisions),
)
) {
continue;
}
if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") {
this.onMessageToSource(source, {
type: "agent_attention_required",
@@ -6182,34 +6157,13 @@ export class Session {
cursor,
limit: pageLimit,
});
const supportsSubmittedPromptRevisions = source
? this.supportsForSource(CLIENT_CAPS.canonicalSubmittedPromptRevisions, source)
: this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions);
let fetchedFullTimeline: AgentTimelineFetchResult | undefined;
if (!supportsSubmittedPromptRevisions) {
fetchedFullTimeline =
direction === "tail" && pageLimit === 0
? fetchedControlTimeline
: this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0 });
}
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,
controlTimeline,
controlTimeline: fetchedControlTimeline,
direction,
...(cursor ? { cursor } : {}),
pageLimit,
...(fullTimeline ? { fullTimeline } : {}),
});
const startCursor =
selectedTimeline.startSeq !== null
@@ -6230,9 +6184,9 @@ export class Session {
direction,
projection,
epoch: selectedTimeline.timeline.epoch,
reset: controlTimeline.reset,
staleCursor: controlTimeline.staleCursor,
gap: controlTimeline.gap,
reset: fetchedControlTimeline.reset,
staleCursor: fetchedControlTimeline.staleCursor,
gap: fetchedControlTimeline.gap,
window: selectedTimeline.timeline.window,
startCursor,
endCursor,
packages/server/src/server/wire-compat.test.ts
-114
View File
@@ -404,7 +404,6 @@ describe("wire compatibility", () => {
protocolVersion: 1,
capabilities: {
[CLIENT_CAPS.projectUpdates]: true,
[CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true,
},
});
@@ -422,7 +421,6 @@ describe("wire compatibility", () => {
protocolVersion: 1,
capabilities: {
project_updates: true,
canonical_submitted_prompt_revisions: true,
},
},
]);
@@ -489,118 +487,6 @@ describe("wire compatibility", () => {
expect(currentParsed.payload.entries[0]?.collapsed).toContain("reasoning_merge");
});
test("projects identity-pending prompt rows coherently for legacy fetches", async () => {
const timestamp = "2026-08-01T00:00:00.000Z";
const prompt = (
seq: number,
options?: { delivery?: AgentTimelineRow["delivery"]; messageId?: string },
) => ({
seq,
timestamp,
...(options?.delivery ? { delivery: options.delivery } : {}),
item: {
type: "user_message" as const,
text: `prompt ${seq}`,
clientMessageId: `client-message-${seq}`,
...(options?.messageId ? { messageId: options.messageId } : {}),
},
});
const pending = await emitTimelineResponse({
rows: [prompt(1, { delivery: "awaiting_provider_identity" })],
request: { projection: "canonical" },
});
const pageShape = (response: typeof pending) => [
[
response.payload.window.minSeq,
response.payload.window.maxSeq,
response.payload.window.nextSeq,
],
response.payload.startCursor?.seq ?? null,
response.payload.endCursor?.seq ?? null,
response.payload.hasOlder,
response.payload.hasNewer,
response.payload.entries.map((entry) => entry.seqStart),
];
expect(pageShape(pending)).toEqual([[0, 0, 1], null, null, false, false, []]);
for (const request of [
{ projection: "canonical" as const },
{
projection: "canonical" as const,
direction: "after" as const,
cursor: { epoch: "epoch-1", seq: 0 },
},
{ projection: "projected" as const },
]) {
const finalized = await emitTimelineResponse({
rows: [prompt(1, { messageId: "provider-message-1" })],
request,
});
expect(pageShape(finalized)).toEqual([[1, 1, 2], 1, 1, false, false, [1]]);
expect(finalized.payload.entries[0]?.item).toMatchObject({
type: "user_message",
messageId: "provider-message-1",
});
}
const daemonHandled = await emitTimelineResponse({
rows: [prompt(1)],
request: { projection: "canonical" },
});
expect(daemonHandled.payload.entries).toHaveLength(1);
const rowsWithPendingSuffix = [
prompt(1),
prompt(2, { delivery: "awaiting_provider_identity" }),
];
for (const projection of ["canonical", "projected"] as const) {
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", () => {
const parsed = LegacySubAgentToolCallSchema.parse({
type: "tool_call",