"""Tests for contextweaver.context.manager."""

from __future__ import annotations

import json

import pytest

from contextweaver.config import ContextBudget, ContextPolicy
from contextweaver.context.classify import HeuristicSensitivityClassifier
from contextweaver.context.manager import ContextManager
from contextweaver.exceptions import (
    ContextWeaverError,
    DuplicateItemError,
    ItemNotFoundError,
    PolicyViolationError,
)
from contextweaver.routing.catalog import Catalog
from contextweaver.routing.router import Router
from contextweaver.routing.tree import TreeBuilder
from contextweaver.store import StoreBundle
from contextweaver.store.event_log import InMemoryEventLog
from contextweaver.store.facts import Fact
from contextweaver.types import (
    ArtifactRef,
    ContextItem,
    ContextPack,
    ItemKind,
    Phase,
    ResultEnvelope,
    SelectableItem,
    Sensitivity,
)


def _make_log(*texts: str) -> InMemoryEventLog:
    log = InMemoryEventLog()
    for i, text in enumerate(texts):
        log.append(ContextItem(id=f"item-{i}", kind=ItemKind.user_turn, text=text))
    return log


class _RecordingHook:
    """Capture build lifecycle callbacks for production-pipeline tests."""

    def __init__(self) -> None:
        self.excluded: list[tuple[list[str], str]] = []
        self.budget_events: list[tuple[int, int]] = []

    def on_context_built(self, pack: ContextPack) -> None:
        _ = pack

    def on_firewall_triggered(self, item: ContextItem, reason: str) -> None:
        _ = item, reason

    def on_items_excluded(self, items: list[ContextItem], reason: str) -> None:
        self.excluded.append(([item.id for item in items], reason))

    def on_budget_exceeded(self, requested: int, budget: int) -> None:
        self.budget_events.append((requested, budget))

    def on_route_completed(self, tool_ids: list[str]) -> None:
        _ = tool_ids


@pytest.mark.asyncio
async def test_build_returns_context_pack() -> None:
    log = _make_log("Hello world", "Search the database")
    mgr = ContextManager(event_log=log)
    pack = await mgr.build(phase=Phase.answer, query="database")
    assert isinstance(pack, ContextPack)
    assert pack.phase == Phase.answer


@pytest.mark.asyncio
async def test_build_empty_log() -> None:
    mgr = ContextManager()
    pack = await mgr.build(phase=Phase.route)
    assert pack.prompt == ""
    assert pack.stats.total_candidates == 0


@pytest.mark.asyncio
async def test_build_includes_items() -> None:
    log = _make_log("User asks about the database")
    mgr = ContextManager(event_log=log)
    pack = await mgr.build(phase=Phase.answer, query="database")
    assert "database" in pack.prompt.lower()


@pytest.mark.asyncio
async def test_build_stats_populated() -> None:
    log = _make_log("item one", "item two", "item three")
    mgr = ContextManager(event_log=log)
    pack = await mgr.build(phase=Phase.answer)
    assert pack.stats.total_candidates == 3


def test_build_stats_attribute_every_drop_and_fire_hooks() -> None:
    log = InMemoryEventLog()
    log.append(ContextItem(id="keep", kind=ItemKind.user_turn, text="keep", token_estimate=20))
    log.append(ContextItem(id="dup-a", kind=ItemKind.doc_snippet, text="same text"))
    log.append(ContextItem(id="dup-b", kind=ItemKind.doc_snippet, text="same text"))
    log.append(ContextItem(id="kind-limit", kind=ItemKind.doc_snippet, text="different text"))
    log.append(
        ContextItem(
            id="secret",
            kind=ItemKind.memory_fact,
            text="secret",
            sensitivity=Sensitivity.confidential,
        )
    )
    log.append(ContextItem(id="budget", kind=ItemKind.agent_msg, text="large", token_estimate=100))
    policy = ContextPolicy()
    policy.max_items_per_kind[ItemKind.doc_snippet] = 1
    hook = _RecordingHook()
    mgr = ContextManager(
        event_log=log,
        budget=ContextBudget(answer=40),
        policy=policy,
        hook=hook,
    )

    pack = mgr.build_sync(phase=Phase.answer, query="keep")

    assert pack.stats.included_count + pack.stats.dropped_count == pack.stats.total_candidates
    assert pack.stats.total_candidates == 6
    dropped = {item.item_id: item.reason for item in pack.stats.dropped_items}
    assert dropped["secret"] == "sensitivity"
    assert dropped["budget"] == "budget"
    assert sorted(dropped.values()) == ["budget", "dedup", "kind_limit", "sensitivity"]
    assert pack.stats.dropped_reasons == {
        "budget": 1,
        "dedup": 1,
        "kind_limit": 1,
        "sensitivity": 1,
    }
    assert sorted(reason for _ids, reason in hook.excluded) == sorted(pack.stats.dropped_reasons)
    assert len(hook.budget_events) == 1


def test_build_sync() -> None:
    log = _make_log("synchronous test item")
    mgr = ContextManager(event_log=log)
    pack = mgr.build_sync(phase=Phase.answer)
    assert isinstance(pack, ContextPack)


@pytest.mark.asyncio
async def test_build_populates_envelopes_for_tool_results() -> None:
    log = InMemoryEventLog()
    log.append(ContextItem(id="u1", kind=ItemKind.user_turn, text="run query"))
    log.append(
        ContextItem(
            id="tr1",
            kind=ItemKind.tool_result,
            text="raw output: rows=[1,2,3]",
        )
    )
    mgr = ContextManager(event_log=log)
    pack = await mgr.build(phase=Phase.answer, query="query")
    assert len(pack.envelopes) == 1
    env = pack.envelopes[0]
    assert env.status == "ok"
    assert env.provenance["source_item_id"] == "tr1"


@pytest.mark.asyncio
async def test_build_sync_inside_running_loop() -> None:
    """build_sync() should work even when an event loop is already running."""
    log = _make_log("works inside async context")
    mgr = ContextManager(event_log=log)
    pack = mgr.build_sync(phase=Phase.answer)
    assert isinstance(pack, ContextPack)


# ---------------------------------------------------------------------------
# Ingestion methods
# ---------------------------------------------------------------------------


