fix(workflow): dedupe progress rows, close leaked coroutines, cap runaway loops (#191)

- render_progress() re-listed agents truncated by the per-phase max_agents
  cap as duplicate "unphased" rows; now filters by phase membership instead
  of a rendered-id set.
- parallel() left already-created agent() coroutines unclosed when its
  argument validation raised, leaking "never awaited" warnings.
- The Python port dropped the reference implementation's 1000-agent
  lifetime backstop, so a workflow script with an unbounded loop and no
  token_budget could spawn subagents forever; restored the cap.

Author: elkaix

CHANGELOG.md

@@ -15,6 +15,9 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- Fix Workflow progress rendering duplicating agents truncated by the per-phase
+  display cap, close leaked `agent()` coroutines when `parallel()` rejects its
+  arguments, and add a 1000-agent lifetime backstop against runaway workflow loops.
 - Fix stale update-success notices so restarting into an older Homebrew install
   shows `/update` again instead of a permanent "Restart to apply" banner.
 

src/pythinker_code/tools/workflow/display.py

@@ -84,7 +84,6 @@ def render_progress(snapshot: WorkflowSnapshot, max_agents: int = 6, max_logs: i
     lines = [
         f"◆ Workflow: {snapshot.name} ({snapshot.done_count}/{len(snapshot.agents)} done{state})"
     ]
-    rendered: set[int] = set()
     phase_order = list(snapshot.phases)
     if snapshot.current_phase and snapshot.current_phase not in phase_order:
         phase_order.append(snapshot.current_phase)
@@ -95,9 +94,10 @@ def render_progress(snapshot: WorkflowSnapshot, max_agents: int = 6, max_logs: i
         done = sum(1 for a in agents if a.status == "done")
         lines.append(f"  {phase} {done}/{len(agents)}")
         for agent in agents[-max_agents:]:
-            rendered.add(agent.id)
             lines.append(f"    #{agent.id} {_STATUS_ICON.get(agent.status, '?')} {agent.label}")
-    unphased = [a for a in snapshot.agents if a.id not in rendered]
+    # Membership in phase_order, not an id set of rendered rows: agents cut by
+    # the per-phase max_agents tail must stay truncated, not reappear here.
+    unphased = [a for a in snapshot.agents if a.phase not in phase_order]
     for agent in unphased[-max_agents:]:
         lines.append(f"    #{agent.id} {_STATUS_ICON.get(agent.status, '?')} {agent.label}")
     for message in snapshot.logs[-max_logs:]:

src/pythinker_code/tools/workflow/engine.py

@@ -191,6 +191,12 @@ class WorkflowRuntimeError(Exception):
     """Raised when a workflow script misuses a primitive at runtime."""
 
 
+# Lifetime cap on agent() calls per workflow run — a runaway-loop backstop
+# (e.g. `while True: await agent(...)` with no token_budget set), mirroring
+# the reference implementation. Set far above any real workflow.
+MAX_TOTAL_AGENTS = 1000
+
+
 class AgentOptions:
     __slots__ = ("label", "phase", "schema", "model", "agent_type")
 
@@ -406,6 +412,11 @@ async def agent(prompt: Any, options: Any = None) -> Any:
             # batch instead of the whole dispatched set.
             if token_budget is not None and budget.remaining() <= 0:
                 raise WorkflowRuntimeError("workflow token budget exhausted")
+            if state["agent_count"] >= MAX_TOTAL_AGENTS:
+                raise WorkflowRuntimeError(
+                    f"workflow exceeded the {MAX_TOTAL_AGENTS}-agent lifetime cap; "
+                    "this is a runaway-loop backstop"
+                )
             state["agent_count"] += 1
             # Captured into a local now, before the first `await` below: other
             # concurrently-dispatched agent() calls can advance
@@ -440,9 +451,14 @@ async def agent(prompt: Any, options: Any = None) -> Any:
 
     async def parallel(items: Sequence[Any]) -> list[Any]:
         if not isinstance(items, (list, tuple)):
+            # Close before raising: the argument may itself be (or contain)
+            # already-created agent() coroutines that would otherwise leak
+            # as "never awaited" warnings.
+            _close_coroutines(items)
             raise WorkflowRuntimeError("parallel() expects a list of awaitables")
         for item in items:
             if callable(item) and not inspect.isawaitable(item):
+                _close_coroutines(items)
                 raise WorkflowRuntimeError(
                     "parallel() expects awaitables, not functions: "
                     "use parallel([agent('...'), agent('...')])"

tests/tools/test_workflow_display.py

@@ -52,6 +52,26 @@ def test_render_progress_shows_recent_log_messages():
     assert "second checkpoint" in text
 
 
+def test_render_progress_truncated_phase_agents_are_not_duplicated():
+    # Regression guard: agents cut by the per-phase max_agents tail must stay
+    # truncated, not reappear at the bottom as "unphased" rows.
+    snap = WorkflowSnapshot(name="n", description="d")
+    for i in range(1, 9):
+        snap.start_agent(i, f"scan {i}", "Scan")
+        snap.end_agent(i)
+    text = render_progress(snap, max_agents=6)
+    assert "#1 " not in text
+    assert "#2 " not in text
+    assert text.count("#8 ") == 1
+
+
+def test_render_progress_still_shows_phaseless_agents():
+    snap = WorkflowSnapshot(name="n", description="d")
+    snap.start_agent(1, "loner", None)
+    text = render_progress(snap)
+    assert "loner" in text
+
+
 def test_render_progress_truncates_to_max_logs():
     snap = WorkflowSnapshot(name="n", description="d")
     for i in range(5):

tests/tools/test_workflow_engine.py

@@ -250,3 +250,36 @@ async def test_cancellation_marks_running_cancelled_and_reraises():
         await task
     assert set(started) == {"a", "b"}
     assert set(skipped) == {"a", "b"}  # in-flight agents reported as cancelled, none completed
+
+
+@pytest.mark.asyncio
+async def test_agent_lifetime_cap_stops_runaway_loop(monkeypatch):
+    from pythinker_code.tools.workflow import engine
+
+    monkeypatch.setattr(engine, "MAX_TOTAL_AGENTS", 3)
+    runner, calls = make_runner()
+    script = 'meta = {"name": "n", "description": "d"}\nwhile True:\n    await agent("go")\n'
+    with pytest.raises(WorkflowRuntimeError, match="lifetime cap"):
+        await run_workflow(script, agent_runner=runner, cwd=".")
+    assert len(calls) == 3
+
+
+@pytest.mark.asyncio
+async def test_parallel_validation_failure_closes_pending_coroutines():
+    # Regression guard: mixing a plain function into parallel() raises, but the
+    # already-created agent() coroutines in the same list must be closed, not
+    # leaked as "coroutine was never awaited" warnings at GC time.
+    import gc
+    import warnings
+
+    runner, calls = make_runner()
+    script = (
+        'meta = {"name": "n", "description": "d"}\nawait parallel([agent("a"), len])\nreturn None\n'
+    )
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        with pytest.raises(WorkflowRuntimeError, match="not functions"):
+            await run_workflow(script, agent_runner=runner, cwd=".")
+        gc.collect()
+    assert not [w for w in caught if "never awaited" in str(w.message)]
+    assert calls == []