FabledScribe/tests/test_integration_shape_classify.py

"""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_uses_edges_are_the_consumer_map(seeded):
    """#2870: a shape keeps ONE snippet_id (what it is) and any number of
    uses edges (what it calls); the snippet's consumer map lists them,
    list_shapes(uses=N) finds them, and a sweep can write them."""
    from scribe.services import snippets as snippets_svc
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    helper = await snippets_svc.create_snippet(
        owner, name="cls_hash_helper", code="def hash_token(raw):\n    return raw\n",
        language="python", repo="Widget", path="src/hash.py", symbol="hash_token",
        project_id=pid,
    )
    hid = int(helper.id)
    out = await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid, "uses": [hid]},
    ], via="audit")
    assert out["classified"] == 1
    rows, total = await list_project_shapes(owner, pid, uses=hid)
    assert total == 1 and rows[0].symbol == "make_app" and rows[0].snippet_id == sid
    consumers = await snippet_consumers(owner, hid)
    assert consumers["instances"] == [] and len(consumers["uses"]) == 1
    assert consumers["uses"][0]["symbol"] == "make_app" and consumers["uses"][0]["basis"] == "audit"
    # A sweep writes uses too; an unknown snippet in uses applies nothing.
    out = await classify_shapes_where(
        owner, pid, path="src/util.py", status="exempt", reason="local", uses=[hid],
    )
    assert out["classified"] == 1
    assert (await list_project_shapes(owner, pid, uses=hid))[1] == 2
    # The proposer writes uses edges for JUDGED rows too: Config (exempt)
    # names hash_token in its body → an edge, no proposal.
    from scribe.services.shape_ledger import propose_for_repo
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "Config", "status": "exempt", "reason": "settings"},
    ])
    defs = _defs(("src/app.py", "sym", "Config", "class Config:", "class Config:\n    token = hash_token(raw)\n"))
    with _quiet_semantic():
        await propose_for_repo(owner, pid, REPO, defs)
    rows, total = await list_project_shapes(owner, pid, uses=hid)
    assert total == 3 and {r.symbol for r in rows} >= {"Config"}
    cfg = next(r for r in rows if r.symbol == "Config")
    assert cfg.status == "exempt" and cfg.proposal is None
    with pytest.raises(ValueError):
        await classify_shapes(owner, pid, [
            {"path": "src/app.py", "symbol": "Config", "status": "exempt", "reason": "x", "uses": [999999]},
        ])


@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": [], "uses": []}


@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_evidence_is_a_proposal_never_a_verdict(seeded):
    """Milestone 439. Pulled + referenced → every named shape of the
    snippet's kind CARRIES the snippet as a proposal (basis "reference"),
    and its status is untouched: the agent that wrote it says what it is.
    A judgment is never touched, not even by a proposal. The outsider
    writes nothing (write-gated like every other ledger write). The uses
    edge — a fact about the code — is still recorded."""
    from datetime import datetime, timezone

    from scribe.services.shape_ledger import suggest_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 suggest_write_path_instances(
        other, pid, path="src/app.py", shapes=[("sym", "make_app")],
        code=code, pulled=pulled,
    ) == []

    suggested = await suggest_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 suggested} == {"make_app", "Config"}   # css skipped: no css canon
    rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
    by_symbol = {r.symbol: r for r in rows}
    for name in ("make_app", "Config"):
        row = by_symbol[name]
        assert row.status == "unclassified" and row.classified_by is None, name
        assert (row.proposed_snippet_id, row.proposal_basis) == (sid, "reference"), name
    # The by-name reference is still recorded as a call-site fact.
    users, _ = await list_project_shapes(owner, pid, uses=sid)
    assert {"make_app", "Config"} <= {r.symbol for r in users}

    # A judgment stands: the next suggestion leaves the judged row alone.
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "exempt",
         "reason": "the app factory is its own thing"},
    ])
    again = await suggest_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")
    assert {r.symbol: r.status for r in rows}["make_app"] == "exempt"

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

    # And nothing the hook does writes classified_by="hook" any more.
    async with async_session() as s:
        hook_rows = (await s.execute(select(CodeShape).where(
            CodeShape.project_id == pid, CodeShape.classified_by == "hook",
        ))).scalars().all()
    assert hook_rows == []


@pytest.mark.integration
async def test_a_brand_new_shape_gets_a_provisional_row_carrying_the_suggestion(seeded):
    """The shape being written right now has no ledger row yet. With the
    hook's repo key it gets a provisional, UNCLASSIFIED one — seen markers
    empty — carrying the suggestion to the end-of-turn question; the sync
    confirms it or stamps it vanished. Without a repo key only existing rows
    carry a suggestion."""
    from datetime import datetime, timezone

    from scribe.services.shape_ledger import suggest_write_path_instances, unjudged_shapes

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

    assert await suggest_write_path_instances(
        owner, pid, path="src/new.py", shapes=[("sym", "build")],
        code=code, pulled=pulled,                              # no repo_key
    ) == []
    suggested = await suggest_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 suggested] == ["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 == "unclassified" and row.classified_by is None
        assert row.proposed_snippet_id == sid
        # "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 end-of-turn question shows the suggestion as evidence.
    asked = await unjudged_shapes(pid, [("src/new.py", "sym", "build")], repo_key=REPO)
    assert asked[0]["evidence"]["looks_like"]["snippet_id"] == sid

    # The sync sees the shape in the tree → confirmed, still unjudged.
    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 == "unclassified" 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
    # but stays a family: a class in two files is already a recipe living in
    # two places (css name floor 2, note 2917). Judge the second and it's gone.
    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", "card"}
    assert {r.path for r in rows if r.symbol == "card"} == {"b/x.css", "b/y.css"}
    await classify_shapes(owner, pid, [
        {"path": "b/y.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"}                   # 1 file < the css name floor


@pytest.mark.integration
async def test_write_time_derive_names_the_family_or_the_canon_for_a_name(seeded):
    """#2900: against real rows — a name in a family → the family (other
    files, count; for CSS a NAME family, never a body one — note 2917); a
    name whose canonical row lives elsewhere → that canon; a judged row at
    the path, the canon's own file, or an unknown name → silence."""
    from scribe.services.shape_ledger import apply_derive_groups, write_time_derive

    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    defs = _defs(
        ("v/A.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
        ("v/B.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
        ("v/C.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
        ("src/factory.py", "sym", "factory", "def factory():", "def factory():\n    return 1"),
    )
    await sync_repo_shapes(pid, REPO, defs, seen_marker="m1")
    await classify_shapes(owner, pid, [
        {"path": "src/factory.py", "symbol": "factory", "status": "canonical", "snippet_id": sid},
    ])
    assert await apply_derive_groups(pid) >= 3

    # A 4th copy about to be written → the family, naming the other files.
    out = await write_time_derive(pid, "v/D.vue", [("css", "log-empty"), ("css", "unknown")])
    assert len(out) == 1 and out[0]["symbol"] == "log-empty" and out[0]["kind"] == "css"
    fam = out[0]["family"]
    # Three identical bodies, and still a NAME family: CSS never groups by body.
    assert fam["identical"] is False and fam["label"] == ".log-empty"
    assert fam["files"] == ["v/A.vue", "v/B.vue", "v/C.vue"] and fam["file_count"] == 3
    assert out[0]["key"] == fam["group"] == "name:css:log-empty"
    # Editing one existing member still names the OTHER members.
    out = await write_time_derive(pid, "v/A.vue", [("css", "log-empty")])
    assert out[0]["family"]["files"] == ["v/B.vue", "v/C.vue"] and out[0]["family"]["size"] == 3
    # The canon's name elsewhere → the canon; in the canon's own file → silence.
    out = await write_time_derive(pid, "src/other.py", [("sym", "factory")])
    assert out == [{"symbol": "factory", "kind": "sym", "key": f"canon:{sid}",
                    "canon": {"snippet_id": sid, "path": "src/factory.py", "label": "factory"}}]
    assert await write_time_derive(pid, "src/factory.py", [("sym", "factory")]) == []
    # A judged row at the path is not re-litigated.
    await classify_shapes(owner, pid, [
        {"path": "v/B.vue", "symbol": "log-empty", "status": "exempt", "reason": "print sheet"},
    ])
    assert await write_time_derive(pid, "v/B.vue", [("css", "log-empty")]) == []


@pytest.mark.integration
async def test_derive_new_names_the_copy_that_joined_a_family_since_the_stamp(seeded):
    """#2899: the first sync seeds one `slug`; a later sync adds an identical
    copy. Against the stamp between them, derive_new counts ONLY the
    newcomer — the drift since the last refresh, not the whole family."""
    from datetime import datetime, timezone

    from scribe.services.shape_ledger import (
        apply_derive_groups, derive_new_summary, live_rows,
    )

    owner, pid = seeded["owner"], seeded["pid"]
    first = _defs(
        ("a/one.py", "sym", "slug", "def slug(t):", "def slug(t):\n    return t.lower()"),
    )
    await sync_repo_shapes(pid, REPO, first, seen_marker="m1")
    stamp = datetime.now(timezone.utc)
    second = _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()"),
    )
    await sync_repo_shapes(pid, REPO, second, seen_marker="m2")
    assert await apply_derive_groups(pid) == 2

    rows = await live_rows(pid)
    out = derive_new_summary(rows, since=stamp)
    assert out["count"] == 1
    assert out["examples"][0]["path"] == "a/two.py"
    assert out["examples"][0]["label"] == "slug"
    assert out["examples"][0]["group"].startswith("dup:")
    assert derive_new_summary(rows, since=None)["count"] == 0


@pytest.mark.integration
async def test_transition_and_prefix_references_clear_the_unused_css_flag(seeded):
    """#2970 end to end: the two class forms a template never spells out —
    a transition `name=` and a concatenated name — travel from the real
    extractor through resolution into the flag, so rules they reach stop
    being reported as unused. Only a rule nothing can reach stays listed."""
    from scribe.services.coverage import class_references
    from scribe.services.shape_ledger import sync_repo_consumers, used_by_map, live_rows

    pid, owner = seeded["pid"], seeded["owner"]
    defs = _defs(
        ("assets/anim.css", "css", "toast-enter-active",
         ".toast-enter-active {", ".toast-enter-active { opacity: 0 }"),
        ("assets/anim.css", "css", "toast-leave-to",
         ".toast-leave-to {", ".toast-leave-to { opacity: 0 }"),
        ("assets/state.css", "css", "status-done",
         ".status-done {", ".status-done { color: green }"),
        ("assets/state.css", "css", "status-todo",
         ".status-todo {", ".status-todo { color: grey }"),
        ("assets/state.css", "css", "really-dead",
         ".really-dead {", ".really-dead { color: red }"),
    )
    await sync_repo_shapes(pid, REPO, defs, seen_marker="t1")

    refs = {
        # Vue applies the toast-* classes itself; the markup names none of them.
        "v/Toast.vue": class_references(
            "v/Toast.vue", '<transition-group name="toast"><li /></transition-group>'
        ),
        # The board builds its status class; which one is unknowable.
        "v/Board.vue": class_references("v/Board.vue", '<b :class="`status-${s}`" />'),
    }
    assert "toast-enter-active" in refs["v/Toast.vue"]
    assert refs["v/Board.vue"] == {"status-*": 1}

    await sync_repo_consumers(pid, REPO, refs)
    live = [r for r in await live_rows(pid) if r.kind == "css"]
    used = await used_by_map(live)
    by = {(r.path, r.symbol): used[r.id]["count"] for r in live}
    assert by[("assets/anim.css", "toast-enter-active")] == 1
    assert by[("assets/anim.css", "toast-leave-to")] == 1
    # the prefix credits BOTH candidates — the template does not say which
    assert by[("assets/state.css", "status-done")] == 1
    assert by[("assets/state.css", "status-todo")] == 1
    assert by[("assets/state.css", "really-dead")] == 0

    unused, n = await list_project_shapes(owner, pid, flag="unused-css")
    assert n == 1 and [(r.path, r.symbol) for r in unused] == [
        ("assets/state.css", "really-dead")
    ]


@pytest.mark.integration
async def test_consumer_map_syncs_edges_from_template_references(seeded):
    """Milestone 302: the consumer edges follow the archive — own-file
    resolution for a scoped class, fan-out to the shared sheet for a class a
    template does not define, counts refreshed and stale edges removed on
    the next sync, and a vanished row's edges gone with it."""
    from scribe.services.shape_ledger import consumers_of, live_rows, sync_repo_consumers

    pid = seeded["pid"]
    defs = _defs(
        ("v/A.vue", "css", "error-msg", ".error-msg {", ".error-msg { color: red }"),
        ("v/B.vue", "css", "error-msg", ".error-msg {", ".error-msg { color: blue }"),
        ("assets/components.css", "css", "btn-primary", ".btn-primary {", ".btn-primary { x: 1 }"),
        ("assets/orphan.css", "css", "orphan", ".orphan {", ".orphan { y: 2 }"),   # no template names it
    )
    await sync_repo_shapes(pid, REPO, defs, seen_marker="m1")
    refs = {
        "v/A.vue": {"error-msg": 2, "btn-primary": 1},
        "v/B.vue": {"error-msg": 1},
        "v/C.vue": {"error-msg": 1, "btn-primary": 4},
    }
    # A and B consume their OWN error-msg; C defines none, so its use fans
    # out to both rows; btn-primary resolves to the shared sheet from A and C.
    assert await sync_repo_consumers(pid, REPO, refs) == 6
    rows = {(r.path, r.symbol): r.id for r in await live_rows(pid) if r.kind == "css"}
    edges = await consumers_of(rows.values())
    view = {(p, s): [(e.path, e.count) for e in edges.get(i, [])] for (p, s), i in rows.items()}
    assert view[("v/A.vue", "error-msg")] == [("v/A.vue", 2), ("v/C.vue", 1)]
    assert view[("v/B.vue", "error-msg")] == [("v/B.vue", 1), ("v/C.vue", 1)]
    assert view[("assets/components.css", "btn-primary")] == [("v/A.vue", 1), ("v/C.vue", 4)]
    assert view[("assets/orphan.css", "orphan")] == []

    # The next tree: C stops using error-msg, A uses btn-primary twice now.
    refs2 = {"v/A.vue": {"error-msg": 2, "btn-primary": 2}, "v/B.vue": {"error-msg": 1}}
    assert await sync_repo_consumers(pid, REPO, refs2) == 3
    edges = await consumers_of(rows.values())
    assert [(e.path, e.count) for e in edges[rows[("v/B.vue", "error-msg")]]] == [("v/B.vue", 1)]
    assert [(e.path, e.count) for e in edges[rows[("assets/components.css", "btn-primary")]]] == [("v/A.vue", 2)]

    # The readout side: used_by per css row, the unused-css flag, and the
    # family's consumers on the write-path check.
    from scribe.services.shape_ledger import (
        apply_derive_groups, used_by_map, write_time_derive,
    )
    owner = seeded["owner"]
    live = [r for r in await live_rows(pid) if r.kind == "css"]
    used = await used_by_map(live)
    assert used[rows[("v/B.vue", "error-msg")]] == {"count": 1, "paths": ["v/B.vue"]}
    assert used[rows[("assets/orphan.css", "orphan")]] == {"count": 0, "paths": []}
    unused, n = await list_project_shapes(owner, pid, flag="unused-css")
    assert n == 1 and [(r.path, r.symbol) for r in unused] == [("assets/orphan.css", "orphan")]
    await apply_derive_groups(pid)
    out = await write_time_derive(pid, "v/New.vue", [("css", "error-msg")])
    assert out and out[0]["family"]["consumers"] == {"count": 2, "paths": ["v/A.vue", "v/B.vue"]}

    # B's rule vanishes from the tree → its edges go with the pass.
    await sync_repo_shapes(pid, REPO, [d for d in defs if d[0] != "v/B.vue"], seen_marker="m2")
    await sync_repo_consumers(pid, REPO, refs2)
    edges = await consumers_of(rows.values())
    assert rows[("v/B.vue", "error-msg")] not in edges


# --- #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")], suggested=[]
    )
    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")], suggested=[]) == []
    assert await write_time_divergence(
        pid, f"{comp}/New.vue", [("sym", "onNew")],
        suggested=[{"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")], suggested=[]) == []

    # 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_weak_stamps_elect_no_canon_and_their_flags_are_withdrawn(seeded):
    """#4608, the poisoned-ledger case as it stood on two projects: a
    directory's "dominant canon" was elected entirely by hook stamps written
    at 0.68-0.72, before the 0.80 floor, so every new shape there was told it
    diverged from a YAML CI snippet. The same fixture as the acceptance case
    above, with one difference — the four instances are weak hook stamps
    rather than an audit — and the prompt must fall silent: at refresh (the
    standing flag is withdrawn) and at write time. The stamps themselves are
    left exactly as they were, for `stamps_to_review` to put before a judge."""
    from datetime import datetime, timedelta, timezone

    from sqlalchemy import update

    from scribe.services import snippets as snippets_svc
    from scribe.services.shape_ledger import (
        _RESEMBLE_REASON, flag_divergence, propose_for_repo, write_time_divergence,
    )

    owner, pid = seeded["owner"], seeded["pid"]
    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"
    names = ("Trash", "Delete", "Remove", "Restore")
    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 names],
    )
    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 names
    ], via="audit")
    previous = datetime.now(timezone.utc)
    later = base + _defs(
        (f"{comp}/Danger.vue", "sym", "confirmDanger", "function confirmDanger() {",
         "function confirmDanger() {\n  return window.confirm('Really?');\n}"),
    )
    await sync_repo_shapes(pid, REPO, later, seen_marker="bbb222")
    with _quiet_semantic():
        await propose_for_repo(owner, pid, REPO, later)
    since = previous - timedelta(seconds=1)
    # Judged by an audit, the canon dominates and the new shape is flagged.
    assert await flag_divergence(pid, since=since) == 1

    # Now make the four what the poisoned ledgers held: unattended stamps
    # below today's floor.
    weak = _RESEMBLE_REASON.format(sid=sid, score=0.72)
    async with async_session() as session:
        await session.execute(
            update(CodeShape)
            .where(CodeShape.project_id == pid, CodeShape.snippet_id == sid)
            .values(classified_by="hook", reason=weak)
        )
        await session.commit()

    assert await flag_divergence(pid, since=since) == 0
    rows, total = await list_project_shapes(owner, pid, flag="divergence")
    assert total == 0, "a flag whose canon no longer dominates is withdrawn"
    assert await write_time_divergence(
        pid, f"{comp}/Danger.vue", [("sym", "confirmDanger")], suggested=[]
    ) == []
    # Nothing was un-judged: the stamps stand until someone reads them.
    async with async_session() as session:
        kept = (await session.execute(
            select(CodeShape).where(CodeShape.project_id == pid, CodeShape.snippet_id == sid)
        )).scalars().all()
    assert len(kept) == 4
    assert all(r.status == "instance" and r.classified_by == "hook" for r in kept)


