tests/test_agent_tools.py

"""Tests for agent_tools.py — per-turn ContextVar accumulation + tool bridge.

RED discipline: this file is written before agent_tools.py exists.
Imports will fail with ModuleNotFoundError until the implementation lands.
"""

from __future__ import annotations

import json
from collections.abc import Iterator
from pathlib import Path

import pytest
from loosecanvas.agent_tools import (
    TOOLS,
    AgentTurnContext,
    _current_turn,
    create_edge,
    create_node,
    finalize_agent_turn,
    get_canvas_state,
    get_turn_context,
    remove_node,
    reveal_node,
    set_turn_context,
    update_edge,
)
from loosecanvas.turn_logic import (
    TurnResult,
    load_fixture_session,
    session_locks,
    session_store,
)

_REPO_ROOT = Path(__file__).resolve().parents[1]
_SMALL_GRAPH = _REPO_ROOT / "fixtures" / "small_graph.json"


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


@pytest.fixture(autouse=True)
def _isolate_sessions() -> Iterator[None]:
    session_store.clear()
    session_locks.clear()
    yield
    session_store.clear()
    session_locks.clear()


@pytest.fixture(autouse=True)
def _reset_agent_context() -> Iterator[None]:
    _current_turn.set(None)
    yield
    _current_turn.set(None)


@pytest.fixture
async def live_session_id() -> str:
    sid, _ = await load_fixture_session(fixture_path=_SMALL_GRAPH)
    return sid


# ── Lifecycle tests ───────────────────────────────────────────────────────────


def test_set_get_turn_context_lifecycle(live_session_id: str) -> None:
    ctx = set_turn_context(live_session_id)
    assert isinstance(ctx, AgentTurnContext)
    assert ctx.session_id == live_session_id
    assert ctx.accumulated_actions == []
    assert ctx.activity == []
    assert get_turn_context() is ctx


def test_get_turn_context_raises_when_unset() -> None:
    # _reset_agent_context ensures no context is active
    with pytest.raises(RuntimeError, match="no active agent turn context"):
        get_turn_context()


# ── Tool tests ────────────────────────────────────────────────────────────────


async def test_reveal_node_accept_path(live_session_id: str) -> None:
    set_turn_context(live_session_id)
    # "learning_rate" is fogged in small_graph initial scene
    result = reveal_node.invoke({"node_id": "learning_rate"})
    data = json.loads(result)
    assert data["status"] == "accepted"
    ctx = get_turn_context()
    assert len(ctx.accumulated_actions) == 1
    assert ctx.accumulated_actions[0].type.value == "reveal"
    assert ctx.accumulated_actions[0].target_id == "learning_rate"
    assert len(ctx.activity) == 1


async def test_get_canvas_state_includes_visible_edges(live_session_id: str) -> None:
    set_turn_context(live_session_id)
    state = json.loads(get_canvas_state.invoke({}))
    assert "visible_edges" in state, "agent must be able to see edges to relabel them"
    for edge in state["visible_edges"]:
        assert set(edge) >= {"id", "source", "target", "label"}
        assert edge["id"] in state["visible_edge_ids"]


async def test_update_edge_accumulates_for_visible_edge(live_session_id: str) -> None:
    set_turn_context(live_session_id)
    state = json.loads(get_canvas_state.invoke({}))
    if not state["visible_edges"]:
        pytest.skip("fixture has no visible edge to relabel")
    edge_id = state["visible_edges"][0]["id"]
    result = json.loads(
        update_edge.invoke({"edge_id": edge_id, "label": "clearer link"})
    )
    assert result["status"] == "accepted"
    ctx = get_turn_context()
    assert any(
        a.type.value == "update_edge"
        and a.edge_id == edge_id
        and a.label == "clearer link"
        for a in ctx.accumulated_actions
    )


async def test_reveal_node_reject_path(live_session_id: str) -> None:
    set_turn_context(live_session_id)
    # Completely nonexistent node id — validator gives id_not_found
    result = reveal_node.invoke({"node_id": "nonexistent_xyz_node"})
    data = json.loads(result)
    assert data["status"] == "rejected"
    assert "hint" in data
    # Hint should list valid fogged ids so the model can self-correct
    hint = data["hint"]
    assert any(
        nid in hint for nid in ("learning_rate", "overfitting", "regularization")
    )
    # Nothing accumulated
    ctx = get_turn_context()
    assert len(ctx.accumulated_actions) == 0


