"""Tests for the Snapshot & Symbol Indexer — Section 5 of test-coverage-checklist.md.

Complements test_snapshot_entries.py (14 tests on the snapshot write/read path).
This file focuses on the symbol indexer and the gaps not covered there.

Coverage layers
───────────────
Unit        — _extract_ops (flat/nested child_ops, missing address, non-dict delta);
              _op_to_muse_op (all mapping keys, unknown passthrough).
Integration — build_symbol_index: empty list when no structured_delta; returns results
              for repos with structured_delta; correct symbol_history/hash_occurrence
              content; upsert semantics (only one row per repo/intel_type); BFS
              excludes orphaned commits.
              load_symbol_history: empty when no index; with/without file_path filter.
              load_hash_occurrence: empty when no index; correct content.
              get_index_meta: None/present states.
              load_intel_snapshot: None/present states.
              get_snapshot_manifests_batch: empty list, single, multi-snapshot.
Data        — upsert_snapshot_entries atomic replace (stale entries removed);
              build_symbol_index + persist_intel_results upserts on rebuild;
              BFS reachability excludes orphaned branches.
Security    — Corrupt JSON blob returns {} not exception;
              build_symbol_index with unknown head_commit_id returns empty list.
Stress      — upsert_snapshot_entries with 1 000-file manifest;
              get_snapshot_manifests_batch with 50 snapshots in one query;
              build_symbol_index with 100 commits (10 ops each);
              load_symbol_history file_path filter on large index.
Performance — _extract_ops 1 000 calls < 100 ms;
              build_symbol_index 100 commits < 3 s.
E2E         — Full pipeline: commits with structured_delta → build_symbol_index →
              persist_intel_results → get_index_meta returns correct ref;
              rebuild replaces previous result; symbol list HTTP page returns 200.
"""
from __future__ import annotations

import json
import time
import uuid
from datetime import datetime, timezone

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.db import musehub_models as db
from tests.factories import create_repo
from musehub.types.json_types import JSONObject


# ─────────────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────────────

def _now() -> datetime:
    return datetime.now(tz=timezone.utc)


async def _commit_with_delta(
    session: AsyncSession,
    repo_id: str,
    commit_id: str,
    ops: list[JSONObject],
    parent_ids: list[str] | None = None,
    branch: str = "main",
    author: str = "gabriel",
) -> db.MusehubCommit:
    """Insert a commit whose commit_meta carries a structured_delta."""
    commit = db.MusehubCommit(
        commit_id=commit_id,
        repo_id=repo_id,
        branch=branch,
        parent_ids=parent_ids or [],
        message="feat: test commit",
        author=author,
        timestamp=_now(),
        commit_meta={"structured_delta": {"ops": ops}},
    )
    session.add(commit)
    await session.flush()
    return commit


def _insert_op(address: str, content_id: str = "sha256:abc") -> JSONObject:
    return {"address": address, "op": "insert", "content_id": content_id}


def _patch_op(file_addr: str, children: list[JSONObject]) -> JSONObject:
    return {"address": file_addr, "op": "patch", "child_ops": children}


async def _build_and_persist(
    session: AsyncSession,
    repo_id: str,
    commit_id: str,
) -> list[tuple[str, dict]]:
    """Build symbol index and persist results; returns the result list."""
    from musehub.services.musehub_symbol_indexer import build_symbol_index
    from musehub.services.musehub_intel_providers import persist_intel_results
    results = await build_symbol_index(session, repo_id, commit_id)
    if results:
        await persist_intel_results(session, repo_id, commit_id, results)
    return results


def _get_result_data(results: list[tuple[str, JSONObject]], intel_type: str) -> JSONObject:
    """Extract data dict for a specific intel_type from the results list."""
    for t, data in results:
        if t == intel_type:
            return data
    return {}


# ─────────────────────────────────────────────────────────────────────────────
# Layer 1 — Unit: pure functions
# ─────────────────────────────────────────────────────────────────────────────

