feat(run-loop): lift the interactive continuation loop — applies to all agents
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) <noreply@anthropic.com>
This commit is contained in:
0xallam
+9
-44
@@ -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.
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
@@ -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}",
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user