jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Stop completed turns from appearing stuck (#2484)

Browse Source 
* fix(app): stop completed turns from appearing stuck

Turn activity was inferred from a user-message row flag, so a stale duplicate row could keep the working footer active after the turn ended. Track submission lifecycle separately and route every send path through the shared submission flow.

* fix(app): guard pending message identity

* test(app): measure submission layout independent of scroll

* fix(app): keep submitted messages consistent through reconnects

Track each in-flight send until its own RPC establishes acceptance, and keep canonical timeline placement authoritative. Restore legacy cached rewind IDs during cache deserialization.

* fix(app): close canonical submission races

* test(app): cover canonical submission races in browser

* fix(app): keep replacement submissions authoritative

Keep current pending rows out of legacy cache migration and leave ambiguous terminal lifecycle events to the daemon snapshot. This prevents fabricated rewind identities and stale completion events from marking replacement turns idle.

* fix(app): keep submitted messages stable across sync

Give submission transport and canonical timeline ingestion separate authority. Preserve every unresolved local send across replacement, reconcile provider identity once, and bridge RPC acceptance to authoritative running state without deriving lifecycle from timeline rows.

* fix(app): settle submissions in either acknowledgement order

Complete submission transactions when RPC and provider acknowledgement arrive in either order. Cache only transaction-owned local rows as transient data, preserve canonical ID-less prompts, and invalidate ambiguous legacy display caches instead of inventing provider identity.

* fix(app): keep agent visible during history handoff

Running and terminal updates could clear create continuity before the initial authoritative timeline arrived, leaving a streaming agent behind the loading screen. End the handoff only when authoritative history is applied.

* fix(app): settle attachment-only submissions

Canonical providers can acknowledge image-only prompts with empty text. Reconcile those events by client identity without rendering a blank canonical row.

* fix(agent): settle out-of-band message submissions

Accepted commands that do not allocate a foreground turn previously had no canonical user acknowledgement. Record the command before its handler runs so submission state and reconnect history converge through the normal timeline producer.

* fix(app): settle out-of-band submissions compatibly

* fix(app): preserve canonical prompt order

* test(app): keep workspace status check timing-independent

The workspace-status scenario asserted footer settlement before initial agent creation had necessarily completed. The dedicated draft-handoff coverage owns that lifecycle contract.
This commit is contained in:
Mohamed Boudra
2026-07-27 21:27:10 +02:00
committed by GitHub
parent 1f253d92e2
commit fa1198c2be
39 changed files with 3416 additions and 1048 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/timeline/session-stream-reducers.test.ts
+282 -259
View File
@@ -1,7 +1,7 @@
import { describe, expect, it } from "vitest";
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
import {
buildOptimisticUserMessage,
createUserMessage,
hydrateStreamState,
type AgentToolCallItem,
type StreamItem,
@@ -119,12 +119,12 @@ function makeAssistantItem(
};
}
function makeOptimisticUserMessage(
function makeSubmittedUserMessage(
text: string,
id = `optimistic-${text.length}`,
id = `submitted-${text.length}`,
): Extract<StreamItem, { kind: "user_message" }> {
return buildOptimisticUserMessage({
id,
return createUserMessage({
clientMessageId: id,
text,
timestamp: new Date(1000),
});
@@ -172,6 +172,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
isInitializing: false,
hasActiveInitDeferred: false,
initRequestDirection: "tail",
sendingClientMessageIds: [],
};
const baseStreamInput: ProcessAgentStreamEventInput = {
@@ -181,7 +182,6 @@ const baseStreamInput: ProcessAgentStreamEventInput = {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
timestamp: new Date(2000),
};
@@ -293,6 +293,75 @@ describe("processTimelineResponse", () => {
expect(result.sideEffects.some((e) => e.type === "flush_pending_updates")).toBe(true);
});
it("keeps a live assistant and submitted head prompt in one lane during replacement", () => {
const submitted = makeSubmittedUserMessage("New prompt", "client-new-prompt");
const liveAssistant = {
...makeAssistantItem("Live answer", "answer-1"),
messageId: "answer-1",
};
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [],
currentHead: [liveAssistant, submitted],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
sendingClientMessageIds: ["client-new-prompt"],
payload: {
...baseTimelineInput.payload,
reset: true,
epoch: "epoch-1",
startCursor: { seq: 1 },
endCursor: { seq: 1 },
entries: [
{
...makeTimelineEntry(1, "Live", "assistant_message"),
item: {
type: "assistant_message",
text: "Live",
messageId: "answer-1",
},
},
],
},
});
expect(result.tail).toEqual([]);
expect(result.head).toEqual([{ ...liveAssistant, text: "Live" }, submitted]);
});
it("preserves newer live head items when canonical replacement ends in a tool call", () => {
const liveThought: StreamItem = {
kind: "thought",
id: "live-thought",
text: "newer reasoning",
timestamp: new Date(3000),
status: "loading",
};
const liveAssistant = makeAssistantItem("newer answer", "live-answer");
const result = processTimelineResponse({
...baseTimelineInput,
currentHead: [liveThought, liveAssistant],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
payload: {
...baseTimelineInput.payload,
reset: true,
epoch: "epoch-1",
startCursor: { seq: 1 },
endCursor: { seq: 1 },
entries: [
makeToolCallTimelineEntry(1, "canonical-call", "completed", {
type: "read",
filePath: "/tmp/older.ts",
}),
],
},
});
expect(result.tail.map((item) => item.kind)).toEqual(["tool_call"]);
expect(result.head).toEqual([liveThought, liveAssistant]);
});
it("uses the timeline entry timestamp as canonical", () => {
const result = processTimelineResponse({
...baseTimelineInput,
@@ -321,12 +390,12 @@ describe("processTimelineResponse", () => {
expect(assistant?.timestamp.toISOString()).toBe("2025-01-01T12:00:04.000Z");
});
it("reconciles an optimistic user message during tail replacement", () => {
it("reconciles a submitted user message during tail replacement", () => {
const image = {
id: "optimistic-image",
id: "submitted-image",
mimeType: "image/png",
storageType: "web-indexeddb" as const,
storageKey: "optimistic-image",
storageKey: "submitted-image",
createdAt: 1000,
};
const attachment = {
@@ -335,8 +404,8 @@ describe("processTimelineResponse", () => {
text: "attached context",
title: "context.txt",
};
const optimistic = buildOptimisticUserMessage({
id: "optimistic-create-user",
const submitted = createUserMessage({
clientMessageId: "submitted-create-user",
text: "Analyze this",
timestamp: new Date(1000),
images: [image],
@@ -345,7 +414,7 @@ describe("processTimelineResponse", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [optimistic],
currentTail: [submitted],
payload: {
...baseTimelineInput.payload,
reset: true,
@@ -358,6 +427,7 @@ describe("processTimelineResponse", () => {
type: "user_message",
text: "server-rendered attachment text",
messageId: "canonical-create-user",
clientMessageId: "submitted-create-user",
},
},
],
@@ -367,13 +437,14 @@ describe("processTimelineResponse", () => {
const userMessages = result.tail.filter((item) => item.kind === "user_message");
expect(userMessages).toHaveLength(1);
expect(userMessages[0]).toMatchObject({
id: "canonical-create-user",
id: "submitted-create-user",
clientMessageId: "submitted-create-user",
messageId: "canonical-create-user",
text: "Analyze this",
timestamp: new Date(1000),
images: [image],
attachments: [attachment],
});
expect(userMessages[0]?.optimistic).toBeUndefined();
const repeated = processTimelineResponse({
...baseTimelineInput,
@@ -390,6 +461,7 @@ describe("processTimelineResponse", () => {
type: "user_message",
text: "server-rendered attachment text",
messageId: "canonical-create-user",
clientMessageId: "submitted-create-user",
},
},
],
@@ -399,24 +471,93 @@ describe("processTimelineResponse", () => {
expect(repeated.tail.filter((item) => item.kind === "user_message")).toEqual(userMessages);
});
it("keeps an unmatched optimistic user message during tail replacement", () => {
const optimistic = makeOptimisticUserMessage("still sending", "optimistic-unmatched");
it("keeps an unmatched submitted user message during tail replacement", () => {
const submitted = makeSubmittedUserMessage("still sending", "submitted-unmatched");
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [optimistic],
currentTail: [submitted],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
sendingClientMessageIds: ["submitted-unmatched"],
payload: {
...baseTimelineInput.payload,
reset: true,
epoch: "epoch-2",
entries: [],
},
});
expect(result.tail).toEqual([optimistic]);
expect(result.tail).toEqual([submitted]);
});
it("does not move an unmatched submission during timeline replacement", () => {
const unmatched = makeOptimisticUserMessage("first submission", "client-first");
it("keeps every unresolved submission during replacement", () => {
const first = makeSubmittedUserMessage("first pending", "client-first");
const second = makeSubmittedUserMessage("second pending", "client-second");
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [first, second],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
sendingClientMessageIds: ["client-first", "client-second"],
payload: {
...baseTimelineInput.payload,
reset: true,
epoch: "epoch-2",
entries: [],
},
});
expect(result.tail).toEqual([first, second]);
});
it("drops an acknowledged local row omitted by a same-epoch replacement", () => {
const acknowledged = createUserMessage({
clientMessageId: "client-local-only",
text: "provider may not echo this",
timestamp: new Date(1000),
});
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [acknowledged],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
sendingClientMessageIds: [],
payload: {
...baseTimelineInput.payload,
reset: true,
epoch: "epoch-1",
entries: [],
},
});
expect(result.tail).toEqual([]);
});
it("drops an acknowledged local row omitted by a known epoch change", () => {
const acknowledged = createUserMessage({
clientMessageId: "client-prior-epoch",
text: "prior prompt",
timestamp: new Date(1000),
});
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [acknowledged],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
sendingClientMessageIds: [],
payload: {
...baseTimelineInput.payload,
reset: true,
epoch: "epoch-2",
entries: [],
},
});
expect(result.tail).toEqual([]);
});
it("keeps an unmatched submission after the canonical replacement range", () => {
const unmatched = makeSubmittedUserMessage("first submission", "client-first");
const acknowledged: StreamItem[] = [
{
kind: "user_message",
@@ -437,6 +578,7 @@ describe("processTimelineResponse", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [unmatched, ...acknowledged],
sendingClientMessageIds: ["client-first"],
payload: {
...baseTimelineInput.payload,
reset: true,
@@ -462,10 +604,10 @@ describe("processTimelineResponse", () => {
},
},
{
...makeTimelineEntry(4, "response to all three submissions"),
...makeTimelineEntry(4, "response to canonical submissions"),
item: {
type: "assistant_message",
text: "response to all three submissions",
text: "response to canonical submissions",
messageId: "assistant-response",
},
},
@@ -480,14 +622,14 @@ describe("processTimelineResponse", () => {
text: "text" in item ? item.text : undefined,
})),
).toEqual([
{ kind: "user_message", id: "client-first", text: "first submission" },
{ kind: "user_message", id: "provider-second", text: "second submission" },
{ kind: "user_message", id: "provider-third", text: "third submission" },
{
kind: "assistant_message",
id: "assistant-response",
text: "response to all three submissions",
text: "response to canonical submissions",
},
{ kind: "user_message", id: "client-first", text: "first submission" },
]);
});
@@ -620,11 +762,11 @@ describe("processTimelineResponse", () => {
startSeq: 1,
endSeq: 1,
};
const optimistic = makeOptimisticUserMessage("sent while catching up", "optimistic-after");
const submitted = makeSubmittedUserMessage("sent while catching up", "submitted-after");
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [optimistic],
currentTail: [submitted],
currentCursor: existingCursor,
payload: {
...baseTimelineInput.payload,
@@ -644,16 +786,19 @@ describe("processTimelineResponse", () => {
const userMessages = result.tail.filter((item) => item.kind === "user_message");
expect(userMessages).toHaveLength(1);
expect(userMessages[0]?.id).toBe("canonical-after");
expect(userMessages[0]?.optimistic).toBeUndefined();
expect(userMessages[0]).toMatchObject({
id: "submitted-after",
clientMessageId: "submitted-after",
messageId: "canonical-after",
});
});
it("reconciles an optimistic user message by client message id", () => {
const optimistic = makeOptimisticUserMessage("local presentation", "client-message");
it("reconciles a submitted user message by client message id", () => {
const submitted = makeSubmittedUserMessage("local presentation", "client-message");
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [optimistic],
currentTail: [submitted],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
payload: {
...baseTimelineInput.payload,
@@ -675,20 +820,21 @@ describe("processTimelineResponse", () => {
const userMessages = result.tail.filter((item) => item.kind === "user_message");
expect(userMessages).toEqual([
expect.objectContaining({
id: "provider-message",
id: "client-message",
clientMessageId: "client-message",
messageId: "provider-message",
text: "local presentation",
}),
]);
expect(userMessages[0]?.optimistic).toBeUndefined();
expect(result.acknowledgedClientMessageIds).toEqual(["client-message"]);
});
it("reconciles multiple optimistic user messages in canonical order", () => {
it("reconciles multiple submitted user messages in canonical order", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [
makeOptimisticUserMessage("first prompt", "optimistic-first"),
makeOptimisticUserMessage("second prompt", "optimistic-second"),
makeSubmittedUserMessage("first prompt", "submitted-first"),
makeSubmittedUserMessage("second prompt", "submitted-second"),
],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
payload: {
@@ -699,14 +845,14 @@ describe("processTimelineResponse", () => {
entries: [
{
...makeTimelineEntry(2, "first prompt", "user_message"),
item: { type: "user_message", text: "first prompt", messageId: "optimistic-first" },
item: { type: "user_message", text: "first prompt", messageId: "submitted-first" },
},
{
...makeTimelineEntry(3, "second prompt", "user_message"),
item: {
type: "user_message",
text: "second prompt",
messageId: "optimistic-second",
messageId: "submitted-second",
},
},
],
@@ -716,15 +862,15 @@ describe("processTimelineResponse", () => {
expect(
result.tail
.filter((item) => item.kind === "user_message")
.map((item) => ({ id: item.id, text: item.text, optimistic: item.optimistic })),
.map((item) => ({ id: item.id, text: item.text, messageId: item.messageId })),
).toEqual([
{ id: "optimistic-first", text: "first prompt", optimistic: undefined },
{ id: "optimistic-second", text: "second prompt", optimistic: undefined },
{ id: "submitted-first", text: "first prompt", messageId: "submitted-first" },
{ id: "submitted-second", text: "second prompt", messageId: "submitted-second" },
]);
});
it("keeps a tail optimistic prompt before a reconciled live assistant head", () => {
const prompt = makeOptimisticUserMessage("new prompt", "optimistic-new-prompt");
it("keeps a tail submitted prompt before a reconciled live assistant head", () => {
const prompt = makeSubmittedUserMessage("new prompt", "submitted-new-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -763,8 +909,8 @@ describe("processTimelineResponse", () => {
).toEqual(["new prompt", "Hello"]);
});
it("keeps a tail optimistic prompt before a live head flushed by catch-up", () => {
const prompt = makeOptimisticUserMessage("new prompt", "optimistic-new-prompt");
it("keeps a tail submitted prompt before a live head flushed by catch-up", () => {
const prompt = makeSubmittedUserMessage("new prompt", "submitted-new-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -835,7 +981,6 @@ describe("processTimelineResponse", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
const result = processTimelineResponse({
@@ -913,7 +1058,6 @@ describe("processTimelineResponse", () => {
currentTail: [],
currentHead: [],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
currentAgent: null,
});
expect(getAssistantTexts(live.tail)).toHaveLength(1);
expect(getAssistantTexts(live.head)).toHaveLength(1);
@@ -961,7 +1105,6 @@ describe("processTimelineResponse", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
const result = processTimelineResponse({
@@ -1000,7 +1143,7 @@ describe("processTimelineResponse", () => {
});
it("does not move a submitted prompt when catch-up history arrives", () => {
const prompt = makeOptimisticUserMessage("New prompt", "new-prompt");
const prompt = makeSubmittedUserMessage("New prompt", "new-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -1024,7 +1167,7 @@ describe("processTimelineResponse", () => {
});
it("does not move an unmatched head prompt when catch-up history arrives", () => {
const prompt = makeOptimisticUserMessage("New prompt", "new-prompt");
const prompt = makeSubmittedUserMessage("New prompt", "new-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -1052,8 +1195,8 @@ describe("processTimelineResponse", () => {
]);
});
it("acknowledges a head prompt in place while catch-up history arrives", () => {
const prompt = makeOptimisticUserMessage("New prompt", "new-prompt");
it("moves an acknowledged head prompt to its catch-up sequence position", () => {
const prompt = makeSubmittedUserMessage("New prompt", "new-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -1084,18 +1227,18 @@ describe("processTimelineResponse", () => {
expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([
"assistant_message",
"user_message",
"tool_call",
"user_message",
]);
expect(
[...result.tail, ...result.head]
.filter((item) => item.kind === "user_message")
.map((item) => item.optimistic),
).toEqual([undefined]);
.map((item) => item.clientMessageId),
).toEqual(["new-prompt"]);
});
it("does not move a prompt around unrelated catch-up history", () => {
const prompt = makeOptimisticUserMessage("New prompt", "new-prompt");
const prompt = makeSubmittedUserMessage("New prompt", "new-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -1141,7 +1284,7 @@ describe("processTimelineResponse", () => {
});
it("does not move a prompt or its live answer around catch-up history", () => {
const prompt = makeOptimisticUserMessage("New prompt", "new-prompt");
const prompt = makeSubmittedUserMessage("New prompt", "new-prompt");
const live = processAgentStreamEvents({
events: [
makeStreamReducerEvent(
@@ -1152,7 +1295,6 @@ describe("processTimelineResponse", () => {
currentTail: [prompt],
currentHead: [],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
currentAgent: null,
});
const result = processTimelineResponse({
@@ -1186,7 +1328,7 @@ describe("processTimelineResponse", () => {
});
it("does not move a prompt or its live answer around catch-up tool history", () => {
const prompt = makeOptimisticUserMessage("New prompt", "new-prompt");
const prompt = makeSubmittedUserMessage("New prompt", "new-prompt");
const live = processAgentStreamEvents({
events: [
makeStreamReducerEvent(
@@ -1197,7 +1339,6 @@ describe("processTimelineResponse", () => {
currentTail: [prompt],
currentHead: [],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
currentAgent: null,
});
const result = processTimelineResponse({
@@ -1228,7 +1369,7 @@ describe("processTimelineResponse", () => {
});
it("never moves submitted messages behind a later assistant response", () => {
const unmatched = makeOptimisticUserMessage("first submission", "client-first");
const unmatched = makeSubmittedUserMessage("first submission", "client-first");
const acknowledged: StreamItem[] = [
{
kind: "user_message",
@@ -1286,8 +1427,8 @@ describe("processTimelineResponse", () => {
]);
});
it("acknowledges a local prompt in place when a remote user row also arrives", () => {
const prompt = makeOptimisticUserMessage("Local prompt", "local-prompt");
it("places a local prompt after an earlier remote canonical row", () => {
const prompt = makeSubmittedUserMessage("Local prompt", "local-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -1320,17 +1461,12 @@ describe("processTimelineResponse", () => {
});
expect(
result.tail
.filter((item) => item.kind === "user_message")
.map((item) => ({ text: item.text, optimistic: item.optimistic })),
).toEqual([
{ text: "Local prompt", optimistic: undefined },
{ text: "Remote prompt", optimistic: undefined },
]);
result.tail.filter((item) => item.kind === "user_message").map((item) => item.text),
).toEqual(["Remote prompt", "Local prompt"]);
});
it("keeps an unmatched optimistic prompt when catch-up contains only a remote user row", () => {
const prompt = makeOptimisticUserMessage("Local prompt", "local-prompt");
it("keeps an unmatched submitted prompt when catch-up contains only a remote user row", () => {
const prompt = makeSubmittedUserMessage("Local prompt", "local-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -1355,17 +1491,12 @@ describe("processTimelineResponse", () => {
});
expect(
result.tail
.filter((item) => item.kind === "user_message")
.map((item) => ({ text: item.text, optimistic: item.optimistic })),
).toEqual([
{ text: "Local prompt", optimistic: true },
{ text: "Remote prompt", optimistic: undefined },
]);
result.tail.filter((item) => item.kind === "user_message").map((item) => item.text),
).toEqual(["Local prompt", "Remote prompt"]);
});
it("does not match equal prompt text when canonical client message ids differ", () => {
const prompt = makeOptimisticUserMessage("continue", "local-prompt");
const prompt = makeSubmittedUserMessage("continue", "local-prompt");
const result = processTimelineResponse({
...baseTimelineInput,
@@ -1393,10 +1524,10 @@ describe("processTimelineResponse", () => {
expect(
result.tail
.filter((item) => item.kind === "user_message")
.map((item) => ({ id: item.id, optimistic: item.optimistic })),
.map((item) => ({ id: item.id, messageId: item.messageId })),
).toEqual([
{ id: "local-prompt", optimistic: true },
{ id: "remote-prompt", optimistic: undefined },
{ id: "local-prompt", messageId: undefined },
{ id: "remote-prompt", messageId: "remote-prompt" },
]);
});
@@ -1639,8 +1770,8 @@ describe("processTimelineResponse", () => {
});
});
it("does not reconcile an active optimistic user message from a before-page response", () => {
const optimistic = makeOptimisticUserMessage("active prompt", "optimistic-active");
it("does not reconcile an active submitted user message from a before-page response", () => {
const submitted = makeSubmittedUserMessage("active prompt", "submitted-active");
const existingCursor: TimelineCursor = {
epoch: "epoch-1",
startSeq: 3,
@@ -1649,7 +1780,7 @@ describe("processTimelineResponse", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [optimistic],
currentTail: [submitted],
currentCursor: existingCursor,
payload: {
...baseTimelineInput.payload,
@@ -1672,8 +1803,8 @@ describe("processTimelineResponse", () => {
const userMessages = result.tail.filter((item) => item.kind === "user_message");
expect(userMessages).toHaveLength(2);
expect(userMessages.map((item) => item.id)).toEqual(["canonical-before", "optimistic-active"]);
expect(userMessages[1]?.optimistic).toBe(true);
expect(userMessages.map((item) => item.id)).toEqual(["canonical-before", "submitted-active"]);
expect(userMessages[1]?.clientMessageId).toBe("submitted-active");
});
it("leaves the cursor alone when a before page makes no progress", () => {
@@ -1813,6 +1944,68 @@ describe("processTimelineResponse", () => {
]);
});
it("removes a reconciled submitted prompt before coalescing a tool call at the pagination seam", () => {
const clientMessageId = "client-boundary-prompt";
const callId = "toolu_submitted_boundary";
const currentTail = [
makeSubmittedUserMessage("Inspect the file", clientMessageId),
...hydrateStreamState(
[
{
event: {
type: "timeline",
provider: "claude",
item: makeToolCallTimelineEntry(3, callId, "completed", {
type: "read",
filePath: "/tmp/example.ts",
}).item,
} as AgentStreamEventPayload,
timestamp: new Date(3000),
},
],
{ source: "canonical" },
),
];
const result = processTimelineResponse({
...baseTimelineInput,
currentTail,
currentCursor: { epoch: "epoch-1", startSeq: 3, endSeq: 5 },
payload: {
...baseTimelineInput.payload,
direction: "before",
epoch: "epoch-1",
startCursor: { seq: 1 },
endCursor: { seq: 2 },
entries: [
{
...makeTimelineEntry(1, "Inspect the file", "user_message"),
item: {
type: "user_message",
text: "Inspect the file",
messageId: "provider-boundary-prompt",
clientMessageId,
},
},
makeToolCallTimelineEntry(2, callId, "running", {
type: "unknown",
input: { file_path: "/tmp/example.ts" },
output: null,
}),
],
},
});
expect(result.tail.filter((item) => item.kind === "user_message")).toHaveLength(1);
expect(getAgentToolCalls(result.tail)).toEqual([
expect.objectContaining({
payload: expect.objectContaining({
data: expect.objectContaining({ callId, status: "completed" }),
}),
}),
]);
});
it("does not coalesce tool call lifecycle rows away from the prepend boundary", () => {
const callId = "toolu_not_boundary";
const currentTail = hydrateStreamState(
@@ -2224,152 +2417,6 @@ describe("processAgentStreamEvent", () => {
endSeq: 1,
});
});
it("derives optimistic idle status on turn_completed for running agent", () => {
const turnCompletedEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCompletedEvent,
currentAgent: {
status: "running",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(true);
expect(result.agent).not.toBe(null);
expect(result.agent!.status).toBe("idle");
expect(result.agent!.updatedAt.getTime()).toBe(2000);
expect(result.agent!.lastActivityAt.getTime()).toBe(2000);
});
it("derives optimistic error status on turn_failed for running agent", () => {
const turnFailedEvent: AgentStreamEventPayload = {
type: "turn_failed",
provider: "claude",
error: "something broke",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnFailedEvent,
currentAgent: {
status: "running",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(true);
expect(result.agent!.status).toBe("error");
});
it("does not derive optimistic idle status on turn_canceled for running agent", () => {
const turnCanceledEvent: AgentStreamEventPayload = {
type: "turn_canceled",
provider: "codex",
reason: "interrupted",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCanceledEvent,
currentAgent: {
status: "running",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(false);
expect(result.agent).toBe(null);
});
it("does not change agent when status is not running", () => {
const turnCompletedEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCompletedEvent,
currentAgent: {
status: "idle",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(false);
expect(result.agent).toBe(null);
});
it("does not change agent when no agent is provided", () => {
const turnCompletedEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCompletedEvent,
currentAgent: null,
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(false);
expect(result.agent).toBe(null);
});
it("preserves updatedAt when agent timestamp is newer than event", () => {
const turnCompletedEvent: AgentStreamEventPayload = {
type: "turn_completed",
provider: "claude",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnCompletedEvent,
currentAgent: {
status: "running",
updatedAt: new Date(5000),
lastActivityAt: new Date(5000),
},
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(true);
expect(result.agent!.updatedAt.getTime()).toBe(5000);
expect(result.agent!.lastActivityAt.getTime()).toBe(5000);
});
it("does not produce agent patch for non-terminal events", () => {
const result = processAgentStreamEvent({
...baseStreamInput,
event: makeTimelineEvent("just text"),
currentAgent: {
status: "running",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
seq: 1,
epoch: "epoch-1",
timestamp: new Date(2000),
});
expect(result.agentChanged).toBe(false);
expect(result.agent).toBe(null);
});
});
describe("processAgentStreamEvents", () => {
@@ -2382,7 +2429,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(false);
@@ -2410,7 +2456,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(false);
@@ -2433,7 +2478,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2451,7 +2495,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2469,7 +2512,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2501,7 +2543,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2527,7 +2568,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2552,7 +2592,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2644,7 +2683,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(getAssistantTexts([...result.tail, ...result.head])).toEqual([
@@ -2655,7 +2693,7 @@ describe("processAgentStreamEvents", () => {
]);
});
it("returns the final optimistic lifecycle patch across a batch", () => {
it("does not derive lifecycle state from a terminal event in a batch", () => {
const result = processAgentStreamEvents({
events: [
makeStreamReducerEvent(makeTimelineEvent("Done"), 1),
@@ -2669,21 +2707,10 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: {
status: "running",
updatedAt: new Date(1000),
lastActivityAt: new Date(1000),
},
});
expect(result.head).toEqual([]);
expect(result.tail).toHaveLength(1);
expect(result.agentChanged).toBe(true);
expect(result.agent).toMatchObject({
status: "idle",
updatedAt: new Date(3000),
lastActivityAt: new Date(3000),
});
});
it("keeps a live Claude assistant paragraph contiguous when init tail hydration lands mid-stream", () => {
@@ -2832,7 +2859,6 @@ describe("createAgentStreamReducerQueue", () => {
currentTail,
currentHead,
currentCursor: undefined,
currentAgent: null,
}),
commit: (agentId, result) => {
currentTail = result.tail;
@@ -2874,7 +2900,6 @@ describe("createAgentStreamReducerQueue", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
}),
commit: (agentId, result) => {
commits.push(
@@ -2904,7 +2929,6 @@ describe("createAgentStreamReducerQueue", () => {
currentTail,
currentHead,
currentCursor,
currentAgent: null,
}),
commit: (_agentId, result) => {
currentTail = result.tail;
@@ -2954,7 +2978,6 @@ describe("createAgentStreamReducerQueue", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
}),
commit: (agentId, result) => {
commits.push(
packages/app/src/timeline/session-stream-reducers.ts
+134 -307
View File
@@ -1,15 +1,15 @@
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
import type { AgentLifecycleStatus } from "@getpaseo/protocol/agent-lifecycle";
import type { Agent } from "@/stores/session-store";
import { useSessionStore } from "@/stores/session-store";
import type { AssistantMessageItem, StreamItem, UserMessageItem } from "@/types/stream";
import type { AssistantMessageItem, StreamItem } from "@/types/stream";
import {
applyStreamEvent,
flushHeadToTail,
hydrateStreamState,
isAgentToolCallItem,
mergeAgentToolCallItem,
replaceWithCanonicalStream,
reduceStreamUpdate,
upsertUserMessageAcrossStream,
} from "@/types/stream";
const AGENT_STREAM_REDUCER_FLUSH_DELAY_MS = 16 * 3;
@@ -88,6 +88,7 @@ export interface ProcessTimelineResponseInput {
isInitializing: boolean;
hasActiveInitDeferred: boolean;
initRequestDirection: InitRequestDirection;
sendingClientMessageIds: readonly string[];
}
export interface ProcessTimelineResponseOutput {
@@ -99,6 +100,7 @@ export interface ProcessTimelineResponseOutput {
clearInitializing: boolean;
error: string | null;
sideEffects: TimelineReducerSideEffect[];
acknowledgedClientMessageIds: string[];
}
interface TimelineUnit {
@@ -115,6 +117,7 @@ interface TimelinePathResult {
cursor: TimelineCursor | null | undefined;
cursorChanged: boolean;
sideEffects: TimelineReducerSideEffect[];
acknowledgedClientMessageIds: string[];
}
function classifySessionTimelineSeq({
@@ -201,76 +204,35 @@ function shouldResolveTimelineInit({
return responseDirection === initRequestDirection;
}
function deriveOptimisticLifecycleStatus(
currentStatus: AgentLifecycleStatus,
event: AgentStreamEventPayload,
): AgentLifecycleStatus | null {
if (currentStatus !== "running") {
return null;
}
switch (event.type) {
case "turn_completed":
return "idle";
case "turn_failed":
return "error";
case "turn_canceled":
// A canceled turn can be either a final user cancel or an interrupt before
// a replacement turn starts. The daemon snapshot is authoritative here.
return null;
default:
return null;
}
}
function preserveReplacePathAssistantHead(params: {
tail: StreamItem[];
currentHead: StreamItem[];
}): {
tail: StreamItem[];
head: StreamItem[];
} {
const { tail, currentHead } = params;
const liveAssistant = currentHead.findLast(
(item): item is Extract<StreamItem, { kind: "assistant_message" }> =>
item.kind === "assistant_message",
);
if (!liveAssistant) {
return { tail, head: [] };
}
const tailAssistant = tail.at(-1);
if (!tailAssistant || tailAssistant.kind !== "assistant_message") {
return { tail, head: currentHead };
}
if (!liveAssistant.text.startsWith(tailAssistant.text)) {
return { tail, head: [] };
}
return {
tail: tail.slice(0, -1),
head: [{ ...liveAssistant, text: tailAssistant.text }],
};
}
function applyTimelineReplacePath(args: {
timelineUnits: TimelineUnit[];
payload: ProcessTimelineResponseInput["payload"];
bootstrapPolicy: ReturnType<typeof deriveBootstrapTailTimelinePolicy>;
currentTail: StreamItem[];
currentHead: StreamItem[];
sendingClientMessageIds: readonly string[];
preserveLiveHead: boolean;
toHydratedEvents: (
units: TimelineUnit[],
) => Array<{ event: AgentStreamEventPayload; timestamp: Date }>;
}): TimelinePathResult {
const { timelineUnits, payload, bootstrapPolicy, currentTail, currentHead, toHydratedEvents } =
args;
const {
timelineUnits,
payload,
bootstrapPolicy,
currentTail,
currentHead,
sendingClientMessageIds,
preserveLiveHead,
toHydratedEvents,
} = args;
const hydratedTail = hydrateStreamState(toHydratedEvents(timelineUnits), { source: "canonical" });
const reconciledTail = reconcileLocalUserPresentationAfterReplace({
canonicalTail: hydratedTail,
const { tail, head, acknowledgedClientMessageIds } = replaceWithCanonicalStream({
canonical: hydratedTail,
previousTail: currentTail,
previousHead: currentHead,
});
const { tail, head } = preserveReplacePathAssistantHead({
tail: reconciledTail,
currentHead,
sendingClientMessageIds,
preserveLiveHead,
});
const cursor: TimelineCursor | null =
payload.startCursor && payload.endCursor
@@ -284,136 +246,16 @@ function applyTimelineReplacePath(args: {
if (bootstrapPolicy.catchUpCursor) {
sideEffects.push({ type: "catch_up", cursor: bootstrapPolicy.catchUpCursor });
}
return { tail, head, cursor, cursorChanged: true, sideEffects };
}
function collectLocallyPresentedUserMessages(items: StreamItem[]): Array<{
ordinal: number;
item: UserMessageItem;
}> {
const localUsers: Array<{ ordinal: number; item: UserMessageItem }> = [];
let ordinal = 0;
for (const item of items) {
if (item.kind !== "user_message") {
continue;
}
if (item.optimistic || item.images?.length || item.attachments?.length) {
localUsers.push({ ordinal, item });
}
ordinal += 1;
}
return localUsers;
}
function mergeCanonicalUserWithLocalPresentation(
canonical: UserMessageItem,
local: UserMessageItem,
): UserMessageItem {
return {
kind: "user_message",
id: canonical.id,
...(canonical.clientMessageId ? { clientMessageId: canonical.clientMessageId } : {}),
text: local.text,
timestamp: local.timestamp,
...(local.images && local.images.length > 0 ? { images: local.images } : {}),
...(local.attachments && local.attachments.length > 0
? { attachments: local.attachments }
: {}),
tail,
head,
cursor,
cursorChanged: true,
sideEffects,
acknowledgedClientMessageIds,
};
}
interface CanonicalUserMessageIdentity {
messageId?: string;
clientMessageId?: string;
text: string;
}
function matchesLocalUserMessageIdentity(
canonical: CanonicalUserMessageIdentity,
optimistic: UserMessageItem,
): boolean {
if (canonical.clientMessageId !== undefined) {
return canonical.clientMessageId === optimistic.id;
}
if (canonical.messageId === optimistic.id) {
return true;
}
// COMPAT(userMessageClientId): added in v0.2.0, remove after 2027-01-20 once
// the supported daemon floor emits clientMessageId on submitted user messages.
return canonical.text.length > 0 && canonical.text === optimistic.text;
}
function reconcileLocalUserPresentationAfterReplace(params: {
canonicalTail: StreamItem[];
previousTail: StreamItem[];
previousHead: StreamItem[];
}): StreamItem[] {
const localUsers = collectLocallyPresentedUserMessages([
...params.previousTail,
...params.previousHead,
]);
if (localUsers.length === 0) {
return params.canonicalTail;
}
const canonicalUserIndexes: number[] = [];
params.canonicalTail.forEach((item, index) => {
if (item.kind === "user_message") {
canonicalUserIndexes.push(index);
}
});
const nextTail = [...params.canonicalTail];
const claimedCanonicalIndexes = new Set<number>();
const unmatched: UserMessageItem[] = [];
for (const local of localUsers) {
const exactIndex = canonicalUserIndexes.find((index) => {
if (claimedCanonicalIndexes.has(index)) return false;
const canonical = params.canonicalTail[index];
return (
canonical?.kind === "user_message" &&
matchesLocalUserMessageIdentity(
{
messageId: canonical.id,
clientMessageId: canonical.clientMessageId,
text: canonical.text,
},
local.item,
)
);
});
const ordinalIndex = canonicalUserIndexes[local.ordinal];
const ordinalItem = ordinalIndex === undefined ? undefined : params.canonicalTail[ordinalIndex];
const canonicalIndex =
exactIndex ??
(ordinalIndex !== undefined &&
!claimedCanonicalIndexes.has(ordinalIndex) &&
ordinalItem?.kind === "user_message" &&
ordinalItem.clientMessageId === undefined
? ordinalIndex
: undefined);
const canonicalItem = canonicalIndex === undefined ? undefined : nextTail[canonicalIndex];
if (canonicalIndex === undefined || !canonicalItem || canonicalItem.kind !== "user_message") {
if (local.item.optimistic) {
unmatched.push(local.item);
}
continue;
}
nextTail[canonicalIndex] = mergeCanonicalUserWithLocalPresentation(canonicalItem, local.item);
claimedCanonicalIndexes.add(canonicalIndex);
}
for (const item of unmatched) {
const insertionIndex = nextTail.findIndex(
(canonical) => canonical.timestamp.getTime() > item.timestamp.getTime(),
);
nextTail.splice(insertionIndex < 0 ? nextTail.length : insertionIndex, 0, item);
}
return nextTail;
}
interface IncrementalAcceptResult {
acceptedUnits: TimelineUnit[];
cursor: TimelineCursor | undefined;
@@ -515,6 +357,34 @@ function mergePrependedCanonicalTail(olderTail: StreamItem[], currentTail: Strea
return olderTail;
}
const remainingOlder: StreamItem[] = [];
let reconciledCurrent = currentTail;
for (const item of olderTail) {
if (item.kind !== "user_message") {
remainingOlder.push(item);
continue;
}
const result = upsertUserMessageAcrossStream({
tail: reconciledCurrent,
head: [],
message: item,
insert: "prepend-tail",
presentation: "existing",
});
if (result.location?.matched) {
remainingOlder.push(result.location.message);
reconciledCurrent = [
...result.tail.slice(0, result.location.index),
...result.tail.slice(result.location.index + 1),
];
} else {
remainingOlder.push(item);
}
}
olderTail = remainingOlder;
currentTail = reconciledCurrent;
if (olderTail.length === 0) return currentTail;
const olderLast = olderTail.at(-1);
const currentFirst = currentTail[0];
@@ -745,9 +615,24 @@ function applyCanonicalForwardUnit(params: {
head: StreamItem[];
unit: TimelineUnit;
epoch: string;
}): { tail: StreamItem[]; head: StreamItem[] } {
}): { tail: StreamItem[]; head: StreamItem[]; acknowledgedClientMessageIds: string[] } {
const { event, timestamp, seqEnd } = params.unit;
const timelineCursor = { epoch: params.epoch, seq: seqEnd };
if (event.type === "timeline" && event.item.type === "user_message") {
const applied = applyStreamEvent({
tail: params.tail,
head: params.head,
event,
timestamp,
source: "canonical",
timelineCursor,
});
return {
tail: applied.tail,
head: applied.head,
acknowledgedClientMessageIds: applied.acknowledgedClientMessageIds ?? [],
};
}
if (params.head.length === 0) {
return {
tail: reduceStreamUpdate(params.tail, event, timestamp, {
@@ -755,6 +640,7 @@ function applyCanonicalForwardUnit(params: {
timelineCursor,
}),
head: params.head,
acknowledgedClientMessageIds: [],
};
}
const replacedHead = replaceLiveAssistantWithProjectedText({
@@ -763,7 +649,9 @@ function applyCanonicalForwardUnit(params: {
timestamp,
timelineCursor,
});
if (replacedHead) return { tail: params.tail, head: replacedHead };
if (replacedHead) {
return { tail: params.tail, head: replacedHead, acknowledgedClientMessageIds: [] };
}
const activeAssistant = params.head.findLast(
(item): item is Extract<StreamItem, { kind: "assistant_message" }> =>
@@ -781,6 +669,7 @@ function applyCanonicalForwardUnit(params: {
source: "canonical",
timelineCursor,
}),
acknowledgedClientMessageIds: [],
};
}
@@ -792,7 +681,11 @@ function applyCanonicalForwardUnit(params: {
source: "canonical",
timelineCursor,
});
return { tail: applied.tail, head: applied.head };
return {
tail: applied.tail,
head: applied.head,
acknowledgedClientMessageIds: applied.acknowledgedClientMessageIds ?? [],
};
}
function applyAcceptedForwardTimelineUnits(params: {
@@ -801,7 +694,7 @@ function applyAcceptedForwardTimelineUnits(params: {
currentTail: StreamItem[];
currentHead: StreamItem[];
currentEndSeq: number | undefined;
}): { tail: StreamItem[]; head: StreamItem[] } {
}): { tail: StreamItem[]; head: StreamItem[]; acknowledgedClientMessageIds: string[] } {
const reconciled = reconcileOverlappingProjectedStreamItems({
tail: params.currentTail,
head: params.currentHead,
@@ -811,15 +704,19 @@ function applyAcceptedForwardTimelineUnits(params: {
});
let tail = reconciled.tail;
let head = reconciled.head;
const acknowledgedClientMessageIds = new Set<string>();
for (const unit of params.units) {
if (reconciled.reconciledUnits.has(unit)) continue;
const applied = applyCanonicalForwardUnit({ tail, head, unit, epoch: params.epoch });
tail = applied.tail;
head = applied.head;
for (const clientMessageId of applied.acknowledgedClientMessageIds) {
acknowledgedClientMessageIds.add(clientMessageId);
}
}
return { tail, head };
return { tail, head, acknowledgedClientMessageIds: [...acknowledgedClientMessageIds] };
}
function applyTimelineIncrementalPath(args: {
@@ -835,9 +732,17 @@ function applyTimelineIncrementalPath(args: {
let nextCursor: TimelineCursor | null | undefined = currentCursor;
let cursorChanged = false;
const sideEffects: TimelineReducerSideEffect[] = [];
let acknowledgedClientMessageIds: string[] = [];
if (timelineUnits.length === 0) {
return { tail: nextTail, head: nextHead, cursor: nextCursor, cursorChanged, sideEffects };
return {
tail: nextTail,
head: nextHead,
cursor: nextCursor,
cursorChanged,
sideEffects,
acknowledgedClientMessageIds,
};
}
const { acceptedUnits, cursor, gapCursor } =
@@ -874,6 +779,7 @@ function applyTimelineIncrementalPath(args: {
});
nextTail = applied.tail;
nextHead = applied.head;
acknowledgedClientMessageIds = applied.acknowledgedClientMessageIds;
}
}
@@ -892,7 +798,14 @@ function applyTimelineIncrementalPath(args: {
sideEffects.push({ type: "catch_up", cursor: gapCursor });
}
return { tail: nextTail, head: nextHead, cursor: nextCursor, cursorChanged, sideEffects };
return {
tail: nextTail,
head: nextHead,
cursor: nextCursor,
cursorChanged,
sideEffects,
acknowledgedClientMessageIds,
};
}
export function processTimelineResponse(
@@ -906,6 +819,7 @@ export function processTimelineResponse(
isInitializing,
hasActiveInitDeferred,
initRequestDirection,
sendingClientMessageIds,
} = input;
// ------------------------------------------------------------------
@@ -921,6 +835,7 @@ export function processTimelineResponse(
clearInitializing: isInitializing,
error: payload.error,
sideEffects: [],
acknowledgedClientMessageIds: [],
};
}
@@ -967,7 +882,6 @@ export function processTimelineResponse(
hasActiveInitDeferred,
});
const replace = bootstrapPolicy.replace;
const sideEffects: TimelineReducerSideEffect[] = [];
const timelineResult = replace
? applyTimelineReplacePath({
@@ -976,6 +890,8 @@ export function processTimelineResponse(
bootstrapPolicy,
currentTail,
currentHead,
sendingClientMessageIds,
preserveLiveHead: currentCursor?.epoch === payload.epoch,
toHydratedEvents,
})
: applyTimelineIncrementalPath({
@@ -1024,6 +940,7 @@ export function processTimelineResponse(
clearInitializing,
error: null,
sideEffects,
acknowledgedClientMessageIds: timelineResult.acknowledgedClientMessageIds,
};
}
@@ -1038,20 +955,9 @@ export interface ProcessAgentStreamEventInput {
currentTail: StreamItem[];
currentHead: StreamItem[];
currentCursor: TimelineCursor | undefined;
currentAgent: {
status: AgentLifecycleStatus;
updatedAt: Date;
lastActivityAt: Date;
} | null;
timestamp: Date;
}
export interface AgentPatch {
status: AgentLifecycleStatus;
updatedAt: Date;
lastActivityAt: Date;
}
export interface ProcessAgentStreamEventOutput {
tail: StreamItem[];
head: StreamItem[];
@@ -1059,8 +965,7 @@ export interface ProcessAgentStreamEventOutput {
changedHead: boolean;
cursor: TimelineCursor | null;
cursorChanged: boolean;
agent: AgentPatch | null;
agentChanged: boolean;
acknowledgedClientMessageIds: string[];
sideEffects: AgentStreamReducerSideEffect[];
}
@@ -1079,18 +984,11 @@ interface TimelineSequencingGateResult {
sideEffects: AgentStreamReducerSideEffect[];
}
export interface AgentStreamReducerAgentSnapshot {
status: AgentLifecycleStatus;
updatedAt: Date;
lastActivityAt: Date;
}
export interface ProcessAgentStreamEventsInput {
events: AgentStreamReducerEvent[];
currentTail: StreamItem[];
currentHead: StreamItem[];
currentCursor: TimelineCursor | undefined;
currentAgent: AgentStreamReducerAgentSnapshot | null;
}
export type AgentStreamReducerSnapshot = Omit<ProcessAgentStreamEventsInput, "events">;
@@ -1114,20 +1012,6 @@ export interface CreateAgentStreamReducerQueueInput {
cancelFlush: (id: number) => void;
}
function applyAgentPatch(
currentAgent: AgentStreamReducerAgentSnapshot | null,
patch: AgentPatch | null,
): AgentStreamReducerAgentSnapshot | null {
if (!currentAgent || !patch) {
return currentAgent;
}
return {
status: patch.status,
updatedAt: patch.updatedAt,
lastActivityAt: patch.lastActivityAt,
};
}
function processTimelineSequencingGate(input: {
event: AgentStreamEventPayload;
seq: number | undefined;
@@ -1201,8 +1085,7 @@ function processTimelineSequencingGate(input: {
export function processAgentStreamEvent(
input: ProcessAgentStreamEventInput,
): ProcessAgentStreamEventOutput {
const { event, seq, epoch, currentTail, currentHead, currentCursor, currentAgent, timestamp } =
input;
const { event, seq, epoch, currentTail, currentHead, currentCursor, timestamp } = input;
const sequencing = processTimelineSequencingGate({ event, seq, epoch, currentCursor });
const timelineCursor =
@@ -1213,7 +1096,7 @@ export function processAgentStreamEvent(
// ------------------------------------------------------------------
// Apply stream event to tail/head
// ------------------------------------------------------------------
const { tail, head, changedTail, changedHead } = sequencing.shouldApplyStreamEvent
const applied = sequencing.shouldApplyStreamEvent
? applyStreamEvent({
tail: sequencing.resetLiveTimeline ? [] : currentTail,
head: sequencing.resetLiveTimeline ? [] : currentHead,
@@ -1229,43 +1112,14 @@ export function processAgentStreamEvent(
changedHead: false,
};
// ------------------------------------------------------------------
// Optimistic lifecycle status
// ------------------------------------------------------------------
let agentPatch: AgentPatch | null = null;
let agentChanged = false;
if (
currentAgent &&
(event.type === "turn_completed" ||
event.type === "turn_canceled" ||
event.type === "turn_failed")
) {
const optimisticStatus = deriveOptimisticLifecycleStatus(currentAgent.status, event);
if (optimisticStatus) {
const nextUpdatedAtMs = Math.max(currentAgent.updatedAt.getTime(), timestamp.getTime());
const nextLastActivityAtMs = Math.max(
currentAgent.lastActivityAt.getTime(),
timestamp.getTime(),
);
agentPatch = {
status: optimisticStatus,
updatedAt: new Date(nextUpdatedAtMs),
lastActivityAt: new Date(nextLastActivityAtMs),
};
agentChanged = true;
}
}
return {
tail,
head,
changedTail,
changedHead,
tail: applied.tail,
head: applied.head,
changedTail: applied.changedTail,
changedHead: applied.changedHead,
cursor: sequencing.nextTimelineCursor,
cursorChanged: sequencing.cursorChanged,
agent: agentPatch,
agentChanged,
acknowledgedClientMessageIds: applied.acknowledgedClientMessageIds ?? [],
sideEffects: sequencing.sideEffects,
};
}
@@ -1276,12 +1130,10 @@ export function processAgentStreamEvents(
let tail = input.currentTail;
let head = input.currentHead;
let cursor = input.currentCursor;
let agent = input.currentAgent;
let changedTail = false;
let changedHead = false;
let cursorChanged = false;
let agentPatch: AgentPatch | null = null;
let agentChanged = false;
const acknowledgedClientMessageIds = new Set<string>();
const sideEffects: AgentStreamReducerSideEffect[] = [];
for (const reducerEvent of input.events) {
@@ -1292,7 +1144,6 @@ export function processAgentStreamEvents(
currentTail: tail,
currentHead: head,
currentCursor: cursor,
currentAgent: agent,
timestamp: reducerEvent.timestamp,
});
@@ -1301,17 +1152,14 @@ export function processAgentStreamEvents(
changedTail = changedTail || result.changedTail;
changedHead = changedHead || result.changedHead;
sideEffects.push(...result.sideEffects);
for (const clientMessageId of result.acknowledgedClientMessageIds) {
acknowledgedClientMessageIds.add(clientMessageId);
}
if (result.cursorChanged) {
cursor = result.cursor ?? undefined;
cursorChanged = true;
}
if (result.agentChanged) {
agentPatch = result.agent;
agentChanged = true;
agent = applyAgentPatch(agent, result.agent);
}
}
return {
@@ -1321,8 +1169,7 @@ export function processAgentStreamEvents(
changedHead,
cursor: cursor ?? null,
cursorChanged,
agent: agentPatch,
agentChanged,
acknowledgedClientMessageIds: [...acknowledgedClientMessageIds],
sideEffects,
};
}
@@ -1405,6 +1252,7 @@ export function createAgentStreamReducerQueue(
interface StreamStatePatch {
tail?: StreamItem[];
head?: StreamItem[];
acknowledgedClientMessageIds?: readonly string[];
}
export interface CreateSessionAgentStreamReducerQueueInput {
@@ -1414,7 +1262,6 @@ export interface CreateSessionAgentStreamReducerQueueInput {
serverId: string,
state: (prev: Map<string, TimelineCursor>) => Map<string, TimelineCursor>,
) => void;
setAgents: (serverId: string, state: (prev: Map<string, Agent>) => Map<string, Agent>) => void;
recoverTimelineGap: (agentId: string, cursor: { epoch: string; endSeq: number }) => void;
}
@@ -1429,31 +1276,29 @@ function cancelAgentStreamReducerFlush(id: number) {
export function createSessionAgentStreamReducerQueue(
input: CreateSessionAgentStreamReducerQueueInput,
): AgentStreamReducerQueue {
const { serverId, setAgentStreamState, setAgentTimelineCursor, setAgents, recoverTimelineGap } =
input;
const { serverId, setAgentStreamState, setAgentTimelineCursor, recoverTimelineGap } = input;
return createAgentStreamReducerQueue({
getSnapshot: (agentId) => {
const session = useSessionStore.getState().sessions[serverId];
const currentAgentEntry = session?.agents.get(agentId);
return {
currentTail: session?.agentStreamTail.get(agentId) ?? [],
currentHead: session?.agentStreamHead.get(agentId) ?? [],
currentCursor: session?.agentTimelineCursor.get(agentId),
currentAgent: currentAgentEntry
? {
status: currentAgentEntry.status,
updatedAt: currentAgentEntry.updatedAt,
lastActivityAt: currentAgentEntry.lastActivityAt,
}
: null,
};
},
commit: (agentId, result, events) => {
if (result.changedTail || result.changedHead) {
if (
result.changedTail ||
result.changedHead ||
result.acknowledgedClientMessageIds.length > 0
) {
setAgentStreamState(serverId, agentId, {
...(result.changedTail ? { tail: result.tail } : {}),
...(result.changedHead ? { head: result.head } : {}),
...(result.acknowledgedClientMessageIds.length > 0
? { acknowledgedClientMessageIds: result.acknowledgedClientMessageIds }
: {}),
});
}
@@ -1486,24 +1331,6 @@ export function createSessionAgentStreamReducerQueue(
return next;
});
}
if (result.agentChanged && result.agent) {
const nextAgent = result.agent;
setAgents(serverId, (prev) => {
const current = prev.get(agentId);
if (!current) {
return prev;
}
const next = new Map(prev);
next.set(agentId, {
...current,
status: nextAgent.status,
updatedAt: nextAgent.updatedAt,
lastActivityAt: nextAgent.lastActivityAt,
});
return next;
});
}
},
handleSideEffects: (agentId, sideEffects) => {
for (const effect of sideEffects) {
packages/app/src/timeline/turn-time.test.ts
+4 -22
View File
@@ -22,34 +22,16 @@ function assistant(id: string, timestamp: Date): StreamItem {
}
describe("deriveStreamTurnTiming", () => {
it("reserves a running footer for an optimistic prompt before the host starts the turn", () => {
const optimisticPrompt = {
...user("optimistic", new Date("2026-05-15T00:00:00.000Z")),
optimistic: true as const,
};
const timing = deriveStreamTurnTiming({
agentStatus: "idle",
tail: [],
head: [optimisticPrompt],
});
assert.equal(timing.isActive, true);
});
it("does not start elapsed time from an optimistic prompt", () => {
const optimisticPrompt = {
...user("optimistic", new Date("2026-05-15T00:00:00.000Z")),
optimistic: true as const,
};
it("starts elapsed time from the submitted prompt", () => {
const submittedAt = new Date("2026-05-15T00:00:00.000Z");
const timing = deriveStreamTurnTiming({
agentStatus: "running",
tail: [],
head: [optimisticPrompt],
head: [user("submitted", submittedAt)],
});
assert.equal(timing.runningStartedAt, null);
assert.equal(timing.runningStartedAt, submittedAt);
});
it("uses the last user message as the running turn start", () => {
packages/app/src/timeline/turn-time.ts
+1 -7
View File
@@ -9,7 +9,6 @@ export interface TurnTiming {
export interface StreamTurnTiming {
byAssistantId: Map<string, TurnTiming>;
runningStartedAt: Date | null;
isActive: boolean;
}
export function deriveStreamTurnTiming(params: {
@@ -19,8 +18,6 @@ export function deriveStreamTurnTiming(params: {
}): StreamTurnTiming {
const byAssistantId = new Map<string, TurnTiming>();
let currentUserAt: Date | null = null;
let currentAuthoritativeUserAt: Date | null = null;
let currentUserIsOptimistic = false;
let currentLastItemAt: Date | null = null;
let currentAssistantIds: string[] = [];
@@ -42,8 +39,6 @@ export function deriveStreamTurnTiming(params: {
if (item.kind === "user_message") {
flushCompletedTurn();
currentUserAt = item.timestamp;
currentAuthoritativeUserAt = item.optimistic ? null : item.timestamp;
currentUserIsOptimistic = item.optimistic === true;
currentLastItemAt = null;
currentAssistantIds = [];
return;
@@ -65,7 +60,7 @@ export function deriveStreamTurnTiming(params: {
}
const isRunning = params.agentStatus === "running";
const runningStartedAt = isRunning ? currentAuthoritativeUserAt : null;
const runningStartedAt = isRunning ? currentUserAt : null;
if (params.agentStatus !== "running") {
flushCompletedTurn();
}
@@ -73,6 +68,5 @@ export function deriveStreamTurnTiming(params: {
return {
byAssistantId,
runningStartedAt,
isActive: isRunning || currentUserIsOptimistic,
};
}