mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-09 21:49:37 +00:00
perf(frontend): avoid redundant chat state snapshots (#5159)
* perf(frontend): avoid redundant chat state snapshots Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> * fix(streaming): preserve incremental chat semantics Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> --------- Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>
This commit is contained in:
PeaceMaker-best
GitHub
parent
83cb6767b3
commit
6022bdf5ae
@ -12,7 +12,7 @@ DeerFlow is a LangGraph-based AI super agent system with a full-stack architectu
|
||||
|
||||
**Runtime**:
|
||||
- `make dev`, Docker dev, and production all run the agent runtime in Gateway via `RunManager` + `run_agent()` + `StreamBridge` (`packages/harness/deerflow/runtime/`). Nginx exposes that runtime at `/api/langgraph/*` and rewrites it to Gateway's native `/api/*` routers.
|
||||
- Gateway streams `write_file` and `str_replace` argument deltas in bounded batches when clients also subscribe to `values`; messages-only consumers retain the original per-chunk contract, while `values` preserves the complete tool call.
|
||||
- Gateway streams `write_file` and `str_replace` argument deltas in bounded batches for multi-mode `messages-tuple` consumers; single-mode message consumers retain the original per-chunk contract. Non-message frames flush pending batches, and `values` remains an optional complete-state snapshot rather than a prerequisite for batching.
|
||||
- With `stream_subgraphs`, subgraph frames keep their namespace in the SSE event name (`values|<ns>`, LangGraph Platform style) instead of impersonating root frames — a delegated subagent inherits the parent checkpoint namespace, so publishing its `values` snapshot as bare `values` replaces the whole thread view in SDK clients (#4399). Root-only consumers (file-tool chunk batcher, subagent event persistence, LLM error-fallback detection) ignore namespaced frames. The web frontend does not request subgraph streaming; subtask progress rides root-namespace `task_*` custom events.
|
||||
- Background subagent identity is deliberately split: the provider `tool_call_id` remains the correlation key for `ToolMessage`, `task_*` SSE events, persisted lifecycle events, frontend cards, and the public `ExtensionData.scope_id` contract (stored as `SubagentResult.external_task_id`), while `SubagentExecutor.execute_async()` generates a full server-side `execution_id` for `SubagentResult.task_id`, the process-wide registry, polling, cancellation, timeout handling, and cleanup. Provider IDs are not globally unique across parent runs, so they must never become registry ownership keys; scheduler closures retain their own `SubagentResult` rather than resolving ownership again through the mutable registry. Terminal subagent token usage travels in the current run's `ToolMessage.additional_kwargs` and is attributed from message state, never through a process-global provider-ID cache.
|
||||
- Scheduled-task executions must reuse that same Gateway run lifecycle. The scheduler may decide *when* work runs, but it must dispatch through the existing run path rather than introducing a parallel execution stack. Scheduled launches pass `scheduler.recursion_limit` (default 1000, matching the web UI's `recursion_limit: 1000`, clamped by `max_recursion_limit`) via `launch_scheduled_thread_run`; the value is read from `get_app_config()` at dispatch.
|
||||
|
||||
@ -1169,7 +1169,7 @@ async def run_agent(
|
||||
|
||||
async def _stream_once(input_payload: Any, stream_config: RunnableConfig) -> None:
|
||||
nonlocal llm_error_fallback_message
|
||||
file_tool_chunk_batcher = _LargeFileToolChunkBatcher() if "values" in requested_modes else None
|
||||
file_tool_chunk_batcher = _LargeFileToolChunkBatcher() if "messages-tuple" in requested_modes else None
|
||||
try:
|
||||
async with _checkpoint_thread_lock(thread_id):
|
||||
if len(lg_modes) == 1 and not stream_subgraphs:
|
||||
|
||||
@ -639,8 +639,15 @@ def test_install_runtime_context_removes_caller_sandbox_execution_identities():
|
||||
assert SANDBOX_COMMAND_SCOPE_CONTEXT_KEY not in config["context"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("stream_modes", "emit_values"),
|
||||
[
|
||||
(["messages-tuple", "values"], True),
|
||||
(["messages-tuple", "updates", "custom"], False),
|
||||
],
|
||||
)
|
||||
@pytest.mark.anyio
|
||||
async def test_run_agent_batches_incremental_file_args_and_keeps_complete_values():
|
||||
async def test_run_agent_batches_incremental_file_args_and_keeps_complete_values(stream_modes: list[str], emit_values: bool):
|
||||
run_manager = RunManager()
|
||||
record = await run_manager.create("thread-file-stream")
|
||||
bridge = SimpleNamespace(
|
||||
@ -696,7 +703,10 @@ async def test_run_agent_batches_incremental_file_args_and_keeps_complete_values
|
||||
{},
|
||||
),
|
||||
)
|
||||
yield ("values", {"messages": [complete_message]})
|
||||
if emit_values:
|
||||
yield ("values", {"messages": [complete_message]})
|
||||
else:
|
||||
yield ("updates", {"agent": {}})
|
||||
|
||||
await run_agent(
|
||||
bridge,
|
||||
@ -706,14 +716,15 @@ async def test_run_agent_batches_incremental_file_args_and_keeps_complete_values
|
||||
agent_factory=lambda **_kwargs: DummyAgent(),
|
||||
graph_input={},
|
||||
config={},
|
||||
stream_modes=["messages-tuple", "values"],
|
||||
stream_modes=stream_modes,
|
||||
)
|
||||
|
||||
message_events = [call.args for call in bridge.publish.await_args_list if call.args[1] == "messages"]
|
||||
assert len(message_events) == 1
|
||||
assert message_events[0][2][0]["tool_calls"][0]["args"]["content"] == "Hello world"
|
||||
values_events = [call.args[2] for call in bridge.publish.await_args_list if call.args[1] == "values"]
|
||||
assert any(event["messages"][0]["tool_calls"][0]["args"]["content"] == "Hello world" for event in values_events)
|
||||
if emit_values:
|
||||
values_events = [call.args[2] for call in bridge.publish.await_args_list if call.args[1] == "values"]
|
||||
assert any(event["messages"][0]["tool_calls"][0]["args"]["content"] == "Hello world" for event in values_events)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
||||
@ -11,7 +11,7 @@
|
||||
full-file action; do not mount CodeMirror for that artifact until the user
|
||||
requests and receives the complete content. The Gateway retains range
|
||||
ownership and returns 206/416 through `FileResponse`.
|
||||
3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. A checkpoint/transient prefix whose canonical position is still behind an unloaded cursor page is woven in before the first shared anchor, not discarded: both the checkpoint and seq-sorted history place it earlier, so that position is known even when the pages between are not. It must never be appended to the tail (#4065) — the tail is provably wrong — but suppressing it entirely is how a user's own question vanished from a long thread once the first 50-row history page no longer reached back to it (#4666). A collapsed unloaded gap is recoverable by paging; a dropped message is not. Weaving alone restores the message but not its exact position — after compaction the live window carries too few anchors — so both sides now carry the backend's thread-global `additional_kwargs.deerflow_seq`: `buildVisibleHistoryMessages` copies each row's `seq`, and the Gateway stamps it onto `values` frame messages it has already persisted. A live message whose seq is below the loaded window's lower bound is placed ahead of everything on screen instead of before the nearest anchor, which is what puts a compaction-rescued first user turn back at the head rather than mid-transcript. That split happens _before_ the anchor walk, not inside it: a compacted checkpoint can share no identity at all with the loaded page — it keeps only the current run's recent tail, while the page on screen was fetched turns earlier — and the anchor walk then never runs at all, which is precisely when a rescued turn most needs its seq. Doing the split inside the walk left that case appending the message after the whole window (#4666), the one arrangement #4065 proved wrong. A message without a seq (still streaming, so not in the feed yet) keeps the weaving path — the tail is already its correct position. Optimistic messages are then added without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys the submitted user message from `X` to `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the submitted frame and checkpoint replacement remain one visible turn. A locally submitted turn also records its pre-submit identity baseline: if `messages-tuple` publishes new AI/tool steps before `values` publishes that turn's human message, render ordering moves only those non-baseline visible steps behind the new human while leaving history, hidden controls, and reconnected runs untouched. Keep that local order anchor through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery.
|
||||
3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. A checkpoint/transient prefix whose canonical position is still behind an unloaded cursor page is woven in before the first shared anchor, not discarded: both the checkpoint and seq-sorted history place it earlier, so that position is known even when the pages between are not. It must never be appended to the tail (#4065) — the tail is provably wrong — but suppressing it entirely is how a user's own question vanished from a long thread once the first 50-row history page no longer reached back to it (#4666). A collapsed unloaded gap is recoverable by paging; a dropped message is not. Weaving alone restores the message but not its exact position — after compaction the live window carries too few anchors — so both sides now carry the backend's thread-global `additional_kwargs.deerflow_seq`: `buildVisibleHistoryMessages` copies each row's `seq`, and the Gateway stamps it onto `values` frame messages it has already persisted. A live message whose seq is below the loaded window's lower bound is placed ahead of everything on screen instead of before the nearest anchor, which is what puts a compaction-rescued first user turn back at the head rather than mid-transcript. That split happens _before_ the anchor walk, not inside it: a compacted checkpoint can share no identity at all with the loaded page — it keeps only the current run's recent tail, while the page on screen was fetched turns earlier — and the anchor walk then never runs at all, which is precisely when a rescued turn most needs its seq. Doing the split inside the walk left that case appending the message after the whole window (#4666), the one arrangement #4065 proved wrong. A message without a seq (still streaming, so not in the feed yet) keeps the weaving path — the tail is already its correct position. Optimistic messages are then added without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys the submitted user message from `X` to `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the submitted frame and checkpoint replacement remain one visible turn. A locally submitted turn also records its pre-submit identity baseline: if `messages-tuple` publishes new AI/tool steps before canonical history (or replay-gap recovery) publishes that turn's human message, render ordering moves only those non-baseline visible steps behind the new human while leaving history, hidden controls, and reconnected runs untouched. Keep that local order anchor through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery.
|
||||
4. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits
|
||||
5. TanStack Query manages server state; localStorage stores user settings. The
|
||||
Settings > Tools MCP switch calls the targeted `PATCH /api/mcp/config`
|
||||
@ -79,7 +79,7 @@ Composer drafts are tab-scoped browser state. `core/threads/composer-draft.ts` s
|
||||
|
||||
Auth UI note: the login page's "keep me signed in" option submits only `remember_me` to the Gateway and may persist only the email address through `core/auth/remember-login.ts`. Passwords and tokens must never be stored in frontend storage; the `HttpOnly access_token` and readable `csrf_token` cookies remain Gateway-owned.
|
||||
|
||||
`/goal` and `/compact` are built-in composer commands, not skill activations. `src/components/workspace/input-box.tsx` intercepts `/goal`, `/goal clear`, and `/goal <condition>` before normal chat submission, calling Gateway `GET/PUT/DELETE /api/threads/{thread_id}/goal`. Setting `/goal <condition>` also submits the condition text as the next user task so the agent starts running immediately; status and clear do not start a run. Goal and compact requests are tied to the current `threadId` with an `AbortController`, so switching threads or unmounting the composer aborts in-flight requests and stale responses cannot update the new thread's composer state. The chat pages render `GoalStatus` above the composer from `AgentThreadState.goal`, with local optimistic state until the next stream `values` update arrives. `/compact` calls `POST /api/threads/{thread_id}/compact` to summarize older active context while leaving the full visible chat history intact; it is skipped on new/empty threads and blocked server-side while a run is in flight. Thread rename uses the same serialized state-write route; the rename dialog stays open and surfaces the server error when an active run returns 409.
|
||||
`/goal` and `/compact` are built-in composer commands, not skill activations. `src/components/workspace/input-box.tsx` intercepts `/goal`, `/goal clear`, and `/goal <condition>` before normal chat submission, calling Gateway `GET/PUT/DELETE /api/threads/{thread_id}/goal`. Setting `/goal <condition>` also submits the condition text as the next user task so the agent starts running immediately; status and clear do not start a run. Goal and compact requests are tied to the current `threadId` with an `AbortController`, so switching threads or unmounting the composer aborts in-flight requests and stale responses cannot update the new thread's composer state. The chat pages render `GoalStatus` above the composer from `AgentThreadState.goal`, with local optimistic state until an incremental goal update or final state reload arrives. `/compact` calls `POST /api/threads/{thread_id}/compact` to summarize older active context while leaving the full visible chat history intact; it is skipped on new/empty threads and blocked server-side while a run is in flight. Thread rename uses the same serialized state-write route; the rename dialog stays open and surfaces the server error when an active run returns 409.
|
||||
|
||||
The `/` skill list stays reachable after a skill is selected: typing `/` in the editable text beside the chip reopens it, and picking an entry swaps the chip rather than adding a second one, because the wire format carries exactly one leading `/skill`. That list offers skills only while a chip is selected — a builtin command owns the whole composer line, so `/goal` behind a selected skill would submit as chat text instead of running the command. The trigger itself is unchanged: a slash only opens the list at the start of the input (`getLeadingSlashSkillQuery`), pinned by `tests/e2e/chat.spec.ts`.
|
||||
|
||||
@ -107,7 +107,7 @@ Edit-and-rerun is deliberately latest-turn-only. `core/messages/utils.ts::getLat
|
||||
- **Thread hooks** (`useThreadStream`, `useSubmitThread`, `useThreads`) are the primary API interface
|
||||
- **Thread routes** — construct Web UI chat paths through `core/threads/utils.ts::pathOfThread()`, which percent-encodes both custom agent names and thread IDs before inserting them into route segments
|
||||
- **LangGraph client** is a singleton obtained via `getAPIClient()` in `core/api/`
|
||||
- **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. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded.
|
||||
- **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 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.
|
||||
- **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.
|
||||
- **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.
|
||||
|
||||
@ -12,7 +12,7 @@ import {
|
||||
import type { AgentThreadState } from "../threads/types";
|
||||
|
||||
import { isStateChangingMethod, readCsrfCookie } from "./fetcher";
|
||||
import { sanitizeRunStreamOptions } from "./stream-mode";
|
||||
import { forceChatRunStreamOptions } from "./stream-mode";
|
||||
|
||||
/**
|
||||
* SDK ``onRequest`` hook that mints the ``X-CSRF-Token`` header from the
|
||||
@ -340,7 +340,7 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient {
|
||||
// this return value with `for await`, so run creation still starts on first
|
||||
// iteration rather than when `runs.stream()` is called.
|
||||
client.runs.stream = async function* (threadId, assistantId, payload) {
|
||||
const sanitizedPayload = sanitizeRunStreamOptions(payload);
|
||||
const sanitizedPayload = forceChatRunStreamOptions(payload);
|
||||
const originalOnRunCreated = sanitizedPayload?.onRunCreated;
|
||||
let runId: string | undefined;
|
||||
const initialStream = originalRunStream(threadId, assistantId, {
|
||||
@ -402,7 +402,7 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient {
|
||||
clearReconnectRun(threadId, runId);
|
||||
return;
|
||||
}
|
||||
const sanitizedOptions = sanitizeRunStreamOptions(options);
|
||||
const sanitizedOptions = forceChatRunStreamOptions(options);
|
||||
yield* handleInactiveRunStream({
|
||||
threadId,
|
||||
expectedRunId: () => runId,
|
||||
|
||||
@ -8,6 +8,12 @@ const SUPPORTED_RUN_STREAM_MODES = new Set([
|
||||
"custom",
|
||||
] as const);
|
||||
|
||||
export const CHAT_RUN_STREAM_MODES = [
|
||||
"messages-tuple",
|
||||
"updates",
|
||||
"custom",
|
||||
] as const;
|
||||
|
||||
const warnedUnsupportedStreamModes = new Set<string>();
|
||||
let warnedUnsupportedStreamResumable = false;
|
||||
|
||||
@ -73,3 +79,34 @@ export function sanitizeRunStreamOptions<T>(options: T): T {
|
||||
|
||||
return sanitizedOptions;
|
||||
}
|
||||
|
||||
/**
|
||||
* Keep chat streams on incremental events only. Without an explicit mode list,
|
||||
* the SDK's lazy message tracking also requests `values`, retransmitting the
|
||||
* full thread state (including message history) after graph steps.
|
||||
*/
|
||||
export function forceChatRunStreamOptions<T>(options: T): T {
|
||||
const sanitizedOptions = sanitizeRunStreamOptions(options);
|
||||
const preservedOptions =
|
||||
typeof AbortSignal !== "undefined" &&
|
||||
sanitizedOptions instanceof AbortSignal
|
||||
? { signal: sanitizedOptions }
|
||||
: typeof sanitizedOptions === "object" && sanitizedOptions !== null
|
||||
? sanitizedOptions
|
||||
: {};
|
||||
const requestedMode = Reflect.get(preservedOptions, "streamMode");
|
||||
const streamModes = new Set<string>([
|
||||
...CHAT_RUN_STREAM_MODES,
|
||||
...((Array.isArray(requestedMode)
|
||||
? requestedMode
|
||||
: requestedMode == null
|
||||
? []
|
||||
: [requestedMode]) as string[]),
|
||||
]);
|
||||
streamModes.delete("values");
|
||||
|
||||
return {
|
||||
...preservedOptions,
|
||||
streamMode: [...streamModes],
|
||||
} as T;
|
||||
}
|
||||
|
||||
@ -40,6 +40,10 @@ import {
|
||||
patchThreadMetadata,
|
||||
type ThreadMetadataPatch,
|
||||
} from "./api";
|
||||
import {
|
||||
hasRenderedThreadStateUpdate,
|
||||
reduceThreadStateUpdates,
|
||||
} from "./stream-state";
|
||||
import {
|
||||
buildThreadsSearchQueryOptions,
|
||||
DEFAULT_THREAD_SEARCH_PARAMS,
|
||||
@ -659,8 +663,8 @@ export function mergeMessages(
|
||||
/**
|
||||
* Keep messages from a locally submitted turn behind that turn's user input.
|
||||
* LangGraph `messages-tuple` events can publish the first AI/tool steps before
|
||||
* the `values` event containing the user message. Those steps are not part of
|
||||
* the pre-submit baseline, so move only that visible pending segment behind the
|
||||
* canonical history contains the user message. Those steps are not part of the
|
||||
* pre-submit baseline, so move only that visible pending segment behind the
|
||||
* first new human message without disturbing established history or hidden
|
||||
* checkpoint controls. The caller keeps the baseline after stream completion
|
||||
* because the SDK may retain its transient event order until the next submit.
|
||||
@ -1774,7 +1778,11 @@ export function useThreadStream({
|
||||
.catch(() => ({}));
|
||||
}
|
||||
},
|
||||
onUpdateEvent(data) {
|
||||
onUpdateEvent(data, { mutate }) {
|
||||
if (hasRenderedThreadStateUpdate(data)) {
|
||||
mutate((previous) => reduceThreadStateUpdates(previous, data) ?? {});
|
||||
}
|
||||
|
||||
const _messages = getSummarizationMiddlewareMessages(data);
|
||||
if (_messages && _messages.length >= 2) {
|
||||
for (const m of _messages) {
|
||||
@ -2062,9 +2070,9 @@ export function useThreadStream({
|
||||
|
||||
// Clear optimistic when server messages arrive.
|
||||
// For messages with a human optimistic message, wait until the server's
|
||||
// human message has arrived to avoid clearing before the input message
|
||||
// appears in the stream (the input message may arrive via "values" events
|
||||
// after individual "messages-tuple" events for AI messages).
|
||||
// human message has arrived to avoid clearing before canonical history (or
|
||||
// replay-gap recovery) reports the input after individual messages-tuple
|
||||
// events for AI messages.
|
||||
const optimisticMessageCount = optimisticMessages.length;
|
||||
const hasHumanOptimistic = optimisticMessages.some((m) => m.type === "human");
|
||||
useEffect(() => {
|
||||
|
||||
108
frontend/src/core/threads/stream-state.ts
Normal file
108
frontend/src/core/threads/stream-state.ts
Normal file
@ -0,0 +1,108 @@
|
||||
import type { AgentThreadState, GoalState } from "./types";
|
||||
|
||||
type ThreadStatePatch = Partial<AgentThreadState>;
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function isGoalState(value: unknown): value is GoalState {
|
||||
return (
|
||||
isRecord(value) &&
|
||||
typeof value.objective === "string" &&
|
||||
value.status === "active" &&
|
||||
typeof value.created_at === "string" &&
|
||||
typeof value.updated_at === "string"
|
||||
);
|
||||
}
|
||||
|
||||
const RENDERED_THREAD_STATE_KEYS = ["title", "artifacts", "todos", "goal"];
|
||||
|
||||
export function hasRenderedThreadStateUpdate(data: unknown): boolean {
|
||||
if (!isRecord(data)) return false;
|
||||
return Object.values(data).some(
|
||||
(update) =>
|
||||
isRecord(update) &&
|
||||
RENDERED_THREAD_STATE_KEYS.some((key) => Object.hasOwn(update, key)),
|
||||
);
|
||||
}
|
||||
|
||||
function mergeArtifacts(
|
||||
existing: AgentThreadState["artifacts"],
|
||||
incoming: unknown,
|
||||
): string[] | undefined {
|
||||
if (incoming == null || !Array.isArray(incoming)) {
|
||||
return undefined;
|
||||
}
|
||||
if (!incoming.every((path) => typeof path === "string")) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return [...new Set([...(existing ?? []), ...incoming])];
|
||||
}
|
||||
|
||||
/**
|
||||
* Fold a LangGraph `updates` frame into the state fields rendered by the chat
|
||||
* UI. Updates are grouped by node name and carry reducer inputs, not complete
|
||||
* state snapshots, so these fields must mirror the reducers in
|
||||
* `deerflow.agents.thread_state` rather than being shallowly assigned.
|
||||
*
|
||||
* `messages` is deliberately excluded. The SDK's `messages-tuple` manager owns
|
||||
* chunk assembly and same-id replacement; applying the node's messages update
|
||||
* through `mutate` as well would duplicate messages and bypass chunk merging.
|
||||
*/
|
||||
export function reduceThreadStateUpdates(
|
||||
previous: AgentThreadState,
|
||||
data: unknown,
|
||||
): ThreadStatePatch | undefined {
|
||||
if (!isRecord(data)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const patch: ThreadStatePatch = {};
|
||||
let artifacts = previous.artifacts;
|
||||
let hasPatch = false;
|
||||
|
||||
for (const update of Object.values(data)) {
|
||||
if (!isRecord(update)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (Object.hasOwn(update, "title") && typeof update.title === "string") {
|
||||
patch.title = update.title;
|
||||
hasPatch = true;
|
||||
}
|
||||
|
||||
if (Object.hasOwn(update, "artifacts")) {
|
||||
const mergedArtifacts = mergeArtifacts(artifacts, update.artifacts);
|
||||
if (mergedArtifacts !== undefined) {
|
||||
artifacts = mergedArtifacts;
|
||||
patch.artifacts = mergedArtifacts;
|
||||
hasPatch = true;
|
||||
}
|
||||
}
|
||||
|
||||
// DeerFlow's merge_todos treats null as "this node did not touch todos"
|
||||
// and an empty list as an explicit clear.
|
||||
if (
|
||||
Object.hasOwn(update, "todos") &&
|
||||
update.todos !== null &&
|
||||
Array.isArray(update.todos)
|
||||
) {
|
||||
patch.todos = update.todos;
|
||||
hasPatch = true;
|
||||
}
|
||||
|
||||
// merge_goal likewise preserves the prior goal for null writes.
|
||||
if (
|
||||
Object.hasOwn(update, "goal") &&
|
||||
update.goal !== null &&
|
||||
isGoalState(update.goal)
|
||||
) {
|
||||
patch.goal = update.goal;
|
||||
hasPatch = true;
|
||||
}
|
||||
}
|
||||
|
||||
return hasPatch ? patch : undefined;
|
||||
}
|
||||
@ -312,6 +312,116 @@ test("proceeds to join when the run is still active", async () => {
|
||||
expect(sessionStorage.removeItem).toHaveBeenCalledWith("lg:stream:thread-1");
|
||||
});
|
||||
|
||||
test("requests incremental modes for initial and rejoined chat streams", async () => {
|
||||
const sessionStorage = makeSessionStorage();
|
||||
let initialStreamBody: Record<string, unknown> | undefined;
|
||||
let joinedStreamModes: unknown;
|
||||
const fetchFn = rs.fn(async (url: string | URL, init?: RequestInit) => {
|
||||
const requestUrl = new URL(url.toString());
|
||||
if (requestUrl.pathname.endsWith("/threads/thread-modes/runs/stream")) {
|
||||
if (typeof init?.body !== "string") {
|
||||
throw new Error("Expected a JSON request body for the initial stream");
|
||||
}
|
||||
initialStreamBody = JSON.parse(init.body);
|
||||
return makeSSEResponse("event: end\ndata: null\n\n", {
|
||||
"Content-Location": "/threads/thread-modes/runs/run-modes",
|
||||
});
|
||||
}
|
||||
if (requestUrl.pathname.endsWith("/runs/run-modes")) {
|
||||
return new Response(JSON.stringify({ status: "running" }), {
|
||||
status: 200,
|
||||
});
|
||||
}
|
||||
if (requestUrl.pathname.endsWith("/runs/run-modes/stream")) {
|
||||
joinedStreamModes = JSON.parse(
|
||||
requestUrl.searchParams.get("stream_mode") ?? "null",
|
||||
);
|
||||
return makeSSEResponse("event: end\ndata: null\n\n");
|
||||
}
|
||||
return new Response(JSON.stringify({ detail: "unexpected request" }), {
|
||||
status: 500,
|
||||
});
|
||||
});
|
||||
rs.stubGlobal("window", {
|
||||
location: { origin: "http://localhost:2026" },
|
||||
sessionStorage,
|
||||
});
|
||||
rs.stubGlobal("fetch", fetchFn);
|
||||
|
||||
for await (const _entry of getAPIClient(true).runs.stream(
|
||||
"thread-modes",
|
||||
"lead_agent",
|
||||
{ streamMode: ["values"] },
|
||||
)) {
|
||||
// Drain the initial stream so its lazy request is issued.
|
||||
void _entry;
|
||||
}
|
||||
for await (const _entry of getAPIClient(true).runs.joinStream(
|
||||
"thread-modes",
|
||||
"run-modes",
|
||||
{ streamMode: ["values"] },
|
||||
)) {
|
||||
// Drain the rejoined stream so its lazy request is issued.
|
||||
void _entry;
|
||||
}
|
||||
|
||||
const incrementalModes = ["messages-tuple", "updates", "custom"];
|
||||
expect(initialStreamBody?.stream_mode).toEqual(incrementalModes);
|
||||
expect(joinedStreamModes).toEqual(incrementalModes);
|
||||
});
|
||||
|
||||
test("passes AbortSignals through initial and directly-signalled join streams", async () => {
|
||||
const sessionStorage = makeSessionStorage();
|
||||
const initialController = new AbortController();
|
||||
const joinController = new AbortController();
|
||||
let initialSignal: AbortSignal | null | undefined;
|
||||
let joinSignal: AbortSignal | null | undefined;
|
||||
const fetchFn = rs.fn(async (url: string | URL, init?: RequestInit) => {
|
||||
const requestUrl = new URL(url.toString());
|
||||
if (requestUrl.pathname.endsWith("/threads/thread-signal/runs/stream")) {
|
||||
initialSignal = init?.signal;
|
||||
return makeSSEResponse("event: end\ndata: null\n\n", {
|
||||
"Content-Location": "/threads/thread-signal/runs/run-signal",
|
||||
});
|
||||
}
|
||||
if (requestUrl.pathname.endsWith("/runs/run-signal")) {
|
||||
return new Response(JSON.stringify({ status: "running" }), {
|
||||
status: 200,
|
||||
});
|
||||
}
|
||||
if (requestUrl.pathname.endsWith("/runs/run-signal/stream")) {
|
||||
joinSignal = init?.signal;
|
||||
return makeSSEResponse("event: end\ndata: null\n\n");
|
||||
}
|
||||
return new Response(JSON.stringify({ detail: "unexpected request" }), {
|
||||
status: 500,
|
||||
});
|
||||
});
|
||||
rs.stubGlobal("window", {
|
||||
location: { origin: "http://localhost:2026" },
|
||||
sessionStorage,
|
||||
});
|
||||
rs.stubGlobal("fetch", fetchFn);
|
||||
|
||||
for await (const _entry of getAPIClient(true).runs.stream(
|
||||
"thread-signal",
|
||||
"lead_agent",
|
||||
{ signal: initialController.signal },
|
||||
)) {
|
||||
void _entry;
|
||||
}
|
||||
for await (const _entry of getAPIClient(true).runs.joinStream(
|
||||
"thread-signal",
|
||||
"run-signal",
|
||||
joinController.signal,
|
||||
)) {
|
||||
void _entry;
|
||||
}
|
||||
|
||||
expect(initialSignal).toBe(initialController.signal);
|
||||
expect(joinSignal).toBe(joinController.signal);
|
||||
});
|
||||
|
||||
test("recovers a join stream gap from durable state and resumes after the retained tail", async () => {
|
||||
const sessionStorage = makeSessionStorage();
|
||||
sessionStorage.setItem("lg:stream:thread-1", "run-1");
|
||||
|
||||
@ -1,6 +1,10 @@
|
||||
import { expect, test } from "@rstest/core";
|
||||
|
||||
import { sanitizeRunStreamOptions } from "@/core/api/stream-mode";
|
||||
import {
|
||||
CHAT_RUN_STREAM_MODES,
|
||||
forceChatRunStreamOptions,
|
||||
sanitizeRunStreamOptions,
|
||||
} from "@/core/api/stream-mode";
|
||||
|
||||
test("rejects mixed supported and unsupported stream modes", () => {
|
||||
expect(() =>
|
||||
@ -61,3 +65,40 @@ test("sanitizes streamResumable while preserving valid stream modes", () => {
|
||||
streamMode: ["values", "custom"],
|
||||
});
|
||||
});
|
||||
|
||||
test("forces incremental modes for chat streams instead of values snapshots", () => {
|
||||
const sanitized = forceChatRunStreamOptions({
|
||||
streamResumable: true,
|
||||
streamMode: ["values", "messages-tuple", "updates", "custom", "debug"],
|
||||
signal: "keep-me",
|
||||
});
|
||||
|
||||
expect(sanitized).toEqual({
|
||||
signal: "keep-me",
|
||||
streamMode: [...CHAT_RUN_STREAM_MODES, "debug"],
|
||||
});
|
||||
expect(sanitized.streamMode).not.toContain("values");
|
||||
});
|
||||
|
||||
test("adds explicit chat stream modes when no options are provided", () => {
|
||||
expect(forceChatRunStreamOptions(undefined)).toEqual({
|
||||
streamMode: [...CHAT_RUN_STREAM_MODES],
|
||||
});
|
||||
});
|
||||
|
||||
test("preserves a direct AbortSignal while adding chat stream modes", () => {
|
||||
const signal = new AbortController().signal;
|
||||
|
||||
expect(forceChatRunStreamOptions(signal)).toEqual({
|
||||
signal,
|
||||
streamMode: [...CHAT_RUN_STREAM_MODES],
|
||||
});
|
||||
});
|
||||
|
||||
test("rejects unsupported chat stream modes before replacing them", () => {
|
||||
expect(() =>
|
||||
forceChatRunStreamOptions({
|
||||
streamMode: ["messages-tuple", "events"],
|
||||
}),
|
||||
).toThrow("Unsupported LangGraph stream mode(s): events");
|
||||
});
|
||||
|
||||
@ -0,0 +1,145 @@
|
||||
import type { Message } from "@langchain/langgraph-sdk";
|
||||
import { expect, rs, test } from "@rstest/core";
|
||||
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
||||
import { act, renderHook } from "@testing-library/react";
|
||||
import { createElement, type ReactNode } from "react";
|
||||
|
||||
import { I18nContext } from "@/core/i18n/context";
|
||||
import { enUS } from "@/core/i18n/locales/en-US";
|
||||
import { DEFAULT_LOCAL_SETTINGS } from "@/core/settings/local";
|
||||
import type { AgentThreadState, GoalState } from "@/core/threads/types";
|
||||
|
||||
type StreamOptions = {
|
||||
onUpdateEvent?: (
|
||||
data: unknown,
|
||||
options: {
|
||||
mutate: (
|
||||
update:
|
||||
| Partial<AgentThreadState>
|
||||
| ((previous: AgentThreadState) => Partial<AgentThreadState>),
|
||||
) => void;
|
||||
},
|
||||
) => void;
|
||||
};
|
||||
|
||||
const streamMockState = rs.hoisted(() => {
|
||||
const existingMessage = {
|
||||
type: "ai",
|
||||
id: "message-1",
|
||||
content: "Already streamed",
|
||||
} as Message;
|
||||
return {
|
||||
existingMessage,
|
||||
options: undefined as StreamOptions | undefined,
|
||||
values: {
|
||||
title: "Before",
|
||||
messages: [existingMessage],
|
||||
artifacts: ["old.md"],
|
||||
todos: [{ content: "old", status: "pending" as const }],
|
||||
goal: null,
|
||||
} as AgentThreadState,
|
||||
};
|
||||
});
|
||||
|
||||
const existingMessage = streamMockState.existingMessage;
|
||||
|
||||
rs.mock("@langchain/langgraph-sdk/react", () => ({
|
||||
useStream: (options: StreamOptions) => {
|
||||
streamMockState.options = options;
|
||||
return {
|
||||
isLoading: true,
|
||||
messages: streamMockState.values.messages,
|
||||
stop: async () => undefined,
|
||||
submit: async () => undefined,
|
||||
values: streamMockState.values,
|
||||
};
|
||||
},
|
||||
}));
|
||||
|
||||
test("updates rendered thread state without receiving a values frame", async () => {
|
||||
const { useThreadStream } = await import("@/core/threads/hooks");
|
||||
const queryClient = new QueryClient({
|
||||
defaultOptions: { queries: { retry: false } },
|
||||
});
|
||||
const wrapper = ({ children }: { children: ReactNode }) =>
|
||||
createElement(
|
||||
QueryClientProvider,
|
||||
{ client: queryClient },
|
||||
createElement(
|
||||
I18nContext.Provider,
|
||||
{
|
||||
value: {
|
||||
locale: "en-US",
|
||||
setLocale: () => undefined,
|
||||
t: enUS,
|
||||
},
|
||||
},
|
||||
children,
|
||||
),
|
||||
);
|
||||
const { rerender, result } = renderHook(
|
||||
() =>
|
||||
useThreadStream({
|
||||
context: DEFAULT_LOCAL_SETTINGS.context,
|
||||
isMock: true,
|
||||
threadId: "thread-1",
|
||||
}),
|
||||
{ wrapper },
|
||||
);
|
||||
|
||||
const goal = {
|
||||
objective: "Finish the report",
|
||||
status: "active",
|
||||
created_at: "2026-09-02T00:00:00Z",
|
||||
updated_at: "2026-09-02T00:01:00Z",
|
||||
continuation_count: 1,
|
||||
max_continuations: 8,
|
||||
no_progress_count: 0,
|
||||
} as GoalState;
|
||||
const todos = [{ content: "Draft", status: "in_progress" as const }];
|
||||
|
||||
act(() => {
|
||||
streamMockState.options?.onUpdateEvent?.(
|
||||
{ agent: { messages: [existingMessage], summary_text: "internal" } },
|
||||
{
|
||||
mutate() {
|
||||
throw new Error("irrelevant updates must not mutate thread values");
|
||||
},
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
streamMockState.options?.onUpdateEvent?.(
|
||||
{
|
||||
agent: {
|
||||
artifacts: ["report.md"],
|
||||
goal,
|
||||
// The same message also arrives through messages-tuple. The update
|
||||
// path must not apply it a second time.
|
||||
messages: [existingMessage],
|
||||
title: "After",
|
||||
todos,
|
||||
},
|
||||
},
|
||||
{
|
||||
mutate(update) {
|
||||
const patch =
|
||||
typeof update === "function"
|
||||
? update(streamMockState.values)
|
||||
: update;
|
||||
streamMockState.values = { ...streamMockState.values, ...patch };
|
||||
},
|
||||
},
|
||||
);
|
||||
rerender();
|
||||
});
|
||||
|
||||
expect(result.current.thread.values).toMatchObject({
|
||||
artifacts: ["old.md", "report.md"],
|
||||
goal,
|
||||
title: "After",
|
||||
todos,
|
||||
});
|
||||
expect(result.current.thread.messages).toEqual([existingMessage]);
|
||||
});
|
||||
Loading…
x
Reference in New Issue
Block a user