tests/test_graph.py

"""tests/test_graph.py — routing tests for the LangGraph agent.

The verify_node-related helpers (_normalize, _get_all_tool_outputs) referenced
by an earlier version of this file were removed when verify_node was retired
in favour of the post-synthesis reliability pass (see agent/post_synthesis.py).
"""
import json

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

from agent.evidence import (
    NO_VERIFIED_SYNTHESIS_MESSAGE,
    evidence_envelope,
    make_evidence_record,
    parse_evidence_envelope,
)
from agent.graph import (
    should_continue,
    nudge_node,
    AgentState,
    _extract_json,
    _cap_signals,
    _format_signals_message,
    MAX_EDGE_SIGNALS,
    MAX_FILING_SIGNALS,
    MAX_TRANSCRIPT_SIGNALS,
    _coverage_report,
    _finalize_synthesis_profile,
    _partial_brief,
    _pop_company_profile,
    profile_evidence_node,
    continuity_node,
    create_graph,
)
from agent.llm import RunConfig


def _state(messages, tool_round_count, nudge_fired: bool = False) -> AgentState:
    return {
        "ticker": "AAPL",
        "messages": messages,
        "tool_round_count": tool_round_count,
        "nudge_fired": nudge_fired,
        "edge_signals": None,
        "profile_payloads": None,
        "open_threads": None,
        "language": "English",
        "brief": None,
        "brief_markdown": None,
        "synthesis_error": None,
        "coverage": None,
        "verification_report": None,
    }


def _ai_with_tools():
    return AIMessage(
        content="",
        tool_calls=[{"id": "c1", "name": "get_financial_metrics", "args": {"ticker": "AAPL"}}],
    )


def _ai_done():
    return AIMessage(content="I have all the information I need.", tool_calls=[])


_SOURCE_BY_TOOL = {
    "get_financial_metrics": "metrics",
    "search_filing": "10-Q",
    "search_transcript": "transcript",
    "search_news": "news",
    "get_analyst_expectations": "analyst",
}


def _tool_msg(name: str, content: str | None = None, suffix: str = "1") -> ToolMessage:
    text = content or f"Verified evidence returned by {name}."
    record = make_evidence_record(
        source=_SOURCE_BY_TOOL[name],
        content=text,
        document_id=f"test:{name}:{suffix}",
        chunk_id="0",
        as_of="2026-04-30",
    )
    return ToolMessage(
        content=evidence_envelope(
            tool=name,
            records=[record],
            query={"ticker": "AAPL"},
        ),
        name=name,
        tool_call_id=f"id_{name}_{suffix}",
    )


def _empty_tool_msg(name: str) -> ToolMessage:
    return ToolMessage(
        content=evidence_envelope(
            tool=name,
            status="EMPTY",
            message="No evidence found.",
            query={"ticker": "AAPL"},
        ),
        name=name,
        tool_call_id=f"id_{name}_empty",
    )


def _error_tool_msg(name: str) -> ToolMessage:
    return ToolMessage(
        content=evidence_envelope(
            tool=name,
            status="ERROR",
            message="Retrieval failed.",
            error_code="TEST_ERROR",
            query={"ticker": "AAPL"},
        ),
        name=name,
        tool_call_id=f"id_{name}_error",
    )


def test_profile_evidence_node_injects_parseable_envelopes(monkeypatch):
    payloads = [
        evidence_envelope(
            tool="search_filing",
            records=[make_evidence_record(
                source="10-K",
                content=f"Profile evidence {index}.",
                document_id=f"sec:AAPL:profile:{index}",
                chunk_id=str(index),
            )],
        )
        for index in range(2)
    ]
    monkeypatch.setattr(
        "agent.company_profile.collect_profile_evidence",
        lambda ticker, include_metrics=True: payloads,
    )

    result = profile_evidence_node(_state([], 0))

    assert result["profile_payloads"] == payloads
    assert len(result["messages"]) == 3
    assert all(isinstance(message, HumanMessage) for message in result["messages"])
    assert result["messages"][0].content == (
        "== COMPANY PROFILE EVIDENCE "
        "(deterministic retrieval, evidence.v1 envelopes follow) =="
    )
    assert parse_evidence_envelope(result["messages"][0]) is None
    assert all(
        parse_evidence_envelope(message) is not None
        for message in result["messages"][1:]
    )
    assert [message.content for message in result["messages"][1:]] == payloads

    monkeypatch.setattr(
        "agent.company_profile.collect_profile_evidence",
        lambda ticker, include_metrics=True: [],
    )
    assert profile_evidence_node(_state([], 0)) == {
        "profile_payloads": [],
        "messages": [],
    }


def test_continuity_node_no_open_threads_returns_empty(monkeypatch):
    monkeypatch.setattr("storage.threads_db.get_open_threads", lambda ticker: [])
    result = continuity_node(_state([], 0))
    assert result == {"open_threads": [], "messages": []}


def test_continuity_node_storage_failure_falls_back_to_no_open_threads(monkeypatch):
    """A threads_db failure (e.g. 'database is locked' on the shared HF Space)
    must not propagate and crash the run -- it degrades to the same shape as
    the no-open-threads case."""
    def _boom(ticker):
        raise RuntimeError("database is locked")
    monkeypatch.setattr("storage.threads_db.get_open_threads", _boom)
    result = continuity_node(_state([], 0))
    assert result == {"open_threads": [], "messages": []}


