From 05db1d66a0cdfbb3d73155b654b106d3e94a390a Mon Sep 17 00:00:00 2001 From: AochenShen99 Date: Sun, 18 Oct 2026 20:26:17 +0800 Subject: [PATCH] fix(workflows): stop idle agent waits from consuming worker turns (#7024) * fix(workflows): stop idle agent waits from consuming worker turns While a script step waited on a child Agent run, the supervisor sent an empty response frame to the QuickJS worker every 20 ms and each round trip counted toward the 10,000-turn cap, so any workflow whose Agent calls kept it waiting for about 209 s failed with "Worker turn limit" regardless of `timeout_seconds`. The worker now reports when its job queue has drained, and the supervisor waits on the pending calls locally while the worker is idle, still re-checking cancellation and faults every 20 ms, without exchanging frames. `timeout_seconds` is the only wall-clock bound again; scripts that spin the job queue still hit the turn cap. The cap is a module constant so tests can lower it. Fixes bytedance/deer-flow#7022 * fix(workflows): default a frame without idle to the polling path Review follow-up: read `idle` with `frame.get("idle", False)` so an outcome-less frame that lacks the key degrades to the previous poll behaviour instead of failing the run with a KeyError, and give the cancel-latency regression test a 3 s budget so store writes and process reaping on a loaded CI runner cannot flake it. * fix(workflows): detect worker exit during idle Agent waits Check worker exit in the existing local poll and clean up pending child runs. Adapt the work-frame budget regression from PR #7025 while retaining precise idle detection. --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 9 ++ backend/tests/test_workflow_plugin.py | 105 ++++++++++++++++++ .../deerflow-extension-workflows/AGENTS.md | 3 + .../deerflow-extension-workflows/README.md | 7 ++ .../deerflow_extension_workflows/service.py | 21 ++-- .../deerflow_extension_workflows/worker.py | 5 +- 6 files changed, 142 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 43a7e49d8b..d2e59cdac5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1265,6 +1265,15 @@ This release closes that milestone with **621 merged pull requests**. ### Fixed +- **workflows:** A script workflow waiting on a long child Agent run no longer + fails with `Worker turn limit` after about 209 seconds regardless of + `timeout_seconds`. The supervisor re-sent an empty frame to the QuickJS worker + every 20 ms poll while a step was pending, and each round trip counted toward + the 10,000-turn cap. The worker now reports when its job queue has drained, + and the supervisor then waits on the pending calls locally (still checking + cancellation, faults and worker exits every 20 ms) without exchanging frames, + so idle waits no longer consume turns and `timeout_seconds` is the only wall-clock bound; + scripts that spin the job queue still hit the turn cap. (issue #7022) - **browser:** `browser_screenshot` no longer replaces a file that already exists in the thread's outputs directory. A screenshot saved as `report.png` used to overwrite the user's own `report.png`, an earlier diff --git a/backend/tests/test_workflow_plugin.py b/backend/tests/test_workflow_plugin.py index 454c89bdc1..0af1678ceb 100644 --- a/backend/tests/test_workflow_plugin.py +++ b/backend/tests/test_workflow_plugin.py @@ -3,6 +3,8 @@ import asyncio import json import os +import sys +import time from pathlib import Path import pytest @@ -510,6 +512,109 @@ async def test_workflow_deadline_is_failed_and_cleans_up(plugin, with_child): assert all(run.status == "interrupted" for run in runs.runs.values()) +def count_sends(service, monkeypatch): + sent = [] + original = service.send + + async def send(process, message): + sent.append(message) + await original(process, message) + + monkeypatch.setattr(service, "send", send) + return sent + + +@pytest.mark.asyncio +async def test_pending_agent_wait_does_not_consume_worker_turns(plugin, monkeypatch): + service, runs = plugin[2], plugin[3] + monkeypatch.setattr(sys.modules[type(service).__module__], "TURN_LIMIT", 20) + sent = count_sends(service, monkeypatch) + result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});") + await asyncio.wait_for(runs.entered.wait(), 5) + await asyncio.sleep(3) + runs.release.set() + final = await done(plugin, result["run_id"]) + assert final["status"] == "completed", final + # Idle waiting must not round-trip to the worker every poll interval. + assert len(sent) <= 5, len(sent) + + +@pytest.mark.asyncio +async def test_worker_exit_while_agent_pending_fails_promptly(plugin): + _, actions, service, runs = plugin + service.config["timeout_seconds"] = 60 + result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});") + await asyncio.wait_for(runs.entered.wait(), 5) + assert runs.active == 1 + process = service.processes[result["run_id"]] + process.kill() + await asyncio.wait_for(process.wait(), 3) + + # A dead idle worker must fail before the workflow's 60-second deadline, + # without waiting for the blocked Agent to complete by itself. + await asyncio.wait_for(service.wait_idle(result["run_id"]), 3) + final = await actions["get"]({"run_id": result["run_id"]}, context(runs)) + assert final["status"] == "failed", final + assert final["error"] == "Worker exited while waiting for Agent calls" + assert final["steps"][0]["child_status"] == "interrupted" + assert all(run.status == "interrupted" for run in runs.runs.values()) + assert not service.processes + + +@pytest.mark.asyncio +async def test_work_frames_still_count_against_worker_turn_budget(plugin, monkeypatch): + """Work-carrying frames retain the turn cap (regression from PR #7025).""" + service = plugin[2] + monkeypatch.setattr(sys.modules[type(service).__module__], "TURN_LIMIT", 2) + result = await launch( + plugin, + """ + await agent({stepId:'a',agentRef:'lead_agent',task:'one'}); + await agent({stepId:'b',agentRef:'lead_agent',task:'two'}); + return 1; + """, + ) + final = await done(plugin, result["run_id"]) + assert final["status"] == "failed", final + assert final["error"] == "Worker turn limit" + assert not service.processes + + +@pytest.mark.asyncio +async def test_cancel_while_agent_pending_stops_promptly(plugin): + _, actions, service, runs = plugin + result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});") + await asyncio.wait_for(runs.entered.wait(), 5) + await asyncio.sleep(0.3) + started = time.monotonic() + await actions["cancel"]({"run_id": result["run_id"]}, context(runs)) + # Cancel latency is ~4 ms locally; the budget also covers the store writes + # and process reaping, which can spike on a loaded CI runner. + await asyncio.wait_for(service.wait_idle(result["run_id"]), 3) + assert time.monotonic() - started < 3 + final = await actions["get"]({"run_id": result["run_id"]}, context(runs)) + assert final["status"] == "cancelled", final + assert not service.processes + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "script", + [ + "for(;;) await null;", + "const p=agent({stepId:'a',agentRef:'lead_agent',task:'blocked'}); for(;;) await null;", + ], +) +async def test_spinning_script_still_hits_turn_limit(plugin, monkeypatch, script): + service = plugin[2] + monkeypatch.setattr(sys.modules[type(service).__module__], "TURN_LIMIT", 50) + result = await launch(plugin, script) + final = await done(plugin, result["run_id"]) + assert final["status"] == "failed", final + assert final["error"] == "Worker turn limit" + assert not service.processes + + @pytest.mark.asyncio async def test_external_supervisor_cancellation_is_interrupted(plugin): service, runs = plugin[2], plugin[3] diff --git a/examples/deerflow-extension-workflows/AGENTS.md b/examples/deerflow-extension-workflows/AGENTS.md index 49e152c536..1abde96292 100644 --- a/examples/deerflow-extension-workflows/AGENTS.md +++ b/examples/deerflow-extension-workflows/AGENTS.md @@ -8,6 +8,9 @@ Generated script runs only in worker.py's isolated, capability-free JS context. Never pass host callables, capability handles, credentials or ambient environment into it. Keep every IPC and stored/public payload bounded. Parent lifecycle must kill and reap the worker and explicitly observe child cancellation. +The worker turn cap counts worker frames only; while the worker is idle (its job +queue drained) and Agent calls are pending, wait locally with bounded cancellation, +fault and worker-exit polls instead of round-tripping empty frames, so wall time stays the only clock bound. Persist stable child thread, input message, attempt and idempotency identity before admission. Workflow cancel is a supervisor signal; do not cancel an admission task diff --git a/examples/deerflow-extension-workflows/README.md b/examples/deerflow-extension-workflows/README.md index 62febe429c..a8dd61e750 100644 --- a/examples/deerflow-extension-workflows/README.md +++ b/examples/deerflow-extension-workflows/README.md @@ -233,6 +233,13 @@ credential-free environment. The parent kills/reaps it on termination. This is a capability-restricted JS interpreter in a separate process, not a general OS sandbox or a guarantee against native engine vulnerabilities. +Wall time (`timeout_seconds`) is the only clock bound on a run. A separate cap +of 10,000 worker turns stops scripts that spin the JS job queue; idle waits on +pending Agent calls do not exchange frames with the worker and do not consume +turns, so long child runs are limited only by the wall deadline. +The local wait checks worker exits every 20 ms alongside cancellation and faults; +an exit triggers prompt failure handling and cleanup of pending child runs. + IPC frames are bounded to 48 KiB. Script JSON encoding is at most 16,000 bytes, arguments 4,000 bytes, task 8,000 bytes, each captured Agent output/final result 12,000 bytes. There are at most 20 progress messages of 160 JSON bytes each and diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py index 030caabe09..a44c3ccd05 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/service.py @@ -21,6 +21,7 @@ from .store import TERMINAL, Store NAMESPACE = "community.workflows" FRAME = 48 * 1024 +TURN_LIMIT = 10000 logger = logging.getLogger(__name__) @@ -539,7 +540,7 @@ class Workflows(Lifecycle, Artifacts): if entry["cancel"].is_set() or entry["fault"]: raise ValueError(entry["fault"] or "Cancelled") turns += 1 - if turns > 10000: + if turns > TURN_LIMIT: raise ValueError("Worker turn limit") raw = await process.stdout.readline() if not raw or len(raw) > FRAME: @@ -587,13 +588,19 @@ class Workflows(Lifecycle, Artifacts): encoded(outcome["value"], 12000) record.update(status="completed", result=outcome["value"]) break - # Poll the supervisor while a response is pending; do not block cancel on a child. + # Poll cancel/fault/worker exit while a response is pending. An idle worker can only progress + # on a response, so keep waiting here instead of sending empty frames (turns). if entry["tasks"]: - await asyncio.wait( - entry["tasks"], - timeout=0.02, - return_when=asyncio.FIRST_COMPLETED, - ) + while True: + finished, _ = await asyncio.wait( + entry["tasks"], + timeout=0.02, + return_when=asyncio.FIRST_COMPLETED, + ) + if process.returncode is not None: + raise ValueError("Worker exited while waiting for Agent calls") + if finished or not frame.get("idle", False) or entry["cancel"].is_set() or entry["fault"]: + break else: await asyncio.sleep(0.002) finished = [t for t in entry["tasks"] if t.done()] diff --git a/examples/deerflow-extension-workflows/deerflow_extension_workflows/worker.py b/examples/deerflow-extension-workflows/deerflow_extension_workflows/worker.py index 57880b3934..0fd05d95e8 100644 --- a/examples/deerflow-extension-workflows/deerflow_extension_workflows/worker.py +++ b/examples/deerflow-extension-workflows/deerflow_extension_workflows/worker.py @@ -89,12 +89,15 @@ def main(): launch(request["script"], json.dumps(request["args"])) while True: # Bound each JS turn and require a parent round trip between batches. + idle = False for _ in range(64): if not ctx.execute_pending_job(): + idle = True break messages = json.loads(take()) outcome = result() - write({"messages": messages, "outcome": json.loads(outcome) if outcome else None}) + # idle: the job queue drained, so only a response can make further progress. + write({"messages": messages, "outcome": json.loads(outcome) if outcome else None, "idle": idle}) if outcome: return response = read()