def test_ingest_appends_to_event_log() -> None:
    mgr = ContextManager()
    item = ContextItem(id="u1", kind=ItemKind.user_turn, text="hello")
    mgr.ingest(item)
    assert mgr.event_log.count() == 1
    assert mgr.event_log.get("u1").text == "hello"


def test_ingest_sync() -> None:
    mgr = ContextManager()
    item = ContextItem(id="u1", kind=ItemKind.user_turn, text="hello")
    mgr.ingest_sync(item)
    assert mgr.event_log.count() == 1


def test_ingest_tool_result_small() -> None:
    mgr = ContextManager()
    item, env = mgr.ingest_tool_result(
        tool_call_id="tc1",
        raw_output="status: ok\ncount: 5",
        tool_name="db_query",
    )
    assert item.kind == ItemKind.tool_result
    assert item.parent_id == "tc1"
    assert env.status == "ok"
    assert mgr.event_log.count() == 1


def test_ingest_tool_result_small_custom_extractor() -> None:
    """Custom extractor is used in the small-output path (below firewall threshold)."""

    class TagExtractor:
        def extract(self, raw: str, metadata: dict) -> list[str]:  # type: ignore[type-arg]
            return [f"[fact]{raw}"]

    mgr = ContextManager(extractor=TagExtractor())
    item, env = mgr.ingest_tool_result(
        tool_call_id="tc_ext",
        raw_output="short output",
        tool_name="small_tool",
    )
    assert env.status == "ok"
    assert env.facts == ["[fact]short output"]
    assert item.parent_id == "tc_ext"


def test_ingest_tool_result_large_triggers_firewall() -> None:
    mgr = ContextManager()
    large_output = "data: " + "x" * 3000
    item, env = mgr.ingest_tool_result(
        tool_call_id="tc2",
        raw_output=large_output,
        tool_name="big_tool",
        firewall_threshold=100,
    )
    assert item.artifact_ref is not None
    assert env.status == "ok"
    assert len(item.text) < len(large_output)
    assert mgr.event_log.count() == 1


def test_ingest_tool_result_sync() -> None:
    mgr = ContextManager()
    item, env = mgr.ingest_tool_result_sync(
        tool_call_id="tc3",
        raw_output="result: 42",
    )
    assert env.status == "ok"


def test_ingest_envelope_canonical_path() -> None:
    # Canonical Frame-shaped seam (issue #352): an already-firewalled
    # ResultEnvelope is appended verbatim, no re-firewalling of raw output.
    mgr = ContextManager()
    ref = ArtifactRef(
        handle="artifact:frame-1",
        media_type="text/plain",
        size_bytes=2048,
        label="upstream-firewalled log",
        content_hash="abc123",
    )
    envelope = ResultEnvelope(
        status="ok",
        summary="2048-byte log summarised upstream",
        facts=["errors: 0"],
        artifacts=[ref],
    )
    item = mgr.ingest_envelope("tc-frame", envelope, tool_name="run_job")

    assert item.kind == ItemKind.tool_result
    assert item.id == "result:tc-frame"
    assert item.parent_id == "tc-frame"
    assert item.text == "2048-byte log summarised upstream"
    assert item.artifact_ref is ref
    assert item.metadata["ingest"] == "envelope"
    assert mgr.event_log.count() == 1


def test_ingest_envelope_without_artifacts() -> None:
    mgr = ContextManager()
    item = mgr.ingest_envelope_sync("tc-bare", ResultEnvelope(status="ok", summary="no artifact"))
    assert item.artifact_ref is None
    assert item.text == "no artifact"


# ---------------------------------------------------------------------------
# Fact / Episode helpers
# ---------------------------------------------------------------------------


def test_add_fact() -> None:
    mgr = ContextManager()
    mgr.add_fact("user_name", "Alice")
    facts = mgr.fact_store.all()
    assert len(facts) == 1
    assert facts[0].key == "user_name"
    assert facts[0].value == "Alice"


def test_add_fact_sync() -> None:
    mgr = ContextManager()
    mgr.add_fact_sync("key", "value")
    assert len(mgr.fact_store.all()) == 1


def test_add_fact_no_overwrite_after_delete() -> None:
    """Delete-then-add must not silently overwrite a surviving fact (#462)."""
    mgr = ContextManager()
    mgr.add_fact("k", "a")  # fact:k:0
    mgr.add_fact("k", "b")  # fact:k:1
    first_id = mgr.fact_store.get_by_key("k")[0].fact_id
    mgr.fact_store.delete(first_id)  # remove fact:k:0 (value "a")
    mgr.add_fact("k", "c")  # must mint fact:k:2, NOT reuse fact:k:1
    survivors = {f.value for f in mgr.fact_store.get_by_key("k")}
    assert survivors == {"b", "c"}  # "b" was not clobbered
    assert len(mgr.fact_store.all()) == 2


def test_add_fact_ids_are_deterministic() -> None:
    """Identical call sequences yield identical fact IDs (#462 determinism)."""
    runs = []
    for _ in range(2):
        mgr = ContextManager()
        mgr.add_fact("x", "1")
        mgr.add_fact("y", "2")
        runs.append([f.fact_id for f in mgr.fact_store.all()])
    assert runs[0] == runs[1]
    assert runs[0] == ["fact:x:0", "fact:y:1"]


def test_add_fact_seeds_counter_from_prepopulated_store() -> None:
    """A pre-populated/persistent store (restart) seeds the counter past existing IDs (#462).

    Regression for the case where a manager is built over a fact store that
    already holds ``add_fact``-minted IDs: the counter must seed past them
    instead of restarting at 0 and clobbering ``fact:{key}:0``.
    """
    mgr = ContextManager()
    # Simulate a persistent store reloaded with prior facts across a restart.
    mgr.fact_store.put(Fact(fact_id="fact:k:0", key="k", value="old"))
    mgr.fact_store.put(Fact(fact_id="fact:k:1", key="k", value="older"))
    mgr.add_fact("k", "new")  # seeds to 2 → fact:k:2, no collision/overwrite
    values = {f.value for f in mgr.fact_store.get_by_key("k")}
    assert values == {"old", "older", "new"}
    assert len(mgr.fact_store.all()) == 3


