tests/test_agent.py

"""Tests for dee/core/agent.py — the Turing OpenRouter orchestrator.

No real network calls: requests.post is monkeypatched. Real end-to-end
verification against the live OpenRouter API is done manually via curl
with a real OPENROUTER_API_KEY, not in the automated suite.
"""

import pytest

from dee.core import agent


class _FakeResponse:
    def __init__(self, status_code=200, payload=None, text=""):
        self.status_code = status_code
        self._payload = payload or {}
        self.text = text

    def json(self):
        return self._payload


def _ok_payload(content="Hello from the model.", cost=0.001, prompt_tokens=0):
    return {
        "choices": [{"message": {"content": content}, "finish_reason": "stop"}],
        "usage": {"total_cost": cost, "prompt_tokens": prompt_tokens},
    }


def _tool_call_payload(call_id="call_1", name="design_crispr_guides", arguments='{"sequence": "ACGT"}', cost=0.001):
    return {
        "choices": [{
            "message": {
                "content": None,
                "tool_calls": [{
                    "id": call_id,
                    "type": "function",
                    "function": {"name": name, "arguments": arguments},
                }],
            },
            "finish_reason": "tool_calls",
        }],
        "usage": {"total_cost": cost},
    }


def test_config_from_env_reads_defaults(monkeypatch):
    monkeypatch.setenv("OPENROUTER_API_KEY", "sk-test")
    monkeypatch.delenv("AGENT_MODEL", raising=False)
    monkeypatch.delenv("AGENT_MAX_STEPS", raising=False)
    monkeypatch.delenv("AGENT_MAX_COST_USD", raising=False)
    config = agent.OpenRouterConfig.from_env()
    assert config is not None
    assert config.api_key == "sk-test"
    assert config.model == "google/gemini-3.5-flash"
    assert config.max_steps == 8
    assert config.max_cost_usd == 0.5


def test_config_from_env_missing_key_returns_none(monkeypatch):
    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
    assert agent.OpenRouterConfig.from_env() is None


def test_config_from_env_respects_overrides(monkeypatch):
    monkeypatch.setenv("OPENROUTER_API_KEY", "sk-test")
    monkeypatch.setenv("AGENT_MODEL", "some/other-model")
    monkeypatch.setenv("AGENT_MAX_STEPS", "3")
    monkeypatch.setenv("AGENT_MAX_COST_USD", "1.25")
    config = agent.OpenRouterConfig.from_env()
    assert config.model == "some/other-model"
    assert config.max_steps == 3
    assert config.max_cost_usd == 1.25


def test_create_session_defaults():
    session = agent.create_session()
    assert session.phase == "design"
    assert session.seed == 42
    assert session.history == []
    assert agent.get_session(session.session_id) is session


def test_get_session_missing_returns_none():
    assert agent.get_session("does-not-exist") is None


def test_run_agent_step_success(monkeypatch):
    def fake_post(url, headers=None, json=None, timeout=None):
        assert url == "https://openrouter.ai/api/v1/chat/completions"
        assert headers["Authorization"] == "Bearer sk-test"
        assert json["model"] == "google/gemini-3.5-flash"
        # System prompt + the one user message, no prior history.
        assert json["messages"][0]["role"] == "system"
        assert json["messages"][-1] == {"role": "user", "content": "hi"}
        return _FakeResponse(200, _ok_payload("hello back"))

    monkeypatch.setattr(agent.requests, "post", fake_post)

    config = agent.OpenRouterConfig(
        api_key="sk-test", model="google/gemini-3.5-flash",
        max_steps=8, max_cost_usd=0.5,
    )
    session = agent.create_session()
    result = agent.run_agent_step(config, session, "hi", False)

    assert result["ok"] is True
    assert result["assistant_message"] == "hello back"
    assert result["phase"] == "design"
    # History persisted on the session for the next turn.
    assert session.history == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello back"},
    ]


def test_run_agent_step_reports_context_window_usage(monkeypatch):
    monkeypatch.setattr(agent.requests, "post",
        lambda *a, **k: _FakeResponse(200, _ok_payload("hello back", prompt_tokens=12345)))

    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()
    result = agent.run_agent_step(config, session, "hi", False)

    assert result["context_tokens_used"] == 12345
    assert result["context_tokens_limit"] == agent.CONTEXT_WINDOW_TOKENS


