From 28553040feffd645b8a99393f5fa0f54f73f01cb Mon Sep 17 00:00:00 2001 From: ajayr Date: Tue, 28 Jul 2026 00:44:59 +0100 Subject: [PATCH 1/4] docs(config): Crawl4AI >= 0.9 requires a bearer token (#4518) The commented Crawl4AI web_fetch example still describes the pre-0.9 server: it says JWT auth is off by default, marks `token:` as only needed "if the server has JWT auth enabled", and pins 0.8.6 in the docker run line. Crawl4AI 0.9.0 made the Docker API server secure-by-default. Auth is on for every request except GET /health, and a server started without CRAWL4AI_API_TOKEN binds 127.0.0.1 only. Following the current example against a current image therefore yields either HTTP 401 on every fetch, or a server a containerised DeerFlow cannot reach -- with no hint that auth is the cause. 0.8.6 is also worth moving off: 0.8.7 fixed two pre-auth RCEs (CVSS 9.8), and 0.8.8/0.8.9 closed SSRF gaps in the same server. No code change is needed -- Crawl4AiClient already sends `Authorization: Bearer ` whenever `token` is set, so this is purely the example catching up with the upstream server. Comment-only, so config_version is unchanged. Co-authored-by: Claude Opus 5 --- config.example.yaml | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/config.example.yaml b/config.example.yaml index 81ef4f7fb..a7ab5a555 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -763,10 +763,16 @@ tools: # # wait_for_timeout_ms: 2000 # Extra wait after page load in milliseconds # # wait_for_selector: "article" # CSS selector to wait for before returning - # Web fetch tool (uses Crawl4AI - self-hosted headless Chromium, no API key) + # Web fetch tool (uses Crawl4AI - self-hosted headless Chromium, no third-party API key) # Crawl4AI returns server-cleaned "fit" markdown directly (no readability step needed), - # ideal for JavaScript-heavy sites. Self-host (JWT auth is off by default, no key): - # docker run -d -p 11235:11235 --shm-size=1g unclecode/crawl4ai:0.8.6 + # ideal for JavaScript-heavy sites. Self-host it: + # docker run -d -p 11235:11235 --shm-size=1g \ + # -e CRAWL4AI_API_TOKEN=$CRAWL4AI_TOKEN unclecode/crawl4ai:0.9.2 + # Crawl4AI >= 0.9 is secure-by-default: a bearer token is REQUIRED on every request + # except GET /health, and a server started without CRAWL4AI_API_TOKEN binds 127.0.0.1 + # only (so a container/remote DeerFlow cannot reach it at all). Set the same value in + # the server env and in `token:` below, or requests fail with HTTP 401. + # Use >= 0.8.7: earlier images, including 0.8.6, carry known pre-auth RCEs. # For Docker deployments, use the Docker service name instead of localhost. # NOTE: Only one web_fetch provider can be active at a time. # Comment out the Jina AI web_fetch entry below before enabling this one. @@ -774,10 +780,10 @@ tools: # group: web # use: deerflow.community.crawl4ai.tools:web_fetch_tool # base_url: http://localhost:11235 # Crawl4AI server URL (Docker: http://crawl4ai:11235) + # token: $CRAWL4AI_TOKEN # Bearer token; required by Crawl4AI >= 0.9 # timeout: 30 # Request timeout in seconds # # allow_private_addresses: false # SSRF guard: keep false in production. Set true ONLY for intentional internal targets. # # filter: fit # Markdown filter: fit (default) | raw | bm25 | llm - # # token: $CRAWL4AI_TOKEN # Bearer token (only if the server has JWT auth enabled) # Web capture tool (uses Browserless /screenshot to render a page as an artifact) # Browserless captures JavaScript-heavy pages with a real headless Chrome and From d455a1815e1af71fe94fa98f8b755029208606e3 Mon Sep 17 00:00:00 2001 From: Aari Date: Tue, 28 Jul 2026 07:49:13 +0800 Subject: [PATCH 2/4] fix(sandbox): allow grep to search a single file (#4512) --- README.md | 2 + backend/AGENTS.md | 4 +- .../community/aio_sandbox/aio_sandbox.py | 62 +++++----- .../harness/deerflow/community/boxlite/box.py | 10 +- .../community/e2b_sandbox/e2b_sandbox.py | 4 +- .../deerflow/community/tenki/sandbox.py | 4 +- .../harness/deerflow/sandbox/sandbox.py | 2 +- .../harness/deerflow/sandbox/search.py | 76 +++++++------ .../harness/deerflow/sandbox/tools.py | 4 +- backend/packages/harness/pyproject.toml | 2 +- backend/tests/test_boxlite_provider.py | 11 ++ backend/tests/test_e2b_sandbox_provider.py | 12 ++ backend/tests/test_sandbox_search_tools.py | 107 +++++++++++++----- backend/tests/test_tenki_provider.py | 10 ++ backend/uv.lock | 2 +- 15 files changed, 202 insertions(+), 110 deletions(-) diff --git a/README.md b/README.md index e7d97c4e0..7d257af76 100644 --- a/README.md +++ b/README.md @@ -893,6 +893,8 @@ DeerFlow doesn't just *talk* about doing things. It has its own computer. Each task gets its own execution environment with a full filesystem view — skills, workspace, uploads, outputs. The agent reads, writes, and edits files. It can view images and, when configured safely, execute shell commands. +The built-in `grep` tool searches either one text file or all matching text files below a directory, so an agent can search an uploaded document directly without first broadening the request to the entire uploads directory. + Image bytes loaded for a vision-model call are transient: DeerFlow removes the hidden base64 message after the model consumes it so later checkpoints do not keep duplicating that payload. After each run, DeerFlow records a workspace change summary for the run-owned `workspace` and `outputs` directories. The Web UI shows a compact "files changed" badge on the assistant turn; opening it reveals created, modified, and deleted files with text diffs when safe to display. Uploads are excluded because they are user inputs, not agent-generated changes. Large, binary, or sensitive-looking files are shown as metadata only. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 3747c920e..2f1fd8ec0 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -536,7 +536,7 @@ copying a raw checkpoint because delta state is not self-contained in one tuple. ### Sandbox System (`packages/harness/deerflow/sandbox/`) -**Interface**: Abstract `Sandbox` with `execute_command(command, env=None)`, `read_file`, `write_file`, `list_dir`. The optional `env` injects per-call environment variables (request-scoped secrets — see Request-Scoped Secrets below); `LocalSandbox` merges it via `subprocess.run(env=...)` and `AioSandbox` routes env-bearing commands through the `bash.exec(env=...)` API on a fresh session. +**Interface**: Abstract `Sandbox` with `execute_command(command, env=None)`, `read_file`, `write_file`, `list_dir`, `glob`, and `grep`. `grep` accepts either one text file or a directory tree. The optional `env` injects per-call environment variables (request-scoped secrets — see Request-Scoped Secrets below); `LocalSandbox` merges it via `subprocess.run(env=...)` and `AioSandbox` routes env-bearing commands through the `bash.exec(env=...)` API on a fresh session. **Provider Pattern**: `SandboxProvider` with `acquire`, `acquire_async`, `get`, `release` lifecycle. Async agent/tool paths call async sandbox lifecycle hooks so Docker sandbox creation, discovery, cross-process locking, readiness polling, and release stay off the event loop. **Environment policy** (`sandbox/env_policy.py`): `execute_command` no longer inherits the full `os.environ`. `build_sandbox_env()` scrubs secret-looking names (`*KEY*`/`*SECRET*`/`*TOKEN*`/`*PASS*`/`*CREDENTIAL*`) from the inherited environment before layering injected request secrets on top, so platform credentials (e.g. `OPENAI_API_KEY`) never leak into skill subprocesses. Benign vars (`PATH`, `HOME`, `LANG`, `VIRTUAL_ENV`, ...) are preserved. **Implementations**: @@ -606,6 +606,8 @@ copying a raw checkpoint because delta state is not self-contained in one tuple. **Sandbox Tools** (in `packages/harness/deerflow/sandbox/tools.py`): - `bash` - Execute commands with path translation and error handling. For `LocalSandbox` (host bash), POSIX output is captured through bounded pipe-drain threads and stdin is `/dev/null`, so a backgrounded long-lived process (`server &`) returns immediately instead of blocking the turn on an inherited pipe, while unredirected background output is drained without growing anonymous temp files. Commands that read stdin get immediate EOF. The command runs in its own process group with a wall-clock timeout (`sandbox.bash_command_timeout`, default 600s); on timeout the whole group is killed and the agent gets a notice telling it to background long-lived processes. The bash tool description itself also instructs the model to background long-lived processes (e.g. servers) up front so it doesn't waste the turn waiting on a foreground server. See `LocalSandbox.execute_command` / `_run_posix_command` and `bash_tool`'s docstring. - `ls` - Directory listing (tree format, max 2 levels) +- `glob` - Find files or directories below a root directory with bounded results +- `grep` - Search one text file or recursively search a directory, with optional glob filtering and bounded line-level results - `read_file` - Read file contents with optional line range - `write_file` - Write/append to files, creates directories; overwrites by default and exposes the `append` argument in the model-facing schema for end-of-file writes; subject to the read-before-write gate when `read_before_write.enabled` (see Middleware Chain) - `str_replace` - Substring replacement (single or all occurrences); same-path serialization is scoped to `(sandbox.id, path)` so isolated sandboxes do not contend on identical virtual paths inside one process; subject to the read-before-write gate when `read_before_write.enabled` (see Middleware Chain) diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py index 5ed2b4bd3..a3be91863 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox.py @@ -422,41 +422,45 @@ class AioSandbox(Sandbox): # (caught by grep_tool's except re.error handler) rather than a # generic remote API error. _re.compile(regex_source, 0 if case_sensitive else _re.IGNORECASE) - regex = regex_source if case_sensitive else f"(?i){regex_source}" - - if glob is not None: - find_result = self._client.file.find_files(path=path, glob=glob) - candidate_paths = find_result.data.files if find_result.data and find_result.data.files else [] - else: - list_result = self._client.file.list_path(path=path, recursive=True, show_hidden=False) - entries = list_result.data.files if list_result.data and list_result.data.files else [] - candidate_paths = [entry.path for entry in entries if not entry.is_directory] + total_cap = max(max_results * 4, max_results + 50) + result = self._client.file.grep_files( + path=path, + pattern=pattern, + case_insensitive=not case_sensitive, + fixed_strings=literal, + max_results=total_cap, + max_file_size="1M", + recursive=True, + ) + data = result.data + provider_matches = data.matches if data and data.matches else [] + root = path.rstrip("/") or "/" + root_prefix = root if root == "/" else f"{root}/" matches: list[GrepMatch] = [] - truncated = False - - for file_path in candidate_paths: + truncated = bool(data and data.truncated) + for match in provider_matches: + file_path = match.file if should_ignore_path(file_path): continue - - search_result = self._client.file.search_in_file(file=file_path, regex=regex) - data = search_result.data - if data is None: + if file_path == root: + rel_path = file_path.rsplit("/", 1)[-1] + elif file_path.startswith(root_prefix): + rel_path = file_path[len(root_prefix) :] + else: continue - - line_numbers = data.line_numbers or [] - matched_lines = data.matches or [] - for line_number, line in zip(line_numbers, matched_lines): - matches.append( - GrepMatch( - path=file_path, - line_number=line_number if isinstance(line_number, int) else 0, - line=truncate_line(line), - ) + if glob is not None and not path_matches(glob, rel_path): + continue + matches.append( + GrepMatch( + path=file_path, + line_number=match.line_number, + line=truncate_line(match.line_content), ) - if len(matches) >= max_results: - truncated = True - return matches, truncated + ) + if len(matches) >= max_results: + truncated = True + break return matches, truncated diff --git a/backend/packages/harness/deerflow/community/boxlite/box.py b/backend/packages/harness/deerflow/community/boxlite/box.py index 2f60dc1ee..534d7e7dd 100644 --- a/backend/packages/harness/deerflow/community/boxlite/box.py +++ b/backend/packages/harness/deerflow/community/boxlite/box.py @@ -325,11 +325,11 @@ class BoxliteBox(Sandbox): re.compile(pattern, 0 if case_sensitive else re.IGNORECASE) resolved = self._resolve_path(path) - # busybox+GNU-portable flags: -r recursive (also prints the filename), - # -n line numbers, -I skip binary, -E/-F regex vs fixed. --include and -m - # are omitted for busybox portability; glob-scoping and the result cap are - # applied in Python below. - flags = ["-r", "-n", "-I"] + # busybox+GNU-portable flags: -r recursive, -H always print the filename + # (including when path is a single file), -n line numbers, -I skip + # binary, -E/-F regex vs fixed. --include and -m are omitted for busybox + # portability; glob-scoping and the result cap are applied in Python. + flags = ["-r", "-H", "-n", "-I"] if not case_sensitive: flags.append("-i") flags.append("-F" if literal else "-E") diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox.py b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox.py index c069ba993..e8607975a 100644 --- a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox.py +++ b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox.py @@ -476,8 +476,8 @@ class E2BSandbox(Sandbox): # ``--include`` flag above only pre-filtered by basename. if file_path != root and not file_path.startswith(root_prefix): continue - rel_path = file_path[len(root) :].lstrip("/") - if not rel_path or not path_matches(glob, rel_path): + rel_path = file_path.rsplit("/", 1)[-1] if file_path == root else file_path[len(root) :].lstrip("/") + if not path_matches(glob, rel_path): continue matches.append( GrepMatch( diff --git a/backend/packages/harness/deerflow/community/tenki/sandbox.py b/backend/packages/harness/deerflow/community/tenki/sandbox.py index 995a7fa62..c227a7cc7 100644 --- a/backend/packages/harness/deerflow/community/tenki/sandbox.py +++ b/backend/packages/harness/deerflow/community/tenki/sandbox.py @@ -448,8 +448,8 @@ class TenkiSandbox(Sandbox): # helper, same relative-to-root semantics as glob() above. if file_path != root and not file_path.startswith(root_prefix): continue - rel_path = file_path[len(root) :].lstrip("/") - if not rel_path or not path_matches(glob, rel_path): + rel_path = posixpath.basename(file_path) if file_path == root else file_path[len(root) :].lstrip("/") + if not path_matches(glob, rel_path): continue matches.append(GrepMatch(path=self._virtual_path(file_path), line_number=line_number, line=truncate_line(line_text))) if len(matches) >= max_results: diff --git a/backend/packages/harness/deerflow/sandbox/sandbox.py b/backend/packages/harness/deerflow/sandbox/sandbox.py index cad0168d7..5147a9ac2 100644 --- a/backend/packages/harness/deerflow/sandbox/sandbox.py +++ b/backend/packages/harness/deerflow/sandbox/sandbox.py @@ -161,7 +161,7 @@ class Sandbox(ABC): case_sensitive: bool = False, max_results: int = 100, ) -> tuple[list[GrepMatch], bool]: - """Search for matches inside text files under a directory.""" + """Search for matches inside a text file or files under a directory.""" pass @abstractmethod diff --git a/backend/packages/harness/deerflow/sandbox/search.py b/backend/packages/harness/deerflow/sandbox/search.py index b4b7b9608..83d7f39a8 100644 --- a/backend/packages/harness/deerflow/sandbox/search.py +++ b/backend/packages/harness/deerflow/sandbox/search.py @@ -169,7 +169,8 @@ def find_grep_matches( if not root.exists(): raise FileNotFoundError(root) - if not root.is_dir(): + root_is_file = root.is_file() + if not root_is_file and not root.is_dir(): raise NotADirectoryError(root) regex_source = re.escape(pattern) if literal else pattern @@ -179,44 +180,47 @@ def find_grep_matches( # Skip lines longer than this to prevent ReDoS on minified / no-newline files. _max_line_chars = line_summary_length * 10 - for current_root, dirs, files in os.walk(root): - dirs[:] = [name for name in dirs if not should_ignore_name(name)] - rel_dir = Path(current_root).relative_to(root) + def candidate_files(): + if root_is_file: + yield root, root.name + return - for name in files: - if should_ignore_name(name): + for current_root, dirs, files in os.walk(root): + dirs[:] = [name for name in dirs if not should_ignore_name(name)] + rel_dir = Path(current_root).relative_to(root) + for name in files: + if should_ignore_name(name): + continue + yield Path(current_root) / name, (rel_dir / name).as_posix() + + for candidate_path, rel_path in candidate_files(): + if glob_pattern is not None and not path_matches(glob_pattern, rel_path): + continue + + try: + if not root_is_file and candidate_path.is_symlink(): continue - - candidate_path = Path(current_root) / name - rel_path = (rel_dir / name).as_posix() - - if glob_pattern is not None and not path_matches(glob_pattern, rel_path): + file_path = candidate_path.resolve() + if not root_is_file and not file_path.is_relative_to(root): continue - - try: - if candidate_path.is_symlink(): - continue - file_path = candidate_path.resolve() - if not file_path.is_relative_to(root): - continue - if file_path.stat().st_size > max_file_size or is_binary_file(file_path): - continue - with file_path.open(encoding="utf-8", errors="replace") as handle: - for line_number, line in enumerate(handle, start=1): - if len(line) > _max_line_chars: - continue - if regex.search(line): - matches.append( - GrepMatch( - path=str(file_path), - line_number=line_number, - line=truncate_line(line, line_summary_length), - ) + if file_path.stat().st_size > max_file_size or is_binary_file(file_path): + continue + with file_path.open(encoding="utf-8", errors="replace") as handle: + for line_number, line in enumerate(handle, start=1): + if len(line) > _max_line_chars: + continue + if regex.search(line): + matches.append( + GrepMatch( + path=str(file_path), + line_number=line_number, + line=truncate_line(line, line_summary_length), ) - if len(matches) >= max_results: - truncated = True - return matches, truncated - except OSError: - continue + ) + if len(matches) >= max_results: + truncated = True + return matches, truncated + except OSError: + continue return matches, truncated diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py index 48db29f7b..85c4f7b17 100644 --- a/backend/packages/harness/deerflow/sandbox/tools.py +++ b/backend/packages/harness/deerflow/sandbox/tools.py @@ -1993,12 +1993,12 @@ def grep_tool( case_sensitive: bool = False, max_results: int = _DEFAULT_GREP_MAX_RESULTS, ) -> str: - """Search for matching lines inside text files under a root directory. + """Search for matching lines inside a text file or files under a root directory. Args: description: Explain why you are searching file contents in short words. ALWAYS PROVIDE THIS PARAMETER FIRST. pattern: The string or regex pattern to search for. - path: The **absolute** root directory to search under. + path: The **absolute** file or root directory to search. glob: Optional glob filter for candidate files, for example `**/*.py`. literal: Whether to treat `pattern` as a plain string. Default is False. case_sensitive: Whether matching is case-sensitive. Default is False. diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index 59ed07530..5dafec0ad 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -5,7 +5,7 @@ description = "DeerFlow agent harness framework" requires-python = ">=3.12" dependencies = [ "agent-client-protocol>=0.4.0", - "agent-sandbox>=0.0.19", + "agent-sandbox>=0.0.30", "croniter>=6.0.0", "dotenv>=0.9.9", "exa-py>=1.0.0", diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py index f163285dd..c151dd83c 100644 --- a/backend/tests/test_boxlite_provider.py +++ b/backend/tests/test_boxlite_provider.py @@ -165,6 +165,17 @@ def test_execute_command_forwards_timeout_to_sdk_and_loop_runner() -> None: assert run_timeouts == [5] +def test_grep_always_prints_filename_for_single_file_paths() -> None: + fake = _FakeBox(name="box-id") + box = BoxliteBox("box-id", box=fake, run=_fake_run) + + box.grep("/mnt/user-data/uploads/report.md", "needle") + + grep_commands = [argv[0][2] for argv in fake._exec_history if argv[0][:2] == ("sh", "-lc") and argv[0][2].startswith("grep ")] + assert grep_commands + assert "-H" in grep_commands[-1].split() + + def test_execute_command_invalidates_box_on_terminal_transport_error() -> None: invalidated: list[tuple[str, str]] = [] diff --git a/backend/tests/test_e2b_sandbox_provider.py b/backend/tests/test_e2b_sandbox_provider.py index b654dffe4..c438ab476 100644 --- a/backend/tests/test_e2b_sandbox_provider.py +++ b/backend/tests/test_e2b_sandbox_provider.py @@ -2072,6 +2072,18 @@ def test_grep_without_glob_is_unaffected(): assert truncated is False +def test_grep_single_file_path_with_matching_glob(): + """A basename glob must also apply when the search root is one file.""" + raw_stdout = "/home/user/uploads/report.md:2:needle here\n" + client = FakeClient(commands=FakeCommandsAPI([SimpleNamespace(stdout=raw_stdout, stderr="", exit_code=0)])) + sb = _make_sandbox(client) + + matches, truncated = sb.grep("/mnt/user-data/uploads/report.md", "needle", glob="*.md") + + assert [m.path for m in matches] == ["/home/user/uploads/report.md"] + assert truncated is False + + # ────────────────────────────────────────────────────────────────────────────── # Capacity enforcement tests (#4339) # ────────────────────────────────────────────────────────────────────────────── diff --git a/backend/tests/test_sandbox_search_tools.py b/backend/tests/test_sandbox_search_tools.py index e3438c823..1acbacefd 100644 --- a/backend/tests/test_sandbox_search_tools.py +++ b/backend/tests/test_sandbox_search_tools.py @@ -101,6 +101,26 @@ def test_grep_tool_filters_by_glob_and_skips_binary_files(tmp_path, monkeypatch) assert str(workspace) not in result +def test_grep_tool_accepts_single_file_path(tmp_path, monkeypatch) -> None: + runtime = _make_runtime(tmp_path) + uploads = tmp_path / "uploads" + report = uploads / "report.md" + report.write_text("Revenue grew 20%\n", encoding="utf-8") + + monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: LocalSandbox(id="local")) + + result = grep_tool.func( + runtime=runtime, + description="find revenue in the uploaded report", + pattern="Revenue", + path="/mnt/user-data/uploads/report.md", + ) + + assert "/mnt/user-data/uploads/report.md:1: Revenue grew 20%" in result + assert "Path is not a directory" not in result + assert str(uploads) not in result + + def test_grep_tool_truncates_results(tmp_path, monkeypatch) -> None: runtime = _make_runtime(tmp_path) workspace = tmp_path / "workspace" @@ -260,24 +280,20 @@ def test_aio_sandbox_grep_parses_json(monkeypatch) -> None: sandbox = AioSandbox(id="test-sandbox", base_url="http://localhost:8080") monkeypatch.setattr( sandbox._client.file, - "list_path", + "grep_files", lambda **kwargs: SimpleNamespace( data=SimpleNamespace( - files=[ + matches=[ SimpleNamespace( - name="app.py", - path="/mnt/user-data/workspace/app.py", - is_directory=False, + file="/mnt/user-data/workspace/app.py", + line_number=7, + line_content="TODO = True", ) - ] + ], + truncated=False, ) ), ) - monkeypatch.setattr( - sandbox._client.file, - "search_in_file", - lambda **kwargs: SimpleNamespace(data=SimpleNamespace(line_numbers=[7], matches=["TODO = True"])), - ) matches, truncated = sandbox.grep("/mnt/user-data/workspace", "TODO") @@ -285,6 +301,37 @@ def test_aio_sandbox_grep_parses_json(monkeypatch) -> None: assert truncated is False +def test_aio_sandbox_grep_accepts_single_file_path(monkeypatch) -> None: + with patch("deerflow.community.aio_sandbox.aio_sandbox.AioSandboxClient"): + sandbox = AioSandbox(id="test-sandbox", base_url="http://localhost:8080") + monkeypatch.setattr( + sandbox._client.file, + "grep_files", + lambda **kwargs: SimpleNamespace( + data=SimpleNamespace( + matches=[ + SimpleNamespace( + file="/mnt/user-data/uploads/report.md", + line_number=3, + line_content="Revenue grew 20%", + ) + ], + truncated=False, + ) + ), + ) + monkeypatch.setattr( + sandbox._client.file, + "list_path", + lambda **kwargs: (_ for _ in ()).throw(AssertionError("single-file grep must not list the path as a directory")), + ) + + matches, truncated = sandbox.grep("/mnt/user-data/uploads/report.md", "Revenue") + + assert matches == [GrepMatch(path="/mnt/user-data/uploads/report.md", line_number=3, line="Revenue grew 20%")] + assert truncated is False + + def test_find_glob_matches_raises_not_a_directory(tmp_path) -> None: file_path = tmp_path / "file.txt" file_path.write_text("x\n", encoding="utf-8") @@ -296,15 +343,14 @@ def test_find_glob_matches_raises_not_a_directory(tmp_path) -> None: pass -def test_find_grep_matches_raises_not_a_directory(tmp_path) -> None: +def test_find_grep_matches_accepts_single_file(tmp_path) -> None: file_path = tmp_path / "file.txt" file_path.write_text("TODO\n", encoding="utf-8") - try: - find_grep_matches(file_path, "TODO") - assert False, "Expected NotADirectoryError" - except NotADirectoryError: - pass + matches, truncated = find_grep_matches(file_path, "TODO") + + assert matches == [GrepMatch(path=str(file_path), line_number=1, line="TODO")] + assert truncated is False def test_find_grep_matches_skips_symlink_outside_root(tmp_path) -> None: @@ -367,29 +413,30 @@ def test_aio_sandbox_glob_include_dirs_enforces_root_boundary(monkeypatch) -> No assert truncated is False -def test_aio_sandbox_grep_skips_mismatched_line_number_payloads(monkeypatch) -> None: +def test_aio_sandbox_grep_drops_matches_outside_requested_root(monkeypatch) -> None: with patch("deerflow.community.aio_sandbox.aio_sandbox.AioSandboxClient"): sandbox = AioSandbox(id="test-sandbox", base_url="http://localhost:8080") monkeypatch.setattr( sandbox._client.file, - "list_path", + "grep_files", lambda **kwargs: SimpleNamespace( data=SimpleNamespace( - files=[ + matches=[ SimpleNamespace( - name="app.py", - path="/mnt/user-data/workspace/app.py", - is_directory=False, - ) - ] + file="/mnt/user-data/workspace/app.py", + line_number=7, + line_content="TODO = True", + ), + SimpleNamespace( + file="/mnt/user-data/workspace-sibling/leak.py", + line_number=9, + line_content="TODO = False", + ), + ], + truncated=False, ) ), ) - monkeypatch.setattr( - sandbox._client.file, - "search_in_file", - lambda **kwargs: SimpleNamespace(data=SimpleNamespace(line_numbers=[7], matches=["TODO = True", "extra"])), - ) matches, truncated = sandbox.grep("/mnt/user-data/workspace", "TODO") diff --git a/backend/tests/test_tenki_provider.py b/backend/tests/test_tenki_provider.py index 2fe8ba93b..77547f302 100644 --- a/backend/tests/test_tenki_provider.py +++ b/backend/tests/test_tenki_provider.py @@ -543,6 +543,16 @@ def test_grep_passes_capital_h_so_single_file_matches_parse() -> None: assert grep_scripts and "-H" in shlex.split(grep_scripts[0]) +def test_grep_single_file_path_with_matching_glob() -> None: + box = TenkiSandbox("sb", _FakeSandbox()) + box.write_file("/mnt/user-data/workspace/a.txt", "needle here\n") + + matches, truncated = box.grep("/mnt/user-data/workspace/a.txt", "needle", glob="*.txt") + + assert [m.path for m in matches] == ["/mnt/user-data/workspace/a.txt"] + assert truncated is False + + def _grep_script(fake: _FakeSandbox) -> list[str]: scripts = [c["argv"][2] for c in fake.exec_calls if c["argv"][:2] == ("sh", "-lc") and c["argv"][2].startswith("grep ")] assert scripts, "no grep command was issued" diff --git a/backend/uv.lock b/backend/uv.lock index bc81bd7ee..de87f4a56 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -952,7 +952,7 @@ tui = [ [package.metadata] requires-dist = [ { name = "agent-client-protocol", specifier = ">=0.4.0" }, - { name = "agent-sandbox", specifier = ">=0.0.19" }, + { name = "agent-sandbox", specifier = ">=0.0.30" }, { name = "aiosqlite", specifier = ">=0.19" }, { name = "alembic", specifier = ">=1.13" }, { name = "asyncpg", marker = "extra == 'postgres'", specifier = ">=0.29" }, From 6456c35675dfbfdfc25ec5346e52ed8a8de1c5ef Mon Sep 17 00:00:00 2001 From: ajayr Date: Tue, 28 Jul 2026 00:56:04 +0100 Subject: [PATCH 3/4] fix(browserless): accept the `timeout` config key and harden coercion (#4519) `browserless` reads `cfg["timeout_s"]`, while its sibling web providers `crawl4ai` and `jina_ai` read `cfg["timeout"]`. Tool configs allow extra fields, so the unrecognised spelling is dropped without a diagnostic: someone adapting one provider's config snippet for another silently gets the 30s default instead of the timeout they set. (Observed in the other direction, on a deployment whose crawl4ai entry carried `timeout_s`.) Accept both keys, preferring the documented `timeout_s` when both are present. While adding coverage, two pre-existing bugs in the same three lines surfaced, both already guarded in crawl4ai/jina_ai but not here: - `timeout_s: "30s"` (or any non-numeric string) raised ValueError out of `float(raw)` during tool construction rather than falling back. - `timeout_s: off` -- YAML parses that as `False`, and `float(False)` is `0.0`, so every request timed out immediately against a healthy server. `_coerce_timeout` now mirrors the sibling providers: booleans and unparsable strings fall back to the default, with a warning for the string case. Tests: five cases in tests/test_browserless_client.py covering both keys, the precedence order, and both coercion bugs. Verified red before the fix (3 of 5 fail) and green after. Co-authored-by: Claude Opus 5 --- .../deerflow/community/browserless/tools.py | 38 ++++++++++++++++++- backend/tests/test_browserless_client.py | 30 +++++++++++++++ config.example.yaml | 2 +- 3 files changed, 67 insertions(+), 3 deletions(-) diff --git a/backend/packages/harness/deerflow/community/browserless/tools.py b/backend/packages/harness/deerflow/community/browserless/tools.py index f5829c84b..7fd981c94 100644 --- a/backend/packages/harness/deerflow/community/browserless/tools.py +++ b/backend/packages/harness/deerflow/community/browserless/tools.py @@ -44,6 +44,41 @@ def _get_tool_config(tool_name: str) -> dict | None: return extras if extras is not None else {} +def _coerce_timeout(value: object, default: float) -> float: + """Coerce a config timeout into seconds, falling back to ``default`` on bad input. + + Mirrors ``crawl4ai._coerce_timeout`` / ``jina_ai._coerce_timeout``: booleans and + non-numeric strings fall back to the default, so ``timeout_s: off`` (YAML ``False``) + does not become ``0.0`` and time out every request against a healthy server, and a + typo'd value does not raise out of tool construction. + """ + if isinstance(value, bool): + return default + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str): + try: + return float(value) + except ValueError: + logger.warning("Browserless: invalid timeout %r in config; using %ss", value, default) + return default + + +def _resolve_timeout(cfg: dict, default: float) -> float: + """Read the timeout, accepting this provider's key and the sibling providers' key. + + ``browserless`` documents ``timeout_s`` while ``crawl4ai`` and ``jina_ai`` read + ``timeout``. An unrecognised key is dropped silently by the extra-fields tool config, + so someone adapting another provider's snippet got the default with no diagnostic. + Accept both, preferring the documented ``timeout_s`` when both are present. + """ + if "timeout_s" in cfg: + return _coerce_timeout(cfg["timeout_s"], default) + if "timeout" in cfg: + return _coerce_timeout(cfg["timeout"], default) + return default + + def _get_browserless_client(tool_name: str = "web_fetch") -> BrowserlessClient: cfg = _get_tool_config(tool_name) base_url = "http://localhost:3032" @@ -52,8 +87,7 @@ def _get_browserless_client(tool_name: str = "web_fetch") -> BrowserlessClient: if cfg is not None: base_url = cfg.get("base_url", base_url) token = cfg.get("token", token) - raw = cfg.get("timeout_s", timeout_s) - timeout_s = float(raw) if not isinstance(raw, float) else raw + timeout_s = _resolve_timeout(cfg, timeout_s) return BrowserlessClient(base_url=base_url, token=token, timeout_s=timeout_s) diff --git a/backend/tests/test_browserless_client.py b/backend/tests/test_browserless_client.py index fe6ebcdfd..a2f80b8ea 100644 --- a/backend/tests/test_browserless_client.py +++ b/backend/tests/test_browserless_client.py @@ -319,6 +319,36 @@ class TestBrowserlessTools: assert client.token == "env-token" + def _client_with_config(self, cfg: dict): + with patch("deerflow.community.browserless.tools._get_tool_config") as mock_cfg: + mock_cfg.return_value = cfg + with patch.dict("os.environ", {}, clear=True): + return tools._get_browserless_client("web_capture") + + async def test_timeout_s_config_key_is_honored(self): + """The documented `timeout_s` key keeps working (back-compat).""" + assert self._client_with_config({"timeout_s": 45}).timeout_s == 45.0 + + async def test_timeout_config_key_is_honored(self): + """`timeout` is accepted too: it is what crawl4ai and jina_ai read. + + Copying that spelling across providers previously produced a silently + ignored key and the 30s default. + """ + assert self._client_with_config({"timeout": 45}).timeout_s == 45.0 + + async def test_timeout_s_wins_when_both_keys_are_present(self): + """`timeout_s` is this provider's documented key, so it takes precedence.""" + assert self._client_with_config({"timeout_s": 45, "timeout": 10}).timeout_s == 45.0 + + async def test_invalid_timeout_falls_back_to_default(self): + """A non-numeric timeout must not crash tool construction.""" + assert self._client_with_config({"timeout_s": "not-a-number"}).timeout_s == 30.0 + + async def test_boolean_timeout_falls_back_to_default(self): + """YAML `timeout_s: off` parses as False; 0.0 would time out every request.""" + assert self._client_with_config({"timeout_s": False}).timeout_s == 30.0 + @patch("deerflow.community.browserless.tools._get_browserless_client") async def test_web_fetch_tool_success(self, mock_get_client): """web_fetch_tool successfully fetches and extracts content.""" diff --git a/config.example.yaml b/config.example.yaml index a7ab5a555..95d3a1f40 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -796,7 +796,7 @@ tools: # use: deerflow.community.browserless.tools:web_capture_tool # base_url: http://localhost:3032 # Browserless instance URL (Docker: http://browserless:3000) # # token: $BROWSERLESS_TOKEN # Required for Browserless Cloud; optional for self-hosted - # timeout_s: 30 # Request timeout in seconds + # timeout_s: 30 # Request timeout in seconds (`timeout` also accepted) # output_format: png # png, jpeg, or webp # full_page: true # Capture entire page instead of viewport only # viewport_width: 1280 From 2654bc60da315ffeda9b7e15959dd0530e62f6aa Mon Sep 17 00:00:00 2001 From: Huixin615 Date: Tue, 28 Jul 2026 13:57:17 +0800 Subject: [PATCH 4/4] fix(frontend): preserve message order during long runs (#4513) * fix(frontend): preserve message order during long runs * test(frontend): fix history pagination regression mock * fix(frontend): validate thread history sequences --- frontend/AGENTS.md | 2 +- frontend/src/core/threads/hooks.ts | 302 +++++++++++++- frontend/src/core/threads/types.ts | 2 +- frontend/tests/e2e/thread-history.spec.ts | 156 +++++++ .../unit/core/threads/message-merge.test.ts | 381 +++++++++++++++++- 5 files changed, 818 insertions(+), 25 deletions(-) diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index a463b5df1..9e76d05e2 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -68,7 +68,7 @@ The frontend is a stateful chat application. Users create **threads** (conversat 1. Optional composer helpers such as `core/input-polish` can rewrite the local draft before submission, and `core/voice-input` can transcribe browser microphone input into that same local draft; confirmed user input then flows to thread hooks (`core/threads/hooks.ts`) → LangGraph SDK streaming 2. Stream events update thread state (messages, artifacts, todos, goal) File-tool artifact auto-open work must run in an effect with timer cleanup; never schedule timers while rendering streamed `write_file` or `str_replace` updates. -3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail), suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. +3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. The resolver suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. 4. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits 5. TanStack Query manages server state; localStorage stores user settings 6. Components subscribe to thread state and render updates diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index a5b312ccc..9399b357c 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -161,6 +161,8 @@ export function buildThreadSubmitMessages({ // Stable identity for "no optimistic messages" so the merged-messages memo // below is not invalidated by a fresh empty array on every render. const EMPTY_MESSAGES: Message[] = []; +const EMPTY_RUN_MESSAGES: RunMessage[] = []; +const EMPTY_MESSAGE_IDENTITIES: readonly string[] = []; const EMPTY_THREAD_VALUES: AgentThreadState = { title: "", @@ -302,6 +304,56 @@ export type ThreadMessagesPageResponse = { next_before_seq: number | null; }; +function isValidThreadMessageSeq(value: unknown): value is number { + return typeof value === "number" && Number.isSafeInteger(value) && value >= 1; +} + +/** + * Validate the sequence fields that history reconciliation and pagination use + * as runtime identities. The static RunMessage type cannot protect this JSON + * boundary from version skew or malformed responses. + */ +export function parseThreadMessagesPageResponse( + value: unknown, +): ThreadMessagesPageResponse { + if (typeof value !== "object" || value === null) { + throw new Error("Thread history returned an invalid response."); + } + + const data = Reflect.get(value, "data"); + const hasMore = Reflect.get(value, "has_more"); + const nextBeforeSeq = Reflect.get(value, "next_before_seq"); + if (!Array.isArray(data) || typeof hasMore !== "boolean") { + throw new Error("Thread history returned an invalid response."); + } + + const seenSeqs = new Set(); + for (const row of data) { + const seq = + typeof row === "object" && row !== null + ? Reflect.get(row, "seq") + : undefined; + if (!isValidThreadMessageSeq(seq)) { + throw new Error("Thread history returned a row with an invalid seq."); + } + if (seenSeqs.has(seq)) { + throw new Error("Thread history returned duplicate seq values."); + } + seenSeqs.add(seq); + } + + if ( + (hasMore && !isValidThreadMessageSeq(nextBeforeSeq)) || + (!hasMore && nextBeforeSeq !== null) + ) { + throw new Error( + "Thread history returned an invalid next_before_seq cursor.", + ); + } + + return value as ThreadMessagesPageResponse; +} + export function getThreadHistoryNextPageParam( lastPage: ThreadMessagesPageResponse, ): number | undefined { @@ -348,6 +400,51 @@ export function flattenThreadHistoryPages( ); } +/** + * Preserve rows that this client has already loaded while newest-first cursor + * pages move forward during a long run. + * + * A background refetch recalculates every loaded page from the refreshed first + * page. When older pages have not all been loaded yet, that can displace rows + * which were visible a moment ago even though they still exist on the server. + * Thread-global seq is the authoritative order; refreshed copies win without + * moving their established position. + */ +export function reconcileThreadHistoryRows( + previousRows: RunMessage[], + currentRows: RunMessage[], + isAuthoritativeComplete: boolean, +): RunMessage[] { + const sourceRows = isAuthoritativeComplete + ? currentRows + : [...previousRows, ...currentRows]; + if (sourceRows.some((row) => !isValidThreadMessageSeq(row.seq))) { + console.error( + "Thread history reconciliation received an invalid sequence value.", + ); + // Never skip an invalid row or feed it into Map/Array.sort: either choice + // can silently lose or misorder messages. A failed refresh keeps the last + // known-good snapshot; an invalid first snapshot degrades to server order. + return previousRows.length > 0 ? previousRows : currentRows; + } + + const rowsBySeq = new Map(); + for (const row of sourceRows) { + rowsBySeq.set(row.seq, row); + } + + const reconciled = dedupeRunMessagesByIdentity( + [...rowsBySeq.values()].sort((left, right) => left.seq - right.seq), + ); + if ( + reconciled.length === previousRows.length && + reconciled.every((row, index) => row === previousRows[index]) + ) { + return previousRows; + } + return reconciled; +} + export function mergeMessages( historyMessages: Message[], threadMessages: Message[], @@ -481,33 +578,78 @@ export function mergeMessages( }); } +/** + * Keep a run-scoped ledger of every visible message that reached a committed + * UI frame. Live checkpoint windows can roll forward between two + * summarization events; replacing this ledger with only the newest window + * would make the intervening steps impossible to rescue at the next + * RemoveMessage(ALL). + * + * The newest visible copy wins by identity without moving its established + * position. Explicitly superseded messages are removed so regeneration cannot + * revive an answer that the UI intentionally hid. + */ +export function mergeRenderedMessageLedger( + previouslyRenderedMessages: Message[], + visibleMessages: Message[], + supersededMessageIds: ReadonlySet = new Set(), +): Message[] { + const isEligible = (message: Message) => + messageIdentity(message) !== undefined && + (!message.id || !supersededMessageIds.has(message.id)); + const retainedPrevious = + supersededMessageIds.size === 0 + ? previouslyRenderedMessages.filter( + (message) => messageIdentity(message) !== undefined, + ) + : previouslyRenderedMessages.filter(isEligible); + const eligibleVisibleMessages = visibleMessages.filter(isEligible); + if (retainedPrevious.length === 0) { + return eligibleVisibleMessages; + } + return mergeMessages(retainedPrevious, eligibleVisibleMessages, []); +} + /** * Derive the live turns that context summarization is about to drop and that * therefore need a short-lived visual bridge until run-event history catches up. * * Summarization emits `RemoveMessage(ALL)` + a hidden summary + the retained - * tail. Everything in the current live thread before the first retained visible - * message is being removed; we keep those (minus the summary control messages - * already tracked) so the UI can still show the full conversation (#3825). + * tail. Everything in the current live thread that is absent from the retained + * visible window is being removed; we keep those (minus the summary control + * messages already tracked) so the UI can still show the full conversation + * (#3825). Comparing identities instead of slicing at the first retained + * message also handles a protected early input followed by a recent tail. */ export function computeSummarizationTransientMessages( currentMessages: Message[], summarizationMessages: Message[], summarizedMessageIds: ReadonlySet, + previouslyRenderedMessages: Message[] = EMPTY_MESSAGES, ): Message[] { - const firstRetainedVisibleIdentity = summarizationMessages - .filter((message) => message.type !== "remove") - .filter((message) => !isHiddenFromUIMessage(message)) - .map(messageIdentity) - .find(isNonEmptyString); + const retainedVisibleIdentities = new Set( + summarizationMessages + .filter((message) => message.type !== "remove") + .filter((message) => !isHiddenFromUIMessage(message)) + .map(messageIdentity) + .filter(isNonEmptyString), + ); + // Updates can outrun React while the SDK applies RemoveMessage(ALL). In that + // case currentMessages may already be the retained post-compaction window + // even though the previous committed UI frame still showed the removed + // processing steps. Use that frame as the chronological base, then overlay + // fresher live copies. This rescues only messages the user actually saw and + // preserves the unloaded-history-gap protection in the bridge resolver. + const captureMessages = + previouslyRenderedMessages.length > 0 + ? mergeMessages(previouslyRenderedMessages, currentMessages, []) + : currentMessages; const moved: Message[] = []; - for (const message of currentMessages) { - if ( - firstRetainedVisibleIdentity && - messageIdentity(message) === firstRetainedVisibleIdentity - ) { - break; + for (const message of captureMessages) { + const identity = messageIdentity(message); + if (identity && retainedVisibleIdentities.has(identity)) { + continue; } if (!summarizedMessageIds.has(message.id ?? "")) { moved.push(message); @@ -538,6 +680,7 @@ export function resolveTransientHistoryBridge( bridgeOrder: readonly string[] = transientMessages .map(messageIdentity) .filter(isNonEmptyString), + previouslyRenderedOrder: readonly string[] = EMPTY_MESSAGE_IDENTITIES, ): Message[] { if (transientMessages.length === 0) { return visibleHistory; @@ -568,20 +711,53 @@ export function resolveTransientHistoryBridge( // intentionally excluded to avoid permanent duplicates. const beforeAnchor = new Map(); const emittedMissingIdentities = new Set(); + const previouslyRenderedIndex = new Map( + previouslyRenderedOrder.map((identity, index) => [identity, index]), + ); let pending: Message[] = []; let lastAnchorIdentity: string | undefined; let hasCanonicalAnchor = false; for (const identity of bridgeOrder) { if (presentIdentities.has(identity)) { - if (pending.length > 0 && hasCanonicalAnchor) { - beforeAnchor.set(identity, [ - ...(beforeAnchor.get(identity) ?? []), - ...pending, - ]); + if (pending.length > 0) { + if (hasCanonicalAnchor) { + beforeAnchor.set(identity, [ + ...(beforeAnchor.get(identity) ?? []), + ...pending, + ]); + } else { + const anchorRenderIndex = previouslyRenderedIndex.get(identity); + if (anchorRenderIndex !== undefined) { + const safeRenderedPrefix = pending + .filter((message) => { + const pendingIdentity = messageIdentity(message); + const renderIndex = pendingIdentity + ? previouslyRenderedIndex.get(pendingIdentity) + : undefined; + return ( + renderIndex !== undefined && renderIndex < anchorRenderIndex + ); + }) + .sort((left, right) => { + const leftIndex = + previouslyRenderedIndex.get(messageIdentity(left) ?? "") ?? + Number.MAX_SAFE_INTEGER; + const rightIndex = + previouslyRenderedIndex.get(messageIdentity(right) ?? "") ?? + Number.MAX_SAFE_INTEGER; + return leftIndex - rightIndex; + }); + if (safeRenderedPrefix.length > 0) { + beforeAnchor.set(identity, safeRenderedPrefix); + } + } + } } // The prefix before the first loaded anchor has no trustworthy position: // cursor pages containing its intervening history may not be loaded yet. + // The sole exception is a prefix whose exact relative position was + // already committed to the previous UI frame. pending = []; hasCanonicalAnchor = true; lastAnchorIdentity = identity; @@ -696,6 +872,7 @@ export function resolveThreadTransientHistoryBridge( bridgeThreadId: string | null, currentThreadId: string | null | undefined, bridgeOrder?: readonly string[], + previouslyRenderedOrder?: readonly string[], ): Message[] { if (!bridgeThreadId || bridgeThreadId !== currentThreadId) { return visibleHistory; @@ -704,6 +881,7 @@ export function resolveThreadTransientHistoryBridge( visibleHistory, transientMessages, bridgeOrder, + previouslyRenderedOrder, ); } @@ -1324,6 +1502,9 @@ export function useThreadStream({ messagesRef.current, _messages, summarizedRef.current ?? new Set(), + renderedMessageSnapshotRef.current.threadId === threadIdRef.current + ? renderedMessageSnapshotRef.current.messages + : EMPTY_MESSAGES, ); transientHistoryOrderRef.current = mergeTransientHistoryBridgeOrder( transientHistoryOrderRef.current, @@ -1535,6 +1716,19 @@ export function useThreadStream({ // anchors so an older rescue can be placed before a newest-first page. const transientHistoryOrderRef = useRef([]); const transientHistoryThreadIdRef = useRef(null); + // The run-scoped committed display ledger supplies message objects when a + // compaction replacement outruns React or several rolling checkpoint windows + // pass between compactions. The merged order separately anchors those + // objects against canonical history. + const renderedMessageSnapshotRef = useRef<{ + threadId: string | null; + messages: Message[]; + order: readonly string[]; + }>({ + threadId: null, + messages: EMPTY_MESSAGES, + order: EMPTY_MESSAGE_IDENTITIES, + }); const summarizedRef = useRef>(null); // Track human message count before sending to prevent clearing optimistic // messages before the server's human message arrives (e.g. when AI messages @@ -1554,6 +1748,11 @@ export function useThreadStream({ transientHistoryBridgeRef.current = []; transientHistoryOrderRef.current = []; transientHistoryThreadIdRef.current = null; + renderedMessageSnapshotRef.current = { + threadId: null, + messages: EMPTY_MESSAGES, + order: EMPTY_MESSAGE_IDENTITIES, + }; summarizedRef.current = new Set(); pendingUsageBaselineMessageIdsRef.current = new Set(); pendingPreparedReplayRef.current = null; @@ -2069,6 +2268,10 @@ export function useThreadStream({ persistedMessages, ) : transientHistoryOrderRef.current; + const previouslyRenderedOrder = + renderedMessageSnapshotRef.current.threadId === threadId + ? renderedMessageSnapshotRef.current.order + : EMPTY_MESSAGE_IDENTITIES; // Commit the extended non-rendering order skeleton after React commits this // render. The local value above keeps this render correctly anchored without @@ -2096,6 +2299,7 @@ export function useThreadStream({ transientHistoryThreadIdRef.current, threadId, transientHistoryOrder, + previouslyRenderedOrder, ); return mergeMessages( effectiveHistory, @@ -2103,12 +2307,36 @@ export function useThreadStream({ visibleOptimisticMessages, ); }, [ + previouslyRenderedOrder, renderMessages, threadId, transientHistoryOrder, visibleHistory, visibleOptimisticMessages, ]); + useEffect(() => { + const visibleMergedMessages = mergedMessages.filter( + (message) => + !isHiddenFromUIMessage(message) && !message.id?.startsWith("opt-"), + ); + const previousLedger = + thread.isLoading && + renderedMessageSnapshotRef.current.threadId === threadId + ? renderedMessageSnapshotRef.current.messages + : EMPTY_MESSAGES; + const renderedMessageLedger = mergeRenderedMessageLedger( + previousLedger, + visibleMergedMessages, + pendingSupersededMessageIds, + ); + renderedMessageSnapshotRef.current = { + threadId: threadId ?? null, + messages: renderedMessageLedger, + order: renderedMessageLedger + .map(messageIdentity) + .filter(isNonEmptyString), + }; + }, [mergedMessages, pendingSupersededMessageIds, thread.isLoading, threadId]); const pendingUsageMessages = thread.isLoading ? getMessagesAfterBaseline( persistedMessages, @@ -2179,15 +2407,47 @@ export function useThreadHistory( ), ); } - return (await response.json()) as ThreadMessagesPageResponse; + return parseThreadMessagesPageResponse(await response.json()); }, getNextPageParam: getThreadHistoryNextPageParam, }); - const messageRows = useMemo( + const currentMessageRows = useMemo( () => flattenThreadHistoryPages(historyQuery.data?.pages ?? []), [historyQuery.data?.pages], ); + const [retainedHistory, setRetainedHistory] = useState<{ + threadId: string; + rows: RunMessage[]; + }>({ threadId, rows: EMPTY_RUN_MESSAGES }); + const previousRows = + retainedHistory.threadId === threadId + ? retainedHistory.rows + : EMPTY_RUN_MESSAGES; + const pages = historyQuery.data?.pages ?? []; + const isAuthoritativeComplete = + historyQuery.isSuccess && + !historyQuery.isFetching && + pages.length > 0 && + pages.at(-1)?.has_more === false; + const messageRows = useMemo( + () => + reconcileThreadHistoryRows( + previousRows, + currentMessageRows, + isAuthoritativeComplete, + ), + [currentMessageRows, isAuthoritativeComplete, previousRows], + ); + + useEffect(() => { + setRetainedHistory((current) => { + if (current.threadId === threadId && current.rows === messageRows) { + return current; + } + return { threadId, rows: messageRows }; + }); + }, [messageRows, threadId]); const messages = useMemo(() => { return buildVisibleHistoryMessages( diff --git a/frontend/src/core/threads/types.ts b/frontend/src/core/threads/types.ts index 6042ed129..e4578e147 100644 --- a/frontend/src/core/threads/types.ts +++ b/frontend/src/core/threads/types.ts @@ -53,7 +53,7 @@ export interface AgentThread extends Thread { export interface RunMessage { run_id: string; - seq?: number; + seq: number; content: Message; metadata: { caller: string; diff --git a/frontend/tests/e2e/thread-history.spec.ts b/frontend/tests/e2e/thread-history.spec.ts index 3ed9b397d..5a92c5f99 100644 --- a/frontend/tests/e2e/thread-history.spec.ts +++ b/frontend/tests/e2e/thread-history.spec.ts @@ -94,6 +94,162 @@ test.describe("Thread history", () => { ).toBeVisible({ timeout: 15_000 }); }); + test("keeps rendered messages ordered when the latest history page advances", async ({ + page, + }) => { + const originalPrompt = "/ppt-master Build the quarterly presentation"; + const followUpPrompt = "Continue with the approved default layout"; + const olderRows = Array.from({ length: 50 }, (_, index) => { + const seq = index + 1; + if (index === 0) { + return { + run_id: "run-initial", + seq, + content: { + type: "human", + id: "history-prompt", + content: [{ type: "text", text: originalPrompt }], + }, + metadata: { caller: "lead_agent" }, + created_at: "2025-06-03T12:00:00Z", + }; + } + if (index === 1) { + return { + run_id: "run-initial", + seq, + content: { + type: "ai", + id: "history-answer", + content: "Initial design is ready", + }, + metadata: { caller: "lead_agent" }, + created_at: "2025-06-03T12:00:01Z", + }; + } + return { + run_id: "run-initial", + seq, + content: { + type: "ai", + id: `history-step-${seq}`, + content: `Historical presentation step ${seq}`, + ...(index === 49 + ? { additional_kwargs: { turn_duration: 704 } } + : {}), + }, + metadata: { caller: "lead_agent" }, + created_at: "2025-06-03T12:00:02Z", + }; + }); + const initialRows = Array.from({ length: 50 }, (_, index) => { + const seq = index + 51; + return { + run_id: "run-initial", + seq, + content: { + type: "ai", + id: `history-step-${seq}`, + content: `Historical presentation step ${seq}`, + }, + metadata: { caller: "lead_agent" }, + created_at: "2025-06-03T12:00:03Z", + }; + }); + const shiftedRows = Array.from({ length: 50 }, (_, index) => { + const seq = index + 101; + return { + run_id: "run-shifted", + seq, + content: { + type: "ai", + id: `shifted-step-${seq}`, + content: `New presentation step ${seq}`, + }, + metadata: { caller: "lead_agent" }, + created_at: "2025-06-03T12:01:00Z", + }; + }); + let latestPageRequestCount = 0; + let cursorPageRequestCount = 0; + + mockLangGraphAPI(page, { + threads: [ + { + thread_id: MOCK_THREAD_ID, + title: "Long presentation task", + updated_at: "2025-06-03T12:00:00Z", + // This scenario exercises persisted run-event pagination. Keep the + // checkpoint empty so its generic mock messages do not interfere + // with optimistic -> server reconciliation after the follow-up. + messages: [], + }, + ], + }); + await page.route( + new RegExp(`/api/threads/${MOCK_THREAD_ID}/messages/page(?:\\?.*)?$`), + async (route) => { + if (route.request().method() !== "GET") { + return route.fallback(); + } + + const beforeSeq = new URL(route.request().url()).searchParams.get( + "before_seq", + ); + const isLatestPage = beforeSeq === null; + const rows = isLatestPage + ? latestPageRequestCount === 0 + ? initialRows + : shiftedRows + : beforeSeq === "101" + ? initialRows + : olderRows; + const hasMore = isLatestPage || beforeSeq === "101"; + await route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify({ + data: rows, + has_more: hasMore, + next_before_seq: hasMore ? (rows[0]?.seq ?? null) : null, + }), + }); + if (isLatestPage) { + latestPageRequestCount += 1; + } else { + cursorPageRequestCount += 1; + } + }, + ); + + await page.goto(`/workspace/chats/${MOCK_THREAD_ID}`); + await expect + .poll(() => cursorPageRequestCount, { timeout: 15_000 }) + .toBeGreaterThan(0); + await expect(page.getByText(originalPrompt)).toBeVisible({ + timeout: 15_000, + }); + await expect(page.getByText("Completed in 11m 44s")).toBeVisible(); + + const latestPageRequestsBeforeSubmit = latestPageRequestCount; + const textarea = page.locator("textarea[name='message']"); + await textarea.fill(followUpPrompt); + await textarea.press("Enter"); + + await expect + .poll(() => latestPageRequestCount, { timeout: 15_000 }) + .toBeGreaterThan(latestPageRequestsBeforeSubmit); + await expect(page.getByText(originalPrompt)).toBeVisible(); + await expect(page.getByText(followUpPrompt)).toBeVisible(); + await expect(page.getByText("Completed in 11m 44s")).toBeVisible(); + + const originalBox = await page.getByText(originalPrompt).boundingBox(); + const followUpBox = await page.getByText(followUpPrompt).boundingBox(); + expect(originalBox).not.toBeNull(); + expect(followUpBox).not.toBeNull(); + expect(originalBox!.y).toBeLessThan(followUpBox!.y); + }); + test("shows a completed run duration once after multi-step history", async ({ page, }) => { diff --git a/frontend/tests/unit/core/threads/message-merge.test.ts b/frontend/tests/unit/core/threads/message-merge.test.ts index 88c5f1270..643267efa 100644 --- a/frontend/tests/unit/core/threads/message-merge.test.ts +++ b/frontend/tests/unit/core/threads/message-merge.test.ts @@ -11,10 +11,13 @@ import { getSummarizationMiddlewareMessages, getThreadHistoryNextPageParam, getVisibleOptimisticMessages, + mergeRenderedMessageLedger, mergeTransientHistoryBridge, mergeTransientHistoryBridgeOrder, mergeMessages, + parseThreadMessagesPageResponse, pruneConfirmedTransientMessages, + reconcileThreadHistoryRows, removeSetItems, resolveThreadTransientHistoryBridge, resolveTransientHistoryBridge, @@ -22,10 +25,10 @@ import { } from "@/core/threads/hooks"; import type { RunMessage } from "@/core/threads/types"; -function runMessage(seq?: number): RunMessage { +function runMessage(seq: number): RunMessage { return { run_id: "run-1", - ...(seq === undefined ? {} : { seq }), + seq, content: {} as Message, metadata: { caller: "" }, created_at: "2026-05-22T00:00:00Z", @@ -108,6 +111,7 @@ test("mergeMessages preserves historical run metadata on a live checkpoint repla [ { run_id: "run-1", + seq: 1, content: persistedAi, metadata: { caller: "lead_agent" }, created_at: "2026-07-21T00:00:00Z", @@ -496,6 +500,54 @@ test("buildThreadMessagesPageUrl returns a relative URL behind nginx", () => { ); }); +test("parseThreadMessagesPageResponse accepts a valid history page", () => { + const response = { + data: [runMessage(1), runMessage(2)], + has_more: true, + next_before_seq: 1, + }; + + expect(parseThreadMessagesPageResponse(response)).toBe(response); +}); + +test.each([ + ["missing", undefined], + ["non-numeric", "2"], + ["fractional", 2.5], + ["unsafe", Number.MAX_SAFE_INTEGER + 1], +])( + "parseThreadMessagesPageResponse rejects a %s row seq", + (_description, seq) => { + expect(() => + parseThreadMessagesPageResponse({ + data: [{ ...runMessage(1), seq }], + has_more: false, + next_before_seq: null, + }), + ).toThrow("invalid seq"); + }, +); + +test("parseThreadMessagesPageResponse rejects duplicate row seq values", () => { + expect(() => + parseThreadMessagesPageResponse({ + data: [runMessage(1), runMessage(1)], + has_more: false, + next_before_seq: null, + }), + ).toThrow("duplicate seq"); +}); + +test("parseThreadMessagesPageResponse rejects an invalid pagination cursor", () => { + expect(() => + parseThreadMessagesPageResponse({ + data: [runMessage(1)], + has_more: true, + next_before_seq: null, + }), + ).toThrow("invalid next_before_seq"); +}); + test("flattenThreadHistoryPages prepends backward pages in global seq order", () => { expect( flattenThreadHistoryPages([ @@ -537,6 +589,96 @@ test("flattenThreadHistoryPages retains backward pages when the latest page refr ).toEqual([1, 2, 3, 4, 5]); }); +test("reconcileThreadHistoryRows retains rows displaced from a moving latest page", () => { + const previousRows = Array.from({ length: 50 }, (_, index) => + runMessage(index + 1), + ); + const currentRows = Array.from({ length: 50 }, (_, index) => + runMessage(index + 51), + ); + + expect( + reconcileThreadHistoryRows(previousRows, currentRows, false).map( + (message) => message.seq, + ), + ).toEqual(Array.from({ length: 100 }, (_, index) => index + 1)); +}); + +test("reconcileThreadHistoryRows refreshes overlapping rows without moving them", () => { + const stale = runMessage(50); + const refreshed = { + ...runMessage(50), + content: { + id: "message-50", + type: "ai", + content: "final response", + additional_kwargs: { turn_duration: 704 }, + } as Message, + }; + + const reconciled = reconcileThreadHistoryRows( + [runMessage(49), stale], + [refreshed, runMessage(51)], + false, + ); + + expect(reconciled.map((message) => message.seq)).toEqual([49, 50, 51]); + expect(reconciled[1]).toBe(refreshed); +}); + +test("reconcileThreadHistoryRows trusts a complete snapshot and prunes missing rows", () => { + const currentRows = [runMessage(2), runMessage(3)]; + + expect( + reconcileThreadHistoryRows( + [runMessage(1), ...currentRows], + currentRows, + true, + ), + ).toEqual(currentRows); +}); + +test("reconcileThreadHistoryRows preserves existing history when a row seq is invalid", () => { + const previousRows = [runMessage(1), runMessage(2)]; + const invalidRow = { + ...runMessage(3), + seq: undefined, + } as unknown as RunMessage; + const consoleError = rs + .spyOn(console, "error") + .mockImplementation(() => undefined); + + expect(reconcileThreadHistoryRows(previousRows, [invalidRow], false)).toBe( + previousRows, + ); + expect(consoleError).toHaveBeenCalledOnce(); + + consoleError.mockRestore(); +}); + +test("reconcileThreadHistoryRows keeps insertion order on an invalid initial snapshot", () => { + const first = { + ...runMessage(1), + seq: undefined, + content: { id: "first", type: "human", content: "first" } as Message, + } as unknown as RunMessage; + const second = { + ...runMessage(2), + seq: undefined, + content: { id: "second", type: "ai", content: "second" } as Message, + } as unknown as RunMessage; + const consoleError = rs + .spyOn(console, "error") + .mockImplementation(() => undefined); + + const currentRows = [first, second]; + + expect(reconcileThreadHistoryRows([], currentRows, false)).toBe(currentRows); + expect(consoleError).toHaveBeenCalledOnce(); + + consoleError.mockRestore(); +}); + test("infinite history refetch recalculates older-page cursors from the refreshed newest page", async () => { const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } }, @@ -651,24 +793,28 @@ test("buildVisibleHistoryMessages filters superseded runs but keeps regenerated const rows: RunMessage[] = [ { run_id: "run-old", + seq: 1, content: oldHuman, metadata: { caller: "lead_agent" }, created_at: "2026-06-18T00:00:00Z", }, { run_id: "run-old", + seq: 2, content: oldAi, metadata: { caller: "lead_agent" }, created_at: "2026-06-18T00:00:01Z", }, { run_id: "run-new", + seq: 3, content: newHuman, metadata: { caller: "lead_agent" }, created_at: "2026-06-18T00:00:02Z", }, { run_id: "run-new", + seq: 4, content: newAi, metadata: { caller: "lead_agent" }, created_at: "2026-06-18T00:00:03Z", @@ -687,6 +833,7 @@ test("buildVisibleHistoryMessages attaches run_id to each content message (#3779 const rows: RunMessage[] = [ { run_id: "run-1", + seq: 1, content: { id: "ai-1", type: "ai", content: "answer" } as Message, metadata: { caller: "lead_agent" }, created_at: "2026-06-26T00:00:00Z", @@ -821,6 +968,53 @@ test("resolveTransientHistoryBridge does not collapse an unloaded gap before its ).toEqual(["event-seq-35", "event-seq-88"]); }); +test("resolveTransientHistoryBridge restores a prefix that was already rendered", () => { + const protectedPrompt = { + id: "protected-prompt", + type: "human", + content: "/ppt-master 帮我做个ppt", + } as Message; + const intermediateReply = { + id: "intermediate-reply", + type: "ai", + content: "正在生成页面", + } as Message; + const pageAnchor = { + id: "page-anchor", + type: "ai", + content: "继续导出 SVG", + } as Message; + const latestAnswer = { + id: "latest-answer", + type: "ai", + content: "任务仍在执行", + } as Message; + const captured = [protectedPrompt, intermediateReply, pageAnchor]; + const bridgeOrder = mergeTransientHistoryBridgeOrder([], captured); + const previouslyRenderedOrder = [ + protectedPrompt, + intermediateReply, + pageAnchor, + latestAnswer, + ] + .map((message) => `message:${message.id}`) + .filter((identity): identity is string => Boolean(identity)); + + expect( + resolveTransientHistoryBridge( + [pageAnchor, latestAnswer], + captured, + bridgeOrder, + previouslyRenderedOrder, + ).map((message) => message.id), + ).toEqual([ + "protected-prompt", + "intermediate-reply", + "page-anchor", + "latest-answer", + ]); +}); + test("resolveTransientHistoryBridge does not duplicate once canonical history catches up", () => { expect( resolveTransientHistoryBridge( @@ -1093,6 +1287,189 @@ test("computeSummarizationTransientMessages captures live turns dropped before t ).toEqual([summarizationHuman1, summarizationAi1, summarizationHuman2]); }); +test("computeSummarizationTransientMessages rescues rendered processing steps missing from a stale live snapshot", () => { + const processingMessages = Array.from({ length: 10 }, (_, index) => { + const step = index + 1; + return { + id: `processing-${step}`, + type: step % 2 === 0 ? "tool" : "ai", + ...(step % 2 === 0 ? { tool_call_id: `call-${step - 1}` } : {}), + content: `step ${step}`, + } as Message; + }); + // The SDK has already advanced to the 10-message post-compaction window, + // while the UI's previous committed frame still contains steps 1..10. + const staleLiveSnapshot = processingMessages.slice(4); + const summarizationMessages = [ + { + id: "__remove_all__", + type: "remove", + content: "", + } as Message, + ...staleLiveSnapshot, + ]; + + expect( + computeSummarizationTransientMessages( + staleLiveSnapshot, + summarizationMessages, + new Set(), + processingMessages, + ), + ).toEqual(processingMessages.slice(0, 4)); +}); + +test("computeSummarizationTransientMessages rescues steps between a protected input and retained tail", () => { + const protectedInput = { + id: "protected-input", + type: "human", + content: "Run a long sequential research task", + } as Message; + const removedSteps = Array.from({ length: 4 }, (_, index) => ({ + id: `protected-window-step-${index + 1}`, + type: "ai", + content: `completed ${index + 1}`, + })) as Message[]; + const retainedTail = { + id: "retained-tail", + type: "tool", + tool_call_id: "retained-call", + content: "latest search result", + } as Message; + const renderedMessages = [protectedInput, ...removedSteps, retainedTail]; + const retainedWindow = [protectedInput, retainedTail]; + + expect( + computeSummarizationTransientMessages( + retainedWindow, + [ + { + id: "__remove_all__", + type: "remove", + content: "", + } as Message, + ...retainedWindow, + ], + new Set(), + renderedMessages, + ), + ).toEqual(removedSteps); +}); + +test("repeated compaction keeps every previously rendered processing step in order", () => { + const processingMessages = Array.from({ length: 12 }, (_, index) => { + const step = index + 1; + return { + id: `repeat-processing-${step}`, + type: step % 2 === 0 ? "tool" : "ai", + ...(step % 2 === 0 ? { tool_call_id: `repeat-call-${step - 1}` } : {}), + content: `step ${step}`, + } as Message; + }); + const removeAll = { + id: "__remove_all__", + type: "remove", + content: "", + } as Message; + + const firstTail = processingMessages.slice(4, 10); + const firstMoved = computeSummarizationTransientMessages( + firstTail, + [removeAll, ...firstTail], + new Set(), + processingMessages.slice(0, 10), + ); + const firstBridge = mergeTransientHistoryBridge([], firstMoved); + const firstMerged = mergeMessages(firstBridge, firstTail, []); + + const secondTail = processingMessages.slice(6); + const secondMoved = computeSummarizationTransientMessages( + secondTail, + [removeAll, ...secondTail], + new Set(), + processingMessages, + ); + const secondBridge = mergeTransientHistoryBridge(firstBridge, secondMoved); + const secondMerged = mergeMessages(secondBridge, secondTail, []); + + expect(firstMerged.map((message) => message.id)).toEqual( + processingMessages.slice(0, 10).map((message) => message.id), + ); + expect(secondMerged.map((message) => message.id)).toEqual( + processingMessages.map((message) => message.id), + ); +}); + +test("rendered message ledger survives rolling live windows before repeated compaction", () => { + const processingMessages = Array.from({ length: 23 }, (_, index) => { + const step = index + 1; + return { + id: `rolling-processing-${step}`, + type: "ai", + content: `completed ${step}/26`, + } as Message; + }); + const firstVisibleWindow = processingMessages.slice(0, 13); + const secondVisibleWindow = processingMessages.slice(8, 18); + const thirdVisibleWindow = processingMessages.slice(13); + + const firstLedger = mergeRenderedMessageLedger([], firstVisibleWindow); + const secondLedger = mergeRenderedMessageLedger( + firstLedger, + secondVisibleWindow, + ); + const thirdLedger = mergeRenderedMessageLedger( + secondLedger, + thirdVisibleWindow, + ); + + expect(thirdLedger.map((message) => message.id)).toEqual( + processingMessages.map((message) => message.id), + ); + + // A later compaction retains only 19..23. The accumulated display ledger + // must still supply 14..18 (and every older displayed step) to the bridge. + const retainedTail = processingMessages.slice(18); + const moved = computeSummarizationTransientMessages( + retainedTail, + [ + { + id: "__remove_all__", + type: "remove", + content: "", + } as Message, + ...retainedTail, + ], + new Set(), + thirdLedger, + ); + + expect(moved.map((message) => message.id)).toEqual( + processingMessages.slice(0, 18).map((message) => message.id), + ); +}); + +test("rendered message ledger does not retain explicitly superseded messages", () => { + const retained = { + id: "retained-answer", + type: "ai", + content: "keep me", + } as Message; + const superseded = { + id: "superseded-answer", + type: "ai", + content: "replace me", + } as Message; + + expect( + mergeRenderedMessageLedger( + [retained, superseded], + [retained], + new Set([superseded.id!]), + ), + ).toEqual([retained]); +}); + test("computeSummarizationTransientMessages excludes already-summarized control messages", () => { const priorSummary = { id: "summary-0",