From f14954328d30f3f22da7462a0bad27873617c22a Mon Sep 17 00:00:00 2001 From: elkaix Date: Wed, 1 Jul 2026 21:13:50 -0400 Subject: [PATCH] fix(workflow): dedupe progress rows, close leaked coroutines, cap runaway loops - 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. --- CHANGELOG.md | 3 ++ src/pythinker_code/tools/workflow/display.py | 6 ++-- src/pythinker_code/tools/workflow/engine.py | 16 ++++++++++ tests/tools/test_workflow_display.py | 20 ++++++++++++ tests/tools/test_workflow_engine.py | 33 ++++++++++++++++++++ 5 files changed, 75 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a956a1e7..c5dc10e8 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/pythinker_code/tools/workflow/display.py b/src/pythinker_code/tools/workflow/display.py index abd1e8e6..f34c3804 100644 --- a/src/pythinker_code/tools/workflow/display.py +++ b/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:]: diff --git a/src/pythinker_code/tools/workflow/engine.py b/src/pythinker_code/tools/workflow/engine.py index 461ffa35..289cde58 100644 --- a/src/pythinker_code/tools/workflow/engine.py +++ b/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('...')])" diff --git a/tests/tools/test_workflow_display.py b/tests/tools/test_workflow_display.py index fedb2cd0..9378cc55 100644 --- a/tests/tools/test_workflow_display.py +++ b/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): diff --git a/tests/tools/test_workflow_engine.py b/tests/tools/test_workflow_engine.py index f9c240eb..96515d97 100644 --- a/tests/tools/test_workflow_engine.py +++ b/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 == []