diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 6893f19ddb..e9ba7e842e 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -96,34 +96,33 @@ footer, and permanently drops attachments because the daemon does not echo them A submitted prompt is one `UserMessageItem` row. That row is the authoritative local presentation: its stable identity, text, timestamp, images, and attachments do not change when the provider acknowledges it. Submission lifecycle is a separate record keyed by agent, not another row shape or -a property inferred from message identity. It begins in `waiting-for-rpc`. Submissions made while a -stored agent is not running require both the RPC and an authoritative non-running-to-running -transition; RPC-first moves to `waiting-for-running`, while running-first moves to -`running-observed`. The second observation removes the record. A force-send that begins while the -agent is already running requires only RPC acceptance, so prior running state cannot count as new -acceptance proof. +a property inferred from message identity. The transaction registry holds every unresolved send and +records RPC acceptance and provider acknowledgement independently. Provider acknowledgement exists +solely so a later transport error cannot roll back a prompt already observed canonically. -An accepted send response does not project agent lifecycle. If it arrives while the replica is -still non-running, the lifecycle record remains `waiting-for-running` until the real transition, so -the footer has no blank frame. A transport error rolls back only `waiting-for-rpc`; -`running-observed` proves the run started and preserves the submitted row. +The daemon's accepted response already waits for the correlated run start, but its response and the +directory update reach client state separately. An accepted transaction remains active until the +directory observes that run or canonical ingestion acknowledges the prompt, bridging those ordered +authorities without inspecting timeline snapshots. Either signal clears only an RPC-accepted +transaction; it cannot settle a fresh send. +Overlapping sends settle independently rather than collapsing to one newest pending message. Canonical submitted user rows carry the provider's `messageId` and Paseo's optional `clientMessageId`. The user-message producer reconciles them by `clientMessageId`, adds provider identity to the existing row, and keeps the local presentation in its original timeline slot. Content matching is limited to the dated compatibility path for daemon timelines created before -that field existed, and it is direction-independent: either the submitted row or its legacy -canonical twin can arrive second. Generic reducers and consumers do not reimplement message -identity matching. +that field existed. Canonical ingestion may match only an explicit unreconciled local candidate; +the draft-create handoff is the one boundary that also permits the legacy canonical twin to have +arrived first. Generic reducers and consumers do not reimplement message identity matching. Ordinary bootstrap, same-epoch reset, and catch-up replacement preserve unmatched locally submitted rows because a provider may never echo them. A known epoch change or rewind replaces history and -drops acknowledged local rows omitted by the new canonical epoch; only the explicitly pending -submission crosses that destructive boundary. +drops acknowledged local rows omitted by the new canonical epoch; every transaction not yet +acknowledged by the provider, and no other local row, crosses that destructive boundary. -Canonical replacement owns both timeline lanes. It reconciles matching submitted head rows in the -head lane and omits their canonical copies from the tail. If a live assistant head is the canonical -assistant prefix, it also stays in the head lane. No row may be returned in both lanes. +Canonical replacement owns both timeline lanes. A matching local row keeps its presentation ID and +payload while taking the canonical row's ordered position. If a live assistant head is the +canonical assistant prefix, it stays in the head lane. No row may be returned in both lanes. ## Relevant code diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts index 608447d663..36d800ff15 100644 --- a/packages/app/e2e/agent-message-submission.spec.ts +++ b/packages/app/e2e/agent-message-submission.spec.ts @@ -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 { 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; } async function beginDraftCreateSubmission( @@ -461,14 +458,13 @@ async function beginDraftCreateSubmission( scenario: DraftCreateScenario, ): Promise { 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 { 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", () => { diff --git a/packages/app/e2e/rewind-menu.ui-contract.spec.ts b/packages/app/e2e/rewind-menu.ui-contract.spec.ts index 2e53357be4..6e3f494487 100644 --- a/packages/app/e2e/rewind-menu.ui-contract.spec.ts +++ b/packages/app/e2e/rewind-menu.ui-contract.spec.ts @@ -61,7 +61,10 @@ async function rewriteCachedMessageAsLegacyRow(page: Page, prompt: string): Prom }, prompt); } -async function waitForCurrentSubmissionInCache(page: Page, prompt: string): Promise { +async function waitForCurrentSubmissionExcludedFromCache( + page: Page, + prompt: string, +): Promise { 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> } | 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 { +async function expectPendingSubmissionNotRestoredAfterReload(page: Page): Promise { 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 }) => { diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index 0ed4fa0313..0bc079e78c 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -240,7 +240,7 @@ export interface AgentStreamViewProps { streamItems: StreamItem[]; streamHead?: StreamItem[]; pendingPermissions: Map; - 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 settings.autoExpandReasoning); const toolCallDetailLevel = useSettings((settings) => settings.toolCallDetailLevel); const viewportRef = useRef(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 ); }, - [context.capabilities, agentId, client, pendingMessageSubmission, resolvedServerId], + [context.capabilities, agentId, client, pendingClientMessageIds, resolvedServerId], ); const renderAssistantMessageItem = useCallback( @@ -885,7 +890,8 @@ const AgentStreamViewComponent = forwardRef 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) diff --git a/packages/app/src/composer/actions.test.ts b/packages/app/src/composer/actions.test.ts index 349d15d493..e42937d386 100644 --- a/packages/app/src/composer/actions.test.ts +++ b/packages/app/src/composer/actions.test.ts @@ -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; tail: Map; - acceptance: MessageSubmissionAcceptance; } -function createFakeStream( - initialHead: Map = new Map(), - options: { acceptance?: MessageSubmissionAcceptance } = {}, -): FakeStream { +function createFakeStream(initialHead: Map = 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 ->(); +const submissionsByFakeStream = new WeakMap>(); -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 }); diff --git a/packages/app/src/composer/draft/create-flow.ts b/packages/app/src/composer/draft/create-flow.ts index eaaddefe37..69895358f8 100644 --- a/packages/app/src/composer/draft/create-flow.ts +++ b/packages/app/src/composer/draft/create-flow.ts @@ -157,12 +157,14 @@ export function useDraftAgentCreateFlow({ }), ]; }, [machine]); - const pendingMessageSubmission = useMemo(() => { - if (machine.tag !== "creating") return null; - return { - clientMessageId: machine.attempt.clientMessageId, - submittedAt: machine.attempt.timestamp, - }; + const pendingMessageSubmissions = useMemo(() => { + if (machine.tag !== "creating") return []; + return [ + { + clientMessageId: machine.attempt.clientMessageId, + submittedAt: machine.attempt.timestamp, + }, + ]; }, [machine]); const draftAgent = useMemo(() => { @@ -212,6 +214,14 @@ export function useDraftAgentCreateFlow({ }), ); 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({ formErrorMessage, isSubmitting, submittedStreamItems, - pendingMessageSubmission, + pendingMessageSubmissions, draftAgent, handleCreateFromInput, continueCreateFromAttempt, diff --git a/packages/app/src/composer/draft/workspace-tab.tsx b/packages/app/src/composer/draft/workspace-tab.tsx index 155bcbdcfb..a4aa272df3 100644 --- a/packages/app/src/composer/draft/workspace-tab.tsx +++ b/packages/app/src/composer/draft/workspace-tab.tsx @@ -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} /> diff --git a/packages/app/src/composer/submission/model.test.ts b/packages/app/src/composer/submission/model.test.ts index a972e44f90..eee4bd4a18 100644 --- a/packages/app/src/composer/submission/model.test.ts +++ b/packages/app/src/composer/submission/model.test.ts @@ -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"); }); }); diff --git a/packages/app/src/composer/submission/model.ts b/packages/app/src/composer/submission/model.ts index f7b2ad67ae..a9e8fc154b 100644 --- a/packages/app/src/composer/submission/model.ts +++ b/packages/app/src/composer/submission/model.ts @@ -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), }; } diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 0ac2a16699..dece8f2ecf 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -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, diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index 565278ee9a..7057169f11 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -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(); 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} /> ); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 2a57a40120..f2d9eea418 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -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 ? { diff --git a/packages/app/src/stores/create-flow-store.ts b/packages/app/src/stores/create-flow-store.ts index 5ca6b95329..2056e42c0f 100644 --- a/packages/app/src/stores/create-flow-store.ts +++ b/packages/app/src/stores/create-flow-store.ts @@ -109,7 +109,10 @@ export const useCreateFlowStore = create((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) { diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index c86c8e2ecb..76c121be0e 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -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; nextAgents: Map; submissions: Map; }): Map { 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()( } } - 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()( 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()( 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()( 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()( [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()( return prev; } const nextAgents = typeof agents === "function" ? agents(session.agents) : agents; - const messageSubmissions = applyAgentStateToMessageSubmissions({ + const messageSubmissions = applyRunningAgentsToAcceptedSubmissions({ previousAgents: session.agents, nextAgents, submissions: session.messageSubmissions, diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index abbda69d71..0b321aa655 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -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( diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 4683b41c89..2690bfb500 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -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; 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 => @@ -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(); 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; @@ -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(); 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) => Map, ) => void; - setAgents: (serverId: string, state: (prev: Map) => Map) => 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) { diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index c2e42af376..645c4c9cc2 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -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, ): 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(); 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] + : [], }; } diff --git a/packages/app/src/utils/agent-directory-sync.test.ts b/packages/app/src/utils/agent-directory-sync.test.ts index 948a8c2619..5330f8aa2a 100644 --- a/packages/app/src/utils/agent-directory-sync.test.ts +++ b/packages/app/src/utils/agent-directory-sync.test.ts @@ -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); }); }); diff --git a/packages/app/src/utils/agent-directory-sync.ts b/packages/app/src/utils/agent-directory-sync.ts index 24e14c4208..532a504788 100644 --- a/packages/app/src/utils/agent-directory-sync.ts +++ b/packages/app/src/utils/agent-directory-sync.ts @@ -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 }); }