mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-26 03:03:04 +00:00
fix(timeline): finalize canonical prompt delivery
Treat provider identity as a single stored-row transition so legacy clients receive one deliverable canonical prompt. Release identity-pending rows before later foreground output or the current terminal while preserving stale-turn isolation.
This commit is contained in:
Mohamed Boudra
@@ -181,13 +181,10 @@ original sequence; clients apply it only to the already-covered row with the sam
|
||||
|
||||
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.
|
||||
the canonical row globally but withholds an identity-awaiting foreground prompt until provider identity
|
||||
arrives or the turn produces later output or a terminal. Finalization optionally adds provider
|
||||
identity, removes the internal delivery marker, and redispatches the same stored sequence. Timeline
|
||||
fetches hide only that trailing marked row; daemon-handled prompts are never marked.
|
||||
Content matching is limited to the dated compatibility path for daemon timelines created before
|
||||
that field existed. Canonical ingestion may match only an explicit unreconciled local candidate;
|
||||
the draft-create handoff is the one boundary that also permits the legacy canonical twin to have
|
||||
|
||||
@@ -15,7 +15,6 @@ 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,
|
||||
@@ -24,10 +23,17 @@ 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,
|
||||
});
|
||||
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 {
|
||||
@@ -566,36 +572,25 @@ describe("HostRuntimeController", () => {
|
||||
expect(controller.getSnapshot().agentDirectoryStatus).toBe("initial_loading");
|
||||
});
|
||||
|
||||
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",
|
||||
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),
|
||||
},
|
||||
});
|
||||
};
|
||||
const controller = new HostRuntimeController({ host: makeHost() });
|
||||
|
||||
await controller.activateConnection({ connectionId: "direct:lan:6767" });
|
||||
|
||||
expect(seenClientIds).toEqual(["cid_runtime_stable"]);
|
||||
expect(capturedDefaultClientOptions[0]).toMatchObject({
|
||||
clientId: "cid_runtime_stable",
|
||||
capabilities: { [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true },
|
||||
});
|
||||
expect(controller.getSnapshot().connectionStatus).toBe("online");
|
||||
await controller.stop();
|
||||
});
|
||||
|
||||
it("keeps browser client lifecycle tied to the active host runtime client", async () => {
|
||||
|
||||
@@ -456,11 +456,6 @@ 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";
|
||||
@@ -473,7 +468,8 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
|
||||
}
|
||||
: undefined;
|
||||
const appCapabilities = {
|
||||
...APP_CLIENT_CAPABILITIES,
|
||||
[CLIENT_CAPS.selectiveAgentTimeline]: true,
|
||||
[CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true,
|
||||
...browserAutomationCapabilities,
|
||||
};
|
||||
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -4633,6 +4633,7 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the
|
||||
const interruptStarted = deferred<void>();
|
||||
const allowInterruptToFinish = deferred<void>();
|
||||
const allowSecondStartToResolve = deferred<void>();
|
||||
const allowSecondTurnToComplete = deferred<void>();
|
||||
let capturedSession: StaleReplacementSession | null = null;
|
||||
|
||||
class StaleReplacementSession extends TestAgentSession {
|
||||
@@ -4651,8 +4652,9 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the
|
||||
|
||||
secondStartEntered.resolve();
|
||||
await allowSecondStartToResolve.promise;
|
||||
setTimeout(() => {
|
||||
setTimeout(async () => {
|
||||
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
|
||||
await allowSecondTurnToComplete.promise;
|
||||
this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
|
||||
}, 0);
|
||||
return { turnId };
|
||||
@@ -4709,7 +4711,9 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the
|
||||
{ agentId: snapshot.id, replayState: false },
|
||||
);
|
||||
|
||||
const firstRun = manager.streamAgent(snapshot.id, "first run");
|
||||
const firstRun = manager.streamAgent(snapshot.id, "first run", {
|
||||
clientMessageId: "first-client-message",
|
||||
});
|
||||
const firstRunDrain = (async () => {
|
||||
for await (const _event of firstRun) {
|
||||
// Drain events so lifecycle updates are applied.
|
||||
@@ -4720,7 +4724,9 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the
|
||||
|
||||
const replaceUpdatesStart = stateUpdates.length;
|
||||
const beforeReplaceUpdatedAt = manager.getAgent(snapshot.id)?.updatedAt.getTime() ?? 0;
|
||||
const secondRunPromise = manager.replaceAgentRun(snapshot.id, "replacement run");
|
||||
const secondRunPromise = manager.replaceAgentRun(snapshot.id, "replacement run", {
|
||||
clientMessageId: "replacement-client-message",
|
||||
});
|
||||
|
||||
await interruptStarted.promise;
|
||||
const replacementUpdates = stateUpdates.slice(replaceUpdatesStart);
|
||||
@@ -4766,6 +4772,18 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the
|
||||
allowSecondStartToResolve.resolve();
|
||||
|
||||
await replacementStart;
|
||||
capturedSession!.pushEvent({ type: "turn_completed", provider: "codex", turnId: "turn-1" });
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
expect(
|
||||
manager
|
||||
.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 })
|
||||
.rows.find(
|
||||
(row) =>
|
||||
row.item.type === "user_message" &&
|
||||
row.item.clientMessageId === "replacement-client-message",
|
||||
)?.delivery,
|
||||
).toBe("awaiting_provider_identity");
|
||||
allowSecondTurnToComplete.resolve();
|
||||
await firstRunDrain;
|
||||
await secondRunDrain;
|
||||
unsubscribe();
|
||||
@@ -8154,17 +8172,19 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
|
||||
const text = typeof prompt === "string" ? prompt : "";
|
||||
const publishProviderIdentity = () => {
|
||||
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
|
||||
this.pushEvent({
|
||||
type: "timeline",
|
||||
provider: this.provider,
|
||||
turnId,
|
||||
item: {
|
||||
type: "user_message",
|
||||
text,
|
||||
messageId: `provider-message-${this.turnCount}`,
|
||||
clientMessageId: options?.clientMessageId,
|
||||
},
|
||||
});
|
||||
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",
|
||||
@@ -8172,11 +8192,18 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
|
||||
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: "turn_completed", provider: this.provider, turnId });
|
||||
};
|
||||
if (this.turnCount === 1) setTimeout(publishProviderIdentity, 0);
|
||||
else publishProviderIdentity();
|
||||
if (this.turnCount === 2) publishProviderIdentity();
|
||||
else setTimeout(publishProviderIdentity, 0);
|
||||
return { turnId };
|
||||
}
|
||||
}
|
||||
@@ -8195,6 +8222,26 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
|
||||
});
|
||||
const events: AgentManagerEvent[] = [];
|
||||
manager.subscribe((event) => events.push(event), { replayState: false });
|
||||
const streamEvents = () =>
|
||||
events.flatMap((event) => {
|
||||
if (event.type !== "agent_stream") return [];
|
||||
if (event.event.type !== "timeline") return [{ type: event.event.type }];
|
||||
const item = event.event.item;
|
||||
return [
|
||||
{
|
||||
type: item.type,
|
||||
seq: event.seq,
|
||||
delivery: event.delivery,
|
||||
...(item.type === "user_message"
|
||||
? {
|
||||
text: item.text,
|
||||
clientMessageId: item.clientMessageId,
|
||||
messageId: item.messageId,
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
];
|
||||
});
|
||||
|
||||
try {
|
||||
const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
@@ -8216,95 +8263,91 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a
|
||||
{ clientMessageId: "msg-client-1" },
|
||||
);
|
||||
|
||||
const liveEvents = events.filter(
|
||||
(event): event is Extract<AgentManagerEvent, { type: "agent_stream" }> =>
|
||||
event.type === "agent_stream" && event.agentId === snapshot.id,
|
||||
);
|
||||
expect(
|
||||
liveEvents.map((event) =>
|
||||
event.event.type === "timeline"
|
||||
? {
|
||||
type: event.event.type,
|
||||
item: event.event.item,
|
||||
seq: event.seq,
|
||||
delivery: event.delivery,
|
||||
}
|
||||
: { type: event.event.type },
|
||||
),
|
||||
).toEqual([
|
||||
expect(streamEvents()).toEqual([
|
||||
{ type: "turn_started" },
|
||||
{
|
||||
type: "timeline",
|
||||
item: {
|
||||
type: "user_message",
|
||||
text: "hello from composer",
|
||||
clientMessageId: "msg-client-1",
|
||||
},
|
||||
type: "user_message",
|
||||
seq: 1,
|
||||
delivery: "awaiting_provider_identity",
|
||||
text: "hello from composer",
|
||||
clientMessageId: "msg-client-1",
|
||||
messageId: undefined,
|
||||
},
|
||||
{
|
||||
type: "timeline",
|
||||
item: {
|
||||
type: "user_message",
|
||||
text: "hello from composer",
|
||||
messageId: "provider-message-1",
|
||||
clientMessageId: "msg-client-1",
|
||||
},
|
||||
type: "user_message",
|
||||
seq: 1,
|
||||
delivery: "awaiting_provider_identity",
|
||||
},
|
||||
{
|
||||
type: "timeline",
|
||||
item: { type: "assistant_message", text: "provider output after identity" },
|
||||
seq: 2,
|
||||
delivery: undefined,
|
||||
text: "hello from composer",
|
||||
clientMessageId: "msg-client-1",
|
||||
messageId: "provider-message-1",
|
||||
},
|
||||
{ type: "assistant_message", seq: 2, delivery: undefined },
|
||||
{ type: "turn_completed" },
|
||||
]);
|
||||
|
||||
const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows;
|
||||
expect(timeline).toEqual([
|
||||
{
|
||||
seq: 1,
|
||||
timestamp: expect.any(String),
|
||||
delivery: "awaiting_provider_identity",
|
||||
item: {
|
||||
expect(timeline.map((row) => [row.seq, row.delivery, row.item])).toEqual([
|
||||
[
|
||||
1,
|
||||
undefined,
|
||||
{
|
||||
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" },
|
||||
},
|
||||
],
|
||||
[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(
|
||||
events.flatMap((event) =>
|
||||
event.type === "agent_stream" && event.event.type === "timeline"
|
||||
? [{ seq: event.seq, delivery: event.delivery, item: event.event.item }]
|
||||
: [],
|
||||
),
|
||||
).toEqual([
|
||||
expect(streamEvents().filter((event) => event.type === "user_message")).toEqual([
|
||||
{
|
||||
type: "user_message",
|
||||
seq: 3,
|
||||
delivery: undefined,
|
||||
item: {
|
||||
type: "user_message",
|
||||
text: "staged identity",
|
||||
messageId: "provider-message-2",
|
||||
clientMessageId: "msg-client-2",
|
||||
},
|
||||
text: "staged identity",
|
||||
messageId: "provider-message-2",
|
||||
clientMessageId: "msg-client-2",
|
||||
},
|
||||
]);
|
||||
|
||||
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" },
|
||||
]);
|
||||
|
||||
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);
|
||||
@@ -8319,6 +8362,11 @@ test("authoritative timeline records a daemon-handled submitted prompt before it
|
||||
const commandCompleted = deferred<void>();
|
||||
|
||||
class DaemonHandledPromptSession extends TestAgentSession {
|
||||
override readonly capabilities = {
|
||||
...TEST_CAPABILITIES,
|
||||
supportsRewindConversation: true,
|
||||
};
|
||||
|
||||
override tryHandleOutOfBand(prompt: AgentPromptInput) {
|
||||
if (prompt !== "/handled") return null;
|
||||
return {
|
||||
|
||||
@@ -2081,14 +2081,13 @@ export class AgentManager {
|
||||
)
|
||||
: undefined;
|
||||
if (options?.clientMessageId) {
|
||||
this.recordSubmittedPrompt(
|
||||
agent,
|
||||
prompt,
|
||||
options.clientMessageId,
|
||||
stagedSubmittedPromptEcho?.item.type === "user_message"
|
||||
? stagedSubmittedPromptEcho.item.messageId
|
||||
: undefined,
|
||||
);
|
||||
this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, {
|
||||
awaitProviderIdentity: true,
|
||||
messageId:
|
||||
stagedSubmittedPromptEcho?.item.type === "user_message"
|
||||
? stagedSubmittedPromptEcho.item.messageId
|
||||
: undefined,
|
||||
});
|
||||
}
|
||||
for (const stagedEvent of pendingRun.stagedEvents.splice(0)) {
|
||||
const isAcceptedTurnStart =
|
||||
@@ -3432,6 +3431,13 @@ 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);
|
||||
@@ -3442,12 +3448,13 @@ export class AgentManager {
|
||||
this.agentStreamCoalescer.flushFor(agent.id);
|
||||
}
|
||||
|
||||
let terminalDisposition: ActiveTurnTerminalDisposition = "untracked";
|
||||
if (isTurnTerminalEvent(event)) {
|
||||
terminalDisposition = options?.fromHistory
|
||||
? "stale"
|
||||
: this.applyActiveTurnTerminal(agent, eventTurnId);
|
||||
}
|
||||
const terminalDisposition = this.applyStreamTerminalDisposition(
|
||||
agent,
|
||||
event,
|
||||
eventTurnId,
|
||||
options?.fromHistory === true,
|
||||
isForegroundEvent,
|
||||
);
|
||||
|
||||
const flags: StreamEventFlags = { shouldDispatchEvent: true, shouldNotifyWaiters: true };
|
||||
|
||||
@@ -3501,6 +3508,21 @@ 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,
|
||||
@@ -3896,8 +3918,9 @@ export class AgentManager {
|
||||
item: AgentTimelineItem,
|
||||
provider: AgentProvider,
|
||||
turnId?: string,
|
||||
delivery?: AgentTimelineRow["delivery"],
|
||||
): AgentStreamEvent {
|
||||
const row = this.recordTimeline(agentId, item);
|
||||
const row = this.recordTimeline(agentId, item, delivery ? { delivery } : undefined);
|
||||
const event: AgentStreamEvent = {
|
||||
type: "timeline",
|
||||
item,
|
||||
@@ -3908,6 +3931,7 @@ export class AgentManager {
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agentId),
|
||||
timestamp: row.timestamp,
|
||||
...(row.delivery ? { delivery: row.delivery } : {}),
|
||||
});
|
||||
|
||||
if (
|
||||
@@ -3929,7 +3953,7 @@ export class AgentManager {
|
||||
agent: ActiveManagedAgent,
|
||||
prompt: AgentPromptInput,
|
||||
clientMessageId: string,
|
||||
messageId?: string,
|
||||
options?: { awaitProviderIdentity: boolean; messageId: string | undefined },
|
||||
): void {
|
||||
if (this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId)) {
|
||||
return;
|
||||
@@ -3940,27 +3964,13 @@ export class AgentManager {
|
||||
type: "user_message",
|
||||
text: submittedPromptText(prompt),
|
||||
clientMessageId,
|
||||
...(messageId ? { messageId } : {}),
|
||||
...(options?.messageId ? { messageId: options.messageId } : {}),
|
||||
};
|
||||
const capabilities = agent.session.capabilities;
|
||||
const delivery =
|
||||
messageId === undefined &&
|
||||
(capabilities.supportsRewindConversation ||
|
||||
capabilities.supportsRewindFiles ||
|
||||
capabilities.supportsRewindBoth)
|
||||
options?.awaitProviderIdentity === true && options.messageId === undefined
|
||||
? 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 },
|
||||
{
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agent.id),
|
||||
timestamp: row.timestamp,
|
||||
...(row.delivery ? { delivery: row.delivery } : {}),
|
||||
},
|
||||
);
|
||||
this.recordAndDispatchTimelineItem(agent.id, item, agent.provider, undefined, delivery);
|
||||
}
|
||||
|
||||
private reconcileSubmittedPromptEcho(
|
||||
@@ -3971,31 +3981,69 @@ export class AgentManager {
|
||||
const { clientMessageId, messageId } = item;
|
||||
if (!clientMessageId) return null;
|
||||
const existing = this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId);
|
||||
if (
|
||||
!existing ||
|
||||
existing.item.type !== "user_message" ||
|
||||
!messageId ||
|
||||
existing.item.messageId === messageId
|
||||
) {
|
||||
return existing;
|
||||
if (!existing || existing.item.type !== "user_message") return null;
|
||||
this.finalizeSubmittedPrompt(agent, event, clientMessageId, messageId);
|
||||
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;
|
||||
}
|
||||
}
|
||||
const row = this.timelineStore.enrichSubmittedUserMessage(agent.id, clientMessageId, {
|
||||
messageId,
|
||||
});
|
||||
if (row) {
|
||||
this.enqueueDurableTimelineUpdate(agent.id, row);
|
||||
this.dispatchStream(
|
||||
agent.id,
|
||||
{ ...event, item: row.item },
|
||||
{
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agent.id),
|
||||
timestamp: row.timestamp,
|
||||
...(row.delivery ? { delivery: row.delivery } : {}),
|
||||
},
|
||||
this.finalizeSubmittedPrompt(agent, event);
|
||||
}
|
||||
|
||||
private finalizeSubmittedPrompt(
|
||||
agent: ActiveManagedAgent,
|
||||
event: AgentStreamEvent,
|
||||
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",
|
||||
);
|
||||
}
|
||||
return row;
|
||||
const clientMessageId =
|
||||
correlatedClientMessageId ??
|
||||
(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(
|
||||
|
||||
@@ -2,16 +2,20 @@ import { describe, expect, it } from "vitest";
|
||||
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
|
||||
|
||||
describe("InMemoryAgentTimelineStore", () => {
|
||||
it("enriches a submitted prompt without allocating a second sequence", () => {
|
||||
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",
|
||||
});
|
||||
store.append(
|
||||
"agent-1",
|
||||
{
|
||||
type: "user_message",
|
||||
text: "Accepted prompt",
|
||||
clientMessageId: "client-message-1",
|
||||
},
|
||||
{ delivery: "awaiting_provider_identity" },
|
||||
);
|
||||
|
||||
const updated = store.enrichSubmittedUserMessage("agent-1", "client-message-1", {
|
||||
const updated = store.finalizeSubmittedUserMessage("agent-1", "client-message-1", {
|
||||
messageId: "provider-message-1",
|
||||
});
|
||||
|
||||
@@ -24,6 +28,7 @@ describe("InMemoryAgentTimelineStore", () => {
|
||||
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,
|
||||
|
||||
@@ -176,20 +176,28 @@ export class InMemoryAgentTimelineStore {
|
||||
return row ? cloneRow(row) : null;
|
||||
}
|
||||
|
||||
enrichSubmittedUserMessage(
|
||||
finalizeSubmittedUserMessage(
|
||||
agentId: string,
|
||||
clientMessageId: string,
|
||||
enrichment: { messageId: string },
|
||||
identity?: { messageId: string },
|
||||
): AgentTimelineRow | null {
|
||||
const state = this.requireState(agentId);
|
||||
const row = state.rows.find(
|
||||
const index = state.rows.findIndex(
|
||||
(candidate) =>
|
||||
candidate.item.type === "user_message" &&
|
||||
candidate.item.clientMessageId === clientMessageId,
|
||||
);
|
||||
if (!row || row.item.type !== "user_message") return null;
|
||||
row.item = { ...row.item, messageId: enrichment.messageId };
|
||||
return cloneRow(row);
|
||||
const row = state.rows[index];
|
||||
if (!row || row.item.type !== "user_message" || row.delivery !== "awaiting_provider_identity") {
|
||||
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);
|
||||
}
|
||||
|
||||
getEpoch(agentId: string): string {
|
||||
|
||||
@@ -1,63 +1,50 @@
|
||||
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 {
|
||||
export function shouldDeliverSubmittedPromptRow(
|
||||
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
|
||||
supportsRevisions ||
|
||||
delivery !== AWAITING_PROVIDER_IDENTITY_DELIVERY ||
|
||||
item.type !== "user_message" ||
|
||||
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;
|
||||
export function projectSubmittedPromptTimeline(
|
||||
timeline: AgentTimelineFetchResult,
|
||||
fullTimeline = timeline,
|
||||
cursorSeq = 0,
|
||||
): AgentTimelineFetchResult {
|
||||
const pending = fullTimeline.rows.at(-1);
|
||||
if (
|
||||
!pending ||
|
||||
pending.seq !== fullTimeline.window.maxSeq ||
|
||||
shouldDeliverSubmittedPromptRow(pending.delivery, pending.item, false)
|
||||
) {
|
||||
return timeline;
|
||||
}
|
||||
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;
|
||||
const visibleMaxSeq = pending.seq - 1;
|
||||
const rows = timeline.rows.filter((row) => row.seq <= visibleMaxSeq);
|
||||
const selectedEndSeq = rows.at(-1)?.seq ?? cursorSeq;
|
||||
return {
|
||||
...input.timeline,
|
||||
...timeline,
|
||||
window: {
|
||||
...input.timeline.window,
|
||||
maxSeq: input.visibleMaxSeq,
|
||||
nextSeq: input.visibleMaxSeq + 1,
|
||||
...timeline.window,
|
||||
minSeq:
|
||||
visibleMaxSeq === 0 || timeline.window.minSeq > visibleMaxSeq ? 0 : timeline.window.minSeq,
|
||||
maxSeq: visibleMaxSeq,
|
||||
nextSeq: visibleMaxSeq + 1,
|
||||
},
|
||||
hasNewer: input.timeline.hasNewer && selectedEndSeq < input.visibleMaxSeq,
|
||||
hasNewer: timeline.hasNewer && selectedEndSeq < visibleMaxSeq,
|
||||
rows,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -22,11 +22,7 @@ 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,
|
||||
AgentTimelineFetchResult,
|
||||
ManagedAgent,
|
||||
} from "./agent/agent-manager.js";
|
||||
import type { AgentManagerEvent } 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";
|
||||
@@ -4887,11 +4883,9 @@ 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 () => {};
|
||||
@@ -4939,7 +4933,6 @@ test("projects canonical submitted prompt revisions per retained socket capabili
|
||||
},
|
||||
seq: 5,
|
||||
epoch: "epoch-1",
|
||||
delivery: "awaiting_provider_identity",
|
||||
});
|
||||
listener({
|
||||
type: "agent_stream",
|
||||
@@ -4956,7 +4949,6 @@ test("projects canonical submitted prompt revisions per retained socket capabili
|
||||
},
|
||||
seq: 6,
|
||||
epoch: "epoch-1",
|
||||
delivery: "awaiting_provider_identity",
|
||||
});
|
||||
listener({
|
||||
type: "agent_stream",
|
||||
@@ -5004,184 +4996,61 @@ test("projects canonical submitted prompt revisions per retained socket capabili
|
||||
{ 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",
|
||||
},
|
||||
},
|
||||
],
|
||||
};
|
||||
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({
|
||||
targetedMessages,
|
||||
messages,
|
||||
agentManager: {
|
||||
waitForAgentClose: vi.fn().mockResolvedValue(undefined),
|
||||
getAgent: vi.fn(() => agent),
|
||||
fetchTimeline: vi.fn(() => timeline),
|
||||
subscribe: vi.fn((next: (event: AgentManagerEvent) => void) => {
|
||||
listener = next;
|
||||
return () => {};
|
||||
}),
|
||||
},
|
||||
});
|
||||
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",
|
||||
},
|
||||
},
|
||||
],
|
||||
session.updateClientCapabilities(capabilities);
|
||||
const item = {
|
||||
type: "user_message" as const,
|
||||
text: "submitted prompt",
|
||||
clientMessageId: "client-message-1",
|
||||
};
|
||||
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",
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
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", () => {
|
||||
|
||||
@@ -77,7 +77,6 @@ import type {
|
||||
ManagedAgent,
|
||||
} from "./agent/agent-manager.js";
|
||||
import {
|
||||
findSubmittedPromptVisibleMaxSeq,
|
||||
projectSubmittedPromptTimeline,
|
||||
shouldDeliverSubmittedPromptRow,
|
||||
} from "./agent/submitted-prompt-delivery.js";
|
||||
@@ -1070,11 +1069,11 @@ export class Session {
|
||||
if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) {
|
||||
if (
|
||||
serializedEvent.type === "timeline" &&
|
||||
!shouldDeliverSubmittedPromptRow({
|
||||
delivery: event.delivery,
|
||||
item: serializedEvent.item,
|
||||
supportsRevisions: this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions),
|
||||
})
|
||||
!shouldDeliverSubmittedPromptRow(
|
||||
event.delivery,
|
||||
serializedEvent.item,
|
||||
this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions),
|
||||
)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
@@ -1105,11 +1104,11 @@ export class Session {
|
||||
const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline);
|
||||
if (
|
||||
serializedEvent.type === "timeline" &&
|
||||
!shouldDeliverSubmittedPromptRow({
|
||||
delivery: event.delivery,
|
||||
item: serializedEvent.item,
|
||||
supportsRevisions: capabilities.has(CLIENT_CAPS.canonicalSubmittedPromptRevisions),
|
||||
})
|
||||
!shouldDeliverSubmittedPromptRow(
|
||||
event.delivery,
|
||||
serializedEvent.item,
|
||||
capabilities.has(CLIENT_CAPS.canonicalSubmittedPromptRevisions),
|
||||
)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
@@ -6120,36 +6119,30 @@ export class Session {
|
||||
direction: AgentTimelineFetchDirection;
|
||||
cursor?: AgentTimelineCursor;
|
||||
pageLimit: number;
|
||||
visibleMaxSeq?: number;
|
||||
fullTimeline?: AgentTimelineFetchResult;
|
||||
}): AgentTimelineProjectionSelection {
|
||||
const selectedTimeline = this.shouldUseFullTimelineForProjectedPage({
|
||||
timeline: input.controlTimeline,
|
||||
pageLimit: input.pageLimit,
|
||||
})
|
||||
? this.agentManager.fetchTimeline(input.agentId, { direction: "tail", limit: 0 })
|
||||
? (input.fullTimeline ??
|
||||
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,
|
||||
rows: selectedTimeline.rows,
|
||||
bounds: selectedTimeline.window,
|
||||
direction: input.controlTimeline.reset ? "tail" : input.direction,
|
||||
...(input.cursor ? { cursorSeq: input.cursor.seq } : {}),
|
||||
limit: input.pageLimit,
|
||||
});
|
||||
|
||||
return {
|
||||
timeline,
|
||||
timeline: selectedTimeline,
|
||||
entries: page.entries,
|
||||
startSeq: page.startSeq,
|
||||
endSeq: page.endSeq,
|
||||
hasOlder: page.hasOlder || (page.startSeq !== null && page.startSeq > timeline.window.minSeq),
|
||||
hasOlder:
|
||||
page.hasOlder || (page.startSeq !== null && page.startSeq > selectedTimeline.window.minSeq),
|
||||
hasNewer: page.hasNewer,
|
||||
};
|
||||
}
|
||||
@@ -6161,7 +6154,7 @@ export class Session {
|
||||
direction: AgentTimelineFetchDirection;
|
||||
cursor?: AgentTimelineCursor;
|
||||
pageLimit: number;
|
||||
visibleMaxSeq?: number;
|
||||
fullTimeline?: AgentTimelineFetchResult;
|
||||
}): AgentTimelineProjectionSelection {
|
||||
if (input.projection === "canonical") {
|
||||
return this.selectCanonicalTimelineProjection({ timeline: input.controlTimeline });
|
||||
@@ -6201,20 +6194,24 @@ export class Session {
|
||||
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,
|
||||
);
|
||||
let fetchedFullTimeline: AgentTimelineFetchResult | undefined;
|
||||
if (!supportsSubmittedPromptRevisions) {
|
||||
fetchedFullTimeline =
|
||||
direction === "tail" && pageLimit === 0
|
||||
? fetchedControlTimeline
|
||||
: this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0 });
|
||||
}
|
||||
const controlTimeline =
|
||||
visibleMaxSeq === undefined
|
||||
fetchedFullTimeline === undefined
|
||||
? fetchedControlTimeline
|
||||
: projectSubmittedPromptTimeline({
|
||||
timeline: fetchedControlTimeline,
|
||||
visibleMaxSeq,
|
||||
...(cursor ? { cursorSeq: cursor.seq } : {}),
|
||||
});
|
||||
: projectSubmittedPromptTimeline(
|
||||
fetchedControlTimeline,
|
||||
fetchedFullTimeline,
|
||||
cursor?.seq,
|
||||
);
|
||||
const fullTimeline = fetchedFullTimeline
|
||||
? projectSubmittedPromptTimeline(fetchedFullTimeline)
|
||||
: undefined;
|
||||
const selectedTimeline = this.selectTimelineProjection({
|
||||
agentId: msg.agentId,
|
||||
projection,
|
||||
@@ -6222,7 +6219,7 @@ export class Session {
|
||||
direction,
|
||||
...(cursor ? { cursor } : {}),
|
||||
pageLimit,
|
||||
...(visibleMaxSeq !== undefined ? { visibleMaxSeq } : {}),
|
||||
...(fullTimeline ? { fullTimeline } : {}),
|
||||
});
|
||||
const startCursor =
|
||||
selectedTimeline.startSeq !== null
|
||||
|
||||
@@ -16,6 +16,8 @@ import {
|
||||
import { Session, type SessionOptions } from "./session.js";
|
||||
import { createProviderSnapshotManagerStub } from "./test-utils/session-stubs.js";
|
||||
import type { AgentTimelineRow } from "./agent/agent-manager.js";
|
||||
import { InMemoryAgentTimelineStore } from "./agent/agent-timeline-store.js";
|
||||
import type { AgentTimelineFetchOptions } from "./agent/agent-timeline-store-types.js";
|
||||
import { handleCreatePaseoWorktreeRequest } from "./worktree-session.js";
|
||||
import { createPersistedProjectRecord } from "./workspace-registry.js";
|
||||
|
||||
@@ -87,7 +89,15 @@ interface SessionInternals {
|
||||
}
|
||||
|
||||
class InMemoryAgentManager {
|
||||
constructor(private readonly rows: AgentTimelineRow[]) {}
|
||||
private readonly timeline = new InMemoryAgentTimelineStore();
|
||||
|
||||
constructor(rows: AgentTimelineRow[]) {
|
||||
this.timeline.initialize("agent-1", {
|
||||
epoch: "epoch-1",
|
||||
rows,
|
||||
nextSeq: (rows.at(-1)?.seq ?? 0) + 1,
|
||||
});
|
||||
}
|
||||
|
||||
getAgent() {
|
||||
return {
|
||||
@@ -133,17 +143,8 @@ class InMemoryAgentManager {
|
||||
};
|
||||
}
|
||||
|
||||
fetchTimeline() {
|
||||
return {
|
||||
epoch: "epoch-1",
|
||||
reset: false,
|
||||
staleCursor: false,
|
||||
gap: false,
|
||||
window: { minSeq: 1, maxSeq: 3, nextSeq: 4 },
|
||||
rows: this.rows,
|
||||
hasOlder: false,
|
||||
hasNewer: false,
|
||||
};
|
||||
fetchTimeline(_agentId: string, options?: AgentTimelineFetchOptions) {
|
||||
return this.timeline.fetch("agent-1", options);
|
||||
}
|
||||
|
||||
listAgents() {
|
||||
@@ -218,6 +219,7 @@ class InMemoryWorktreeWorkflow {
|
||||
function createSessionForWireCompatTest(options?: {
|
||||
clientCapabilities?: Record<string, unknown> | null;
|
||||
messages?: SessionOutboundMessage[];
|
||||
rows?: AgentTimelineRow[];
|
||||
}): Session {
|
||||
const messages = options?.messages ?? [];
|
||||
const rows: AgentTimelineRow[] = [
|
||||
@@ -247,7 +249,9 @@ function createSessionForWireCompatTest(options?: {
|
||||
downloadTokenStore: {} as SessionOptions["downloadTokenStore"],
|
||||
pushTokenStore: {} as SessionOptions["pushTokenStore"],
|
||||
paseoHome: "/tmp/paseo-home",
|
||||
agentManager: new InMemoryAgentManager(rows) as unknown as SessionOptions["agentManager"],
|
||||
agentManager: new InMemoryAgentManager(
|
||||
options?.rows ?? rows,
|
||||
) as unknown as SessionOptions["agentManager"],
|
||||
agentStorage: new EmptyAgentStorage() as unknown as SessionOptions["agentStorage"],
|
||||
projectRegistry: new EmptyProjectRegistry() as unknown as SessionOptions["projectRegistry"],
|
||||
workspaceRegistry:
|
||||
@@ -308,11 +312,19 @@ function createSessionForWireCompatTest(options?: {
|
||||
return session;
|
||||
}
|
||||
|
||||
async function emitTimelineResponse(
|
||||
clientCapabilities?: Record<string, unknown> | null,
|
||||
): Promise<Extract<SessionOutboundMessage, { type: "fetch_agent_timeline_response" }>> {
|
||||
async function emitTimelineResponse(options?: {
|
||||
clientCapabilities?: Record<string, unknown> | null;
|
||||
rows?: AgentTimelineRow[];
|
||||
request?: Partial<
|
||||
Extract<z.infer<typeof SessionInboundMessageSchema>, { type: "fetch_agent_timeline_request" }>
|
||||
>;
|
||||
}): Promise<Extract<SessionOutboundMessage, { type: "fetch_agent_timeline_response" }>> {
|
||||
const messages: SessionOutboundMessage[] = [];
|
||||
const session = createSessionForWireCompatTest({ clientCapabilities, messages });
|
||||
const session = createSessionForWireCompatTest({
|
||||
clientCapabilities: options?.clientCapabilities,
|
||||
rows: options?.rows,
|
||||
messages,
|
||||
});
|
||||
const internals = session as unknown as SessionInternals;
|
||||
|
||||
await internals.handleFetchAgentTimelineRequest({
|
||||
@@ -320,6 +332,7 @@ async function emitTimelineResponse(
|
||||
requestId: "req-timeline",
|
||||
agentId: "agent-1",
|
||||
projection: "projected",
|
||||
...options?.request,
|
||||
});
|
||||
|
||||
const response = messages[0];
|
||||
@@ -377,7 +390,7 @@ describe("wire compatibility", () => {
|
||||
]);
|
||||
});
|
||||
|
||||
test("hello parses with and without the project update capability", () => {
|
||||
test("hello parses with and without current client capabilities", () => {
|
||||
const legacy = WSHelloMessageSchema.parse({
|
||||
type: "hello",
|
||||
clientId: "legacy-client",
|
||||
@@ -389,7 +402,10 @@ describe("wire compatibility", () => {
|
||||
clientId: "capable-client",
|
||||
clientType: "mobile",
|
||||
protocolVersion: 1,
|
||||
capabilities: { [CLIENT_CAPS.projectUpdates]: true },
|
||||
capabilities: {
|
||||
[CLIENT_CAPS.projectUpdates]: true,
|
||||
[CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true,
|
||||
},
|
||||
});
|
||||
|
||||
expect([legacy, capable]).toEqual([
|
||||
@@ -404,7 +420,10 @@ describe("wire compatibility", () => {
|
||||
clientId: "capable-client",
|
||||
clientType: "mobile",
|
||||
protocolVersion: 1,
|
||||
capabilities: { project_updates: true },
|
||||
capabilities: {
|
||||
project_updates: true,
|
||||
canonical_submitted_prompt_revisions: true,
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
@@ -463,13 +482,94 @@ describe("wire compatibility", () => {
|
||||
|
||||
test("preserves reasoning_merge for clients that declare the capability", async () => {
|
||||
const response = await emitTimelineResponse({
|
||||
[CLIENT_CAPS.reasoningMergeEnum]: true,
|
||||
clientCapabilities: { [CLIENT_CAPS.reasoningMergeEnum]: true },
|
||||
});
|
||||
|
||||
const currentParsed = FetchAgentTimelineResponseMessageSchema.parse(response);
|
||||
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 = (options?: { delivery?: AgentTimelineRow["delivery"]; messageId?: string }) => ({
|
||||
seq: 1,
|
||||
timestamp,
|
||||
...(options?.delivery ? { delivery: options.delivery } : {}),
|
||||
item: {
|
||||
type: "user_message" as const,
|
||||
text: "hello",
|
||||
clientMessageId: "client-message-1",
|
||||
...(options?.messageId ? { messageId: options.messageId } : {}),
|
||||
},
|
||||
});
|
||||
|
||||
const pending = await emitTimelineResponse({
|
||||
rows: [prompt({ 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({ 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()],
|
||||
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}`,
|
||||
},
|
||||
}));
|
||||
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]]);
|
||||
}
|
||||
});
|
||||
|
||||
test("sub_agent tool-call payload still parses against the v0.1.65-beta.3 schema", () => {
|
||||
const parsed = LegacySubAgentToolCallSchema.parse({
|
||||
type: "tool_call",
|
||||
|
||||
Reference in New Issue
Block a user