def test_add_fact_collision_guard_is_defensive_backstop() -> None:
    """If the counter desyncs from the store, add_fact raises rather than overwriting (#462).

    Seeding makes a single-threaded collision unreachable, so this drives the
    guard directly by forcing a stale counter (e.g. a concurrent writer minted
    the same ID after this manager seeded).
    """
    mgr = ContextManager()
    mgr.fact_store.put(Fact(fact_id="fact:k:0", key="k", value="other"))
    # Force a stale, already-seeded counter pointing at an existing ID.
    mgr._fact_seq = 0
    mgr._fact_seq_seeded = True
    with pytest.raises(DuplicateItemError, match="already exists"):
        mgr.add_fact("k", "new")


def test_add_episode() -> None:
    mgr = ContextManager()
    mgr.add_episode("ep1", "User searched for data")
    episodes = mgr.episodic_store.all()
    assert len(episodes) == 1
    assert episodes[0].episode_id == "ep1"


def test_add_episode_sync() -> None:
    mgr = ContextManager()
    mgr.add_episode_sync("ep1", "summary")
    assert len(mgr.episodic_store.all()) == 1


# ---------------------------------------------------------------------------
# Facts + episodic in build output
# ---------------------------------------------------------------------------


def test_build_includes_facts_in_prompt() -> None:
    mgr = ContextManager()
    mgr.add_fact("user_lang", "Python")
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="hello"))
    pack = mgr.build_sync(phase=Phase.answer)
    assert "user_lang" in pack.prompt
    assert "Python" in pack.prompt


def test_build_caps_facts_by_line_count() -> None:
    """Fact injection is capped at 64 lines; excess produces an omitted notice."""
    mgr = ContextManager()
    # Zero-padded keys so lexicographic == numeric order
    for i in range(80):
        mgr.add_fact(f"k{i:03d}", f"v{i}")
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="hello"))
    pack = mgr.build_sync(phase=Phase.answer)
    assert "more facts omitted" in pack.prompt
    # Key 063 (last within cap) should be present, key 064 should not
    assert "k063" in pack.prompt
    assert "- k064:" not in pack.prompt


def test_build_caps_facts_by_char_budget() -> None:
    """Fact injection truncates when total chars exceed 2000."""
    mgr = ContextManager()
    # Each fact line is ~210 chars → 10 facts ≈ 2100 chars, exceeds 2000
    for i in range(15):
        mgr.add_fact(f"key{i}", "x" * 200)
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="hello"))
    pack = mgr.build_sync(phase=Phase.answer)
    assert "facts truncated to fit header budget" in pack.prompt


def test_build_includes_episodic_in_prompt() -> None:
    mgr = ContextManager()
    mgr.add_episode("ep1", "Previously searched for billing data")
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="hello"))
    pack = mgr.build_sync(phase=Phase.answer)
    assert "billing" in pack.prompt.lower()


# ---------------------------------------------------------------------------
# StoreBundle constructor
# ---------------------------------------------------------------------------


def test_stores_bundle_constructor() -> None:
    log = InMemoryEventLog()
    log.append(ContextItem(id="u1", kind=ItemKind.user_turn, text="pre-loaded"))
    bundle = StoreBundle(event_log=log)
    mgr = ContextManager(stores=bundle)
    assert mgr.event_log.count() == 1


# ---------------------------------------------------------------------------
# Budget override
# ---------------------------------------------------------------------------


def test_build_with_budget_override() -> None:
    log = _make_log("item about database queries")
    mgr = ContextManager(event_log=log)
    pack = mgr.build_sync(phase=Phase.answer, budget_tokens=50)
    assert isinstance(pack, ContextPack)


# ---------------------------------------------------------------------------
# Header/footer budget accounting
# ---------------------------------------------------------------------------


def test_header_footer_tokens_recorded_in_stats() -> None:
    """BuildStats.header_footer_tokens reflects injected facts/episodes cost."""
    mgr = ContextManager()
    mgr.add_fact("lang", "Python")
    mgr.add_episode("ep1", "Searched billing data")
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="hello"))
    pack = mgr.build_sync(phase=Phase.answer)
    assert pack.stats.header_footer_tokens > 0
    assert "[FACTS]" in pack.prompt
    assert "[EPISODIC MEMORY]" in pack.prompt


def test_header_footer_tokens_zero_without_injection() -> None:
    """Without facts or episodes, header_footer_tokens is 0."""
    log = _make_log("hello world")
    mgr = ContextManager(event_log=log)
    pack = mgr.build_sync(phase=Phase.answer)
    assert pack.stats.header_footer_tokens == 0


def test_facts_budget_subtracted_from_selection() -> None:
    """Injected facts reduce the budget available for context items."""
    # Tight budget: 100 tokens total. With facts injected, fewer items fit.
    mgr_no_facts = ContextManager()
    mgr_no_facts.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="a " * 50))
    mgr_no_facts.ingest(ContextItem(id="u2", kind=ItemKind.user_turn, text="b " * 50))
    pack_no = mgr_no_facts.build_sync(phase=Phase.answer, budget_tokens=100)

    mgr_with_facts = ContextManager()
    for i in range(10):
        mgr_with_facts.add_fact(f"k{i}", f"value-{i}")
    mgr_with_facts.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="a " * 50))
    mgr_with_facts.ingest(ContextItem(id="u2", kind=ItemKind.user_turn, text="b " * 50))
    pack_with = mgr_with_facts.build_sync(phase=Phase.answer, budget_tokens=100)

    # With facts consuming part of the budget, fewer items should be included
    # OR the total context-item tokens should be lower.
    assert pack_with.stats.header_footer_tokens > 0
    items_tokens_no = sum(pack_no.stats.tokens_per_section.values())
    items_tokens_with = sum(pack_with.stats.tokens_per_section.values())
    assert items_tokens_with <= items_tokens_no


# ---------------------------------------------------------------------------
# Per-phase build
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_build_route_phase() -> None:
    log = InMemoryEventLog()
    log.append(ContextItem(id="u1", kind=ItemKind.user_turn, text="search data"))
    mgr = ContextManager(event_log=log)
    pack = await mgr.build(phase=Phase.route, query="search")
    assert pack.phase == Phase.route


@pytest.mark.asyncio
async def test_build_call_phase() -> None:
    log = InMemoryEventLog()
    log.append(ContextItem(id="u1", kind=ItemKind.user_turn, text="search data"))
    mgr = ContextManager(event_log=log)
    pack = await mgr.build(phase=Phase.call, query="search")
    assert pack.phase == Phase.call


