tests/test_full_turn_loop.py

"""M22 — Full turn-loop integration proofs (offline, always-run).

The live-server acceptance harness lives in ``tests/test_turn_loop_integration.py`` (gated
by ``LOOSECANVAS_RUN_LLM_INTEGRATION``). This module holds M22's **offline** automatable
proofs so they run in CI without a model:

* item 12 — one-directional import lint (core must not import extractors).
* items 7/10/11 — the SAME turn loop, curator, validator, reducer and adapter succeed on
  BOTH the M04a wiki fixture and an M04b text-extracted graph, with no "is this a repo"
  branch. The model is mocked via ``httpx.MockTransport``.
* item 13 — only the evidence service may transition ``support_state`` to
  ``source_supported``.
* item 1 — the M23 repo leg (skipped unless ``LOOSECANVAS_TEST_REPO`` is set).
"""

from __future__ import annotations

import importlib.util
import json
import os
from collections.abc import Iterator
from pathlib import Path
from types import ModuleType
from typing import Any

import httpx
import pytest
from loosecanvas.claim_state_machine import InvalidTransitionError, transition_support
from loosecanvas.config import ALLOWED_BASE_DIR
from loosecanvas.contracts import (
    Claim,
    EvidenceRef,
    Node,
    SourceFile,
    SourceSnapshot,
)
from loosecanvas.evidence_service import EvidenceService, compute_snippet_hash
from loosecanvas.extraction_pipeline import ExtractionOrchestrator, ExtractionResult
from loosecanvas.extractors.text_graph_adapter import extract_graph_from_text
from loosecanvas.graph_repository import LooseGraphRepository
from loosecanvas.llm_client import LLMClient
from loosecanvas.scene_curator import SceneCurator
from loosecanvas.sceneplan_generator import ScenePlanGenerator
from loosecanvas.turn_logic import (
    TurnResult,
    load_fixture_session,
    run_turn,
    session_locks,
    session_store,
)

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


# ── Fixtures / helpers ─────────────────────────────────────────────────────────


@pytest.fixture(autouse=True)
def _isolate_sessions() -> Iterator[None]:
    """Keep the module-level session store from leaking between tests."""
    session_store.clear()
    session_locks.clear()
    yield
    session_store.clear()
    session_locks.clear()


def _mock_client(content: str) -> LLMClient:
    """An ``LLMClient`` whose every completion returns ``content`` verbatim."""

    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200, json={"choices": [{"message": {"content": content}}]}
        )

    return LLMClient(base_url="http://test", transport=httpx.MockTransport(handler))


def _sceneplan_json(target_id: str) -> str:
    """A schema-valid ``ScenePlan`` targeting a real visible node (passes M14)."""
    return json.dumps(
        {
            "actions": [
                {
                    "type": "center_on",
                    "target_id": target_id,
                    "reasoning": "Focus the user's question on the selected concept.",
                },
                {
                    "type": "highlight",
                    "target_id": target_id,
                    "reasoning": "Mark the node of interest so it stands out.",
                },
            ],
            "summary": "Centering on the selected concept so its role is visible.",
        }
    )