async def test_get_canvas_state_no_mutation(live_session_id: str) -> None:
    set_turn_context(live_session_id)
    result = get_canvas_state.invoke({})
    data = json.loads(result)
    assert "visible_nodes" in data
    assert "fogged_count" in data
    assert "visible_edge_ids" in data
    visible_ids = {n["id"] for n in data["visible_nodes"]}
    assert "gradient_descent" in visible_ids
    assert data["fogged_count"] == 3  # learning_rate, overfitting, regularization
    # READ-ONLY: no accumulation
    ctx = get_turn_context()
    assert len(ctx.accumulated_actions) == 0


def test_finalize_agent_turn_no_actions_returns_none(live_session_id: str) -> None:
    set_turn_context(live_session_id)
    # No tool calls — accumulated_actions empty → pure-speech turn
    result = finalize_agent_turn(live_session_id, {}, "Nothing to do on canvas.")
    assert result is None


async def test_finalize_agent_turn_with_reveal_returns_turn_result(
    live_session_id: str,
) -> None:
    set_turn_context(live_session_id)
    reveal_result = reveal_node.invoke({"node_id": "learning_rate"})
    assert json.loads(reveal_result)["status"] == "accepted"
    turn_result = finalize_agent_turn(live_session_id, {}, "Revealed learning rate.")
    assert turn_result is not None
    assert isinstance(turn_result, TurnResult)
    assert turn_result.renderer_patch is not None
    assert turn_result.status in ("success", "partial_success")


async def test_create_edge_resolves_same_turn_created_node(
    live_session_id: str,
) -> None:
    """A create_edge to a node create_node'd EARLIER THIS TURN must be accepted.

    The per-tool gate validates against the turn-so-far (accumulated actions), not just
    the persisted graph/scene — patches only apply at finalize. Without this, the model
    is told its same-turn edge was rejected and stalls (the "latency of truth" bug).
    """
    set_turn_context(live_session_id)
    node_result = create_node.invoke(
        {"node_id": "concept::new", "label": "New Concept"}
    )
    assert json.loads(node_result)["status"] == "accepted"
    edge_result = create_edge.invoke(
        {
            "source_id": "concept::new",
            "target_id": "gradient_descent",
            "label": "relates to",
        }
    )
    assert json.loads(edge_result)["status"] == "accepted"
    ctx = get_turn_context()
    assert len(ctx.accumulated_actions) == 2
    assert [a.type.value for a in ctx.accumulated_actions] == [
        "create_node",
        "create_edge",
    ]


async def test_duplicate_create_node_rejected_same_turn(
    live_session_id: str,
) -> None:
    """A repeated create_node for an id minted earlier this turn must be rejected.

    Without turn-so-far validation the node isn't in the persisted graph yet, so a retry
    is accepted twice — producing the duplicate canvas mutations / receipts seen live.
    """
    set_turn_context(live_session_id)
    first = create_node.invoke({"node_id": "concept::dup", "label": "Dup"})
    assert json.loads(first)["status"] == "accepted"
    second = create_node.invoke({"node_id": "concept::dup", "label": "Dup"})
    data = json.loads(second)
    assert data["status"] == "rejected"
    assert "id_exists" in data["reason"]
    ctx = get_turn_context()
    assert len(ctx.accumulated_actions) == 1


async def test_remove_node_accept_path(live_session_id: str) -> None:
    set_turn_context(live_session_id)
    # "gradient_descent" is visible; 5 total visible — removing one leaves 4 (> 0)
    result = remove_node.invoke({"node_id": "gradient_descent"})
    data = json.loads(result)
    assert data["status"] == "accepted"
    ctx = get_turn_context()
    assert len(ctx.accumulated_actions) == 1
    assert ctx.accumulated_actions[0].type.value == "remove_node"


def test_tools_list_populated() -> None:
    assert (
        len(TOOLS) >= 14
    )  # at least one tool per SceneActionType plus get_canvas_state