@pytest.mark.asyncio
async def test_build_interpret_phase() -> None:
    log = InMemoryEventLog()
    log.append(ContextItem(id="u1", kind=ItemKind.user_turn, text="query"))
    log.append(ContextItem(id="tr1", kind=ItemKind.tool_result, text="result data"))
    mgr = ContextManager(event_log=log)
    pack = await mgr.build(phase=Phase.interpret, query="interpret result")
    assert pack.phase == Phase.interpret


# ---------------------------------------------------------------------------
# build_route_prompt
# ---------------------------------------------------------------------------


def _make_selectable_items() -> list[SelectableItem]:
    """Build a small catalog for route-prompt tests."""
    return [
        SelectableItem(
            id="db_read",
            kind="tool",
            name="read_db",
            description="Read from database",
            tags=["data"],
        ),
        SelectableItem(
            id="send_email",
            kind="tool",
            name="send_email",
            description="Send email notification",
            tags=["comm"],
        ),
        SelectableItem(
            id="search_docs",
            kind="tool",
            name="search_docs",
            description="Search documentation pages",
            tags=["search"],
        ),
    ]


def test_build_route_prompt_returns_tuple() -> None:
    """build_route_prompt returns (ContextPack, cards, RouteResult)."""
    items = _make_selectable_items()
    graph = TreeBuilder(max_children=10).build(items)
    router = Router(graph, items=items, beam_width=2, top_k=5)

    log = InMemoryEventLog()
    log.append(ContextItem(id="u1", kind=ItemKind.user_turn, text="read database"))
    mgr = ContextManager(event_log=log)

    pack, cards, route_result = mgr.build_route_prompt(
        goal="Find data tools",
        query="read database",
        router=router,
    )

    # ContextPack for route phase
    assert isinstance(pack, ContextPack)
    assert pack.phase == Phase.route

    # Cards list corresponds to route candidates
    assert isinstance(cards, list)
    assert len(cards) == len(route_result.candidate_ids)

    # RouteResult has matching lengths
    assert len(route_result.scores) == len(route_result.candidate_ids)

    # Prompt includes GOAL header and AVAILABLE TOOLS footer
    assert "[GOAL]" in pack.prompt
    assert "Find data tools" in pack.prompt
    assert "[AVAILABLE TOOLS]" in pack.prompt


def test_build_route_prompt_sync_alias() -> None:
    """build_route_prompt_sync is a working alias."""
    items = _make_selectable_items()
    graph = TreeBuilder(max_children=10).build(items)
    router = Router(graph, items=items)

    mgr = ContextManager()
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="search docs"))

    pack, cards, result = mgr.build_route_prompt_sync(
        goal="Search goal",
        query="search docs",
        router=router,
    )
    assert pack.phase == Phase.route
    assert len(cards) > 0
    assert len(result.candidate_ids) > 0


# ---------------------------------------------------------------------------
# build_route_prompt history-aware (issue #27)
# ---------------------------------------------------------------------------


def test_build_route_prompt_with_no_tool_results_produces_no_history_adjustments() -> None:
    """First routing call in a session: log has no tool results → no history."""
    items = _make_selectable_items()
    graph = TreeBuilder(max_children=10).build(items)
    router = Router(graph, items=items)

    log = InMemoryEventLog()
    log.append(ContextItem(id="u1", kind=ItemKind.user_turn, text="read database"))
    mgr = ContextManager(event_log=log)
    _, _, result = mgr.build_route_prompt(
        goal="Find data tools", query="read database", router=router
    )
    assert result.history_adjustments == {}


def test_build_route_prompt_auto_constructs_history_from_event_log() -> None:
    """When the event log has a prior tool_result, the router sees a RouteHistory."""
    items = _make_selectable_items()
    graph = TreeBuilder(max_children=10).build(items)
    router = Router(graph, items=items, top_k=5)

    log = InMemoryEventLog()
    log.append(ContextItem(id="u1", kind=ItemKind.user_turn, text="read database"))
    log.append(
        ContextItem(
            id="tc1",
            parent_id=None,
            kind=ItemKind.tool_call,
            text="db_read invoked",
            metadata={"function_name": "db_read"},
        )
    )
    # In production, tool_result.parent_id points to the tool_call item id
    # ("tc1"), not the tool catalog id ("db_read"). The resolver derives the
    # catalog id via the tool_call's metadata["function_name"].
    log.append(
        ContextItem(
            id="tr1",
            parent_id="tc1",
            kind=ItemKind.tool_result,
            text="rows: id, name, email",
        )
    )

    mgr = ContextManager(event_log=log)
    _, _, result = mgr.build_route_prompt(
        goal="Find next tool", query="read database again", router=router
    )
    # db_read was already called → it should get a negative adjustment
    assert result.history_adjustments.get("db_read", 0.0) < 0.0


def test_build_route_prompt_history_from_log_can_be_disabled() -> None:
    """Setting history_from_log=False preserves stateless pre-#27 behaviour."""
    items = _make_selectable_items()
    graph = TreeBuilder(max_children=10).build(items)
    router = Router(graph, items=items, top_k=5)

    log = InMemoryEventLog()
    log.append(
        ContextItem(
            id="tc1",
            kind=ItemKind.tool_call,
            text="db_read",
            metadata={"function_name": "db_read"},
        )
    )
    log.append(ContextItem(id="tr1", parent_id="tc1", kind=ItemKind.tool_result, text="rows"))
    mgr = ContextManager(event_log=log)
    _, _, result = mgr.build_route_prompt(
        goal="g",
        query="read database",
        router=router,
        history_from_log=False,
    )
    assert result.history_adjustments == {}


def test_build_route_prompt_explicit_history_overrides_event_log_construction() -> None:
    """An explicit ``history=`` wins over the auto-built one."""
    from contextweaver.routing.history import RouteHistory

    items = _make_selectable_items()
    graph = TreeBuilder(max_children=10).build(items)
    # beam_width has to be wide enough that send_email is one of the candidates
    # for the query we send, otherwise it never reaches the history-adjustment
    # phase.
    router = Router(graph, items=items, beam_width=5, top_k=5)

    log = InMemoryEventLog()
    log.append(ContextItem(id="tr1", parent_id="db_read", kind=ItemKind.tool_result, text="rows"))
    mgr = ContextManager(event_log=log)
    explicit = RouteHistory(called_tool_ids=["send_email"], repeat_penalty=0.1)
    _, _, result = mgr.build_route_prompt(
        goal="g", query="send email notification", router=router, history=explicit
    )
    # The explicit history names send_email, so we expect send_email to get the
    # adjustment.  db_read is not in the explicit called list so it must NOT
    # receive a repeat penalty even though the log mentions db_read.
    assert "send_email" in result.history_adjustments
    assert "db_read" not in result.history_adjustments


