diff --git a/README.md b/README.md index 76d887d23b..8c60cdff3f 100644 --- a/README.md +++ b/README.md @@ -2474,9 +2474,10 @@ Backend response cleanup preserves unrelated tag names such as `` an The Web UI composer can polish draft input before sending. The rewrite runs as a short Gateway LLM request using the `input_polish` model configuration, keeps slash skill prefixes such as `/data-analysis`, and only replaces the local draft after the user clicks the polish button; it does not create a thread run or persist a message. -Input polishing, follow-up suggestions, and background project-file summaries -construct their model providers off the Gateway event loop, so local credential -reads do not stall unrelated requests. +Input polishing, follow-up suggestions, background project-file summaries, the +LLM skill scanner, and the `/goal` evaluator construct their model providers off +the Gateway event loop, so local credential reads do not stall unrelated +requests. When the agent asks for clarification, the Web UI shows the structured response card but keeps the normal composer available. Users can complete the card or send a free-form chat message to bypass it; that message closes the latest pending clarification and becomes the agent's next input. Accompanying answer text stays outside the execution steps panel, and answering the request keeps previously completed text in the conversation while the agent continues. diff --git a/backend/packages/harness/deerflow/runtime/goal.py b/backend/packages/harness/deerflow/runtime/goal.py index c7fb367919..0247577aed 100644 --- a/backend/packages/harness/deerflow/runtime/goal.py +++ b/backend/packages/harness/deerflow/runtime/goal.py @@ -31,6 +31,7 @@ from deerflow.models import create_chat_model from deerflow.runtime.goal_progress import previous_goal_operations from deerflow.runtime.keyed_lock import AsyncKeyedLockTable from deerflow.tracing import inject_langfuse_metadata +from deerflow.utils.assembly_io import run_assembly from deerflow.utils.file_io import await_drained from deerflow.utils.goal_objective import MAX_GOAL_OBJECTIVE_CHARS as MAX_GOAL_OBJECTIVE_CHARS from deerflow.utils.goal_objective import normalize_goal_objective as normalize_goal_objective @@ -430,6 +431,9 @@ def create_goal_evaluator_model( call to inherit tracing from. It must attach its own model-level tracing callbacks, same as the other standalone, non-graph callers (``oneshot_llm.run_oneshot_llm``, ``MemoryUpdater``). + + Provider construction can read local credentials, so async callers build + the evaluator through ``run_assembly`` instead of on the event loop. """ return create_chat_model( name=model_name, @@ -509,7 +513,7 @@ async def evaluate_goal_completion( user_content = f"Active goal:\n{goal['objective']}\n\nVisible conversation evidence:\n{conversation}\n\nIs the active goal fully satisfied?" if model is None: - model = create_goal_evaluator_model(model_name=model_name, app_config=app_config) + model = await run_assembly(create_goal_evaluator_model, model_name=model_name, app_config=app_config) invoke_config: dict[str, Any] = {"run_name": "goal_evaluator"} if usage_callback is not None: # This critic must not inherit graph callbacks: its usage crosses the diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 09589378d7..0033efeb61 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -2825,7 +2825,9 @@ async def _prepare_goal_continuation_input( return None try: - evaluator_model = evaluator_model_factory() if evaluator_model_factory is not None else None + # The factory builds the evaluator on first use, which can read local + # credentials; keep that off the event loop (see utils/AGENTS.md). + evaluator_model = await run_assembly(evaluator_model_factory) if evaluator_model_factory is not None else None for attempt in range(2): try: evaluation = await evaluate_goal_completion( diff --git a/backend/packages/harness/deerflow/skills/security_scanner.py b/backend/packages/harness/deerflow/skills/security_scanner.py index 35a6ff3322..20c83f3c0d 100644 --- a/backend/packages/harness/deerflow/skills/security_scanner.py +++ b/backend/packages/harness/deerflow/skills/security_scanner.py @@ -15,6 +15,7 @@ from deerflow.models import create_chat_model from deerflow.runtime.user_context import get_effective_user_id from deerflow.skills.types import SKILL_MD_FILE from deerflow.tracing import inject_langfuse_metadata +from deerflow.utils.assembly_io import run_assembly from deerflow.utils.llm_text import extract_response_text logger = logging.getLogger(__name__) @@ -27,10 +28,9 @@ class ScanResult: available: bool = True -def _resolve_fail_closed(app_config: AppConfig | None) -> bool: +def _resolve_fail_closed(config: AppConfig | None) -> bool: """Resolve the fail-closed policy, defaulting to True if config is unavailable.""" try: - config = app_config or get_app_config() return bool(getattr(config.skill_evolution, "security_fail_closed", True)) except Exception: return True @@ -126,11 +126,16 @@ async def scan_skill_content( prompt = f"Location: {location}\nExecutable: {str(executable).lower()}\nDeterministic SkillScan findings:\n{_format_static_findings_context(static_findings or [])}\n\nReview this content:\n-----\n{content}\n-----" model_responded = False + config = app_config try: - config = app_config or get_app_config() + # Off the event loop (see utils/AGENTS.md): reading the live config + # stats and hashes config.yaml, and provider construction can read + # local credentials. The fail-closed fallback below reuses this config. + if config is None: + config = await run_assembly(get_app_config) model_name = config.skill_evolution.moderation_model_name model_kwargs = {"thinking_enabled": False, "app_config": config, "attach_tracing": attach_tracing} - model = create_chat_model(name=model_name, **model_kwargs) if model_name else create_chat_model(**model_kwargs) + model = await run_assembly(create_chat_model, name=model_name, **model_kwargs) if model_name else await run_assembly(create_chat_model, **model_kwargs) invoke_config: dict[str, Any] = {"run_name": "security_agent"} if attach_tracing: # Standalone callers own the trace root, so they must inject their own @@ -175,7 +180,7 @@ async def scan_skill_content( return ScanResult("block", "Security scan produced unparseable output; manual review required.", available=False) if executable: return ScanResult("block", "Security scan unavailable for executable content; manual review required.", available=False) - if _resolve_fail_closed(app_config): + if _resolve_fail_closed(config): return ScanResult("block", "Security scan unavailable for skill content; manual review required.", available=False) logger.warning("Security scan unavailable; failing open for non-executable skill content at %s (manual review recommended)", location) return ScanResult("warn", "Security scan unavailable for non-executable skill content; manual review recommended.", available=False) diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index c9540c4f66..235cbeeedf 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -120,14 +120,20 @@ Display summaries use `strip_leading_think_blocks` before limiting text; ### Agent / Tool Assembly Off-Load -One-shot input-polish, follow-up-suggestion, and project-file-summary helpers -build providers through `run_assembly` before awaiting inference. Provider -construction can read local credentials; keep it off the event loop and the -default executor while preserving request context. Cancellation during -construction must not start inference. This contract covers callers of -`oneshot_llm.py`; standalone skill-moderation and goal-evaluator constructors -are separate sites and are not covered by this offload. -Coverage: `tests/blocking_io/test_oneshot_llm_offloop.py`. +One-shot input-polish, follow-up-suggestion, and project-file-summary helpers, +skill moderation (`scan_skill_content`) and the goal evaluator build providers +through `run_assembly` before awaiting inference; the run worker calls its +cached evaluator factory the same way. Provider construction can read local +credentials; keep it off the event loop and the default executor while +preserving request context. Cancellation during construction must not start +inference. A scan without `app_config` (in-graph `skill_manage`) reads the live +config in the same pool, and its fail-closed fallback reuses that config rather +than reading it again on the loop. Other model construction is separate and +not covered here: manual compaction (`compact_thread_context` builds the +summarization middleware and its models inline) and, inside graph runs, title +generation, summarization and subagent construction. +Coverage: `tests/blocking_io/test_oneshot_llm_offloop.py` and +`tests/blocking_io/test_system_model_offloop.py`. `start_run` also resolves the factory through `run_assembly` before admission: cold imports cannot occupy the loop or its default pool, and import errors diff --git a/backend/tests/blocking_io/test_system_model_offloop.py b/backend/tests/blocking_io/test_system_model_offloop.py new file mode 100644 index 0000000000..a50bed1334 --- /dev/null +++ b/backend/tests/blocking_io/test_system_model_offloop.py @@ -0,0 +1,304 @@ +"""Skill-moderation and goal-evaluator models must not read credentials on the loop. + +Both sites build a provider right before a standalone model call: the skill +security scan behind the Gateway skill routes, the installer and the in-graph +``skill_manage`` tool, and the goal evaluator that runs after Gateway runs. +Provider construction can read local credentials (Codex CLI ``auth.json``); +the real Codex provider is built here so that IO stays inside the strict gate, +while tracing and inference stay offline. +""" + +import asyncio +import json +import threading +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from langchain_core.messages import AIMessage, HumanMessage +from langgraph.checkpoint.base import empty_checkpoint +from langgraph.checkpoint.memory import InMemorySaver + +# evaluate_goal_completion imports these lazily. A Gateway run has loaded them +# long before its goal is evaluated, so a cold import here would only time the +# import itself inside the strict gate. +import deerflow.agents.middlewares.memory_middleware # noqa: F401 +import deerflow.agents.middlewares.pii_redaction_middleware # noqa: F401 +import deerflow.tools.skill_manage_tool as skill_manage_module +from deerflow.config.app_config import AppConfig, get_app_config, reset_app_config +from deerflow.config.extensions_config import reset_extensions_config +from deerflow.models import factory +from deerflow.models.openai_codex_provider import CodexChatModel +from deerflow.runtime import goal +from deerflow.runtime.checkpoint_state import CheckpointStateAccessor, build_state_mutation_graph +from deerflow.runtime.runs import worker +from deerflow.runtime.user_context import get_effective_user_id, reset_current_user, set_current_user +from deerflow.skills import security_scanner + +pytestmark = pytest.mark.asyncio + +_SCAN_VERDICT = '{"decision": "allow", "reason": "clean"}' +_GOAL_VERDICT = json.dumps({"satisfied": True, "blocker": "none", "reason": "Reported.", "evidence_summary": "The report was sent."}) +_CONFIG_YAML = """\ +sandbox: + use: deerflow.sandbox.local:LocalSandboxProvider +models: + - name: fixture-codex + model: fixture-model + use: deerflow.models.openai_codex_provider:CodexChatModel +""" + + +def _codex_config(**overrides) -> AppConfig: + return AppConfig.model_validate( + { + "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}, + "models": [{"name": "fixture-codex", "model": "fixture-model", "use": "deerflow.models.openai_codex_provider:CodexChatModel"}], + **overrides, + } + ) + + +@pytest.fixture +def codex(tmp_path, monkeypatch): + """Real Codex construction (credential file IO included); offline tracing and inference.""" + credentials = tmp_path / "auth.json" + credentials.write_text(json.dumps({"access_token": "fixture-token", "account_id": "fixture-account"}), encoding="utf-8") + monkeypatch.setenv("CODEX_AUTH_PATH", str(credentials)) + monkeypatch.setattr(factory, "build_tracing_callbacks", lambda **kwargs: []) + state = SimpleNamespace(reply="", calls=[]) + + async def invoke(model, messages, *, config): + state.calls.append((model, threading.get_ident())) + return AIMessage(content=state.reply) + + monkeypatch.setattr(CodexChatModel, "ainvoke", invoke) + return state + + +def _scan(config): + return security_scanner.scan_skill_content("# Fixture skill", app_config=config) + + +def _evaluate(config): + return goal.evaluate_goal_completion(goal.build_goal_state("Report"), [HumanMessage(content="Report"), AIMessage(content="Done.")], model_name="fixture-codex", app_config=config) + + +_SITES = [ + pytest.param(SimpleNamespace(module=security_scanner, call=_scan, verdict=_SCAN_VERDICT), id="skill-moderation"), + pytest.param(SimpleNamespace(module=goal, call=_evaluate, verdict=_GOAL_VERDICT), id="goal-evaluator"), +] + + +@pytest.mark.parametrize("moderation_model_name", ["fixture-codex", None], ids=["named-model", "default-model"]) +async def test_skill_scan_builds_codex_model_off_loop(codex, moderation_model_name): + codex.reply = _SCAN_VERDICT + + result = await _scan(_codex_config(skill_evolution={"moderation_model_name": moderation_model_name})) + + # On the loop, the gate rejects the credential read and the scan falls back + # to its fail-closed "unavailable" block instead of the model's verdict. + assert (result.decision, result.reason, result.available) == ("allow", "clean", True) + assert len(codex.calls) == 1 + assert isinstance(codex.calls[0][0], CodexChatModel) + assert codex.calls[0][1] == threading.get_ident() + + +@pytest.mark.parametrize("warm_cache", [False, True], ids=["cold-config", "cached-config"]) +async def test_in_graph_skill_scan_reads_live_config_off_loop(tmp_path, monkeypatch, codex, warm_cache): + config_path = tmp_path / "config.yaml" + extensions_path = tmp_path / "extensions_config.json" + + def seed_files() -> None: + config_path.write_text(_CONFIG_YAML, encoding="utf-8") + extensions_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8") + + await asyncio.to_thread(seed_files) + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setattr("deerflow.config.paths._paths", None) + monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) + reset_app_config() + reset_extensions_config() + codex.reply = _SCAN_VERDICT + try: + if warm_cache: + await asyncio.to_thread(get_app_config) + # skill_manage's in-graph scan passes no app_config, so the scanner + # resolves the live config, which stats and hashes config.yaml. + verdict = await skill_manage_module._scan_or_raise("# Fixture skill", executable=False, location="fixture-skill/SKILL.md") + finally: + reset_app_config() + reset_extensions_config() + + assert verdict == {"decision": "allow", "reason": "clean"} + assert len(codex.calls) == 1 + assert codex.calls[0][1] == threading.get_ident() + + +async def test_unreadable_live_config_fails_closed_without_a_loop_read(tmp_path, monkeypatch): + monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(tmp_path / "missing.yaml")) + reset_app_config() + reads = [] + + def read_live_config(): + reads.append(threading.get_ident()) + return get_app_config() + + monkeypatch.setattr(security_scanner, "get_app_config", read_live_config) + try: + result = await security_scanner.scan_skill_content("# Fixture skill", attach_tracing=False) + finally: + reset_app_config() + + assert (result.decision, result.available) == ("block", False) + assert result.reason == "Security scan unavailable for skill content; manual review required." + # One read, in the assembly pool: the fail-closed fallback does not retry it on the loop. + assert len(reads) == 1 + assert reads[0] != threading.get_ident() + + +async def test_goal_evaluator_builds_codex_model_off_loop(codex): + codex.reply = _GOAL_VERDICT + + evaluation = await _evaluate(_codex_config()) + + assert evaluation["satisfied"] is True + assert evaluation["reason"] == "Reported." + assert len(codex.calls) == 1 + assert isinstance(codex.calls[0][0], CodexChatModel) + assert codex.calls[0][1] == threading.get_ident() + + +class _Bridge: + async def publish(self, _run_id, _event, _payload): + pass + + +async def test_worker_builds_goal_evaluator_off_loop(codex): + codex.reply = _GOAL_VERDICT + config = _codex_config() + checkpointer = InMemorySaver() + thread_id = "goal-thread" + checkpoint = empty_checkpoint() + checkpoint["channel_values"] = {"messages": [HumanMessage(content="Send the report."), AIMessage(content="The report was sent.")]} + checkpoint["channel_versions"] = {"messages": 1} + checkpointer.put({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}, checkpoint, {"step": 1}, {"messages": 1}) + await goal.write_thread_goal(checkpointer, thread_id, goal.build_goal_state("Send the report")) + factory_threads = [] + + def build_evaluator(): + # Mirrors run_agent's cached factory around create_goal_evaluator_model. + factory_threads.append(threading.get_ident()) + return goal.create_goal_evaluator_model(model_name="fixture-codex", app_config=config) + + candidate = await worker._prepare_goal_continuation_input( + accessor=CheckpointStateAccessor.bind(build_state_mutation_graph("goal_evaluator", "full"), checkpointer, mode="full"), + bridge=_Bridge(), + checkpointer=checkpointer, + thread_id=thread_id, + run_id="run-1", + model_name="fixture-codex", + app_config=config, + evaluator_model_factory=build_evaluator, + ) + + # On the loop, the gate rejects the factory's credential read and the goal + # stands down as "evaluator_failed" without a verdict. + assert isinstance(candidate, worker._GoalCompletionCandidate) + assert candidate.evaluation["satisfied"] is True + assert len(factory_threads) == 1 + assert factory_threads[0] != threading.get_ident() + assert len(codex.calls) == 1 + assert codex.calls[0][1] == threading.get_ident() + + +@pytest.mark.parametrize("site", _SITES) +async def test_construction_preserves_context_and_keeps_default_pool_available(monkeypatch, site): + loop = asyncio.get_running_loop() + loop.set_default_executor(ThreadPoolExecutor(max_workers=1)) + started = asyncio.Event() + release = threading.Event() + observed = [] + model = SimpleNamespace(ainvoke=AsyncMock(return_value=AIMessage(content=site.verdict))) + + def construct(**kwargs): + observed.append((get_effective_user_id(), threading.get_ident())) + loop.call_soon_threadsafe(started.set) + if not release.wait(timeout=10): + raise TimeoutError("test construction was not released") + return model + + monkeypatch.setattr(site.module, "create_chat_model", construct) + token = set_current_user(SimpleNamespace(id="request-owner")) + task = asyncio.create_task(site.call(_codex_config())) + try: + await asyncio.wait_for(started.wait(), timeout=5) + assert not task.done() + assert await asyncio.wait_for(asyncio.to_thread(lambda: "unrelated work"), timeout=5) == "unrelated work" + finally: + release.set() + try: + await asyncio.wait_for(task, timeout=5) + finally: + reset_current_user(token) + + assert observed == [("request-owner", observed[0][1])] + assert observed[0][1] != threading.get_ident() + model.ainvoke.assert_awaited_once() + + +@pytest.mark.parametrize("site", _SITES) +async def test_cancelled_construction_does_not_start_inference(monkeypatch, site): + loop = asyncio.get_running_loop() + started = asyncio.Event() + finished = asyncio.Event() + release = threading.Event() + model = SimpleNamespace(ainvoke=AsyncMock()) + + def construct(**kwargs): + loop.call_soon_threadsafe(started.set) + try: + if not release.wait(timeout=10): + raise TimeoutError("test construction was not released") + return model + finally: + loop.call_soon_threadsafe(finished.set) + + monkeypatch.setattr(site.module, "create_chat_model", construct) + task = asyncio.create_task(site.call(_codex_config())) + try: + await asyncio.wait_for(started.wait(), timeout=5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + finally: + release.set() + await asyncio.wait_for(finished.wait(), timeout=5) + model.ainvoke.assert_not_awaited() + + +async def test_moderation_construction_failure_still_fails_closed(monkeypatch): + def construct(**kwargs): + raise ValueError("fixture credential unavailable") + + monkeypatch.setattr(security_scanner, "create_chat_model", construct) + + result = await _scan(_codex_config()) + + assert (result.decision, result.available) == ("block", False) + assert result.reason == "Security scan unavailable for skill content; manual review required." + + +async def test_evaluator_construction_failure_propagates(monkeypatch): + failure = ValueError("fixture credential unavailable") + + def construct(**kwargs): + raise failure + + monkeypatch.setattr(goal, "create_chat_model", construct) + + with pytest.raises(ValueError) as caught: + await _evaluate(_codex_config()) + assert caught.value is failure