def test_continuity_node_formats_open_threads_as_message(monkeypatch):
    open_threads = [{
        "thread_id": "abc123", "topic": "Cloud margin compression", "category": "Operational",
        "status": "ongoing", "last_updated_period": "2026-01-30",
        "history": [{"period": "2026-01-30", "status": "new", "note": "First flagged.", "evidence_ref": None, "evidence_snippet": None}],
    }]
    monkeypatch.setattr("storage.threads_db.get_open_threads", lambda ticker: open_threads)

    result = continuity_node(_state([], 0))

    assert result["open_threads"] == open_threads
    assert len(result["messages"]) == 1
    assert isinstance(result["messages"][0], HumanMessage)
    assert "== OPEN THREADS FROM PRIOR QUARTERS ==" in result["messages"][0].content
    assert "abc123" in result["messages"][0].content
    assert "Cloud margin compression" in result["messages"][0].content
    assert "First flagged." in result["messages"][0].content


def test_partial_brief_has_company_profile_none():
    partial = _partial_brief(_state([], 0), "No usable evidence.")
    assert partial["company_profile"] is None


def test_synthesis_zero_verified_keeps_filtered_brief_not_partial_skeleton(monkeypatch):
    fact = {
        "text": "Revenue increased 12%.",
        "source": "10-Q",
        "reliability": "HIGH",
        "evidence_snippet": "Revenue increased 12%.",
    }
    brief_json = json.dumps({
        "ticker": "AAPL",
        "company_name": "Apple Inc.",
        "filing_date": "2026-04-30",
        "what_matters_most": "Unsupported synthesis commentary.",
        "standout_number": fact,
        "what_changed": [],
        "bull_points": [fact],
        "bear_points": [],
        "what_to_watch": ["Watch Q3 gross margin"],
        "trends": [],
        "mda_summary": {
            "drivers": [],
            "headwinds": [],
            "language_shift": "No prior-period comparison was available.",
            "key_quote": fact,
        },
        "risks_categorized": [],
        "management_commentary": [],
        "guidance_history": [],
        "sentiment": None,
        "market_expectations": None,
    })

    class FakeLLM:
        def bind_tools(self, tools):
            return self

        def invoke(self, messages):
            return AIMessage(content="done", tool_calls=[])

        def stream(self, messages):
            yield AIMessage(content=brief_json)

    monkeypatch.setattr("analysis.textdiff.compute", lambda ticker: [])
    monkeypatch.setattr("analysis.tone_drift.compute", lambda ticker: [])
    monkeypatch.setattr(
        "agent.company_profile.collect_profile_evidence",
        lambda ticker, include_metrics=True: [],
    )
    monkeypatch.setattr("agent.graph.make_chat_model", lambda *args, **kwargs: FakeLLM())

    cfg = RunConfig(
        provider="anthropic",
        model="claude-haiku-4-5-20251001",
        api_key="sk-ant-test",
    )
    initial_state = _state([
        HumanMessage(content="Generate a research brief for AAPL."),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _tool_msg("search_transcript"),
    ], 0)

    final = create_graph(cfg).invoke(initial_state)
    brief = final["brief"]

    assert brief["status"] == "PARTIAL"
    assert brief["what_matters_most"] == NO_VERIFIED_SYNTHESIS_MESSAGE
    assert brief["bull_points"] == []
    assert brief["what_to_watch"] == ["Watch Q3 gross margin"]
    assert brief["filing_date"] == "2026-04-30"
    assert brief["model"] == cfg.model
    assert brief["generated_at"]
    assert brief["evidence_coverage"]["verified"] == 0
    assert brief["coverage"]


def test_synthesis_survives_continuity_thread_persistence_failure(monkeypatch):
    """A threads_db failure inside apply_continuity_threads (e.g. 'database is
    locked' on the shared HF Space) must not propagate to synthesis_node's
    blanket except-Exception and discard an otherwise fully-synthesized brief."""
    fact = {
        "text": "Revenue increased due to strong demand.",
        "source": "10-Q",
        "reliability": "HIGH",
        "evidence_snippet": "Revenue increased due to strong demand.",
    }
    brief_json = json.dumps({
        "ticker": "AAPL",
        "company_name": "Apple Inc.",
        "filing_date": "2026-04-30",
        "what_matters_most": "x",
        "standout_number": fact,
        "what_changed": [],
        "bull_points": [],
        "bear_points": [],
        "what_to_watch": ["Watch Q3 gross margin"],
        "trends": [],
        "continuity_threads": [],
        "mda_summary": {
            "drivers": [], "headwinds": [],
            "language_shift": "x", "key_quote": fact,
        },
        "risks_categorized": [], "management_commentary": [], "guidance_history": [],
        "sentiment": None, "market_expectations": None,
    })

    class FakeLLM:
        def bind_tools(self, tools):
            return self

        def invoke(self, messages):
            return AIMessage(content="done", tool_calls=[])

        def stream(self, messages):
            yield AIMessage(content=brief_json)

    def _boom(ticker, period, continuity_threads):
        raise RuntimeError("database is locked")

    monkeypatch.setattr("analysis.textdiff.compute", lambda ticker: [])
    monkeypatch.setattr("analysis.tone_drift.compute", lambda ticker: [])
    monkeypatch.setattr(
        "agent.company_profile.collect_profile_evidence",
        lambda ticker, include_metrics=True: [],
    )
    monkeypatch.setattr("agent.graph.make_chat_model", lambda *a, **k: FakeLLM())
    monkeypatch.setattr("agent.graph.apply_continuity_threads", _boom)

    cfg = RunConfig(provider="anthropic", model="claude-haiku-4-5-20251001", api_key="sk-ant-test")
    initial_state = _state([
        HumanMessage(content="Generate a research brief for AAPL."),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _tool_msg("search_transcript"),
    ], 0)

    final = create_graph(cfg).invoke(initial_state)
    brief = final["brief"]

    # The brief is the real synthesized one (not the _partial_brief skeleton
    # produced by the blanket except in synthesis_node), with continuity_threads
    # left untouched (unpersisted) rather than crashing the whole run.
    assert final["synthesis_error"] is None
    assert brief["company_name"] == "Apple Inc."
    assert brief["continuity_threads"] == []


