tests/test_context_packer.py

"""M13 — Context Packer + Deictic Resolver. Red/green tests from the M13 success criteria.

Success criteria under test:
- The produced packet is ≤1500 tokens via the ``len(json.dumps(packet)) / 4.5`` char proxy.
- The packet includes the selection plus visible nodes plus fog candidates with summaries,
  and every fog candidate is drawn from ``scene.fogged_node_ids``.
- Deictic terms ("this", "it") resolve to the selection id, are recorded in
  ``resolved_deixis``, and the prepended ``(Selected: ...)`` anchor appears.
- An oversized graph triggers the truncation cascade (and the extreme case raises
  ``ContextBudgetExceededError``).
- Serialized edges use the property-graph alias ``type`` for ``Edge.kind``.
"""

from __future__ import annotations

import json

import pytest
from loosecanvas.context_packer import (
    ContextBudgetExceededError,
    ContextPacker,
    ContextPacket,
)
from loosecanvas.contracts import Edge, MapEvent, Node, SceneState
from loosecanvas.graph_repository import LooseGraphRepository

# ── Fixtures ──────────────────────────────────────────────────────────────────


def _node(node_id: str, *, label: str | None = None, summary: str = "") -> Node:
    return Node(id=node_id, kind="concept", label=label or node_id, summary=summary)


def _edge(edge_id: str, source: str, target: str) -> Edge:
    return Edge(id=edge_id, source=source, target=target, kind="relates_to")


def _proxy_tokens(packet: ContextPacket) -> float:
    return len(json.dumps(packet.model_dump())) / 4.5


def _small_repo() -> LooseGraphRepository:
    repo = LooseGraphRepository()
    repo.upsert_nodes(
        [
            _node("a", label="Alpha", summary="the alpha node summary"),
            _node("b", label="Beta", summary="the beta node summary"),
            _node("c", label="Gamma", summary="the gamma node summary"),
            _node("d", label="Delta", summary="a fogged delta candidate summary"),
            _node("e", label="Epsilon", summary="a fogged epsilon candidate summary"),
        ]
    )
    repo.upsert_edges([_edge("e_ab", "a", "b"), _edge("e_ad", "a", "d")])
    return repo


def _scene() -> SceneState:
    return SceneState(
        scene_id="s1",
        visible_node_ids=["a", "b", "c"],
        visible_edge_ids=["e_ab"],
        fogged_node_ids=["d", "e"],
    )


# ── Budget & content ──────────────────────────────────────────────────────────


def test_simple_packet_within_budget_and_has_core_fields() -> None:
    packer = ContextPacker(_small_repo())
    packet = packer.pack(
        _scene(), selection_id="a", message="tell me about Alpha", recent_events=[]
    )

    assert _proxy_tokens(packet) <= 1500
    assert packet.selection["id"] == "a"
    assert packet.selection["label"] == "Alpha"
    assert {n["id"] for n in packet.visible_nodes} == {"a", "b", "c"}
    assert packet.user_message  # never empty when a message is supplied


def test_fog_candidates_only_from_fog_set_and_carry_summaries() -> None:
    scene = _scene()
    packer = ContextPacker(_small_repo())
    packet = packer.pack(scene, selection_id="a", message="overview", recent_events=[])

    fog_ids = {c["id"] for c in packet.fog_candidates}
    assert fog_ids  # fog candidates are produced
    assert fog_ids <= set(scene.fogged_node_ids)
    for cand in packet.fog_candidates:
        assert "summary" in cand
        assert len(cand["summary"]) <= 80


def test_fog_candidates_sorted_by_proximity_to_selection() -> None:
    # ``d`` is a 1-hop neighbour of selection ``a`` (edge e_ad); ``e`` is not.
    scene = _scene()
    packer = ContextPacker(_small_repo())
    packet = packer.pack(scene, selection_id="a", message="overview", recent_events=[])

    assert packet.fog_candidates[0]["id"] == "d"


def test_visible_edges_use_type_alias_for_edge_kind() -> None:
    packer = ContextPacker(_small_repo())
    packet = packer.pack(
        _scene(), selection_id="a", message="overview", recent_events=[]
    )

    assert packet.visible_edges
    edge = packet.visible_edges[0]
    assert edge["type"] == "relates_to"
    assert "kind" not in edge


# ── Deictic resolution ────────────────────────────────────────────────────────


def test_deixis_this_it_resolve_to_selection_and_prepend_anchor() -> None:
    packer = ContextPacker(_small_repo())
    packet = packer.pack(
        _scene(), selection_id="a", message="what is this?", recent_events=[]
    )

    assert packet.resolved_deixis == {"this": "a", "it": "a"}
    assert packet.user_message.startswith("(Selected: Alpha, id=a.)")
    assert "what is this?" in packet.user_message