@pytest.mark.integration
async def test_a_semantic_miss_does_not_silence_the_prompt(seeded):
    """#4208, reversed on measurement: a semantic miss is not evidence.

    THE FIXTURE IS THE ACCEPTANCE CASE ABOVE, DELIBERATELY. A hand-rolled
    `confirmDanger` beside an async confirm canon is structurally identical to
    a registry helper beside an async service canon, and the arm's miss was
    meant to tell them apart. Measured live, true instances of a canon rarely
    clear the arm's floor either, so a miss fires on both — silencing on
    it silenced #2793's own case. Here the arm has read the body, found
    nothing, and the prompt is still RAISED.

    Asserted on the row, not a flag count: `since` sits a second early to
    absorb clock skew, which also admits other fixture rows as "new".
    """
    from datetime import datetime, timedelta, timezone
    from unittest.mock import AsyncMock, patch

    from scribe.services import shape_ledger
    from scribe.services import snippets as snippets_svc
    from scribe.services.shape_ledger import flag_divergence, live_rows, propose_for_repo

    owner, pid = seeded["owner"], seeded["pid"]
    canon = await snippets_svc.create_snippet(
        owner, name="cls_confirm_factory_meaning",
        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)

    later = base + _defs(
        (f"{comp}/Danger.vue", "sym", "confirmDanger", "function confirmDanger() {",
         "function confirmDanger() {\n  return window.confirm('Really?');\n}"),
    )
    await sync_repo_shapes(pid, REPO, later, seen_marker="bbb222")

    miss = AsyncMock(return_value=None)
    with patch.object(shape_ledger, "_semantic_canon", miss):
        await propose_for_repo(owner, pid, REPO, later)
    assert miss.await_count >= 1, "the arm must actually have read a body"

    await flag_divergence(pid, since=previous - timedelta(seconds=1))
    danger = next(r for r in await live_rows(pid) if r.symbol == "confirmDanger")
    assert danger.proposed_snippet_id is None
    assert danger.diverges_from == sid, "a miss at the arm's floor is 'cannot tell'"


