Merge c5b6fbf6b0 into e21f9ce4cc

strix/core/execution.py

@@ -103,6 +103,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
 
@@ -627,6 +683,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

@@ -1167,6 +1167,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")