def _load_lint_module() -> ModuleType:
    """Import ``scripts/lint_import_dependencies.py`` by path (it is not a package)."""
    path = _REPO_ROOT / "scripts" / "lint_import_dependencies.py"
    spec = importlib.util.spec_from_file_location("lint_import_dependencies", path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _assert_successful_turn(result: TurnResult) -> None:
    """Shared post-conditions for a substrate-agnostic happy-path turn (items 7/10)."""
    assert result.status != "failed", result.assistant_message
    assert result.scene_patch is not None
    patch = result.scene_patch
    # Non-empty ScenePatch: the highlight/center actions must have moved scene state.
    assert patch.set_highlighted_ids or patch.set_camera_target_id is not None
    operations = result.renderer_patch.operations
    assert operations, "renderer patch carried no operations"
    assert len(operations) < _MAX_RENDERER_OPS


# ── item 12: import lint ───────────────────────────────────────────────────────


def test_import_lint_core_does_not_import_extractors() -> None:
    lint = _load_lint_module()
    violations = lint.find_forbidden_imports()
    assert violations == [], "\n".join(violations)


# ── items 7/10/11: substrate-agnostic turn loop ────────────────────────────────


async def test_turn_loop_on_m04a_wiki_fixture() -> None:
    session_id, _snapshot = await load_fixture_session(fixture_path=_SMALL_GRAPH)
    record = session_store[session_id]
    target = record.scene.visible_node_ids[0]
    generator = ScenePlanGenerator(_mock_client(_sceneplan_json(target)))

    result = await run_turn(
        session_id,
        {"kind": "node", "id": target},
        "explain this concept's role",
        generator,
    )
    _assert_successful_turn(result)


async def test_turn_loop_on_m04b_text_extracted_graph() -> None:
    # An M04b text-extracted graph must flow through the IDENTICAL turn loop as the
    # repo/wiki leg — no code path may branch on "is this a repo" (guardrail 8).
    article = (
        "# Gradient Descent\n\n"
        "Gradient descent is an optimization method. "
        "A loss function measures prediction error."
    )
    concept_json = json.dumps(
        {
            "concept_nodes": [
                {
                    "id": "concept::gradient_descent",
                    "label": "Gradient Descent",
                    "summary": "An iterative optimization method.",
                    "start_char": 2,
                    "end_char": 18,
                },
                {
                    "id": "concept::loss_function",
                    "label": "Loss Function",
                    "summary": "Measures prediction error.",
                    "start_char": 58,
                    "end_char": 71,
                },
            ],
            "semantic_edges": [
                {
                    "source_id": "concept::gradient_descent",
                    "target_id": "concept::loss_function",
                    "type": "related_to",
                    "label": "minimizes",
                }
            ],
        }
    )
    graph = await extract_graph_from_text(
        article, "Gradient Descent", _mock_client(concept_json)
    )
    assert graph.nodes, "M04b adapter produced no nodes"

    # Same curator (M12) as the repo leg builds the initial scene.
    repo = LooseGraphRepository.from_extraction(
        graph.nodes, graph.edges, graph.claims, SourceSnapshot()
    )
    scene, _pathways = SceneCurator(repo).curate_initial_scene()
    assert scene.visible_node_ids, "curator produced no visible nodes"

    fixture_data: dict[str, Any] = {
        "graph": graph.model_dump(by_alias=True),
        "scene": scene.model_dump(),
    }
    session_id, _snapshot = await load_fixture_session(fixture_data=fixture_data)
    record = session_store[session_id]
    target = record.scene.visible_node_ids[0]
    generator = ScenePlanGenerator(_mock_client(_sceneplan_json(target)))

    result = await run_turn(
        session_id,
        {"kind": "node", "id": target},
        "explain this concept's role",
        generator,
    )
    _assert_successful_turn(result)


# ── item 13: evidence-service actor gate ───────────────────────────────────────


async def test_only_evidence_service_can_corroborate() -> None:
    content = "intro line\nGradient descent minimizes the loss.\ntrailing line"
    snippet = "\n".join(content.splitlines()[1:2])  # line 2, the evidence span
    snapshot = SourceSnapshot(
        snapshot_id="snap1",
        root_path="",
        files={
            "src1": SourceFile(
                source_id="src1",
                path="doc.md",
                content=content,
                line_count=len(content.splitlines()),
            )
        },
    )
    ref = EvidenceRef(
        source_id="src1",
        node_id="n1",
        start_line=2,
        end_line=2,
        snippet_hash=compute_snippet_hash(snippet),
    )
    node = Node(id="n1", kind="concept", label="Gradient Descent", evidence_refs=[ref])
    claim = Claim(
        id="cl1",
        claim_type="note",
        target_id="n1",
        text="Gradient descent minimizes the loss.",
        origin="model_inferred",
        support_state="unverified",
        review_state="pending",
        evidence_refs=[ref],
    )
    repo = LooseGraphRepository.from_extraction([node], [], [claim], snapshot)

    # A non-evidence actor may NOT promote the claim to source_supported.
    with pytest.raises(InvalidTransitionError):
        transition_support(claim, "source_supported", actor="not_evidence_service")

    # The evidence service (correct actor) corroborates the fresh-evidence claim.
    service = EvidenceService(snapshot)
    updated = await service.corroborate_claims("n1", repo)
    assert any(c.support_state == "source_supported" for c in updated)
    assert repo.claims["cl1"].support_state == "source_supported"
    # Trust model stays unflattened: origin is unchanged and still immutable.
    assert repo.claims["cl1"].origin == "model_inferred"


# ── item 1: repo leg via M23 (skipped unless a repo is provided) ───────────────


async def test_repo_leg_via_m23_extraction() -> None:
    repo_env = os.environ.get("LOOSECANVAS_TEST_REPO")
    if not repo_env:
        pytest.skip(
            "set LOOSECANVAS_TEST_REPO to a repo under ALLOWED_BASE_DIR to run the "
            "repo-leg integration test"
        )
    repo_path = Path(repo_env).resolve()
    assert repo_path.is_relative_to(
        ALLOWED_BASE_DIR
    ), f"LOOSECANVAS_TEST_REPO must be under ALLOWED_BASE_DIR ({ALLOWED_BASE_DIR})"
    result = await ExtractionOrchestrator().run(repo_path)
    assert isinstance(result, ExtractionResult)
    assert result.scene.visible_node_ids, "extraction produced an empty scene"