FabledScribe/src/scribe/models/code_shape.py

from datetime import datetime, timezone

from sqlalchemy import (
    BigInteger,
    DateTime,
    Float,
    ForeignKey,
    Index,
    Integer,
    Text,
    UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column

from scribe.models import Base
from scribe.models.base import TimestampMixin, iso

# The classification vocabulary (note 2786). `unclassified` is the default and
# THE todo state; every other status is a judgment, stamped with who made it —
# except `scoped` (#2869): the coverage sync's mechanical stamp on shapes that
# are one-offs BY CONSTRUCTION (a Vue component's scoped <style> rules and its
# <script setup> functions — unreachable from any other file). Scoped rows are
# accounted for without a human judging them, so `exempt` keeps meaning "a
# person looked"; the proposer, derive grouping and divergence still see them,
# and any judgment (instance/variant/exempt) overrides the stamp.
SHAPE_STATUSES = ("canonical", "instance", "variant", "exempt", "scoped", "unclassified")
SHAPE_CLASSIFIERS = ("agent", "audit", "hook", "mechanical", "import")

# The reason catalogue (#2874): an OPTIONAL code beside the prose reason on
# variant/exempt rows, so the ledger can be filtered and aggregated by kind
# of one-off. The prose remains the record; the code is the index.
REASON_CODES = (
    "scoped-css",            # a scoped rule styling one element (pre-#2869 rows)
    "one-off-handler",       # a view/component handler or loader, one per surface
    "test-helper",           # a test module's stub, driver or fixture data
    "convention-plumbing",   # registration, wiring, app factory — one of each
    "pure-helper",           # a sync module-private helper with no session
    "generated",             # generated source (theme.css, protos, bundles)
    "script",                # a standalone dev/CI script
    "typed-record",          # a NamedTuple / dataclass / error class — one each
)

# How the mechanical proposer (#2792) arrived at a proposal, strongest first.
# `derive` is the odd one out: not "this is an instance of #N" but "this
# shape repeats with NO canon — derive one first" (note 2786's derive-first
# rule), so it carries a group key instead of a snippet.
PROPOSAL_BASES = ("symbol", "text", "reference", "signature", "semantic", "derive")


class CodeShape(Base, TimestampMixin):
    """One extracted code shape and its classification against canon (#2787).

    The accounting half of the pattern system (governing note 2786): the
    snippet library records CANON (small); this ledger accounts for EVERY
    shape the coverage extractor finds in a bound repo (total). A row's
    status says how the shape relates to canon — it IS a snippet's reference
    (`canonical`), conforms to one (`instance` — snippet_id may point at
    another project's snippet, so family canon counts), departs deliberately
    (`variant`, with the why in `reason`), was judged one-off (`exempt`,
    a recorded judgment rather than silence), or awaits judgment
    (`unclassified` — the todo).

    Identity is (project, repo_key, path, symbol, kind) — kind is part of it
    because one file can define `.foo` (css) and `foo` (sym) as distinct
    shapes. A rename therefore reads as vanish + new row: accepted for v1,
    because chasing renames needs content identity the extractor doesn't
    have. `vanished_at` keeps the history instead of deleting it.

    snippet_id is SET NULL on snippet deletion: the classification's target
    is gone but the judgment happened; the sync pass (step 2) re-files such
    rows as unclassified so they rejoin the todo instead of dangling.

    `signature` / `body_sha` (#2792) are the shape's content fingerprint —
    its definition line and a whitespace/comment-insensitive hash of its
    block — refreshed by every sync. They are what the mechanical proposer
    matches on and what a later drift recheck compares against; the ledger
    still never stores code bodies.

    `classified_sha` remembers the fingerprint a judgment was made at;
    when a later sync sees the body change under an instance/variant, the
    row is flagged `recheck_at` (the judgment stands, it just asks to be
    confirmed again) and a `drifted` event is written. `diverges_from`
    (#2793) is the button-B flag: a shape new since the previous refresh, in
    a directory+kind where one canon dominates the judged siblings, that the
    proposer did not match to that canon — "button B appeared where button
    A is canon: divergence or variant? classify it." Both clear on judgment.

    The proposal columns hold the proposer's standing suggestion for an
    UNCLASSIFIED row: `proposed_snippet_id` + `proposal_basis` + score for
    "looks like an instance of #N", or `proposal_basis="derive"` +
    `proposal_group` for "repeats with no canon". `proposed_sha` is the
    body_sha the row was last examined at, so a refresh re-examines only
    what changed. A judgment clears the proposal — the machine proposes,
    judgment classifies.
    """

    __tablename__ = "code_shapes"
    __table_args__ = (
        UniqueConstraint(
            "project_id", "repo_key", "path", "symbol", "kind",
            name="uq_code_shapes_identity",
        ),
        Index("ix_code_shapes_project_status", "project_id", "status"),
        Index("ix_code_shapes_snippet", "snippet_id"),
        Index("ix_code_shapes_proposed", "project_id", "proposed_snippet_id"),
        Index("ix_code_shapes_diverges", "project_id", "diverges_from"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    project_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("projects.id", ondelete="CASCADE"), nullable=False
    )
    repo_key: Mapped[str] = mapped_column(Text, nullable=False)
    path: Mapped[str] = mapped_column(Text, nullable=False)
    symbol: Mapped[str] = mapped_column(Text, nullable=False)
    kind: Mapped[str] = mapped_column(Text, nullable=False)  # "css" | "sym"
    status: Mapped[str] = mapped_column(Text, nullable=False, default="unclassified")
    snippet_id: Mapped[int | None] = mapped_column(
        BigInteger, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
    )
    reason: Mapped[str | None] = mapped_column(Text, nullable=True)
    reason_code: Mapped[str | None] = mapped_column(Text, nullable=True)   # REASON_CODES (#2874)
    classified_by: Mapped[str | None] = mapped_column(Text, nullable=True)
    classified_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    first_seen_commit: Mapped[str] = mapped_column(Text, default="")
    last_seen_commit: Mapped[str] = mapped_column(Text, default="")
    vanished_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    signature: Mapped[str] = mapped_column(Text, default="")
    body_sha: Mapped[str] = mapped_column(Text, default="")
    proposed_snippet_id: Mapped[int | None] = mapped_column(
        BigInteger, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
    )
    proposal_basis: Mapped[str | None] = mapped_column(Text, nullable=True)
    proposal_score: Mapped[float | None] = mapped_column(Float, nullable=True)
    proposal_group: Mapped[str | None] = mapped_column(Text, nullable=True)
    proposed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    proposed_sha: Mapped[str] = mapped_column(Text, default="")
    classified_sha: Mapped[str] = mapped_column(Text, default="")
    recheck_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    diverges_from: Mapped[int | None] = mapped_column(
        BigInteger, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
    )

    @property
    def proposal(self) -> dict | None:
        """The standing proposal as one object, or None when the proposer
        has nothing to say about this row."""
        if self.proposed_snippet_id is None and not self.proposal_group:
            return None
        out: dict = {"basis": self.proposal_basis, "score": self.proposal_score}
        if self.proposed_snippet_id is not None:
            out["snippet_id"] = self.proposed_snippet_id
        if self.proposal_group:
            out["group"] = self.proposal_group
        return out

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "project_id": self.project_id,
            "repo_key": self.repo_key,
            "path": self.path,
            "symbol": self.symbol,
            "kind": self.kind,
            "status": self.status,
            "snippet_id": self.snippet_id,
            "reason": self.reason,
            "reason_code": self.reason_code,
            "classified_by": self.classified_by,
            "classified_at": iso(self.classified_at),
            "first_seen_commit": self.first_seen_commit,
            "last_seen_commit": self.last_seen_commit,
            "vanished_at": iso(self.vanished_at),
            "signature": self.signature,
            "body_sha": self.body_sha,
            "proposal": self.proposal,
            "classified_sha": self.classified_sha,
            "recheck_at": iso(self.recheck_at),
            "diverges_from": self.diverges_from,
            "created_at": iso(self.created_at),
            "updated_at": iso(self.updated_at),
        }

    def to_compact(self) -> dict:
        """The row as an audit reads it (#2868): identity, standing, the
        definition line and the proposer's word — none of the bookkeeping
        (commits, shas, timestamps). A 500-row page of these fits the tool
        budget; a page of to_dict() does not."""
        out = {
            "path": self.path,
            "symbol": self.symbol,
            "kind": self.kind,
            "status": self.status,
            "signature": self.signature,
        }
        if self.snippet_id is not None:
            out["snippet_id"] = self.snippet_id
        if self.classified_by:
            out["by"] = self.classified_by
        if self.reason_code:
            out["reason_code"] = self.reason_code
        proposal = self.proposal
        if proposal:
            out["proposal"] = proposal
        if self.diverges_from is not None:
            out["diverges_from"] = self.diverges_from
        if self.recheck_at is not None:
            out["recheck"] = True
        return out


# How a uses edge was established (#2870): who/what said "this shape calls
# that canon". `reference` is the proposer's mechanical by-name hit on the
# body (language-gated, #2871); `hook` is write-path evidence (pulled the
# snippet, then wrote code naming its symbol); agent/audit/import are
# judgments carried on classify_shapes(..., uses=[...]).
USE_BASES = ("reference", "hook", "agent", "audit", "import")


class CodeShapeUse(Base):
    """One consumption edge: shape → canonical snippet it calls/uses (#2870).

    Conformance (CodeShape.status/snippet_id) answers "what shape is this";
    this table answers "what does it use" — many per shape. A service function
    that is an instance of the service-function convention AND a consumer of
    hash_token has one snippet_id and one uses edge. Cascades with both ends:
    a use of a deleted snippet is no longer a fact worth keeping.
    """

    __tablename__ = "code_shape_uses"
    __table_args__ = (
        UniqueConstraint("shape_id", "snippet_id", name="uq_code_shape_uses_shape_snippet"),
        Index("ix_code_shape_uses_snippet", "snippet_id"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    shape_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("code_shapes.id", ondelete="CASCADE"), nullable=False
    )
    snippet_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("notes.id", ondelete="CASCADE"), nullable=False
    )
    basis: Mapped[str] = mapped_column(Text, nullable=False)
    evidence: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)
    )

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "shape_id": self.shape_id,
            "snippet_id": self.snippet_id,
            "basis": self.basis,
            "evidence": self.evidence,
            "created_at": iso(self.created_at),
        }