async def _two_generations(seeded, tag: str):
    """A canon-dense directory, an OLD unjudged helper, then one NEW shape.

    The same density as the acceptance fixture above, with an extra unjudged
    row from the first sync so there is an old row competing with the new one
    for the capped semantic pass.
    """
    from datetime import datetime, timezone

    from scribe.services import snippets as snippets_svc

    owner, pid = seeded["owner"], seeded["pid"]
    canon = await snippets_svc.create_snippet(
        owner, name=f"cls_confirm_factory_{tag}",
        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")],
        (f"{comp}/Old.vue", "sym", "oldHelper", "function oldHelper() {",
         "function oldHelper() {\n  return document.title.length > 0;\n}"),
    )
    await sync_repo_shapes(pid, REPO, base, seen_marker=f"{tag}-1")
    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)
    later = base + _defs(
        (f"{comp}/Danger.vue", "sym", "confirmDanger", "function confirmDanger() {",
         "function confirmDanger() {\n  return window.confirm('Really?');\n}"),
    )
    await sync_repo_shapes(pid, REPO, later, seen_marker=f"{tag}-2")
    return owner, pid, later, previous


@pytest.mark.integration
async def test_the_capped_semantic_pass_reaches_the_new_shape_first(seeded):
    """Measured live: after the #4208 version bump, 1,533 rows competed for 150
    semantic checks in row order and the six new shapes — the only ones the
    divergence check acts on — were all left unread and flagged. With room for
    ONE check, the new shape must be the one that gets it."""
    from unittest.mock import AsyncMock, patch

    from scribe.services import shape_ledger
    from scribe.services.shape_ledger import live_rows, propose_for_repo

    owner, pid, later, _previous = await _two_generations(seeded, "cap")
    with patch.object(shape_ledger, "_semantic_canon", AsyncMock(return_value=None)):
        stats = await propose_for_repo(owner, pid, REPO, later, semantic_cap=1)
    assert stats["semantic_checked"] == 1

    rows = {r.symbol: r for r in await live_rows(pid)}
    assert rows["confirmDanger"].proposed_sha != "", "the new shape got the one check"
    # The old row is left for the next refresh, unexamined — never stamped.
    assert rows["oldHelper"].proposed_sha == ""