def test_synthesis_requests_expanded_max_tokens(monkeypatch):
    fact = {
        "text": "Revenue increased 12%.",
        "source": "10-Q",
        "reliability": "HIGH",
        "evidence_snippet": "Revenue increased 12%.",
    }
    brief_json = json.dumps({
        "ticker": "AAPL",
        "company_name": "Apple Inc.",
        "filing_date": "2026-04-30",
        "what_matters_most": "Unsupported synthesis commentary.",
        "standout_number": fact,
        "what_changed": [],
        "bull_points": [fact],
        "bear_points": [],
        "what_to_watch": ["Watch Q3 gross margin"],
        "trends": [],
        "mda_summary": {
            "drivers": [],
            "headwinds": [],
            "language_shift": "No prior-period comparison was available.",
            "key_quote": fact,
        },
        "risks_categorized": [],
        "management_commentary": [],
        "guidance_history": [],
        "sentiment": None,
        "market_expectations": None,
    })

    class FakeLLM:
        def bind_tools(self, tools):
            return self

        def invoke(self, messages):
            return AIMessage(content="done", tool_calls=[])

        def stream(self, messages):
            yield AIMessage(content=brief_json)

    calls = []

    def spy_make_chat_model(*args, **kwargs):
        calls.append(kwargs)
        return FakeLLM()

    monkeypatch.setattr("analysis.textdiff.compute", lambda ticker: [])
    monkeypatch.setattr("analysis.tone_drift.compute", lambda ticker: [])
    monkeypatch.setattr(
        "agent.company_profile.collect_profile_evidence",
        lambda ticker, include_metrics=True: [],
    )
    monkeypatch.setattr("agent.graph.make_chat_model", spy_make_chat_model)

    cfg = RunConfig(
        provider="anthropic",
        model="claude-haiku-4-5-20251001",
        api_key="sk-ant-test",
    )
    initial_state = _state([
        HumanMessage(content="Generate a research brief for AAPL."),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _tool_msg("search_transcript"),
    ], 0)

    create_graph(cfg).invoke(initial_state)

    from agent.graph import SYNTHESIS_MAX_TOKENS

    assert SYNTHESIS_MAX_TOKENS == 64000
    assert any(
        kwargs.get("max_tokens") == SYNTHESIS_MAX_TOKENS
        for kwargs in calls
    )


def test_synthesis_truncation_produces_explicit_partial_reason(monkeypatch):
    class FakeLLM:
        def bind_tools(self, tools):
            return self

        def invoke(self, messages):
            return AIMessage(content="done", tool_calls=[])

        def stream(self, messages):
            yield AIMessage(
                content='{"ticker": "AAPL", "company_name": "Apple',
                response_metadata={"stop_reason": "max_tokens"},
            )

    monkeypatch.setattr("analysis.textdiff.compute", lambda ticker: [])
    monkeypatch.setattr("analysis.tone_drift.compute", lambda ticker: [])
    monkeypatch.setattr(
        "agent.company_profile.collect_profile_evidence",
        lambda ticker, include_metrics=True: [],
    )
    monkeypatch.setattr("agent.graph.make_chat_model", lambda *args, **kwargs: FakeLLM())

    cfg = RunConfig(
        provider="anthropic",
        model="claude-haiku-4-5-20251001",
        api_key="sk-ant-test",
    )
    initial_state = _state([
        HumanMessage(content="Generate a research brief for AAPL."),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _tool_msg("search_transcript"),
    ], 0)

    final = create_graph(cfg).invoke(initial_state)
    brief = final["brief"]

    assert brief["status"] == "PARTIAL"
    assert "truncated" in brief["evidence_notes"][0]
    assert "token limit" in brief["evidence_notes"][0]
    assert not brief["evidence_notes"][0].startswith("Synthesis failed validation")
    assert "truncated" in final["synthesis_error"]