# ---------------------------------------------------------------------------
# build_call_prompt
# ---------------------------------------------------------------------------


def _make_catalog() -> Catalog:
    """Build a small catalog with schema data for call-prompt tests."""
    catalog = Catalog()
    catalog.register(
        SelectableItem(
            id="db_read",
            kind="tool",
            name="read_db",
            description="Read from database",
            tags=["data"],
            args_schema={
                "query": {"type": "string", "description": "SQL query"},
                "limit": {"type": "integer", "default": 100},
            },
            examples=["read_db(query='SELECT * FROM users', limit=10)"],
            constraints={"max_rows": 1000},
            cost_hint=0.1,
        )
    )
    catalog.register(
        SelectableItem(
            id="send_email",
            kind="tool",
            name="send_email",
            description="Send email notification",
            tags=["comm"],
            side_effects=True,
        )
    )
    return catalog


def test_build_call_prompt_injects_schema() -> None:
    """build_call_prompt_sync injects the tool schema into the prompt header."""
    catalog = _make_catalog()
    mgr = ContextManager()
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="read user data"))

    pack = mgr.build_call_prompt_sync(
        tool_id="db_read",
        query="read user data",
        catalog=catalog,
    )
    assert isinstance(pack, ContextPack)
    assert pack.phase == Phase.call
    assert "[TOOL SCHEMA]" in pack.prompt
    assert "read_db" in pack.prompt
    assert "SQL query" in pack.prompt
    assert "max_rows" in pack.prompt
    assert "read_db(query='SELECT * FROM users'" in pack.prompt


def test_build_call_prompt_missing_tool_raises() -> None:
    """build_call_prompt raises ItemNotFoundError for unknown tool IDs."""
    catalog = _make_catalog()
    mgr = ContextManager()
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="hello"))

    with pytest.raises(ItemNotFoundError):
        mgr.build_call_prompt_sync(
            tool_id="nonexistent",
            query="hello",
            catalog=catalog,
        )


def test_build_call_prompt_schema_override() -> None:
    """build_call_prompt accepts a schema override."""
    catalog = _make_catalog()
    mgr = ContextManager()
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="send email"))

    custom_schema = {"to": {"type": "string"}, "body": {"type": "string"}}
    pack = mgr.build_call_prompt_sync(
        tool_id="send_email",
        query="send email",
        catalog=catalog,
        schema=custom_schema,
    )
    assert '"to"' in pack.prompt
    assert '"body"' in pack.prompt


def test_build_call_prompt_budget_enforcement() -> None:
    """Schema token cost is subtracted from the call-phase budget."""
    catalog = _make_catalog()
    mgr = ContextManager()
    # Add multiple items so budget pressure is visible
    for i in range(20):
        mgr.ingest(ContextItem(id=f"u{i}", kind=ItemKind.user_turn, text=f"item {i} " * 20))

    pack = mgr.build_call_prompt_sync(
        tool_id="db_read",
        query="read data",
        catalog=catalog,
        budget_tokens=200,
    )
    assert pack.phase == Phase.call
    # Schema header should consume part of the budget
    assert pack.stats.header_footer_tokens > 0


def test_build_call_prompt_side_effects_flag() -> None:
    """Tools with side_effects=True include a side-effects notice."""
    catalog = _make_catalog()
    mgr = ContextManager()
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="send email"))

    pack = mgr.build_call_prompt_sync(
        tool_id="send_email",
        query="send email",
        catalog=catalog,
    )
    assert "Side effects: yes" in pack.prompt


@pytest.mark.asyncio
async def test_build_call_prompt_async() -> None:
    """Async build_call_prompt wrapper works."""
    catalog = _make_catalog()
    mgr = ContextManager()
    mgr.ingest(ContextItem(id="u1", kind=ItemKind.user_turn, text="read data"))

    pack = await mgr.build_call_prompt(
        tool_id="db_read",
        query="read data",
        catalog=catalog,
    )
    assert isinstance(pack, ContextPack)
    assert pack.phase == Phase.call
    assert "[TOOL SCHEMA]" in pack.prompt


# ---------------------------------------------------------------------------
# ingest_mcp_result
# ---------------------------------------------------------------------------


def test_ingest_mcp_result_text_only() -> None:
    """Text-only MCP result: no artifacts stored."""
    mgr = ContextManager()
    mcp_result = {
        "content": [{"type": "text", "text": "status: ok\ncount: 42"}],
    }
    item, env = mgr.ingest_mcp_result("call-1", mcp_result, "search_tool")
    assert item.kind == ItemKind.tool_result
    assert item.parent_id == "call-1"
    assert env.status == "ok"
    assert "42" in env.summary
    assert mgr.event_log.count() == 1
    # No binary artifacts → nothing in store
    assert mgr.artifact_store.list_refs() == []


def test_ingest_mcp_result_image() -> None:
    """Image content is persisted in the artifact store."""
    import base64

    mgr = ContextManager()
    png_bytes = b"\x89PNG_test_image_data"
    b64_data = base64.b64encode(png_bytes).decode()
    mcp_result = {
        "content": [
            {"type": "image", "data": b64_data, "mimeType": "image/png"},
        ],
    }
    item, env = mgr.ingest_mcp_result("call-2", mcp_result, "screenshot")
    assert len(env.artifacts) == 1
    assert env.artifacts[0].media_type == "image/png"
    # Artifact is actually persisted
    handle = "mcp:screenshot:image:0"
    assert mgr.artifact_store.exists(handle)
    assert mgr.artifact_store.get(handle) == png_bytes


def test_ingest_mcp_result_resource() -> None:
    """Resource content is persisted in the artifact store."""
    mgr = ContextManager()
    mcp_result = {
        "content": [
            {
                "type": "resource",
                "resource": {
                    "uri": "file:///data.csv",
                    "mimeType": "text/csv",
                    "text": "a,b\n1,2",
                },
            }
        ],
    }
    item, env = mgr.ingest_mcp_result("call-3", mcp_result, "read_file")
    assert len(env.artifacts) == 1
    handle = "mcp:read_file:resource:0"
    assert mgr.artifact_store.exists(handle)
    assert mgr.artifact_store.get(handle) == b"a,b\n1,2"
    assert "a,b" in env.summary


