diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index f7c3dcb9..bb5924fb 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -31,19 +31,16 @@ INTER-AGENT MESSAGES: {% if interactive %} INTERACTIVE BEHAVIOR: -- You are in an interactive conversation with a user -- CRITICAL: A message WITHOUT a tool call IMMEDIATELY STOPS your entire execution and waits for user input. This is a HARD SYSTEM CONSTRAINT, not a suggestion. - - Statements like "Planning the assessment..." or "I'll now scan..." or "Starting with..." WITHOUT a tool call will HALT YOUR WORK COMPLETELY. The system interprets no-tool-call as "I'm done, waiting for the user." - - If you want to plan, call the think tool. If you want to act, call the appropriate tool. There is NO valid reason to output text without a tool call while working on a task. - - The ONLY time you may send a message without a tool call is when you are genuinely DONE and presenting final results, or when you NEED the user to answer a question before continuing. -- EVERY message while working MUST contain exactly one tool call — this is what keeps execution moving. No tool call = execution stops. -- You may include brief explanatory text BEFORE the tool call -- Respond naturally when the user asks questions or gives instructions -- For simple conversation, acknowledgements, or direct questions that you can answer from current context, reply in plain text and stop. Do NOT call think just to prepare wording. -- If you use a tool to answer a user question (for example list_todos, view_agent_graph, or a file read), then after the tool result arrives, provide the answer in plain text and stop unless the user explicitly asked you to continue working. -- Never loop through think or other tools just to prepare, polish, confirm, or announce a final answer. Once you know the answer, say it. -- NEVER send empty messages — if you have nothing to do or say, call the wait_for_message tool -- If you catch yourself about to describe multiple steps without a tool call, STOP and call the think tool instead +- You are in an interactive conversation with a user. +- HOW EXECUTION ENDS: your turn ends ONLY when you make an explicit lifecycle tool call. Plain text NEVER ends your turn and NEVER hands control to the user — text is shown to the user, and then execution continues. + - To hand control back to the user (you answered them, or you need their input before continuing), call the wait_for_message tool. This is the ONLY sanctioned way to yield to the user; it parks you until the user's next message arrives. + - To end the whole engagement, call the lifecycle tool: finish_scan (root) or agent_finish (subagent). + - A turn that ends with plain text and no tool call does NOT stop you: the system nudges you to continue and will re-run you. Do not rely on going silent to pause — it will not pause you. +- Answering a user question: put your answer in text, then call wait_for_message in the SAME turn (the text is delivered to the user and you park for their reply). Do not answer and then fall silent — that just triggers a continuation nudge. +- You may include brief explanatory text before a tool call. +- Respond naturally when the user asks questions or gives instructions. +- While actively working on a task, every turn should carry exactly one tool call — use think to plan, the appropriate tool to act, and wait_for_message only when you genuinely need the user. +- Never loop through think or other tools just to prepare, polish, confirm, or announce an answer. Once you know the answer, say it (then wait_for_message). {% else %} AUTONOMOUS BEHAVIOR: - Work autonomously by default diff --git a/strix/core/execution.py b/strix/core/execution.py index 762fcd3f..b16041bb 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -208,22 +208,8 @@ async def run_agent_loop( await coordinator.send(agent_id, _reserve_notice()) if not (start_parked and interactive): - if interactive: - with contextlib.suppress(BudgetPausedError): - result = await _run_cycle_parked( - agent, - coordinator, - agent_id, - input_data=first_cycle_input, - run_config=run_config, - context=context, - max_turns=max_turns, - session=session, - event_sink=event_sink, - hooks=hooks, - ) - else: - result = await _run_noninteractive_until_lifecycle( + with contextlib.suppress(BudgetPausedError): + result = await _run_until_lifecycle( agent, coordinator, agent_id, @@ -232,6 +218,7 @@ async def run_agent_loop( context=context, max_turns=max_turns, session=session, + interactive=interactive, event_sink=event_sink, hooks=hooks, ) @@ -268,15 +255,16 @@ async def run_agent_loop( await coordinator.consume_pending(agent_id) with contextlib.suppress(BudgetPausedError): - result = await _run_cycle_parked( + result = await _run_until_lifecycle( agent, coordinator, agent_id, - input_data=[], + initial_input=[], run_config=run_config, context=context, max_turns=max_turns, session=session, + interactive=True, event_sink=event_sink, hooks=hooks, ) @@ -427,7 +415,10 @@ async def respawn_subagents( await coordinator.set_status(child_id, "crashed") -async def _run_noninteractive_until_lifecycle( +_INTERACTIVE_TOOL_RECOVERY_LIMIT = 3 + + +async def _run_until_lifecycle( agent: Any, coordinator: AgentCoordinator, agent_id: str, @@ -437,14 +428,21 @@ async def _run_noninteractive_until_lifecycle( context: dict[str, Any], max_turns: int, session: Session | None, + interactive: bool, event_sink: StreamEventSink | None, hooks: RunHooks[dict[str, Any]] | None, ) -> RunResultBase | None: - """Non-chat mode keeps running until finish_scan / agent_finish settles status.""" + """Drive an agent until an explicit lifecycle tool settles its status. + + A turn that ends without ``finish_scan``, ``agent_finish``, or + ``wait_for_message`` leaves the agent ``running``: plain text never + terminates a run and never yields to the user. Such a turn is nudged back + into a tool call, bounded by a recovery limit. + """ result: RunResultBase | None = None input_data: Any = initial_input - invalid_final_outputs = 0 - invalid_final_output_limit = max(1, max_turns) + recoveries = 0 + recovery_limit = _INTERACTIVE_TOOL_RECOVERY_LIMIT if interactive else max(1, max_turns) while True: if coordinator.budget_stopped: @@ -455,50 +453,89 @@ async def _run_noninteractive_until_lifecycle( await coordinator.set_status(agent_id, "stopped") raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve") - result = await _run_cycle( - agent, - coordinator, - agent_id, - input_data=input_data, - run_config=run_config, - context=context, - max_turns=max_turns, - session=session, - interactive=False, - event_sink=event_sink, - hooks=hooks, - ) + if interactive: + result = await _run_cycle_parked( + agent, + coordinator, + agent_id, + input_data=input_data, + run_config=run_config, + context=context, + max_turns=max_turns, + session=session, + event_sink=event_sink, + hooks=hooks, + ) + else: + result = await _run_cycle( + agent, + coordinator, + agent_id, + input_data=input_data, + run_config=run_config, + context=context, + max_turns=max_turns, + session=session, + interactive=False, + event_sink=event_sink, + hooks=hooks, + ) status = await _agent_status(coordinator, agent_id) if status != "running": return result - invalid_final_outputs += 1 + recoveries += 1 logger.warning( - "agent %s produced non-lifecycle final output in non-interactive mode; " + "agent %s ended a turn without a lifecycle tool call (interactive=%s); " "forcing tool continuation (%d/%d): %s", agent_id, - invalid_final_outputs, - invalid_final_output_limit, + interactive, + recoveries, + recovery_limit, _final_output_preview(result), ) - if invalid_final_outputs >= invalid_final_output_limit: - await coordinator.set_status(agent_id, "crashed") - await _notify_parent_on_terminal(coordinator, agent_id, "crashed") - raise MaxTurnsExceeded( - "Agent exhausted non-interactive recovery attempts without calling " - "finish_scan or agent_finish." - ) + if recoveries >= recovery_limit: + return await _exhausted_recovery(coordinator, agent_id, result, interactive=interactive) - input_data = await _append_noninteractive_tool_required_message( + input_data = await _append_tool_required_message( session=session, context=context, - attempt=invalid_final_outputs, - limit=invalid_final_output_limit, + attempt=recoveries, + limit=recovery_limit, + interactive=interactive, ) +async def _exhausted_recovery( + coordinator: AgentCoordinator, + agent_id: str, + result: RunResultBase | None, + *, + interactive: bool, +) -> RunResultBase | None: + """Settle an agent that never recovered into a tool call. + + Interactive runs park instead of dying: a human is present, so the scan + stays resumable by sending another message. Autonomous runs have nobody to + resume them, so they fail loudly. + """ + if not interactive: + await coordinator.set_status(agent_id, "crashed") + await _notify_parent_on_terminal(coordinator, agent_id, "crashed") + raise MaxTurnsExceeded( + "Agent exhausted recovery attempts without calling finish_scan or agent_finish." + ) + + logger.warning( + "agent %s exhausted tool-call recovery attempts; parking until a message arrives", + agent_id, + ) + await coordinator.set_status(agent_id, "waiting") + return result + + _WAITING_AUTO_RESUME_TIMEOUT_S = 600.0 @@ -724,27 +761,9 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 await _notify_parent_on_terminal(coordinator, agent_id, status) return None else: - await _settle_run_result(coordinator, agent_id, interactive) return cast("RunResultBase | None", stream) -async def _settle_run_result( - coordinator: AgentCoordinator, - agent_id: str, - interactive: bool, -) -> None: - async with coordinator._lock: - current_status = coordinator.statuses.get(agent_id) - - if current_status != "running": - return - - if not interactive: - return - - await coordinator.set_status(agent_id, "waiting") - - async def _agent_status(coordinator: AgentCoordinator, agent_id: str) -> Status | None: async with coordinator._lock: return coordinator.statuses.get(agent_id) @@ -760,23 +779,36 @@ def _final_output_preview(result: RunResultBase | None) -> str: return text[:300] -async def _append_noninteractive_tool_required_message( +async def _append_tool_required_message( *, session: Session | None, context: dict[str, Any], attempt: int, limit: int, + interactive: bool, ) -> list[dict[str, str]]: finish_tool = "finish_scan" if context.get("parent_id") is None else "agent_finish" - message = ( - "Your previous response ended the autonomous Strix run without a lifecycle tool call. " - "That is invalid in non-interactive mode; plain text final answers are ignored. " - "Continue immediately and call exactly one tool. " - f"If your work is complete, call {finish_tool}. " - "If you are blocked waiting for another agent, call wait_for_message. " - "Otherwise use the appropriate execution or planning tool. " - f"This is recovery attempt {attempt}/{limit}." - ) + if interactive: + message = ( + "Your previous message ended a turn without a tool call. Plain text never ends " + "execution and never hands control to the user: it is shown to the user, and the " + "run continues. Continue immediately and call exactly one tool. " + "If you have finished responding and want the user's next message, call " + "wait_for_message. " + f"If the whole engagement is complete, call {finish_tool}. " + "Otherwise use the appropriate execution or planning tool. " + f"This is recovery attempt {attempt}/{limit}." + ) + else: + message = ( + "Your previous response ended the autonomous Strix run without a lifecycle tool " + "call. That is invalid in non-interactive mode; plain text final answers are " + "ignored. Continue immediately and call exactly one tool. " + f"If your work is complete, call {finish_tool}. " + "If you are blocked waiting for another agent, call wait_for_message. " + "Otherwise use the appropriate execution or planning tool. " + f"This is recovery attempt {attempt}/{limit}." + ) item = {"role": "user", "content": message} if session is None: return [item] diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index 7f937400..b5d1af9d 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -232,11 +232,19 @@ async def wait_for_message( # noqa: PLR0911 message arrives, so pick a ``timeout_seconds`` proportional to the work you're awaiting. + In an interactive/chat session this is also the ONLY sanctioned way + to hand control back to the user: plain text does not end your turn + or yield to the user, so after you answer the user (or when you need + their input before continuing) call this to park until their next + message arrives. + **Critical caveats:** - - **Never** call this if you finished your own task and have **no** - child agents running — that's a permanent stall. Call - ``finish_scan`` (root) or ``agent_finish`` (subagent) instead. + - In an autonomous (non-interactive) run, **never** call this if you + finished your own task and have **no** child agents running — that's a + permanent stall. Call ``finish_scan`` (root) or ``agent_finish`` + (subagent) instead. (In an interactive session there is always a user + who can message you, so parking to await the user is expected.) - If you're waiting on an agent that **isn't your child**, message it first asking it to ping you when done — otherwise it has no reason to send to your inbox and you'll wait the full timeout. diff --git a/tests/test_execution.py b/tests/test_execution.py index c1a13a72..7fc73782 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -9,6 +9,7 @@ from typing import Any, cast from unittest.mock import MagicMock import pytest +from agents.exceptions import MaxTurnsExceeded from agents.items import MessageOutputItem from agents.memory import SQLiteSession from agents.tool_context import ToolContext @@ -785,3 +786,148 @@ async def test_run_agent_loop_seeds_identity_before_first_cycle( stored = await session.get_items() assert any("recon" in str(cast("dict[str, Any]", i).get("content", "")) for i in stored) session.close() + + +def _scripted_cycle( + coordinator: AgentCoordinator, + agent_id: str, + statuses: list[str], + calls: list[Any], +) -> Any: + """Fake run cycle that leaves ``agent_id`` in a scripted status per call.""" + + async def _cycle(*_args: Any, **kwargs: Any) -> Any: + calls.append(kwargs.get("input_data")) + status = statuses[min(len(calls) - 1, len(statuses) - 1)] + await coordinator.set_status(agent_id, status) + return MagicMock(final_output="plain text, no tool call") + + return _cycle + + +async def _drive( + coordinator: AgentCoordinator, + agent_id: str, + *, + interactive: bool, + max_turns: int = 5, +) -> Any: + return await execution._run_until_lifecycle( + MagicMock(), + coordinator, + agent_id, + initial_input=[], + run_config=MagicMock(), + context={"agent_id": agent_id, "parent_id": None}, + max_turns=max_turns, + session=None, + interactive=interactive, + event_sink=None, + hooks=None, + ) + + +@pytest.mark.asyncio +async def test_interactive_text_only_turn_is_nudged_instead_of_parking( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A no-tool-call turn must not silently hand control back to the user.""" + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + calls: list[Any] = [] + monkeypatch.setattr( + execution, + "_run_cycle_parked", + _scripted_cycle(coordinator, "root", ["running", "completed"], calls), + ) + + await _drive(coordinator, "root", interactive=True) + + assert len(calls) == 2 + # The retry carries an explicit "call a tool" nudge rather than empty input. + nudge = calls[1][0]["content"] + assert "without a tool call" in nudge + assert "wait_for_message" in nudge + assert coordinator.statuses["root"] == "completed" + + +@pytest.mark.asyncio +async def test_interactive_wait_for_message_parks_without_a_nudge( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """``waiting`` is now only reachable by an explicit wait_for_message call.""" + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + calls: list[Any] = [] + monkeypatch.setattr( + execution, + "_run_cycle_parked", + _scripted_cycle(coordinator, "root", ["waiting"], calls), + ) + + await _drive(coordinator, "root", interactive=True) + + assert len(calls) == 1 + assert coordinator.statuses["root"] == "waiting" + + +@pytest.mark.asyncio +async def test_interactive_recovery_exhaustion_parks_instead_of_crashing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A human can resume an interactive scan, so exhaustion parks rather than dies.""" + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + calls: list[Any] = [] + monkeypatch.setattr( + execution, + "_run_cycle_parked", + _scripted_cycle(coordinator, "root", ["running"], calls), + ) + + await _drive(coordinator, "root", interactive=True) + + assert len(calls) == execution._INTERACTIVE_TOOL_RECOVERY_LIMIT + assert coordinator.statuses["root"] == "waiting" + + +@pytest.mark.asyncio +async def test_noninteractive_recovery_exhaustion_crashes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """No user is present to resume an autonomous run, so it still fails loudly.""" + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + calls: list[Any] = [] + monkeypatch.setattr( + execution, + "_run_cycle", + _scripted_cycle(coordinator, "root", ["running"], calls), + ) + + with pytest.raises(MaxTurnsExceeded): + await _drive(coordinator, "root", interactive=False, max_turns=2) + + assert len(calls) == 2 + assert coordinator.statuses["root"] == "crashed" + + +@pytest.mark.asyncio +async def test_tool_required_message_is_persisted_to_the_session(tmp_path: Any) -> None: + session = SQLiteSession("root", tmp_path / "agents.db") + + assert ( + await execution._append_tool_required_message( + session=session, + context={"parent_id": None}, + attempt=1, + limit=3, + interactive=True, + ) + == [] + ) + + stored = [cast("dict[str, Any]", i) for i in await session.get_items()] + assert "finish_scan" in stored[0]["content"] + assert "wait_for_message" in stored[0]["content"] + session.close()