fix(execution): fail autonomous agents fast on plain-text refusals (#1155)

When a model refuses in plain text instead of returning a structured
`refusal` item, the turn ends without a lifecycle tool call. Autonomous
runs nudge that up to `max_turns` times and then crash as if the agent
had stalled, spending the whole budget on a reply that will not change.

Detect refusal text with action-anchored phrases (ambiguous markers such
as "as an AI" only count next to an inability word, so ordinary turns
like "I cannot reproduce the issue" are not caught) and mark the agent
failed with the refusal preview, notifying its parent.

Interactive runs are left alone: plain text is how the agent talks to
the user there, and declining a request is a legitimate reply that the
existing nudge and park handling already cover.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

strix/core/execution.py

@@ -86,6 +86,62 @@ def _structured_provider_refusal(result: Any) -> str | None:
     return None
 
 
+# Plain-text refusals: the model answers with refusal text (no structured
+# ``refusal`` content) instead of calling a lifecycle tool. An autonomous run
+# would nudge that up to ``max_turns`` times and then crash as if the agent had
+# merely stalled (#1155). Single words like "cannot" or "as an AI" also appear
+# in ordinary turns ("I cannot reproduce the issue", quoted page content), so
+# only action-anchored phrases count on their own.
+_STRONG_REFUSAL_PHRASES: frozenset[str] = frozenset(
+    {
+        "i cannot help",
+        "i cannot assist",
+        "i can't help",
+        "i can't assist",
+        "cannot help with",
+        "cannot assist",
+        "unable to help",
+        "unable to assist",
+        "i will not help",
+        "i will not assist",
+        "refuse to help",
+        "refuse to assist",
+        "i must decline",
+        "i have to decline",
+        "can't help with",
+        "won't help with",
+    }
+)
+
+
+def _extract_final_output_text(result: Any) -> str | None:
+    """Return the final output as stripped text if it is a non-empty string."""
+    final = getattr(result, "final_output", None)
+    if isinstance(final, str) and final.strip():
+        return final.strip()
+    return None
+
+
+def _is_text_refusal(text: str) -> bool:
+    """Heuristic: does plain-text output look like a model safety refusal?
+
+    Uses action-anchored strong phrases plus guarded co-occurrence checks
+    for ambiguous markers to avoid false positives on ordinary turns
+    (e.g. quoted page content or 'I cannot reproduce the issue').
+    """
+    lower = text.lower()
+    if any(phrase in lower for phrase in _STRONG_REFUSAL_PHRASES):
+        return True
+    # Ambiguous markers only count when paired with an inability signal.
+    if "i apologize" in lower and any(k in lower for k in ("cannot", "unable", "won't", "can't")):
+        return True
+    if "as an ai" in lower and any(k in lower for k in ("cannot", "unable", "not able", "won't")):
+        return True
+    return ("safety guidelines" in lower or "my safety" in lower) and any(
+        k in lower for k in ("cannot", "unable", "refuse", "not able", "won't")
+    )
+
+
 def _run_config_model(run_config: RunConfig) -> str | None:
     return run_config.model if isinstance(run_config.model, str) else None
 
@@ -584,6 +640,23 @@ async def _run_until_lifecycle(
             await coordinator.reset_recovery(agent_id)
             return result
 
+        # A refusal won't turn into a tool call however often it is nudged, so an
+        # autonomous run fails it now instead of spending the recovery budget.
+        # Interactive runs keep the nudge: plain text is how the agent talks to
+        # the user there, and declining a request is a legitimate reply.
+        refusal_text = None if interactive else _extract_final_output_text(result)
+        if refusal_text is not None and _is_text_refusal(refusal_text):
+            preview = _final_output_preview(result)
+            logger.error(
+                "agent %s refused task (plain-text refusal); failing fast instead of "
+                "retrying without tool call: %s",
+                agent_id,
+                preview,
+            )
+            await coordinator.set_status(agent_id, "failed", error=f"model refusal: {preview}")
+            await notify_parent_on_terminal(coordinator, agent_id, "failed")
+            return result
+
         recoveries = await coordinator.record_recovery(agent_id)
         _log_recovery(
             agent_id,

tests/test_execution.py

@@ -1144,6 +1144,103 @@ async def test_noninteractive_recovery_exhaustion_crashes(
     assert coordinator.statuses["root"] == "crashed"
 
 
+def _text_cycle(
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    text: str,
+    calls: list[Any],
+) -> Any:
+    """Fake run cycle whose turn ends in ``text`` with no lifecycle tool call."""
+
+    async def _cycle(*_args: Any, **kwargs: Any) -> Any:
+        calls.append(kwargs.get("input_data"))
+        await coordinator.set_status(agent_id, "running")
+        return MagicMock(final_output=text)
+
+    return _cycle
+
+
+@pytest.mark.asyncio
+async def test_noninteractive_text_refusal_fails_without_retrying(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A refusal never becomes a tool call, so nudging it only burns the budget."""
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("child", "recon", parent_id="root")
+    calls: list[Any] = []
+    monkeypatch.setattr(
+        execution,
+        "_run_cycle",
+        _text_cycle(coordinator, "child", "I can't help with testing this target.", calls),
+    )
+
+    await _drive(coordinator, "child", interactive=False, max_turns=50)
+
+    assert len(calls) == 1
+    assert coordinator.statuses["child"] == "failed"
+    pending, _ = await coordinator.consume_pending("root")
+    assert pending == 1
+
+
+@pytest.mark.asyncio
+async def test_noninteractive_ordinary_text_is_still_nudged(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Text that merely contains "cannot" keeps the normal recovery path."""
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    calls: list[Any] = []
+    monkeypatch.setattr(
+        execution,
+        "_run_cycle",
+        _text_cycle(coordinator, "root", "I cannot reproduce the issue yet.", 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_interactive_text_refusal_is_nudged_not_failed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Interactive agents talk to the user in plain text, declining included."""
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    calls: list[Any] = []
+    monkeypatch.setattr(
+        execution,
+        "_run_cycle_parked",
+        _text_cycle(coordinator, "root", "I can't help with scanning that host.", calls),
+    )
+
+    await _drive(coordinator, "root", interactive=True)
+
+    assert len(calls) == execution._INTERACTIVE_TOOL_RECOVERY_LIMIT
+    assert coordinator.statuses["root"] == "waiting"
+
+
+@pytest.mark.parametrize(
+    ("text", "expected"),
+    [
+        ("I can't help with that request.", True),
+        ("I must decline to attack this system.", True),
+        ("I apologize, but I'm unable to continue with this.", True),
+        ("As an AI, I cannot perform intrusive testing.", True),
+        ("I cannot reproduce the issue on the staging host.", False),
+        ("The page says: 'As an AI language model...'", False),
+        ("I apologize for the noise; the scan finished.", False),
+        ("Found 3 open ports on 10.0.0.5.", False),
+    ],
+)
+def test_is_text_refusal(text: str, expected: bool) -> None:
+    assert execution._is_text_refusal(text) is expected
+
+
 @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")