diff --git a/strix/core/execution.py b/strix/core/execution.py index 87a2bf02..23784e61 100644 --- a/strix/core/execution.py +++ b/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, diff --git a/tests/test_execution.py b/tests/test_execution.py index 0daf0da2..81f0a147 100644 --- a/tests/test_execution.py +++ b/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")