def test_deixis_implicit_anchor_on_expansion_without_selection() -> None:
    packer = ContextPacker(_small_repo())
    packet = packer.pack(
        _scene(), selection_id=None, message="show me more", recent_events=[]
    )

    assert "implicit_anchor" in packet.resolved_deixis
    assert packet.resolved_deixis["implicit_anchor"] in {"a", "b", "c"}


def test_deixis_recent_highlight_reference() -> None:
    packer = ContextPacker(_small_repo())
    events = [
        MapEvent(
            actor="curator",
            action="highlight",
            target_id="c",
            detail="highlighted gamma",
        )
    ]
    packet = packer.pack(
        _scene(),
        selection_id=None,
        message="explain the thing you showed",
        recent_events=events,
    )

    assert packet.resolved_deixis == {"that": "c"}
    assert packet.user_message.startswith("(Selected: Gamma, id=c.)")


def test_message_never_truncated_and_no_deixis_when_irrelevant() -> None:
    packer = ContextPacker(_small_repo())
    packet = packer.pack(
        _scene(), selection_id="a", message="overview please", recent_events=[]
    )

    assert packet.resolved_deixis == {}
    assert packet.user_message == "overview please"


# ── Truncation cascade ────────────────────────────────────────────────────────


def _oversized_repo() -> LooseGraphRepository:
    repo = LooseGraphRepository()
    long_label = "L" * 400
    long_summary = "S" * 250
    nodes = [
        _node("v" + str(i) * 50, label=long_label, summary=long_summary)
        for i in range(10)
    ]
    fog = [
        _node("f" + str(i) * 50, label=long_label, summary=long_summary)
        for i in range(5)
    ]
    repo.upsert_nodes(nodes + fog)
    # a couple of fat, non-truncatable visible edges
    repo.upsert_edges([_edge("g" * 150, nodes[0].id, nodes[1].id)])
    return repo


def _oversized_scene(repo: LooseGraphRepository) -> SceneState:
    node_ids = [n.id for n in repo.nodes.values() if n.id.startswith("v")]
    fog_ids = [n.id for n in repo.nodes.values() if n.id.startswith("f")]
    edge_ids = list(repo.edges.keys())
    return SceneState(
        scene_id="big",
        visible_node_ids=node_ids,
        visible_edge_ids=edge_ids,
        fogged_node_ids=fog_ids,
    )


def test_oversized_graph_triggers_truncation_but_stays_within_budget() -> None:
    repo = _oversized_repo()
    scene = _oversized_scene(repo)
    selection_id = scene.visible_node_ids[0]
    packer = ContextPacker(repo)

    packet = packer.pack(
        scene, selection_id=selection_id, message="overview", recent_events=[]
    )

    assert _proxy_tokens(packet) <= 1500
    # truncation actually fired: fewer than the full 10 visible nodes survive.
    assert len(packet.visible_nodes) < 10


def test_extreme_message_raises_budget_exceeded() -> None:
    repo = _small_repo()
    packer = ContextPacker(repo)
    huge_message = "x" * 10000  # never truncated → no cascade step can recover

    with pytest.raises(ContextBudgetExceededError):
        packer.pack(_scene(), selection_id="a", message=huge_message, recent_events=[])


# ── Conversation memory (recent_dialogue / recent_actions) ─────────────────────


def test_recent_dialogue_populates_packet_field() -> None:
    packer = ContextPacker(_small_repo())
    dialogue = [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
    ]
    packet = packer.pack(
        _scene(),
        selection_id="a",
        message="overview",
        recent_events=[],
        recent_dialogue=dialogue,
    )

    assert packet.recent_dialogue == dialogue


def test_recent_dialogue_defaults_to_empty_when_omitted() -> None:
    packer = ContextPacker(_small_repo())
    packet = packer.pack(
        _scene(), selection_id="a", message="overview", recent_events=[]
    )

    assert packet.recent_dialogue == []


def test_recent_actions_override_packet_recent_events() -> None:
    packer = ContextPacker(_small_repo())
    packet = packer.pack(
        _scene(),
        selection_id="a",
        message="overview",
        recent_events=[],
        recent_actions=["revealed X", "centered Y"],
    )

    assert packet.recent_events == ["revealed X", "centered Y"]


def test_recent_dialogue_trimmed_before_visible_nodes_reduced() -> None:
    packer = ContextPacker(_small_repo())
    # An oversized dialogue list pushes the packet over budget; the memory
    # trimming steps must fire BEFORE the visible-node reduction steps.
    oversized_dialogue = [{"role": "user", "content": "Q" * 200} for _ in range(40)]

    packet = packer.pack(
        _scene(),
        selection_id="a",
        message="overview",
        recent_events=[],
        recent_dialogue=oversized_dialogue,
    )

    assert _proxy_tokens(packet) <= 1500
    # dialogue shrank...
    assert len(packet.recent_dialogue) < len(oversized_dialogue)
    # ...while all visible nodes were preserved (visible reduction never reached).
    assert {n["id"] for n in packet.visible_nodes} == {"a", "b", "c"}