From 46ef0c59c3020566c9c4c103f6a87bc323224716 Mon Sep 17 00:00:00 2001 From: haochen <3033940900@qq.com> Date: Thu, 24 Sep 2026 15:11:49 +0800 Subject: [PATCH 1/6] fix(channels): persist runtime configuration as UTF-8 (#5803) Co-authored-by: Zhangchenhao-111 <285911435+Zhangchenhao-111@users.noreply.github.com> --- backend/app/channels/runtime_config_store.py | 1 + .../test_channel_runtime_config_encoding.py | 35 +++++++++++++++++++ 2 files changed, 36 insertions(+) create mode 100644 backend/tests/test_channel_runtime_config_encoding.py diff --git a/backend/app/channels/runtime_config_store.py b/backend/app/channels/runtime_config_store.py index 2c6e599b3f..ec7b810bfb 100644 --- a/backend/app/channels/runtime_config_store.py +++ b/backend/app/channels/runtime_config_store.py @@ -46,6 +46,7 @@ class ChannelRuntimeConfigStore: def _save(self) -> None: fd = tempfile.NamedTemporaryFile( mode="w", + encoding="utf-8", dir=self._path.parent, suffix=".tmp", delete=False, diff --git a/backend/tests/test_channel_runtime_config_encoding.py b/backend/tests/test_channel_runtime_config_encoding.py new file mode 100644 index 0000000000..a628a79803 --- /dev/null +++ b/backend/tests/test_channel_runtime_config_encoding.py @@ -0,0 +1,35 @@ +"""Runtime channel configuration must round-trip independently of host locale.""" + +import json + +import pytest + +from app.channels import runtime_config_store + + +@pytest.mark.parametrize( + ("locale_encoding", "relay_url"), + [ + ("cp1252", "wss://relay.example/caf\u00e9"), + ("gbk", "wss://relay.example/\u804a\u5929"), + ("cp1252", "wss://relay.example/\u804a\u5929"), + ], +) +def test_runtime_config_round_trips_non_ascii_with_legacy_locale(tmp_path, monkeypatch, locale_encoding, relay_url): + named_temporary_file = runtime_config_store.tempfile.NamedTemporaryFile + + def locale_temporary_file(*args, **kwargs): + # Simulate a legacy host default only when the writer omits encoding. + kwargs.setdefault("encoding", locale_encoding) + return named_temporary_file(*args, **kwargs) + + monkeypatch.setattr(runtime_config_store.tempfile, "NamedTemporaryFile", locale_temporary_file) + path = tmp_path / "channels" / "runtime-config.json" + config = {"enabled": True, "relay_url": relay_url} + store = runtime_config_store.ChannelRuntimeConfigStore(path) + + store.set_provider_config("buzz", config) + + assert json.loads(path.read_text(encoding="utf-8")) == {"buzz": config} + reloaded = runtime_config_store.ChannelRuntimeConfigStore(path) + assert reloaded.get_provider_config("buzz") == config From 7cf0abe58430f018c3ea025fcaca9bdc98c05408 Mon Sep 17 00:00:00 2001 From: Zeren Wang <53075619+Vanzeren@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:16:08 +0200 Subject: [PATCH 2/6] test(checkpoint): delta-history cache migration coverage + pin two open DeltaChannel defects (#5797) * test(checkpoint): cover delta-history cache migration and pin two open DeltaChannel defects - Extend the full -> delta migration contract to {cache disabled, cold, warm} with a differential raw-saver oracle; the warm state also asserts a cache-served re-read with no extra inner history walks. - Pin two still-unfixed upstream DeltaChannel defects as tripwire regressions: #8382 (parallel-superstep delta replay order diverges from live order) and #8448 (Postgres paged delta walk hydrates old checkpoints empty). Each asserts the correct contract and skips, naming the issue and pinned version, while the defect is present - so a later dependency bump that lands the fix turns it into a live gate. The contract is never asserted as buggy behaviour (see backend/tests/AGENTS.md). * test(checkpoint): scope the #8448 skip to the postgres parameter The paged stage-1 walk that langgraph#8448 poisons only exists in langgraph.checkpoint.postgres; memory and sqlite never paginate. An unscoped skip let a future regression in long-chain materialization on those backends hide behind the Postgres-only upstream issue instead of failing the InMemorySaver differential oracle. Keep the assertion live there and trip only where the defect can occur. --- .../test_cached_history_saver_integration.py | 87 +++++++++++ .../tests/test_delta_channel_checkpointers.py | 142 +++++++++++++++++- 2 files changed, 228 insertions(+), 1 deletion(-) diff --git a/backend/tests/test_cached_history_saver_integration.py b/backend/tests/test_cached_history_saver_integration.py index 942f8622a0..0c3a6ba24e 100644 --- a/backend/tests/test_cached_history_saver_integration.py +++ b/backend/tests/test_cached_history_saver_integration.py @@ -27,6 +27,7 @@ from langchain_core.messages import AIMessage, AnyMessage, HumanMessage from langgraph.channels import DeltaChannel from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph +from langgraph.graph.message import add_messages from langgraph.types import Command, interrupt from deerflow.agents.thread_state import merge_message_writes @@ -362,3 +363,89 @@ async def test_rollback_supersede_does_not_pollute() -> None: oracle_reread = await oracle_accessor.aget(oracle_by_next[next_key].config) assert _digest(reread.values) == _expected_final_digest()[:expected_len] assert _digest(reread.values) == _digest(oracle_reread.values) + + +class _FullState(TypedDict): + """Pre-delta (full) thread schema: ``messages`` is a plain list channel.""" + + messages: Annotated[list[AnyMessage], add_messages] + + +def _noop(state: dict[str, Any]) -> dict[str, Any]: + return {} + + +def _build_noop_graph(saver: Any, schema: Any) -> Any: + builder = StateGraph(schema) + builder.add_node("noop", _noop) + builder.set_entry_point("noop") + builder.set_finish_point("noop") + return builder.compile(checkpointer=saver) + + +_MIGRATION_DELTA_STEPS = 3 + + +async def _migrate_full_to_delta(saver: Any, config: dict[str, Any]) -> tuple[list, list, list]: + """Seed a thread in full (pre-delta) mode, then swap to delta on it. + + Returns ``(migrated, replayed, history)``: the digest right after the mode + swap (the plain-value seed must read back intact), the head after + ``_MIGRATION_DELTA_STEPS`` delta appends, and every checkpoint's digest. + """ + full_graph = _build_noop_graph(saver, _FullState) + await full_graph.ainvoke(_input(), config) + + delta_graph = _build_noop_graph(saver, _state_schema()) + accessor = CheckpointStateAccessor.bind(delta_graph, saver, mode="delta") + migrated = _digest((await accessor.aget(config)).values) + + for i in range(_MIGRATION_DELTA_STEPS): + await delta_graph.ainvoke({"messages": [AIMessage(content=f"delta-{i}", id=f"ai-d{i}")]}, config) + + replayed = _digest((await accessor.aget(config)).values) + history = _history_digests(await accessor.ahistory(config)) + return migrated, replayed, history + + +_CACHE_STATES = ("disabled", "cold", "warm") + + +@pytest.mark.parametrize("cache_state", _CACHE_STATES) +@pytest.mark.anyio +async def test_full_to_delta_migration_under_cache_states(cache_state: str) -> None: + """The delta-history cache must not change full -> delta migration + semantics: with the cache disabled, cold, or already warm, the migrated + plain-value seed and the replayed delta chain must stay digest-identical to + the raw saver (differential oracle). + + Regression guard for the cache composing ``writes + seed`` on its own + (``CachedHistorySaver._aresolve``): a full-mode ``messages`` blob is stored + as a plain channel value, not a ``_DeltaSnapshot``, so it is exactly the + seed path the cache reconstructs without a saver walk.""" + inner = _CountingInMemorySaver() + saver = CachedHistorySaver( + inner, + MemoryCheckpointHistoryCache(0 if cache_state == "disabled" else 128), + key_prefix=f"itest-migrate-{uuid4().hex}", + ) + config = _config() + migrated, replayed, history = await _migrate_full_to_delta(saver, config) + + oracle_inner = _CountingInMemorySaver() + oracle_config = _config() + oracle_migrated, oracle_replayed, oracle_history = await _migrate_full_to_delta(oracle_inner, oracle_config) + + assert migrated == [("human", "kickoff", "h-0")] + assert migrated == oracle_migrated + assert replayed == oracle_replayed + assert history == oracle_history + + if cache_state == "warm": + # A second identical pass over the SAME saver must be served from the + # populated cache: same digests, no extra inner history walks. + accessor = CheckpointStateAccessor.bind(_build_noop_graph(saver, _state_schema()), saver, mode="delta") + walks_before = inner.history_walks + assert _digest((await accessor.aget(config)).values) == replayed + assert _history_digests(await accessor.ahistory(config)) == history + assert inner.history_walks == walks_before, "warm re-read triggered an inner walk" diff --git a/backend/tests/test_delta_channel_checkpointers.py b/backend/tests/test_delta_channel_checkpointers.py index 0c4ebfd6b8..c7c0106f2f 100644 --- a/backend/tests/test_delta_channel_checkpointers.py +++ b/backend/tests/test_delta_channel_checkpointers.py @@ -15,6 +15,15 @@ relies on: Every contract runs against InMemorySaver, AsyncSqliteSaver, and - when ``TEST_POSTGRES_URI`` is set - AsyncPostgresSaver, because each backend implements blob/version handling slightly differently. + +Two further tests pin currently-unfixed upstream defects (langgraph #8382 +parallel-superstep replay order, #8448 Postgres paginated delta walk). They +assert the correct contract and trip - skip, naming the issue - while the +defect is present, becoming live gates once a dependency bump lands a fix. The +#8448 trip is scoped to the Postgres parameter, the only backend that pages its +stage-1 scan: on memory/sqlite it stays a live differential assertion against +``InMemorySaver``, so a regression there fails instead of blaming the +Postgres-only upstream issue. """ from __future__ import annotations @@ -31,12 +40,13 @@ from langgraph.channels import DeltaChannel from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.serde.types import _DeltaSnapshot from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver -from langgraph.graph import StateGraph +from langgraph.graph import END, START, StateGraph from langgraph.graph.message import add_messages from langgraph.types import Overwrite from deerflow.agents.goal_state import GoalState from deerflow.agents.thread_state import DeltaThreadState, merge_message_writes +from deerflow.config.database_config import DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY from deerflow.runtime.checkpoint_mode import CHECKPOINT_MODE_METADATA_KEY, checkpoint_tuple_uses_delta from deerflow.runtime.checkpoint_state import CheckpointStateAccessor from deerflow.runtime.goal import write_thread_goal @@ -318,3 +328,133 @@ async def test_non_delta_writers_preserve_delta_messages_and_markers(saver_env: assert checkpoint_tuple_uses_delta(latest_tuple) assert latest_tuple.metadata.get(CHECKPOINT_MODE_METADATA_KEY) == "delta" assert latest_tuple.metadata.get("run_durations", {}).get("run-1") == 7 + + +# --------------------------------------------------------------------------- +# Known-unfixed upstream DeltaChannel defects. +# +# langgraph #8382 (parallel-superstep delta replay order) and #8448 (Postgres +# paginated delta-walk cursor) are both still open against the pinned +# langgraph-checkpoint 4.2.0 / langgraph-checkpoint-postgres 3.1.2, with fixes +# (PR #8544 / PR #8556) unmerged and no released version carrying them. Each +# test below asserts the CORRECT contract; while the defect is present it trips +# (skips, naming the upstream issue) instead of failing, so a later dependency +# bump that lands the fix flips it into a live gate. The contract is never +# asserted as buggy behaviour. +# --------------------------------------------------------------------------- + +_PARALLEL_LABELS = tuple("abcdefgh") +_PAGINATION_STEPS = 12 +# Shrink the Postgres stage-1 page so a dozen checkpoints cross it. The defect +# is decided by which page the target lands on, not the absolute count; upstream +# PR #8556 makes the same substitution in its own regression rather than writing +# 1024+ real checkpoints. +_PAGINATION_PAGE_SIZE = 3 + + +class LongChainState(TypedDict): + """Delta ``messages`` at the production snapshot cadence (long walks).""" + + messages: Annotated[ + list[AnyMessage], + DeltaChannel(merge_message_writes, snapshot_frequency=DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY), + ] + + +def _message_digest(values: dict[str, Any]) -> list[tuple[str, str, str | None]]: + return [(m.type, m.content, m.id) for m in values["messages"]] + + +def _append_node(index: int) -> Any: + def node(state: dict[str, Any]) -> dict[str, Any]: + return {"messages": [AIMessage(id=f"l{index}", content=f"long-{index}")]} + + return node + + +def _build_linear_graph(saver: Any, steps: int) -> Any: + builder = StateGraph(LongChainState) + for i in range(steps): + builder.add_node(f"l{i}", _append_node(i)) + builder.set_entry_point("l0") + for i in range(steps - 1): + builder.add_edge(f"l{i}", f"l{i + 1}") + builder.set_finish_point(f"l{steps - 1}") + return builder.compile(checkpointer=saver) + + +def _expected_long_chain(steps: int) -> list[tuple[str, str, str]]: + return [("human", "kickoff", "h0"), *[("ai", f"long-{i}", f"l{i}") for i in range(steps)]] + + +async def _long_chain_history(saver: Any, thread_id: str, steps: int) -> list[tuple[list, tuple]]: + """Materialize every checkpoint of a linear delta chain (newest first).""" + graph = _build_linear_graph(saver, steps) + config = _config(thread_id) + await graph.ainvoke({"messages": [HumanMessage(id="h0", content="kickoff")]}, config) + accessor = CheckpointStateAccessor.bind(graph, saver, mode="delta") + return [(_message_digest(s.values), tuple(s.next)) for s in await accessor.ahistory(config)] + + +@pytest.mark.anyio +async def test_long_chain_history_survives_pagination(saver_env: _SaverEnv, monkeypatch: pytest.MonkeyPatch) -> None: + """Every checkpoint of a long delta chain must materialize its own ancestor + prefix. Postgres resolves the history through a stage-1 scan paged + newest-first; when the target checkpoint is not on the first page the + unpatched cursor is derived from a not-yet-loaded parent and parks at + ``None`` for good, so old checkpoints hydrate empty (upstream #8448).""" + if saver_env.kind == "postgres": + pg_aio = pytest.importorskip("langgraph.checkpoint.postgres.aio", reason="postgres extra not installed") + monkeypatch.setattr(pg_aio, "_DELTA_PAGE_SIZE", _PAGINATION_PAGE_SIZE) + + got = await _long_chain_history(saver_env.saver, _thread_id(), _PAGINATION_STEPS) + oracle = await _long_chain_history(InMemorySaver(), _thread_id(), _PAGINATION_STEPS) + + # Reference sanity: newest-first, the head carries the whole chain, and + # only the pre-input root checkpoint is empty. + assert oracle[0] == (_expected_long_chain(_PAGINATION_STEPS), ()) + assert all(digest for digest, next_key in oracle if next_key != ("__start__",)) + + if saver_env.kind == "postgres" and got != oracle: + pytest.skip("upstream langgraph#8448 unfixed in langgraph-checkpoint-postgres 3.1.2: the paged delta walk poisons the channel cursor for a target past the first stage-1 page, hydrating old checkpoints empty") + assert got == oracle + + +def _fanout_node(label: str) -> Any: + def node(state: dict[str, Any]) -> dict[str, Any]: + return {"messages": [AIMessage(id=f"fan-{label}", content=label)]} + + return node + + +def _build_fanout_graph(saver: Any) -> Any: + builder = StateGraph(LongChainState) + for label in _PARALLEL_LABELS: + builder.add_node(label, _fanout_node(label)) + builder.add_edge(START, label) + builder.add_edge(label, END) + return builder.compile(checkpointer=saver) + + +@pytest.mark.anyio +async def test_parallel_superstep_replay_matches_live_write_order(saver_env: _SaverEnv) -> None: + """When several parallel tasks write the same delta channel in one + superstep, the live ``apply_writes`` order is path-sorted, but the delta + history replay sorts by the hashed ``(task_id, idx)`` - an unrelated order + (upstream #8382). ``get_state`` on the same thread must report the order the + run actually produced.""" + graph = _build_fanout_graph(saver_env.saver) + config = _config(_thread_id()) + live = await graph.ainvoke({"messages": []}, config) + live_order = [m.content for m in live["messages"]] + + # Live superstep writes are deterministic and path-sorted, so the fan-out + # lands in label order regardless of task completion order. + assert live_order == list(_PARALLEL_LABELS) + + replayed = await graph.aget_state(config) + replayed_order = [m.content for m in replayed.values["messages"]] + + if replayed_order != live_order: + pytest.skip("upstream langgraph#8382 unfixed in langgraph-checkpoint 4.2.0: delta replay orders writes by hashed (task_id, idx) instead of the path order apply_writes used live, so get_state/resume reorders same-superstep writes") + assert replayed_order == live_order From 54c65d8ef60aa732c52e5118fb6a2f3cb867b67e Mon Sep 17 00:00:00 2001 From: lihongyuan99 <64824864+lihongyuan99@users.noreply.github.com> Date: Thu, 24 Sep 2026 15:19:50 +0800 Subject: [PATCH 3/6] fix(sandbox): resolve the async acquire lock path off the event loop (#5760) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(sandbox): resolve the async acquire lock path off the event loop _discover_or_create_with_lock_async offloads every step of the acquire — ensure_thread_dirs, the lock file, cache re-checks, backend discovery and registration — except the one line that builds the lock path itself, which resolves through Paths.base_dir and therefore runs a syscall on the loop. The sibling anchor documented that as "a pre-existing blocking call in this coroutine" and stubbed the path layer to get past it. With the resolution in a worker thread the stub is no longer needed, so this removes it and pins the behaviour with an anchor of its own. * test(sandbox): word the offload anchor against the state after the stub removal * test(sandbox): pin that the gate still catches an inline thread_dir call The lock-path anchor only asserts execution reached the lock file, so it would stay green if the gate stopped instrumenting the syscalls behind Path.resolve() and the inline call were restored. Add a meta-check next to the store probe's: call thread_dir() directly on the event loop under the same isolation and assert BlockingError. --- .../aio_sandbox/aio_sandbox_provider.py | 9 ++- .../tests/blocking_io/test_aio_sandbox_get.py | 59 ++++++++++++++++--- 2 files changed, 59 insertions(+), 9 deletions(-) diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py index 9620b799b8..c343d25483 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py @@ -2175,7 +2175,14 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): paths = get_paths() effective_user_id = self._effective_acquire_user_id(user_id) await asyncio.to_thread(paths.ensure_thread_dirs, thread_id, user_id=effective_user_id) - lock_path = paths.thread_dir(thread_id, user_id=effective_user_id) / f"{sandbox_id}.lock" + + def _lock_path(): + # Worker thread: thread_dir() resolves through Paths.base_dir, which is + # a syscall — the same reason ensure_thread_dirs directly above it, and + # every later step of this coroutine, is offloaded. + return paths.thread_dir(thread_id, user_id=effective_user_id) / f"{sandbox_id}.lock" + + lock_path = await asyncio.to_thread(_lock_path) lock_file = await asyncio.to_thread(_open_lock_file, lock_path) locked = False diff --git a/backend/tests/blocking_io/test_aio_sandbox_get.py b/backend/tests/blocking_io/test_aio_sandbox_get.py index 07f2b1a9e3..40ecd126af 100644 --- a/backend/tests/blocking_io/test_aio_sandbox_get.py +++ b/backend/tests/blocking_io/test_aio_sandbox_get.py @@ -133,7 +133,6 @@ async def test_async_acquire_offloads_ownership_publish(tmp_path, monkeypatch): these two were called directly, putting a Redis round trip on the event loop for every discover/create. """ - import deerflow.community.aio_sandbox.aio_sandbox_provider as aio_mod from deerflow.community.aio_sandbox.sandbox_info import SandboxInfo provider = _make_provider(tmp_path) @@ -145,14 +144,58 @@ async def test_async_acquire_offloads_ownership_publish(tmp_path, monkeypatch): ) provider._backend.discover = MagicMock(return_value=info) - # Stub the path layer: `get_paths()` resolves the base dir via os.getcwd on - # the event loop, which is a pre-existing blocking call in this coroutine and - # not what this anchor is about. Scoping it out keeps the test pinned to the - # ownership publish this diff added. - fake_paths = MagicMock() - fake_paths.thread_dir.return_value = tmp_path - monkeypatch.setattr(aio_mod, "get_paths", lambda: fake_paths) + # The lock path is resolved in a worker thread, so the real `Paths` object can + # stay in place; DEER_FLOW_HOME keeps the thread directories and the lock file + # inside the test's own tmp_path. + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setattr("deerflow.config.paths._paths", None) sandbox_id = await provider._discover_or_create_with_lock_async("t-async", "sb-async", user_id="u1") assert sandbox_id == "sb-async" + + +async def test_async_lock_path_resolution_stays_off_the_loop(tmp_path, monkeypatch) -> None: + """The async acquire resolves its lock path from ``Paths.base_dir``, which syscalls. + + The sibling anchor used to stub the path layer out and documented why — "a + pre-existing blocking call in this coroutine". The resolution now runs in a + worker thread, so that stub is gone; this anchor keeps it from coming back. + """ + import deerflow.community.aio_sandbox.aio_sandbox_provider as aio_mod + + provider = _make_provider(tmp_path) + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setattr("deerflow.config.paths._paths", None) + + class _ReachedLockFile(Exception): + pass + + def _stop_at_lock_file(_lock_path): + raise _ReachedLockFile + + monkeypatch.setattr(aio_mod, "_open_lock_file", _stop_at_lock_file) + + # Red before the fix: the gate raises BlockingError from the inline + # ``paths.thread_dir(...)`` before the sentinel is ever reached. + with pytest.raises(_ReachedLockFile): + await provider._discover_or_create_with_lock_async("t-lock", "sb-lock", user_id="u1") + + +async def test_blocking_probe_thread_dir_actually_trips_the_gate(tmp_path, monkeypatch) -> None: + """Meta-check: prove the gate still catches an inline ``thread_dir`` call. + + The anchor above only asserts execution reached the lock file, so if the gate + went blind to the syscalls behind ``Path.resolve()`` the anchor would pass with + the inline call restored. Calling ``thread_dir`` straight from the event loop + pins that the same isolation is armed and blocking here. + """ + from blockbuster import BlockingError + + from deerflow.config.paths import get_paths + + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setattr("deerflow.config.paths._paths", None) + + with pytest.raises(BlockingError): + get_paths().thread_dir("aio-sandbox-lock-wait") From 7c948a38c4915ad3ac2b147fe4026bf2d167f3c0 Mon Sep 17 00:00:00 2001 From: Ryanba <92616678+Gujiassh@users.noreply.github.com> Date: Thu, 24 Sep 2026 15:34:32 +0800 Subject: [PATCH 4/6] fix(frontend): avoid subtask render state mutation (#3157) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(frontend): avoid subtask render state mutation * fix(frontend): avoid subtask render state mutation * Fix the lint error * 修复子任务终态回退渲染 * fix(frontend): preserve subtask loading semantics * fix(frontend): harden subtask render merging --------- Co-authored-by: Willem Jiang --- frontend/src/AGENTS.md | 1 + .../workspace/messages/message-list.tsx | 83 +++-- .../workspace/messages/subtask-card.tsx | 6 +- frontend/src/core/tasks/subtask-render.ts | 110 ++++++ .../messages/subtask-card.dom.test.tsx | 216 ++++++++++++ .../workspace/messages/subtask-card.test.ts | 104 ++++++ .../unit/core/tasks/subtask-render.test.ts | 315 ++++++++++++++++++ 7 files changed, 790 insertions(+), 45 deletions(-) create mode 100644 frontend/src/core/tasks/subtask-render.ts create mode 100644 frontend/tests/unit/components/workspace/messages/subtask-card.dom.test.tsx create mode 100644 frontend/tests/unit/components/workspace/messages/subtask-card.test.ts create mode 100644 frontend/tests/unit/core/tasks/subtask-render.test.ts diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 9a0e249f90..8f35c824cc 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -152,6 +152,7 @@ Array previews coalesce consecutive generated markers only at the end into one o - Tool-step links in `message-group.tsx` (`web_fetch` args, `web_search` / `image_search` result URLs) are model- or provider-controlled, so they pass the markdown `isSafeHref` allowlist; every surface renders a rejected href through the shared `UnsafeLink` marker. - **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1` - **Subtask step history and runtime metadata** (`core/tasks/`) — the subtask card shows a subagent's full step timeline (#3779): its assistant reasoning turns interleaved with the tools it ran. The task tool's model-visible `description` is an optional progress label; `MessageList` uses the required `prompt` (then the localized generic subtask label) when a provider omits it, so a valid task call never renders a blank card title. `Subtask.steps[]` is accumulated live from `task_running` events (appended via `mergeSteps`, not overwritten) and backfilled on expand for historical runs by `fetchSubtaskSteps`, which pages the events endpoint scoped to one task (GET `/runs/{runId}/events?event_types=subagent.step&task_id=…&after_seq=…`) until a short page, so the run-wide limit can't truncate the timeline. `task_started` carries the effective `model_name`; `task_running` carries a cumulative usage snapshot after each completed LLM call. `core/tasks/lifecycle.ts` normalizes these additive events, and `computeNextSubtask` keeps the largest cumulative total so replayed or late SSE frames cannot double-count or roll the folded card backward. Terminal ToolMessage metadata (`subagent_model_name` / `subagent_token_usage`) restores the same values from normal history after reload; no per-card event fetch is needed. `core/tasks/steps.ts` is the pure step model: `messageToStep` (live), `eventsToSteps` (reload), `mergeSteps` (dedup by `message_index`), and `stepsForDisplay` (what the card renders — keeps tool steps + AI steps with text, drops the trailing final-answer AI step when completed since it's shown as `result`). `core/tasks/context.tsx`'s `useUpdateSubtask` applies updates against a `tasksRef` mirroring the latest state (not a closure snapshot), so a late-resolving `fetchSubtaskSteps` backfill merges into current state instead of clobbering SSE steps or sibling subtasks that arrived meanwhile. The owning `run_id` is carried onto history content messages in `buildVisibleHistoryMessages` so the card can resolve the events endpoint. +- **Subtask render synchronization** (`core/tasks/subtask-render.ts`) — `MessageList` derives message-backed task snapshots in `useMemo` and publishes them to task context from an effect, never during render. Current message arguments own the card title and prompt; live task state retains lifecycle and runtime metadata until a terminal message snapshot supplies status/result/error. Cards receive the same snapshot as a render fallback so effect timing cannot leave streamed arguments stale. ### Interaction Ownership diff --git a/frontend/src/components/workspace/messages/message-list.tsx b/frontend/src/components/workspace/messages/message-list.tsx index 407a36d172..1c7fc3a38b 100644 --- a/frontend/src/components/workspace/messages/message-list.tsx +++ b/frontend/src/components/workspace/messages/message-list.tsx @@ -55,7 +55,6 @@ import { areStreamMetadataSnapshotsEqual, extractContentFromMessage, extractPresentFilesFromMessage, - extractTextFromMessage, getAssistantTurnCopyData, getBranchableAssistantGroupIds, getLatestEditableTurn, @@ -76,11 +75,7 @@ import { } from "@/core/sidecar"; import type { Subtask } from "@/core/tasks"; import { useUpdateSubtask } from "@/core/tasks/context"; -import { resolveSubtaskDescription } from "@/core/tasks/presentation"; -import { - derivePendingSubtaskStatus, - parseSubtaskResult, -} from "@/core/tasks/subtask-result"; +import { collectRenderedSubtasks } from "@/core/tasks/subtask-render"; import type { AgentThreadState } from "@/core/threads"; import { cn } from "@/lib/utils"; @@ -503,6 +498,30 @@ export function MessageList({ }, [groupedMessages]); const updateSubtask = useUpdateSubtask(); const lastGroupIndex = groupedMessages.length - 1; + const renderedSubtasks = useMemo( + () => + collectRenderedSubtasks( + groupedMessages, + (groupIndex) => thread.isLoading && groupIndex === lastGroupIndex, + t.subtasks.failed, + t.subtasks.subtask, + ), + [ + groupedMessages, + lastGroupIndex, + t.subtasks.failed, + t.subtasks.subtask, + thread.isLoading, + ], + ); + + useEffect(() => { + // Synchronize message-derived snapshots after render. The task context + // ignores identical updates, so streamed re-renders remain idempotent. + for (const task of renderedSubtasks.updates) { + updateSubtask(task); + } + }, [renderedSubtasks, updateSubtask]); const previousTurnUsageStateRef = useRef( undefined, ); @@ -1316,44 +1335,15 @@ export function MessageList({ } else if (group.type === "assistant:subagent") { const tasks = new Set(); for (const message of group.messages) { - if (message.type === "ai") { - for (const toolCall of message.tool_calls ?? []) { - if (toolCall.name === "task") { - const taskId = toolCall.id; - if (!taskId) { - continue; - } - const status = derivePendingSubtaskStatus( - taskId, - group.messages, - groupIsLoading, - ); - const task: Subtask = { - id: taskId, - subagent_type: toolCall.args.subagent_type, - description: resolveSubtaskDescription( - toolCall.args.description, - toolCall.args.prompt, - t.subtasks.subtask, - ), - prompt: toolCall.args.prompt, - status, - ...(status === "failed" - ? { error: t.subtasks.failed } - : {}), - }; - updateSubtask(task); - tasks.add(task); - } - } - } else if (message.type === "tool") { - const taskId = message.tool_call_id; - if (taskId) { - const parsed = parseSubtaskResult( - extractTextFromMessage(message), - message.additional_kwargs, - ); - updateSubtask({ id: taskId, ...parsed }); + if (message.type !== "ai") { + continue; + } + for (const toolCall of message.tool_calls ?? []) { + const task = toolCall.id + ? renderedSubtasks.tasks.get(toolCall.id) + : undefined; + if (toolCall.name === "task" && task) { + tasks.add(task); } } } @@ -1395,6 +1385,10 @@ export function MessageList({ : [], ); for (const taskId of taskIds ?? []) { + const fallbackTask = renderedSubtasks.tasks.get(taskId); + if (!fallbackTask) { + continue; + } results.push( , ); } diff --git a/frontend/src/components/workspace/messages/subtask-card.tsx b/frontend/src/components/workspace/messages/subtask-card.tsx index 9b42ffce6a..980a53383b 100644 --- a/frontend/src/components/workspace/messages/subtask-card.tsx +++ b/frontend/src/components/workspace/messages/subtask-card.tsx @@ -28,6 +28,7 @@ import { SafeStreamdown, toStreamdownComponents, } from "@/core/streamdown/components"; +import type { Subtask } from "@/core/tasks"; import { fetchSubtaskSteps } from "@/core/tasks/api"; import { useSubtask, useUpdateSubtask } from "@/core/tasks/context"; import { @@ -35,6 +36,7 @@ import { resolveSubtaskModelLabel, } from "@/core/tasks/presentation"; import { stepsForDisplay } from "@/core/tasks/steps"; +import { resolveRenderedSubtask } from "@/core/tasks/subtask-render"; import { explainLastToolCall } from "@/core/tools/utils"; import { cn } from "@/lib/utils"; @@ -49,16 +51,18 @@ export function SubtaskCard({ threadId, runId, isLoading, + fallbackTask, }: { className?: string; taskId: string; threadId?: string; runId?: string; isLoading: boolean; + fallbackTask: Subtask; }) { const { t } = useI18n(); const [collapsed, setCollapsed] = useState(true); - const task = useSubtask(taskId)!; + const task = resolveRenderedSubtask(useSubtask(taskId), fallbackTask)!; const { models, tokenUsageEnabled } = useModels(); const updateSubtask = useUpdateSubtask(); const modelLabel = resolveSubtaskModelLabel(task.modelName, models); diff --git a/frontend/src/core/tasks/subtask-render.ts b/frontend/src/core/tasks/subtask-render.ts new file mode 100644 index 0000000000..db73364636 --- /dev/null +++ b/frontend/src/core/tasks/subtask-render.ts @@ -0,0 +1,110 @@ +import { extractTextFromMessage, type MessageGroup } from "../messages/utils"; + +import { resolveSubtaskDescription } from "./presentation"; +import { + derivePendingSubtaskStatus, + parseSubtaskResult, +} from "./subtask-result"; +import { isTerminalSubtaskStatus } from "./subtask-update"; +import type { Subtask } from "./types"; + +export interface RenderedSubtasks { + tasks: Map; + updates: Array & { id: string }>; +} + +export function resolveRenderedSubtask( + liveTask: Subtask | undefined, + fallbackTask: Subtask | undefined, +): Subtask | undefined { + if (!fallbackTask) { + return liveTask; + } + + if (!liveTask) { + return fallbackTask; + } + + if (!isTerminalSubtaskStatus(fallbackTask.status)) { + return { + ...fallbackTask, + ...liveTask, + subagent_type: fallbackTask.subagent_type ?? liveTask.subagent_type, + description: fallbackTask.description ?? liveTask.description, + prompt: fallbackTask.prompt ?? liveTask.prompt, + status: liveTask.status ?? fallbackTask.status, + }; + } + + return { + ...liveTask, + ...fallbackTask, + steps: liveTask.steps ?? fallbackTask.steps, + latestMessage: liveTask.latestMessage ?? fallbackTask.latestMessage, + }; +} + +export function collectRenderedSubtasks( + groups: MessageGroup[], + isGroupLoading: (groupIndex: number) => boolean, + failedLabel: string, + subtaskLabel: string, +): RenderedSubtasks { + const tasks = new Map(); + const updates: Array & { id: string }> = []; + + for (const [groupIndex, group] of groups.entries()) { + if (group.type !== "assistant:subagent") { + continue; + } + + const groupIsLoading = isGroupLoading(groupIndex); + + for (const message of group.messages) { + if (message.type === "ai") { + for (const toolCall of message.tool_calls ?? []) { + if (toolCall.name !== "task" || !toolCall.id) { + continue; + } + const status = derivePendingSubtaskStatus( + toolCall.id, + group.messages, + groupIsLoading, + ); + const task: Subtask = { + id: toolCall.id, + subagent_type: toolCall.args.subagent_type, + description: resolveSubtaskDescription( + toolCall.args.description, + toolCall.args.prompt, + subtaskLabel, + ), + prompt: toolCall.args.prompt, + status, + ...(status === "failed" ? { error: failedLabel } : {}), + }; + tasks.set(task.id, { ...(tasks.get(task.id) ?? {}), ...task }); + updates.push(task); + } + continue; + } + + if (message.type === "tool" && message.tool_call_id) { + const update = { + id: message.tool_call_id, + ...parseSubtaskResult( + extractTextFromMessage(message), + message.additional_kwargs, + ), + }; + const existingTask = tasks.get(message.tool_call_id); + if (existingTask) { + tasks.set(message.tool_call_id, { ...existingTask, ...update }); + } + updates.push(update); + } + } + } + + return { tasks, updates }; +} diff --git a/frontend/tests/unit/components/workspace/messages/subtask-card.dom.test.tsx b/frontend/tests/unit/components/workspace/messages/subtask-card.dom.test.tsx new file mode 100644 index 0000000000..db932c90cb --- /dev/null +++ b/frontend/tests/unit/components/workspace/messages/subtask-card.dom.test.tsx @@ -0,0 +1,216 @@ +import type { Message } from "@langchain/langgraph-sdk"; +import { afterEach, describe, expect, it, rs } from "@rstest/core"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { + act, + cleanup, + fireEvent, + render, + screen, +} from "@testing-library/react"; +import { useEffect, useMemo, useState } from "react"; + +import { SubtaskCard } from "@/components/workspace/messages/subtask-card"; +import { I18nContext } from "@/core/i18n/context"; +import { enUS } from "@/core/i18n/locales/en-US"; +import { fetchSubtaskSteps } from "@/core/tasks/api"; +import { + SubtasksProvider, + useSubtaskContext, + useUpdateSubtask, +} from "@/core/tasks/context"; +import { collectRenderedSubtasks } from "@/core/tasks/subtask-render"; +import type { Subtask } from "@/core/tasks/types"; + +rs.mock("@/core/tasks/api", () => ({ fetchSubtaskSteps: rs.fn() })); + +afterEach(() => { + cleanup(); + rs.clearAllMocks(); +}); + +function setup(history = false) { + let completeArguments: () => void; + let publish: ReturnType; + let readTask: () => Subtask; + function Messages() { + const [args, setArgs] = useState({ description: "Res", prompt: "" }); + const update = useUpdateSubtask(); + const { tasksRef } = useSubtaskContext(); + const rendered = useMemo( + () => + collectRenderedSubtasks( + [ + { + id: "group-1", + type: "assistant:subagent", + messages: [ + { + type: "ai", + id: "message-1", + content: "", + tool_calls: [ + { + id: "task-1", + name: "task", + args: { subagent_type: "researcher", ...args }, + }, + ], + } as Message, + ], + }, + ], + () => true, + "Subtask failed", + "Subtask", + ), + [args], + ); + useEffect(() => { + for (const task of rendered.updates) update(task); + }, [rendered, update]); + useEffect(() => { + publish = update; + readTask = () => tasksRef.current["task-1"]!; + completeArguments = () => { + update({ id: "task-1", modelName: "test-model" }); + setArgs({ + description: "Research the topic", + prompt: "Find and compare reliable sources", + }); + }; + }); + return ( + + ); + } + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + client.setQueryData(["models"], { + models: [], + token_usage: { enabled: true }, + }); + const view = render( + + undefined, t: enUS }} + > + + + + + , + ); + return { + ...view, + complete: () => completeArguments!(), + publish: (task: Partial & { id: string }) => publish!(task), + read: () => readTask!(), + }; +} + +describe("SubtaskCard message synchronization", () => { + it("merges a delayed history backfill with live steps after argument synchronization", async () => { + let finishBackfill!: (steps: NonNullable) => void; + rs.mocked(fetchSubtaskSteps).mockImplementation( + () => + new Promise((resolve) => { + finishBackfill = resolve; + }), + ); + const view = setup(true); + fireEvent.click(screen.getByRole("button")); + expect(fetchSubtaskSteps).toHaveBeenCalledWith( + "thread-1", + "run-1", + "task-1", + ); + const live = { + kind: "tool" as const, + message_index: 2, + text: "live", + tool_name: "web_search", + }; + const historical = { + kind: "tool" as const, + message_index: 1, + text: "history", + tool_name: "read_file", + }; + await act(async () => view.publish({ id: "task-1", steps: [live] })); + await act(async () => view.complete()); + await act(async () => finishBackfill([historical])); + expect(view.read().steps).toEqual([historical, live]); + expect(view.container.textContent).toContain("read_file"); + expect(view.container.textContent).toContain("web_search"); + expect(view.container.textContent).toContain( + "Find and compare reliable sources", + ); + }); + + it("updates the folded title when final arguments and task_started arrive together", async () => { + const view = setup(); + expect(screen.getByRole("button").textContent).toContain("Res"); + await act(async () => view.complete()); + expect(view.read().description).toBe("Research the topic"); + expect(screen.getByRole("button").textContent).toContain( + "Research the topic", + ); + expect(view.container.textContent).toContain("test-model"); + }); + + it("updates an already-expanded prompt without another event or click", async () => { + const view = setup(); + fireEvent.click(screen.getByRole("button")); + await act(async () => view.complete()); + expect(view.read().prompt).toBe("Find and compare reliable sources"); + expect(view.container.textContent).toContain( + "Find and compare reliable sources", + ); + expect(screen.getByRole("button").textContent).toContain( + "Research the topic", + ); + }); + + it("preserves live steps, usage and terminal state while completing arguments", async () => { + const view = setup(); + const step = { + kind: "tool" as const, + message_index: 1, + text: "Collected sources", + tool_name: "web_search", + truncated: false, + }; + const usage = { inputTokens: 800, outputTokens: 200, totalTokens: 1000 }; + await act(async () => view.publish({ id: "task-1", steps: [step], usage })); + await act(async () => + view.publish({ + id: "task-1", + status: "completed", + result: "Finished research", + }), + ); + await act(async () => view.complete()); + expect(screen.getByRole("button").textContent).toContain( + "Research the topic", + ); + expect(view.container.textContent).toContain("Subtask completed"); + expect(view.container.textContent).toContain("1,000 Tokens"); + expect(view.container.textContent).not.toContain("Running subtask"); + fireEvent.click(screen.getByRole("button")); + expect(view.container.textContent).toContain("web_search"); + expect(view.container.textContent).toContain("Finished research"); + expect(view.read()).toMatchObject({ + status: "completed", + steps: [step], + usage, + }); + }); +}); diff --git a/frontend/tests/unit/components/workspace/messages/subtask-card.test.ts b/frontend/tests/unit/components/workspace/messages/subtask-card.test.ts new file mode 100644 index 0000000000..fc07473518 --- /dev/null +++ b/frontend/tests/unit/components/workspace/messages/subtask-card.test.ts @@ -0,0 +1,104 @@ +import { describe, expect, it } from "@rstest/core"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { createElement } from "react"; +import { renderToStaticMarkup } from "react-dom/server"; + +import { SubtaskCard } from "@/components/workspace/messages/subtask-card"; +import { I18nContext } from "@/core/i18n/context"; +import { enUS } from "@/core/i18n/locales/en-US"; +import { SubtaskContext } from "@/core/tasks/context"; +import type { Subtask } from "@/core/tasks/types"; + +function baseTask(overrides: Partial = {}): Subtask { + return { + id: "task-1", + status: "in_progress", + subagent_type: "researcher", + description: "Research the topic", + prompt: "Find sources", + ...overrides, + }; +} + +describe("SubtaskCard", () => { + it("renders a failed fallback snapshot ahead of stale in-progress context state", () => { + const html = renderCard({ + contextTask: baseTask({ status: "in_progress" }), + fallbackTask: baseTask({ + status: "failed", + error: "Subtask failed", + }), + }); + + expect(html).toContain("Subtask failed"); + expect(html).not.toContain("Running subtask"); + }); + + it("renders a completed fallback snapshot ahead of stale in-progress context state", () => { + const html = renderCard({ + contextTask: baseTask({ + status: "in_progress", + steps: [ + { + kind: "tool", + message_index: 1, + text: "Collected sources", + truncated: false, + }, + ], + }), + fallbackTask: baseTask({ + status: "completed", + result: "done", + }), + }); + + expect(html).toContain("Subtask completed"); + expect(html).not.toContain("Running subtask"); + }); +}); + +function renderCard({ + contextTask, + fallbackTask, +}: { + contextTask?: Subtask; + fallbackTask: Subtask; +}) { + const tasks = contextTask ? { [contextTask.id]: contextTask } : {}; + const queryClient = new QueryClient(); + + return renderToStaticMarkup( + createElement( + QueryClientProvider, + { + client: queryClient, + }, + createElement( + I18nContext.Provider, + { + value: { + locale: "en-US", + setLocale: () => undefined, + t: enUS, + }, + }, + createElement( + SubtaskContext.Provider, + { + value: { + tasks, + tasksRef: { current: tasks }, + setTasks: () => undefined, + }, + }, + createElement(SubtaskCard, { + taskId: fallbackTask.id, + isLoading: false, + fallbackTask, + }), + ), + ), + ), + ); +} diff --git a/frontend/tests/unit/core/tasks/subtask-render.test.ts b/frontend/tests/unit/core/tasks/subtask-render.test.ts new file mode 100644 index 0000000000..a138cf1170 --- /dev/null +++ b/frontend/tests/unit/core/tasks/subtask-render.test.ts @@ -0,0 +1,315 @@ +import type { Message } from "@langchain/langgraph-sdk"; +import { describe, expect, it } from "@rstest/core"; + +import type { MessageGroup } from "@/core/messages/utils"; +import { + collectRenderedSubtasks, + resolveRenderedSubtask, +} from "@/core/tasks/subtask-render"; +import type { Subtask } from "@/core/tasks/types"; + +function baseTask(overrides: Partial = {}): Subtask { + return { + id: "task-1", + status: "in_progress", + subagent_type: "researcher", + description: "Research the topic", + prompt: "Find sources", + ...overrides, + }; +} + +function subagentGroup(messages: Message[]) { + return { + id: "group-1", + type: "assistant:subagent", + messages, + } as const; +} + +describe("collectRenderedSubtasks", () => { + it("derives an in-progress task while the subagent turn is still streaming", () => { + const groups: MessageGroup[] = [ + { + id: "group-0", + type: "human", + messages: [], + }, + subagentGroup([ + { + type: "ai", + tool_calls: [ + { + id: "task-1", + name: "task", + args: { + subagent_type: "researcher", + description: "Research the topic", + prompt: "Find sources", + }, + }, + ], + content: "", + } as unknown as Message, + ]), + ]; + + const threadIsLoading = true; + const rendered = collectRenderedSubtasks( + groups, + (groupIndex) => threadIsLoading && groupIndex === groups.length - 1, + "failed", + "Subtask", + ); + + expect(rendered.tasks.get("task-1")).toMatchObject({ + id: "task-1", + status: "in_progress", + subagent_type: "researcher", + }); + expect(rendered.updates).toHaveLength(1); + }); + + it("marks a task as failed once its turn ended without a tool result", () => { + const groups = [ + subagentGroup([ + { + type: "ai", + tool_calls: [ + { + id: "task-1", + name: "task", + args: { + subagent_type: "researcher", + description: "Research the topic", + prompt: "Find sources", + }, + }, + ], + content: "", + } as unknown as Message, + ]), + ]; + + const rendered = collectRenderedSubtasks( + groups, + () => false, + "Subtask failed", + "Subtask", + ); + + expect(rendered.tasks.get("task-1")).toMatchObject({ + id: "task-1", + status: "failed", + error: "Subtask failed", + }); + }); + + it("merges a tool result into the fallback task snapshot", () => { + const groups = [ + subagentGroup([ + { + type: "ai", + tool_calls: [ + { + id: "task-1", + name: "task", + args: { + subagent_type: "researcher", + description: "Research the topic", + prompt: "Find sources", + }, + }, + ], + content: "", + } as unknown as Message, + { + type: "tool", + tool_call_id: "task-1", + content: "ignored by structured status", + additional_kwargs: { + subagent_status: "completed", + subagent_result_brief: "done", + }, + } as unknown as Message, + ]), + ]; + + const rendered = collectRenderedSubtasks( + groups, + () => false, + "failed", + "Subtask", + ); + + expect(rendered.tasks.get("task-1")).toMatchObject({ + id: "task-1", + status: "completed", + result: "done", + }); + expect(rendered.updates).toHaveLength(2); + }); +}); + +describe("collectRenderedSubtasks description fallback", () => { + it.each([ + [undefined, "Use the prompt", "Use the prompt"], + [" ", " Trimmed prompt ", "Trimmed prompt"], + [undefined, "", "Localized subtask"], + ])( + "preserves the current-main description fallback for %s / %s", + (description, prompt, expected) => { + const rendered = collectRenderedSubtasks( + [ + subagentGroup([ + { + type: "ai", + content: "", + tool_calls: [ + { + id: "task-1", + name: "task", + args: { description, prompt, subagent_type: "researcher" }, + }, + ], + } as Message, + ]), + ], + () => true, + "failed", + "Localized subtask", + ); + expect(rendered.tasks.get("task-1")?.description).toBe(expected); + }, + ); +}); + +describe("resolveRenderedSubtask", () => { + it("keeps live arguments when a partial snapshot omits them", () => { + const live = baseTask(); + const fallback = { id: "task-1", status: "in_progress" } as Subtask; + expect(resolveRenderedSubtask(live, fallback)).toEqual(live); + }); + + it.each(["completed", "failed"] as const)( + "preserves %s lifecycle and runtime data while refreshing message arguments", + (status) => { + const live = baseTask({ + status, + result: "done", + error: "failed", + stopReason: "token_capped", + modelName: "test-model", + usage: { inputTokens: 8, outputTokens: 2, totalTokens: 10 }, + steps: [{ kind: "tool", message_index: 1, text: "done" }], + latestMessage: { id: "live", type: "ai", content: "working" }, + }); + const resolved = resolveRenderedSubtask( + live, + baseTask({ + description: "Current title", + prompt: "Current prompt", + subagent_type: "general-purpose", + }), + ); + expect(resolved).toEqual({ + ...live, + description: "Current title", + prompt: "Current prompt", + subagent_type: "general-purpose", + }); + }, + ); + + it("uses an in-progress fallback before the live task is available", () => { + const fallbackTask = baseTask({ + status: "in_progress", + description: "fallback", + }); + + expect(resolveRenderedSubtask(undefined, fallbackTask)).toBe(fallbackTask); + }); + + it("prefers a terminal fallback snapshot over a stale live task", () => { + const resolved = resolveRenderedSubtask( + baseTask({ + status: "in_progress", + modelName: "claude-3-7-sonnet", + usage: { + inputTokens: 800, + outputTokens: 200, + totalTokens: 1_000, + }, + latestMessage: { id: "live-1" } as Subtask["latestMessage"], + }), + baseTask({ + status: "failed", + error: "Subtask failed", + }), + ); + + expect(resolved).toMatchObject({ + status: "failed", + error: "Subtask failed", + modelName: "claude-3-7-sonnet", + usage: { + inputTokens: 800, + outputTokens: 200, + totalTokens: 1_000, + }, + latestMessage: { id: "live-1" }, + }); + }); + + it("uses current message arguments while preserving live runtime state", () => { + const resolved = resolveRenderedSubtask( + baseTask({ + status: "in_progress", + description: "live", + modelName: "test-model", + prompt: "old prompt", + }), + baseTask({ status: "in_progress", description: "fallback", prompt: "" }), + ); + + expect(resolved).toMatchObject({ + status: "in_progress", + description: "fallback", + prompt: "", + modelName: "test-model", + }); + }); + + it("fills a metadata-only live task from an in-progress fallback", () => { + const resolved = resolveRenderedSubtask( + { + id: "task-1", + modelName: "claude-3-7-sonnet", + } as Subtask, + baseTask({ status: "in_progress", description: "fallback" }), + ); + + expect(resolved).toMatchObject({ + id: "task-1", + status: "in_progress", + description: "fallback", + modelName: "claude-3-7-sonnet", + }); + }); + + it("preserves fields available only on a live task after completion", () => { + const resolved = resolveRenderedSubtask( + { + ...baseTask({ status: "in_progress" }), + futureLiveMetadata: "preserved", + } as Subtask & { futureLiveMetadata: string }, + baseTask({ status: "completed", result: "done" }), + ) as Subtask & { futureLiveMetadata?: string }; + + expect(resolved).toMatchObject({ + status: "completed", + result: "done", + futureLiveMetadata: "preserved", + }); + }); +}); From c77e8744fabb061b5fd4e3a0bd7897a06ea2138b Mon Sep 17 00:00:00 2001 From: "Grapette.L" <90967079+Lesereingrape@users.noreply.github.com> Date: Thu, 24 Sep 2026 15:38:52 +0800 Subject: [PATCH 5/6] fix(memory): keep a zero confidence from scoring as the default (#5586) or 0.5 cannot distinguish an explicit 0.0 from an unset field, so a discounted fact got the default confidence weight in both the FTS5 document and the substring fallback, while the same row indexed through FTS5Retrieval.index_fact kept its zero weight. --- .../deermem/deermem/core/retrieval.py | 4 ++- .../backends/deermem/deermem/core/storage.py | 5 ++- .../tests/test_memory_retrieval_adapter.py | 31 +++++++++++++++++++ 3 files changed, 38 insertions(+), 2 deletions(-) diff --git a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/retrieval.py b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/retrieval.py index dca66aa7d3..a619f8e0d0 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/retrieval.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/retrieval.py @@ -578,11 +578,13 @@ class FTS5RetrievalAdapter: payload = dict(fact) payload["scope"] = {"userId": scope.get("userId"), "agentName": scope.get("agentName")} source = payload.get("source") + confidence = payload.get("confidence") return { "fact_id": self._document_id(fact_id, scope), "content": content, "category": str(payload.get("category") or "context"), - "confidence": float(payload.get("confidence") or 0.5), + # 0.0 is a persisted confidence storage._normalize_fact accepts; only unset/null defaults. + "confidence": 0.5 if confidence is None else float(confidence), "created_at": payload.get("createdAt") if isinstance(payload.get("createdAt"), str) else None, "scope_user": scope_user, "scope_agent": scope_agent, diff --git a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/storage.py b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/storage.py index f79233fbb8..490beb971c 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/storage.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/storage.py @@ -1822,7 +1822,10 @@ class FileMemoryStorage(MemoryStorage): for fact in facts: content = fact.get("content") if isinstance(content, str) and query_lower in content.lower(): - results.append({"fact": fact, "score": float(fact.get("confidence") or 0.5), "matchType": "substring"}) + # 0.0 is a persisted confidence _normalize_fact accepts; only unset/null defaults. + confidence = fact.get("confidence") + score = 0.5 if confidence is None else float(confidence) + results.append({"fact": fact, "score": score, "matchType": "substring"}) results.sort(key=lambda result: result["score"], reverse=True) return results[:top_k] diff --git a/backend/tests/test_memory_retrieval_adapter.py b/backend/tests/test_memory_retrieval_adapter.py index bef09cf50a..b47cb0dea9 100644 --- a/backend/tests/test_memory_retrieval_adapter.py +++ b/backend/tests/test_memory_retrieval_adapter.py @@ -287,3 +287,34 @@ def test_deermem_create_and_restart_use_retrieval_adapter(tmp_path: Path) -> Non restarted = DeerMem(backend_config=config) assert any(fact["id"] == fact_id for fact in restarted.search("restart retrieval", user_id="alice")) + + +def test_indexed_zero_confidence_keeps_its_retrieval_weight(tmp_path: Path) -> None: + adapter = FTS5RetrievalAdapter(tmp_path / "facts.sqlite3") + scope = {"userId": "alice", "agentName": "agent-a"} + try: + adapter.upsert({**_fact("zero", "confidence weight probe"), "confidence": 0.0}, scope=scope, path="") + adapter.upsert({**_fact("half", "confidence weight probe"), "confidence": 0.5}, scope=scope, path="") + unset_fact = _fact("unset", "confidence weight probe") + del unset_fact["confidence"] + adapter.upsert(unset_fact, scope=scope, path="") + scores = {item["fact"]["id"]: item["score"] for item in adapter.search("confidence", scopes=[scope], top_k=5, mode="fts5", filters=None)} + + assert scores["zero"] < scores["half"] + # A missing confidence still falls back to the default, so only the explicit 0.0 is discounted. + assert scores["unset"] == pytest.approx(scores["half"]) + finally: + adapter.close() + + +def test_substring_fallback_keeps_zero_confidence_weight(tmp_path: Path) -> None: + storage = FileMemoryStorage(DeerMemConfig(storage_path=str(tmp_path))) + scope = {"userId": "alice", "agentName": "agent-a"} + try: + storage.upsert_fact({**_fact("zero", "substring confidence probe"), "confidence": 0.0}, user_id="alice", agent_name="agent-a") + storage.upsert_fact({**_fact("half", "substring confidence probe"), "confidence": 0.5}, user_id="alice", agent_name="agent-a") + scores = {item["fact"]["id"]: item["score"] for item in storage.search_facts("substring", scopes=[scope], top_k=5)} + + assert scores == {"zero": 0.0, "half": 0.5} + finally: + storage.close() From 7833801690905e924d93326ee2423b9a94281ed7 Mon Sep 17 00:00:00 2001 From: "Grapette.L" <90967079+Lesereingrape@users.noreply.github.com> Date: Thu, 24 Sep 2026 15:47:33 +0800 Subject: [PATCH 6/6] fix(skills): load SKILL.md saved as UTF-8 with a BOM (#5588) * fix(skills): load SKILL.md saved as UTF-8 with a BOM The front-matter anchor began at ^---, so the U+FEFF that Windows Notepad and PowerShell's `Set-Content -Encoding UTF8` prepend made the match fail. parse_skill_file then returned None and the catalog builder dropped the skill without logging, leaving an author with a valid SKILL.md that simply never appears. split_skill_markdown reported "No YAML frontmatter found" for the same file at install time. Let the anchor consume one leading byte-order mark, and reuse the shared compiled pattern in parser.py instead of the duplicated inline copy so the two cannot drift apart again. * fix(skills): capture the description of a BOM-prefixed SKILL.md The reader in skill_context anchored on ^---, so the byte-order mark that this change already tolerates in the loader and validator made the match fail and build_skill_entry_metadata_from_read capture an empty description. render_skill_context then listed the skill without the one line the active-skill reminder shows for it. Consume one leading byte-order mark there and pin the capture with a regression test. --- .../agents/middlewares/skill_context.py | 5 ++- .../harness/deerflow/skills/frontmatter.py | 5 ++- .../harness/deerflow/skills/parser.py | 3 +- backend/tests/test_skill_context.py | 11 +++++ backend/tests/test_skills_parser.py | 38 +++++++++++++++- backend/tests/test_skills_validation.py | 45 ++++++++++++++++++- 6 files changed, 100 insertions(+), 7 deletions(-) diff --git a/backend/packages/harness/deerflow/agents/middlewares/skill_context.py b/backend/packages/harness/deerflow/agents/middlewares/skill_context.py index 75163148e4..414e547cf2 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/skill_context.py +++ b/backend/packages/harness/deerflow/agents/middlewares/skill_context.py @@ -15,7 +15,10 @@ from langchain_core.messages import AIMessage, AnyMessage, ToolMessage from deerflow.agents.thread_state import _SKILL_DESCRIPTION_MAX_CHARS, SkillEntry _SKILL_FILE_NAME = "SKILL.md" -_FRONT_MATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL) +# `\ufeff` is the UTF-8 byte-order mark a Windows-saved SKILL.md can carry. The loader and +# validator already tolerate it (see `deerflow.skills.frontmatter._FRONTMATTER_RE`); without it +# here the match fails on the leading marker and the captured description comes back empty. +_FRONT_MATTER_RE = re.compile(r"^\ufeff?---\s*\n(.*?)\n---\s*\n", re.DOTALL) SKILL_CONTEXT_ENTRY_KEY = "skill_context_entry" logger = logging.getLogger(__name__) diff --git a/backend/packages/harness/deerflow/skills/frontmatter.py b/backend/packages/harness/deerflow/skills/frontmatter.py index 4c5f96dacd..b1187a2616 100644 --- a/backend/packages/harness/deerflow/skills/frontmatter.py +++ b/backend/packages/harness/deerflow/skills/frontmatter.py @@ -26,7 +26,10 @@ ALLOWED_FRONTMATTER_PROPERTIES = { "author", } -_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n?", re.DOTALL) +# `\ufeff` is the UTF-8 byte-order mark that Windows Notepad and PowerShell's +# `Set-Content -Encoding UTF8` prepend to a saved file. It is consumed by the match, so +# neither the front matter nor the body carries it downstream. +_FRONTMATTER_RE = re.compile(r"^\ufeff?---\s*\n(.*?)\n---\s*\n?", re.DOTALL) @dataclass(frozen=True) diff --git a/backend/packages/harness/deerflow/skills/parser.py b/backend/packages/harness/deerflow/skills/parser.py index 7c2aa58048..cc2d71fe4c 100644 --- a/backend/packages/harness/deerflow/skills/parser.py +++ b/backend/packages/harness/deerflow/skills/parser.py @@ -4,6 +4,7 @@ from pathlib import Path import yaml +from .frontmatter import _FRONTMATTER_RE from .types import SKILL_MD_FILE, SecretRequirement, Skill, SkillCategory logger = logging.getLogger(__name__) @@ -217,7 +218,7 @@ def parse_skill_file(skill_file: Path, category: SkillCategory, relative_path: P # Keep parser diagnostics richer than the pure helper's host-path-free # error string; tests and authoring UX depend on the line-specific hint. - front_matter_match = re.match(r"^---\s*\n(.*?)\n---\s*\n?", content, re.DOTALL) + front_matter_match = _FRONTMATTER_RE.match(content) if not front_matter_match: return None front_matter_text = front_matter_match.group(1) diff --git a/backend/tests/test_skill_context.py b/backend/tests/test_skill_context.py index 2564547b89..c4e4c60d09 100644 --- a/backend/tests/test_skill_context.py +++ b/backend/tests/test_skill_context.py @@ -57,6 +57,17 @@ class TestExtractSkills: } assert "ALWAYS_USE_PANDAS_SENTINEL" not in repr(entry) + def test_build_skill_entry_metadata_from_read_captures_description_behind_a_bom(self): + entry = build_skill_entry_metadata_from_read( + "/mnt/skills/public/data-analysis/SKILL.md", + f"\ufeff{_SKILL_BODY}", + skills_root=_ROOT, + ) + assert entry == { + "path": "/mnt/skills/public/data-analysis/SKILL.md", + "description": "Analyze data with pandas and charts.", + } + def test_captures_skill_reference_with_description(self): msgs = [ HumanMessage(content="use the analysis skill"), diff --git a/backend/tests/test_skills_parser.py b/backend/tests/test_skills_parser.py index 90f6dbc062..c128a77ad6 100644 --- a/backend/tests/test_skills_parser.py +++ b/backend/tests/test_skills_parser.py @@ -22,12 +22,12 @@ from deerflow.skills.parser import parse_skill_file # --------------------------------------------------------------------------- -def _write_skill(tmp_path: Path, front_matter: str, body: str = "# My Skill\n") -> Path: +def _write_skill(tmp_path: Path, front_matter: str, body: str = "# My Skill\n", encoding: str = "utf-8") -> Path: """Write a minimal SKILL.md and return the path.""" skill_dir = tmp_path / "my-skill" skill_dir.mkdir() skill_file = skill_dir / "SKILL.md" - skill_file.write_text(f"---\n{front_matter}\n---\n{body}", encoding="utf-8") + skill_file.write_text(f"---\n{front_matter}\n---\n{body}", encoding=encoding) return skill_file @@ -417,3 +417,37 @@ def test_parse_unquoted_colon_value_escapes_regex_in_hint(tmp_path, caplog): assert skill is None combined = "\n".join(rec.getMessage() for rec in caplog.records) assert r'description: "match: \\d+ digits"' in combined + + +# --------------------------------------------------------------------------- +# UTF-8 byte-order mark +# --------------------------------------------------------------------------- + + +def test_parse_skill_file_saved_with_utf8_bom(tmp_path): + """A SKILL.md saved as "UTF-8 with BOM" is a skill, not a broken file. + + Windows Notepad and PowerShell's ``Set-Content -Encoding UTF8`` prepend U+FEFF to + every file they save. The front-matter anchor began at ``^---``, so that mark made + ``parse_skill_file`` return ``None`` and the catalog builder dropped the skill with + no log line and no error - the skill simply never existed as far as DeerFlow was + concerned, while the exact same file edited on Linux loaded fine. + """ + skill_file = _write_skill(tmp_path, "name: my-skill\ndescription: A test skill", encoding="utf-8-sig") + assert skill_file.read_bytes().startswith(b"\xef\xbb\xbf"), "fixture must carry a real BOM" + + skill = parse_skill_file(skill_file, category="custom") + assert skill is not None + assert skill.name == "my-skill" + assert skill.description == "A test skill" + + +def test_parse_skill_file_without_bom_is_unaffected(tmp_path): + """Control: the same document written as plain UTF-8 parses exactly as before.""" + skill_file = _write_skill(tmp_path, "name: my-skill\ndescription: A test skill", encoding="utf-8") + assert not skill_file.read_bytes().startswith(b"\xef\xbb\xbf") + + skill = parse_skill_file(skill_file, category="custom") + assert skill is not None + assert skill.name == "my-skill" + assert skill.description == "A test skill" diff --git a/backend/tests/test_skills_validation.py b/backend/tests/test_skills_validation.py index 070adbcaf9..a759f0e427 100644 --- a/backend/tests/test_skills_validation.py +++ b/backend/tests/test_skills_validation.py @@ -8,13 +8,14 @@ from pathlib import Path import pytest +from deerflow.skills.frontmatter import split_skill_markdown from deerflow.skills.validation import ALLOWED_FRONTMATTER_PROPERTIES, _validate_skill_frontmatter -def _write_skill(tmp_path: Path, content: str) -> Path: +def _write_skill(tmp_path: Path, content: str, encoding: str = "utf-8") -> Path: """Write a SKILL.md file and return its parent directory.""" skill_file = tmp_path / "SKILL.md" - skill_file.write_text(content, encoding="utf-8") + skill_file.write_text(content, encoding=encoding) return tmp_path @@ -296,3 +297,43 @@ class TestValidateSkillFrontmatter: assert valid is True assert msg == "Skill is valid!" assert name == "demo-skill" + + def test_valid_when_saved_with_utf8_bom(self, tmp_path): + """A SKILL.md saved as "UTF-8 with BOM" installs like any other file. + + Windows Notepad and PowerShell's ``Set-Content -Encoding UTF8`` prepend U+FEFF, + which used to fall outside the ``^---`` anchor and report "No YAML frontmatter + found" for a file that is byte-for-byte a valid skill. + """ + skill_dir = _write_skill( + tmp_path, + "---\nname: my-skill\ndescription: A valid skill\n---\n\nBody\n", + encoding="utf-8-sig", + ) + assert (skill_dir / "SKILL.md").read_bytes().startswith(b"\xef\xbb\xbf"), "fixture must carry a real BOM" + + valid, msg, name = _validate_skill_frontmatter(skill_dir) + assert valid is True + assert msg == "Skill is valid!" + assert name == "my-skill" + + +class TestSplitSkillMarkdownBom: + def test_bom_is_consumed_and_reaches_neither_metadata_nor_body(self): + """The mark is swallowed by the anchor instead of leaking into the parsed parts.""" + parts, error = split_skill_markdown("\ufeff---\nname: my-skill\ndescription: A valid skill\n---\nBody\n") + + assert error is None + assert parts is not None + assert parts.metadata["name"] == "my-skill" + assert parts.frontmatter_text == "name: my-skill\ndescription: A valid skill" + assert parts.body == "Body\n" + + def test_control_document_without_bom_is_parsed_identically(self): + parts, error = split_skill_markdown("---\nname: my-skill\ndescription: A valid skill\n---\nBody\n") + + assert error is None + assert parts is not None + assert parts.metadata["name"] == "my-skill" + assert parts.frontmatter_text == "name: my-skill\ndescription: A valid skill" + assert parts.body == "Body\n"