def test_ingest_mcp_result_error() -> None:
    """Error MCP results set status='error'."""
    mgr = ContextManager()
    mcp_result = {
        "content": [{"type": "text", "text": "something went wrong"}],
        "isError": True,
    }
    item, env = mgr.ingest_mcp_result("call-4", mcp_result, "fail_tool")
    assert env.status == "error"
    assert mgr.event_log.count() == 1


def test_ingest_mcp_result_large_output_firewall() -> None:
    """Large text output triggers the context firewall; full raw text is preserved."""
    mgr = ContextManager()
    large_text = "row: " + "x" * 3000
    mcp_result = {
        "content": [{"type": "text", "text": large_text}],
    }
    item, env = mgr.ingest_mcp_result("call-5", mcp_result, "big_tool", firewall_threshold=100)
    # Firewall should have truncated/summarized the item text
    assert len(item.text) < len(large_text)
    assert item.artifact_ref is not None
    assert mgr.event_log.count() == 1
    # Full raw text (not the truncated 500-char summary) must be in the artifact store
    stored = mgr.artifact_store.get(item.artifact_ref.handle)
    assert len(stored) >= len(large_text.encode("utf-8"))


def test_ingest_mcp_result_sync() -> None:
    """Sync alias works."""
    mgr = ContextManager()
    mcp_result = {
        "content": [{"type": "text", "text": "result: 42"}],
    }
    item, env = mgr.ingest_mcp_result_sync("call-6", mcp_result, "tool")
    assert env.status == "ok"
    assert mgr.event_log.count() == 1


def test_ingest_mcp_result_mixed_content() -> None:
    """Mixed content (text + image + resource) persists all artifacts."""
    import base64

    mgr = ContextManager()
    img_bytes = b"\x89PNG_mix"
    mcp_result = {
        "content": [
            {"type": "text", "text": "Found 5 results"},
            {
                "type": "image",
                "data": base64.b64encode(img_bytes).decode(),
                "mimeType": "image/png",
            },
            {
                "type": "resource",
                "resource": {
                    "uri": "file:///report.txt",
                    "mimeType": "text/plain",
                    "text": "Report content here",
                },
            },
        ],
    }
    item, env = mgr.ingest_mcp_result("call-7", mcp_result, "multi_tool")
    assert len(env.artifacts) == 2
    assert mgr.artifact_store.exists("mcp:multi_tool:image:1")
    assert mgr.artifact_store.exists("mcp:multi_tool:resource:2")
    assert mgr.artifact_store.get("mcp:multi_tool:image:1") == img_bytes
    assert "Found 5 results" in env.summary
    assert "Report content" in env.summary


# ---------------------------------------------------------------------------
# Drilldown
# ---------------------------------------------------------------------------


def test_drilldown_basic() -> None:
    """drilldown() returns a slice of a stored artifact."""
    mgr = ContextManager()
    large_output = json.dumps({"users": [{"id": i, "name": f"User {i}"} for i in range(100)]})
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-dd1",
        raw_output=large_output,
        tool_name="search_users",
        media_type="application/json",
        firewall_threshold=100,
    )
    assert len(env.artifacts) >= 1
    handle = env.artifacts[0].handle
    result = mgr.drilldown(handle, {"type": "head", "chars": 50})
    assert len(result) <= 50
    assert result == large_output[:50]


def test_drilldown_json_keys() -> None:
    """drilldown() with json_keys selector returns filtered JSON."""
    mgr = ContextManager()
    data = json.dumps({"name": "Alice", "age": 30, "role": "admin"})
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-dd2",
        raw_output=data,
        tool_name="get_user",
        media_type="application/json",
        firewall_threshold=10,
    )
    handle = env.artifacts[0].handle
    result = mgr.drilldown(handle, {"type": "json_keys", "keys": ["name"]})
    parsed = json.loads(result)
    assert parsed == {"name": "Alice"}


def test_drilldown_inject_into_event_log() -> None:
    """drilldown(inject=True) appends the result as a new ContextItem."""
    mgr = ContextManager()
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-dd3",
        raw_output="line0\nline1\nline2\nline3\nline4",
        tool_name="list_files",
        firewall_threshold=10,
    )
    handle = env.artifacts[0].handle
    initial_count = mgr.event_log.count()
    result = mgr.drilldown(
        handle,
        {"type": "lines", "start": 1, "end": 3},
        inject=True,
        parent_id=_item.id,
    )
    assert result == "line1\nline2"
    assert mgr.event_log.count() == initial_count + 1
    injected = mgr.event_log.get(f"drilldown:{handle}:lines:{initial_count}")
    assert injected.text == "line1\nline2"
    assert injected.parent_id == _item.id


def test_drilldown_inject_repeated_same_selector() -> None:
    """Repeated drilldown(inject=True) with the same selector must not crash."""
    mgr = ContextManager()
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-dd-rep",
        raw_output="line0\nline1\nline2\nline3\nline4",
        tool_name="list_files",
        firewall_threshold=10,
    )
    handle = env.artifacts[0].handle
    selector = {"type": "lines", "start": 0, "end": 2}
    count_before = mgr.event_log.count()
    r1 = mgr.drilldown(handle, selector, inject=True)
    r2 = mgr.drilldown(handle, selector, inject=True)
    assert r1 == r2 == "line0\nline1"
    assert mgr.event_log.count() == count_before + 2


def test_drilldown_without_inject() -> None:
    """drilldown(inject=False) does not modify event log."""
    mgr = ContextManager()
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-dd4",
        raw_output="hello world",
        tool_name="echo",
        firewall_threshold=5,
    )
    handle = env.artifacts[0].handle
    count_before = mgr.event_log.count()
    mgr.drilldown(handle, {"type": "head", "chars": 5})
    assert mgr.event_log.count() == count_before


def test_drilldown_sync() -> None:
    """drilldown_sync() is a synchronous alias."""
    mgr = ContextManager()
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-dd5",
        raw_output="sync test data",
        tool_name="echo",
        firewall_threshold=5,
    )
    handle = env.artifacts[0].handle
    result = mgr.drilldown_sync(handle, {"type": "head", "chars": 4})
    assert result == "sync"