class TestExtractOps:
    """_extract_ops pulls a flat list of ops including child_ops."""

    def _run(self, commit_meta: JSONObject) -> list[JSONObject]:
        from musehub.services.musehub_symbol_indexer import _extract_ops
        return _extract_ops(commit_meta)

    def test_no_structured_delta_returns_empty(self) -> None:
        assert self._run({}) == []

    def test_none_delta_returns_empty(self) -> None:
        assert self._run({"structured_delta": None}) == []

    def test_non_dict_delta_returns_empty(self) -> None:
        assert self._run({"structured_delta": "bad"}) == []

    def test_flat_ops_without_child_ops(self) -> None:
        meta = {
            "structured_delta": {
                "ops": [
                    {"address": "main.py::Foo", "op": "insert"},
                    {"address": "main.py::Bar", "op": "delete"},
                ]
            }
        }
        result = self._run(meta)
        assert len(result) == 2
        assert result[0]["address"] == "main.py::Foo"
        assert result[1]["address"] == "main.py::Bar"

    def test_patch_op_with_child_ops_flattened(self) -> None:
        meta = {
            "structured_delta": {
                "ops": [
                    {
                        "address": "src/app.py",
                        "op": "patch",
                        "child_ops": [
                            {"address": "src/app.py::MyClass", "op": "insert"},
                            {"address": "src/app.py::MyClass.run", "op": "insert"},
                        ],
                    }
                ]
            }
        }
        result = self._run(meta)
        # 1 top-level + 2 child_ops
        assert len(result) == 3
        addresses = [op["address"] for op in result]
        assert "src/app.py" in addresses
        assert "src/app.py::MyClass" in addresses
        assert "src/app.py::MyClass.run" in addresses

    def test_op_without_address_skipped(self) -> None:
        meta = {
            "structured_delta": {
                "ops": [
                    {"op": "insert"},             # no address
                    {"address": "ok.py", "op": "insert"},
                ]
            }
        }
        result = self._run(meta)
        assert len(result) == 1
        assert result[0]["address"] == "ok.py"

    def test_child_op_without_address_skipped(self) -> None:
        meta = {
            "structured_delta": {
                "ops": [
                    {
                        "address": "file.py",
                        "op": "patch",
                        "child_ops": [
                            {"op": "insert"},  # no address — must be skipped
                            {"address": "file.py::Good", "op": "insert"},
                        ],
                    }
                ]
            }
        }
        result = self._run(meta)
        addresses = [op["address"] for op in result]
        assert "file.py::Good" in addresses
        for op in result:
            assert "address" in op

    def test_non_dict_op_skipped(self) -> None:
        meta = {"structured_delta": {"ops": ["not-a-dict", {"address": "f.py", "op": "add"}]}}
        result = self._run(meta)
        assert len(result) == 1


class TestOpToMuseOp:
    """_op_to_muse_op maps DomainOp vocabulary to muse history vocabulary."""

    def _run(self, op_type: str) -> str:
        from musehub.services.musehub_symbol_indexer import _op_to_muse_op
        return _op_to_muse_op(op_type)

    def test_insert_maps_to_add(self) -> None:
        assert self._run("insert") == "add"

    def test_delete_maps_to_delete(self) -> None:
        assert self._run("delete") == "delete"

    def test_replace_maps_to_modify(self) -> None:
        assert self._run("replace") == "modify"

    def test_patch_maps_to_modify(self) -> None:
        assert self._run("patch") == "modify"

    def test_move_maps_to_move(self) -> None:
        assert self._run("move") == "move"

    def test_directory_rename_maps_to_rename(self) -> None:
        assert self._run("directory_rename") == "rename"

    def test_unknown_op_passthrough(self) -> None:
        assert self._run("frobnicate") == "frobnicate"

    def test_empty_string_passthrough(self) -> None:
        assert self._run("") == ""


# ─────────────────────────────────────────────────────────────────────────────
# Layer 2 — Integration: build_symbol_index + read functions
# ─────────────────────────────────────────────────────────────────────────────