@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") == {}


# --- milestone 439: the writer judges what it wrote, before the sync has seen it

@pytest.mark.integration
async def test_a_shape_written_this_turn_is_judged_before_the_sync_sees_it(seeded):
    """The end-of-turn verdict lands on a shape no sync has read yet: with a
    bound `repo` it becomes a provisional row and is judged, the sync that
    then reads the shape keeps the judgment, and a sync that does NOT have it
    yet (the code has not reached the bound ref) keeps it through the vanish
    and the reappearance. Without `repo` the old contract holds."""
    from scribe.services import repo_bindings as repo_bindings_svc

    owner, pid = seeded["owner"], seeded["pid"]
    await repo_bindings_svc.set_binding(owner, REPO, pid)
    verdict = [{"path": "web/src/lib/StatusChip.svelte", "symbol": "tone",
                "kind": "sym", "status": "exempt",
                "reason": "the chip's own colour switch", "reason_code": "pure-helper"}]

    out = await classify_shapes(owner, pid, verdict)
    assert out["unmatched"] == [{"path": "web/src/lib/StatusChip.svelte", "symbol": "tone"}]

    with pytest.raises(ValueError, match="not bound"):
        await classify_shapes(owner, pid, verdict, repo="git.example.com/alice/elsewhere")

    out = await classify_shapes(owner, pid, verdict, repo=REPO)
    assert out == {"classified": 1, "unmatched": [], "provisional": 1}

    # A sync that has not got the code yet: the row vanishes, judgment intact…
    await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="c1")
    # …and the sync that has it brings the row back with the verdict.
    await sync_repo_shapes(
        pid, REPO, SHAPES + [("web/src/lib/StatusChip.svelte", "sym", "tone")], seen_marker="c2",
    )
    rows, _ = await list_project_shapes(owner, pid, path="web/src/lib/StatusChip.svelte")
    assert len(rows) == 1
    row = rows[0]
    assert (row.status, row.classified_by, row.reason_code) == ("exempt", "agent", "pure-helper")
    assert row.vanished_at is None and row.last_seen_commit == "c2"


