diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index a19e399488..f1d8998919 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -192,13 +192,20 @@ export on a page error or non-advancing cursor; demos keep their fixture path. - **Run creation retries** are disabled only for the initial `POST /runs/stream` in `core/api/api-client.ts`, because an ambiguous failure can otherwise create duplicate runs. Automatic SSE recovery keeps the SDK's normal HTTP retry budget for `GET` requests to the server-provided `Location`, including transient 5xx responses, and forwards `Last-Event-ID`. Reads and explicit SSE joins keep their normal retry behavior. Regression tests cover both a failed creation POST and POST success → stream interruption → GET 504 → GET success. - **Run stream options** are sanitized by `core/api/stream-mode.ts`: the Gateway-supported set is `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom`; any request containing an unsupported mode throws before HTTP instead of being partially forwarded or silently defaulting to `values`. `streamResumable` is retained by thread hooks only for SDK-side reconnect bookkeeping but stripped before the HTTP request because the Gateway does not accept that request option; actual replay uses the SSE `Last-Event-ID` cursor. The main chat client's initial and rejoined streams are forced to incremental `messages-tuple`, `updates`, and `custom` modes so SDK lazy tracking cannot add repeated full-state `values` snapshots; explicitly requested non-snapshot modes such as `debug`, `tasks`, and `checkpoints` are preserved. `values` remains supported outside this chat wrapper for explicit state inspection and by the replay-gap durable-state reload below. The backend file-tool chunk batcher is keyed to `messages-tuple`, not `values`, so omitting snapshots must not regress `write_file` / `str_replace` streaming into one SSE frame per model token. `core/threads/stream-state.ts` folds the user-visible non-message fields from `updates` with the matching DeerFlow reducer semantics, including `tool_artifacts` capture/consumption updates (same-handle replacement preserves order, retention directives trim the oldest entries, and the absolute ceiling is 1000) and rejects irrelevant frames before calling the SDK mutator, while the SDK's `messages-tuple` manager remains the sole owner of live message chunk assembly and deduplication. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded. - **Active-run recovery** in `core/threads/hooks.ts` discovers pending/running runs when a reopened chat has no tab-local reconnect pointer. Remember SDK-completed run IDs for the mounted hook, including submitted and natively reconnected runs: stale or delayed runs reads must not rejoin a completed run and interrupt follow-up suggestions. Keep recovery retries bounded and continue allowing a different active run to reconnect. The SDK's same-tab reconnect keeps its pointer on error and never retries, so a failed reconnect (not a failed submit) releases the pointer and recovery takes over with its own bounded retries. -- **SSE replay gaps** are handled in `core/api/api-client.ts`, which wraps both initial and joined run streams because the upstream SDK ignores unknown event names. An id-less backend `gap` control frame clears stale reconnect metadata, emits an internal `stream_replay_gap` custom event, reloads durable thread values, and resumes after the server-provided retained tail when one exists (or rejoins without a cursor if the buffer is empty), with up to five recovery rejoins after the original stream (six total stream calls on an all-gap exhaustion path). The wrapper remains a lazy async iterable because the SDK consumes it with `for await`. `core/threads/hooks.ts` clears optimistic/transient/subtask state, invalidates durable history caches, and shows the localized recovery warning; never let a gap fall through as a normal stream finish or cancel the still-running backend run. - Replay-gap recovery restores its tab-local run pointer only if the slot is empty - or still belongs to that run; never overwrite another run's pointer after an - asynchronous durable-state read. A stale foreign pointer may cost an extra - run lookup/rejoin on reload; recovery adds no server probe. The same-named - `rememberReconnectRun` in `core/threads/hooks.ts` stays unconditional: - active-run rejoin makes a fresh claim that must win over stored metadata. +- **SSE replay gaps** (`core/api/api-client.ts`) wrap initial/joined streams as a lazy + async iterable: the SDK ignores unknown events. An id-less `gap` clears stale + reconnect metadata, emits a `stream_replay_gap` custom event, reloads durable values, + then rejoins after the server-provided retained tail (without a cursor for an empty + buffer). Allow five recovery rejoins after the original stream (six calls total). + `core/threads/hooks.ts` clears optimistic/transient/subtask state, invalidates durable + history caches and shows the localized warning; never finish normally on a gap or + cancel the running backend run. Merge submitted initial-stream input or saved + joined-stream input into each recovered values frame: the first checkpoint may not + yet contain the human message. + Restore the tab-local run pointer only if empty or still owned by this run, never + over another run's pointer after a durable-state read. A foreign pointer can cost + an extra lookup/rejoin on reload; add no server probe. The hooks' same-named + `rememberReconnectRun` remains unconditional: active-run rejoin claims ownership. - **Streaming Markdown rendering** is owned by `core/streamdown`: Streamdown's `animated` / `isAnimating` API handles incremental word animation, while the shared `streamdownRenderingPlugins` config registers the named code-highlighting and Mermaid plugins required by Streamdown 2.5. Keep wrappers and derived configs wired to that shared object; do not reintroduce a rehype plugin that wraps every word, because reparsing a growing block remounts old words and replays their animation. - Citation links in message and artifact Markdown must derive their `citation:` label from the full `ReactNode` children tree, since Streamdown may provide element or array children during streaming rather than a plain string. - Web-search steps accept either a provider result array or a Webz-style `{ query, returned_results, results }` envelope. Unwrap only in the web-search renderer; skip entries without string titles/URLs, preserve other tool payloads and apply the same link allowlist to both formats. UI regressions pin source titles/links and browser reloads, including malformed neighboring entries. diff --git a/frontend/src/core/api/api-client.ts b/frontend/src/core/api/api-client.ts index 8791a064d4..af095132ac 100644 --- a/frontend/src/core/api/api-client.ts +++ b/frontend/src/core/api/api-client.ts @@ -111,10 +111,12 @@ function streamOptionSignal(options: unknown): AbortSignal | undefined { * Recover the submitted input before replaying an active run. The incremental * chat stream intentionally omits `values`, so a page reload can otherwise * receive the run's AI/tool chunks before its human message has reached the - * durable history feed. `runs.get` retains the original graph input in - * `kwargs.input`; merge it into the latest durable values for one synthetic - * snapshot. Any read failure is deliberately ignored so reconnect semantics - * remain unchanged for deployments without run metadata. + * durable history feed. Initial streams supply the sanitized `submittedInput` + * directly, avoiding a run-metadata read. Otherwise, use `kwargs.input` from + * the supplied run record, falling back to `runs.get` when needed. Merge the + * input into the latest durable values for one synthetic snapshot. Any read + * failure is deliberately ignored so reconnect semantics remain unchanged for + * deployments without run metadata. */ async function loadReconnectInputSnapshot( client: LangGraphClient, @@ -123,15 +125,19 @@ async function loadReconnectInputSnapshot( run?: Awaited>, durableValues?: unknown, signal?: AbortSignal, + submittedInput?: unknown, ): Promise { try { - const resolvedRun = - run ?? (await client.runs.get(threadId, runId, { signal })); - const runKwargs = Reflect.get(resolvedRun, "kwargs"); - const input = - typeof runKwargs === "object" && runKwargs !== null - ? Reflect.get(runKwargs, "input") - : undefined; + let input = submittedInput; + if (input === undefined) { + const resolvedRun = + run ?? (await client.runs.get(threadId, runId, { signal })); + const runKwargs = Reflect.get(resolvedRun, "kwargs"); + input = + typeof runKwargs === "object" && runKwargs !== null + ? Reflect.get(runKwargs, "input") + : undefined; + } const inputMessages = typeof input === "object" && input !== null ? Reflect.get(input, "messages") @@ -347,6 +353,7 @@ async function* recoverStreamReplayGaps({ resume, signal, reconnectRun, + submittedInput, }: { client: LangGraphClient; threadId: string | null | undefined; @@ -355,6 +362,7 @@ async function* recoverStreamReplayGaps({ resume: (runId: string, lastEventId?: string) => AsyncIterable; signal?: AbortSignal; reconnectRun?: Awaited>; + submittedInput?: unknown; }): AsyncGenerator { let stream = initialStream; let recoveryAttempts = 0; @@ -404,18 +412,20 @@ async function* recoverStreamReplayGaps({ }); if (durableState.values != null) { // A gap can arrive after the initial hydration frame but before the - // input reaches the checkpoint. Rebuild the snapshot from run metadata - // so this recovery path cannot overwrite the rescued human message. - const recoveredSnapshot = reconnectRun - ? await loadReconnectInputSnapshot( - client, - threadId, - runId, - reconnectRun, - durableState.values, - signal, - ) - : undefined; + // input reaches the checkpoint. Rebuild the snapshot from submitted or + // saved run input so recovery cannot overwrite the human message. + const recoveredSnapshot = + reconnectRun || submittedInput !== undefined + ? await loadReconnectInputSnapshot( + client, + threadId, + runId, + reconnectRun, + durableState.values, + signal, + submittedInput, + ) + : undefined; yield { event: "values", data: recoveredSnapshot ?? durableState.values, @@ -507,6 +517,7 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient { initialStream, signal: streamOptionSignal(sanitizedPayload), reconnectRun: undefined, + submittedInput: sanitizedPayload?.input, resume: (resolvedRunId, lastEventId) => { // Keep the recovery run id available to the shared inactive-stream // handler even if the SDK omitted its onRunCreated callback. diff --git a/frontend/tests/unit/core/api/api-client.test.ts b/frontend/tests/unit/core/api/api-client.test.ts index b46ef4d52e..1c95aa322a 100644 --- a/frontend/tests/unit/core/api/api-client.test.ts +++ b/frontend/tests/unit/core/api/api-client.test.ts @@ -1053,7 +1053,18 @@ test("recovers a gap emitted by the initial run stream", async () => { for await (const entry of getAPIClient(true).runs.stream( "thread-2", "lead_agent", - { streamResumable: true }, + { + input: { + messages: [ + { + id: "human-2", + type: "human", + content: "Question awaiting checkpoint", + }, + ], + }, + streamResumable: true, + }, )) { received.push(entry); } @@ -1065,7 +1076,16 @@ test("recovers a gap emitted by the initial run stream", async () => { }, { event: "values", - data: { messages: [{ type: "ai", content: "checkpoint" }] }, + data: { + messages: [ + { type: "ai", content: "checkpoint" }, + { + id: "human-2", + type: "human", + content: "Question awaiting checkpoint", + }, + ], + }, }, { event: "end", data: null }, ]);