# How a consumer edge was established (milestone 302). `template` is the
# sync's mechanical read of a file's markup (class= / :class= / className=);
# the vocabulary is a list so a later basis (a stylesheet `@apply`, a script's
# classList) has a name without a schema change.
CONSUMER_BASES = ("template",)


class CodeShapeConsumer(Base):
    """One consumer edge: CSS shape → the file whose markup names its class
    (milestone 302; note 2917 — CSS is watched by name, by recipe, by token
    and by WHAT USES IT). The analogue of CodeShapeUse for styling: `uses`
    says what a shape calls, this says who renders a class. Rows, not prose,
    so "is this recipe shared or scoped?" is a count, not a guess.

    Mechanical and fully recomputable: every coverage sync rebuilds a repo's
    edges from its archive, so the table is not backed up (see
    services/backup._NOT_INCLUDED). Cascades with the shape.
    """

    __tablename__ = "code_shape_consumers"
    __table_args__ = (
        UniqueConstraint("shape_id", "path", name="uq_code_shape_consumers_shape_path"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    shape_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("code_shapes.id", ondelete="CASCADE"), nullable=False
    )
    path: Mapped[str] = mapped_column(Text, nullable=False)
    count: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    basis: Mapped[str] = mapped_column(Text, nullable=False, default="template")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)
    )

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "shape_id": self.shape_id,
            "path": self.path,
            "count": self.count,
            "basis": self.basis,
            "created_at": iso(self.created_at),
        }