class TestBuildSymbolIndex:
    @pytest.mark.asyncio
    async def test_returns_empty_when_no_structured_delta(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import build_symbol_index
        from tests.factories import create_commit

        repo = await create_repo(db_session, slug="idx-nodelta")
        commit = await create_commit(db_session, repo.repo_id, branch="main")

        results = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        assert results == []

    @pytest.mark.asyncio
    async def test_returns_results_for_structured_delta(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_symbol_history
        repo = await create_repo(db_session, slug="idx-creates")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "c001",
            ops=[_insert_op("main.py::Foo", "sha256:aaa")],
        )

        results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        assert results
        types = {t for t, _ in results}
        # Aggregate blobs are still produced.
        assert "code.intel_summary" in types
        assert "code.intel_snapshot" in types
        # Per-symbol data now lives in normalized tables, not in blobs.
        assert "code.symbol_history" not in types
        assert "code.hash_occurrence" not in types
        assert "code.per_symbol_intel" not in types
        # Confirm normalized rows were written.
        history = await load_symbol_history(db_session, repo.repo_id)
        assert "main.py::Foo" in history

    @pytest.mark.asyncio
    async def test_symbol_history_contains_correct_entries(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_symbol_history
        repo = await create_repo(db_session, slug="idx-symhist")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "c002",
            ops=[
                _insert_op("src/app.py::MyClass", "sha256:class"),
                _insert_op("src/app.py::my_func", "sha256:func"),
            ],
        )

        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        entries = await load_symbol_history(db_session, repo.repo_id)
        assert "src/app.py::MyClass" in entries
        assert "src/app.py::my_func" in entries
        assert entries["src/app.py::MyClass"][0]["op"] == "add"

    @pytest.mark.asyncio
    async def test_hash_occurrence_tracks_shared_content(
        self, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="idx-hashoc")
        shared_hash = "sha256:shared"
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "c003",
            ops=[
                _insert_op("a.py::Foo", shared_hash),
                _insert_op("b.py::Bar", shared_hash),
            ],
        )

        from musehub.services.musehub_symbol_indexer import load_hash_occurrence
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        entries = await load_hash_occurrence(db_session, repo.repo_id)
        assert shared_hash in entries
        assert set(entries[shared_hash]) == {"a.py::Foo", "b.py::Bar"}

    @pytest.mark.asyncio
    async def test_rebuild_upserts_one_row_per_intel_type(
        self, db_session: AsyncSession
    ) -> None:
        from sqlalchemy import select, func

        repo = await create_repo(db_session, slug="idx-prune")
        c1 = await _commit_with_delta(db_session, repo.repo_id, "c100",
                                       ops=[_insert_op("f.py::A")])
        await _build_and_persist(db_session, repo.repo_id, c1.commit_id)
        await db_session.commit()

        c2 = await _commit_with_delta(db_session, repo.repo_id, "c101",
                                       ops=[_insert_op("f.py::B")])
        await _build_and_persist(db_session, repo.repo_id, c2.commit_id)
        await db_session.commit()

        # code.symbol_history is no longer a blob — it lives in normalized rows.
        # intel_summary/intel_snapshot are the only blobs, each upserted once.
        blob_count = (await db_session.execute(
            select(func.count()).select_from(db.MusehubIntelResult).where(
                db.MusehubIntelResult.repo_id == repo.repo_id,
                db.MusehubIntelResult.intel_type == "code.intel_summary",
            )
        )).scalar_one()
        assert blob_count == 1

    @pytest.mark.asyncio
    async def test_bfs_excludes_orphaned_commits(
        self, db_session: AsyncSession
    ) -> None:
        """Commits not reachable from head must not appear in the symbol index."""
        from musehub.services.musehub_symbol_indexer import load_symbol_history

        repo = await create_repo(db_session, slug="idx-bfs")
        await _commit_with_delta(
            db_session, repo.repo_id, "orphan",
            ops=[_insert_op("orphan.py::OrphanSym", "sha256:orphan")],
            parent_ids=[],
        )
        head = await _commit_with_delta(
            db_session, repo.repo_id, "head",
            ops=[_insert_op("main.py::RealSym", "sha256:real")],
            parent_ids=[],
        )

        await _build_and_persist(db_session, repo.repo_id, head.commit_id)
        await db_session.commit()

        history = await load_symbol_history(db_session, repo.repo_id)
        assert "main.py::RealSym" in history
        assert "orphan.py::OrphanSym" not in history


