"""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);
              msgpack round-trip schema version preservation.
Integration — build_symbol_index: no-op when no structured_delta; creates index row;
              correct symbol_history/hash_occurrence content; prunes stale index;
              invalidates MusehubFileIntelCache; 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 only one row per repo after rebuild;
              FileIntelCache invalidated on every rebuild;
              BFS reachability excludes orphaned branches.
Security    — Corrupt msgpack blob returns {} not exception;
              build_symbol_index with unknown head_commit_id returns None gracefully.
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;
              upsert_snapshot_entries 1 000 files < 500 ms;
              get_snapshot_manifests_batch 50 snapshots < 300 ms.
E2E         — Full pipeline: commits with structured_delta → build_symbol_index →
              get_index_meta returns correct ref; rebuild replaces previous row;
              symbol list HTTP page returns 200 for repo with index.
"""
from __future__ import annotations

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

import msgpack
import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from musehub.db import musehub_models as db
from tests.factories import create_repo
from musehub.muse_contracts.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",
) -> 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="gabriel",
        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}


# ─────────────────────────────────────────────────────────────────────────────
# 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
        # The child without address must not appear
        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("") == ""


class TestMsgpackRoundTrip:
    """Verify the msgpack blob schema produced by build_symbol_index is stable."""

    def test_schema_version_preserved(self) -> None:
        data = {
            "schema_version": "musehub-v1",
            "index": "symbol_history",
            "updated_at": "2026-04-04T00:00:00+00:00",
            "entries": {"main.py::Foo": [{"commit_id": "abc", "op": "add"}]},
        }
        blob = msgpack.packb(data, use_bin_type=True)
        recovered = msgpack.unpackb(blob, raw=False)
        assert recovered["schema_version"] == "musehub-v1"
        assert recovered["entries"]["main.py::Foo"][0]["op"] == "add"

    def test_bytes_values_survive_round_trip(self) -> None:
        data = {"entries": {"k": b"\x00\x01\x02"}}
        blob = msgpack.packb(data, use_bin_type=True)
        recovered = msgpack.unpackb(blob, raw=False)
        assert recovered["entries"]["k"] == b"\x00\x01\x02"


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

