feat(subagents): enable historical upload discovery (#5170)

Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>
Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>

README.md

@@ -1202,6 +1202,8 @@ Sub-agents are an optimization, not the default response to a complex request.
 
 The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions — when delegation has clear net benefit from real parallel latency, specialist capability, or context isolation. It keeps interdependent scopes and overlapping side effects out of parallel dispatch; a bounded sequential chain can still run in one sub-agent when specialist or context-isolation benefit clearly wins. The lead uses the fewest useful sub-agents and re-evaluates later batches instead of fanning out solely because a task is large or multi-step. Sub-agents report back structured results, and the lead agent verifies and synthesizes them into a coherent output. Deterministic tool receipts cover both direct tool messages and state-updating `Command` results such as delegated `task` responses; when the receipt ledger reaches its context budget, it retains the newest actions and their original receipt IDs. Operators can disable this provenance layer with `verification.receipts_enabled: false`. Their configured skills are resolved from the same user-scoped catalog as the lead agent, so user-owned custom skills remain available without exposing another user's version. Their internal AI and tool messages stay scoped to the delegated graph instead of entering the parent chat stream. Reloaded thread history enforces the same boundary: callback-captured sub-agent AI responses remain available in run-event diagnostics but are excluded from the parent transcript, while the parent `task` result remains attached to its subtask card. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. Concurrent parent runs also receive independent server-side sub-agent execution IDs, so a provider that reuses a tool-call ID cannot make one run poll, cancel, or clean up another run's background task. Collapsed sub-agent cards show the effective model and, when the provider returns usage metadata, a cumulative token total that updates after each completed sub-agent LLM call and persists after a reload. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step from that run's terminal tool-message metadata rather than a process-global provider-ID cache.
 
+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.
+
 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.
 
 Direct `create_deerflow_agent(...)` integrations can own the same boundary explicitly instead of relying on Gateway startup. Construct one `SubagentRuntime` and share it across every graph in that application; its `max_running`, ordinary per-run total, bound `task` tool, and optional durable-batch tools then use the same caller-owned snapshot and execution controller. A runtime with a batch repository owns a worker and must be started before graph construction and stopped during application shutdown:

backend/packages/harness/deerflow/subagents/AGENTS.md

@@ -4,6 +4,7 @@
 **Registry and managed definitions**: Runtime resolution is built-in → `config.yaml custom_agents` → enabled administrator-managed definitions, followed by explicit `subagents.agents.<name>` overrides. Managed definitions are deployment-wide, persist through the same `agent_storage.backend` selection as Custom Agent definitions, and remain stored but are excluded from runtime when a built-in or later-added config definition owns the same name. The default Lead Agent sees the whole enabled catalog. A Custom Agent's `allowed_subagents` is snapshotted into run metadata (`None` = all, `[]` = hard deny, list = allowlist) and must filter both prompt discovery and `task` execution; never reload caller policy from mutable agent config inside the tool.
 **Benefit-based routing policy**: Enabling subagents exposes delegation as an optimization, not a default response to complexity. The lead prompt defaults to direct execution and permits `task` only when parallel latency, specialist capability, or context-isolation benefit clearly exceeds startup, duplicate-discovery, synthesis, state-conflict, and side-effect costs. Inter-agent output dependencies and overlapping mutable state are hard vetoes for parallel dispatch, while duplicate discovery and a cheap direct path remain costs rather than categorical vetoes; a bounded sequential chain may run in one subagent when specialist or context-isolation benefit clearly wins. Parallel scopes must be independent and non-overlapping, the lead uses the fewest useful subagents, and every later batch is re-evaluated while retaining any within-batch parallel benefit. When the enforced per-response limit is 1, the rendered prompt removes parallel and multi-batch benefit guidance and permits delegation only for material specialist or context-isolation benefit. Keep this policy aligned across `lead_agent/prompt.py`, the `task` tool description, and both built-in role descriptions; routing regressions are pinned in `tests/test_subagent_routing_prompt.py`, `tests/test_subagent_prompt_security.py`, and `tests/test_lead_agent_prompt.py`.
 **User-scoped Skills**: Subagents resolve their configured skills through `get_or_new_user_skill_storage(user_id)` using the parent runtime identity, with `DEFAULT_USER_ID` only when no identity is available. This keeps custom-skill shadowing and visibility aligned with the lead agent instead of reading the global-only catalog.
+**Upload-state boundary**: Ordinary `task` delegation snapshots a valid parent `ThreadState.uploaded_files` list at dispatch, deep-copies it across the isolated-loop boundary, seeds it into the child's fresh state, and only then makes `list_uploaded_files` eligible for normal tool-policy filtering. An explicit empty list is valid and must be preserved because it means every upload in the thread is historical for this run. Missing or malformed state fails closed with the tool disabled. Durable `batch_task` execution intentionally keeps the tool disabled: delayed and recovered items have no valid parent-run upload boundary, and supporting that case requires a separate persisted-state contract.
 **Date context (#4781)**: Every built-in subagent execution registers `SubagentDateContextMiddleware` immediately before `SystemMessageCoalescingMiddleware`. Its one-time `before_agent` hook adds a hidden framework-owned `SystemMessage` containing only `<current_date>` before the first model call; it does not read `AppConfig.memory`, call the memory manager, rewrite the task `HumanMessage`, or inherit the lead agent's frozen-conversation/midnight lifecycle. The coalescer merges that reminder with the subagent's static prompt so strict providers still receive exactly one leading `SystemMessage`. The lead-only `DynamicContextMiddleware` registration and its date, optional-memory, and midnight-update behavior remain unchanged.
 **Execution**: Ordinary and durable-batch native subagents submit coroutines directly to one persistent isolated event loop. Gateway/embedded startup installs one process-wide async FIFO admission controller (default 3 running, bounded queue). Direct `create_deerflow_agent` callers can instead pass a caller-owned `SubagentRuntime`; reuse the same instance across graphs so its bound `task`, optional batch tools/service, middleware limits, and `SubagentExecutor` all share one controller without reading global YAML. An owned batch service must be started before graph construction and stopped at application shutdown. Waiters hold no scheduler thread, and cancellation/timeout release queue/slot ownership.
 **Shared sandbox execution lifecycle** (#5128): every admitted subagent run carries a stable task-derived `sandbox_lease_owner_id` and matching `sandbox_command_scope_id` in its runtime context. Sandbox middleware retains that execution against the lead thread's active provider client, so one child finishing cannot close the sandbox while siblings still run; the final holder performs any pending provider release. A rollback/fork-restored child reusing the parent's live client binds a non-releasing holder: it fences parent cleanup and owns its command scope without requesting a park itself; a parent's earlier park request waits for the child, while a missing inherited client falls through to a normal fresh acquire. On AIO, the command scope selects one explicit persistent shell session per subagent, allowing independent scopes to run concurrently while preserving in-order shell state within one child. Sync sandbox tool bodies offloaded with `asyncio.to_thread` are shielded and drained across repeated cancellation before the outer execution can clean its holder; a cancelled worker can therefore neither re-admit an already-released owner nor run after subagent terminalization. Middleware performs the normal release, and `SubagentExecutor` repeats it idempotently in `finally` so exceptions, cooperative cancellation, and timeout unwind paths cannot leak a lease or scoped session.

backend/packages/harness/deerflow/subagents/executor.py

@@ -12,6 +12,7 @@ from collections.abc import Callable, Coroutine, Mapping
 from concurrent.futures import Future
 from concurrent.futures import TimeoutError as FuturesTimeoutError
 from contextvars import Context, copy_context
+from copy import deepcopy
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from enum import Enum
@@ -772,6 +773,7 @@ class SubagentExecutor:
         parent_model: str | None = None,
         sandbox_state: SandboxState | None = None,
         thread_data: ThreadDataState | None = None,
+        uploaded_files: list[dict[str, Any]] | None = None,
         thread_id: str | None = None,
         trace_id: str | None = None,
         user_id: str | None = None,
@@ -799,6 +801,9 @@ class SubagentExecutor:
             parent_model: The parent agent's model name for inheritance.
             sandbox_state: Sandbox state from parent agent.
             thread_data: Thread data from parent agent.
+            uploaded_files: Snapshot of files uploaded in the parent's current
+                run. Seeded into the child graph state so ``list_uploaded_files``
+                can exclude them from historical-upload results.
             thread_id: Thread ID for sandbox operations.
             trace_id: Trace ID from parent for distributed tracing.
             user_id: User ID captured from the parent tool's runtime context.
@@ -845,6 +850,7 @@ class SubagentExecutor:
             self.model_name = None
         self.sandbox_state = sandbox_state
         self.thread_data = thread_data
+        self.uploaded_files = deepcopy(uploaded_files) if uploaded_files is not None else None
         self.thread_id = thread_id
         # Generate trace_id if not provided (for top-level calls)
         self.trace_id = trace_id or str(uuid.uuid4())[:8]
@@ -1256,11 +1262,15 @@ class SubagentExecutor:
             "messages": messages,
         }
 
-        # Pass through sandbox and thread data from parent
+        # Pass through the parent runtime state that tools need. Each child
+        # receives fresh containers so graph writes never mutate the snapshot
+        # held by another execution.
         if self.sandbox_state is not None:
             state["sandbox"] = self.sandbox_state
         if self.thread_data is not None:
             state["thread_data"] = self.thread_data
+        if self.uploaded_files is not None:
+            state["uploaded_files"] = deepcopy(self.uploaded_files)
 
         return state, final_tools, deferred_setup
 

backend/packages/harness/deerflow/tools/builtins/task_tool.py

@@ -8,6 +8,7 @@ import time
 import uuid
 from contextvars import ContextVar
 from dataclasses import replace
+from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, Any, cast
 
 from langchain.tools import InjectedToolCallId, tool
@@ -763,6 +764,8 @@ async def task_tool(
     # Extract parent context from runtime
     sandbox_state = None
     thread_data = None
+    uploaded_files = None
+    upload_state_available = False
     thread_id = None
     parent_model = None
     trace_id = None
@@ -771,6 +774,15 @@ async def task_tool(
     if runtime is not None:
         sandbox_state = runtime.state.get("sandbox")
         thread_data = runtime.state.get("thread_data")
+        parent_uploaded_files = runtime.state.get("uploaded_files")
+        if isinstance(parent_uploaded_files, list) and all(
+            isinstance(entry, dict) and isinstance(entry.get("filename"), str) and bool(entry["filename"]) and Path(entry["filename"]).name == entry["filename"] for entry in parent_uploaded_files
+        ):
+            # Only a complete, validated boundary can safely exclude same-run
+            # files. SubagentExecutor snapshots it synchronously before work
+            # crosses to the isolated event loop.
+            uploaded_files = parent_uploaded_files
+            upload_state_available = True
         thread_id = runtime.context.get("thread_id") if runtime.context else None
         if thread_id is None:
             thread_id = runtime.config.get("configurable", {}).get("thread_id")
@@ -833,15 +845,16 @@ async def task_tool(
         resolved_app_config = get_app_config()
     effective_model = resolve_subagent_model_name(config, parent_model, app_config=resolved_app_config)
 
-    # Subagents should not have subagent tools enabled (prevent recursive nesting).
-    # Subagents also must not get list_uploaded_files — they have an independent
-    # ThreadState where runtime.state["uploaded_files"] is absent, so the
-    # current-run file exclusion would not work.
+    # Subagents should not have subagent tools enabled (prevent recursive
+    # nesting). Ordinary task subagents receive a snapshot of the parent's
+    # current-run uploads below, so historical upload discovery is safe when
+    # that state channel is present. Non-standard callers without the channel
+    # remain fail-closed instead of misclassifying current uploads as history.
     available_tools_kwargs = {
         "model_name": effective_model,
         "groups": parent_tool_groups,
         "subagent_enabled": False,
-        "include_upload_tool": False,
+        "include_upload_tool": upload_state_available,
     }
     if resolved_app_config is not None:
         available_tools_kwargs["app_config"] = resolved_app_config
@@ -854,6 +867,7 @@ async def task_tool(
         "parent_model": parent_model,
         "sandbox_state": sandbox_state,
         "thread_data": thread_data,
+        "uploaded_files": uploaded_files,
         "thread_id": thread_id,
         "trace_id": trace_id,
         "user_id": user_id,

backend/packages/harness/deerflow/tools/tools.py

@@ -76,8 +76,9 @@ def get_available_tools(
         model_name: Optional model name to determine if vision tools should be included.
         subagent_enabled: Whether to include subagent tools (task, task_status).
         include_upload_tool: Whether to include ``list_uploaded_files`` (default: True).
-            Set to False for subagent tool assembly — subagents have independent
-            ThreadState and cannot exclude current-run files.
+            Ordinary task subagents enable it only after snapshotting the
+            parent's current-run upload state. Durable batch and non-standard
+            subagent callers without that state keep it disabled.
 
     Returns:
         List of available tools.

backend/tests/test_subagent_executor.py

@@ -534,6 +534,63 @@ class TestAgentConstruction:
         assert base_config.system_prompt in messages[0].content
         assert isinstance(messages[1], HumanMessage)
 
+    @pytest.mark.anyio
+    async def test_build_initial_state_seeds_current_upload_snapshot(
+        self,
+        classes,
+        base_config,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        """A delegated graph receives the parent's current-run upload boundary."""
+        SubagentExecutor = classes["SubagentExecutor"]
+
+        monkeypatch.setattr(
+            sys.modules["deerflow.skills.storage"],
+            "get_or_new_user_skill_storage",
+            lambda user_id, *, app_config=None: SimpleNamespace(load_skills=lambda *, enabled_only: []),
+        )
+        parent_uploads = [
+            {
+                "filename": "fresh.pdf",
+                "size": 128,
+                "path": "/mnt/user-data/uploads/fresh.pdf",
+                "extension": ".pdf",
+                "outline": [{"line": 1, "title": "Summary"}],
+            }
+        ]
+        executor = SubagentExecutor(
+            config=base_config,
+            tools=[],
+            thread_id="test-thread",
+            uploaded_files=parent_uploads,
+        )
+
+        parent_uploads[0]["filename"] = "mutated-after-dispatch.pdf"
+        parent_uploads[0]["outline"][0]["title"] = "Mutated"
+        state, _final_tools, _deferred_setup = await executor._build_initial_state("Do the task")
+
+        assert state["uploaded_files"] == [
+            {
+                "filename": "fresh.pdf",
+                "size": 128,
+                "path": "/mnt/user-data/uploads/fresh.pdf",
+                "extension": ".pdf",
+                "outline": [{"line": 1, "title": "Summary"}],
+            }
+        ]
+        assert state["uploaded_files"] is not parent_uploads
+        assert state["uploaded_files"][0] is not parent_uploads[0]
+
+        empty_executor = SubagentExecutor(
+            config=base_config,
+            tools=[],
+            thread_id="test-thread",
+            uploaded_files=[],
+        )
+        empty_state, _final_tools, _deferred_setup = await empty_executor._build_initial_state("Find earlier uploads")
+        assert "uploaded_files" in empty_state
+        assert empty_state["uploaded_files"] == []
+
     @pytest.mark.anyio
     async def test_build_initial_state_no_system_prompt_with_skills(
         self,

backend/tests/test_task_tool_core_logic.py

@@ -776,6 +776,14 @@ def test_task_tool_threads_runtime_app_config_to_subagent_dependencies(monkeypat
 def test_task_tool_emits_running_and_completed_events(monkeypatch):
     config = _make_subagent_config()
     runtime = _make_runtime()
+    runtime.state["uploaded_files"] = [
+        {
+            "filename": "fresh.pdf",
+            "size": 128,
+            "path": "/mnt/user-data/uploads/fresh.pdf",
+            "extension": ".pdf",
+        }
+    ]
     runtime.context["deerflow_trace_id"] = "task-trace-1"
     events = []
     dispatched_events = []
@@ -839,12 +847,13 @@ def test_task_tool_emits_running_and_completed_events(monkeypatch):
     assert captured["executor_kwargs"]["thread_id"] == "thread-1"
     assert captured["executor_kwargs"]["parent_model"] == "ark-model"
     assert captured["executor_kwargs"]["deerflow_trace_id"] == "task-trace-1"
+    assert captured["executor_kwargs"]["uploaded_files"] == runtime.state["uploaded_files"]
     assert captured["executor_kwargs"]["config"].max_turns == config.max_turns
     # Skills are no longer appended to system_prompt; they are loaded per-session
     # by SubagentExecutor and injected as conversation items (Codex pattern).
     assert captured["executor_kwargs"]["config"].system_prompt == "Base system prompt"
 
-    get_available_tools.assert_called_once_with(model_name="ark-model", groups=None, subagent_enabled=False, include_upload_tool=False)
+    get_available_tools.assert_called_once_with(model_name="ark-model", groups=None, subagent_enabled=False, include_upload_tool=True)
 
     event_types = [e["type"] for e in events]
     assert event_types == ["task_started", "task_running", "task_running", "task_completed"]
@@ -923,16 +932,20 @@ def test_task_tool_propagates_tool_groups_to_subagent(monkeypatch):
         state={
             "sandbox": {"sandbox_id": "local"},
             "thread_data": {"workspace_path": "/tmp/workspace"},
+            # An empty current-run snapshot is valid and is the common case
+            # when the subagent needs to discover uploads from an earlier turn.
+            "uploaded_files": [],
         },
         context={"thread_id": "thread-1"},
         config={"metadata": {"model_name": "ark-model", "trace_id": "trace-1", "tool_groups": parent_tool_groups}},
     )
     events = []
+    captured = {}
     get_available_tools = MagicMock(return_value=["tool-a"])
 
     class DummyExecutor:
         def __init__(self, **kwargs):
-            pass
+            captured.update(kwargs)
 
         def execute_async(self, prompt, task_id=None):
             return task_id or "generated-task-id"
@@ -958,8 +971,9 @@ def test_task_tool_propagates_tool_groups_to_subagent(monkeypatch):
     )
 
     assert _task_tool_message(output).content == "Task Succeeded. Result: done"
+    assert captured["uploaded_files"] == []
     # The key assertion: groups should be propagated from parent metadata
-    get_available_tools.assert_called_once_with(model_name="ark-model", groups=parent_tool_groups, subagent_enabled=False, include_upload_tool=False)
+    get_available_tools.assert_called_once_with(model_name="ark-model", groups=parent_tool_groups, subagent_enabled=False, include_upload_tool=True)
 
 
 def test_task_tool_uses_subagent_model_override_for_tool_loading(monkeypatch):
@@ -974,6 +988,9 @@ def test_task_tool_uses_subagent_model_override_for_tool_loading(monkeypatch):
     )
     runtime = _make_runtime()
     runtime.config["metadata"]["model_name"] = "parent-text-model"
+    # Do not enable upload discovery when the parent boundary is malformed:
+    # doing so could expose a same-run upload as if it were historical.
+    runtime.state["uploaded_files"] = [{"filename": ""}]
     events = []
     get_available_tools = MagicMock(return_value=[])