def test_partial_brief_falls_back_to_metrics_db(monkeypatch):
    monkeypatch.setattr(
        "storage.metrics_db.get_metrics",
        lambda ticker: {
            "company_name": "Apple Inc.",
            "filing_date": "2026-07-31",
            "period": "Q32026",
            "form_type": "10-Q",
        },
    )

    partial = _partial_brief(
        _state([], 0), "Required evidence was unavailable or invalid."
    )

    assert partial["company_name"] == "Apple Inc."
    assert partial["filing_date"] == "2026-07-31"
    assert partial["data_as_of"] == "2026-07-31"
    assert partial["what_matters_most"] == NO_VERIFIED_SYNTHESIS_MESSAGE
    assert partial["evidence_notes"][0] == "Required evidence was unavailable or invalid."


def test_partial_brief_metrics_db_failure_falls_back_to_uppercase_ticker(monkeypatch):
    def fail_metrics_lookup(ticker):
        raise RuntimeError("metrics unavailable")

    monkeypatch.setattr("storage.metrics_db.get_metrics", fail_metrics_lookup)

    partial = _partial_brief(_state([], 0), "No usable evidence.")

    assert partial["company_name"] == "AAPL"
    assert partial["filing_date"] == ""


def test_synthesis_pops_company_profile_before_validation(monkeypatch):
    from agent.post_synthesis import apply_reliability
    from agent.schemas import BriefOutput

    record = make_evidence_record(
        source="10-Q",
        content="Revenue increased due to higher demand.",
        document_id="sec:AAPL:brief",
        chunk_id="mda:0",
        as_of="2026-04-30",
    )
    fact = {
        "text": "Revenue increased due to higher demand.",
        "source": "10-Q",
        "reliability": "HIGH",
        "evidence_snippet": "Revenue increased due to higher demand.",
        "evidence_ref": record.ref.model_dump(mode="json"),
    }
    payload = evidence_envelope(tool="search_filing", records=[record])
    data = {
        "ticker": "AAPL",
        "company_name": "Apple Inc.",
        "filing_date": "2026-04-30",
        "what_matters_most": "Verified demand evidence is the central fact.",
        "standout_number": fact,
        "what_changed": [],
        "bull_points": [],
        "bear_points": [],
        "what_to_watch": [],
        "trends": [],
        "mda_summary": {
            "drivers": [],
            "headwinds": [],
            "language_shift": "No verified cross-period shift.",
            "key_quote": fact,
        },
        "risks_categorized": [],
        "management_commentary": [],
        "guidance_history": [],
        "company_profile": {
            "business_lines": [{
                "name": "Unsupported",
                "description": {
                    **fact,
                    "text": "This profile fact is not in the record.",
                },
            }],
        },
    }

    profile_section = _pop_company_profile(data)
    brief = BriefOutput.model_validate(data)
    verified = apply_reliability(brief.model_dump(), evidence_payloads=[payload])

    assert "company_profile" not in data
    assert profile_section["business_lines"][0]["name"] == "Unsupported"
    assert verified["evidence_coverage"] == {
        "status": "VERIFIED",
        "verified": 2,
        "unverified": 0,
        "failed": 0,
        "total": 2,
    }

    finalized = {"ticker": "AAPL", "status": "PARTIAL"}
    calls = {}

    def fake_finalize(ticker, section, payloads, model):
        calls["finalize"] = (ticker, section, payloads, model)
        return finalized

    def fake_save(ticker, profile):
        calls["save"] = (ticker, profile)

    monkeypatch.setattr(
        "agent.company_profile.finalize_profile_from_synthesis", fake_finalize
    )
    monkeypatch.setattr("storage.company_profiles.save_profile", fake_save)
    state = _state([_tool_msg("search_filing")], 1)
    state["profile_payloads"] = [payload]

    assert _finalize_synthesis_profile(state, profile_section, "test-model") == finalized
    assert calls["finalize"][0] == "AAPL"
    assert calls["finalize"][2] == state["messages"] + [payload]
    assert calls["save"] == ("AAPL", finalized)

    monkeypatch.setattr(
        "agent.company_profile.finalize_profile_from_synthesis",
        lambda *args: (_ for _ in ()).throw(RuntimeError("profile failed")),
    )
    before_failure = {k: v for k, v in verified.items() if k != "company_profile"}
    verified["company_profile"] = _finalize_synthesis_profile(
        state, profile_section, "test-model"
    )
    assert verified["company_profile"] is None
    assert {key: value for key, value in verified.items() if key != "company_profile"} == before_failure


# ── Cap-based routing (existing tests, renamed) ──────────────────────────────


def test_routes_to_synthesis_when_no_tool_calls():
    # Coverage satisfied with valid evidence.v1 records.
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _tool_msg("search_transcript"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=3)
    assert should_continue(state) == "synthesis"


def test_routes_to_tools_when_tool_calls_under_cap():
    state = _state([HumanMessage(content="brief"), _ai_with_tools()], tool_round_count=3)
    assert should_continue(state) == "tools"


def test_routes_to_partial_at_cap_without_evidence_floor():
    state = _state([HumanMessage(content="brief"), _ai_with_tools()], tool_round_count=10)
    assert should_continue(state) == "partial"


def test_routes_to_partial_above_cap_without_evidence_floor():
    state = _state([HumanMessage(content="brief"), _ai_with_tools()], tool_round_count=11)
    assert should_continue(state) == "partial"


# ── Nudge / minimum-evidence floor ───────────────────────────────────────────


def test_routes_to_nudge_when_no_filing_or_transcript_evidence():
    """Agent stops, but filing + transcript never called → force one more round."""
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _tool_msg("get_analyst_expectations"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=2, nudge_fired=False)
    assert should_continue(state) == "nudge"