def test_run_agent_step_second_turn_includes_history(monkeypatch):
    seen = {}

    def fake_post(url, headers=None, json=None, timeout=None):
        seen["messages"] = json["messages"]
        return _FakeResponse(200, _ok_payload("second reply"))

    monkeypatch.setattr(agent.requests, "post", fake_post)

    config = agent.OpenRouterConfig(
        api_key="sk-test", model="m", max_steps=8, max_cost_usd=0.5,
    )
    session = agent.create_session()
    session.history = [
        {"role": "user", "content": "first"},
        {"role": "assistant", "content": "first reply"},
    ]
    agent.run_agent_step(config, session, "second", False)

    roles = [m["role"] for m in seen["messages"]]
    assert roles == ["system", "user", "assistant", "user"]


def test_run_agent_step_http_error_raises_agent_error(monkeypatch):
    def fake_post(url, headers=None, json=None, timeout=None):
        return _FakeResponse(500, text="upstream exploded")

    monkeypatch.setattr(agent.requests, "post", fake_post)
    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()

    with pytest.raises(agent.AgentError) as excinfo:
        agent.run_agent_step(config, session, "hi", False)
    assert excinfo.value.kind == "agent_error"


def test_run_agent_step_insufficient_credits_raises_friendly_unavailable_error(monkeypatch):
    # OpenRouter's exact error shape for an out-of-credits account/key —
    # https://openrouter.ai/docs, confirmed 2026-07-10.
    def fake_post(url, headers=None, json=None, timeout=None):
        return _FakeResponse(402, payload={"error": {
            "code": 402,
            "message": "Your account or API key has insufficient credits. Add more credits and retry the request.",
        }}, text='{"error": {"code": 402, "message": "insufficient credits"}}')

    monkeypatch.setattr(agent.requests, "post", fake_post)
    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()

    with pytest.raises(agent.AgentError) as excinfo:
        agent.run_agent_step(config, session, "hi", False)

    assert excinfo.value.kind == "agent_unavailable"
    # Never surface billing language to the end user.
    message = str(excinfo.value)
    assert "credit" not in message.lower()
    assert "insufficient" not in message.lower()
    assert "development" in message.lower()


def test_run_agent_step_empty_choices_raises(monkeypatch):
    def fake_post(url, headers=None, json=None, timeout=None):
        return _FakeResponse(200, {"choices": []})

    monkeypatch.setattr(agent.requests, "post", fake_post)
    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()

    with pytest.raises(agent.AgentError):
        agent.run_agent_step(config, session, "hi", False)


def test_run_agent_step_cost_cap_exceeded_raises(monkeypatch):
    def fake_post(url, headers=None, json=None, timeout=None):
        return _FakeResponse(200, _ok_payload("expensive", cost=5.0))

    monkeypatch.setattr(agent.requests, "post", fake_post)
    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()

    with pytest.raises(agent.AgentError):
        agent.run_agent_step(config, session, "hi", False)
    # Nothing should be persisted for a call that got rejected on cost.
    assert session.history == []


def test_run_agent_step_network_error_raises_agent_error(monkeypatch):
    import requests as requests_lib

    def fake_post(url, headers=None, json=None, timeout=None):
        raise requests_lib.ConnectionError("dns broke")

    monkeypatch.setattr(agent.requests, "post", fake_post)
    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()

    with pytest.raises(agent.AgentError):
        agent.run_agent_step(config, session, "hi", False)