class TestBuildSymbolIndex:
    @pytest.mark.asyncio
    async def test_returns_none_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")

        result = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        assert result is None

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

        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")],
        )

        result = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        assert result is not None
        assert result.repo_id == repo.repo_id
        assert result.ref == commit.commit_id

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

        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"),
            ],
        )

        row = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        assert row is not None and row.symbol_history is not None

        doc = msgpack.unpackb(row.symbol_history, raw=False)
        entries = doc["entries"]
        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:
        from musehub.services.musehub_symbol_indexer import build_symbol_index

        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),
            ],
        )

        row = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        assert row is not None and row.hash_occurrence is not None

        doc = msgpack.unpackb(row.hash_occurrence, raw=False)
        entries = doc["entries"]
        assert shared_hash in entries
        assert set(entries[shared_hash]) == {"a.py::Foo", "b.py::Bar"}

    @pytest.mark.asyncio
    async def test_rebuild_prunes_old_index_row(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import build_symbol_index
        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_symbol_index(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_symbol_index(db_session, repo.repo_id, c2.commit_id)
        await db_session.commit()

        count = (await db_session.execute(
            select(func.count()).where(db.MusehubSymbolIndex.repo_id == repo.repo_id)
        )).scalar_one()
        assert count == 1

    @pytest.mark.asyncio
    async def test_rebuild_invalidates_file_intel_cache(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import build_symbol_index
        from sqlalchemy import select, func

        repo = await create_repo(db_session, slug="idx-cache-inv")
        # Seed a stale cache entry
        cache = db.MusehubFileIntelCache(
            repo_id=repo.repo_id,
            file_path="old.py",
            ref="stale-ref",
            intel_json="{}",
            symbol_history_json="{}",
        )
        db_session.add(cache)
        await db_session.flush()

        commit = await _commit_with_delta(db_session, repo.repo_id, "c200",
                                           ops=[_insert_op("new.py::X")])
        await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        remaining = (await db_session.execute(
            select(func.count()).where(db.MusehubFileIntelCache.repo_id == repo.repo_id)
        )).scalar_one()
        assert remaining == 0  # all cache entries invalidated

    @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 build_symbol_index

        repo = await create_repo(db_session, slug="idx-bfs")
        # Main chain: orphan → (not linked to main chain)
        orphan = await _commit_with_delta(
            db_session, repo.repo_id, "orphan",
            ops=[_insert_op("orphan.py::OrphanSym", "sha256:orphan")],
            parent_ids=[],
        )
        # Head chain
        head = await _commit_with_delta(
            db_session, repo.repo_id, "head",
            ops=[_insert_op("main.py::RealSym", "sha256:real")],
            parent_ids=[],
        )

        row = await build_symbol_index(db_session, repo.repo_id, head.commit_id)
        assert row is not None
        history = msgpack.unpackb(row.symbol_history, raw=False)["entries"]
        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 build_symbol_index, 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"),  # file-level address
                _insert_op("b.py::Bar", "sha256:y"),
            ],
        )
        await build_symbol_index(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 build_symbol_index, 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_symbol_index(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 build_symbol_index, 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_symbol_index(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 build_symbol_index, 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")],
        )
        row = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        await db_session.commit()

        if row is not None and row.intel_full_json:
            snap = await load_intel_snapshot(db_session, repo.repo_id)
            assert snap is not None
        else:
            # intel computation may be skipped if musehub_intel unavailable
            pytest.skip("intel_full_json not populated — compute_intel unavailable")


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 build_symbol_index, 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_symbol_index(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 build_symbol_index, 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_symbol_index(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_symbol_index(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 build_symbol_index, 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_symbol_index(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:
        """upsert_snapshot_entries must delete old entries before inserting new ones."""
        from musehub.services.musehub_snapshot import (
            get_snapshot_manifest,
            upsert_snapshot_entries,
        )
        repo = await create_repo(db_session, slug="di-atomic")
        snap_id = "snap-atomic"
        await upsert_snapshot_entries(
            db_session, repo.repo_id, snap_id,
            {"old_file.py": "sha256:old", "shared.py": "sha256:shared"},
        )
        await db_session.commit()

        # Replace with entirely different manifest
        await upsert_snapshot_entries(
            db_session, repo.repo_id, snap_id,
            {"new_file.py": "sha256:new"},
        )
        await db_session.commit()

        manifest = await get_snapshot_manifest(db_session, snap_id)
        assert "new_file.py" in manifest
        assert "old_file.py" not in manifest  # atomically removed
        assert "shared.py" not in manifest

    @pytest.mark.asyncio
    async def test_only_one_index_row_per_repo_after_multiple_builds(
        self, db_session: AsyncSession
    ) -> None:
        from musehub.services.musehub_symbol_indexer import build_symbol_index
        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_symbol_index(db_session, repo.repo_id, c.commit_id)
            await db_session.commit()

        count = (await db_session.execute(
            select(func.count()).where(db.MusehubSymbolIndex.repo_id == repo.repo_id)
        )).scalar_one()
        assert count == 1

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

        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")],
        )
        row = await build_symbol_index(db_session, repo.repo_id, commit.commit_id)
        assert row is not None

        history = msgpack.unpackb(row.symbol_history, raw=False)["entries"]
        entry = history["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_msgpack_returns_empty_not_exception(
        self, db_session: AsyncSession
    ) -> None:
        """A corrupt symbol_history blob must return {} — not raise."""
        from musehub.services.musehub_symbol_indexer import load_symbol_history

        repo = await create_repo(db_session, slug="sec-corrupt")
        # Manually insert a row with garbage blob
        index_row = db.MusehubSymbolIndex(
            index_id=str(uuid.uuid4()),
            repo_id=repo.repo_id,
            ref="bad-ref",
            symbol_history=b"\xff\xfe\xfd\xfc garbage",
            hash_occurrence=b"\xff\xfe garbage",
            built_at=_now(),
        )
        db_session.add(index_row)
        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_none(
        self, db_session: AsyncSession
    ) -> None:
        """Unknown head_commit_id must return None, not raise."""
        from musehub.services.musehub_symbol_indexer import build_symbol_index

        repo = await create_repo(db_session, slug="sec-unknown-head")
        result = await build_symbol_index(
            db_session, repo.repo_id, "nonexistent-commit-id"
        )
        assert result is None

    @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

        repo = await create_repo(db_session, slug="sec-corrupt-hash")
        index_row = db.MusehubSymbolIndex(
            index_id=str(uuid.uuid4()),
            repo_id=repo.repo_id,
            ref="bad-ref2",
            symbol_history=None,
            hash_occurrence=b"\x00corrupt",
            built_at=_now(),
        )
        db_session.add(index_row)
        await db_session.commit()

        result = await load_hash_occurrence(db_session, repo.repo_id)
        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 build_symbol_index

        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

        row = await build_symbol_index(db_session, repo.repo_id, head_id)
        await db_session.commit()
        assert row is not None
        history = msgpack.unpackb(row.symbol_history, raw=False)["entries"]
        # 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 build_symbol_index, load_symbol_history

        repo = await create_repo(db_session, slug="stress-filter-large")
        # 50 files, each with 10 symbols
        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_symbol_index(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")
        # Only the 10 symbols for module_05 should be returned
        assert len(result) == 10
        for key in result:
            assert key.startswith("src/module_05.py")


# ─────────────────────────────────────────────────────────────────────────────
# Layer 7 — Performance
# ─────────────────────────────────────────────────────────────────────────────

class TestPerformance:
    def test_extract_ops_1000_calls_under_100ms(self) -> None:
        from musehub.services.musehub_symbol_indexer import _extract_ops
        meta = {
            "structured_delta": {
                "ops": [
                    _patch_op("main.py", [_insert_op("main.py::Foo"), _insert_op("main.py::Bar")]),
                    _insert_op("util.py::Helper"),
                ]
            }
        }
        t0 = time.perf_counter()
        for _ in range(1000):
            _extract_ops(meta)
        elapsed_ms = (time.perf_counter() - t0) * 1000
        assert elapsed_ms < 100, f"_extract_ops 1000 calls took {elapsed_ms:.1f}ms"

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

        repo = await create_repo(db_session, slug="perf-idx-100c")
        prev_id: str | None = None
        for i in range(100):
            cid = f"perf-{i:04d}"
            ops = [_insert_op(f"p{i}.py::S{j}") for j in range(5)]
            c = await _commit_with_delta(db_session, repo.repo_id, cid, ops=ops,
                                          parent_ids=[prev_id] if prev_id else [])
            prev_id = c.commit_id

        t0 = time.perf_counter()
        row = await build_symbol_index(db_session, repo.repo_id, prev_id)
        elapsed_ms = (time.perf_counter() - t0) * 1000
        assert row is not None
        assert elapsed_ms < 3000, f"build_symbol_index 100 commits took {elapsed_ms:.0f}ms"

    @pytest.mark.asyncio
    async def test_upsert_1000_files_under_500ms(self, db_session: AsyncSession) -> None:
        from musehub.services.musehub_snapshot import upsert_snapshot_entries

        repo = await create_repo(db_session, slug="perf-upsert-1k")
        manifest = {f"f{i:04d}.py": f"sha256:{i}" for i in range(1000)}

        t0 = time.perf_counter()
        await upsert_snapshot_entries(db_session, repo.repo_id, "perf-snap", manifest)
        elapsed_ms = (time.perf_counter() - t0) * 1000
        assert elapsed_ms < 500, f"upsert 1000 entries took {elapsed_ms:.0f}ms"

    @pytest.mark.asyncio
    async def test_batch_manifest_50_snapshots_under_300ms(
        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="perf-batch-50")
        ids: list[str] = []
        for i in range(50):
            sid = f"perf-batch-{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()

        # Warm-up
        await get_snapshot_manifests_batch(db_session, ids)

        t0 = time.perf_counter()
        result = await get_snapshot_manifests_batch(db_session, ids)
        elapsed_ms = (time.perf_counter() - t0) * 1000
        assert len(result) == 50
        assert elapsed_ms < 300, f"batch 50 snapshots took {elapsed_ms:.0f}ms"