def test_routes_to_nudge_when_only_filing_present():
    """Filing touched but transcript missing → still nudge."""
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("search_filing"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=2, nudge_fired=False)
    assert should_continue(state) == "nudge"


def test_routes_to_partial_after_nudge_when_evidence_floor_is_still_missing():
    """The nudge is one-shot, but missing primary evidence still fails closed."""
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=3, nudge_fired=True)
    assert should_continue(state) == "partial"


def test_routes_to_synthesis_after_nudge_with_metrics_and_primary_filing():
    """Incomplete secondary coverage may degrade the brief, not fabricate it."""
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=3, nudge_fired=True)

    assert should_continue(state) == "synthesis"
    assert _coverage_report(messages)["status"] == "PARTIAL"


def test_routes_to_synthesis_when_filing_and_transcript_present():
    """Floor satisfied (≥2 filing calls + transcript) → no nudge, go to synthesis."""
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _tool_msg("search_transcript"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=4, nudge_fired=False)
    assert should_continue(state) == "synthesis"


def test_nudge_not_triggered_at_cap_but_missing_floor_routes_partial():
    """At cap no extra round is attempted and missing primary evidence is explicit."""
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=10, nudge_fired=False)
    assert should_continue(state) == "partial"


def test_nudge_node_sets_flag_and_appends_message():
    """nudge_node must set nudge_fired=True and inject a HumanMessage."""
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=2, nudge_fired=False)
    result = nudge_node(state)

    assert result["nudge_fired"] is True
    assert len(result["messages"]) == 1
    new_msg = result["messages"][0]
    assert isinstance(new_msg, HumanMessage)
    # Should mention both missing tools when neither was called.
    assert "search_filing" in new_msg.content
    assert "search_transcript" in new_msg.content


def test_nudge_node_mentions_only_missing_tool():
    """If only the transcript requirement is missing, nudge mentions just that one."""
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=3, nudge_fired=False)
    result = nudge_node(state)

    new_msg = result["messages"][0]
    assert "search_transcript" in new_msg.content
    assert "search_filing" not in new_msg.content


