puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-06 16:34:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(app): keep submitted messages stable across sync

Browse Source 
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.
This commit is contained in:
Mohamed Boudra
2026-07-27 17:26:49 +02:00
parent 0c5a052980
commit 292cea4bcc
19 changed files with 599 additions and 892 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/e2e/agent-message-submission.spec.ts
+4 -14
View File
@@ -20,6 +20,7 @@ import { getServerId } from "./helpers/server-id";
import { buildHostWorkspaceRoute } from "@/utils/host-routes";
import { delayBrowserAgentCreatedStatus } from "./helpers/new-workspace";
import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate";
import { selectModel } from "./helpers/app";
const IMAGE = {
name: "message-submission.png",
@@ -438,7 +439,6 @@ async function expectCreatedAgentHandoff(
page: Page,
prompt: string,
userMessage: Locator,
submittedGeometry: MessageGeometry,
): Promise<void> {
await expect(page.getByTestId("turn-working-indicator")).toBeVisible();
await expect(page.getByTestId(/^workspace-tab-agent_/).first()).toBeVisible({ timeout: 30_000 });
@@ -446,14 +446,11 @@ async function expectCreatedAgentHandoff(
await expect(page.getByTestId("turn-working-indicator")).toBeVisible();
await expect(page.getByTestId("user-message").filter({ hasText: prompt })).toHaveCount(1);
await expect(userMessage.getByRole("button", { name: "Open image attachment" })).toBeVisible();
expect(await readMessageGeometry(page, userMessage)).toEqual(submittedGeometry);
}
interface DraftCreatePendingSubmission {
prompt: string;
userMessage: Locator;
submittedGeometry: MessageGeometry;
finishFooterContinuityCheck: () => Promise<void>;
}
async function beginDraftCreateSubmission(
@@ -461,14 +458,13 @@ async function beginDraftCreateSubmission(
scenario: DraftCreateScenario,
): Promise<DraftCreatePendingSubmission> {
await openWorkspaceDraft(page, scenario.workspaceId);
await selectModel(page, "one-minute-stream");
const prompt = "Keep this row through create handoff.";
const userMessage = await submitMessageWithImage(page, prompt);
await scenario.agentCreatedDelay.waitForCreateRequest();
await scenario.agentCreatedDelay.waitForDelayedCreatedStatus();
await expectPendingSubmission(page, userMessage);
const submittedGeometry = await readMessageGeometry(page, userMessage);
const finishFooterContinuityCheck = await beginWorkingFooterContinuityCheck(page);
return { prompt, userMessage, submittedGeometry, finishFooterContinuityCheck };
return { prompt, userMessage };
}
async function completeDraftCreateSubmission(
@@ -477,13 +473,7 @@ async function completeDraftCreateSubmission(
pending: DraftCreatePendingSubmission,
): Promise<void> {
scenario.agentCreatedDelay.release();
await expectCreatedAgentHandoff(
page,
pending.prompt,
pending.userMessage,
pending.submittedGeometry,
);
await pending.finishFooterContinuityCheck();
await expectCreatedAgentHandoff(page, pending.prompt, pending.userMessage);
}
test.describe("Agent message submission", () => {
packages/app/e2e/rewind-menu.ui-contract.spec.ts
+10 -10
View File
@@ -61,7 +61,10 @@ async function rewriteCachedMessageAsLegacyRow(page: Page, prompt: string): Prom
}, prompt);
}
async function waitForCurrentSubmissionInCache(page: Page, prompt: string): Promise<void> {
async function waitForCurrentSubmissionExcludedFromCache(
page: Page,
prompt: string,
): Promise<void> {
await expect
.poll(() =>
page.evaluate((messageText) => {
@@ -70,7 +73,7 @@ async function waitForCurrentSubmissionInCache(page: Page, prompt: string): Prom
const cache = JSON.parse(raw) as {
hosts?: Array<{ timeline?: { items?: Array<Record<string, unknown>> } | null }>;
};
return cache.hosts
return !cache.hosts
?.flatMap((host) => host.timeline?.items ?? [])
.some(
(item) =>
@@ -84,7 +87,7 @@ async function waitForCurrentSubmissionInCache(page: Page, prompt: string): Prom
.toBe(true);
}
async function expectPendingSubmissionNotRewindableAfterReload(page: Page): Promise<void> {
async function expectPendingSubmissionNotRestoredAfterReload(page: Page): Promise<void> {
const prompt = "Keep this cached submission pending.";
const gate = await installDaemonWebSocketGate(page);
const session = await seedMockAgentWorkspace({
@@ -98,14 +101,11 @@ async function expectPendingSubmissionNotRewindableAfterReload(page: Page): Prom
gate.holdNextClientRequest("send_agent_message_request");
await submitMessage(page, prompt);
await gate.waitForHeldClientRequest();
await waitForCurrentSubmissionInCache(page, prompt);
await waitForCurrentSubmissionExcludedFromCache(page, prompt);
await gate.drop();
await page.reload();
const restoredMessage = userMessage(page, prompt);
await expect(restoredMessage).toBeVisible();
await restoredMessage.hover();
await expect(restoredMessage.getByTestId("rewind-menu-trigger")).toHaveCount(0);
await expect(userMessage(page, prompt)).toHaveCount(0);
} finally {
gate.restore();
await session.cleanup();
@@ -113,8 +113,8 @@ async function expectPendingSubmissionNotRewindableAfterReload(page: Page): Prom
}
test.describe("Rewind sheet", () => {
test("does not restore a pending submission as a rewindable legacy message", async ({ page }) => {
await expectPendingSubmissionNotRewindableAfterReload(page);
test("does not restore a local-only submission from the display cache", async ({ page }) => {
await expectPendingSubmissionNotRestoredAfterReload(page);
});
test("keeps rewind available for a message restored from the legacy cache", async ({ page }) => {
packages/app/src/agent-stream/view.tsx
+14 -8
View File
@@ -240,7 +240,7 @@ export interface AgentStreamViewProps {
streamItems: StreamItem[];
streamHead?: StreamItem[];
pendingPermissions: Map<string, PendingPermission>;
pendingMessageSubmission?: PendingMessageSubmission | null;
pendingMessageSubmissions?: readonly PendingMessageSubmission[];
routeBottomAnchorRequest?: BottomAnchorRouteRequest | null;
isAuthoritativeHistoryReady?: boolean;
toast?: ToastApi | null;
@@ -267,6 +267,7 @@ const AGENT_CAPABILITY_FLAG_KEYS: (keyof AgentCapabilityFlags)[] = [
];
const EMPTY_STREAM_HEAD: StreamItem[] = [];
const EMPTY_PENDING_MESSAGE_SUBMISSIONS: readonly PendingMessageSubmission[] = [];
const GROUPED_TOOL_CALL_DETAIL_MAX_HEIGHT = 200;
function buildChatHistoryAttachment(input: {
@@ -329,7 +330,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
streamItems,
streamHead: providedStreamHead,
pendingPermissions,
pendingMessageSubmission = null,
pendingMessageSubmissions = EMPTY_PENDING_MESSAGE_SUBMISSIONS,
routeBottomAnchorRequest = null,
isAuthoritativeHistoryReady = true,
toast,
@@ -344,6 +345,10 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
const autoExpandReasoning = useSettings((settings) => settings.autoExpandReasoning);
const toolCallDetailLevel = useSettings((settings) => settings.toolCallDetailLevel);
const viewportRef = useRef<StreamViewportHandle | null>(null);
const pendingClientMessageIds = useMemo(
() => new Set(pendingMessageSubmissions.map((submission) => submission.clientMessageId)),
[pendingMessageSubmissions],
);
const isMobile = useIsCompactFormFactor();
const streamRenderStrategy = useMemo(
() =>
@@ -669,13 +674,13 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
isFirstInGroup={layoutItem.isFirstInUserGroup}
isLastInGroup={layoutItem.isLastInUserGroup}
isPending={
pendingMessageSubmission !== null &&
pendingMessageSubmission.clientMessageId === item.clientMessageId
item.clientMessageId !== undefined &&
pendingClientMessageIds.has(item.clientMessageId)
}
/>
);
},
[context.capabilities, agentId, client, pendingMessageSubmission, resolvedServerId],
[context.capabilities, agentId, client, pendingClientMessageIds, resolvedServerId],
);
const renderAssistantMessageItem = useCallback(
@@ -885,7 +890,8 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
[pendingPermissions, agentId],
);
const showRunningTurnFooter = context.status === "running" || pendingMessageSubmission !== null;
const showRunningTurnFooter =
context.status === "running" || pendingMessageSubmissions.length > 0;
const pendingPermissionsNode = useMemo(
() =>
renderPendingPermissionsNode({
@@ -1164,8 +1170,8 @@ function agentStreamViewPropsEqual(
if (left.streamItems !== right.streamItems) reasons.push("streamItems");
if (left.streamHead !== right.streamHead) reasons.push("streamHead");
if (left.pendingPermissions !== right.pendingPermissions) reasons.push("pendingPermissions");
if (left.pendingMessageSubmission !== right.pendingMessageSubmission) {
reasons.push("pendingMessageSubmission");
if (left.pendingMessageSubmissions !== right.pendingMessageSubmissions) {
reasons.push("pendingMessageSubmissions");
}
if (
!bottomAnchorRouteRequestsEqual(left.routeBottomAnchorRequest, right.routeBottomAnchorRequest)
packages/app/src/composer/actions.test.ts
+44 -27
View File
@@ -6,13 +6,16 @@ import type {
UserComposerAttachment,
WorkspaceComposerAttachment,
} from "@/attachments/types";
import type { StreamItem } from "@/types/stream";
import {
appendSubmittedUserMessage,
removeSubmittedUserMessage,
type StreamItem,
} from "@/types/stream";
import {
acceptMessageSubmission,
beginMessageSubmission,
rejectMessageSubmission,
type MessageSubmissionAcceptance,
type MessageSubmissionState,
type MessageSubmissionRecord,
} from "@/composer/submission/model";
import {
cancelComposerAgent,
@@ -195,47 +198,61 @@ function createFakeSendClient(
interface FakeStream extends MessageSubmissionWriter {
head: Map<string, StreamItem[]>;
tail: Map<string, StreamItem[]>;
acceptance: MessageSubmissionAcceptance;
}
function createFakeStream(
initialHead: Map<string, StreamItem[]> = new Map(),
options: { acceptance?: MessageSubmissionAcceptance } = {},
): FakeStream {
function createFakeStream(initialHead: Map<string, StreamItem[]> = new Map()): FakeStream {
const fake: FakeStream = {
head: new Map(initialHead),
tail: new Map(),
acceptance: options.acceptance ?? "rpc-and-running",
begin: (agentId, message) => {
const next = beginMessageSubmission(readSubmission(fake, agentId), {
const current = readSubmission(fake, agentId);
const stream = appendSubmittedUserMessage({
tail: current.tail,
head: current.head,
message,
submittedAt: message.timestamp,
acceptance: fake.acceptance,
});
writeSubmission(fake, agentId, next);
writeSubmission(fake, agentId, {
...stream,
submissions: beginMessageSubmission(current.submissions, {
clientMessageId: message.clientMessageId!,
submittedAt: message.timestamp,
}),
});
},
accept: (agentId, clientMessageId) => {
writeSubmission(
fake,
agentId,
acceptMessageSubmission(readSubmission(fake, agentId), clientMessageId),
);
const current = readSubmission(fake, agentId);
writeSubmission(fake, agentId, {
...current,
submissions: acceptMessageSubmission(current.submissions, clientMessageId, true),
});
},
reject: (agentId, clientMessageId) => {
const result = rejectMessageSubmission(readSubmission(fake, agentId), clientMessageId);
writeSubmission(fake, agentId, result.state);
const current = readSubmission(fake, agentId);
const result = rejectMessageSubmission(current.submissions, clientMessageId);
const stream =
result.outcome === "rejected"
? removeSubmittedUserMessage({
tail: current.tail,
head: current.head,
clientMessageId,
})
: current;
writeSubmission(fake, agentId, { ...stream, submissions: result.submissions });
return result.outcome;
},
};
return fake;
}
const submissionsByFakeStream = new WeakMap<
FakeStream,
Map<string, MessageSubmissionState["submissions"]>
>();
const submissionsByFakeStream = new WeakMap<FakeStream, Map<string, MessageSubmissionRecord[]>>();
function readSubmission(fake: FakeStream, agentId: string): MessageSubmissionState {
interface FakeSubmissionState {
tail: StreamItem[];
head: StreamItem[];
submissions: MessageSubmissionRecord[];
}
function readSubmission(fake: FakeStream, agentId: string): FakeSubmissionState {
return {
tail: fake.tail.get(agentId) ?? [],
head: fake.head.get(agentId) ?? [],
@@ -243,7 +260,7 @@ function readSubmission(fake: FakeStream, agentId: string): MessageSubmissionSta
};
}
function writeSubmission(fake: FakeStream, agentId: string, state: MessageSubmissionState): void {
function writeSubmission(fake: FakeStream, agentId: string, state: FakeSubmissionState): void {
fake.tail = new Map(fake.tail).set(agentId, state.tail);
fake.head = new Map(fake.head).set(agentId, state.head);
const submissions = submissionsByFakeStream.get(fake) ?? new Map();
@@ -405,7 +422,7 @@ describe("dispatchComposerAgentMessage", () => {
});
it("rolls back an already-running force send when its RPC fails", async () => {
const stream = createFakeStream(new Map(), { acceptance: "rpc-only" });
const stream = createFakeStream();
const transportError = new Error("Force send failed while the prior turn was running");
const client = createFakeSendClient({ rejection: transportError });
packages/app/src/composer/draft/create-flow.ts
+17 -7
View File
@@ -157,12 +157,14 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
}),
];
}, [machine]);
const pendingMessageSubmission = useMemo<PendingMessageSubmission | null>(() => {
if (machine.tag !== "creating") return null;
return {
clientMessageId: machine.attempt.clientMessageId,
submittedAt: machine.attempt.timestamp,
};
const pendingMessageSubmissions = useMemo<readonly PendingMessageSubmission[]>(() => {
if (machine.tag !== "creating") return [];
return [
{
clientMessageId: machine.attempt.clientMessageId,
submittedAt: machine.attempt.timestamp,
},
];
}, [machine]);
const draftAgent = useMemo<TDraftAgent | null>(() => {
@@ -212,6 +214,14 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
}),
);
markPendingCreateLifecycle({ draftId, lifecycle: "sent" });
if (
useSessionStore.getState().sessions[pendingServerId]?.agents.get(createResult.agentId)
?.status === "running"
) {
useCreateFlowStore
.getState()
.clearByAgent({ serverId: pendingServerId, agentId: createResult.agentId });
}
}
await onCreateSuccess({ result: createResult.result, attempt });
@@ -335,7 +345,7 @@ export function useDraftAgentCreateFlow<TDraftAgent, TCreateResult>({
formErrorMessage,
isSubmitting,
submittedStreamItems,
pendingMessageSubmission,
pendingMessageSubmissions,
draftAgent,
handleCreateFromInput,
continueCreateFromAttempt,
packages/app/src/composer/draft/workspace-tab.tsx
+2 -2
View File
@@ -480,7 +480,7 @@ export function WorkspaceDraftAgentTab({
formErrorMessage,
isSubmitting,
submittedStreamItems,
pendingMessageSubmission,
pendingMessageSubmissions,
draftAgent,
handleCreateFromInput,
continueCreateFromAttempt,
@@ -644,7 +644,7 @@ export function WorkspaceDraftAgentTab({
serverId={serverId}
context={draftAgent}
streamItems={submittedStreamItems}
pendingMessageSubmission={pendingMessageSubmission}
pendingMessageSubmissions={pendingMessageSubmissions}
pendingPermissions={EMPTY_PENDING_PERMISSIONS}
onOpenWorkspaceFile={onOpenWorkspaceFile}
/>
packages/app/src/composer/submission/model.test.ts
+81 -217
View File
@@ -1,247 +1,111 @@
import { describe, expect, it } from "vitest";
import { createUserMessage, type StreamItem } from "@/types/stream";
import {
acceptMessageSubmission,
beginMessageSubmission,
getPendingMessageSubmission,
observeMessageSubmissionRunning,
getActiveMessageSubmissions,
getSendingClientMessageIds,
observeAcceptedMessageSubmissionsRunning,
observeMessageSubmissionCanonical,
rejectMessageSubmission,
type MessageSubmissionState,
} from "./model";
function initialState(head: StreamItem[] = []): MessageSubmissionState {
return { tail: [], head, submissions: [] };
}
const submittedAt = new Date("2026-07-26T10:00:00.000Z");
describe("message submission model", () => {
it("waits for authoritative running after RPC acceptance", () => {
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const begun = beginMessageSubmission(initialState(), {
message,
submittedAt: message.timestamp,
acceptance: "rpc-and-running",
describe("message submission transactions", () => {
it("tracks every in-flight submission independently", () => {
const first = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
const both = beginMessageSubmission(first, {
clientMessageId: "client-2",
submittedAt: new Date(submittedAt.getTime() + 1),
});
const accepted = acceptMessageSubmission(begun, "client-1");
const runningObserved = observeMessageSubmissionRunning(accepted.submissions);
expect(getActiveMessageSubmissions(both).map((item) => item.clientMessageId)).toEqual([
"client-1",
"client-2",
]);
expect(getSendingClientMessageIds(both)).toEqual(["client-1", "client-2"]);
});
expect(accepted.submissions).toEqual([
it("removes only the RPC-accepted transaction", () => {
const both = beginMessageSubmission(
beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }),
{ clientMessageId: "client-2", submittedAt },
);
expect(acceptMessageSubmission(both, "client-1", true)).toEqual([
{
clientMessageId: "client-1",
submittedAt: message.timestamp,
phase: "waiting-for-running",
clientMessageId: "client-2",
submittedAt,
rpcAccepted: false,
providerAcknowledged: false,
},
]);
expect(getPendingMessageSubmission(accepted.submissions)).toEqual(accepted.submissions[0]);
expect(runningObserved).toEqual([]);
});
it("closes RPC-accepted pending state when canonical history proves acceptance", () => {
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const begun = beginMessageSubmission(initialState(), {
message,
submittedAt: message.timestamp,
acceptance: "rpc-and-running",
});
const accepted = acceptMessageSubmission(begun, "client-1");
it("bridges an accepted RPC until the correlated running state is observed", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
const accepted = acceptMessageSubmission(sending, "client-1", false);
expect(observeMessageSubmissionCanonical(accepted.submissions, ["client-1"])).toEqual([]);
expect(getActiveMessageSubmissions(accepted)).toHaveLength(1);
expect(accepted[0].rpcAccepted).toBe(true);
expect(observeAcceptedMessageSubmissionsRunning(accepted)).toEqual([]);
});
it("remembers canonical acceptance until an in-flight RPC rejection settles", () => {
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const begun = beginMessageSubmission(initialState(), {
message,
submittedAt: message.timestamp,
acceptance: "rpc-and-running",
});
const canonicalObserved = observeMessageSubmissionCanonical(begun.submissions, ["client-1"]);
it("settles an accepted RPC when provider acknowledgement arrives after running was missed", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
const accepted = acceptMessageSubmission(sending, "client-1", false);
expect(getPendingMessageSubmission(canonicalObserved)).toBeNull();
expect(
rejectMessageSubmission({ ...begun, submissions: canonicalObserved }, "client-1"),
).toEqual({
outcome: "accepted",
state: { ...begun, submissions: [] },
});
expect(observeMessageSubmissionCanonical(accepted, ["client-1"])).toEqual([]);
});
it("rejects a row when an unrelated running transition precedes its RPC error", () => {
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const begun = beginMessageSubmission(initialState(), {
message,
submittedAt: message.timestamp,
acceptance: "rpc-and-running",
});
const runningObserved = observeMessageSubmissionRunning(begun.submissions);
const rejected = rejectMessageSubmission(
{ ...begun, submissions: runningObserved },
"client-1",
it("records provider acknowledgement without settling another transaction", () => {
const both = beginMessageSubmission(
beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }),
{ clientMessageId: "client-2", submittedAt },
);
const observed = observeMessageSubmissionCanonical(both, ["client-1"]);
expect(runningObserved).toBe(begun.submissions);
expect(getPendingMessageSubmission(runningObserved)).toEqual(runningObserved[0]);
expect(rejected).toEqual({
outcome: "rejected",
state: { tail: [], head: [], submissions: [] },
});
});
it("does not accept an already-running force send from agent status alone", () => {
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const begun = beginMessageSubmission(initialState(), {
message,
submittedAt: message.timestamp,
acceptance: "rpc-only",
});
expect(observeMessageSubmissionRunning(begun.submissions)).toBe(begun.submissions);
});
it("begins and accepts without changing the submitted row", () => {
const submittedAt = new Date("2026-07-26T10:00:00.000Z");
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: submittedAt,
});
const begun = beginMessageSubmission(initialState(), {
message,
submittedAt,
acceptance: "rpc-only",
});
const accepted = acceptMessageSubmission(begun, "client-1");
expect(begun).toEqual({
tail: [message],
head: [],
submissions: [
{
clientMessageId: "client-1",
submittedAt,
phase: "waiting-for-rpc",
acceptance: "rpc-only",
},
],
});
expect(accepted.tail).toBe(begun.tail);
expect(accepted.head).toBe(begun.head);
expect(accepted.submissions).toEqual([]);
});
it("rejects by removing the submitted row and closing pending", () => {
const submittedAt = new Date("2026-07-26T10:00:00.000Z");
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: submittedAt,
});
const existingHead = createUserMessage({
id: "provider-prior",
messageId: "provider-prior",
text: "prior",
timestamp: new Date(0),
});
const begun = beginMessageSubmission(initialState([existingHead]), {
message,
submittedAt,
acceptance: "rpc-and-running",
});
expect(rejectMessageSubmission(begun, "client-1")).toEqual({
outcome: "rejected",
state: {
tail: [],
head: [existingHead],
submissions: [],
},
});
});
it("treats RPC acceptance waiting for running as authoritatively accepted", () => {
const message = createUserMessage({
clientMessageId: "client-1",
text: "hello",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const state = initialState([message]);
const waitingForRunning = {
...state,
submissions: [
{
clientMessageId: "client-1",
submittedAt: message.timestamp,
phase: "waiting-for-running" as const,
},
],
};
expect(rejectMessageSubmission(waitingForRunning, "client-1")).toEqual({
outcome: "accepted",
state: { ...waitingForRunning, submissions: [] },
});
});
it("reports unknown when neither pending nor its submitted row exists", () => {
const state = initialState();
expect(rejectMessageSubmission(state, "client-1")).toEqual({
outcome: "unknown",
state,
});
});
});
describe("submitting alongside repeated history", () => {
it("appends a resubmitted prompt instead of adopting an identical older row", () => {
// A row from a daemon that never recorded clientMessageId, e.g. history created
// before the field existed. Sending the same text again must be a new message.
const olderRow = createUserMessage({
id: "provider-1",
messageId: "provider-1",
text: "continue",
timestamp: new Date("2026-07-26T10:00:00.000Z"),
});
const resubmitted = createUserMessage({
clientMessageId: "client-2",
text: "continue",
timestamp: new Date("2026-07-26T10:05:00.000Z"),
attachments: [{ type: "text", mimeType: "text/plain", text: "attachment" }],
});
const begun = beginMessageSubmission(
{ tail: [olderRow], head: [], submissions: [] },
expect(observed).toEqual([
{
message: resubmitted,
submittedAt: resubmitted.timestamp,
acceptance: "rpc-and-running",
clientMessageId: "client-1",
submittedAt,
rpcAccepted: false,
providerAcknowledged: true,
},
);
{
clientMessageId: "client-2",
submittedAt,
rpcAccepted: false,
providerAcknowledged: false,
},
]);
expect(getSendingClientMessageIds(observed)).toEqual(["client-2"]);
});
expect(begun.tail).toEqual([olderRow, resubmitted]);
it("does not roll back a provider-acknowledged prompt on a later transport error", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
const observed = observeMessageSubmissionCanonical(sending, ["client-1"]);
expect(rejectMessageSubmission(observed, "client-1")).toEqual({
outcome: "accepted",
submissions: [],
});
});
it("rejects an unacknowledged transaction", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
expect(rejectMessageSubmission(sending, "client-1")).toEqual({
outcome: "rejected",
submissions: [],
});
});
it("does not create duplicate transaction identity", () => {
const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt });
expect(() =>
beginMessageSubmission(sending, { clientMessageId: "client-1", submittedAt }),
).toThrow("Message submission already exists");
});
});
packages/app/src/composer/submission/model.ts
+44 -109
View File
@@ -1,105 +1,68 @@
import type { StreamItem, UserMessageItem } from "@/types/stream";
export interface PendingMessageSubmission {
clientMessageId: string;
submittedAt: Date;
}
export type MessageSubmissionAcceptance = "rpc-only" | "rpc-and-running";
export type MessageSubmissionRecord = PendingMessageSubmission & {
rpcAccepted: boolean;
providerAcknowledged: boolean;
};
export type MessageSubmissionRecord = PendingMessageSubmission &
(
| { phase: "waiting-for-rpc"; acceptance: MessageSubmissionAcceptance }
| { phase: "waiting-for-running" }
| { phase: "canonical-observed" }
);
const EMPTY_MESSAGE_SUBMISSIONS: readonly MessageSubmissionRecord[] = [];
export function getPendingMessageSubmission(
export function getActiveMessageSubmissions(
submissions: readonly MessageSubmissionRecord[] | null | undefined,
): PendingMessageSubmission | null {
return submissions?.findLast((submission) => submission.phase !== "canonical-observed") ?? null;
): readonly PendingMessageSubmission[] {
return submissions ?? EMPTY_MESSAGE_SUBMISSIONS;
}
export interface MessageSubmissionState {
tail: StreamItem[];
head: StreamItem[];
submissions: MessageSubmissionRecord[];
export function getSendingClientMessageIds(
submissions: readonly MessageSubmissionRecord[] | null | undefined,
): string[] {
return (submissions ?? [])
.filter((submission) => !submission.providerAcknowledged)
.map((submission) => submission.clientMessageId);
}
export type MessageSubmissionRejectionOutcome = "rejected" | "accepted" | "unknown";
export interface MessageSubmissionRejectionResult {
state: MessageSubmissionState;
submissions: MessageSubmissionRecord[];
outcome: MessageSubmissionRejectionOutcome;
}
export function beginMessageSubmission(
state: MessageSubmissionState,
input: {
message: UserMessageItem;
submittedAt: Date;
acceptance: MessageSubmissionAcceptance;
},
): MessageSubmissionState {
if (!input.message.clientMessageId) {
throw new Error("Submitted user message requires client identity");
submissions: readonly MessageSubmissionRecord[],
input: PendingMessageSubmission,
): MessageSubmissionRecord[] {
if (submissions.some((submission) => submission.clientMessageId === input.clientMessageId)) {
throw new Error(`Message submission already exists: ${input.clientMessageId}`);
}
const pending: MessageSubmissionRecord = {
clientMessageId: input.message.clientMessageId,
submittedAt: input.submittedAt,
phase: "waiting-for-rpc",
acceptance: input.acceptance,
};
// A submission carries a freshly generated client identity, so it cannot already be in
// the timeline. Append it. Reconciling here would let the legacy content fallback adopt
// an older row that happens to repeat the same text.
return state.head.length > 0
? {
tail: state.tail,
head: [...state.head, input.message],
submissions: [...state.submissions, pending],
}
: {
tail: [...state.tail, input.message],
head: state.head,
submissions: [...state.submissions, pending],
};
return [...submissions, { ...input, rpcAccepted: false, providerAcknowledged: false }];
}
export function acceptMessageSubmission(
state: MessageSubmissionState,
submissions: readonly MessageSubmissionRecord[],
clientMessageId: string,
isAgentRunning = false,
): MessageSubmissionState {
const index = state.submissions.findIndex(
isAgentRunning: boolean,
): MessageSubmissionRecord[] {
const index = submissions.findIndex(
(submission) => submission.clientMessageId === clientMessageId,
);
if (index === -1) return state;
const submission = state.submissions[index];
if (submission.phase === "canonical-observed") {
const submissions = state.submissions.slice();
submissions.splice(index, 1);
return { ...state, submissions };
if (index < 0) return submissions as MessageSubmissionRecord[];
if (isAgentRunning) {
return submissions.filter((_, submissionIndex) => submissionIndex !== index);
}
if (submission.phase === "waiting-for-running") return state;
const nextSubmission: MessageSubmissionRecord | null =
submission.acceptance === "rpc-only" || isAgentRunning
? null
: {
clientMessageId: submission.clientMessageId,
submittedAt: submission.submittedAt,
phase: "waiting-for-running",
};
const submissions = state.submissions.slice();
if (nextSubmission) submissions[index] = nextSubmission;
else submissions.splice(index, 1);
return { ...state, submissions };
if (submissions[index].rpcAccepted) return submissions as MessageSubmissionRecord[];
const next = submissions.slice();
next[index] = { ...next[index], rpcAccepted: true };
return next;
}
export function observeMessageSubmissionRunning(
export function observeAcceptedMessageSubmissionsRunning(
submissions: readonly MessageSubmissionRecord[],
): MessageSubmissionRecord[] {
const next = submissions.filter((submission) => submission.phase !== "waiting-for-running");
const next = submissions.filter((submission) => !submission.rpcAccepted);
return next.length === submissions.length ? (submissions as MessageSubmissionRecord[]) : next;
}
@@ -111,53 +74,25 @@ export function observeMessageSubmissionCanonical(
const observed = new Set(clientMessageIds);
let changed = false;
const next = submissions.flatMap((submission): MessageSubmissionRecord[] => {
if (!observed.has(submission.clientMessageId)) return [submission];
if (submission.phase === "waiting-for-running") {
changed = true;
return [];
if (submission.providerAcknowledged || !observed.has(submission.clientMessageId)) {
return [submission];
}
if (submission.phase === "waiting-for-rpc") {
changed = true;
return [
{
clientMessageId: submission.clientMessageId,
submittedAt: submission.submittedAt,
phase: "canonical-observed",
},
];
}
return [submission];
changed = true;
return submission.rpcAccepted ? [] : [{ ...submission, providerAcknowledged: true }];
});
return changed ? next : (submissions as MessageSubmissionRecord[]);
}
function removeSubmittedMessage(items: StreamItem[], clientMessageId: string): StreamItem[] {
const next = items.filter(
(item) => item.kind !== "user_message" || item.clientMessageId !== clientMessageId,
);
return next.length === items.length ? items : next;
}
export function rejectMessageSubmission(
state: MessageSubmissionState,
submissions: readonly MessageSubmissionRecord[],
clientMessageId: string,
): MessageSubmissionRejectionResult {
const index = state.submissions.findIndex(
(submission) => submission.clientMessageId === clientMessageId,
);
if (index === -1) return { outcome: "unknown", state };
const submission = state.submissions[index];
const submissions = state.submissions.slice();
submissions.splice(index, 1);
if (submission.phase === "waiting-for-running" || submission.phase === "canonical-observed") {
return { outcome: "accepted", state: { ...state, submissions } };
const submission = submissions.find((item) => item.clientMessageId === clientMessageId);
if (!submission) {
return { outcome: "unknown", submissions: submissions as MessageSubmissionRecord[] };
}
return {
outcome: "rejected",
state: {
tail: removeSubmittedMessage(state.tail, clientMessageId),
head: removeSubmittedMessage(state.head, clientMessageId),
submissions,
},
outcome: submission.providerAcknowledged || submission.rpcAccepted ? "accepted" : "rejected",
submissions: submissions.filter((item) => item.clientMessageId !== clientMessageId),
};
}
packages/app/src/contexts/session-context.tsx
+23 -8
View File
@@ -53,7 +53,7 @@ import {
} from "@/utils/agent-initialization";
import { encodeImages } from "@/utils/encode-images";
import { derivePendingPermissionKey } from "@/utils/agent-snapshots";
import { getPendingMessageSubmission } from "@/composer/submission/model";
import { getSendingClientMessageIds } from "@/composer/submission/model";
import type { AttachmentMetadata } from "@/attachments/types";
import { patchWorkspaceScripts } from "@/contexts/session-workspace-scripts";
import { useToast } from "@/contexts/toast-context";
@@ -248,10 +248,17 @@ function applyTimelineStreamPatches(input: {
setAgentTimelineCursor,
} = input;
if (result.tail !== currentTail || result.head !== currentHead) {
if (
result.tail !== currentTail ||
result.head !== currentHead ||
result.acknowledgedClientMessageIds.length > 0
) {
setAgentStreamState(serverId, agentId, {
...(result.tail !== currentTail ? { tail: result.tail } : {}),
...(result.head !== currentHead ? { head: result.head } : {}),
...(result.acknowledgedClientMessageIds.length > 0
? { acknowledgedClientMessageIds: result.acknowledgedClientMessageIds }
: {}),
});
}
@@ -320,10 +327,12 @@ function finalizeTimelineApplication(input: {
}
if (shouldMarkAuthoritativeHistoryApplied) {
setAgentAuthoritativeHistoryApplied(serverId, agentId, true);
useCreateFlowStore.getState().clearByAgent({ serverId, agentId });
markAgentHistorySynchronized(serverId, agentId);
const session = useSessionStore.getState().sessions[serverId];
const agent = session?.agents.get(agentId) ?? session?.agentDetails.get(agentId);
if (agent?.status === "running") {
useCreateFlowStore.getState().clearByAgent({ serverId, agentId });
}
if (agent && agent.status !== "running") {
getHostRuntimeStore().drainQueuedAgentMessage(serverId, agentId);
}
@@ -640,9 +649,9 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const currentCursor = session?.agentTimelineCursor.get(agentId);
const currentTail = session?.agentStreamTail.get(agentId) ?? [];
const currentHead = session?.agentStreamHead.get(agentId) ?? [];
const pendingClientMessageId =
getPendingMessageSubmission(session?.messageSubmissions.get(agentId))?.clientMessageId ??
null;
const sendingClientMessageIds = getSendingClientMessageIds(
session?.messageSubmissions.get(agentId),
);
setAgentTimelineHasOlder(serverId, (prev) => {
if (prev.get(agentId) === payload.hasOlder) {
@@ -662,7 +671,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
isInitializing,
hasActiveInitDeferred,
initRequestDirection: activeInitDeferred?.requestDirection ?? "tail",
pendingClientMessageId,
sendingClientMessageIds,
});
if (result.error) {
@@ -790,7 +799,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
serverId,
setAgentStreamState,
setAgentTimelineCursor,
setAgents,
recoverTimelineGap,
});
@@ -807,6 +815,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
) {
voiceRuntime?.onTurnEvent(serverId, agentId, event.type);
}
if (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
) {
useCreateFlowStore.getState().clearByAgent({ serverId, agentId });
}
agentStreamReducerQueue.enqueue(agentId, {
event: streamEvent,
packages/app/src/panels/agent-panel.tsx
+6 -5
View File
@@ -25,7 +25,7 @@ import { FileDropZone } from "@/components/file-drop/file-drop-zone";
import { useRetainedPanelActive } from "@/components/retained-panel";
import { SidebarCallout } from "@/components/sidebar-callout";
import { Composer } from "@/composer";
import { getPendingMessageSubmission } from "@/composer/submission/model";
import { getActiveMessageSubmissions } from "@/composer/submission/model";
import { RewindComposerRestoreProvider } from "@/components/rewind/composer-restore";
import { getProviderIcon } from "@/components/provider-icons";
import {
@@ -443,6 +443,7 @@ export function useDraftPanelDescriptor(
}
const EMPTY_STREAM_ITEMS: StreamItem[] = [];
const EMPTY_MESSAGE_SUBMISSIONS = [] as const;
const EMPTY_PENDING_PERMISSIONS = new Map<string, PendingPermission>();
const EMPTY_PENDING_PERMISSION_LIST: PendingPermission[] = [];
@@ -1289,10 +1290,10 @@ const AgentStreamSection = memo(function AgentStreamSection({
const streamItemsRaw = useSessionStore((state) =>
agentId ? state.sessions[serverId]?.agentStreamTail?.get(agentId) : undefined,
);
const pendingMessageSubmission = useSessionStore((state) =>
const pendingMessageSubmissions = useSessionStore((state) =>
agentId
? getPendingMessageSubmission(state.sessions[serverId]?.messageSubmissions.get(agentId))
: null,
? getActiveMessageSubmissions(state.sessions[serverId]?.messageSubmissions.get(agentId))
: EMPTY_MESSAGE_SUBMISSIONS,
);
const streamItems = streamItemsRaw ?? EMPTY_STREAM_ITEMS;
const pendingPermissionList = useStoreWithEqualityFn(
@@ -1333,7 +1334,7 @@ const AgentStreamSection = memo(function AgentStreamSection({
routeBottomAnchorRequest={routeBottomAnchorRequest}
isAuthoritativeHistoryReady={hasAppliedAuthoritativeHistory}
toast={toast}
pendingMessageSubmission={pendingMessageSubmission}
pendingMessageSubmissions={pendingMessageSubmissions}
onOpenWorkspaceFile={onOpenWorkspaceFile}
/>
);
packages/app/src/runtime/replica-cache/index.ts
+9 -5
View File
@@ -129,10 +129,10 @@ function decodeDates(value: unknown): unknown {
return Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, decodeDates(entry)]));
}
function restoreCachedStreamItem(item: StreamItem): StreamItem {
if (item.kind !== "user_message" || item.messageId || item.clientMessageId) return item;
function restoreCachedStreamItem(item: StreamItem): StreamItem | null {
if (item.kind !== "user_message" || item.messageId) return item;
const legacyItem = item as StreamItem & { optimistic?: true };
if (legacyItem.optimistic) return item;
if (legacyItem.optimistic || item.clientMessageId) return null;
// COMPAT(replicaCacheUserMessageId): before v0.2.3, cache v1 stored the provider ID
// only in `id`. Remove after 2027-01-27, once cache v1 predating v0.2.3 is unsupported.
return { ...item, messageId: item.id };
@@ -148,7 +148,7 @@ function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["ti
}
return {
agentId: stored.agentId,
items: decoded.map(restoreCachedStreamItem),
items: decoded.flatMap((item) => restoreCachedStreamItem(item) ?? []),
cursor: stored.cursor,
hasOlder: stored.hasOlder,
};
@@ -379,7 +379,11 @@ export class ReplicaCache {
(workspace) => workspace.workspaceDirectory === focusedAgent.cwd,
))
: undefined;
const items = focusedAgentId ? session.agentStreamTail.get(focusedAgentId) : undefined;
const items = focusedAgentId
? session.agentStreamTail
.get(focusedAgentId)
?.filter((item) => item.kind !== "user_message" || item.messageId !== undefined)
: undefined;
const timeline =
focusedAgent && items
? {
packages/app/src/stores/create-flow-store.ts
+4 -1
View File
@@ -109,7 +109,10 @@ export const useCreateFlowStore = create<CreateFlowState>((set) => ({
set((state) => {
const next = Object.fromEntries(
Object.entries(state.pendingByDraftId).filter(
([, pending]) => pending.serverId !== serverId || pending.agentId !== agentId,
([, pending]) =>
pending.lifecycle !== "sent" ||
pending.serverId !== serverId ||
pending.agentId !== agentId,
),
);
if (Object.keys(next).length === Object.keys(state.pendingByDraftId).length) {
packages/app/src/stores/session-store.ts
+60 -74
View File
@@ -5,14 +5,16 @@ import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
import type { ViewedTimelineUiBridge } from "@/timeline/viewed-timeline-sync";
import type { AgentDirectoryEntry } from "@/types/agent-directory";
import {
appendSubmittedUserMessage,
handoffCreatedAgentUserMessageToStream,
removeSubmittedUserMessage,
type StreamItem,
type UserMessageItem,
} from "@/types/stream";
import {
acceptMessageSubmission,
beginMessageSubmission,
observeMessageSubmissionRunning,
observeAcceptedMessageSubmissionsRunning,
observeMessageSubmissionCanonical,
rejectMessageSubmission,
type MessageSubmissionRecord,
@@ -469,7 +471,11 @@ interface SessionStoreActions {
setAgentStreamState: (
serverId: string,
agentId: string,
state: { tail?: StreamItem[]; head?: StreamItem[] },
state: {
tail?: StreamItem[];
head?: StreamItem[];
acknowledgedClientMessageIds?: readonly string[];
},
) => void;
beginAgentMessageSubmission: (
serverId: string,
@@ -592,35 +598,23 @@ type SessionStore = SessionStoreState & SessionStoreActions;
const agentLastActivityCoalescer = createAgentLastActivityCoalescer();
function applyAgentStateToMessageSubmissions(input: {
function applyRunningAgentsToAcceptedSubmissions(input: {
previousAgents: Map<string, Agent>;
nextAgents: Map<string, Agent>;
submissions: Map<string, MessageSubmissionRecord[]>;
}): Map<string, MessageSubmissionRecord[]> {
let nextSubmissions = input.submissions;
const write = (agentId: string, submissions: MessageSubmissionRecord[]): void => {
if (nextSubmissions === input.submissions) nextSubmissions = new Map(input.submissions);
if (submissions.length > 0) {
nextSubmissions.set(agentId, submissions);
} else {
nextSubmissions.delete(agentId);
}
};
for (const [agentId, submissions] of input.submissions) {
const previousAgent = input.previousAgents.get(agentId);
const nextAgent = input.nextAgents.get(agentId);
if (!nextAgent) {
write(agentId, []);
if (!nextAgent || previousAgent?.status === "running" || nextAgent.status !== "running") {
continue;
}
const previousAgent = input.previousAgents.get(agentId);
const observedRunningTransition =
previousAgent !== undefined &&
previousAgent.status !== "running" &&
nextAgent.status === "running";
if (!observedRunningTransition) continue;
const nextSubmission = observeMessageSubmissionRunning(submissions);
if (nextSubmission !== submissions) write(agentId, nextSubmission);
const remaining = observeAcceptedMessageSubmissionsRunning(submissions);
if (remaining === submissions) continue;
if (nextSubmissions === input.submissions) nextSubmissions = new Map(input.submissions);
if (remaining.length > 0) nextSubmissions.set(agentId, remaining);
else nextSubmissions.delete(agentId);
}
return nextSubmissions;
}
@@ -1106,18 +1100,10 @@ export const useSessionStore = create<SessionStore>()(
}
}
const canonicalClientMessageIds = [
...(state.tail ?? session.agentStreamTail.get(agentId) ?? []),
...(state.head ?? session.agentStreamHead.get(agentId) ?? []),
].flatMap((item) =>
item.kind === "user_message" && item.clientMessageId && item.messageId
? [item.clientMessageId]
: [],
);
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
const observedSubmissions = observeMessageSubmissionCanonical(
currentSubmissions,
canonicalClientMessageIds,
state.acknowledgedClientMessageIds ?? [],
);
const changedSubmissions = observedSubmissions !== currentSubmissions;
@@ -1154,36 +1140,36 @@ export const useSessionStore = create<SessionStore>()(
set((prev) => {
const session = prev.sessions[serverId];
if (!session) return prev;
const submittedAt = message.timestamp;
const currentAgent = session.agents.get(agentId);
const next = beginMessageSubmission(
{
tail: session.agentStreamTail.get(agentId) ?? [],
head: session.agentStreamHead.get(agentId) ?? [],
submissions: session.messageSubmissions.get(agentId) ?? [],
},
{
message,
submittedAt,
acceptance:
currentAgent !== undefined && currentAgent.status !== "running"
? "rpc-and-running"
: "rpc-only",
},
if (!message.clientMessageId) {
throw new Error("Beginning a message submission requires client identity");
}
const currentTail = session.agentStreamTail.get(agentId) ?? [];
const currentHead = session.agentStreamHead.get(agentId) ?? [];
const stream = appendSubmittedUserMessage({
tail: currentTail,
head: currentHead,
message,
});
const submissions = beginMessageSubmission(
session.messageSubmissions.get(agentId) ?? [],
{ clientMessageId: message.clientMessageId, submittedAt: message.timestamp },
);
const messageSubmissions = new Map(session.messageSubmissions);
if (next.submissions.length === 0) {
throw new Error("Beginning a message submission must create lifecycle state");
}
messageSubmissions.set(agentId, next.submissions);
messageSubmissions.set(agentId, submissions);
return {
...prev,
sessions: {
...prev.sessions,
[serverId]: {
...session,
agentStreamTail: new Map(session.agentStreamTail).set(agentId, next.tail),
agentStreamHead: new Map(session.agentStreamHead).set(agentId, next.head),
agentStreamTail:
stream.tail === currentTail
? session.agentStreamTail
: new Map(session.agentStreamTail).set(agentId, stream.tail),
agentStreamHead:
stream.head === currentHead
? session.agentStreamHead
: new Map(session.agentStreamHead).set(agentId, stream.head),
messageSubmissions,
},
},
@@ -1196,19 +1182,15 @@ export const useSessionStore = create<SessionStore>()(
const session = prev.sessions[serverId];
if (!session) return prev;
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
const next = acceptMessageSubmission(
{
tail: session.agentStreamTail.get(agentId) ?? [],
head: session.agentStreamHead.get(agentId) ?? [],
submissions: currentSubmissions,
},
const submissions = acceptMessageSubmission(
currentSubmissions,
clientMessageId,
session.agents.get(agentId)?.status === "running",
);
if (next.submissions === currentSubmissions) return prev;
if (submissions === currentSubmissions) return prev;
const messageSubmissions = new Map(session.messageSubmissions);
if (next.submissions.length > 0) {
messageSubmissions.set(agentId, next.submissions);
if (submissions.length > 0) {
messageSubmissions.set(agentId, submissions);
} else {
messageSubmissions.delete(agentId);
}
@@ -1230,16 +1212,20 @@ export const useSessionStore = create<SessionStore>()(
const currentTail = session.agentStreamTail.get(agentId) ?? [];
const currentHead = session.agentStreamHead.get(agentId) ?? [];
const currentSubmissions = session.messageSubmissions.get(agentId) ?? [];
const result = rejectMessageSubmission(
{ tail: currentTail, head: currentHead, submissions: currentSubmissions },
clientMessageId,
);
const result = rejectMessageSubmission(currentSubmissions, clientMessageId);
outcome = result.outcome;
const next = result.state;
if (outcome === "unknown") return prev;
const stream =
outcome === "rejected"
? removeSubmittedUserMessage({
tail: currentTail,
head: currentHead,
clientMessageId,
})
: { tail: currentTail, head: currentHead };
const messageSubmissions = new Map(session.messageSubmissions);
if (next.submissions.length > 0) {
messageSubmissions.set(agentId, next.submissions);
if (result.submissions.length > 0) {
messageSubmissions.set(agentId, result.submissions);
} else {
messageSubmissions.delete(agentId);
}
@@ -1250,13 +1236,13 @@ export const useSessionStore = create<SessionStore>()(
[serverId]: {
...session,
agentStreamTail:
next.tail === currentTail
stream.tail === currentTail
? session.agentStreamTail
: new Map(session.agentStreamTail).set(agentId, next.tail),
: new Map(session.agentStreamTail).set(agentId, stream.tail),
agentStreamHead:
next.head === currentHead
stream.head === currentHead
? session.agentStreamHead
: new Map(session.agentStreamHead).set(agentId, next.head),
: new Map(session.agentStreamHead).set(agentId, stream.head),
messageSubmissions,
},
},
@@ -1498,7 +1484,7 @@ export const useSessionStore = create<SessionStore>()(
return prev;
}
const nextAgents = typeof agents === "function" ? agents(session.agents) : agents;
const messageSubmissions = applyAgentStateToMessageSubmissions({
const messageSubmissions = applyRunningAgentsToAcceptedSubmissions({
previousAgents: session.agents,
nextAgents,
submissions: session.messageSubmissions,
packages/app/src/timeline/session-stream-reducers.test.ts
+29 -178
View File
@@ -172,7 +172,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = {
isInitializing: false,
hasActiveInitDeferred: false,
initRequestDirection: "tail",
pendingClientMessageId: null,
sendingClientMessageIds: [],
};
const baseStreamInput: ProcessAgentStreamEventInput = {
@@ -182,7 +182,6 @@ const baseStreamInput: ProcessAgentStreamEventInput = {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
timestamp: new Date(2000),
};
@@ -306,7 +305,7 @@ describe("processTimelineResponse", () => {
currentTail: [],
currentHead: [liveAssistant, submitted],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
pendingClientMessageId: "client-new-prompt",
sendingClientMessageIds: ["client-new-prompt"],
payload: {
...baseTimelineInput.payload,
reset: true,
@@ -479,7 +478,7 @@ describe("processTimelineResponse", () => {
...baseTimelineInput,
currentTail: [submitted],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
pendingClientMessageId: "submitted-unmatched",
sendingClientMessageIds: ["submitted-unmatched"],
payload: {
...baseTimelineInput.payload,
reset: true,
@@ -491,7 +490,27 @@ describe("processTimelineResponse", () => {
expect(result.tail).toEqual([submitted]);
});
it("keeps an acknowledged local row omitted by a same-epoch replacement", () => {
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",
@@ -502,7 +521,7 @@ describe("processTimelineResponse", () => {
...baseTimelineInput,
currentTail: [acknowledged],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
pendingClientMessageId: null,
sendingClientMessageIds: [],
payload: {
...baseTimelineInput.payload,
reset: true,
@@ -511,7 +530,7 @@ describe("processTimelineResponse", () => {
},
});
expect(result.tail).toEqual([acknowledged]);
expect(result.tail).toEqual([]);
});
it("drops an acknowledged local row omitted by a known epoch change", () => {
@@ -525,7 +544,7 @@ describe("processTimelineResponse", () => {
...baseTimelineInput,
currentTail: [acknowledged],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
pendingClientMessageId: null,
sendingClientMessageIds: [],
payload: {
...baseTimelineInput.payload,
reset: true,
@@ -559,7 +578,7 @@ describe("processTimelineResponse", () => {
const result = processTimelineResponse({
...baseTimelineInput,
currentTail: [unmatched, ...acknowledged],
pendingClientMessageId: "client-first",
sendingClientMessageIds: ["client-first"],
payload: {
...baseTimelineInput.payload,
reset: true,
@@ -807,6 +826,7 @@ describe("processTimelineResponse", () => {
text: "local presentation",
}),
]);
expect(result.acknowledgedClientMessageIds).toEqual(["client-message"]);
});
it("reconciles multiple submitted user messages in canonical order", () => {
@@ -961,7 +981,6 @@ describe("processTimelineResponse", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
const result = processTimelineResponse({
@@ -1039,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);
@@ -1087,7 +1105,6 @@ describe("processTimelineResponse", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
const result = processTimelineResponse({
@@ -1278,7 +1295,6 @@ describe("processTimelineResponse", () => {
currentTail: [prompt],
currentHead: [],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
currentAgent: null,
});
const result = processTimelineResponse({
@@ -1323,7 +1339,6 @@ describe("processTimelineResponse", () => {
currentTail: [prompt],
currentHead: [],
currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 },
currentAgent: null,
});
const result = processTimelineResponse({
@@ -2402,150 +2417,6 @@ describe("processAgentStreamEvent", () => {
endSeq: 1,
});
});
it("does not mark a running replacement idle on an earlier turn_completed", () => {
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(false);
expect(result.agent).toBe(null);
});
it("derives submitted 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 submitted 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 a failing event is older than the agent", () => {
const turnFailedEvent: AgentStreamEventPayload = {
type: "turn_failed",
provider: "claude",
error: "something broke",
};
const result = processAgentStreamEvent({
...baseStreamInput,
event: turnFailedEvent,
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", () => {
@@ -2558,7 +2429,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(false);
@@ -2586,7 +2456,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(false);
@@ -2609,7 +2478,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2627,7 +2495,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2645,7 +2512,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2677,7 +2543,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2703,7 +2568,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2728,7 +2592,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(result.changedTail).toBe(true);
@@ -2820,7 +2683,6 @@ describe("processAgentStreamEvents", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
});
expect(getAssistantTexts([...result.tail, ...result.head])).toEqual([
@@ -2845,17 +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(false);
expect(result.agent).toBe(null);
});
it("keeps a live Claude assistant paragraph contiguous when init tail hydration lands mid-stream", () => {
@@ -3004,7 +2859,6 @@ describe("createAgentStreamReducerQueue", () => {
currentTail,
currentHead,
currentCursor: undefined,
currentAgent: null,
}),
commit: (agentId, result) => {
currentTail = result.tail;
@@ -3046,7 +2900,6 @@ describe("createAgentStreamReducerQueue", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
}),
commit: (agentId, result) => {
commits.push(
@@ -3076,7 +2929,6 @@ describe("createAgentStreamReducerQueue", () => {
currentTail,
currentHead,
currentCursor,
currentAgent: null,
}),
commit: (_agentId, result) => {
currentTail = result.tail;
@@ -3126,7 +2978,6 @@ describe("createAgentStreamReducerQueue", () => {
currentTail: [],
currentHead: [],
currentCursor: undefined,
currentAgent: null,
}),
commit: (agentId, result) => {
commits.push(
packages/app/src/timeline/session-stream-reducers.ts
+92 -159
View File
@@ -1,6 +1,4 @@
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 } from "@/types/stream";
import {
@@ -90,7 +88,7 @@ export interface ProcessTimelineResponseInput {
isInitializing: boolean;
hasActiveInitDeferred: boolean;
initRequestDirection: InitRequestDirection;
pendingClientMessageId: string | null;
sendingClientMessageIds: readonly string[];
}
export interface ProcessTimelineResponseOutput {
@@ -102,6 +100,7 @@ export interface ProcessTimelineResponseOutput {
clearInitializing: boolean;
error: string | null;
sideEffects: TimelineReducerSideEffect[];
acknowledgedClientMessageIds: string[];
}
interface TimelineUnit {
@@ -118,6 +117,7 @@ interface TimelinePathResult {
cursor: TimelineCursor | null | undefined;
cursorChanged: boolean;
sideEffects: TimelineReducerSideEffect[];
acknowledgedClientMessageIds: string[];
}
function classifySessionTimelineSeq({
@@ -204,34 +204,13 @@ function shouldResolveTimelineInit({
return responseDirection === initRequestDirection;
}
function deriveProjectedLifecycleStatus(
currentStatus: AgentLifecycleStatus,
event: AgentStreamEventPayload,
): AgentLifecycleStatus | null {
if (currentStatus !== "running") {
return null;
}
switch (event.type) {
case "turn_failed":
return "error";
case "turn_completed":
case "turn_canceled":
// A terminal event can belong to an interrupted turn while its replacement
// is still running. The daemon snapshot is authoritative here.
return null;
default:
return null;
}
}
function applyTimelineReplacePath(args: {
timelineUnits: TimelineUnit[];
payload: ProcessTimelineResponseInput["payload"];
bootstrapPolicy: ReturnType<typeof deriveBootstrapTailTimelinePolicy>;
currentTail: StreamItem[];
currentHead: StreamItem[];
pendingClientMessageId: string | null;
replacementPolicy: "preserve-local-submissions" | "replace-history";
sendingClientMessageIds: readonly string[];
preserveLiveHead: boolean;
toHydratedEvents: (
units: TimelineUnit[],
@@ -243,18 +222,16 @@ function applyTimelineReplacePath(args: {
bootstrapPolicy,
currentTail,
currentHead,
pendingClientMessageId,
replacementPolicy,
sendingClientMessageIds,
preserveLiveHead,
toHydratedEvents,
} = args;
const hydratedTail = hydrateStreamState(toHydratedEvents(timelineUnits), { source: "canonical" });
const { tail, head } = replaceWithCanonicalStream({
const { tail, head, acknowledgedClientMessageIds } = replaceWithCanonicalStream({
canonical: hydratedTail,
previousTail: currentTail,
previousHead: currentHead,
pendingClientMessageId,
policy: replacementPolicy,
sendingClientMessageIds,
preserveLiveHead,
});
const cursor: TimelineCursor | null =
@@ -269,7 +246,14 @@ function applyTimelineReplacePath(args: {
if (bootstrapPolicy.catchUpCursor) {
sideEffects.push({ type: "catch_up", cursor: bootstrapPolicy.catchUpCursor });
}
return { tail, head, cursor, cursorChanged: true, sideEffects };
return {
tail,
head,
cursor,
cursorChanged: true,
sideEffects,
acknowledgedClientMessageIds,
};
}
interface IncrementalAcceptResult {
@@ -631,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, {
@@ -641,6 +640,7 @@ function applyCanonicalForwardUnit(params: {
timelineCursor,
}),
head: params.head,
acknowledgedClientMessageIds: [],
};
}
const replacedHead = replaceLiveAssistantWithProjectedText({
@@ -649,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" }> =>
@@ -667,6 +669,7 @@ function applyCanonicalForwardUnit(params: {
source: "canonical",
timelineCursor,
}),
acknowledgedClientMessageIds: [],
};
}
@@ -678,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: {
@@ -687,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,
@@ -697,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: {
@@ -721,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 } =
@@ -760,6 +779,7 @@ function applyTimelineIncrementalPath(args: {
});
nextTail = applied.tail;
nextHead = applied.head;
acknowledgedClientMessageIds = applied.acknowledgedClientMessageIds;
}
}
@@ -778,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(
@@ -792,7 +819,7 @@ export function processTimelineResponse(
isInitializing,
hasActiveInitDeferred,
initRequestDirection,
pendingClientMessageId,
sendingClientMessageIds,
} = input;
// ------------------------------------------------------------------
@@ -808,6 +835,7 @@ export function processTimelineResponse(
clearInitializing: isInitializing,
error: payload.error,
sideEffects: [],
acknowledgedClientMessageIds: [],
};
}
@@ -854,11 +882,6 @@ export function processTimelineResponse(
hasActiveInitDeferred,
});
const replace = bootstrapPolicy.replace;
const replacementPolicy =
currentCursor && currentCursor.epoch !== payload.epoch
? "replace-history"
: "preserve-local-submissions";
const sideEffects: TimelineReducerSideEffect[] = [];
const timelineResult = replace
? applyTimelineReplacePath({
@@ -867,8 +890,7 @@ export function processTimelineResponse(
bootstrapPolicy,
currentTail,
currentHead,
pendingClientMessageId,
replacementPolicy,
sendingClientMessageIds,
preserveLiveHead: currentCursor?.epoch === payload.epoch,
toHydratedEvents,
})
@@ -918,6 +940,7 @@ export function processTimelineResponse(
clearInitializing,
error: null,
sideEffects,
acknowledgedClientMessageIds: timelineResult.acknowledgedClientMessageIds,
};
}
@@ -932,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[];
@@ -953,8 +965,7 @@ export interface ProcessAgentStreamEventOutput {
changedHead: boolean;
cursor: TimelineCursor | null;
cursorChanged: boolean;
agent: AgentPatch | null;
agentChanged: boolean;
acknowledgedClientMessageIds: string[];
sideEffects: AgentStreamReducerSideEffect[];
}
@@ -973,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">;
@@ -1008,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;
@@ -1095,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 =
@@ -1107,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,
@@ -1123,43 +1112,14 @@ export function processAgentStreamEvent(
changedHead: false,
};
// ------------------------------------------------------------------
// Projected 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 projectedStatus = deriveProjectedLifecycleStatus(currentAgent.status, event);
if (projectedStatus) {
const nextUpdatedAtMs = Math.max(currentAgent.updatedAt.getTime(), timestamp.getTime());
const nextLastActivityAtMs = Math.max(
currentAgent.lastActivityAt.getTime(),
timestamp.getTime(),
);
agentPatch = {
status: projectedStatus,
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,
};
}
@@ -1170,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) {
@@ -1186,7 +1144,6 @@ export function processAgentStreamEvents(
currentTail: tail,
currentHead: head,
currentCursor: cursor,
currentAgent: agent,
timestamp: reducerEvent.timestamp,
});
@@ -1195,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 {
@@ -1215,8 +1169,7 @@ export function processAgentStreamEvents(
changedHead,
cursor: cursor ?? null,
cursorChanged,
agent: agentPatch,
agentChanged,
acknowledgedClientMessageIds: [...acknowledgedClientMessageIds],
sideEffects,
};
}
@@ -1299,6 +1252,7 @@ export function createAgentStreamReducerQueue(
interface StreamStatePatch {
tail?: StreamItem[];
head?: StreamItem[];
acknowledgedClientMessageIds?: readonly string[];
}
export interface CreateSessionAgentStreamReducerQueueInput {
@@ -1308,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;
}
@@ -1323,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 }
: {}),
});
}
@@ -1380,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/types/stream.ts
+108 -26
View File
@@ -123,12 +123,45 @@ export function createUserMessage(input: UserMessageInput): UserMessageItem {
};
}
export function appendSubmittedUserMessage(input: {
tail: StreamItem[];
head: StreamItem[];
message: UserMessageItem;
}): { tail: StreamItem[]; head: StreamItem[] } {
const clientMessageId = input.message.clientMessageId;
if (!clientMessageId) {
throw new Error("Submitted user message requires client identity");
}
const alreadyExists = [...input.tail, ...input.head].some(
(item) => item.kind === "user_message" && item.clientMessageId === clientMessageId,
);
if (alreadyExists) {
throw new Error(`Submitted user message already exists: ${clientMessageId}`);
}
return input.head.length > 0
? { tail: input.tail, head: [...input.head, input.message] }
: { tail: [...input.tail, input.message], head: input.head };
}
export function removeSubmittedUserMessage(input: {
tail: StreamItem[];
head: StreamItem[];
clientMessageId: string;
}): { tail: StreamItem[]; head: StreamItem[] } {
const remove = (items: StreamItem[]) => {
const next = items.filter(
(item) => item.kind !== "user_message" || item.clientMessageId !== input.clientMessageId,
);
return next.length === items.length ? items : next;
};
return { tail: remove(input.tail), head: remove(input.head) };
}
// COMPAT(userMessageClientId): added in v0.2.0, remove after 2027-01-20 once the
// supported daemon floor emits clientMessageId on submitted user messages. Until then a
// locally submitted row (clientMessageId, no messageId) and its canonical twin from an
// old daemon (messageId, no clientMessageId) share no identifier, so they are matched by
// the id the client supplied over the wire or by text. Direction-independent: either row
// can already be in the timeline when the other arrives.
// old daemon (messageId, no clientMessageId) share no identifier, so canonical ingestion
// may match an explicit local candidate by the id supplied over the wire or by text.
function matchesLegacyCanonicalUserMessage(
submitted: UserMessageItem,
canonical: UserMessageItem,
@@ -138,17 +171,21 @@ function matchesLegacyCanonicalUserMessage(
return canonical.messageId === submitted.clientMessageId || canonical.text === submitted.text;
}
function matchesUserMessage(existing: UserMessageItem, incoming: UserMessageItem): boolean {
type UserMessageMatchPolicy = "canonical-incoming" | "handoff";
function matchesUserMessage(
existing: UserMessageItem,
incoming: UserMessageItem,
policy: UserMessageMatchPolicy,
): boolean {
if (existing.clientMessageId && incoming.clientMessageId) {
return existing.clientMessageId === incoming.clientMessageId;
}
if (existing.messageId && incoming.messageId) {
return existing.messageId === incoming.messageId;
}
return (
matchesLegacyCanonicalUserMessage(existing, incoming) ||
matchesLegacyCanonicalUserMessage(incoming, existing)
);
if (matchesLegacyCanonicalUserMessage(existing, incoming)) return true;
return policy === "handoff" && matchesLegacyCanonicalUserMessage(incoming, existing);
}
export function upsertUserMessage(
@@ -173,9 +210,10 @@ function produceUserMessage(
incoming: UserMessageItem,
insertAt: number | null,
presentationPolicy: UserMessagePresentationPolicy,
matchPolicy: UserMessageMatchPolicy = "canonical-incoming",
): UserMessageProductionResult {
const index = items.findIndex(
(item) => item.kind === "user_message" && matchesUserMessage(item, incoming),
(item) => item.kind === "user_message" && matchesUserMessage(item, incoming, matchPolicy),
);
if (index < 0) {
if (insertAt === null) {
@@ -221,6 +259,7 @@ export interface UserMessageStreamUpsertInput {
message: UserMessageItem;
insert: "tail" | "head" | "prepend-tail" | "none";
presentation: UserMessagePresentationPolicy;
matchPolicy?: UserMessageMatchPolicy;
}
export interface UserMessageStreamUpsertResult extends ApplyStreamEventResult {
@@ -235,7 +274,13 @@ export interface UserMessageStreamUpsertResult extends ApplyStreamEventResult {
export function upsertUserMessageAcrossStream(
input: UserMessageStreamUpsertInput,
): UserMessageStreamUpsertResult {
const tailResult = produceUserMessage(input.tail, input.message, null, input.presentation);
const tailResult = produceUserMessage(
input.tail,
input.message,
null,
input.presentation,
input.matchPolicy,
);
if (tailResult.matched) {
return {
tail: tailResult.items,
@@ -250,7 +295,13 @@ export function upsertUserMessageAcrossStream(
},
};
}
const headResult = produceUserMessage(input.head, input.message, null, input.presentation);
const headResult = produceUserMessage(
input.head,
input.message,
null,
input.presentation,
input.matchPolicy,
);
if (headResult.matched) {
return {
tail: input.tail,
@@ -280,6 +331,7 @@ export function upsertUserMessageAcrossStream(
input.message,
input.head.length,
input.presentation,
input.matchPolicy,
);
return {
tail: input.tail,
@@ -299,6 +351,7 @@ export function upsertUserMessageAcrossStream(
input.message,
input.insert === "prepend-tail" ? 0 : input.tail.length,
input.presentation,
input.matchPolicy,
);
return {
tail: inserted.items,
@@ -318,14 +371,14 @@ export interface CanonicalStreamReplacementInput {
canonical: StreamItem[];
previousTail: StreamItem[];
previousHead: StreamItem[];
pendingClientMessageId: string | null;
policy: "preserve-local-submissions" | "replace-history";
sendingClientMessageIds: readonly string[];
preserveLiveHead: boolean;
}
export interface CanonicalStreamReplacementResult {
tail: StreamItem[];
head: StreamItem[];
acknowledgedClientMessageIds: string[];
}
function removeUserMessageAt(items: UserMessageItem[], index: number): UserMessageItem[] {
@@ -336,11 +389,15 @@ function preserveReplacementHead(
tail: StreamItem[],
currentHead: StreamItem[],
preserveLiveHead: boolean,
sendingClientMessageIds: ReadonlySet<string>,
): CanonicalStreamReplacementResult {
const retainedHead = preserveLiveHead
? currentHead
: currentHead.filter(
(item) => item.kind === "user_message" && item.clientMessageId !== undefined,
(item) =>
item.kind === "user_message" &&
item.clientMessageId !== undefined &&
sendingClientMessageIds.has(item.clientMessageId),
);
const tailIds = new Set(tail.map((item) => item.id));
const unreconciledHead = retainedHead.filter(
@@ -349,7 +406,9 @@ function preserveReplacementHead(
const liveAssistantIndex = unreconciledHead.findLastIndex(
(item) => item.kind === "assistant_message",
);
if (liveAssistantIndex < 0) return { tail, head: unreconciledHead };
if (liveAssistantIndex < 0) {
return { tail, head: unreconciledHead, acknowledgedClientMessageIds: [] };
}
const liveAssistant = unreconciledHead[liveAssistantIndex];
const tailAssistant = tail.at(-1);
@@ -359,7 +418,7 @@ function preserveReplacementHead(
tailAssistant.kind !== "assistant_message" ||
!liveAssistant.text.startsWith(tailAssistant.text)
) {
return { tail, head: unreconciledHead };
return { tail, head: unreconciledHead, acknowledgedClientMessageIds: [] };
}
const head = [
@@ -367,18 +426,20 @@ function preserveReplacementHead(
{ ...liveAssistant, text: tailAssistant.text },
...unreconciledHead.slice(liveAssistantIndex + 1),
];
return { tail: tail.slice(0, -1), head };
return { tail: tail.slice(0, -1), head, acknowledgedClientMessageIds: [] };
}
export function replaceWithCanonicalStream(
input: CanonicalStreamReplacementInput,
): CanonicalStreamReplacementResult {
const sendingClientMessageIds = new Set(input.sendingClientMessageIds);
let unmatchedTailMessages = input.previousTail.filter(
(item): item is UserMessageItem =>
item.kind === "user_message" && item.clientMessageId !== undefined,
);
let nextHead = input.previousHead;
const nextTail: StreamItem[] = [];
const acknowledgedClientMessageIds = new Set<string>();
for (const item of input.canonical) {
if (item.kind !== "user_message") {
@@ -390,6 +451,12 @@ export function replaceWithCanonicalStream(
if (tailResult.matched) {
unmatchedTailMessages = removeUserMessageAt(unmatchedTailMessages, tailResult.index);
nextTail.push(tailResult.message);
if (
tailResult.message.clientMessageId &&
sendingClientMessageIds.has(tailResult.message.clientMessageId)
) {
acknowledgedClientMessageIds.add(tailResult.message.clientMessageId);
}
continue;
}
@@ -400,6 +467,12 @@ export function replaceWithCanonicalStream(
...headResult.items.slice(headResult.index + 1),
];
nextTail.push(headResult.message);
if (
headResult.message.clientMessageId &&
sendingClientMessageIds.has(headResult.message.clientMessageId)
) {
acknowledgedClientMessageIds.add(headResult.message.clientMessageId);
}
continue;
}
@@ -407,10 +480,7 @@ export function replaceWithCanonicalStream(
}
for (const local of unmatchedTailMessages) {
if (
input.policy === "replace-history" &&
local.clientMessageId !== input.pendingClientMessageId
) {
if (!local.clientMessageId || !sendingClientMessageIds.has(local.clientMessageId)) {
continue;
}
const insertionIndex = nextTail.findIndex(
@@ -421,13 +491,19 @@ export function replaceWithCanonicalStream(
nextHead = nextHead.filter((item) => {
if (item.kind !== "user_message" || !item.clientMessageId) return true;
return (
input.policy === "preserve-local-submissions" ||
item.clientMessageId === input.pendingClientMessageId
);
return sendingClientMessageIds.has(item.clientMessageId);
});
return preserveReplacementHead(nextTail, nextHead, input.preserveLiveHead);
const replacement = preserveReplacementHead(
nextTail,
nextHead,
input.preserveLiveHead,
sendingClientMessageIds,
);
return {
...replacement,
acknowledgedClientMessageIds: [...acknowledgedClientMessageIds],
};
}
export interface AssistantMessageItem {
@@ -572,6 +648,7 @@ export function handoffCreatedAgentUserMessageToStream(params: {
...params,
insert: "tail",
presentation: "incoming",
matchPolicy: "handoff",
});
}
@@ -1409,6 +1486,7 @@ export interface ApplyStreamEventResult {
head: StreamItem[];
changedTail: boolean;
changedHead: boolean;
acknowledgedClientMessageIds?: string[];
}
function applyCanonicalUserMessageEvent(params: {
@@ -1447,6 +1525,10 @@ function applyCanonicalUserMessageEvent(params: {
head: reconciled.head,
changedTail: flushedTail !== tail || reconciled.changedTail,
changedHead: flushedHead !== head || reconciled.changedHead,
acknowledgedClientMessageIds:
reconciled.location?.matched && reconciled.location.message.clientMessageId
? [reconciled.location.message.clientMessageId]
: [],
};
}
packages/app/src/utils/agent-directory-sync.test.ts
+25 -24
View File
@@ -96,9 +96,9 @@ function applyAgentStatus(input: {
});
}
describe("message submission lifecycle ordering", () => {
it("keeps idle and pending when acceptance precedes authoritative running", () => {
const serverId = "server-accept-before-running";
describe("message submission authority", () => {
it("does not settle a submission from an unrelated running transition", () => {
const serverId = "server-running-is-not-submission-ack";
const agentId = "agent-1";
const store = useSessionStore.getState();
store.initializeSession(serverId, null as unknown as DaemonClient);
@@ -110,18 +110,6 @@ describe("message submission lifecycle ordering", () => {
});
const clientMessageId = beginPendingSubmission(serverId, agentId);
store.acceptAgentMessageSubmission(serverId, agentId, clientMessageId);
const session = useSessionStore.getState().sessions[serverId];
expect(session?.agents.get(agentId)?.status).toBe("idle");
expect(session?.messageSubmissions.get(agentId)).toEqual([
{
clientMessageId,
submittedAt: new Date("2026-07-27T10:00:00.000Z"),
phase: "waiting-for-running",
},
]);
applyAgentStatus({
serverId,
agentId,
@@ -129,14 +117,19 @@ describe("message submission lifecycle ordering", () => {
updatedAt: "2026-07-27T10:00:01.000Z",
});
expect(
useSessionStore.getState().sessions[serverId]?.messageSubmissions.get(agentId),
).toBeUndefined();
expect(useSessionStore.getState().sessions[serverId]?.messageSubmissions.get(agentId)).toEqual([
{
clientMessageId,
submittedAt: new Date("2026-07-27T10:00:00.000Z"),
rpcAccepted: false,
providerAcknowledged: false,
},
]);
store.clearSession(serverId);
});
it("clears accepted pending state when canonical history arrives after a missed run", () => {
const serverId = "server-canonical-after-missed-run";
it("settles provider acknowledgement only when timeline ingestion reports it", () => {
const serverId = "server-explicit-provider-ack";
const agentId = "agent-1";
const store = useSessionStore.getState();
store.initializeSession(serverId, null as unknown as DaemonClient);
@@ -147,8 +140,6 @@ describe("message submission lifecycle ordering", () => {
updatedAt: "2026-07-27T10:00:00.000Z",
});
const clientMessageId = beginPendingSubmission(serverId, agentId);
store.acceptAgentMessageSubmission(serverId, agentId, clientMessageId);
store.setAgentStreamState(serverId, agentId, {
tail: [
createUserMessage({
@@ -163,8 +154,18 @@ describe("message submission lifecycle ordering", () => {
});
expect(
useSessionStore.getState().sessions[serverId]?.messageSubmissions.get(agentId),
).toBeUndefined();
useSessionStore.getState().sessions[serverId]?.messageSubmissions.get(agentId)?.[0]
?.providerAcknowledged,
).toBe(false);
store.setAgentStreamState(serverId, agentId, {
acknowledgedClientMessageIds: [clientMessageId],
});
expect(
useSessionStore.getState().sessions[serverId]?.messageSubmissions.get(agentId)?.[0]
?.providerAcknowledged,
).toBe(true);
store.clearSession(serverId);
});
});
packages/app/src/utils/agent-directory-sync.ts
+10
View File
@@ -9,6 +9,7 @@ import { acceptAgentDirectoryUpdate } from "@/utils/agent-directory-update-polic
import { buildDraftStoreKey } from "@/stores/draft-keys";
import { useDraftStore } from "@/stores/draft-store";
import { getInitDeferred, getInitKey, rejectInitDeferred } from "@/utils/agent-initialization";
import { useCreateFlowStore } from "@/stores/create-flow-store";
type AgentDirectoryFetchEntry = FetchAgentsEntry;
export type AgentDirectoryDelta = Extract<
@@ -52,6 +53,9 @@ function upsertAgentDirectoryReplica(
previousAgent?.projectPlacement,
};
const acceptedAgent = upsertAgentReplica(serverId, agent);
if (acceptedAgent.status === "running") {
useCreateFlowStore.getState().clearByAgent({ serverId, agentId: acceptedAgent.id });
}
if (acceptedAgent.archivedAt) {
clearArchiveAgentPending({ queryClient, serverId, agentId: acceptedAgent.id });
}
@@ -174,6 +178,12 @@ export function replaceFetchedAgentDirectory(input: {
const store = useSessionStore.getState();
for (const agent of fetchedAgents.values()) {
if (agent.status === "running") {
useCreateFlowStore.getState().clearByAgent({
serverId: input.serverId,
agentId: agent.id,
});
}
if (agent.archivedAt) {
clearArchiveAgentPending({ queryClient, serverId: input.serverId, agentId: agent.id });
}