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 <willem.jiang@gmail.com>

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

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]

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

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

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()]

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()