def test_plain_text_tool_outputs_never_satisfy_coverage():
    messages = [
        HumanMessage(content="brief"),
        ToolMessage(content="ok", name="get_financial_metrics", tool_call_id="metrics"),
        ToolMessage(content="ok", name="search_filing", tool_call_id="filing"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=3, nudge_fired=True)

    assert should_continue(state) == "partial"
    assert _coverage_report(messages)["metrics"]["status"] == "NOT_CALLED"


def test_tampered_ok_envelope_is_reported_invalid_and_fails_closed():
    valid_message = _tool_msg("get_financial_metrics")
    tampered = json.loads(valid_message.content)
    tampered["records"][0]["content"] = "Tampered after hashing."
    invalid_message = ToolMessage(
        content=json.dumps(tampered),
        name="get_financial_metrics",
        tool_call_id="metrics_tampered",
    )
    messages = [HumanMessage(content="brief"), invalid_message, _ai_done()]
    state = _state(messages, tool_round_count=3, nudge_fired=True)

    assert should_continue(state) == "partial"
    coverage = _coverage_report(messages)
    assert coverage["metrics"]["status"] == "INVALID"
    assert coverage["metrics"]["evidence_count"] == 0


def test_envelope_tool_name_and_record_source_must_match_call():
    filing_record = make_evidence_record(
        source="10-Q",
        content="A filing cannot masquerade as the metrics tool.",
        document_id="sec:AAPL:wrong-tool",
    )
    filing_payload = json.loads(evidence_envelope(
        tool="search_filing", records=[filing_record]
    ))
    filing_payload["tool"] = "get_financial_metrics"
    wrong_source = ToolMessage(
        content=json.dumps(filing_payload),
        name="get_financial_metrics",
        tool_call_id="wrong_source",
    )
    wrong_name = ToolMessage(
        content=evidence_envelope(
            tool="search_filing", records=[filing_record]
        ),
        name="get_financial_metrics",
        tool_call_id="wrong_name",
    )

    source_coverage = _coverage_report([wrong_source])
    name_coverage = _coverage_report([wrong_name])
    assert source_coverage["metrics"]["status"] == "INVALID"
    assert source_coverage["metrics"]["evidence_count"] == 0
    assert name_coverage["metrics"]["status"] == "NOT_CALLED"


def test_error_envelopes_never_satisfy_evidence_floor():
    messages = [
        HumanMessage(content="brief"),
        _error_tool_msg("get_financial_metrics"),
        _error_tool_msg("search_filing"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=3, nudge_fired=True)

    assert should_continue(state) == "partial"
    assert _coverage_report(messages)["filings"]["status"] == "ERROR"


def test_empty_transcript_is_a_confirmed_gap_and_does_not_block_synthesis():
    messages = [
        HumanMessage(content="brief"),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _empty_tool_msg("search_transcript"),
        _ai_done(),
    ]
    state = _state(messages, tool_round_count=4)

    assert should_continue(state) == "synthesis"
    coverage = _coverage_report(messages)
    assert coverage["transcripts"]["status"] == "EMPTY"
    assert coverage["status"] == "PARTIAL"


# ── _extract_json ─────────────────────────────────────────────────────────────


def test_extract_json_simple_object():
    """Well-formed single object passes through intact."""
    raw = '{"ticker": "AAPL", "value": 42}'
    result = _extract_json(raw)
    assert json.loads(result) == {"ticker": "AAPL", "value": 42}


def test_extract_json_nested_braces():
    """Nested objects are not truncated at the first closing brace."""
    raw = '{"a": {"b": 1}, "c": 2}'
    result = _extract_json(raw)
    assert json.loads(result) == {"a": {"b": 1}, "c": 2}


def test_extract_json_two_objects_concatenated():
    """Two JSON objects concatenated — only the first is returned.

    This is the exact failure mode from 'Extra data: line 478 column 1'.
    """
    raw = '{"ticker": "NVDA", "value": 1}\n{"ticker": "AAPL", "value": 2}'
    result = _extract_json(raw)
    parsed = json.loads(result)   # must not raise Extra data
    assert parsed == {"ticker": "NVDA", "value": 1}


def test_extract_json_trailing_prose():
    """Object followed by LLM commentary text — only the object is returned."""
    raw = '{"x": 1}\n\nNote: This brief covers Q1 2025 results.'
    result = _extract_json(raw)
    assert json.loads(result) == {"x": 1}


def test_extract_json_markdown_fence():
    """JSON wrapped in a markdown code fence is correctly extracted."""
    raw = "```json\n{\"ticker\": \"MSFT\"}\n```"
    result = _extract_json(raw)
    assert json.loads(result) == {"ticker": "MSFT"}


# ── _cap_signals ──────────────────────────────────────────────────────────────


def _sig(kind: str, source: str, significance: str) -> dict:
    return {"kind": kind, "source": source, "significance": significance, "term": ""}


def test_cap_signals_global_cap():
    signals = [_sig("risk_added", "10-Q", "HIGH") for _ in range(20)]
    capped = _cap_signals(signals)
    assert len(capped) <= MAX_EDGE_SIGNALS
    assert len(capped) <= MAX_FILING_SIGNALS  # all filing-sourced here


def test_cap_signals_per_source_caps():
    signals = (
        [_sig("risk_added", "10-Q", "HIGH") for _ in range(10)]
        + [_sig("recurring_evasion", "transcript", "HIGH") for _ in range(10)]
    )
    capped = _cap_signals(signals)
    filing = [s for s in capped if s["source"] != "transcript"]
    transcript = [s for s in capped if s["source"] == "transcript"]
    assert len(filing) <= MAX_FILING_SIGNALS
    assert len(transcript) <= MAX_TRANSCRIPT_SIGNALS
    assert len(capped) <= MAX_EDGE_SIGNALS


def test_cap_signals_high_significance_first():
    signals = [
        _sig("term_frequency", "10-Q", "MEDIUM"),
        _sig("risk_added", "10-Q", "HIGH"),
        _sig("kpi_dropped", "10-Q", "LOW"),
    ]
    capped = _cap_signals(signals)
    assert [s["significance"] for s in capped] == ["HIGH", "MEDIUM", "LOW"]


def test_cap_signals_empty():
    assert _cap_signals([]) == []


# ── _format_signals_message — transcript kinds ────────────────────────────────


def test_format_signals_message_labels_transcript_kinds():
    signals = [
        {"kind": "tone_trend", "significance": "HIGH", "term": "hedging language",
         "computed_metric": "hedge-word rate 31→44→59 per 10k words", "source": "transcript",
         "period_from": "Q32025", "period_to": "Q12026", "before_text": "", "after_text": ""},
        {"kind": "recurring_evasion", "significance": "HIGH", "term": "china / pricing",
         "computed_metric": "asked in Q32025, Q12026; 2/2 answers non-quantitative",
         "source": "transcript", "period_from": "Q32025", "period_to": "Q12026",
         "before_text": "Analyst: question?", "after_text": "Too early to say."},
        {"kind": "topic_arc", "significance": "MEDIUM", "term": "inventory",
         "computed_metric": "1→4→7 mentions", "source": "transcript",
         "period_from": "Q32025", "period_to": "Q12026", "before_text": "", "after_text": ""},
        {"kind": "topic_fade", "significance": "MEDIUM", "term": "backlog",
         "computed_metric": "'backlog' absent in Q12026", "source": "transcript",
         "period_from": "Q32025", "period_to": "Q12026", "before_text": "Backlog grew.", "after_text": ""},
    ]
    msg = _format_signals_message(signals)
    assert "MANAGEMENT TONE TREND" in msg
    assert "RECURRING Q&A EVASION" in msg
    assert "TRANSCRIPT TOPIC ARC" in msg
    assert "PREPARED-REMARKS TOPIC FADE" in msg
    assert "hedge-word rate 31→44→59" in msg


# ── create_graph(config) — provider/model/key threading ───────────────────────


def test_create_graph_compiles_with_no_config_no_network():
    from agent.graph import create_graph
    graph = create_graph()
    assert graph is not None


def test_create_graph_compiles_with_explicit_anthropic_config_no_network():
    from agent.graph import create_graph
    from agent.llm import RunConfig
    cfg = RunConfig(provider="anthropic", model="claude-haiku-4-5-20251001", api_key="sk-ant-test")
    graph = create_graph(cfg)
    assert graph is not None


def test_create_graph_compiles_with_openai_config_no_network():
    from agent.graph import create_graph
    from agent.llm import RunConfig
    cfg = RunConfig(provider="openai", model="gpt-5-mini", api_key="sk-test")
    graph = create_graph(cfg)
    assert graph is not None


def test_run_brief_accepts_legacy_two_arg_call(monkeypatch):
    """run_brief(ticker, language) — the pre-refactor call shape — must still work."""
    import agent.graph as graph_module

    captured = {}

    class _FakeGraph:
        def invoke(self, initial):
            captured["initial"] = initial
            return {"brief": {"ticker": "NVDA"}}

    def _fake_create_graph(config=None):
        captured["config"] = config
        return _FakeGraph()

    monkeypatch.setattr(graph_module, "create_graph", _fake_create_graph)
    result = graph_module.run_brief("NVDA", "English")

    assert result == {"ticker": "NVDA"}
    assert captured["config"] is None
    assert captured["initial"]["ticker"] == "NVDA"


def test_two_runs_persist_and_surface_continuity_thread(monkeypatch, tmp_path):
    import storage.threads_db as threads_db_module
    monkeypatch.setattr(threads_db_module, "DB_PATH", tmp_path / "threads.db")
    threads_db_module.init_db()

    monkeypatch.setattr("analysis.textdiff.compute", lambda ticker: [])
    monkeypatch.setattr("analysis.tone_drift.compute", lambda ticker: [])
    monkeypatch.setattr(
        "agent.company_profile.collect_profile_evidence",
        lambda ticker, include_metrics=True: [],
    )

    # Give `fact` a real evidence_ref matching the search_filing tool_msg record
    # below, so it passes deterministic verification -- otherwise
    # _filter_unverified_claims (agent/evidence.py) now correctly drops the
    # continuity thread using it before apply_continuity_threads ever runs.
    filing_record = make_evidence_record(
        source="10-Q",
        content="Verified evidence returned by search_filing.",
        document_id="test:search_filing:1",
        chunk_id="0",
        as_of="2026-04-30",
    )
    fact = {
        # No digits here: the evidence content has none, and verify_fact fails
        # closed on any claim number the cited record doesn't contain.
        "text": "Revenue increased due to strong demand.", "source": "10-Q", "reliability": "HIGH",
        "evidence_snippet": "Verified evidence returned by search_filing.",
        "evidence_ref": filing_record.ref.model_dump(mode="json"),
    }

    def _brief_json(filing_date, continuity_threads):
        return json.dumps({
            "ticker": "AAPL", "company_name": "Apple Inc.", "filing_date": filing_date,
            "what_matters_most": "x", "standout_number": fact,
            "what_changed": [], "bull_points": [fact], "bear_points": [],
            "what_to_watch": ["Watch Q3 gross margin"], "trends": [],
            "continuity_threads": continuity_threads,
            "mda_summary": {"drivers": [], "headwinds": [], "language_shift": "x", "key_quote": fact},
            "risks_categorized": [], "management_commentary": [], "guidance_history": [],
            "sentiment": None, "market_expectations": None,
        })

    cfg = RunConfig(provider="anthropic", model="claude-haiku-4-5-20251001", api_key="sk-ant-test")
    base_messages = [
        HumanMessage(content="Generate a research brief for AAPL."),
        _tool_msg("get_financial_metrics"),
        _tool_msg("search_filing"),
        _tool_msg("search_filing", suffix="2"),
        _tool_msg("search_transcript"),
    ]

    class FakeLLM1:
        def bind_tools(self, tools):
            return self

        def invoke(self, messages):
            return AIMessage(content="done", tool_calls=[])

        def stream(self, messages):
            yield AIMessage(content=_brief_json("2026-01-30", [
                {"thread_id": None, "topic": "Cloud margin compression", "category": "Operational",
                 "status": "new", "note": "First flagged.", "evidence": fact}
            ]))

    monkeypatch.setattr("agent.graph.make_chat_model", lambda *a, **k: FakeLLM1())
    first = create_graph(cfg).invoke(_state(base_messages, 0))
    thread_id = first["brief"]["continuity_threads"][0]["thread_id"]
    assert thread_id

    captured = {}

    class FakeLLM2:
        def bind_tools(self, tools):
            return self

        def invoke(self, messages):
            captured["messages"] = messages
            return AIMessage(content="done", tool_calls=[])

        def stream(self, messages):
            yield AIMessage(content=_brief_json("2026-04-30", [
                {"thread_id": thread_id, "topic": "Cloud margin compression", "category": "Operational",
                 "status": "resolved", "note": "Fixed per Q2 call.", "evidence": fact}
            ]))

    monkeypatch.setattr("agent.graph.make_chat_model", lambda *a, **k: FakeLLM2())
    second = create_graph(cfg).invoke(_state(base_messages, 0))

    injected = "\n".join(m.content for m in captured["messages"] if isinstance(m, HumanMessage))
    assert thread_id in injected
    assert "Cloud margin compression" in injected
    assert second["brief"]["continuity_threads"][0]["status"] == "resolved"

    from storage.threads_db import get_open_threads
    assert get_open_threads("AAPL") == []  # resolved thread is no longer open


def test_market_expectations_aligned_reflects_the_consensus_gate(monkeypatch):
    """The flag is computed from the archive, not hardcoded False."""
    import agent.graph as graph

    monkeypatch.setattr(
        graph, "aligned_consensus",
        lambda ticker, period, metric: ({"value": 1.24}, "ALIGNED"),
    )
    assert graph._market_expectations_aligned("NVDA", "Q32027") is True

    monkeypatch.setattr(
        graph, "aligned_consensus",
        lambda ticker, period, metric: (None, "basis_mismatch"),
    )
    assert graph._market_expectations_aligned("NVDA", "Q32027") is False


def test_market_expectations_aligned_is_false_without_a_period():
    import agent.graph as graph
    assert graph._market_expectations_aligned("NVDA", None) is False


def test_market_expectations_aligned_swallows_errors(monkeypatch):
    """An archive read failure must not fail a brief."""
    import agent.graph as graph

    def _boom(*a, **k):
        raise RuntimeError("archive unreadable")

    monkeypatch.setattr(graph, "aligned_consensus", _boom)
    assert graph._market_expectations_aligned("NVDA", "Q32027") is False


def test_market_expectations_aligned_metrics_lists_only_what_passed(monkeypatch):
    """The boolean collapses per-metric alignment, and by design only revenue
    will normally pass — basis_mismatch is the documented, expected outcome
    for eps. A consumer reading only the boolean and rendering an
    actual-vs-consensus EPS line would render a comparison the gate refused.
    The list is what a per-metric consumer must check."""
    import agent.graph as graph

    def _gate(ticker, period, metric):
        if metric == "revenue":
            return {"value": 54e9}, "ALIGNED"
        return None, "basis_mismatch"

    monkeypatch.setattr(graph, "aligned_consensus", _gate)
    assert graph._market_expectations_aligned_metrics("NVDA", "Q32027") == ["revenue"]
    assert graph._market_expectations_aligned("NVDA", "Q32027") is True


def test_market_expectations_aligned_metrics_is_empty_when_nothing_passes(monkeypatch):
    import agent.graph as graph

    monkeypatch.setattr(
        graph, "aligned_consensus",
        lambda ticker, period, metric: (None, "no_live_snapshot"),
    )
    assert graph._market_expectations_aligned_metrics("NVDA", "Q32027") == []
    assert graph._market_expectations_aligned("NVDA", "Q32027") is False


def test_market_expectations_aligned_metrics_is_empty_without_a_period_or_on_error(
    monkeypatch,
):
    import agent.graph as graph

    assert graph._market_expectations_aligned_metrics("NVDA", None) == []

    def _boom(*a, **k):
        raise RuntimeError("archive unreadable")

    monkeypatch.setattr(graph, "aligned_consensus", _boom)
    assert graph._market_expectations_aligned_metrics("NVDA", "Q32027") == []


def test_market_expectations_aligned_metrics_lists_both_when_both_pass(monkeypatch):
    """Order is stable and metric-named, so a consumer can index it directly."""
    import agent.graph as graph

    monkeypatch.setattr(
        graph, "aligned_consensus",
        lambda ticker, period, metric: ({"value": 1.0}, "ALIGNED"),
    )
    assert graph._market_expectations_aligned_metrics("NVDA", "Q32027") == [
        "revenue", "eps",
    ]


def test_partial_brief_display_policy_carries_an_empty_aligned_metrics_list():
    """A partial brief has no verified market data, so the list is empty and
    the key is still present — a consumer must never have to distinguish
    "absent" from "nothing aligned"."""
    partial = _partial_brief(_state([], 0), "No usable evidence.")
    policy = partial["display_policy"]
    assert policy["market_expectations_aligned"] is False
    assert policy["market_expectations_aligned_metrics"] == []


def test_langgraph_json_entrypoint_resolves():
    """langgraph.json names a module:symbol pair that LangGraph Studio
    imports directly. A symbol that does not exist makes run_studio.ps1 fail
    at load time with no useful message.

    The resolved symbol must also be invocable with zero arguments, since
    LangGraph Studio calls a factory bare. We check that by inspecting the
    signature only — we deliberately never call the factory here: building
    a real graph (create_graph()) costs several seconds, and paying that on
    every test run is exactly the import-time cost Ruling A-2 declined to
    accept."""
    import importlib
    import inspect
    import json
    from pathlib import Path

    config = json.loads(Path("langgraph.json").read_text(encoding="utf-8"))
    target = config["graphs"]["brief_agent"]
    module_path, _, symbol = target.partition(":")

    module_name = (
        module_path.removeprefix("./").removesuffix(".py").replace("/", ".")
    )
    module = importlib.import_module(module_name)

    assert hasattr(module, symbol), (
        f"langgraph.json points at {target!r} but {module_name} has no {symbol!r}"
    )

    obj = getattr(module, symbol)
    assert callable(obj), f"{module_name}.{symbol} is not callable"

    params = inspect.signature(obj).parameters.values()
    no_default = [
        p.name
        for p in params
        if p.default is inspect.Parameter.empty
        and p.kind not in (p.VAR_POSITIONAL, p.VAR_KEYWORD)
    ]
    assert no_default == [], (
        f"{module_name}.{symbol} requires argument(s) {no_default} with no "
        "default — LangGraph Studio calls the factory with zero arguments"
    )