diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index 28c86b1e38..57f2320745 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -94,6 +94,69 @@ _explicit_app_config: ContextVar[Any | None] = ContextVar( ) +def _preflight_delegation_checks( + *, + subagent_type: str, + context_mode: str, + runtime_app_config: Any | None, + allowed_subagents: Any | None, +) -> tuple[Any | None, str | None]: + """Read-only preflight (validate-then-mutate). + + Runs every precondition check in one pass before any side-effecting + delegation work (tool assembly, SubagentExecutor creation, background task + registration) is triggered. + + This helper has zero side effects of its own: + - Does not create a SubagentExecutor + - Does not register background tasks + - Does not assemble MCP/business tools + - Does not snapshot the parent context + - Does not preempt or cancel any source task + + Returns (config, None) when every check passes; returns (None, error_str) + when rejected, and the caller should return the failure directly. The + helper's internal state is unchanged, so callers can retry safely. + Note: before this helper runs, the caller has already registered a child + gateway entry on RunInputControl; that entry can be reused for benign + retries and is not a side effect of this helper itself. + + If the recovery feature lands later, add read-only source-task + recoverability checks here (type compatibility, extractable history, + non-empty), and have the caller run supersede_and_claim_resumable_task only + after all checks pass, so a rejected recovery request does not consume the + source task (see issue #7633). + """ + config: Any | None = None + + if context_mode not in {"isolated", "snapshot"}: + return None, f"Unknown context_mode '{context_mode}'. Use isolated or snapshot." + + if subagent_type == "bash": + host_bash_allowed = is_host_bash_allowed(runtime_app_config) if runtime_app_config is not None else is_host_bash_allowed() + if not host_bash_allowed: + return None, LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE + + # Catalog discovery only after the cheapest guards pass: an invalid + # context_mode must not reach the managed-subagent store lookup. + if allowed_subagents is None: + available_subagent_names = get_available_subagent_names(app_config=runtime_app_config) if runtime_app_config is not None else get_available_subagent_names() + else: + available_subagent_names = get_available_subagent_names(app_config=runtime_app_config, allowed_subagents=allowed_subagents) if runtime_app_config is not None else get_available_subagent_names(allowed_subagents=allowed_subagents) + + config = get_subagent_config(subagent_type, app_config=runtime_app_config) if runtime_app_config is not None else get_subagent_config(subagent_type) + if config is None or subagent_type not in available_subagent_names: + if available_subagent_names: + available = ", ".join(available_subagent_names) + elif allowed_subagents is not None: + available = "none permitted by caller policy" + else: + available = "none" + return None, f"Unknown subagent type '{subagent_type}'. Available: {available}" + + return config, None + + def _record_middleware_on_parent_loop(journal: Any, kwargs: dict[str, Any]) -> None: """Run one subagent middleware-journal append on the journal owner's loop.""" try: @@ -880,42 +943,30 @@ async def task_tool( input_control = run_input_control(parent_context) input_gate = input_control.child(tool_call_id) if input_control is not None else None result_command = partial(_task_result_command, message_id=input_gate.result_id) if input_gate is not None else _task_result_command - if context_mode not in {"isolated", "snapshot"}: - return result_command(tool_call_id=tool_call_id, status="failed", error=f"Unknown context_mode '{context_mode}'. Use isolated or snapshot.") runtime_app_config = _get_runtime_app_config(runtime) metadata: dict = runtime.config.get("metadata", {}) if runtime is not None else {} allowed_subagents = metadata.get("allowed_subagents") - if allowed_subagents is None: - available_subagent_names = get_available_subagent_names(app_config=runtime_app_config) if runtime_app_config is not None else get_available_subagent_names() - else: - available_subagent_names = get_available_subagent_names(app_config=runtime_app_config, allowed_subagents=allowed_subagents) if runtime_app_config is not None else get_available_subagent_names(allowed_subagents=allowed_subagents) - # Preserve the dedicated sandbox-policy guidance before the generic - # registry/policy membership gate filters bash from the visible catalog. - if subagent_type == "bash": - host_bash_allowed = is_host_bash_allowed(runtime_app_config) if runtime_app_config is not None else is_host_bash_allowed() - if not host_bash_allowed: - return result_command( - tool_call_id=tool_call_id, - status="failed", - error=LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE, - ) - - # Get subagent configuration - config = get_subagent_config(subagent_type, app_config=runtime_app_config) if runtime_app_config is not None else get_subagent_config(subagent_type) - if config is None or subagent_type not in available_subagent_names: - if available_subagent_names: - available = ", ".join(available_subagent_names) - elif allowed_subagents is not None: - available = "none permitted by caller policy" - else: - available = "none" - error = f"Unknown subagent type '{subagent_type}'. Available: {available}" + # Preflight: read-only checks with zero side effects. Complete every + # precondition check in one pass before any side-effecting delegation work + # (tool assembly, executor creation, background task registration). Return + # immediately on failure; the helper's internal state is unchanged, so the + # caller can retry safely. The context_mode guard is the first branch + # executed inside the helper, and catalog discovery was moved after the + # guard so an invalid mode never triggers a managed-subagent store access. + preflight_config, preflight_error = _preflight_delegation_checks( + subagent_type=subagent_type, + context_mode=context_mode, + runtime_app_config=runtime_app_config, + allowed_subagents=allowed_subagents, + ) + if preflight_error is not None: return result_command( tool_call_id=tool_call_id, status="failed", - error=error, + error=preflight_error, ) + config = preflight_config # Rejected delegations must not serialize the retained history. Capture # after delegation validation, before child setup (including tool loading). context_snapshot = ParentContextSnapshot.from_state(runtime.state) if context_mode == "snapshot" and runtime is not None else None diff --git a/backend/tests/test_run_input_graph.py b/backend/tests/test_run_input_graph.py index b451a5d3fe..73cd34c3bb 100644 --- a/backend/tests/test_run_input_graph.py +++ b/backend/tests/test_run_input_graph.py @@ -6,6 +6,7 @@ from typing import Annotated import pytest from langchain.agents import create_agent +from langchain.tools import ToolRuntime from langchain_core.language_models import BaseChatModel from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage, ToolMessage from langchain_core.outputs import ChatGeneration, ChatResult @@ -119,7 +120,7 @@ async def test_native_task_pending_and_final_results_share_an_id_per_invocation( inputs = RunInputs(store, "run", "owner", "worker") await inputs.submit("update", "Handle this first", "steer") control = RunInputControl(inputs, agent_runs=None, thread_id="thread", checkpointer=None) - runtime = SimpleNamespace(context={INPUT_CONTEXT_KEY: control}) + runtime = ToolRuntime(state={}, context={INPUT_CONTEXT_KEY: control}, config={}, stream_writer=lambda _: None, tools=[], tool_call_id="reused", store=None) pending_ids, results = [], [] try: for _ in range(2): diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py index fd91c18088..ec98aa0465 100644 --- a/backend/tests/test_task_tool_core_logic.py +++ b/backend/tests/test_task_tool_core_logic.py @@ -3759,3 +3759,143 @@ def test_parallel_task_choices_do_not_mutate_parent_or_shared_config(monkeypatch assert {entry["config"].model for entry in seen} == {"B", "C"} assert shared.model == "inherit" assert runtime.config["metadata"]["model_name"] == "ark-model" + + +def _preflight(**overrides): + kwargs = { + "subagent_type": "general-purpose", + "context_mode": "isolated", + "runtime_app_config": None, + "allowed_subagents": None, + } + kwargs.update(overrides) + return task_tool_module._preflight_delegation_checks(**kwargs) + + +def test_preflight_resolves_catalog_after_guards(monkeypatch): + """Illegal context_mode must short-circuit before the catalog store lookup. + + This guards against the regression where catalog discovery was hoisted out + of the helper and ran before any guard, so a typo in context_mode could + hit the managed-subagent store instead of returning failed.""" + get_names = MagicMock(return_value=["general-purpose"]) + monkeypatch.setattr(task_tool_module, "get_available_subagent_names", get_names) + + config, error = _preflight(context_mode="thread") + + assert config is None + assert error == "Unknown context_mode 'thread'. Use isolated or snapshot." + get_names.assert_not_called() + + +def test_preflight_returns_the_resolved_config_on_success(monkeypatch): + resolved = _make_subagent_config() + monkeypatch.setattr(task_tool_module, "get_available_subagent_names", lambda **kwargs: ["general-purpose"]) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda *args, **kwargs: resolved) + + config, error = _preflight() + + assert error is None + assert config is resolved + + +def test_preflight_rejects_unknown_context_mode(monkeypatch): + get_config = MagicMock() + monkeypatch.setattr(task_tool_module, "get_subagent_config", get_config) + + config, error = _preflight(context_mode="thread") + + assert config is None + assert error == "Unknown context_mode 'thread'. Use isolated or snapshot." + # The cheapest guard must fail before any registry lookup. + get_config.assert_not_called() + + +def test_preflight_rejects_bash_when_host_bash_is_disabled(monkeypatch): + monkeypatch.setattr(task_tool_module, "is_host_bash_allowed", lambda: False) + + config, error = _preflight(subagent_type="bash") + + assert config is None + assert error == LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE + + +def test_preflight_allows_bash_when_host_bash_is_enabled(monkeypatch): + resolved = _make_subagent_config(name="bash") + monkeypatch.setattr(task_tool_module, "is_host_bash_allowed", lambda: True) + monkeypatch.setattr(task_tool_module, "get_available_subagent_names", lambda **kwargs: ["bash"]) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda *args, **kwargs: resolved) + + config, error = _preflight(subagent_type="bash") + + assert error is None + assert config is resolved + + +def test_preflight_rejects_a_subagent_outside_the_caller_policy(monkeypatch): + monkeypatch.setattr(task_tool_module, "get_available_subagent_names", lambda **kwargs: ["planner"]) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda *args, **kwargs: _make_subagent_config()) + + config, error = _preflight(allowed_subagents=["planner"]) + + assert config is None + assert error == "Unknown subagent type 'general-purpose'. Available: planner" + + +def test_preflight_rejects_when_no_subagent_is_permitted(monkeypatch): + monkeypatch.setattr(task_tool_module, "get_available_subagent_names", lambda **kwargs: []) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda *args, **kwargs: _make_subagent_config()) + + config, error = _preflight(allowed_subagents=[]) + + assert config is None + assert error == "Unknown subagent type 'general-purpose'. Available: none permitted by caller policy" + + +def test_rejected_preflight_creates_no_executor_and_assembles_no_tools(monkeypatch): + """Validate-then-mutate: a rejected delegation must not reach tool assembly + or executor construction.""" + monkeypatch.setattr(task_tool_module, "get_available_subagent_names", lambda **kwargs: ["general-purpose"]) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda *args, **kwargs: None) + executor = MagicMock() + assemble = MagicMock() + monkeypatch.setattr(task_tool_module, "SubagentExecutor", executor) + monkeypatch.setattr("deerflow.tools.get_available_tools", assemble) + + message = _task_tool_message(_run_task_tool(runtime=_make_runtime(), prompt="do work", subagent_type="general-purpose", tool_call_id="tc-rejected")) + + assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "failed" + executor.assert_not_called() + assemble.assert_not_called() + + +def test_task_tool_binds_the_preflight_config_on_the_happy_path(monkeypatch): + """Regression: the preflight helper must hand its resolved config back to + task_tool. When config stayed scoped inside the helper, every successful + delegation raised UnboundLocalError at the skills/model lookups while only + the rejection paths kept passing.""" + runtime = _make_runtime() + resolved = _make_subagent_config() + assert resolved.model == "inherit" # the branch hit on every success path + captured = {} + + class DummyExecutor: + def __init__(self, **kwargs): + captured["config"] = kwargs["config"] + + def execute_async(self, prompt, task_id=None): + return task_id or "generated-task-id" + + monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus) + monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor) + monkeypatch.setattr(task_tool_module, "get_available_subagent_names", lambda **kwargs: ["general-purpose"]) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda *args, **kwargs: resolved) + monkeypatch.setattr(task_tool_module, "get_background_task_result", lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done")) + monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _: None) + monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep) + monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: []) + + message = _task_tool_message(_run_task_tool(runtime=runtime, description="happy path", prompt="do work", subagent_type="general-purpose", tool_call_id="tc-happy")) + + assert message.additional_kwargs[SUBAGENT_STATUS_KEY] == "completed" + assert captured["config"].model == "inherit"