def test_drilldown_missing_handle_raises() -> None:
    """drilldown() raises ArtifactNotFoundError for unknown handles."""
    from contextweaver.exceptions import ArtifactNotFoundError

    mgr = ContextManager()
    with pytest.raises(ArtifactNotFoundError):
        mgr.drilldown("no-such-handle", {"type": "head", "chars": 10})


def test_drilldown_unknown_selector_raises() -> None:
    """drilldown() raises ContextWeaverError for unknown selector types."""
    mgr = ContextManager()
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-dd6",
        raw_output="some data",
        tool_name="echo",
        firewall_threshold=5,
    )
    handle = env.artifacts[0].handle
    with pytest.raises(ContextWeaverError, match="Unknown drilldown"):
        mgr.drilldown(handle, {"type": "unknown_type"})


# ---------------------------------------------------------------------------
# Drilldown redaction enforcement (issue #451)
# ---------------------------------------------------------------------------


def _seed_sensitive_artifact(
    mgr: ContextManager, handle: str, content: str, sensitivity: Sensitivity
) -> None:
    """Store *content* under *handle* with a source event-log item at *sensitivity*."""
    ref = mgr.artifact_store.put(handle, content.encode("utf-8"), media_type="text/plain")
    mgr.event_log.append(
        ContextItem(
            id=f"src:{handle}",
            kind=ItemKind.tool_result,
            text="summary",
            artifact_ref=ref,
            sensitivity=sensitivity,
        )
    )


def test_drilldown_denied_for_confidential_source_by_default() -> None:
    """drilldown() on a source item that meets the floor is denied (closed default)."""
    mgr = ContextManager()
    _seed_sensitive_artifact(
        mgr, "artifact:secret-1", "TOP SECRET payload", Sensitivity.confidential
    )
    with pytest.raises(PolicyViolationError, match="denied"):
        mgr.drilldown("artifact:secret-1", {"type": "head", "chars": 10})


def test_drilldown_allowed_for_public_source() -> None:
    """A non-sensitive source is never over-blocked."""
    mgr = ContextManager()
    _seed_sensitive_artifact(mgr, "artifact:ok-1", "plain text", Sensitivity.public)
    assert mgr.drilldown("artifact:ok-1", {"type": "head", "chars": 5}) == "plain"


def test_drilldown_allowed_for_confidential_source_with_optout() -> None:
    """The documented opt-out re-enables recovery of filtered content."""
    mgr = ContextManager(policy=ContextPolicy(allow_redacted_drilldown=True))
    _seed_sensitive_artifact(mgr, "artifact:secret-2", "TOP SECRET", Sensitivity.confidential)
    assert mgr.drilldown("artifact:secret-2", {"type": "head", "chars": 3}) == "TOP"


def test_drilldown_inject_inherits_source_sensitivity() -> None:
    """An injected slice inherits the source label, not the default public."""
    mgr = ContextManager(policy=ContextPolicy(allow_redacted_drilldown=True))
    _seed_sensitive_artifact(mgr, "artifact:secret-3", "alpha\nbeta\ngamma", Sensitivity.restricted)
    count = mgr.event_log.count()
    mgr.drilldown("artifact:secret-3", {"type": "lines", "start": 0, "end": 1}, inject=True)
    injected = mgr.event_log.get(f"drilldown:artifact:secret-3:lines:{count}")
    assert injected.sensitivity == Sensitivity.restricted


def test_drilldown_unknown_handle_not_over_blocked() -> None:
    """A handle with no matching source item is treated as public (injects as public)."""
    mgr = ContextManager()
    ref = mgr.artifact_store.put("artifact:orphan", b"orphan body", media_type="text/plain")
    assert ref.handle == "artifact:orphan"
    count = mgr.event_log.count()
    mgr.drilldown("artifact:orphan", {"type": "head", "chars": 6}, inject=True)
    injected = mgr.event_log.get(f"drilldown:artifact:orphan:head:{count}")
    assert injected.sensitivity == Sensitivity.public


# ---------------------------------------------------------------------------
# Classifier audit metadata (issue #542)
# ---------------------------------------------------------------------------


class _AlwaysInternal:
    """Test classifier that always votes ``internal``."""

    def classify(self, item: ContextItem) -> Sensitivity:
        _ = item
        return Sensitivity.internal


def test_classify_items_records_audit_metadata() -> None:
    """A raised label records metadata["sensitivity_raised_by"] (issue #542 AC)."""
    from contextweaver.context.build import _classify_items

    items = [
        ContextItem(id="a", kind=ItemKind.tool_result, text="x", sensitivity=Sensitivity.public)
    ]
    out = _classify_items(items, _AlwaysInternal())
    assert out[0].sensitivity == Sensitivity.internal
    assert out[0].metadata["sensitivity_raised_by"] == "_AlwaysInternal"


def test_classify_items_no_metadata_when_label_not_raised() -> None:
    """No audit metadata is written when the classifier does not raise the label."""
    from contextweaver.context.build import _classify_items

    items = [
        ContextItem(id="a", kind=ItemKind.tool_result, text="x", sensitivity=Sensitivity.restricted)
    ]
    out = _classify_items(items, _AlwaysInternal())
    assert out[0].sensitivity == Sensitivity.restricted
    assert "sensitivity_raised_by" not in out[0].metadata


# ---------------------------------------------------------------------------
# Auto-generated views on ingest
# ---------------------------------------------------------------------------


def test_ingest_tool_result_auto_views_json_large() -> None:
    """Large JSON tool results have auto-generated views after firewall."""
    mgr = ContextManager()
    data = json.dumps({"users": [1, 2, 3], "count": 3})
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-av1",
        raw_output=data,
        tool_name="api_call",
        media_type="application/json",
        firewall_threshold=10,
    )
    assert len(env.views) > 0
    view_ids = [v.view_id for v in env.views]
    assert any("json_keys" in vid for vid in view_ids)


def test_ingest_tool_result_auto_views_small_output() -> None:
    """Small outputs also get auto-generated views and artifact refs."""
    mgr = ContextManager()
    data = json.dumps({"status": "ok"})
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-av2",
        raw_output=data,
        tool_name="ping",
        media_type="application/json",
        firewall_threshold=5000,
    )
    assert len(env.views) > 0
    assert len(env.artifacts) > 0
    assert _item.artifact_ref is not None