def test_run_agent_step_never_leaks_secret_in_error_message(monkeypatch):
    """Regression test for a real incident (2026-07-08): a malformed API key
    made the underlying HTTP client raise an exception whose message embeds
    the raw header value VERBATIM (a stdlib/urllib3 behavior, not a bug in
    this codebase) — that string reached the client through this route's
    error response, live, in production. Reproduces that exact failure shape
    and asserts the secret never appears anywhere in the raised message."""
    secret = "sk-or-v1-should-never-appear-in-any-error-message"

    def fake_post(url, headers=None, json=None, timeout=None):
        # Exactly the real failure: an exception whose str() embeds the
        # Authorization header value, malformed key and all.
        raise ValueError(
            f"Invalid leading whitespace, reserved character(s), or return "
            f"character(s) in header value: 'Bearer {secret}\\n\\n'"
        )

    monkeypatch.setattr(agent.requests, "post", fake_post)
    config = agent.OpenRouterConfig(api_key=secret, model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()

    with pytest.raises(agent.AgentError) as excinfo:
        agent.run_agent_step(config, session, "hi", False)

    assert secret not in str(excinfo.value)


def test_run_agent_step_malformed_json_never_leaks_response_text(monkeypatch):
    def fake_post(url, headers=None, json=None, timeout=None):
        class _BadJson(_FakeResponse):
            def json(self):
                raise ValueError("not json: token=super-secret-should-not-leak")

        return _BadJson(200, text="token=super-secret-should-not-leak")

    monkeypatch.setattr(agent.requests, "post", fake_post)
    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()

    with pytest.raises(agent.AgentError) as excinfo:
        agent.run_agent_step(config, session, "hi", False)

    assert "super-secret" not in str(excinfo.value)


def test_run_agent_step_executes_tool_call_then_returns_final_reply(monkeypatch):
    calls = [
        _tool_call_payload(arguments='{"sequence": "ACGTACGT"}'),
        _ok_payload("Here are the top guides."),
    ]

    def fake_post(url, headers=None, json=None, timeout=None):
        return _FakeResponse(200, calls.pop(0))

    monkeypatch.setattr(agent.requests, "post", fake_post)

    seen_calls = []

    def fake_execute_tool(name, args, auth_anonymous):
        seen_calls.append((name, args, auth_anonymous))
        return {"ok": True, "n_guides": 1, "guides": [{"spacer": "ACGTACGTACGTACGTACGT"}]}

    monkeypatch.setattr(agent._tools, "execute_tool", fake_execute_tool)

    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()
    result = agent.run_agent_step(config, session, "design me some guides", False)

    assert seen_calls == [("design_crispr_guides", {"sequence": "ACGTACGT"}, False)]
    assert result["assistant_message"] == "Here are the top guides."
    assert result["tool_result"] == {"ok": True, "n_guides": 1, "guides": [{"spacer": "ACGTACGTACGTACGTACGT"}]}
    assert result["tool_name"] == "design_crispr_guides"
    # Full turn — user ask, the assistant's tool-call, the tool's own
    # result, and the final reply — is what the NEXT turn replays as
    # context, so all four must be persisted in order.
    roles = [m["role"] for m in session.history]
    assert roles == ["user", "assistant", "tool", "assistant"]
    assert session.history[2]["tool_call_id"] == "call_1"


def test_run_agent_step_passes_auth_anonymous_true_to_tool(monkeypatch):
    calls = [_tool_call_payload(), _ok_payload("done")]
    monkeypatch.setattr(agent.requests, "post", lambda *a, **k: _FakeResponse(200, calls.pop(0)))

    seen = {}
    monkeypatch.setattr(agent._tools, "execute_tool",
                         lambda name, args, auth_anonymous: seen.setdefault("anon", auth_anonymous) or {"ok": False})

    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()
    agent.run_agent_step(config, session, "hi", True)

    assert seen["anon"] is True


def test_run_agent_step_cost_cap_checked_across_multiple_calls(monkeypatch):
    # Neither call alone exceeds the 0.5 cap, but the tool-call round trip's
    # SECOND OpenRouter call pushes the running total over it — must still
    # be caught, not just checked once against a single call's cost.
    calls = [
        _tool_call_payload(cost=0.3),
        _ok_payload("final", cost=0.3),
    ]
    monkeypatch.setattr(agent.requests, "post", lambda *a, **k: _FakeResponse(200, calls.pop(0)))
    monkeypatch.setattr(agent._tools, "execute_tool", lambda *a, **k: {"ok": True})

    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=8, max_cost_usd=0.5)
    session = agent.create_session()

    with pytest.raises(agent.AgentError):
        agent.run_agent_step(config, session, "hi", False)


def test_run_agent_step_exhausts_max_steps_returns_fallback_message(monkeypatch):
    # The model keeps requesting tool calls and never gives a final
    # content-only reply — must degrade gracefully within max_steps calls,
    # not hang or raise.
    monkeypatch.setattr(agent.requests, "post", lambda *a, **k: _FakeResponse(200, _tool_call_payload(cost=0.0)))
    monkeypatch.setattr(agent._tools, "execute_tool", lambda *a, **k: {"ok": True})

    config = agent.OpenRouterConfig(api_key="k", model="m", max_steps=3, max_cost_usd=0.5)
    session = agent.create_session()
    result = agent.run_agent_step(config, session, "hi", False)

    assert result["ok"] is True
    assert "ran out of steps" in result["assistant_message"]