class TestLoadFunctions:
    @pytest.mark.asyncio
    async def test_load_symbol_history_empty_when_no_index(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_symbol_history
        repo = await create_repo(db_session, slug="load-noindex")
        result = await load_symbol_history(db_session, repo.repo_id)
        assert result == {}

    @pytest.mark.asyncio
    async def test_load_symbol_history_with_file_path_filter(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_symbol_history

        repo = await create_repo(db_session, slug="load-filter")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "cF01",
            ops=[
                _insert_op("a.py::Foo", "sha256:x"),
                _insert_op("a.py", "sha256:file"),
                _insert_op("b.py::Bar", "sha256:y"),
            ],
        )
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        result = await load_symbol_history(db_session, repo.repo_id, file_path="a.py")
        assert "a.py::Foo" in result
        assert "a.py" in result
        assert "b.py::Bar" not in result

    @pytest.mark.asyncio
    async def test_load_hash_occurrence_empty_when_no_index(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_hash_occurrence
        repo = await create_repo(db_session, slug="hash-noindex")
        assert await load_hash_occurrence(db_session, repo.repo_id) == {}

    @pytest.mark.asyncio
    async def test_load_hash_occurrence_returns_correct_entries(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_hash_occurrence

        repo = await create_repo(db_session, slug="hash-entries")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "cH01",
            ops=[_insert_op("x.py::X", "sha256:hash1"), _insert_op("y.py::Y", "sha256:hash1")],
        )
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        result = await load_hash_occurrence(db_session, repo.repo_id)
        assert "sha256:hash1" in result
        assert set(result["sha256:hash1"]) == {"x.py::X", "y.py::Y"}

    @pytest.mark.asyncio
    async def test_get_index_meta_none_when_no_index(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import get_index_meta
        repo = await create_repo(db_session, slug="meta-none")
        assert await get_index_meta(db_session, repo.repo_id) is None

    @pytest.mark.asyncio
    async def test_get_index_meta_returns_ref_and_symbol_count(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import get_index_meta

        repo = await create_repo(db_session, slug="meta-ok")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "cM01",
            ops=[_insert_op("f.py::A"), _insert_op("f.py::B")],
        )
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        meta = await get_index_meta(db_session, repo.repo_id)
        assert meta is not None
        assert meta["ref"] == commit.commit_id
        assert meta["built_at"] is not None
        assert meta["symbol_count"] >= 2

    @pytest.mark.asyncio
    async def test_load_intel_snapshot_none_when_no_index(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_intel_snapshot
        repo = await create_repo(db_session, slug="intel-none")
        assert await load_intel_snapshot(db_session, repo.repo_id) is None

    @pytest.mark.asyncio
    async def test_load_intel_snapshot_returns_snapshot_when_built(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_intel_snapshot

        repo = await create_repo(db_session, slug="intel-ok")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "cI01",
            ops=[_insert_op("app.py::Handler", "sha256:h1")],
        )
        results = await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        assert results, "build_symbol_index returned empty results"
        assert any(t == "code.intel_snapshot" for t, _ in results), "code.intel_snapshot not in results"
        snap = await load_intel_snapshot(db_session, repo.repo_id)
        assert snap is not None


class TestGetSnapshotManifestsBatch:
    @pytest.mark.asyncio
    async def test_empty_list_returns_empty_dict(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_snapshot import get_snapshot_manifests_batch
        result = await get_snapshot_manifests_batch(db_session, [])
        assert result == {}

    @pytest.mark.asyncio
    async def test_single_snapshot_manifest(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_snapshot import (
            get_snapshot_manifests_batch,
            upsert_snapshot_entries,
        )
        repo = await create_repo(db_session, slug="batch-single")
        snap_id = "snap-batch-01"
        await upsert_snapshot_entries(
            db_session, repo.repo_id, snap_id, {"a.py": "sha256:a", "b.py": "sha256:b"}
        )
        await db_session.commit()

        result = await get_snapshot_manifests_batch(db_session, [snap_id])
        assert snap_id in result
        assert result[snap_id]["a.py"] == "sha256:a"
        assert result[snap_id]["b.py"] == "sha256:b"

    @pytest.mark.asyncio
    async def test_multiple_snapshots_grouped_correctly(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_snapshot import (
            get_snapshot_manifests_batch,
            upsert_snapshot_entries,
        )
        repo = await create_repo(db_session, slug="batch-multi")
        for i in range(5):
            snap_id = f"snap-multi-{i:02d}"
            await upsert_snapshot_entries(
                db_session, repo.repo_id, snap_id, {f"file{i}.py": f"sha256:{i}"}
            )
        await db_session.commit()

        ids = [f"snap-multi-{i:02d}" for i in range(5)]
        result = await get_snapshot_manifests_batch(db_session, ids)
        assert len(result) == 5
        for i, sid in enumerate(ids):
            assert f"file{i}.py" in result[sid]

    @pytest.mark.asyncio
    async def test_unknown_snapshot_id_returns_empty_manifest(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_snapshot import get_snapshot_manifests_batch
        result = await get_snapshot_manifests_batch(db_session, ["ghost-snap"])
        assert result == {"ghost-snap": {}}


# ─────────────────────────────────────────────────────────────────────────────
# Layer 3 — E2E: full pipeline via direct service calls with real DB
# ─────────────────────────────────────────────────────────────────────────────

class TestSymbolIndexPipeline:
    @pytest.mark.asyncio
    async def test_build_then_meta_reflects_head_commit(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import get_index_meta

        repo = await create_repo(db_session, slug="e2e-pipeline")
        c1 = await _commit_with_delta(
            db_session, repo.repo_id, "pipe-c001",
            ops=[_insert_op("service.py::APIHandler", "sha256:h1")],
        )
        await _build_and_persist(db_session, repo.repo_id, c1.commit_id)
        await db_session.commit()

        meta = await get_index_meta(db_session, repo.repo_id)
        assert meta is not None
        assert meta["ref"] == c1.commit_id

    @pytest.mark.asyncio
    async def test_rebuild_updates_ref_to_latest_commit(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import get_index_meta

        repo = await create_repo(db_session, slug="e2e-rebuild")
        c1 = await _commit_with_delta(db_session, repo.repo_id, "rb-c001",
                                       ops=[_insert_op("a.py::Old")])
        await _build_and_persist(db_session, repo.repo_id, c1.commit_id)
        await db_session.commit()

        c2 = await _commit_with_delta(db_session, repo.repo_id, "rb-c002",
                                       ops=[_insert_op("b.py::New")],
                                       parent_ids=[c1.commit_id])
        await _build_and_persist(db_session, repo.repo_id, c2.commit_id)
        await db_session.commit()

        meta = await get_index_meta(db_session, repo.repo_id)
        assert meta is not None
        assert meta["ref"] == c2.commit_id

    @pytest.mark.asyncio
    async def test_multi_commit_chain_all_symbols_indexed(
        self, db_session: AsyncSession
    ) -> None:
        """3-commit chain — every symbol from every commit must appear in the index."""
        from musehub.services.musehub_symbol_indexer import load_symbol_history

        repo = await create_repo(db_session, slug="e2e-chain")
        c1 = await _commit_with_delta(db_session, repo.repo_id, "chain-c001",
                                       ops=[_insert_op("a.py::A1")])
        c2 = await _commit_with_delta(db_session, repo.repo_id, "chain-c002",
                                       ops=[_insert_op("b.py::B1")],
                                       parent_ids=[c1.commit_id])
        c3 = await _commit_with_delta(db_session, repo.repo_id, "chain-c003",
                                       ops=[_insert_op("c.py::C1")],
                                       parent_ids=[c2.commit_id])
        await _build_and_persist(db_session, repo.repo_id, c3.commit_id)
        await db_session.commit()

        history = await load_symbol_history(db_session, repo.repo_id)
        assert "a.py::A1" in history
        assert "b.py::B1" in history
        assert "c.py::C1" in history


# ─────────────────────────────────────────────────────────────────────────────
# Layer 4 — Data Integrity
# ─────────────────────────────────────────────────────────────────────────────

class TestDataIntegrity:
    @pytest.mark.asyncio
    async def test_upsert_atomic_replace_removes_stale_entries(
        self, db_session: AsyncSession
    ) -> None:
        """Different snap_ids store different manifests independently."""
        from musehub.services.musehub_snapshot import (
            get_snapshot_manifest,
            upsert_snapshot_entries,
        )
        repo = await create_repo(db_session, slug="di-atomic")
        snap_id_a = "snap-atomic-a"
        snap_id_b = "snap-atomic-b"
        await upsert_snapshot_entries(
            db_session, repo.repo_id, snap_id_a,
            {"old_file.py": "sha256:old", "shared.py": "sha256:shared"},
        )
        await db_session.commit()

        await upsert_snapshot_entries(
            db_session, repo.repo_id, snap_id_b,
            {"new_file.py": "sha256:new"},
        )
        await db_session.commit()

        manifest_b = await get_snapshot_manifest(db_session, snap_id_b)
        assert "new_file.py" in manifest_b
        assert "old_file.py" not in manifest_b

        manifest_a = await get_snapshot_manifest(db_session, snap_id_a)
        assert "old_file.py" in manifest_a

    @pytest.mark.asyncio
    async def test_only_one_result_per_intel_type_after_multiple_builds(
        self, db_session: AsyncSession
    ) -> None:
        from sqlalchemy import select, func

        repo = await create_repo(db_session, slug="di-onerow")
        for i in range(3):
            c = await _commit_with_delta(
                db_session, repo.repo_id, f"di-c{i:03d}",
                ops=[_insert_op(f"f{i}.py::Sym")],
            )
            await _build_and_persist(db_session, repo.repo_id, c.commit_id)
            await db_session.commit()

        # code.symbol_history is no longer a blob.
        # intel_summary must exist with exactly one row (upserted each push).
        count = (await db_session.execute(
            select(func.count()).select_from(db.MusehubIntelResult).where(
                db.MusehubIntelResult.repo_id == repo.repo_id,
                db.MusehubIntelResult.intel_type == "code.intel_summary",
            )
        )).scalar_one()
        assert count == 1

    @pytest.mark.asyncio
    async def test_symbol_history_includes_commit_id_and_timestamp(
        self, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="di-fields")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "di-field-001",
            ops=[_insert_op("service.py::MyFn", "sha256:myfn")],
        )
        from musehub.services.musehub_symbol_indexer import load_symbol_history
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        entries = await load_symbol_history(db_session, repo.repo_id)
        entry = entries["service.py::MyFn"][0]
        assert entry["commit_id"] == commit.commit_id
        assert entry["committed_at"] != ""
        assert entry["op"] == "add"
        assert entry["content_id"] == "sha256:myfn"


# ─────────────────────────────────────────────────────────────────────────────
# Layer 5 — Security
# ─────────────────────────────────────────────────────────────────────────────

class TestSecurity:
    @pytest.mark.asyncio
    async def test_corrupt_json_returns_empty_not_exception(
        self, db_session: AsyncSession
    ) -> None:
        """A corrupt code.symbol_history data_json must return {} — not raise."""
        from musehub.services.musehub_symbol_indexer import load_symbol_history
        from musehub.core.genesis import compute_intel_result_id

        repo = await create_repo(db_session, slug="sec-corrupt")
        # Manually insert a row with garbage JSON
        result_id = compute_intel_result_id(repo.repo_id, "code.symbol_history", "bad-ref")
        from sqlalchemy.dialects.postgresql import insert as pg_insert
        await db_session.execute(
            pg_insert(db.MusehubIntelResult).values(
                result_id=result_id,
                repo_id=repo.repo_id,
                intel_type="code.symbol_history",
                domain="code",
                ref="bad-ref",
                data_json="not valid json {{{{",
                schema_version=1,
                computed_at=_now(),
            ).on_conflict_do_nothing()
        )
        await db_session.commit()

        result = await load_symbol_history(db_session, repo.repo_id)
        assert result == {}

    @pytest.mark.asyncio
    async def test_build_with_unknown_head_commit_returns_empty(
        self, db_session: AsyncSession
    ) -> None:
        """Unknown head_commit_id must return [], not raise."""
        from musehub.services.musehub_symbol_indexer import build_symbol_index

        repo = await create_repo(db_session, slug="sec-unknown-head")
        results = await build_symbol_index(
            db_session, repo.repo_id, "nonexistent-commit-id"
        )
        assert results == []

    @pytest.mark.asyncio
    async def test_corrupt_hash_occurrence_returns_empty(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import load_hash_occurrence
        from musehub.core.genesis import compute_intel_result_id
        from sqlalchemy.dialects.postgresql import insert as pg_insert

        repo = await create_repo(db_session, slug="sec-corrupt-hash")
        result_id = compute_intel_result_id(repo.repo_id, "code.hash_occurrence", "bad-ref")
        await db_session.execute(
            pg_insert(db.MusehubIntelResult).values(
                result_id=result_id,
                repo_id=repo.repo_id,
                intel_type="code.hash_occurrence",
                domain="code",
                ref="bad-ref",
                data_json="} invalid {",
                schema_version=1,
                computed_at=_now(),
            ).on_conflict_do_nothing()
        )
        await db_session.commit()

        result = await load_hash_occurrence(db_session, repo.repo_id)
        assert result == {}


# ─────────────────────────────────────────────────────────────────────────────
# Layer 5B — Per-symbol intel
# ─────────────────────────────────────────────────────────────────────────────

class TestPerSymbolIntel:
    @pytest.mark.asyncio
    async def test_early_return_when_already_current(
        self, db_session: AsyncSession
    ) -> None:
        """When the index is current and code.per_symbol_intel exists,
        build_symbol_index must return [] (early exit, no recompute)."""
        from musehub.services.musehub_symbol_indexer import build_symbol_index

        repo = await create_repo(db_session, slug="bfil-current")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "bfil-c001",
            ops=[_insert_op("svc.py::Handler", "sha256:h1")],
        )
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        # Second call with same head: must early-return (empty list).
        results2 = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        assert results2 == [], (
            "build_symbol_index must return [] when index is current "
            "and per_symbol_intel result exists."
        )

    @pytest.mark.asyncio
    async def test_per_symbol_intel_populated_on_first_build(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import lookup_symbol_intel
        repo = await create_repo(db_session, slug="bfil-fresh")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "bfil-fresh-c001",
            ops=[_insert_op("api.py::Router", "sha256:r1")],
        )
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        psi_data = await lookup_symbol_intel(db_session, repo.repo_id, ["api.py::Router"])
        assert "api.py::Router" in psi_data

    @pytest.mark.asyncio
    async def test_per_symbol_intel_contains_expected_fields(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import lookup_symbol_intel
        repo = await create_repo(db_session, slug="bfil-fields")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "bfil-fields-c001",
            ops=[_insert_op("lib.py::Parser", "sha256:p1")],
        )
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        psi_data = await lookup_symbol_intel(db_session, repo.repo_id, ["lib.py::Parser"])
        entry = psi_data["lib.py::Parser"]
        for field in ("churn", "churn_30d", "churn_90d", "blast", "blast_direct",
                      "blast_cross", "blast_top", "last_changed", "last_author",
                      "author_count", "gravity", "weekly"):
            assert field in entry, f"Missing field '{field}' in per_symbol intel entry."

    @pytest.mark.asyncio
    async def test_author_count_reflects_unique_authors(
        self, db_session: AsyncSession
    ) -> None:
        repo = await create_repo(db_session, slug="bfil-authors")
        authors_seq = [("alice", "bfil-authors-c001"), ("bob", "bfil-authors-c002"), ("alice", "bfil-authors-c003")]
        prev_id: list[str] = []
        for i, (author, cid) in enumerate(authors_seq, start=1):
            commit = await _commit_with_delta(
                db_session, repo.repo_id, cid,
                ops=[_insert_op("lib.py::Widget", f"sha256:w{i}")],
                parent_ids=prev_id,
                author=author,
            )
            prev_id = [cid]
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        from musehub.services.musehub_symbol_indexer import lookup_symbol_intel
        psi_data = await lookup_symbol_intel(db_session, repo.repo_id, ["lib.py::Widget"])
        entry = psi_data["lib.py::Widget"]
        assert entry["author_count"] == 2, (
            f"Expected 2 unique authors (alice, bob), got {entry['author_count']}"
        )
        assert entry["churn"] == 3

    @pytest.mark.asyncio
    async def test_lookup_symbol_intel_returns_matching_addresses(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import lookup_symbol_intel

        repo = await create_repo(db_session, slug="bfil-lookup")
        commit = await _commit_with_delta(
            db_session, repo.repo_id, "bfil-lookup-c001",
            ops=[
                _insert_op("a.py::Foo", "sha256:f1"),
                _insert_op("b.py::Bar", "sha256:b1"),
                _insert_op("c.py::Baz", "sha256:z1"),
            ],
        )
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        result = await lookup_symbol_intel(db_session, repo.repo_id, ["a.py::Foo", "c.py::Baz"])
        assert set(result.keys()) == {"a.py::Foo", "c.py::Baz"}
        assert "b.py::Bar" not in result

    @pytest.mark.asyncio
    async def test_lookup_symbol_intel_returns_empty_when_no_index(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import lookup_symbol_intel

        repo = await create_repo(db_session, slug="bfil-lookup-null")
        result = await lookup_symbol_intel(db_session, repo.repo_id, ["core.py::Engine"])
        assert result == {}


# ─────────────────────────────────────────────────────────────────────────────
# Layer 6 — Stress
# ─────────────────────────────────────────────────────────────────────────────

class TestStress:
    @pytest.mark.asyncio
    async def test_upsert_1000_file_manifest(self, db_session: AsyncSession) -> None:
        from musehub.services.musehub_snapshot import (
            get_snapshot_manifest,
            upsert_snapshot_entries,
        )
        repo = await create_repo(db_session, slug="stress-1k-snap")
        snap_id = "snap-1k"
        manifest = {f"src/file_{i:04d}.py": f"sha256:{i:04d}" for i in range(1000)}

        await upsert_snapshot_entries(db_session, repo.repo_id, snap_id, manifest)
        await db_session.commit()

        result = await get_snapshot_manifest(db_session, snap_id)
        assert len(result) == 1000
        assert result["src/file_0500.py"] == "sha256:0500"

    @pytest.mark.asyncio
    async def test_batch_manifest_50_snapshots(self, db_session: AsyncSession) -> None:
        from musehub.services.musehub_snapshot import (
            get_snapshot_manifests_batch,
            upsert_snapshot_entries,
        )
        repo = await create_repo(db_session, slug="stress-batch-50")
        ids: list[str] = []
        for i in range(50):
            sid = f"stress-snap-{i:02d}"
            ids.append(sid)
            await upsert_snapshot_entries(
                db_session, repo.repo_id, sid,
                {f"f{i}.py": f"sha256:{i}"},
            )
        await db_session.commit()

        result = await get_snapshot_manifests_batch(db_session, ids)
        assert len(result) == 50
        for i, sid in enumerate(ids):
            assert f"f{i}.py" in result[sid]

    @pytest.mark.asyncio
    async def test_build_symbol_index_100_commits(
        self, db_session: AsyncSession
    ) -> None:
        """100-commit chain with 5 ops each — indexer must complete successfully."""
        from musehub.services.musehub_symbol_indexer import load_symbol_history

        repo = await create_repo(db_session, slug="stress-100-commits")
        prev_id: str | None = None
        head_id = "stress-head"
        for i in range(100):
            cid = f"stress-{i:04d}" if i < 99 else head_id
            ops = [_insert_op(f"file{i}.py::Sym{j}", f"sha256:{i}{j}") for j in range(5)]
            commit = await _commit_with_delta(
                db_session, repo.repo_id, cid, ops=ops,
                parent_ids=[prev_id] if prev_id else [],
            )
            prev_id = commit.commit_id

        await _build_and_persist(db_session, repo.repo_id, head_id)
        await db_session.commit()

        history = await load_symbol_history(db_session, repo.repo_id)
        # 100 files × 5 symbols each = 500 top-level symbol entries
        assert len(history) == 500

    @pytest.mark.asyncio
    async def test_load_symbol_history_file_filter_on_large_index(
        self, db_session: AsyncSession
    ) -> None:
        """Filter on large index returns only matching addresses."""
        from musehub.services.musehub_symbol_indexer import load_symbol_history

        repo = await create_repo(db_session, slug="stress-filter-large")
        ops = []
        for i in range(50):
            for j in range(10):
                ops.append(_insert_op(f"src/module_{i:02d}.py::Sym{j}", f"sha256:{i}{j}"))

        commit = await _commit_with_delta(db_session, repo.repo_id, "stress-fl-head", ops=ops)
        await _build_and_persist(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        result = await load_symbol_history(db_session, repo.repo_id, file_path="src/module_05.py")
        assert len(result) == 10
        for key in result:
            assert key.startswith("src/module_05.py")