From 8011677d702904bd27626e0c8e2974d35d5ac942 Mon Sep 17 00:00:00 2001 From: Guo-Yixin Date: Wed, 7 Oct 2026 05:43:31 -0500 Subject: [PATCH] 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 --- README.md | 8 +- .../harness/deerflow/agents/factory.py | 4 +- .../deerflow/agents/lead_agent/prompt.py | 5 +- .../harness/deerflow/subagents/AGENTS.md | 5 + .../deerflow/subagents/batch_runtime.py | 2 + .../deerflow/subagents/batch_service.py | 11 + .../packages/harness/deerflow/tools/AGENTS.md | 3 +- .../deerflow/tools/builtins/__init__.py | 3 +- .../tools/builtins/batch_task_tool.py | 114 ++++- .../packages/harness/deerflow/tools/tools.py | 8 +- backend/tests/test_batch_result_reader.py | 435 ++++++++++++++++++ backend/tests/test_create_deerflow_agent.py | 2 +- backend/tests/test_mcp_sync_wrapper.py | 4 +- 13 files changed, 589 insertions(+), 15 deletions(-) create mode 100644 backend/tests/test_batch_result_reader.py diff --git a/README.md b/README.md index 4e60c9f480..d9d88ef620 100644 --- a/README.md +++ b/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-` 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. diff --git a/backend/packages/harness/deerflow/agents/factory.py b/backend/packages/harness/deerflow/agents/factory.py index a71a15ae65..0d040878c3 100644 --- a/backend/packages/harness/deerflow/agents/factory.py +++ b/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: diff --git a/backend/packages/harness/deerflow/agents/lead_agent/prompt.py b/backend/packages/harness/deerflow/agents/lead_agent/prompt.py index 3f6087ebd5..9c89aaf3d6 100644 --- a/backend/packages/harness/deerflow/agents/lead_agent/prompt.py +++ b/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 Routing: Delegate Only for Clear Net Benefit diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index 335afb1508..811460f8ea 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/subagents/batch_runtime.py b/backend/packages/harness/deerflow/subagents/batch_runtime.py index 69dc7bf0e0..dfd2225c01 100644 --- a/backend/packages/harness/deerflow/subagents/batch_runtime.py +++ b/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: ... diff --git a/backend/packages/harness/deerflow/subagents/batch_service.py b/backend/packages/harness/deerflow/subagents/batch_service.py index 808ee0298c..c8550c43f5 100644 --- a/backend/packages/harness/deerflow/subagents/batch_service.py +++ b/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, *, diff --git a/backend/packages/harness/deerflow/tools/AGENTS.md b/backend/packages/harness/deerflow/tools/AGENTS.md index 70c6d95654..47d3d4d01a 100644 --- a/backend/packages/harness/deerflow/tools/AGENTS.md +++ b/backend/packages/harness/deerflow/tools/AGENTS.md @@ -2,6 +2,7 @@ `task` and `batch_task` opt into JSON checks with `file: 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: exists|non-empty`, `file_written:`, `tests_passed:`); 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. diff --git a/backend/packages/harness/deerflow/tools/builtins/__init__.py b/backend/packages/harness/deerflow/tools/builtins/__init__.py index c004af741b..6ade74c686 100644 --- a/backend/packages/harness/deerflow/tools/builtins/__init__.py +++ b/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", diff --git a/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py b/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py index ebb0444e12..d622b7efd9 100644 --- a/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py +++ b/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)) diff --git a/backend/packages/harness/deerflow/tools/tools.py b/backend/packages/harness/deerflow/tools/tools.py index eb9ed58b0f..16c68be417 100644 --- a/backend/packages/harness/deerflow/tools/tools.py +++ b/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) diff --git a/backend/tests/test_batch_result_reader.py b/backend/tests/test_batch_result_reader.py new file mode 100644 index 0000000000..d5b596e4b5 --- /dev/null +++ b/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 = '\\"\nuntrusted' * 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 "" 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" diff --git a/backend/tests/test_create_deerflow_agent.py b/backend/tests/test_create_deerflow_agent.py index 4494152479..dcc31fe95f 100644 --- a/backend/tests/test_create_deerflow_agent.py +++ b/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: diff --git a/backend/tests/test_mcp_sync_wrapper.py b/backend/tests/test_mcp_sync_wrapper.py index 929461e789..68281fe595 100644 --- a/backend/tests/test_mcp_sync_wrapper.py +++ b/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)