mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-01 08:00:40 +00:00
Merge branch 'main' into codex/ark-evolving-default
This commit is contained in:
laundry
GitHub
commit
0bae3b1d36
@ -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,
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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]
|
||||
|
||||
|
||||
@ -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__)
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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")
|
||||
|
||||
@ -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"
|
||||
|
||||
35
backend/tests/test_channel_runtime_config_encoding.py
Normal file
35
backend/tests/test_channel_runtime_config_encoding.py
Normal file
@ -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
|
||||
@ -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
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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"),
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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"
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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<AssistantTurnUsageState | undefined>(
|
||||
undefined,
|
||||
);
|
||||
@ -1316,44 +1335,15 @@ export function MessageList({
|
||||
} else if (group.type === "assistant:subagent") {
|
||||
const tasks = new Set<Subtask>();
|
||||
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(
|
||||
<SubtaskCard
|
||||
key={"task-group-" + taskId}
|
||||
@ -1402,6 +1396,7 @@ export function MessageList({
|
||||
threadId={threadId}
|
||||
runId={(message as { run_id?: string }).run_id}
|
||||
isLoading={groupIsLoading}
|
||||
fallbackTask={fallbackTask}
|
||||
/>,
|
||||
);
|
||||
}
|
||||
|
||||
@ -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);
|
||||
|
||||
110
frontend/src/core/tasks/subtask-render.ts
Normal file
110
frontend/src/core/tasks/subtask-render.ts
Normal file
@ -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<string, Subtask>;
|
||||
updates: Array<Partial<Subtask> & { 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<string, Subtask>();
|
||||
const updates: Array<Partial<Subtask> & { 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 };
|
||||
}
|
||||
@ -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<typeof useUpdateSubtask>;
|
||||
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 (
|
||||
<SubtaskCard
|
||||
taskId="task-1"
|
||||
threadId={history ? "thread-1" : undefined}
|
||||
runId={history ? "run-1" : undefined}
|
||||
isLoading
|
||||
fallbackTask={rendered.tasks.get("task-1")!}
|
||||
/>
|
||||
);
|
||||
}
|
||||
const client = new QueryClient({
|
||||
defaultOptions: { queries: { retry: false } },
|
||||
});
|
||||
client.setQueryData(["models"], {
|
||||
models: [],
|
||||
token_usage: { enabled: true },
|
||||
});
|
||||
const view = render(
|
||||
<QueryClientProvider client={client}>
|
||||
<I18nContext.Provider
|
||||
value={{ locale: "en-US", setLocale: () => undefined, t: enUS }}
|
||||
>
|
||||
<SubtasksProvider>
|
||||
<Messages />
|
||||
</SubtasksProvider>
|
||||
</I18nContext.Provider>
|
||||
</QueryClientProvider>,
|
||||
);
|
||||
return {
|
||||
...view,
|
||||
complete: () => completeArguments!(),
|
||||
publish: (task: Partial<Subtask> & { 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<Subtask["steps"]>) => 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,
|
||||
});
|
||||
});
|
||||
});
|
||||
@ -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> = {}): 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,
|
||||
}),
|
||||
),
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
315
frontend/tests/unit/core/tasks/subtask-render.test.ts
Normal file
315
frontend/tests/unit/core/tasks/subtask-render.test.ts
Normal file
@ -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> = {}): 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",
|
||||
});
|
||||
});
|
||||
});
|
||||
Loading…
x
Reference in New Issue
Block a user