@pytest.mark.integration
async def test_unjudged_is_what_nobody_read(seeded):
    """The end-of-turn question: a shape with no row, an unclassified one, a
    sync-scoped one and a hook stamp are all asked about; an agent's or an
    audit's verdict is not. Evidence travels; a verdict is never inferred."""
    from sqlalchemy import update

    from scribe.services.shape_ledger import _RESEMBLE_REASON, unjudged_shapes

    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": "src/util.py", "symbol": "helper", "status": "exempt", "reason": "x"},
    ])
    async with async_session() as session:
        await session.execute(
            update(CodeShape)
            .where(CodeShape.project_id == pid, CodeShape.symbol == "helper")
            .values(classified_by="hook", status="instance", snippet_id=sid,
                    reason=_RESEMBLE_REASON.format(sid=sid, score=0.7))
        )
        await session.commit()

    written = [
        ("src/app.py", "sym", "make_app"),      # judged by an agent → not asked
        ("src/app.py", "sym", "Config"),        # unclassified → asked
        ("src/util.py", "sym", "helper"),       # a hook stamp → asked, with its evidence
        ("src/new.py", "sym", "fresh"),         # no row → asked as new
        ("src/new.py", "sym", "fresh"),         # duplicates collapse
    ]
    got = await unjudged_shapes(pid, written, repo_key=REPO)
    assert [(g["symbol"], g["status"]) for g in got] == [
        ("Config", "unclassified"), ("helper", "instance"), ("fresh", "new"),
    ]
    assert got[1]["evidence"]["stamped"] == {"snippet_id": sid, "score": 0.7}
    assert got[2]["evidence"] == {}
    assert await unjudged_shapes(pid, []) == []