# What a shape's history records (#2793). Not "appeared" — first_seen and
# created_at already say that on the row; history is for what CHANGED:
SHAPE_EVENTS = ("classified", "vanished", "reappeared", "drifted")


class CodeShapeEvent(Base):
    """One state change in a shape's life — the what-was-used-when record.

    "We used #N here from <date>, #M replaced it at commit C, reason R" is a
    question the ledger row alone cannot answer once it has moved on; this
    table keeps each judgment (status, snippet, who, why, at which commit)
    and each presence change (vanished / reappeared / drifted) as it
    happened. Denormalised path/symbol/kind so a directory's history reads
    without joining; `snippet_id` is deliberately FK-free — history outlives
    the snippet it names, which is the point.
    """

    __tablename__ = "code_shape_events"
    __table_args__ = (
        Index("ix_code_shape_events_shape", "shape_id", "at"),
        Index("ix_code_shape_events_project_path", "project_id", "path"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    shape_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("code_shapes.id", ondelete="CASCADE"), nullable=False
    )
    project_id: Mapped[int] = mapped_column(Integer, nullable=False)
    path: Mapped[str] = mapped_column(Text, nullable=False)
    symbol: Mapped[str] = mapped_column(Text, nullable=False)
    kind: Mapped[str] = mapped_column(Text, nullable=False)
    event: Mapped[str] = mapped_column(Text, nullable=False)
    status: Mapped[str | None] = mapped_column(Text, nullable=True)
    snippet_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    classified_by: Mapped[str | None] = mapped_column(Text, nullable=True)
    reason: Mapped[str | None] = mapped_column(Text, nullable=True)
    commit: Mapped[str] = mapped_column(Text, default="")
    at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "shape_id": self.shape_id,
            "project_id": self.project_id,
            "path": self.path,
            "symbol": self.symbol,
            "kind": self.kind,
            "event": self.event,
            "status": self.status,
            "snippet_id": self.snippet_id,
            "classified_by": self.classified_by,
            "reason": self.reason,
            "commit": self.commit,
            "at": iso(self.at),
        }