diff --git a/README.md b/README.md index aff33b3795..7be0c0bc7e 100644 --- a/README.md +++ b/README.md @@ -1634,6 +1634,11 @@ Content-less sub-agent final messages report `No response generated` instead of An ordinary `task` also receives a defensive snapshot of the dispatching run's current uploads. This lets eligible sub-agents use `list_uploaded_files` to find earlier-turn files without returning same-turn attachments as historical. Delayed or recovered `batch_task` workers leave this tool disabled because they have no valid turn-local upload boundary. +历史上传发现支持稳定续页：`list_uploaded_files` 默认每页 20 项、最多 100 项， +将返回的 `next_cursor` 作为下一次调用的 `cursor`，保留相同过滤条件，直到末页不再 +返回游标。目录元数据或调用上下文变化时会明确要求重新枚举；详情见 +[文件上传文档](backend/docs/FILE_UPLOAD.md)。 + Durable `batch_task` workers use one app-owned plugin snapshot for tool assembly and execution. Recovered tasks adopt the new worker's plugin snapshot after a Gateway restart; plugin objects are never stored in durable task records. Ordinary `task` delegation and explicit durable `batch_task` execution share the startup-scoped `subagent_runtime` process capacity. Batch mode keeps large independent item sets in SQL with separate total, live, and running limits, restart recovery, bounded results, and a thread-scoped Web UI panel. The panel pages through bounded previews on demand; full stored result text is available only through the owner-scoped JSONL export, while internal execution and authorization context never enters owner-facing responses. If the batch worker is later stopped or disabled, threads with persisted batches retain read-only item inspection and JSONL export; execution controls remain disabled until the worker is running again. See `config.example.yaml` and [the implementation contract](docs/plans/2026-08-24-subagent-batch-capacity-implementation.md) for limits and recovery semantics. diff --git a/backend/docs/FILE_UPLOAD.md b/backend/docs/FILE_UPLOAD.md index 2e878c0d01..6c55bd97c3 100644 --- a/backend/docs/FILE_UPLOAD.md +++ b/backend/docs/FILE_UPLOAD.md @@ -141,6 +141,23 @@ To work with these files: `extensions` 按类型过滤；过滤发生在默认 20 条上限之前）。如果已知文件名，也可直接使用 `read_file` 或 `grep` 访问 `/mnt/user-data/uploads/` 下的文件。 +历史上传支持有界续页：`max_results` 默认 20、每页最多 100。 +返回 `next_cursor` 时，将其作为下一次调用的 `cursor`，并保留相同的 +`query` / `extensions`；末页没有 `next_cursor`。可在续页时调整每页数量和 +`include_outline`，大纲只针对当前页提取。结果按修改时间倒序、同时间按原始文件名 +排序；`total_count` 是完整过滤结果数，`omitted_summary` 只统计当前页之后剩余的文件。 + +例如 250 个匹配附件可按 100 → 100 → 50 枚举。每页都排除本轮上传、staging、 +符号链接以及现有规则识别的转换 companion。规范化后的过滤条件、用户、线程、 +本轮上传排除集合或目录中的普通文件清单元数据改变时，旧游标返回 +`error: stale_cursor`；畸形或过长游标返回 `error: invalid_cursor`。 +两种情况都有 `restart_required: true`，应丢弃此前收集的页，省略 `cursor` +重新开始，避免把两次不同枚举混合起来。 + +游标仅用于一致性校验，不是授权凭据；工具仍从可信 runtime 解析当前用户和线程。 +每页重新扫描目录，校验文件名、大小及纳秒级修改/变更时间，不保存持久快照， +不保证文件字节不变或扫描期间的原子快照，也不限制任意大目录的扫描开销。 + ### 使用上传的文件 Agent 在沙箱中运行，使用虚拟路径访问文件。Agent 可以直接使用 `read_file` 工具读取上传的文件： diff --git a/backend/packages/harness/deerflow/tools/AGENTS.md b/backend/packages/harness/deerflow/tools/AGENTS.md index 4e4d2ad0c0..23d4af91d6 100644 --- a/backend/packages/harness/deerflow/tools/AGENTS.md +++ b/backend/packages/harness/deerflow/tools/AGENTS.md @@ -1,5 +1,11 @@ ### Tool System (`packages/harness/deerflow/tools/`) +`list_uploaded_files` 的续页契约见 [FILE_UPLOAD.md](../../../../docs/FILE_UPLOAD.md)。 +游标绑定可信用户/线程、规范化过滤条件、本轮上传排除集合和目录元数据；身份与目录 +始终由 runtime 解析，游标只负责一致性校验。先过滤，再按修改时间降序和原始文件名 +排序分页；失效返回 `restart_required`，不得静默回到第一页或伪报末页。 +页大小及大纲选项不参与清单绑定；`total_count` 表示完整过滤结果，摘要只统计剩余项。 + `conversation.py` supplies the optional `read_conversation` tool. Ordinary lead assembly opts in only with a host reader; default, bootstrap, embedded and subagent assembly withhold it. The tool requires the worker-owned diff --git a/backend/packages/harness/deerflow/tools/builtins/list_uploaded_files_tool.py b/backend/packages/harness/deerflow/tools/builtins/list_uploaded_files_tool.py index a3ff64b1ce..b73110b858 100644 --- a/backend/packages/harness/deerflow/tools/builtins/list_uploaded_files_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/list_uploaded_files_tool.py @@ -6,10 +6,14 @@ this tool lets the agent discover files uploaded in previous turns on demand. from __future__ import annotations +import hashlib +import json import logging import os +import re from collections import Counter from pathlib import Path +from stat import S_ISREG from typing import Annotated, Any from langchain.tools import tool @@ -26,6 +30,23 @@ logger = logging.getLogger(__name__) _DEFAULT_MAX_RESULTS = 20 _MAX_MAX_RESULTS = 100 +_CURSOR_PATTERN = re.compile(r"v1\.([1-9][0-9]{0,19})\.([0-9a-f]{64})", re.ASCII) + + +def _cursor_error(code: str) -> dict: + """用固定消息返回续页失败，避免回显游标或主机路径。""" + return { + "files": [], + "error": code, + "restart_required": True, + "message": "Upload listing continuation is invalid or has changed. Restart without cursor using the desired filters; discard earlier pages before combining results.", + } + + +def _make_cursor(offset: int, revision: str) -> str: + """把页位置绑定到清单及调用上下文；此校验值不承担授权职责。""" + digest = hashlib.sha256(f"{offset}:{revision}".encode("ascii")).hexdigest() + return f"v1.{offset}.{digest}" def _extension_label(file_path: Path) -> str: @@ -114,9 +135,16 @@ def _list_uploaded_files_impl( *, query: str | None = None, extensions: list[str] | None = None, + cursor: str | None = None, _paths: Any | None = None, ) -> dict: """Core implementation — testable without the @tool wrapper.""" + offset = 0 + if cursor is not None: + if not isinstance(cursor, str) or len(cursor) > 88 or (match := _CURSOR_PATTERN.fullmatch(cursor)) is None: + return _cursor_error("invalid_cursor") + offset = int(match[1]) + if runtime is None: return {"files": [], "message": "No runtime context available."} @@ -129,6 +157,8 @@ def _list_uploaded_files_impl( uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=user_id) if not uploads_dir.exists(): + if cursor is not None: + return _cursor_error("stale_cursor") return {"files": [], "message": "No uploads directory for this thread."} # Resolve the set of filenames uploaded in the current run so we can exclude them. @@ -138,9 +168,12 @@ def _list_uploaded_files_impl( uploaded = state.get("uploaded_files") if isinstance(state, dict) else getattr(state, "uploaded_files", None) if isinstance(uploaded, list): for entry in uploaded: - if isinstance(entry, dict) and entry.get("filename"): - current_run_filenames.add(entry["filename"]) + filename = entry.get("filename") if isinstance(entry, dict) else None + if isinstance(filename, str) and filename: + current_run_filenames.add(filename) except Exception: + if cursor is not None: + return _cursor_error("stale_cursor") logger.warning( "Failed to read uploaded_files from runtime.state; current-run files may appear in list_uploaded_files results", exc_info=True, @@ -159,13 +192,19 @@ def _list_uploaded_files_impl( # Collect historical files (sorted by mtime descending). # Skip .md files that are conversion artifacts (have a same-stem non-.md sibling). - candidates: list[tuple[float, Path, int]] = [] + candidates: list[tuple[int, Path, int]] = [] + listing_metadata: list[tuple[str, int, int, int]] = [] try: # Collect file entries once to build the name set and iterate. - entries = [e for e in os.scandir(uploads_dir) if e.is_file() and not e.is_symlink() and not is_upload_staging_file(e.name)] + with os.scandir(uploads_dir) as scan: + entries = [e for e in scan if e.is_file(follow_symlinks=False) and not is_upload_staging_file(e.name)] all_names: set[str] = {e.name for e in entries} for entry in entries: + stat = entry.stat(follow_symlinks=False) + if not S_ISREG(stat.st_mode): + continue + listing_metadata.append((entry.name, stat.st_mtime_ns, stat.st_ctime_ns, stat.st_size)) if entry.name in current_run_filenames: continue # Skip .md files that are conversion artifacts of another file. @@ -178,32 +217,51 @@ def _list_uploaded_files_impl( non_md_siblings = {n for n in all_names if n != entry.name and Path(n).stem == stem} if non_md_siblings: continue - stat = entry.stat() - candidates.append((stat.st_mtime, Path(entry.path), stat.st_size)) + candidates.append((stat.st_mtime_ns, Path(entry.path), stat.st_size)) except OSError: - return {"files": [], "message": f"Failed to read uploads directory: {uploads_dir}"} - - if not candidates: - return {"files": [], "message": "No historical uploaded files in this thread."} + if cursor is not None: + return _cursor_error("stale_cursor") + return {"files": [], "message": "Failed to read uploads directory."} query_filter = _normalize_query(query) extension_filter = _normalize_extensions(extensions) + # 绑定原始文件名和元数据，不把标识或路径编码到模型可见的游标中。 + revision = hashlib.sha256( + json.dumps( + [str(uploads_dir), user_id, thread_id, query_filter.casefold() if query_filter else None, sorted(extension_filter or ()), sorted(current_run_filenames), sorted(listing_metadata)], + ensure_ascii=True, + separators=(",", ":"), + ).encode("ascii") + ).hexdigest() + if cursor is not None and cursor != _make_cursor(offset, revision): + return _cursor_error("stale_cursor") + + if not candidates: + if cursor is not None: + return _cursor_error("stale_cursor") + return {"files": [], "message": "No historical uploaded files in this thread."} + if query_filter is not None or extension_filter is not None: candidates = [item for item in candidates if _matches_filters(item[1].name, item[1].suffix, query_filter, extension_filter)] if not candidates: + if cursor is not None: + return _cursor_error("stale_cursor") return { "files": [], "total_count": 0, "message": "No uploaded files matched the given filters.", } - # Sort by mtime descending (most recent first) - candidates.sort(key=lambda item: item[0], reverse=True) + # 同一时间戳按原始文件名排序，避免依赖 scandir 的不稳定顺序。 + candidates.sort(key=lambda item: (-item[0], item[1].name)) total_count = len(candidates) - truncated = total_count > max_results - visible = candidates[:max_results] - omitted_paths = [p.name for _, p, _ in candidates[max_results:]] + if cursor is not None and offset >= total_count: + return _cursor_error("stale_cursor") + page_end = offset + max_results + truncated = total_count > page_end + visible = candidates[offset:page_end] + omitted_paths = [p.name for _, p, _ in candidates[page_end:]] files: list[dict] = [] for _, file_path, st_size in visible: @@ -233,6 +291,7 @@ def _list_uploaded_files_impl( if truncated: result["truncated"] = True result["omitted_summary"] = _format_omitted_summary(omitted_paths) + result["next_cursor"] = _make_cursor(page_end, revision) if files: result["message"] = f"Found {total_count} historical file(s)." @@ -264,6 +323,10 @@ def list_uploaded_files( list[str] | None, 'Optional file extensions to keep, e.g. ["pdf", ".PNG"]. With or without a leading dot; matching is case-insensitive. Combined with query using AND. Omit to skip type filtering.', ] = None, + cursor: Annotated[ + str | None, + "The next_cursor returned by the previous page; keep the same query/extensions when continuing. Omit for the first page. On restart_required, discard previously collected pages and restart from the first page.", + ] = None, ) -> dict: """Discover historical uploaded files available in this thread. @@ -281,6 +344,11 @@ def list_uploaded_files( Optional filters (`query`, `extensions`) run before the max_results cap, so older matching files are not displaced by newer unrelated uploads. + + Continue with next_cursor when present; the last page has no next_cursor. + Changes to directory metadata, user, thread, filters, or current-run upload + exclusions invalidate the cursor. Each page contains at most 100 files; + this enumerates a listing, not a snapshot of file contents. """ return _list_uploaded_files_impl( include_outline=include_outline, @@ -288,4 +356,5 @@ def list_uploaded_files( runtime=runtime, query=query, extensions=extensions, + cursor=cursor, ) diff --git a/backend/tests/test_list_uploaded_files_tool.py b/backend/tests/test_list_uploaded_files_tool.py index 04fdb5f069..9af3dea0bc 100644 --- a/backend/tests/test_list_uploaded_files_tool.py +++ b/backend/tests/test_list_uploaded_files_tool.py @@ -978,7 +978,7 @@ class TestToolSchema: def test_runtime_excluded_from_model_facing_args(self): from deerflow.tools.builtins.list_uploaded_files_tool import list_uploaded_files - assert set(list_uploaded_files.args) == {"include_outline", "max_results", "query", "extensions"} + assert set(list_uploaded_files.args) == {"include_outline", "max_results", "query", "extensions", "cursor"} assert "runtime" not in list_uploaded_files.args def test_openai_schema_generation_succeeds(self): @@ -989,4 +989,190 @@ class TestToolSchema: # This raised PydanticInvalidForJsonSchema before the fix. oai = convert_to_openai_tool(list_uploaded_files) params = oai["function"]["parameters"]["properties"] - assert set(params) == {"include_outline", "max_results", "query", "extensions"} + assert set(params) == {"include_outline", "max_results", "query", "extensions", "cursor"} + + +class TestUploadPagination: + """通过工具入口验证稳定枚举、隔离和失效语义。""" + + def test_250_files_have_bounded_complete_pages(self, tmp_path): + directory = _uploads_dir(tmp_path) + expected = [f"report-{i:03}.pdf" for i in range(250)] + for name in reversed(expected): + path = directory / name + path.write_bytes(b"pdf") + os.utime(path, ns=(1_000_000_000, 1_000_000_000)) + kwargs = dict(runtime=_runtime(), _paths=_paths(tmp_path), max_results=100, extensions=["pdf"]) + first = _list_uploaded_files_impl(**kwargs) + second = _list_uploaded_files_impl(**kwargs, cursor=first["next_cursor"]) + assert second == _list_uploaded_files_impl(**kwargs, cursor=first["next_cursor"]) + third = _list_uploaded_files_impl(**kwargs, cursor=second["next_cursor"]) + assert [len(p["files"]) for p in [first, second, third]] == [100, 100, 50] + assert [f["filename"] for p in [first, second, third] for f in p["files"]] == expected + assert all(p["total_count"] == 250 for p in [first, second, third]) + assert first["omitted_summary"] == "150 .pdf" + assert second["omitted_summary"] == "50 .pdf" + assert "next_cursor" not in third + assert "truncated" not in third + + @pytest.mark.parametrize("mutation", ["add", "delete", "rename", "mtime", "size", "empty", "remove_directory"]) + def test_directory_changes_invalidate_cursor(self, tmp_path, mutation): + directory = _uploads_dir(tmp_path) + for name in ["a.txt", "b.txt"]: + (directory / name).write_text("old", encoding="utf-8") + kwargs = dict(runtime=_runtime(), _paths=_paths(tmp_path), max_results=1) + first = _list_uploaded_files_impl(**kwargs) + path = directory / "a.txt" + if mutation == "add": + (directory / "c.txt").write_text("new", encoding="utf-8") + elif mutation == "delete": + path.unlink() + elif mutation == "rename": + path.rename(directory / "c.txt") + elif mutation == "mtime": + os.utime(path, ns=(1_000_000_000, 1_000_000_000)) + elif mutation == "size": + path.write_text("longer", encoding="utf-8") + else: + for child in directory.iterdir(): + child.unlink() + if mutation == "remove_directory": + directory.rmdir() + result = _list_uploaded_files_impl(**kwargs, cursor=first["next_cursor"]) + assert result["files"] == [] + assert result["error"] == "stale_cursor" + assert result["restart_required"] is True + assert "next_cursor" not in result + assert str(directory) not in str(result) + + @pytest.mark.parametrize("change", ["query", "extensions", "excluded", "thread", "owner"]) + def test_context_changes_reject_cursor(self, tmp_path, change): + directory = _uploads_dir(tmp_path) + for name in ["a.txt", "b.txt", "c.pdf"]: + (directory / name).touch() + rt = _runtime() + paths = MagicMock() + paths.sandbox_uploads_dir.return_value = directory + kwargs = dict(runtime=rt, _paths=paths, max_results=1) + first = _list_uploaded_files_impl(**kwargs) + if change == "query": + kwargs["query"] = "a" + elif change == "extensions": + kwargs["extensions"] = ["txt"] + elif change == "excluded": + rt.state["uploaded_files"] = [{"filename": "unrelated.txt"}] + elif change == "thread": + rt.context["thread_id"] = "other-thread" + else: + rt.context["user_id"] = "other-owner" + result = _list_uploaded_files_impl(**kwargs, cursor=first["next_cursor"]) + assert result["files"] == [] + assert result["error"] == "stale_cursor" + + def test_changed_position_and_digest_are_rejected(self, tmp_path): + directory = _uploads_dir(tmp_path) + for name in ["a.txt", "b.txt", "c.txt"]: + (directory / name).touch() + kwargs = dict(runtime=_runtime(), _paths=_paths(tmp_path), max_results=1) + cursor = _list_uploaded_files_impl(**kwargs)["next_cursor"] + version, position, digest = cursor.split(".") + for changed in [f"{version}.2.{digest}", f"{version}.{position}.{'0' * 64}"]: + result = _list_uploaded_files_impl(**kwargs, cursor=changed) + assert result["files"] == [] + assert result["error"] == "stale_cursor" + assert len(cursor) <= 88 + assert str(directory) not in cursor + assert "thread-abc" not in cursor + + def test_page_size_stays_capped_on_continuation(self, tmp_path): + directory = _uploads_dir(tmp_path) + for i in range(125): + (directory / f"{i:03}.txt").touch() + kwargs = dict(runtime=_runtime(), _paths=_paths(tmp_path)) + first = _list_uploaded_files_impl(**kwargs) + assert len(first["files"]) == 20 + second = _list_uploaded_files_impl(**kwargs, max_results=10000, cursor=first["next_cursor"]) + assert len(second["files"]) == 100 + last = _list_uploaded_files_impl(**kwargs, cursor=second["next_cursor"]) + assert len(last["files"]) == 5 + assert "next_cursor" not in last + + def test_tool_wrapper_passes_cursor(self, tmp_path, monkeypatch): + from deerflow.tools.builtins.list_uploaded_files_tool import list_uploaded_files + + directory = _uploads_dir(tmp_path) + for name in ["a.txt", "b.txt"]: + (directory / name).touch() + monkeypatch.setattr("deerflow.tools.builtins.list_uploaded_files_tool.get_paths", lambda: _paths(tmp_path)) + kwargs = dict(runtime=_runtime(), max_results=1) + first = list_uploaded_files.func(**kwargs) + second = list_uploaded_files.func(**kwargs, cursor=first["next_cursor"]) + assert first["files"][0]["filename"] != second["files"][0]["filename"] + assert "next_cursor" not in second + + def test_malformed_upload_entries_do_not_break_revision(self, tmp_path): + directory = _uploads_dir(tmp_path) + for name in ["a.txt", "b.txt", "fresh.txt"]: + (directory / name).touch() + rt = _runtime(state_uploaded=[{"filename": 5}, {"filename": ["invalid"]}, {"filename": "fresh.txt"}]) + kwargs = dict(runtime=rt, _paths=_paths(tmp_path), max_results=1) + first = _list_uploaded_files_impl(**kwargs) + second = _list_uploaded_files_impl(**kwargs, cursor=first["next_cursor"]) + assert {f["filename"] for page in [first, second] for f in page["files"]} == {"a.txt", "b.txt"} + + def test_directory_read_error_does_not_leak_host_path(self, tmp_path, monkeypatch): + directory = _uploads_dir(tmp_path) + for name in ["a.txt", "b.txt"]: + (directory / name).touch() + kwargs = dict(runtime=_runtime(), _paths=_paths(tmp_path), max_results=1) + cursor = _list_uploaded_files_impl(**kwargs)["next_cursor"] + monkeypatch.setattr("deerflow.tools.builtins.list_uploaded_files_tool.os.scandir", MagicMock(side_effect=PermissionError(str(directory)))) + result = _list_uploaded_files_impl(**kwargs, cursor=cursor) + assert result["error"] == "stale_cursor" + assert result["files"] == [] + assert str(directory) not in str(result) + + @pytest.mark.parametrize("cursor", ["", "invalid", "v2.1." + "a" * 64, "v1.-1." + "a" * 64, "v1.0." + "a" * 64, "x" * 10000, 5, {}, "oops"]) + def test_malformed_cursor_is_controlled(self, tmp_path, cursor): + _uploads_dir(tmp_path) + result = _list_uploaded_files_impl(runtime=_runtime(), _paths=_paths(tmp_path), cursor=cursor) + assert result["files"] == [] + assert result["error"] == "invalid_cursor" + assert result["restart_required"] is True + + def test_normalized_filters_and_page_size_change(self, tmp_path): + directory = _uploads_dir(tmp_path) + for i in range(5): + (directory / f"Report-{i}.pdf").touch() + kwargs = dict(runtime=_runtime(), _paths=_paths(tmp_path)) + first = _list_uploaded_files_impl(**kwargs, query=" REPORT ", extensions=["*.PDF"], max_results=1) + second = _list_uploaded_files_impl(**kwargs, query="report", extensions=["pdf", ".PDF"], max_results=1000, cursor=first["next_cursor"]) + assert len(second["files"]) == 4 + assert {f["filename"] for f in first["files"]}.isdisjoint(f["filename"] for f in second["files"]) + assert "next_cursor" not in second + + def test_exclusions_and_outlines_apply_on_every_page(self, tmp_path, monkeypatch): + directory = _uploads_dir(tmp_path) + for name in ["a.pdf", "a.md", "b.txt", "c.txt", "fresh.txt", ".upload-hidden.part"]: + (directory / name).touch() + (directory / "folder").mkdir() + try: + (directory / "link.txt").symlink_to(directory / "b.txt") + except OSError: + pytest.skip("当前平台无法创建符号链接") + outline = MagicMock(return_value=([{"title": "heading", "line": 1}], [])) + monkeypatch.setattr("deerflow.tools.builtins.list_uploaded_files_tool.extract_outline_for_file", outline) + kwargs = dict(runtime=_runtime(state_uploaded=[{"filename": "fresh.txt"}]), _paths=_paths(tmp_path), max_results=1, include_outline=True) + names = [] + cursor = None + while True: + page = _list_uploaded_files_impl(**kwargs, cursor=cursor) + assert len(page["files"]) == 1 + assert page["total_count"] == 3 + names.append(page["files"][0]["filename"]) + cursor = page.get("next_cursor") + if cursor is None: + break + assert len(names) < 4 + assert set(names) == {"a.pdf", "b.txt", "c.txt"} + assert [call.args[0].name for call in outline.call_args_list] == names diff --git a/backend/tests/test_tool_args_schema_no_pydantic_warning.py b/backend/tests/test_tool_args_schema_no_pydantic_warning.py index 6c0af334ec..da2bd3c358 100644 --- a/backend/tests/test_tool_args_schema_no_pydantic_warning.py +++ b/backend/tests/test_tool_args_schema_no_pydantic_warning.py @@ -250,7 +250,7 @@ def test_list_uploaded_files_model_schema_excludes_injected_runtime() -> None: """The model-facing schema must not expose ToolRuntime internals.""" parameters = convert_to_openai_tool(list_uploaded_files)["function"]["parameters"] - assert set(parameters["properties"]) == {"include_outline", "max_results", "query", "extensions"} + assert set(parameters["properties"]) == {"include_outline", "max_results", "query", "extensions", "cursor"} @pytest.mark.parametrize("tool_obj", [case[0] for case in _TOOL_CASES], ids=[case[0].name for case in _TOOL_CASES])