@pytest.mark.integration
async def test_a_whole_file_snippet_makes_its_file_row_canonical(seeded):
    """Milestone 439: a snippet recorded at a component's path, with no
    symbol, is the canon for that file row — the claim a whole-file record
    makes and a definition row never could. It still covers no definition
    inside the file."""
    from scribe.services import snippets as snippets_svc
    from scribe.services.coverage import _recorded_locations
    from scribe.services.shape_ledger import mark_canonicals

    owner, pid = seeded["owner"], seeded["pid"]
    await sync_repo_shapes(pid, REPO, SHAPES + [
        ("web/lib/StatusChip.svelte", "file", "StatusChip"),
        ("web/lib/StatusChip.svelte", "css", "chip"),
    ], seen_marker="c1")
    chip = await snippets_svc.create_snippet(
        owner, name="StatusChip — the one status pill",
        code="<span class=\"chip\">{label}</span>", language="svelte",
        repo="Widget", path="web/lib/StatusChip.svelte", project_id=pid,
    )
    await mark_canonicals(pid, await _recorded_locations(owner, pid))
    rows, _ = await list_project_shapes(owner, pid, path="web/lib/StatusChip.svelte")
    by = {(r.kind, r.symbol): r for r in rows}
    assert by[("file", "StatusChip")].status == "canonical"
    assert by[("file", "StatusChip")].snippet_id == int(chip.id)
    assert by[("css", "chip")].status == "unclassified"