FabledGrimoire/src/scribe/services/shape_ledger.py

"""The shape ledger's write side — sync and mechanical marking (#2788).

The coverage walk (services/coverage.py) is the only feed that sees every
shape, so it is the ledger's sync point: each refresh upserts one repo's
extracted shapes — new shapes arrive `unclassified` (THE todo state, note
2786), surviving shapes bump their last-seen marker, vanished shapes get
stamped rather than deleted (history is the point). Classifications survive
recompute by construction: the upsert never touches a judgment, with two
deliberate exceptions —

  - a judgment whose snippet target is gone (SET NULL on snippet deletion)
    is re-filed as unclassified so it rejoins the todo instead of dangling;
  - a MECHANICALLY-stamped canonical row whose snippet location no longer
    covers it falls back to unclassified. Only mechanical stamps self-heal;
    an agent's judgment is never unwound by machinery.

`location_covers` is the one covering predicate — the same doctrine the
recorded-location drift check uses — shared by the sync's canonical marking
and by anything else that must decide whether a recorded location speaks for
an extracted shape.
"""
from __future__ import annotations

from datetime import datetime, timezone

from sqlalchemy import select

from scribe.models import async_session
from scribe.models.code_shape import CodeShape

# Statuses whose meaning requires a snippet target.
_NEEDS_TARGET = ("canonical", "instance", "variant")


def _norm_symbol(kind_or_symbol: str) -> str:
    # CSS shapes and recorded CSS symbols may or may not carry the leading
    # dot; compare without it so ".btn-primary" and "btn-primary" agree.
    return kind_or_symbol.lstrip(".").strip()


def location_covers(loc_path: str, loc_symbol: str, path: str, name: str) -> bool:
    """Does a recorded (path, symbol) location speak for this shape?

    Symbol-less locations never cover a shape — a whole-file record makes no
    claim about any particular definition inside it. Path semantics are the
    drift check's own: exact file, or the recorded path is a directory the
    file lives under.
    """
    if not (loc_symbol or "").strip():
        return False
    if _norm_symbol(loc_symbol) != _norm_symbol(name):
        return False
    if not loc_path:
        # Symbol-only record: the symbol match is all the claim there is.
        return True
    from scribe.services.snippets import _path_touches

    return _path_touches(loc_path, path)


async def sync_repo_shapes(
    project_id: int,
    repo_key: str,
    shapes: list[tuple[str, str, str]],
    *,
    seen_marker: str,
) -> None:
    """Upsert one repo's extracted (path, kind, name) shapes into the ledger.

    ``seen_marker`` is the commit the archive was read at when the forge can
    say, else the ref name — provenance sugar; the row timestamps carry the
    when.
    """
    now = datetime.now(timezone.utc)
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.repo_key == repo_key,
                )
            )
        ).scalars().all()
        by_key = {(r.path, r.symbol, r.kind): r for r in rows}
        seen: set[tuple[str, str, str]] = set()
        for path, kind, name in shapes:
            key = (path, name, kind)
            if key in seen:
                continue
            seen.add(key)
            row = by_key.get(key)
            if row is None:
                session.add(CodeShape(
                    project_id=project_id, repo_key=repo_key,
                    path=path, symbol=name, kind=kind,
                    first_seen_commit=seen_marker, last_seen_commit=seen_marker,
                ))
                continue
            row.last_seen_commit = seen_marker
            # A shape that vanished and came back is live again — the vanish
            # stays visible in history via updated_at, not as a dead flag.
            row.vanished_at = None
            if row.status in _NEEDS_TARGET and row.snippet_id is None:
                row.status = "unclassified"
                row.classified_by = None
                row.classified_at = None
                row.reason = None
        for key, row in by_key.items():
            if key not in seen and row.vanished_at is None:
                row.vanished_at = now
        await session.commit()


