"""Real-Postgres integration tests for shape classification (#2789).

What mocks can't prove: the all-or-nothing batch against real rows, the
write-ACL gate, the todo query's filters, and the consumer map riding
get_snippet. Ledger rows are seeded through the same sync the coverage walk
uses — no forge needed, the sync takes extracted shapes directly.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from scribe.models import async_session
from scribe.models.code_shape import CodeShape
from scribe.models.project import Project
from scribe.models.user import User
from scribe.services.shape_ledger import (
    classify_shapes,
    classify_shapes_where,
    list_project_shapes,
    snippet_consumers,
    sync_repo_shapes,
)
from tests.helpers import ensure_user

pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]

REPO = "git.example.com/alice/widget"
SHAPES = [
    ("src/app.py", "sym", "make_app"),
    ("src/app.py", "sym", "Config"),
    ("src/util.py", "sym", "helper"),
    ("web/button.css", "css", "btn"),
]


@pytest_asyncio.fixture
async def seeded():
    """Owner + outsider, a project with a synced 4-shape ledger, one snippet."""
    from scribe.services import snippets as snippets_svc

    async with async_session() as s:
        owner = await ensure_user(s, "classify_owner")
        other = await ensure_user(s, "classify_other")
        project = Project(user_id=owner.id, title="Classify target")
        s.add(project)
        await s.flush()
        ids = {"owner": owner.id, "other": other.id, "pid": project.id}
        await s.commit()

    await sync_repo_shapes(ids["pid"], REPO, SHAPES, seen_marker="main")
    snippet = await snippets_svc.create_snippet(
        ids["owner"], name="cls_make_app", code="def make_app():\n    pass\n",
        language="python", repo="Widget", path="src/factory.py",
        symbol="factory", project_id=ids["pid"],
    )
    ids["snippet"] = int(snippet.id)
    return ids


@pytest.mark.integration
async def test_classify_applies_judgments_and_reports_unmatched(seeded):
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    out = await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid},
        {"path": "src/util.py", "symbol": "helper", "status": "exempt",
         "reason": "test scaffolding, deliberately local"},
        {"path": "web/button.css", "symbol": "btn", "status": "variant",
         "snippet_id": sid, "reason": "darker focus ring for the toolbar"},
        {"path": "gone.py", "symbol": "nothing", "status": "exempt",
         "reason": "x"},
    ], via="audit")
    assert out["classified"] == 3
    assert out["unmatched"] == [{"path": "gone.py", "symbol": "nothing"}]

    rows, total = await list_project_shapes(owner, pid)
    by_symbol = {r.symbol: r for r in rows}
    assert total == 4
    assert by_symbol["make_app"].status == "instance"
    assert by_symbol["make_app"].snippet_id == sid
    assert by_symbol["make_app"].classified_by == "audit"
    assert by_symbol["helper"].status == "exempt"
    assert by_symbol["helper"].reason == "test scaffolding, deliberately local"
    assert by_symbol["btn"].status == "variant"
    assert by_symbol["Config"].status == "unclassified"

    # Withdrawing a judgment returns the shape to the todo, fields cleared.
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "unclassified"},
    ])
    rows, _ = await list_project_shapes(owner, pid, status="unclassified")
    assert {r.symbol for r in rows} == {"Config", "make_app"}
    make_app = next(r for r in rows if r.symbol == "make_app")
    assert make_app.snippet_id is None and make_app.classified_by is None


@pytest.mark.integration
async def test_a_bad_batch_applies_nothing(seeded):
    """All-or-nothing (#2709's lesson): a caller must never learn later that
    half a batch silently happened."""
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    with pytest.raises(ValueError) as err:
        await classify_shapes(owner, pid, [
            {"path": "src/app.py", "symbol": "make_app", "status": "instance",
             "snippet_id": sid},
            {"path": "src/util.py", "symbol": "helper", "status": "variant",
             "snippet_id": sid},  # variant with no reason: structural error
        ])
    assert "needs a reason" in str(err.value)
    rows, _ = await list_project_shapes(owner, pid, status="unclassified")
    assert len(rows) == 4  # including make_app — the valid half did NOT apply

    # A snippet target the caller can't read is the same: nothing applies.
    with pytest.raises(ValueError):
        await classify_shapes(owner, pid, [
            {"path": "src/app.py", "symbol": "make_app", "status": "instance",
             "snippet_id": 999999999},
        ])


@pytest.mark.integration
async def test_classification_is_write_gated_and_listing_read_gated(seeded):
    other, pid, sid = seeded["other"], seeded["pid"], seeded["snippet"]
    with pytest.raises(ValueError):
        await classify_shapes(other, pid, [
            {"path": "src/app.py", "symbol": "make_app", "status": "exempt",
             "reason": "not their call to make"},
        ])
    assert await list_project_shapes(other, pid) == ([], 0)


@pytest.mark.integration
async def test_rule_form_sweeps_unclassified_rows_only_and_applies_whole(seeded):
    """#2868: one judgment over a directory + glob; judged rows are left
    alone unless include_judged; the same gates as the row form."""
    owner, other, pid, sid = seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "Config", "status": "exempt", "reason": "settings holder"},
    ])
    out = await classify_shapes_where(
        owner, pid, path="src", status="instance", snippet_id=sid, via="audit",
    )
    # make_app + helper swept; Config (already judged) untouched; css not under src/.
    assert out["classified"] == 2
    assert out["sample"] == ["src/app.py::make_app", "src/util.py::helper"]
    rows, _ = await list_project_shapes(owner, pid)
    by_symbol = {r.symbol: r for r in rows}
    assert by_symbol["make_app"].status == "instance" and by_symbol["make_app"].classified_by == "audit"
    assert by_symbol["Config"].status == "exempt" and by_symbol["Config"].reason == "settings holder"
    assert by_symbol["btn"].status == "unclassified"
    # Glob + kind narrow; include_judged re-judges.
    out = await classify_shapes_where(
        owner, pid, path="web", status="exempt", pattern="btn*", kind="css",
        reason="one toolbar button", include_judged=True,
    )
    assert out["classified"] == 1
    out = await classify_shapes_where(
        owner, pid, path="src", status="unclassified", include_judged=True,
    )
    assert out["classified"] == 3   # withdrawal sweeps judged rows when asked
    # Gates: reason for exempt, snippet for instance, write access, a path.
    with pytest.raises(ValueError):
        await classify_shapes_where(owner, pid, path="src", status="exempt")
    with pytest.raises(ValueError):
        await classify_shapes_where(owner, pid, path="src", status="instance")
    with pytest.raises(ValueError):
        await classify_shapes_where(owner, pid, path="", status="exempt", reason="x")
    with pytest.raises(ValueError):
        await classify_shapes_where(other, pid, path="src", status="exempt", reason="x")


@pytest.mark.integration
async def test_sync_stamps_scoped_rows_and_unstamps_when_they_become_reachable(seeded):
    """#2869: by-construction one-offs arrive `scoped` (mechanical), count as
    accounted, are reached by the sweep's default, and go back to the todo
    if a later tree makes them ordinary. A judgment overrides the stamp."""
    from scribe.services.coverage import ArchiveShape
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    scoped_shapes = [
        ArchiveShape("web/Card.vue", "css", "card", ".card {", "s1", ".card { x: 1 }", True),
        ArchiveShape("web/Card.vue", "sym", "load", "function load() {", "s2", "function load() {}", True),
        ArchiveShape("src/util.py", "sym", "helper", "def helper():", "s3", "def helper(): pass", False),
    ]
    await sync_repo_shapes(pid, REPO, scoped_shapes, seen_marker="main")
    rows, _ = await list_project_shapes(owner, pid, path="web/Card.vue")
    assert {r.status for r in rows} == {"scoped"}
    assert all(r.classified_by == "mechanical" and "by construction" in (r.reason or "") for r in rows)
    # The human todo excludes them; the sweep's default still reaches them.
    assert (await list_project_shapes(owner, pid, status="unclassified", path="web/Card.vue"))[1] == 0
    out = await classify_shapes_where(
        owner, pid, path="web/Card.vue", status="instance", snippet_id=sid, kind="css",
    )
    assert out["classified"] == 1
    # Re-synced as ordinary: the stamped sym returns to the todo; the
    # judged css keeps its judgment.
    plain = [ArchiveShape(s.path, s.kind, s.name, s.signature, s.body_sha, s.body, False) for s in scoped_shapes]
    await sync_repo_shapes(pid, REPO, plain, seen_marker="main")
    rows, _ = await list_project_shapes(owner, pid, path="web/Card.vue")
    by_symbol = {r.symbol: r for r in rows}
    assert by_symbol["load"].status == "unclassified" and by_symbol["load"].classified_by is None
    assert by_symbol["card"].status == "instance" and by_symbol["card"].snippet_id == sid


@pytest.mark.integration
async def test_list_filters_compose(seeded):
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid},
    ])
    rows, total = await list_project_shapes(owner, pid, path="src")
    assert total == 3 and all(r.path.startswith("src/") for r in rows)
    # Directory semantics, not string prefix: "sr" matches nothing.
    assert (await list_project_shapes(owner, pid, path="sr"))[1] == 0
    rows, total = await list_project_shapes(owner, pid, snippet_id=sid)
    assert total == 1 and rows[0].symbol == "make_app"
    rows, total = await list_project_shapes(
        owner, pid, status="unclassified", limit=2
    )
    assert total == 3 and len(rows) == 2  # paged, with the true total


@pytest.mark.integration
async def test_get_snippet_carries_the_structured_consumer_map(seeded):
    from scribe.mcp._context import _user_id_ctx
    from scribe.mcp.tools.snippets import get_snippet

    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid},
        {"path": "web/button.css", "symbol": "btn", "status": "variant",
         "snippet_id": sid, "reason": "darker focus ring"},
    ])
    token = _user_id_ctx.set(owner)
    try:
        data = await get_snippet(snippet_id=sid)
    finally:
        _user_id_ctx.reset(token)
    assert [i["path"] for i in data["instances"]] == ["src/app.py"]
    assert data["variants"][0]["reason"] == "darker focus ring"

    # The map is caller-scoped: an outsider asking the service directly gets
    # silence, not another project's file layout.
    consumers = await snippet_consumers(seeded["other"], sid)
    assert consumers == {"instances": [], "variants": []}


@pytest.mark.integration
async def test_sync_refiles_rows_whose_snippet_was_purged(seeded):
    """The SET NULL companion (#2787): a judgment whose target is hard-deleted
    rejoins the todo on the next sync instead of dangling target-less."""
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid},
    ])
    from scribe.models.note import Note

    async with async_session() as s:
        note = await s.get(Note, sid)
        await s.delete(note)  # hard delete, as purge_trash would
        await s.commit()
    await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="main")
    async with async_session() as s:
        row = (await s.execute(select(CodeShape).where(
            CodeShape.project_id == pid, CodeShape.symbol == "make_app",
        ))).scalar_one()
    assert row.status == "unclassified"
    assert row.snippet_id is None


# --- #2791: the write-path feed lands hook evidence as rows -------------------


@pytest.mark.integration
async def test_write_path_stamp_is_evidence_that_yields_to_judgment(seeded):
    """Pulled + referenced → every named shape of the snippet's kind becomes
    an instance row, classified_by=hook, carrying the evidence as reason. A
    later agent judgment on one of them stands against a re-stamp; the hook
    may only overwrite nobody's judgment or its own. The outsider stamps
    nothing (write-gated like every other ledger write)."""
    from datetime import datetime, timezone

    from scribe.services.shape_ledger import stamp_write_path_instances

    owner, other, pid, sid = (
        seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
    )
    pulled = {sid: datetime.now(timezone.utc)}
    code = "app = factory()\nreturn app\n"          # references the snippet's symbol

    assert await stamp_write_path_instances(
        other, pid, path="src/app.py", shapes=[("sym", "make_app")],
        code=code, pulled=pulled,
    ) == []

    stamped = await stamp_write_path_instances(
        owner, pid, path="src/app.py",
        shapes=[("sym", "make_app"), ("sym", "Config"), ("css", "nope")],
        code=code, pulled=pulled,
    )
    assert {s["symbol"] for s in stamped} == {"make_app", "Config"}   # css skipped: no css canon
    rows, _ = await list_project_shapes(owner, pid, snippet_id=sid)
    by_symbol = {r.symbol: r for r in rows}
    assert by_symbol["make_app"].status == "instance"
    assert by_symbol["make_app"].classified_by == "hook"
    assert by_symbol["make_app"].reason == f"hook: pulled #{sid}; payload references `factory`"

    # A judgment lands; the next stamp must leave it alone but may re-stamp
    # its own earlier row.
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "exempt",
         "reason": "the app factory is its own thing"},
    ])
    again = await stamp_write_path_instances(
        owner, pid, path="src/app.py",
        shapes=[("sym", "make_app"), ("sym", "Config")], code=code, pulled=pulled,
    )
    assert {s["symbol"] for s in again} == {"Config"}
    rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
    by_symbol = {r.symbol: r for r in rows}
    assert by_symbol["make_app"].status == "exempt"
    assert by_symbol["Config"].status == "instance"

    # Neither pulled nor in play → nothing, even with shapes named.
    assert await stamp_write_path_instances(
        owner, pid, path="src/util.py", shapes=[("sym", "helper")],
        code="print('unrelated')", pulled=pulled,
    ) == []


@pytest.mark.integration
async def test_a_brand_new_shape_gets_a_provisional_row_the_sync_settles(seeded):
    """The shape being written right now has no ledger row yet. With the
    hook's repo key it gets a provisional one — seen markers empty — so the
    stamp survives until the next sync, which confirms it (sets the marker)
    or stamps it vanished. Without a repo key only existing rows are touched."""
    from datetime import datetime, timezone

    from scribe.services.shape_ledger import stamp_write_path_instances

    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    pulled = {sid: datetime.now(timezone.utc)}
    code = "def build():\n    return factory()\n"

    assert await stamp_write_path_instances(
        owner, pid, path="src/new.py", shapes=[("sym", "build")],
        code=code, pulled=pulled,                              # no repo_key
    ) == []
    stamped = await stamp_write_path_instances(
        owner, pid, path="src/new.py", shapes=[("sym", "build")],
        code=code, pulled=pulled, repo_key=REPO,
    )
    assert [s["symbol"] for s in stamped] == ["build"]
    async with async_session() as s:
        row = (await s.execute(select(CodeShape).where(
            CodeShape.project_id == pid, CodeShape.path == "src/new.py",
        ))).scalar_one()
        assert row.status == "instance" and row.classified_by == "hook"
        # "Unset" is the column's empty default — the markers are non-null
        # Text, and the sync is what first fills them.
        assert row.first_seen_commit == "" and row.last_seen_commit == ""

    # The sync sees the shape in the tree → confirmed, stamp intact.
    await sync_repo_shapes(
        pid, REPO, SHAPES + [("src/new.py", "sym", "build")], seen_marker="abc123",
    )
    rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
    assert rows[0].status == "instance" and rows[0].last_seen_commit == "abc123"

    # The sync no longer sees it → vanished, out of the live accounting.
    await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="def456")
    rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
    assert rows == []
    rows, _ = await list_project_shapes(owner, pid, path="src/new.py", include_vanished=True)
    assert rows[0].vanished_at is not None


@pytest.mark.integration
async def test_recent_pulls_reads_the_usage_stream(seeded):
    """The "actually pulled it" half is the PULLED usage event, inside the
    window; a surfacing alone is not a pull."""
    from datetime import datetime, timedelta, timezone

    from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
    from scribe.services.shape_ledger import recent_pulls

    owner, sid = seeded["owner"], seeded["snippet"]
    now = datetime.now(timezone.utc)
    async with async_session() as s:
        s.add_all([
            NoteUsageEvent(user_id=owner, note_id=sid, event=PULLED, source="mcp_get_snippet"),
            NoteUsageEvent(user_id=owner, note_id=sid + 1000, event=SURFACED, source="auto_inject"),
            NoteUsageEvent(user_id=owner, note_id=sid + 2000, event=PULLED,
                           source="mcp_get_snippet", created_at=now - timedelta(days=2)),
        ])
        await s.commit()
    pulls = await recent_pulls(owner)
    assert sid in pulls
    assert sid + 1000 not in pulls
    assert sid + 2000 not in pulls


# --- #2792: the mechanical proposer against real rows -------------------------


def _quiet_semantic():
    """The semantic basis needs the embedder; these tests prove the other
    bases and the bookkeeping, so it answers "nothing" here."""
    from unittest.mock import AsyncMock, patch

    from scribe.services import shape_ledger
    return patch.object(shape_ledger, "_semantic_canon", AsyncMock(return_value=None))


def _defs(*items):
    """ArchiveShape-like records: (path, kind, name, signature, body_sha, body)."""
    import hashlib
    out = []
    for path, kind, name, signature, body in items:
        sha = hashlib.sha1(" ".join(body.split()).encode()).hexdigest()[:16]
        out.append((path, kind, name, signature, sha, body))
    return out


@pytest.mark.integration
async def test_proposer_proposes_and_confirm_classifies(seeded):
    """Bodies in hand, the proposer records proposals on unclassified rows —
    symbol (a second `factory` elsewhere), reference (a call site), and
    nothing for the unrelated — skips rows whose content it already judged,
    and a scoped confirm turns proposals into agent instances while a
    classify on another retires its proposal."""
    from scribe.services.shape_ledger import (
        confirm_proposals, propose_for_repo,
    )

    owner, other, pid, sid = (
        seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
    )
    defs = _defs(
        ("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n    app = factory()\n    return app"),
        ("src/app.py", "sym", "Config", "class Config:", "class Config:\n    debug = False"),
        ("src/util.py", "sym", "helper", "def helper(x):", "def helper(x):\n    return x"),
        ("src/dup.py", "sym", "factory", "def factory():", "def factory():\n    return 1"),
        ("web/button.css", "css", "btn", ".btn {", ".btn {\n  color: red;\n}"),
    )
    await sync_repo_shapes(pid, REPO, defs, seen_marker="main")

    with _quiet_semantic():
        stats = await propose_for_repo(owner, pid, REPO, defs)
    assert stats == {"examined": 5, "proposed": 2, "semantic_checked": 2}
    rows, total = await list_project_shapes(owner, pid, proposal="canon")
    by_symbol = {r.symbol: r for r in rows}
    assert total == 2
    assert by_symbol["factory"].proposal == {"basis": "symbol", "score": 1.0, "snippet_id": sid}
    assert by_symbol["make_app"].proposal == {"basis": "reference", "score": 0.9, "snippet_id": sid}
    rows, _ = await list_project_shapes(owner, pid, proposal="reference")
    assert [r.symbol for r in rows] == ["make_app"]

    # Same content again → nothing re-examined (the semantic cap would
    # otherwise be spent on the same rows every refresh). A cap that leaves
    # rows unreached leaves them UNexamined, so the next refresh gets them.
    with _quiet_semantic():
        assert (await propose_for_repo(owner, pid, REPO, defs))["examined"] == 0
        await classify_shapes(owner, pid, [
            {"path": "src/util.py", "symbol": "helper", "status": "unclassified"},
        ])
        assert (await propose_for_repo(owner, pid, REPO, defs, semantic_cap=0))["semantic_checked"] == 0
        assert (await propose_for_repo(owner, pid, REPO, defs))["examined"] == 1

    # Outsider can't confirm; the owner confirms by snippet, scoped.
    with pytest.raises(ValueError):
        await confirm_proposals(other, pid, snippet_id=sid)
    assert await confirm_proposals(owner, pid, basis="symbol") == {"confirmed": 1}
    rows, _ = await list_project_shapes(owner, pid, snippet_id=sid)
    factory = next(r for r in rows if r.symbol == "factory")
    assert factory.status == "instance" and factory.classified_by == "agent"
    assert factory.reason == "confirmed symbol proposal (1.00)"
    assert factory.proposal is None

    # A judgment on a proposed row retires the proposal; withdrawing a
    # judgment forgets the examination so the next pass proposes afresh.
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "exempt", "reason": "bootstrap"},
    ])
    rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
    make_app = next(r for r in rows if r.symbol == "make_app")
    assert make_app.status == "exempt" and make_app.proposal is None
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "unclassified"},
    ])
    with _quiet_semantic():
        assert (await propose_for_repo(owner, pid, REPO, defs))["proposed"] == 1
    rows, _ = await list_project_shapes(owner, pid, proposal="canon")
    assert [r.symbol for r in rows] == ["make_app"]


@pytest.mark.integration
async def test_derive_groups_land_on_rows_and_in_the_summary(seeded):
    """Shapes with no canon hit that repeat — identical bodies in two files,
    the same name in three — carry a derive proposal, and the readout ranks
    the families. A canon proposal keeps a row out of any derive group."""
    from scribe.services.shape_ledger import (
        apply_derive_groups, live_rows, propose_for_repo, proposal_summary,
    )

    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    defs = _defs(
        ("a/one.py", "sym", "slug", "def slug(t):", "def slug(t):\n    return t.lower()"),
        ("a/two.py", "sym", "slug", "def slug(t):", "def slug(t):\n    return t.lower()"),
        ("b/x.css", "css", "card", ".card {", ".card { padding: 1px }"),
        ("b/y.css", "css", "card", ".card {", ".card { padding: 2px }"),
        ("b/z.css", "css", "card", ".card {", ".card { padding: 3px }"),
        ("c/only.py", "sym", "alone", "def alone():", "def alone():\n    return 0"),
        ("c/use.py", "sym", "boot", "def boot():", "def boot():\n    return factory()"),
    )
    await sync_repo_shapes(pid, REPO, defs, seen_marker="main")
    with _quiet_semantic():
        await propose_for_repo(owner, pid, REPO, defs)
    assert await apply_derive_groups(pid) == 5

    rows, total = await list_project_shapes(owner, pid, proposal="derive")
    assert total == 5
    groups = {(r.path, r.symbol): r.proposal for r in rows}
    assert groups[("a/one.py", "slug")]["group"] == groups[("a/two.py", "slug")]["group"]
    assert groups[("a/one.py", "slug")]["group"].startswith("dup:")
    assert groups[("b/x.css", "card")] == {"basis": "derive", "score": 3.0, "group": "name:css:card"}
    rows, _ = await list_project_shapes(owner, pid, proposal="any")
    assert {r.symbol for r in rows} == {"slug", "card", "boot"}   # boot: reference proposal

    summary = proposal_summary(await live_rows(pid))
    assert summary["proposed"] == 1
    # #2872: the body-identical group (a real copy) outranks the bigger name
    # group (usually convention), even at size 2 vs 3.
    order = [g["group"] for g in summary["derive_groups"]]
    assert order[0].startswith("dup:") and order[1] == "name:css:card"
    assert summary["derive_groups"][0]["files"] == 2
    assert summary["derive_groups"][0]["label"] == "slug (identical body)"
    assert summary["derive_groups"][1]["label"] == ".card"
    assert summary["derive_groups"][0]["size"] == 2 and summary["derive_groups"][1]["size"] == 3

    # One of the css copies gets judged → the group shrinks on the next pass.
    await classify_shapes(owner, pid, [
        {"path": "b/z.css", "symbol": "card", "status": "exempt", "reason": "print sheet"},
    ])
    await apply_derive_groups(pid)
    rows, _ = await list_project_shapes(owner, pid, proposal="derive")
    assert {r.symbol for r in rows} == {"slug"}                   # 2 files < the name floor


# --- #2793: the divergence readout against real rows -------------------------


@pytest.mark.integration
async def test_a_second_confirm_dialog_is_detected_and_named(seeded):
    """The milestone's acceptance case. A directory where one canon dominates
    the judged siblings (a confirm helper with four instance call sites);
    after a previous refresh, a new shape lands there that the proposer does
    not match to the canon — it is flagged `diverges_from` the canon, the
    readout names it, and the in-band check names it at write time. A
    judgment clears the flag; a shape proposed AS the canon is not flagged."""
    from datetime import datetime, timedelta, timezone

    from scribe.services.shape_ledger import (
        divergence_summary, flag_divergence, live_rows, propose_for_repo,
        write_time_divergence,
    )

    from scribe.services import snippets as snippets_svc

    owner, pid = seeded["owner"], seeded["pid"]
    # The confirm helper is TS canon: since #2871 a sym basis only proposes
    # within the shape's language family, so the fixture's Python snippet
    # says nothing about these Vue bodies — the canon must be one of theirs.
    canon = await snippets_svc.create_snippet(
        owner, name="cls_confirm_factory",
        code="export async function factory(): Promise<boolean> {\n  return true;\n}\n",
        language="typescript", repo="Widget",
        path="frontend/src/composables/useConfirm.ts", symbol="factory",
        project_id=pid,
    )
    sid = int(canon.id)
    comp = "frontend/src/components"
    base = _defs(
        *[(f"{comp}/{n}.vue", "sym", f"on{n}", f"async function on{n}() {{",
           f"async function on{n}() {{\n  const ok = await factory();\n  if (!ok) return;\n}}")
          for n in ("Trash", "Delete", "Remove", "Restore")],
    )
    await sync_repo_shapes(pid, REPO, base, seen_marker="aaa111")
    await classify_shapes(owner, pid, [
        {"path": f"{comp}/{n}.vue", "symbol": f"on{n}", "status": "instance", "snippet_id": sid}
        for n in ("Trash", "Delete", "Remove", "Restore")
    ], via="audit")
    previous = datetime.now(timezone.utc)

    # Button B: a hand-rolled confirm that never touches the canon, plus a
    # proper new instance (references the canon → the proposer claims it).
    later = base + _defs(
        (f"{comp}/Danger.vue", "sym", "confirmDanger", "function confirmDanger() {",
         "function confirmDanger() {\n  return window.confirm('Really?');\n}"),
        (f"{comp}/Proper.vue", "sym", "onPurge", "async function onPurge() {",
         "async function onPurge() {\n  const ok = await factory();\n  if (!ok) return;\n}"),
    )
    await sync_repo_shapes(pid, REPO, later, seen_marker="bbb222")
    with _quiet_semantic():
        await propose_for_repo(owner, pid, REPO, later)
    assert await flag_divergence(pid, since=None) == 0          # a first seed flags nothing
    assert await flag_divergence(pid, since=previous - timedelta(seconds=1)) == 1

    rows, total = await list_project_shapes(owner, pid, flag="divergence")
    assert total == 1
    assert rows[0].symbol == "confirmDanger" and rows[0].diverges_from == sid
    summary = divergence_summary(await live_rows(pid))
    assert summary["divergent"] == 1
    assert summary["divergence"][0]["symbol"] == "confirmDanger"
    assert summary["divergence"][0]["canon_snippet_id"] == sid

    # In-band: the hook names the shape at write time → the check names the canon.
    named = await write_time_divergence(
        pid, f"{comp}/Danger.vue", [("sym", "confirmDanger")], stamped=[]
    )
    assert named == [{"symbol": "confirmDanger", "kind": "sym", "canon_snippet_id": sid,
                      "instances": 4, "judged": 4}]
    # ...but an already-judged shape, or one just stamped as the canon's
    # instance, is not re-litigated.
    assert await write_time_divergence(pid, f"{comp}/Trash.vue", [("sym", "onTrash")], stamped=[]) == []
    assert await write_time_divergence(
        pid, f"{comp}/New.vue", [("sym", "onNew")],
        stamped=[{"symbol": "onNew", "kind": "sym", "snippet_id": sid}],
    ) == []
    # A directory with no dominant canon is silent.
    assert await write_time_divergence(pid, "src/other.py", [("sym", "thing")], stamped=[]) == []

    # The judgment answers the question and clears the flag.
    await classify_shapes(owner, pid, [
        {"path": f"{comp}/Danger.vue", "symbol": "confirmDanger", "status": "variant",
         "snippet_id": sid, "reason": "native confirm is fine in the dev-only panel"},
    ])
    rows, total = await list_project_shapes(owner, pid, flag="divergence")
    assert total == 0


@pytest.mark.integration
async def test_history_records_what_was_used_when_and_drift_asks_for_a_recheck(seeded):
    from scribe.services.shape_ledger import shape_history

    owner, other, pid, sid = (
        seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
    )
    v1 = _defs(("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n    return factory()"))
    await sync_repo_shapes(pid, REPO, v1, seen_marker="c1")
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance", "snippet_id": sid},
    ])
    # The body moves under the judgment → drifted + recheck; re-judging clears it.
    v2 = _defs(("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n    return factory(debug=True)"))
    await sync_repo_shapes(pid, REPO, v2, seen_marker="c2")
    rows, total = await list_project_shapes(owner, pid, flag="recheck")
    assert total == 1 and rows[0].symbol == "make_app" and rows[0].status == "instance"
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "variant", "snippet_id": sid,
         "reason": "debug flag is deliberate here"},
    ])
    rows, total = await list_project_shapes(owner, pid, flag="recheck")
    assert total == 0
    # Then it vanishes from the tree.
    await sync_repo_shapes(pid, REPO, [], seen_marker="c3")

    history = await shape_history(owner, pid, "src/app.py", symbol="make_app")
    shape = history["shapes"][0]
    assert shape["status"] == "variant" and shape["vanished_at"] is not None
    # The seeded fixture synced this row first (marker "main"); v1/v2 are
    # later sightings — first_seen keeps the first.
    assert shape["first_seen_commit"] == "main" and shape["last_seen_commit"] == "c2"
    timeline = [(e["event"], e["status"], e["snippet_id"], e["commit"]) for e in history["events"]]
    assert timeline == [
        ("classified", "instance", sid, "c1"),
        ("drifted", "instance", sid, "c2"),
        ("classified", "variant", sid, "c2"),
        ("vanished", "variant", sid, "c2"),
    ]
    assert history["events"][2]["reason"] == "debug flag is deliberate here"
    assert history["events"][0]["classified_by"] == "agent"
    # Directory-wide read works (the empty sync also vanished the seeded
    # Config and helper rows under src/ — two more events); an outsider
    # reads nothing.
    assert len((await shape_history(owner, pid, "src"))["events"]) == 6
    assert await shape_history(other, pid, "src/app.py") == {}