def test_ingest_tool_result_views_drilldown_chain() -> None:
    """Views from ingest can be used to drive drilldown calls."""
    mgr = ContextManager()
    data = json.dumps({"alpha": "aaa", "beta": "bbb", "gamma": "ccc"})
    _item, env = mgr.ingest_tool_result(
        tool_call_id="tc-chain",
        raw_output=data,
        tool_name="get_data",
        media_type="application/json",
        firewall_threshold=10,
    )
    # Pick a view and use its selector to drilldown
    key_views = [v for v in env.views if v.selector.get("type") == "json_keys"]
    assert len(key_views) > 0
    view = key_views[0]
    handle = env.artifacts[0].handle
    result = mgr.drilldown(handle, view.selector)
    assert len(result) > 0
    parsed = json.loads(result)
    assert isinstance(parsed, dict)


def test_view_registry_accessible() -> None:
    """ContextManager exposes view_registry property."""
    mgr = ContextManager()
    from contextweaver.context.views import ViewRegistry

    assert isinstance(mgr.view_registry, ViewRegistry)


# ---------------------------------------------------------------------------
# Issue #45 — profile / mode wiring
# ---------------------------------------------------------------------------


def test_default_mode_is_strict() -> None:
    from contextweaver.profiles import Mode

    mgr = ContextManager()
    assert mgr.mode == Mode.strict
    assert mgr.profile is None


def test_profile_seeds_budget_policy_scoring() -> None:
    from contextweaver.profiles import Mode, ProfileConfig

    profile = ProfileConfig.from_preset("fast")
    mgr = ContextManager(profile=profile)
    # ``fast`` preset sets answer budget to 3000.
    assert mgr._budget.answer == 3000
    assert mgr.profile is profile
    assert mgr.mode == Mode.strict  # preset doesn't change mode


def test_profile_explicit_kwargs_take_priority() -> None:
    from contextweaver.config import ContextBudget
    from contextweaver.profiles import ProfileConfig

    profile = ProfileConfig.from_preset("accurate")  # answer=8000
    custom_budget = ContextBudget(answer=1234)
    mgr = ContextManager(budget=custom_budget, profile=profile)
    assert mgr._budget.answer == 1234


def test_profile_with_seeded_mode() -> None:
    from contextweaver.profiles import Mode, ProfileConfig

    profile = ProfileConfig(mode=Mode.seeded, seed=42)
    mgr = ContextManager(profile=profile)
    assert mgr.mode == Mode.seeded


# ------------------------------------------------------------------
# Header memory routed through sensitivity + phase policy (issue #450)
# ------------------------------------------------------------------


def test_header_facts_dropped_by_sensitivity_floor() -> None:
    mgr = ContextManager()  # default floor confidential, action drop
    mgr.add_fact("public_key", "public_value")
    mgr.add_fact("secret_key", "secret_value", sensitivity=Sensitivity.restricted)
    prompt = mgr.build_sync(phase=Phase.answer).prompt
    assert "public_value" in prompt
    assert "secret_value" not in prompt


def test_header_facts_redacted_when_action_redact() -> None:
    policy = ContextPolicy(sensitivity_floor=Sensitivity.confidential, sensitivity_action="redact")
    mgr = ContextManager(policy=policy)
    mgr.add_fact("secret_key", "secret_value", sensitivity=Sensitivity.confidential)
    prompt = mgr.build_sync(phase=Phase.answer).prompt
    assert "secret_value" not in prompt
    assert "[REDACTED: confidential]" in prompt


def test_header_facts_excluded_in_phase_without_memory_fact() -> None:
    """A phase that excludes memory_fact must not receive fact text via the
    header side-channel (issue #450)."""
    mgr = ContextManager()
    mgr.add_fact("k", "v")
    assert "[FACTS]" in mgr.build_sync(phase=Phase.answer).prompt  # answer allows memory_fact
    assert "[FACTS]" not in mgr.build_sync(phase=Phase.call).prompt  # call does not


def test_header_episode_dropped_by_sensitivity() -> None:
    mgr = ContextManager()
    mgr.add_episode("e1", "benign summary")
    mgr.add_episode("e2", "leaked secret summary", sensitivity=Sensitivity.restricted)
    prompt = mgr.build_sync(phase=Phase.answer).prompt
    assert "benign summary" in prompt
    assert "leaked secret summary" not in prompt


# ------------------------------------------------------------------
# Ingestion sensitivity classification (issue #542)
# ------------------------------------------------------------------


def _secret_log() -> InMemoryEventLog:
    log = InMemoryEventLog()
    log.append(
        ContextItem(id="t1", kind=ItemKind.tool_result, text="leak key=AKIAIOSFODNN7EXAMPLE")
    )
    return log


def test_unlabeled_secret_item_survives_without_classifier() -> None:
    prompt = ContextManager(event_log=_secret_log()).build_sync(phase=Phase.answer).prompt
    assert "AKIAIOSFODNN7EXAMPLE" in prompt


def test_classifier_raises_and_drops_unlabeled_secret_item() -> None:
    mgr = ContextManager(
        event_log=_secret_log(), sensitivity_classifier=HeuristicSensitivityClassifier()
    )
    prompt = mgr.build_sync(phase=Phase.answer).prompt
    assert "AKIAIOSFODNN7EXAMPLE" not in prompt


def test_classifier_applies_to_header_facts() -> None:
    mgr = ContextManager(sensitivity_classifier=HeuristicSensitivityClassifier())
    mgr.add_fact("creds", "key=AKIAIOSFODNN7EXAMPLE")
    prompt = mgr.build_sync(phase=Phase.answer).prompt
    assert "AKIAIOSFODNN7EXAMPLE" not in prompt


# ------------------------------------------------------------------
# Manager-level firewall secret scrubbing (issue #428)
# ------------------------------------------------------------------


def test_manager_redact_secrets_scrubs_built_prompt() -> None:
    log = InMemoryEventLog()
    log.append(
        ContextItem(id="t1", kind=ItemKind.tool_result, text="config key=AKIAIOSFODNN7EXAMPLE")
    )
    mgr = ContextManager(event_log=log, redact_secrets=True)
    prompt = mgr.build_sync(phase=Phase.answer).prompt
    assert "AKIAIOSFODNN7EXAMPLE" not in prompt