async def mark_canonicals(
    project_id: int, recorded: list[tuple[int, str, str]]
) -> None:
    """Stamp snippet reference locations as `canonical` — the one mechanical
    rule that is always safe (the judgment happened when the snippet was
    minted; this row just makes it queryable).

    ``recorded`` is (snippet_note_id, path, symbol) for every live snippet
    location in the project. Touches only rows machinery owns: unclassified
    rows gain the stamp; mechanically-stamped canonicals no longer covered
    fall back to unclassified. Agent judgments are never overwritten.
    """
    usable = [(nid, p, s) for nid, p, s in recorded if (s or "").strip()]
    now = datetime.now(timezone.utc)
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
        for row in rows:
            covering = next(
                (
                    nid for nid, lp, ls in usable
                    if location_covers(lp, ls, row.path, row.symbol)
                ),
                None,
            )
            if covering is not None and row.status == "unclassified":
                row.status = "canonical"
                row.snippet_id = covering
                row.classified_by = "mechanical"
                row.classified_at = now
            elif (
                covering is None
                and row.status == "canonical"
                and row.classified_by == "mechanical"
            ):
                row.status = "unclassified"
                row.snippet_id = None
                row.classified_by = None
                row.classified_at = None
        await session.commit()


async def live_rows(project_id: int) -> list[CodeShape]:
    """Every un-vanished ledger row for a project — the accounting readout's
    input, across ALL its repos (a repo unreachable this refresh still counts;
    accounting is project-wide)."""
    async with async_session() as session:
        return list(
            (
                await session.execute(
                    select(CodeShape).where(
                        CodeShape.project_id == project_id,
                        CodeShape.vanished_at.is_(None),
                    )
                )
            ).scalars().all()
        )


# --- classification (#2789): the judgment write path --------------------------

# Statuses an explicit classification may set. All five: setting a row back to
# `unclassified` is how a judgment is deliberately withdrawn.
_SETTABLE = ("canonical", "instance", "variant", "exempt", "unclassified")

# Who may appear as the classifier on this path. `hook` and `mechanical` are
# server-internal feeds (steps 5-6) — a caller claiming them would launder a
# judgment as machinery.
_CALLER_VIAS = ("agent", "audit", "import")


def validate_classifications(items: list[dict]) -> str | None:
    """The structural error a classification batch would earn, or None.

    Pure and checked BEFORE anything is touched: a batch either applies or
    errors whole — the StrictArgs lesson (#2709), a caller must never learn
    later that half a batch silently happened.
    """
    if not items:
        return "classifications is empty — nothing to apply"
    for i, item in enumerate(items):
        if not isinstance(item, dict):
            return f"classifications[{i}] is not an object"
        path = (item.get("path") or "").strip()
        symbol = (item.get("symbol") or "").strip()
        if not path or not symbol:
            return f"classifications[{i}] needs both path and symbol"
        status = item.get("status") or ""
        if status not in _SETTABLE:
            return (
                f"classifications[{i}] has unknown status {status!r} "
                f"(one of: {', '.join(_SETTABLE)})"
            )
        snippet_id = item.get("snippet_id") or 0
        if status in _NEEDS_TARGET and not snippet_id:
            return (
                f"classifications[{i}]: status {status!r} needs snippet_id — "
                "the snippet this shape is (or departs from)"
            )
        if status in ("variant", "exempt") and not (item.get("reason") or "").strip():
            return (
                f"classifications[{i}]: status {status!r} needs a reason — "
                "the WHY is the record (note 2786)"
            )
    return None


