feat(subagents): add bounded same-thread batch result inspection (#6416)

* feat(subagents): add bounded same-thread batch result reader

* test(subagents): validate batch result reader on PostgreSQL

* docs(subagents): explain batch reader budget and position contracts

* fix(tests): adapt batch reader PostgreSQL CI connection options

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -1889,7 +1889,13 @@ An ordinary `task` also receives a defensive snapshot of the dispatching run's c
 
 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.
+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; bulk stored results are available 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.
+
+While durable batch tools are available, a later owner turn can explicitly inspect selected results with `read_batch_result(batch_id, position=0, offset=0, max_chars=4000)`. The reader is restricted to the current user and thread, including in explicitly bound SDK graphs. It returns one window of a JSON document containing the stored report, execution/error state and separate acceptance criteria/verdict, without starting or waiting for work. Concatenate `content` windows before parsing: pass `next_offset` and the returned `revision` as `expected_revision` until `next_offset` is null. A changed item returns `restart_required`; discard prior windows and start at zero. Use `next_position` for a selected next item within `batch_status.total_items`. The maximum window is 8,192 characters; metadata shares the window budget, and JSON response framing adds overhead. `result_truncated` means storage already capped the report; continuation cannot recover discarded content. A succeeded item is not an accepted task, and unchecked criteria remain UNVERIFIED. No automatic result ingestion, waiting loop or UI change is added.
+
+The complete escaped response also fits within 10,000 characters and the bound host's positive `tool_output` externalization/fallback limits, so normal budgeting preserves continuation JSON. Smaller limits can shorten a window; `budget_too_small` means the operator must raise the budget instead of retrying a zero-length page. Limits too small even for that error envelope cannot provide a usable structured response. Explicit SDK graphs with a custom output-budget middleware should supply the same `AppConfig` through `SubagentRuntime`; default graphs use default reader limits.
+
+Offline acceptance: `cd backend && PYTHONPATH=. uv run --locked pytest tests/test_batch_result_reader.py -k 'real_submission_worker or sdk_factory_graph' -q`. This exercises real submission, deterministic worker completion, database reopen and registered graph consumption; external model execution is replaced by a fake model/executor.
 
 Durable knowledge-research results also export nullable `result_artifact.knowledge_sources` snapshots. Each retained source pairs a citation ID in the stored report with the exact excerpt captured by `knowledge_search` or a delegated `task`, plus its RAGFlow dataset/document/chunk locator and pages. The existing authenticated `GET /api/threads/{thread_id}/subagent-batches/{batch_id}/results.jsonl` remains the entry point: find a report's `#knowledge-<id>` link in that row's `sources` by `id` to inspect the evidence after a Gateway restart or provider change. Compact item queries and the batch panel do not include these snapshots; this does not add clickable citations to batch previews.
 

backend/packages/harness/deerflow/agents/factory.py

@@ -410,9 +410,9 @@ def _assemble_from_features(
             from deerflow.subagents.batch_runtime import is_subagent_batch_runtime_available
 
             if is_subagent_batch_runtime_available():
-                from deerflow.tools.builtins import batch_status, batch_task, cancel_batch
+                from deerflow.tools.builtins import batch_status, batch_task, cancel_batch, read_batch_result
 
-                extra_tools.extend((batch_task, batch_status, cancel_batch))
+                extra_tools.extend((batch_task, batch_status, cancel_batch, read_batch_result))
 
     # --- [12] LoopDetection ---
     if feat.loop_detection is not False:

backend/packages/harness/deerflow/agents/lead_agent/prompt.py

@@ -481,7 +481,10 @@ count and never emulate it by repeatedly calling `task`.
   implies that all items become live or run at once.
 - Use `batch_status` for compact progress and `cancel_batch` for cancellation.
 - Do not wait for or paste all item results into this run. The Web UI and results
-  export API own progress and result inspection.
+  export API own bulk inspection. When the owner explicitly asks to inspect or
+  synthesize stored results, use `read_batch_result` for selected items in this
+  thread. Follow its bounded continuation/revision contract; never poll for
+  completion or equate execution success with acceptance.
 """
     return f"""<subagent_system>
 ## Subagent Routing: Delegate Only for Clear Net Benefit

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

@@ -1,11 +1,16 @@
 ### Subagent System (`packages/harness/deerflow/subagents/`)
 
+Result-reader acceptance runs on SQLite and, with TEST_POSTGRES_URI, an isolated
+PostgreSQL schema; both exercise submission, completion, reopen and ToolNode
+continuation without paid model work. Always clean up the temporary schema.
+
 **JSON**: See README. Authorize before local metadata and reads; inaccessible paths stay UNVERIFIED.
 
 Apply each subagent's prompt overlay after assembling its full SystemMessage.
 Registry overrides must not mutate `BUILTIN_SUBAGENTS`.
 Direct returns use the compiled tool registry, including middleware tools. Match current-turn IDs in call order; error ToolMessages fail the task with outputs preserved.
 Durable batch specs store overlays as JSON and restore them before execution.
+`read_batch_item` projects only report/state/acceptance fields after owner and current-thread checks. Positions are immutable submission order; no status filtering. The bounded tool hashes that projection for continuation, detecting completion/retry/cancel changes. It does not persist snapshots, schedule work or expose execution specs; HTTP bulk export stays independent.
 
 Batch evidence uses nullable `result_artifact` (migration 0033), persisted with
 successful results under the same lease fence. Only full-result exports expose

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

@@ -33,6 +33,8 @@ class SubagentBatchSubmitter(Protocol):
 
     async def get_batch(self, *, batch_id: str, user_id: str) -> dict[str, Any] | None: ...
 
+    async def read_batch_item(self, *, batch_id: str, user_id: str, thread_id: str, position: int) -> dict[str, Any] | None: ...
+
     async def cancel_batch(self, *, batch_id: str, user_id: str) -> dict[str, Any] | None: ...
 
 

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

@@ -173,6 +173,17 @@ class SubagentBatchService:
     ) -> dict[str, Any] | None:
         return await self._repository.get_batch(batch_id, user_id=user_id)
 
+    async def read_batch_item(self, *, batch_id: str, user_id: str, thread_id: str, position: int) -> dict[str, Any] | None:
+        """Read one current-thread result; never expose worker execution context."""
+        batch = await self._repository.get_batch(batch_id, user_id=user_id)
+        if batch is None or batch["thread_id"] != thread_id:
+            return None
+        items = await self._repository.list_items(batch_id, user_id=user_id, offset=position, limit=1, include_result=True)
+        if not items:
+            return None
+        fields = ("id", "item_key", "position", "status", "attempt", "result", "result_truncated", "error", "stop_reason", "acceptance_criteria", "acceptance_verdict")
+        return {key: items[0].get(key) for key in fields}
+
     async def cancel_batch(
         self,
         *,

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

@@ -2,6 +2,7 @@
 
 `task` and `batch_task` opt into JSON checks with `file:<path> json-valid`.
 See [subagents/AGENTS.md](../subagents/AGENTS.md) for read limits and UNVERIFIED semantics.
+Batch readers bound by assembly use its AppConfig output budget. Fit the complete escaped response under 10K and active per-tool/fallback limits; too-small envelopes stop reading. Numeric reader inputs are strict integers. Escape `<` in outer JSON to preserve untrusted data without exposing framework tags; decoding restores the document unchanged.
 
 `list_uploaded_files` 的续页契约见 [FILE_UPLOAD.md](../../../../docs/FILE_UPLOAD.md)。
 游标绑定可信用户/线程、规范化过滤条件、本轮上传排除集合和目录元数据；身份与目录
@@ -38,7 +39,7 @@ Only standalone tool discovery without a model falls back to the base profile.
 4. **Subagent tool** (if enabled):
    - `task` - Delegate to subagent (`prompt`, `subagent_type`, optional `acceptance_criteria`, and an optional model-visible `description` used only as a short progress label). Execution never depends on `description`; lifecycle display falls back to `prompt` when a provider omits it. Subagent reports are self-reports: the docstring directs the lead to expect `[rN]` receipt citations and verifiable handles while `verification.receipts_enabled` (and explicitly qualifies that disabled receipts mean no citations and no citation verdict), to read the delegation ledger's citation cross-check as execution evidence only, and to attach `acceptance_criteria` for objectively checkable outcomes (canonical forms `file:<path> exists|non-empty`, `file_written:<path>`, `tests_passed:<command>`); criteria are handed to the executor and appended to the subagent's task message as untrusted data (see `subagents/report_contract.py`).
      Polling safety timeouts carry the latest published tool receipts into the terminal task metadata before requesting background cancellation.
-   - `batch_task`, `batch_status`, `cancel_batch` - Explicit durable batch submission/progress/cancellation. Added only while the startup SQL-backed batch submitter is installed; large results stay in the owner-scoped API/JSONL export rather than the lead context. Items accept optional `acceptance_criteria`; item queries and exports expose the separate `acceptance_verdict`. Progress counts describe execution, not acceptance; unmet and UNVERIFIED conditions never trigger automatic retries.
+   - `batch_task`, `batch_status`, `cancel_batch`, `read_batch_result` - Explicit durable batch operations, added only while the SQL-backed submitter is installed. Bulk results stay in JSONL; explicit reads return one current-user/thread item document window (≤8192 chars, metadata included). Concatenate windows before parsing; nonzero offsets require the previous content revision, changed state requires restart. Serialization/hash run off-loop. Bound SDK tools never fall back to a global worker. Criteria/verdicts describe acceptance separately from execution; reads neither wait nor retry.
    - Direct `create_deerflow_agent` integrations receive cloned tools bound to their explicit `SubagentRuntime`. The bound `task` forwards that runtime's exact execution controller and optional caller-owned `AppConfig` into registry/model/tool resolution and `SubagentExecutor`; bound batch tools use the same config snapshot and resolve only that runtime's submitter before falling back to no other application's active worker. Keep the original tool name/schema unchanged so model contracts and user-tool deduplication remain stable.
 
 The ordinary `task` boundary carries one narrow parent-loop middleware recorder into the isolated subagent runtime under separate loop-detection, tool-promotion, and tool-progress keys. It schedules only `record_middleware` calls back onto the loop that owns `RunJournal`, keeps an execution-local atomic promotion claim so parallel searches do not double-report one new schema, is fenced and drained once before `task` returns, and never exposes the journal or event store to the child loop. Durable batch tasks have no parent run journal and do not use this bridge.

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

@@ -1,5 +1,5 @@
 from .background_tasks_tool import cancel_background_task, list_background_tasks
-from .batch_task_tool import batch_status, batch_task, cancel_batch
+from .batch_task_tool import batch_status, batch_task, cancel_batch, read_batch_result
 from .clarification_tool import ask_clarification_tool
 from .list_uploaded_files_tool import list_uploaded_files
 from .present_file_tool import present_file_tool
@@ -20,6 +20,7 @@ __all__ = [
     "batch_task",
     "batch_status",
     "cancel_batch",
+    "read_batch_result",
     "list_uploaded_files",
     "list_background_tasks",
     "cancel_background_task",

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

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
+import asyncio
+import hashlib
 import json
+import re
 from collections.abc import Callable
 from contextvars import ContextVar
 from dataclasses import asdict, replace
@@ -11,9 +14,10 @@ from typing import Annotated, Any, cast
 from langchain.tools import InjectedToolCallId, tool
 from langchain_core.messages import ToolMessage
 from langgraph.types import Command
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, StrictInt
 
 from deerflow.authz.principal import normalize_authz_attributes
+from deerflow.config.tool_output_config import ToolOutputConfig
 from deerflow.knowledge_scope import KNOWLEDGE_SCOPE_RUNTIME_KEY, execution_scope
 from deerflow.mcp_scope import (
     THREAD_INCARNATION_CONTEXT_KEY,
@@ -110,7 +114,7 @@ def bind_batch_tools(
         raise ValueError("Provide exactly one of submitter or submitter_provider")
     provider = submitter_provider if submitter_provider is not None else lambda: submitter
 
-    return tuple(_bind_batch_tool(tool, provider, app_config) for tool in (batch_task, batch_status, cancel_batch))
+    return tuple(_bind_batch_tool(tool, provider, app_config) for tool in (batch_task, batch_status, cancel_batch, read_batch_result))
 
 
 def _result(tool_call_id: str, *, content: str, batch: dict[str, Any] | None = None, error: bool = False) -> Command:
@@ -303,3 +307,109 @@ async def cancel_batch(runtime: Runtime, batch_id: str) -> str:
     if batch is None:
         return "Batch not found."
     return f"Batch {batch_id} cancellation requested."
+
+
+def _batch_result_window(item: dict[str, Any], offset: int, max_chars: int, expected_revision: str | None, response_limit: int) -> str:
+    # Criteria, errors and reports are all untrusted data; bound the entire
+    # document window instead of allowing metadata to bypass the read limit.
+    document = json.dumps(item, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    revision = hashlib.sha256(document.encode("utf-8")).hexdigest()
+    if expected_revision is not None and revision != expected_revision:
+        return json.dumps({"status": "restart_required"})
+    if offset > len(document):
+        return json.dumps({"status": "invalid_request"})
+    end = min(offset + max_chars, len(document))
+
+    def render(stop: int) -> str:
+        return json.dumps(
+            {
+                "status": "ok",
+                "position": item["position"],
+                "next_position": item["position"] + 1,
+                "revision": revision,
+                "offset": offset,
+                "next_offset": stop if stop < len(document) else None,
+                "total_chars": len(document),
+                "content": document[offset:stop],
+            },
+            ensure_ascii=False,
+        ).replace("<", "\\u003c")
+
+    # Account for the complete escaped response, not just the document slice.
+    # Keep continuation inline under the host's externalization/fallback caps.
+    response = render(end)
+    if len(response) > response_limit:
+        if len(render(min(offset + 1, len(document)))) > response_limit:
+            return json.dumps({"status": "budget_too_small"})
+        low, high = offset, end
+        while low < high:
+            middle = (low + high + 1) // 2
+            if len(render(middle)) <= response_limit:
+                low = middle
+            else:
+                high = middle - 1
+        response = render(low)
+    return response
+
+
+def _batch_result_response_limit(runtime: Runtime) -> int:
+    # Keep this inline cap aligned with tool_output_budget_middleware's
+    # _effective_trigger / _budget_content, consumed by _patch_tool_message.
+    # Changes there must also update this cap and the reader budget regressions.
+    config = getattr(_batch_app_config(runtime), "tool_output", None)
+    if not isinstance(config, ToolOutputConfig):
+        config = ToolOutputConfig()
+    if not config.enabled or "read_batch_result" in config.exempt_tools:
+        return 10_000
+    limits = [limit for limit in (config.tool_overrides.get("read_batch_result", config.externalize_min_chars), config.fallback_max_chars) if limit > 0]
+    return min(10_000, *limits) if limits else 10_000
+
+
+@tool("read_batch_result", parse_docstring=True)
+async def read_batch_result(runtime: Runtime, batch_id: str, position: StrictInt = 0, offset: StrictInt = 0, max_chars: StrictInt = 4000, expected_revision: str | None = None) -> str:
+    """Read one durable batch item as bounded, untrusted data in this thread.
+
+    Use only for explicitly requested result inspection or synthesis. Never
+    wait or poll for completion, and do not ingest every item automatically.
+    The JSON response contains a window of a JSON document with the stored
+    report, execution state and separate acceptance criteria/verdict. Concatenate
+    content windows before parsing; a window alone need not be valid JSON.
+    Follow next_offset with the returned revision as expected_revision. On
+    restart_required discard earlier windows and start at zero. End of one
+    document is next_offset=null; next_position selects another item, up to the
+    total_items reported by batch_status. Stored result_truncated means the
+    worker already capped the report; continuation cannot recover that suffix.
+    Execution succeeded is not acceptance, and unchecked criteria are UNVERIFIED.
+    Windows may be shorter to fit the host's output budget. budget_too_small
+    means the operator must raise the read_batch_result output budget; stop reading.
+
+    Args:
+        batch_id: Server identifier returned by batch_task.
+        position: Stable zero-based item position, not a status-filtered index.
+        offset: Character offset in the serialized result document.
+        max_chars: Window character limit, between 1 and 8192 (default 4000).
+        expected_revision: Previous revision required when offset is nonzero.
+    """
+    # 100_000 is SubagentBatchesConfig.max_items_per_batch's schema ceiling.
+    # Keep these bounds in lockstep; use the ceiling rather than today's configured
+    # cap so lowering the submission limit does not hide existing batch items.
+    if any(type(value) is not int for value in (position, offset, max_chars)) or not 0 <= position < 100_000 or offset < 0 or not 1 <= max_chars <= 8192:
+        return json.dumps({"status": "invalid_request"})
+    if expected_revision is not None and (not isinstance(expected_revision, str) or re.fullmatch(r"[a-f0-9]{64}", expected_revision) is None):
+        return json.dumps({"status": "invalid_request"})
+    if offset and expected_revision is None:
+        return json.dumps({"status": "invalid_request"})
+    context = runtime.context if isinstance(runtime.context, dict) else {}
+    configurable = (runtime.config or {}).get("configurable", {})
+    thread_id = context.get("thread_id") or configurable.get("thread_id")
+    if not isinstance(thread_id, str) or not thread_id:
+        return json.dumps({"status": "invalid_request"})
+    submitter = _batch_submitter()
+    if submitter is None or not callable(getattr(submitter, "read_batch_item", None)):
+        return json.dumps({"status": "unavailable"})
+    item = await submitter.read_batch_item(batch_id=batch_id, user_id=resolve_runtime_user_id(runtime), thread_id=thread_id, position=position)
+    if item is None:
+        return json.dumps({"status": "not_found"})
+    # Reports can reach the configured million-character cap; serializing and
+    # hashing the snapshot must not run on the streaming event loop.
+    return await asyncio.to_thread(_batch_result_window, item, offset, max_chars, expected_revision, _batch_result_response_limit(runtime))

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

@@ -16,10 +16,7 @@ from deerflow.scheduler.runtime import SchedulerRunCapability, is_scheduler_capa
 from deerflow.subagents.batch_runtime import is_subagent_batch_runtime_available
 from deerflow.tools.builtins import (
     ask_clarification_tool,
-    batch_status,
-    batch_task,
     cancel_background_task,
-    cancel_batch,
     list_background_tasks,
     list_uploaded_files,
     present_file_tool,
@@ -199,7 +196,10 @@ def get_available_tools(
     if subagent_enabled:
         builtin_tools.extend(SUBAGENT_TOOLS)
         if is_subagent_batch_runtime_available():
-            builtin_tools.extend((batch_task, batch_status, cancel_batch))
+            from deerflow.subagents.batch_runtime import get_subagent_batch_submitter
+            from deerflow.tools.builtins.batch_task_tool import bind_batch_tools
+
+            builtin_tools.extend(bind_batch_tools(submitter_provider=get_subagent_batch_submitter, app_config=config))
         logger.info("Including native subagent tools")
 
     # If no model_name specified, use the first model (default)

backend/tests/test_batch_result_reader.py

@@ -0,0 +1,435 @@
+"""Owner/thread-bound durable results reach the real graph as bounded data."""
+
+import asyncio
+import importlib
+import json
+import os
+import uuid
+from datetime import UTC, datetime
+from enum import Enum
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import pytest_asyncio
+from langchain_core.messages import AIMessage
+from langgraph.graph import END, START, MessagesState, StateGraph
+from langgraph.prebuilt import ToolNode
+from support.postgres import asyncpg_test_url
+
+from deerflow.config.database_config import DatabaseConfig
+from deerflow.config.subagent_batches_config import SubagentBatchesConfig
+from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig
+from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
+from deerflow.persistence.subagent_batches import SubagentBatchRepository
+from deerflow.subagents import batch_service
+
+tools_module = importlib.import_module("deerflow.tools.builtins.batch_task_tool")
+
+
+class Status(Enum):
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+    @property
+    def is_terminal(self):
+        return self is not Status.RUNNING
+
+
+@pytest_asyncio.fixture
+async def env(monkeypatch, tmp_path, request):
+    schema = None
+    if getattr(request, "param", "sqlite") == "postgres":
+        uri = os.environ.get("TEST_POSTGRES_URI")
+        if not uri:
+            pytest.skip("requires TEST_POSTGRES_URI (real Postgres result reader)")
+        schema = f"batch_reader_{uuid.uuid4().hex}"
+        db = DatabaseConfig(backend="postgres", postgres_url=asyncpg_test_url(uri), postgres_schema=schema)
+    else:
+        db = DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path / "db"))
+    service = None
+    try:
+        await init_engine_from_config(db)
+        if schema:
+            from sqlalchemy import text
+
+            async with get_session_factory()() as session:
+                assert (await session.execute(text("SELECT current_schema()"))).scalar_one() == schema
+        repo = SubagentBatchRepository(get_session_factory())
+        state = SimpleNamespace(db=db, repo=repo, launches=0)
+        state.result = SimpleNamespace(status=Status.COMPLETED, result='Report "引用"\n' * 1500, error=None, stop_reason=None, token_usage_records=[], bash_executions=[])
+
+        class Executor:
+            def __init__(self, **kwargs):
+                pass
+
+            def execute_async(self, prompt, task_id=None):
+                state.launches += 1
+                return task_id
+
+        monkeypatch.setattr(batch_service, "SubagentExecutor", Executor)
+        monkeypatch.setattr(batch_service, "SubagentStatus", Status)
+        monkeypatch.setattr(batch_service, "get_background_task_result", lambda _: state.result)
+        monkeypatch.setattr(batch_service, "cleanup_background_task", lambda _: None)
+        monkeypatch.setattr(batch_service, "resolve_subagent_model_name", lambda *args, **kwargs: "offline-model")
+        monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
+        state.service = batch_service.SubagentBatchService(repository=repo, config=SubagentBatchesConfig(max_attempts=1), runtime_config=SubagentRuntimeConfig(), app_config=SimpleNamespace())
+        service = state.service
+        yield state
+    finally:
+        try:
+            if service is not None:
+                await service.stop()
+        finally:
+            try:
+                await close_engine()
+            finally:
+                if schema:
+                    import sqlalchemy as sa
+                    from sqlalchemy.ext.asyncio import create_async_engine
+
+                    engine = create_async_engine(db.app_sqlalchemy_url)
+                    try:
+                        async with engine.begin() as conn:
+                            await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
+                    finally:
+                        await engine.dispose()
+
+
+def runtime(user="user-1", thread="thread-1"):
+    return SimpleNamespace(context={"user_id": user, "thread_id": thread}, config={"configurable": {"thread_id": thread}})
+
+
+def reader(service):
+    tools = {tool.name: tool for tool in tools_module.bind_batch_tools(service)}
+    assert "read_batch_result" in tools, "No native graph tool can consume a stored batch result"
+    return tools["read_batch_result"]
+
+
+async def submit(env):
+    # Use the real batch submission tool; the fake executor replaces only paid work.
+    from deerflow.subagents.config import SubagentConfig
+
+    tool = {t.name: t for t in tools_module.bind_batch_tools(env.service)}["batch_task"]
+    with pytest.MonkeyPatch.context() as patcher:
+        patcher.setattr(tools_module, "get_available_subagent_names", lambda **kwargs: ["general-purpose"])
+        patcher.setattr(tools_module, "get_subagent_config", lambda *args, **kwargs: SubagentConfig(name="general-purpose", description="Worker"))
+        command = await tool.coroutine(
+            runtime=runtime(),
+            title="Research",
+            subagent_type="general-purpose",
+            tool_call_id="submit-1",
+            items=[tools_module.BatchTaskItem(key="one", prompt="Private item prompt", acceptance_criteria=["Human review required"]), tools_module.BatchTaskItem(key="two", prompt="Second")],
+            max_live_items=2,
+            max_running_items=2,
+        )
+    return command.update["messages"][0].additional_kwargs["subagent_batch_id"]
+
+
+async def read(service, batch_id, **args):
+    return json.loads(await reader(service).coroutine(runtime=args.pop("runtime", runtime()), batch_id=batch_id, **args))
+
+
+async def complete(env, batch):
+    await env.service.run_once(now=datetime.now(UTC))
+    await asyncio.gather(*list(env.service._executions.values()))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("env", ["sqlite", "postgres"], indirect=True)
+async def test_real_submission_worker_reopen_and_toolnode_continuation(env):
+    batch = await submit(env)
+    await complete(env, batch)
+    launches = env.launches
+    await env.service.stop()
+    await close_engine()
+    await init_engine_from_config(env.db)
+    repo = SubagentBatchRepository(get_session_factory())
+    service = batch_service.SubagentBatchService(repository=repo, config=SubagentBatchesConfig(), runtime_config=SubagentRuntimeConfig(), app_config=SimpleNamespace())
+    tool = reader(service)
+    graph = StateGraph(MessagesState, context_schema=dict)
+    graph.add_node("tools", ToolNode([tool]))
+    graph.add_edge(START, "tools")
+    graph.add_edge("tools", END)
+    compiled = graph.compile()
+    assert "runtime" not in tool.tool_call_schema.model_json_schema()["properties"]
+    for user, thread, position in [("other", "thread-1", 0), ("user-1", "other", 0), ("user-1", "thread-1", 2)]:
+        denied = await compiled.ainvoke(
+            {"messages": [AIMessage(content="", tool_calls=[{"name": "read_batch_result", "args": {"batch_id": batch, "position": position}, "id": "denied", "type": "tool_call"}])]},
+            context={"user_id": user, "thread_id": thread},
+        )
+        assert json.loads(denied["messages"][-1].content) == {"status": "not_found"}
+    chunks = []
+    offset, revision = 0, None
+    while True:
+        args = {"batch_id": batch, "position": 0, "offset": offset, "max_chars": 512, "expected_revision": revision}
+        output = await compiled.ainvoke({"messages": [AIMessage(content="", tool_calls=[{"name": "read_batch_result", "args": args, "id": "read-1", "type": "tool_call"}])]}, context={"user_id": "user-1", "thread_id": "thread-1"})
+        message = output["messages"][-1]
+        assert message.status == "success"
+        window = json.loads(message.content)
+        assert window["status"] == "ok"
+        assert len(window["content"]) <= 512
+        assert window["offset"] == offset and window["position"] == 0
+        assert window["next_position"] == 1
+        assert "Private item prompt" not in message.content
+        chunks.append(window["content"])
+        revision = window["revision"]
+        if window["next_offset"] is None:
+            break
+        offset = window["next_offset"]
+    document = json.loads("".join(chunks))
+    assert document["result"] == env.result.result
+    assert document["status"] == "succeeded"
+    assert document["acceptance_criteria"] == ["Human review required"]
+    assert document["acceptance_verdict"]["all_hold"] is False
+    assert document["acceptance_verdict"]["leaves"][0]["checked"] is False
+    assert env.launches == launches  # Reads never schedule or re-execute work.
+    assert document.keys() == {"id", "item_key", "position", "status", "attempt", "result", "result_truncated", "error", "stop_reason", "acceptance_criteria", "acceptance_verdict"}
+    await service.stop()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("user,thread", [("other", "thread-1"), ("user-1", "other"), ("other", "other")])
+async def test_reader_rejects_other_user_or_thread(env, user, thread):
+    batch = await submit(env)
+    await complete(env, batch)
+    output = await read(env.service, batch, runtime=runtime(user, thread))
+    assert output == {"status": "not_found"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "args", [{"position": -1}, {"position": True}, {"position": 100_000}, {"position": 2**100}, {"offset": -1}, {"offset": True}, {"max_chars": 0}, {"max_chars": 8193}, {"max_chars": True}, {"offset": 1}, {"expected_revision": "x"}]
+)
+async def test_invalid_reader_arguments_do_not_query_storage(args):
+    submitter = AsyncMock()
+    output = await read(submitter, "batch", **args)
+    assert output["status"] == "invalid_request"
+    submitter.read_batch_item.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_pending_empty_and_position_boundaries(env):
+    batch = await submit(env)
+    first = await read(env.service, batch, max_chars=8192)
+    assert json.loads(first["content"])["status"] == "pending"
+    assert json.loads(first["content"])["result"] is None
+    assert env.launches == 0
+    assert await read(env.service, batch, position=2) == {"status": "not_found"}
+    end = await read(env.service, batch, offset=first["total_chars"], expected_revision=first["revision"])
+    assert end["content"] == "" and end["next_offset"] is None
+    past = await read(env.service, batch, offset=first["total_chars"] + 1, expected_revision=first["revision"])
+    assert past["status"] == "invalid_request"
+
+
+@pytest.mark.asyncio
+async def test_completion_changes_revision_and_requires_restart(env):
+    batch = await submit(env)
+    before = await read(env.service, batch, max_chars=100)
+    await complete(env, batch)
+    after = await read(env.service, batch, offset=before["next_offset"], expected_revision=before["revision"])
+    assert after == {"status": "restart_required"}
+    fresh = await read(env.service, batch)
+    assert fresh["revision"] != before["revision"]
+
+
+@pytest.mark.asyncio
+async def test_failed_report_retry_and_cancel_states(env):
+    env.result.status = Status.FAILED
+    env.result.error = "Worker failed"
+    batch = await submit(env)
+    await complete(env, batch)
+    before = await read(env.service, batch)
+    failed = json.loads(before["content"])
+    assert failed["status"] == "failed" and failed["error"] == "Worker failed"
+    assert failed["result"] is None  # Existing failed-worker policy is preserved.
+    await env.repo.retry_item(batch, failed["id"], user_id="user-1")
+    assert await read(env.service, batch, offset=1, expected_revision=before["revision"]) == {"status": "restart_required"}
+    pending = json.loads((await read(env.service, batch))["content"])
+    assert pending["status"] == "pending" and pending["error"] is None
+    await env.service.cancel_batch(batch_id=batch, user_id="user-1")
+    cancelled = json.loads((await read(env.service, batch))["content"])
+    assert cancelled["status"] == "cancelled"
+
+
+@pytest.mark.asyncio
+async def test_bound_reader_stop_does_not_fall_back_to_global(monkeypatch):
+    current = [None]
+    global_submitter = AsyncMock()
+    monkeypatch.setattr(tools_module, "get_subagent_batch_submitter", lambda: global_submitter)
+    tools = {t.name: t for t in tools_module.bind_batch_tools(submitter_provider=lambda: current[0])}
+    assert "read_batch_result" in tools
+    output = json.loads(await tools["read_batch_result"].coroutine(runtime=runtime(), batch_id="batch"))
+    assert output == {"status": "unavailable"}
+    global_submitter.read_batch_item.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_continuation_accepts_changed_window_size(env):
+    batch = await submit(env)
+    await complete(env, batch)
+    first = await read(env.service, batch, max_chars=50)
+    changed = await read(env.service, batch, offset=50, max_chars=8192, expected_revision=first["revision"])
+    document = json.dumps(await env.service.read_batch_item(batch_id=batch, user_id="user-1", thread_id="thread-1", position=0), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    assert changed["status"] == "ok" and changed["content"] == document[50 : 50 + len(changed["content"])]
+
+
+@pytest.mark.asyncio
+async def test_long_criteria_and_error_are_inside_the_window(env):
+    batch = await env.repo.create_batch(
+        batch_id="long",
+        user_id="user-1",
+        thread_id="thread-1",
+        run_id=None,
+        tool_call_id=None,
+        submission_key="long",
+        title="Long",
+        subagent_type="general-purpose",
+        items=[{"key": "one", "prompt": "private", "acceptance_criteria": ["文" * 500 for _ in range(20)]}],
+        max_live_items=1,
+        max_running_items=1,
+        max_attempts=1,
+        execution_spec={"secret": "private"},
+    )
+    rows = await env.repo.claim_items(now=datetime.now(UTC), lease_owner="worker", lease_seconds=60, limit=1)
+    assert await env.repo.finalize_item(
+        rows[0]["id"], lease_owner="worker", succeeded=False, result=None, result_preview=None, result_truncated=False, error="error\n" * 3000, stop_reason=None, token_usage=None, model_name=None, completed_at=datetime.now(UTC)
+    )
+    page = await read(env.service, batch["id"], max_chars=8192)
+    assert len(page["content"]) == 8192 and page["next_offset"] == 8192
+    assert len(json.dumps(page, ensure_ascii=False)) < 17_000
+    assert page["total_chars"] > 8192 and "private" not in page["content"]
+
+
+@pytest.mark.asyncio
+async def test_sdk_factory_graph_consumes_reader(env):
+    from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+    from langchain_core.messages import HumanMessage, ToolMessage
+
+    from deerflow.agents.factory import create_deerflow_agent
+    from deerflow.agents.features import RuntimeFeatures
+    from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware
+    from deerflow.config.app_config import AppConfig
+    from deerflow.config.tool_output_config import ToolOutputConfig
+    from deerflow.subagents.runtime import SubagentRuntime
+
+    class Model(FakeMessagesListChatModel):
+        def bind_tools(self, tools, **kwargs):
+            return self
+
+    batch = await submit(env)
+    budget = ToolOutputConfig(tool_overrides={"read_batch_result": 500})
+    owned = SubagentRuntime(SubagentRuntimeConfig(), batch_submitter=env.service, app_config=AppConfig(sandbox={"use": "deerflow.sandbox.local:LocalSandboxProvider"}, tool_output=budget))
+    model = Model(responses=[AIMessage(content="", tool_calls=[{"name": "read_batch_result", "args": {"batch_id": batch}, "id": "reader", "type": "tool_call"}]), AIMessage(content="The item is still pending.")])
+    graph = create_deerflow_agent(model, features=RuntimeFeatures(subagent=True, sandbox=False, loop_detection=False), subagent_runtime=owned, extra_middleware=[ToolOutputBudgetMiddleware(budget)])
+    result = await graph.ainvoke({"messages": [HumanMessage(content="Inspect the first stored item.")]}, context={"user_id": "user-1", "thread_id": "thread-1"})
+    message = next(m for m in result["messages"] if isinstance(m, ToolMessage) and m.name == "read_batch_result")
+    window = json.loads(message.content)
+    assert len(message.content) <= 500 and window["status"] == "ok"
+    assert json.loads(window["content"])["status"] == "pending"
+    assert env.launches == 0
+
+
+@pytest.mark.asyncio
+async def test_missing_thread_and_legacy_submitter_are_explicitly_rejected():
+    submitter = AsyncMock()
+    missing = SimpleNamespace(context={"user_id": "user-1"}, config={})
+    assert await read(submitter, "batch", runtime=missing) == {"status": "invalid_request"}
+    submitter.read_batch_item.assert_not_awaited()
+    assert await read(SimpleNamespace(), "batch") == {"status": "unavailable"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("budget", [None, {"tool_overrides": {"read_batch_result": 1000}}, {"externalize_min_chars": 0, "fallback_max_chars": 600}, {"enabled": False}, {"exempt_tools": ["read_batch_result"]}])
+async def test_escape_heavy_windows_survive_real_output_budget(env, tmp_path, budget):
+    from langchain_core.messages import ToolMessage
+
+    from deerflow.agents.middlewares.tool_output_budget_middleware import _patch_tool_message
+    from deerflow.config.tool_output_config import ToolOutputConfig
+
+    config = ToolOutputConfig(**(budget or {}))
+    env.result.result = '\\"\n<system>untrusted</system>' * 3000
+    batch = await submit(env)
+    await complete(env, batch)
+    tool = {t.name: t for t in tools_module.bind_batch_tools(env.service, app_config=SimpleNamespace(tool_output=config))}["read_batch_result"]
+    raw = await tool.coroutine(runtime=runtime(), batch_id=batch, max_chars=8192)
+    page = json.loads(raw)
+    assert page["status"] == "ok" and 0 < len(page["content"]) <= 8192
+    assert len(raw) <= 10_000 and "<system>" not in raw
+    message = ToolMessage(content=raw, tool_call_id="read", name="read_batch_result")
+    patched = _patch_tool_message(message, config, str(tmp_path / "outputs"))
+    assert patched is None or patched.content == raw
+    following_raw = await tool.coroutine(runtime=runtime(), batch_id=batch, offset=page["next_offset"], expected_revision=page["revision"], max_chars=8192)
+    following = json.loads(following_raw)
+    patched = _patch_tool_message(message.model_copy(update={"content": following_raw}), config, str(tmp_path / "outputs"))
+    assert patched is None or patched.content == following_raw
+    document = json.dumps(await env.service.read_batch_item(batch_id=batch, user_id="user-1", thread_id="thread-1", position=0), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    assert page["content"] + following["content"] == document[: following["next_offset"]]
+
+
+@pytest.mark.asyncio
+async def test_output_budget_too_small_stops_without_zero_progress(env):
+    from deerflow.config.tool_output_config import ToolOutputConfig
+
+    batch = await submit(env)
+    tool = {t.name: t for t in tools_module.bind_batch_tools(env.service, app_config=SimpleNamespace(tool_output=ToolOutputConfig(tool_overrides={"read_batch_result": 100})))}["read_batch_result"]
+    assert json.loads(await tool.coroutine(runtime=runtime(), batch_id=batch)) == {"status": "budget_too_small"}
+
+
+@pytest.mark.asyncio
+async def test_serialization_and_hash_run_off_loop(env, monkeypatch):
+    import threading
+
+    batch = await submit(env)
+    loop_thread = threading.get_ident()
+    real = tools_module._batch_result_window
+    observed = []
+
+    def record(*args):
+        observed.append(threading.get_ident())
+        return real(*args)
+
+    monkeypatch.setattr(tools_module, "_batch_result_window", record)
+    assert (await read(env.service, batch))["status"] == "ok"
+    assert len(observed) == 1 and observed[0] != loop_thread
+
+
+@pytest.mark.parametrize("field", ["position", "offset", "max_chars"])
+def test_real_tool_schema_rejects_boolean_integers(field):
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError):
+        reader(AsyncMock()).tool_call_schema.model_validate({"batch_id": "batch", field: True})
+
+
+@pytest.mark.asyncio
+async def test_gateway_tool_assembly_gating_and_config_binding(env, monkeypatch):
+    from deerflow.config.app_config import AppConfig
+    from deerflow.config.tool_output_config import ToolOutputConfig
+    from deerflow.subagents.batch_runtime import set_subagent_batch_submitter
+    from deerflow.tools.tools import get_available_tools
+
+    config = AppConfig(sandbox={"use": "deerflow.sandbox.local:LocalSandboxProvider"}, tool_output=ToolOutputConfig(tool_overrides={"read_batch_result": 500}))
+    batch = await submit(env)
+    set_subagent_batch_submitter(env.service)
+    try:
+        tools = {t.name: t for t in get_available_tools(app_config=config, subagent_enabled=True)}
+        raw = await tools["read_batch_result"].coroutine(runtime=runtime(), batch_id=batch)
+        assert json.loads(raw)["status"] == "ok" and len(raw) <= 500
+        assert "read_batch_result" not in {t.name for t in get_available_tools(app_config=config, subagent_enabled=False)}
+        set_subagent_batch_submitter(None)
+        assert json.loads(await tools["read_batch_result"].coroutine(runtime=runtime(), batch_id=batch)) == {"status": "unavailable"}
+        assert "read_batch_result" not in {t.name for t in get_available_tools(app_config=config, subagent_enabled=True)}
+    finally:
+        set_subagent_batch_submitter(None)
+
+
+@pytest.mark.asyncio
+async def test_runtime_identity_is_not_a_model_argument(env):
+    schema = reader(env.service).tool_call_schema.model_json_schema()["properties"]
+    assert set(schema) == {"batch_id", "position", "offset", "max_chars", "expected_revision"}
+    trusted = runtime("other")
+    trusted.server_info = SimpleNamespace(user=SimpleNamespace(identity="user-1"))
+    batch = await submit(env)
+    assert (await read(env.service, batch, runtime=trusted))["status"] == "ok"

backend/tests/test_create_deerflow_agent.py

@@ -311,7 +311,7 @@ def test_explicit_subagent_runtime_aligns_factory_middleware_and_tools(mock_crea
     assert limit.max_concurrent == 7
     assert limit.max_total == 12
     tool_names = {tool.name for tool in call_kwargs["tools"]}
-    assert {"task", "batch_task", "batch_status", "cancel_batch"} <= tool_names
+    assert {"task", "batch_task", "batch_status", "cancel_batch", "read_batch_result"} <= tool_names
 
 
 def test_explicit_subagent_runtime_requires_the_subagent_feature() -> None:

backend/tests/test_mcp_sync_wrapper.py

@@ -323,9 +323,9 @@ def test_sync_wrapped_builtin_tools_still_resolve_runtime():
         cancel_background_task,
         list_background_tasks,
     )
-    from deerflow.tools.builtins.batch_task_tool import batch_status, cancel_batch
+    from deerflow.tools.builtins.batch_task_tool import batch_status, cancel_batch, read_batch_result
 
-    for tool in (list_background_tasks, cancel_background_task, batch_status, cancel_batch):
+    for tool in (list_background_tasks, cancel_background_task, batch_status, cancel_batch, read_batch_result):
         patched = copy.copy(tool)
         # _ensure_sync_invocable_tool does exactly this to async-only tools.
         patched.func = make_sync_tool_wrapper(patched.coroutine, patched.name)