From 1afd1766cb7e91901c7bc1e58ce364e85c51b34f Mon Sep 17 00:00:00 2001 From: 0xallam Date: Sat, 25 Apr 2026 17:02:44 -0700 Subject: [PATCH] =?UTF-8?q?feat(run-loop):=20lift=20the=20interactive=20co?= =?UTF-8?q?ntinuation=20loop=20=E2=80=94=20applies=20to=20all=20agents?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit only kept the root agent alive across cycles. But ``interactive`` propagates to children via ``make_child_factory``, and the legacy harness's continuation loop applied to every interactive agent in the tree — children also stayed alive after ``agent_finish``, ready to receive follow-up messages from the parent or siblings. Lift the demo-loop pattern out of ``entry.run_strix_scan`` into a shared helper :func:`strix.run_loop.run_with_continuation` and use it at both call sites: - ``entry.run_strix_scan`` for the root agent. - ``tools.agents_graph.tools.create_agent`` for child agents — the ``asyncio.create_task(Runner.run(...))`` becomes ``asyncio.create_task(run_with_continuation(...))``. ``StrixOrchestrationHooks.on_agent_end`` drops the ``parent_id is None`` constraint — any interactive agent parks instead of finalizing. Children that crash still finalize so parents stop waiting on them. Cancellation propagates correctly: ``bus.cancel_descendants`` cancels the task; ``run_with_continuation``'s ``await bus.wait_for_message`` catches ``CancelledError`` and returns the last result. Lint at baseline. Co-Authored-By: Claude Opus 4.7 (1M context) --- strix/entry.py | 53 +++-------------- strix/orchestration/hooks.py | 25 ++++---- strix/run_loop.py | 98 +++++++++++++++++++++++++++++++ strix/tools/agents_graph/tools.py | 12 ++-- 4 files changed, 127 insertions(+), 61 deletions(-) create mode 100644 strix/run_loop.py diff --git a/strix/entry.py b/strix/entry.py index 57bfbb8..f825ed7 100644 --- a/strix/entry.py +++ b/strix/entry.py @@ -15,13 +15,11 @@ from __future__ import annotations -import asyncio import logging import uuid from pathlib import Path from typing import TYPE_CHECKING, Any, Literal -from agents import Runner from agents.memory import SQLiteSession from strix.agents.factory import build_strix_agent, make_child_factory @@ -33,6 +31,7 @@ from strix.run_config_factory import ( make_agent_context, make_run_config, ) +from strix.run_loop import run_with_continuation from strix.runtime import session_manager @@ -284,52 +283,18 @@ async def run_strix_scan( session_db.parent.mkdir(parents=True, exist_ok=True) session = SQLiteSession(session_id=scan_id, db_path=session_db) - task_text = _build_root_task(scan_config) - hooks = StrixOrchestrationHooks() - - result = await Runner.run( - root_agent, - input=task_text, - session=session, + return await run_with_continuation( + agent=root_agent, + initial_input=_build_root_task(scan_config), run_config=run_config, context=context, - hooks=hooks, + hooks=StrixOrchestrationHooks(), max_turns=max_turns, + bus=bus, + agent_id=root_id, + interactive=interactive, + session=session, ) - - if not interactive: - return result - - # Interactive mode: SDK demo-loop pattern. The root agent is - # parked (not finalized) by ``StrixOrchestrationHooks.on_agent_end`` - # at the end of each cycle, so ``bus.send`` from the TUI still - # accepts user messages between cycles. Wake on the next message, - # drain it, and re-invoke ``Runner.run`` with the appended input — - # the SQLite session preserves the prior conversation, so the - # agent picks up with full context. - while True: - try: - await bus.wait_for_message(root_id) - except asyncio.CancelledError: - return result - pending = await bus.drain(root_id) - if not pending: - continue - next_input = "\n\n".join( - str(msg.get("content", "")).strip() for msg in pending if msg.get("content") - ) - if not next_input: - continue - - result = await Runner.run( - root_agent, - input=next_input, - session=session, - run_config=run_config, - context=context, - hooks=hooks, - max_turns=max_turns, - ) except BaseException: # Cancel any descendant tasks the root spawned before unwinding. # cancel_descendants is idempotent and handles the empty-tree case. diff --git a/strix/orchestration/hooks.py b/strix/orchestration/hooks.py index d83217b..fb08f81 100644 --- a/strix/orchestration/hooks.py +++ b/strix/orchestration/hooks.py @@ -147,17 +147,15 @@ class StrixOrchestrationHooks(RunHooks[Any]): return crashed = (output is None) or not ctx.get("agent_finish_called", False) - # Interactive root agents stay alive across ``Runner.run`` cycles — - # ``entry.run_strix_scan``'s outer loop re-invokes ``Runner.run`` - # whenever the user sends a follow-up message, so we just park - # the agent (status=waiting) instead of finalizing. - is_interactive_root = ( - ctx.get("parent_id") is None and bool(ctx.get("interactive", False)) and not crashed - ) + # Interactive agents (root and children) stay alive across + # ``Runner.run`` cycles — ``run_with_continuation`` re-invokes + # ``Runner.run`` whenever the agent receives a follow-up + # message, so we just park (status=waiting) instead of + # finalizing. Crashed runs always finalize so the parent + # learns to stop waiting. + stays_alive = bool(ctx.get("interactive", False)) and not crashed - final_status = ( - "waiting" if is_interactive_root else ("crashed" if crashed else "completed") - ) + final_status = "waiting" if stays_alive else ("crashed" if crashed else "completed") tracer = ctx.get("tracer") if tracer is not None and me in tracer.agents: @@ -178,10 +176,11 @@ class StrixOrchestrationHooks(RunHooks[Any]): }, ) - if is_interactive_root: + if stays_alive: await bus.park(me) - # Reset the per-cycle flag so the next ``Runner.run`` invocation - # can detect a fresh ``finish_scan`` call. + # Reset per-cycle flags so the next ``Runner.run`` invocation + # can detect a fresh finish-tool call and re-trigger budget + # warnings against its own iteration count. ctx["agent_finish_called"] = False ctx["turn_count"] = 0 else: diff --git a/strix/run_loop.py b/strix/run_loop.py new file mode 100644 index 0000000..811a7bd --- /dev/null +++ b/strix/run_loop.py @@ -0,0 +1,98 @@ +"""``run_with_continuation`` — interactive-mode demo-loop wrapper around ``Runner.run``. + +Pre-migration ``BaseAgent.agent_loop`` ran forever in interactive mode, +re-entering a "waiting state" after each finish-tool call so the agent +could pick up follow-up messages from its parent (or from the user, in +the root's case). Post-migration ``Runner.run`` returns on +``StopAtTools(...)`` and the agent is gone. + +This helper restores the legacy semantics using the SDK's canonical +demo-loop pattern (``agents/repl.py:run_demo_loop``): after each +``Runner.run`` cycle, ``await bus.wait_for_message(agent_id)``, drain +new messages, and re-invoke ``Runner.run`` with them as the next turn's +input. The session (if provided) preserves prior conversation across +cycles. + +Used by both the root scan loop in ``entry.run_strix_scan`` and the +child-agent loop in ``tools.agents_graph.tools.create_agent`` so every +interactive agent on the bus stays alive. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import TYPE_CHECKING, Any + +from agents import Runner + + +if TYPE_CHECKING: + from agents.lifecycle import RunHooks + from agents.memory import Session + from agents.result import RunResult + from agents.run_config import RunConfig + + from strix.orchestration.bus import AgentMessageBus + + +logger = logging.getLogger(__name__) + + +async def run_with_continuation( + *, + agent: Any, + initial_input: Any, + run_config: RunConfig, + context: dict[str, Any], + hooks: RunHooks[Any], + max_turns: int, + bus: AgentMessageBus, + agent_id: str, + interactive: bool, + session: Session | None = None, +) -> RunResult: + """Run an agent once (non-interactive) or in a continuation loop (interactive). + + For non-interactive runs this is a thin wrapper around + ``Runner.run`` and returns its result. + + For interactive runs the function loops: after each ``Runner.run`` + returns, it awaits ``bus.wait_for_message(agent_id)``, drains any + accumulated messages from the inbox, formats them as the next + turn's user input, and invokes ``Runner.run`` again. The loop ends + when the wait gets cancelled (e.g. parent ``cancel_descendants`` or + user-issued KeyboardInterrupt). + """ + kwargs: dict[str, Any] = { + "input": initial_input, + "run_config": run_config, + "context": context, + "hooks": hooks, + "max_turns": max_turns, + } + if session is not None: + kwargs["session"] = session + + result: RunResult = await Runner.run(agent, **kwargs) + + if not interactive: + return result + + while True: + try: + await bus.wait_for_message(agent_id) + except asyncio.CancelledError: + return result + + pending = await bus.drain(agent_id) + if not pending: + continue + next_input = "\n\n".join( + str(msg.get("content", "")).strip() for msg in pending if msg.get("content") + ) + if not next_input: + continue + + kwargs["input"] = next_input + result = await Runner.run(agent, **kwargs) diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index 2d400e1..b56cdeb 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -21,11 +21,12 @@ import logging import uuid from typing import TYPE_CHECKING, Any, Literal -from agents import RunContextWrapper, Runner, function_tool +from agents import RunContextWrapper, function_tool from agents.items import TResponseInputItem from strix.orchestration.hooks import StrixOrchestrationHooks from strix.run_config_factory import make_agent_context, make_run_config +from strix.run_loop import run_with_continuation if TYPE_CHECKING: @@ -464,13 +465,16 @@ async def create_agent( ) task_handle = asyncio.create_task( - Runner.run( - child_agent, - input=initial_input, + run_with_continuation( + agent=child_agent, + initial_input=initial_input, run_config=child_run_config, context=child_ctx, hooks=StrixOrchestrationHooks(), max_turns=int(inner.get("max_turns", 300)), + bus=bus, + agent_id=child_id, + interactive=bool(inner.get("interactive", False)), ), name=f"agent-{name}-{child_id}", )