async def classify_shapes(
    user_id: int,
    project_id: int,
    classifications: list[dict],
    *,
    via: str = "agent",
) -> dict:
    """Apply a batch of judgments to a project's live ledger rows.

    All-or-nothing on errors: the whole batch is validated (structure, write
    access, every snippet target readable by the caller) before any row is
    touched. Rows are matched by exact (path, symbol) — plus kind when the
    item carries one — and a target no live row matches is reported in
    ``unmatched``, not an error: the tree may simply have moved since the
    caller listed. Idempotent by construction.
    """
    from scribe.services import access
    from scribe.services import snippets as snippets_svc

    if via not in _CALLER_VIAS:
        raise ValueError(f"via must be one of: {', '.join(_CALLER_VIAS)}")
    error = validate_classifications(classifications)
    if error:
        raise ValueError(error)
    if not await access.can_write_project(user_id, project_id):
        raise ValueError(f"project {project_id} not found or no write access")

    # Snippet targets resolve through the caller's own read access — a
    # family-canon snippet in another project counts (note 2786), a snippet
    # the caller cannot read does not exist for them.
    target_ids = {
        int(item["snippet_id"])
        for item in classifications
        if item.get("status") in _NEEDS_TARGET
    }
    for sid in sorted(target_ids):
        if await snippets_svc.get_snippet(user_id, sid) is None:
            raise ValueError(f"snippet {sid} not found (or not readable)")

    now = datetime.now(timezone.utc)
    classified = 0
    unmatched: list[dict] = []
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.project_id == project_id,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
        by_key: dict[tuple[str, str], list[CodeShape]] = {}
        for row in rows:
            by_key.setdefault((row.path, row.symbol), []).append(row)
        for item in classifications:
            matches = by_key.get(
                ((item.get("path") or "").strip(), (item.get("symbol") or "").strip())
            ) or []
            kind = (item.get("kind") or "").strip()
            if kind:
                matches = [r for r in matches if r.kind == kind]
            if not matches:
                unmatched.append({
                    "path": item.get("path"), "symbol": item.get("symbol"),
                })
                continue
            status = item["status"]
            for row in matches:
                row.status = status
                if status == "unclassified":
                    row.snippet_id = None
                    row.reason = None
                    row.classified_by = None
                    row.classified_at = None
                else:
                    row.snippet_id = (
                        int(item["snippet_id"]) if status in _NEEDS_TARGET else None
                    )
                    row.reason = (item.get("reason") or "").strip() or None
                    row.classified_by = via
                    row.classified_at = now
                classified += 1
        await session.commit()
    return {"classified": classified, "unmatched": unmatched}


async def list_project_shapes(
    user_id: int,
    project_id: int,
    *,
    status: str = "",
    path: str = "",
    snippet_id: int = 0,
    include_vanished: bool = False,
    limit: int = 100,
    offset: int = 0,
) -> tuple[list[CodeShape], int]:
    """A filtered page of a project's ledger, with the unfiltered-match total.

    ([], 0) when the caller can't read the project — the same silence every
    other project list gives. ``path`` matches the exact file or anything
    beneath it, mirroring recorded-location semantics.
    """
    from sqlalchemy import func, or_

    from scribe.services import access

    if not await access.can_read_project(user_id, project_id):
        return [], 0
    conds = [CodeShape.project_id == project_id]
    if not include_vanished:
        conds.append(CodeShape.vanished_at.is_(None))
    if status:
        conds.append(CodeShape.status == status)
    if path:
        clean = path.strip().strip("/")
        conds.append(or_(
            CodeShape.path == clean, CodeShape.path.like(clean + "/%")
        ))
    if snippet_id:
        conds.append(CodeShape.snippet_id == snippet_id)
    async with async_session() as session:
        total = (
            await session.execute(
                select(func.count()).select_from(CodeShape).where(*conds)
            )
        ).scalar_one()
        rows = (
            await session.execute(
                select(CodeShape).where(*conds)
                .order_by(CodeShape.path, CodeShape.symbol, CodeShape.kind)
                .limit(max(1, min(limit, 500))).offset(max(0, offset))
            )
        ).scalars().all()
    return list(rows), int(total)


def _consumer_dict(row: CodeShape) -> dict:
    """The compact shape a snippet's consumer map carries — enough to open
    the file, none of the ledger bookkeeping."""
    out = {
        "project_id": row.project_id,
        "repo": row.repo_key,
        "path": row.path,
        "symbol": row.symbol,
        "kind": row.kind,
        "classified_by": row.classified_by,
    }
    if row.reason:
        out["reason"] = row.reason
    return out


async def snippet_consumers(user_id: int, note_id: int) -> dict:
    """The structured consumer map for one snippet (#2789): its `instances`
    (rows judged to conform) and `variants` (named departures, each carrying
    its why). Rows are filtered to projects the CALLER can read — a shared
    snippet must not become a side channel into someone else's project
    layout. Empty lists mean "attach nothing" (#2483)."""
    from scribe.services import access

    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape).where(
                    CodeShape.snippet_id == note_id,
                    CodeShape.status.in_(("instance", "variant")),
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).scalars().all()
    readable: dict[int, bool] = {}
    out: dict[str, list[dict]] = {"instances": [], "variants": []}
    for row in rows:
        if row.project_id not in readable:
            readable[row.project_id] = await access.can_read_project(
                user_id, row.project_id
            )
        if not readable[row.project_id]:
            continue
        out["instances" if row.status == "instance" else "variants"].append(
            _consumer_dict(row)
        )
    return out