bvandeusenandClaude Opus 5.5
db2427a3dc
CI & Build / Python lint (push) Successful in 2s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / TypeScript typecheck (push) Successful in 55s
CI & Build / integration (push) Successful in 1m2s
CI & Build / Python tests (push) Successful in 1m56s
CI & Build / Build & push image (push) Successful in 22s
The derive check matches by name, so a renamed copy of a recorded snippet passed it and showed only as "similar" prior art. When a write introduces a definition the file does not hold (shape_ledger.new_shapes_at) and an unseen snippet on the menu clears the write-path floor by WRITEPATH_COPY_MARGIN (0.80 at the default floor), the hint now says to reuse it or extract the shared shape before another copy. Names the derive check or the stamping feed already spoke for are left to them. Covers recorded snippets only: the ledger keeps no bodies to compare an unrecorded sibling against. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3047 lines
133 KiB
Python
3047 lines
133 KiB
Python
"""The shape ledger's write side — sync and mechanical marking.
|
||
|
||
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
|
||
|
||
import difflib
|
||
import logging
|
||
import re
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Iterable, NamedTuple
|
||
|
||
from sqlalchemy import select
|
||
|
||
from grimoire.models import async_session
|
||
from grimoire.models.code_shape import (
|
||
REASON_CODES, CodeShape, CodeShapeConsumer, CodeShapeEvent, CodeShapeUse,
|
||
)
|
||
from grimoire.models.base import iso
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 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 grimoire.services.snippets import _path_touches
|
||
|
||
return _path_touches(loc_path, path)
|
||
|
||
|
||
# The rows the machine may still speak about: nobody's judgment stands on
|
||
# them. `scoped` is the sync's own by-construction stamp — the
|
||
# proposer, derive grouping, divergence, hook evidence and sweeps treat it
|
||
# like the todo; only the human todo (`unclassified`) excludes it.
|
||
_MECHANICAL_TODO = ("unclassified", "scoped")
|
||
_SCOPED_REASON = (
|
||
"by construction: a Vue component's scoped <style> rule / <script setup> "
|
||
"function — unreachable from any other file (stamped by the coverage sync)"
|
||
)
|
||
|
||
|
||
async def sync_repo_shapes(
|
||
project_id: int,
|
||
repo_key: str,
|
||
shapes: list,
|
||
*,
|
||
seen_marker: str,
|
||
) -> None:
|
||
"""Upsert one repo's extracted shapes into the ledger.
|
||
|
||
``shapes`` are (path, kind, name) triples, or the richer ArchiveShape
|
||
records whose 4th/5th fields — signature, body_sha — refresh the
|
||
row's content fingerprint, and whose 7th says the shape is a
|
||
one-off by construction: such rows are stamped `scoped` (mechanical)
|
||
while unjudged, and un-stamped if a later tree makes them reachable. ``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 shape in shapes:
|
||
path, kind, name = shape[0], shape[1], shape[2]
|
||
signature = shape[3] if len(shape) > 3 else ""
|
||
body_sha = shape[4] if len(shape) > 4 else ""
|
||
scoped = bool(shape[6]) if len(shape) > 6 else False
|
||
key = (path, name, kind)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
row = by_key.get(key)
|
||
if row is None:
|
||
row = 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,
|
||
signature=signature, body_sha=body_sha,
|
||
)
|
||
if scoped:
|
||
await _judge(session, row, status="scoped", snippet_id=None,
|
||
by="mechanical", reason=_SCOPED_REASON, at=now)
|
||
else:
|
||
session.add(row)
|
||
continue
|
||
row.last_seen_commit = seen_marker
|
||
if scoped and row.status == "unclassified":
|
||
await _judge(session, row, status="scoped", snippet_id=None,
|
||
by="mechanical", reason=_SCOPED_REASON, at=now)
|
||
elif not scoped and row.status == "scoped":
|
||
await _judge(session, row, status="unclassified", snippet_id=None,
|
||
by=None, reason=None, at=now)
|
||
if signature:
|
||
row.signature = signature
|
||
if body_sha and body_sha != row.body_sha:
|
||
judged = row.status in ("instance", "variant")
|
||
if judged and not row.classified_sha:
|
||
# Judged before fingerprints existed: the first sync
|
||
# that sees a body adopts it as the judged content.
|
||
row.classified_sha = body_sha
|
||
elif judged and row.classified_sha != body_sha and row.recheck_at is None:
|
||
# The body moved under a standing judgment: the judgment
|
||
# stands, but asks to be confirmed again.
|
||
row.recheck_at = now
|
||
session.add(_event(row, "drifted", now, commit=seen_marker))
|
||
row.body_sha = body_sha
|
||
elif row.status in ("instance", "variant") and not row.classified_sha:
|
||
row.classified_sha = row.body_sha
|
||
# A shape that vanished and came back is live again — the vanish
|
||
# stays visible in history via updated_at, not as a dead flag.
|
||
if row.vanished_at is not None:
|
||
row.vanished_at = None
|
||
session.add(_event(row, "reappeared", now, commit=seen_marker))
|
||
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
|
||
session.add(_event(row, "vanished", now, commit=row.last_seen_commit))
|
||
await session.commit()
|
||
|
||
|
||
async def move_shapes(
|
||
user_id: int, project_id: int, from_path: str, to_path: str,
|
||
) -> dict:
|
||
"""Carry the ledger rows under ``from_path`` (a file or a directory) to
|
||
the same place under ``to_path``, in every repo of the project.
|
||
|
||
Rows are keyed by path, so a directory moved in the tree would otherwise
|
||
vanish every judged row at the next sync and start the same shapes again,
|
||
unclassified, under the new path. A moved row is the same row — its id,
|
||
judgment, uses edges and history ride along — and a `moved` event
|
||
records the path it left. A row already at the target (the sync ran
|
||
first, or a provisional judgment landed there) is replaced when nobody
|
||
judged it; a judged one stands, and the old row is left where it is and
|
||
reported in ``kept``. Bodies are not compared here: the next sync does
|
||
that as always, and asks for a recheck where a judged body changed.
|
||
"""
|
||
from sqlalchemy import or_
|
||
|
||
from grimoire.services import access
|
||
from grimoire.services.snippets import _path_touches
|
||
|
||
old = (from_path or "").strip().strip("/")
|
||
new = (to_path or "").strip().strip("/")
|
||
if not old or not new:
|
||
raise ValueError("from_path and to_path are both required")
|
||
if _path_touches(old, new) or _path_touches(new, old):
|
||
raise ValueError(f"`{old}` and `{new}` overlap; move to a path outside the source")
|
||
if not await access.can_write_project(user_id, project_id):
|
||
raise ValueError(f"project {project_id} not found or no write access")
|
||
now = datetime.now(timezone.utc)
|
||
async with async_session() as session:
|
||
def beneath(prefix: str):
|
||
return select(CodeShape).where(
|
||
CodeShape.project_id == project_id,
|
||
or_(CodeShape.path == prefix, CodeShape.path.like(prefix + "/%")),
|
||
)
|
||
|
||
# LIKE reads `_` as a wildcard; the prefix test is the real filter.
|
||
sources = [
|
||
r for r in (await session.execute(beneath(old))).scalars().all()
|
||
if _path_touches(old, r.path)
|
||
]
|
||
targets = {
|
||
(r.repo_key, r.path, r.symbol, r.kind): r
|
||
for r in (await session.execute(beneath(new))).scalars().all()
|
||
}
|
||
moved, replaced, kept = 0, 0, []
|
||
for row in sources:
|
||
dest = new + row.path[len(old):]
|
||
there = targets.get((row.repo_key, dest, row.symbol, row.kind))
|
||
if there is not None:
|
||
if there.status not in _MECHANICAL_TODO:
|
||
kept.append({"path": row.path, "symbol": row.symbol, "kind": row.kind})
|
||
continue
|
||
# The target is the live sighting when the sync got there
|
||
# first; the moved row takes over its presence.
|
||
row.vanished_at = there.vanished_at
|
||
row.last_seen_commit = there.last_seen_commit or row.last_seen_commit
|
||
await session.delete(there)
|
||
await session.flush()
|
||
replaced += 1
|
||
session.add(CodeShapeEvent(
|
||
shape_id=row.id, project_id=project_id,
|
||
path=row.path, symbol=row.symbol, kind=row.kind,
|
||
event="moved", status=row.status, snippet_id=row.snippet_id,
|
||
classified_by=row.classified_by, reason=row.reason,
|
||
commit=row.last_seen_commit or "", at=now,
|
||
))
|
||
row.path = dest
|
||
moved += 1
|
||
await session.commit()
|
||
return {"moved": moved, "replaced": replaced, "kept": kept}
|
||
|
||
|
||
def _event(row: CodeShape, event: str, at: datetime, *, commit: str = "") -> CodeShapeEvent:
|
||
"""A history row for a state change on ``row`` — status/snippet/by/reason
|
||
are the row's CURRENT values, which for `classified` is the judgment
|
||
just made and for presence events is the standing one."""
|
||
return CodeShapeEvent(
|
||
shape_id=row.id, project_id=row.project_id,
|
||
path=row.path, symbol=row.symbol, kind=row.kind,
|
||
event=event, status=row.status, snippet_id=row.snippet_id,
|
||
classified_by=row.classified_by, reason=row.reason,
|
||
commit=commit or row.last_seen_commit or "", at=at,
|
||
)
|
||
|
||
|
||
async def _judge(
|
||
session, row: CodeShape, *, status: str, snippet_id: int | None,
|
||
by: str | None, reason: str | None, at: datetime, reason_code: str | None = None,
|
||
) -> None:
|
||
"""Apply a judgment to a row — the ONE place a status is set — and write
|
||
its history. Clears what a judgment settles: the standing proposal, the
|
||
recheck ask, the divergence flag; remembers the fingerprint judged.
|
||
`unclassified` is the withdrawal: fields clear, the examination is
|
||
forgotten so the proposer looks again, and history records the
|
||
withdrawal too."""
|
||
row.status = status
|
||
row.snippet_id = snippet_id if status in _NEEDS_TARGET else None
|
||
row.reason = (reason or "").strip() or None
|
||
row.reason_code = (reason_code or "").strip() or None
|
||
row.classified_by = by if status != "unclassified" else None
|
||
row.classified_at = at if status != "unclassified" else None
|
||
row.classified_sha = row.body_sha if status != "unclassified" else ""
|
||
row.recheck_at = None
|
||
row.diverges_from = None
|
||
_clear_proposal(row, reexamine=(status == "unclassified"))
|
||
if row.id is None:
|
||
# A provisional row (hook stamp on a shape not yet synced): flush so
|
||
# the event can point at it.
|
||
session.add(row)
|
||
await session.flush()
|
||
session.add(_event(row, "classified", at))
|
||
|
||
|
||
async def record_uses(
|
||
session, row: CodeShape, snippet_ids, *, basis: str, evidence: str | None = None,
|
||
) -> int:
|
||
"""Upsert consumption edges shape → snippet. A judgment-grade
|
||
basis (agent/audit/import) overwrites a mechanical one (reference/hook)
|
||
on the same edge; mechanical never overwrites a judgment. Returns the
|
||
number of edges written or refreshed. The row must be persisted (flushed)
|
||
so it has an id."""
|
||
wanted = {int(x) for x in (snippet_ids or []) if x}
|
||
if not wanted:
|
||
return 0
|
||
if row.id is None:
|
||
session.add(row)
|
||
await session.flush()
|
||
existing = {
|
||
e.snippet_id: e
|
||
for e in (
|
||
await session.execute(
|
||
select(CodeShapeUse).where(CodeShapeUse.shape_id == row.id)
|
||
)
|
||
).scalars().all()
|
||
}
|
||
judged = basis in _CALLER_VIAS
|
||
n = 0
|
||
for sid in wanted:
|
||
edge = existing.get(sid)
|
||
if edge is None:
|
||
session.add(CodeShapeUse(shape_id=row.id, snippet_id=sid, basis=basis, evidence=evidence))
|
||
n += 1
|
||
elif judged or edge.basis not in _CALLER_VIAS:
|
||
edge.basis, edge.evidence = basis, evidence
|
||
n += 1
|
||
return n
|
||
|
||
|
||
async def uses_of(shape_ids) -> dict[int, list[CodeShapeUse]]:
|
||
"""{shape_id: [edges]} for a set of rows — the read side of record_uses."""
|
||
ids = [int(x) for x in shape_ids if x]
|
||
if not ids:
|
||
return {}
|
||
async with async_session() as session:
|
||
edges = (
|
||
await session.execute(
|
||
select(CodeShapeUse).where(CodeShapeUse.shape_id.in_(ids))
|
||
.order_by(CodeShapeUse.shape_id, CodeShapeUse.snippet_id)
|
||
)
|
||
).scalars().all()
|
||
out: dict[int, list[CodeShapeUse]] = {}
|
||
for e in edges:
|
||
out.setdefault(e.shape_id, []).append(e)
|
||
return out
|
||
|
||
|
||
# --- the CSS consumer map ---------------------------------------------------
|
||
|
||
|
||
def resolve_consumers(
|
||
css_rows: Iterable[tuple[int, str, str]],
|
||
references: dict[str, dict[str, int]],
|
||
) -> dict[tuple[int, str], int]:
|
||
"""{(shape_id, consumer_path): count} — which CSS rows each file's markup
|
||
consumes. ``css_rows`` are (id, path, symbol) of the repo's live css rows;
|
||
``references`` is scan_archive's path → class token → count.
|
||
|
||
Resolution: a class named in file F resolves to F's OWN row
|
||
of that name when F defines it (a scoped rule is consumed by its own
|
||
template); otherwise to every other file's row of that name — a shared
|
||
sheet, or, when several files define it, all of them: the map says
|
||
"ambiguous" by fanning out rather than guessing one.
|
||
|
||
A token ending in ``PREFIX_MARK`` is a PREFIX reference — the
|
||
static head of a name the template concatenates, `status-*` from
|
||
`` `status-${s}` ``. It stands for every row whose symbol starts with
|
||
that head, each resolved by the same own-file-else-fan-out rule. The
|
||
template cannot tell us WHICH of them it built, so the map credits all
|
||
of them rather than calling live rules unused."""
|
||
# Lazy, like the extract_definitions import below: coverage reaches into
|
||
# this module during a refresh, so neither may import the other at load.
|
||
from grimoire.services.coverage import PREFIX_MARK
|
||
|
||
by_symbol: dict[str, list[tuple[int, str]]] = {}
|
||
for sid, path, symbol in css_rows:
|
||
by_symbol.setdefault(symbol, []).append((sid, path))
|
||
out: dict[tuple[int, str], int] = {}
|
||
|
||
def credit(rows: list[tuple[int, str]], consumer: str, count: int) -> None:
|
||
own = [sid for sid, path in rows if path == consumer]
|
||
for sid in own or [sid for sid, _path in rows]:
|
||
out[(sid, consumer)] = out.get((sid, consumer), 0) + int(count)
|
||
|
||
for consumer, tokens in references.items():
|
||
for token, count in tokens.items():
|
||
if token.endswith(PREFIX_MARK):
|
||
head = token[: -len(PREFIX_MARK)]
|
||
for symbol, rows in by_symbol.items():
|
||
if symbol.startswith(head):
|
||
credit(rows, consumer, count)
|
||
continue
|
||
rows = by_symbol.get(token)
|
||
if rows:
|
||
credit(rows, consumer, count)
|
||
return out
|
||
|
||
|
||
async def sync_repo_consumers(
|
||
project_id: int, repo_key: str, references: dict[str, dict[str, int]]
|
||
) -> int:
|
||
"""Rebuild one repo's consumer edges from its archive's class references:
|
||
insert the new, refresh changed counts, delete what the tree no longer
|
||
says (a template rewritten, a class renamed, a file gone). Edges hang on
|
||
live rows only; a vanished row's edges go with this pass. Returns how
|
||
many edges stand afterwards."""
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape.id, CodeShape.path, CodeShape.symbol, CodeShape.vanished_at).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.repo_key == repo_key,
|
||
CodeShape.kind == "css",
|
||
)
|
||
)
|
||
).all()
|
||
live = [(r[0], r[1], r[2]) for r in rows if r[3] is None]
|
||
all_ids = [r[0] for r in rows]
|
||
wanted = resolve_consumers(live, references)
|
||
existing = (
|
||
await session.execute(
|
||
select(CodeShapeConsumer).where(CodeShapeConsumer.shape_id.in_(all_ids))
|
||
)
|
||
).scalars().all() if all_ids else []
|
||
have = {(e.shape_id, e.path): e for e in existing}
|
||
for key, edge in have.items():
|
||
if key not in wanted:
|
||
await session.delete(edge)
|
||
elif edge.count != wanted[key]:
|
||
edge.count = wanted[key]
|
||
for (sid, path), count in wanted.items():
|
||
if (sid, path) not in have:
|
||
session.add(CodeShapeConsumer(shape_id=sid, path=path, count=count, basis="template"))
|
||
await session.commit()
|
||
return len(wanted)
|
||
|
||
|
||
async def consumers_of(shape_ids) -> dict[int, list[CodeShapeConsumer]]:
|
||
"""{shape_id: [edges]} for a set of rows — the read side of the map,
|
||
ordered by path so a readout is stable."""
|
||
ids = [int(x) for x in shape_ids if x]
|
||
if not ids:
|
||
return {}
|
||
async with async_session() as session:
|
||
edges = (
|
||
await session.execute(
|
||
select(CodeShapeConsumer).where(CodeShapeConsumer.shape_id.in_(ids))
|
||
.order_by(CodeShapeConsumer.shape_id, CodeShapeConsumer.path)
|
||
)
|
||
).scalars().all()
|
||
out: dict[int, list[CodeShapeConsumer]] = {}
|
||
for e in edges:
|
||
out.setdefault(e.shape_id, []).append(e)
|
||
return out
|
||
|
||
|
||
# How many consumer files a readout names before "+N more".
|
||
_CONSUMERS_SHOWN = 4
|
||
|
||
|
||
def consumer_summary(paths: Iterable[str]) -> dict:
|
||
"""{"count", "paths"} — distinct consumer files, sorted, the first few
|
||
named. The one shape every surface uses for "used by N template(s)"."""
|
||
files = sorted(set(paths))
|
||
return {"count": len(files), "paths": files[:_CONSUMERS_SHOWN]}
|
||
|
||
|
||
async def used_by_map(rows: Iterable[CodeShape]) -> dict[int, dict]:
|
||
"""{shape_id: consumer_summary} for every css row given — a row with no
|
||
consumer gets {"count": 0, "paths": []}: "no template names it" is a
|
||
finding, not an absence."""
|
||
css = [r for r in rows if r.kind == "css"]
|
||
if not css:
|
||
return {}
|
||
edges = await consumers_of([r.id for r in css])
|
||
return {r.id: consumer_summary(e.path for e in edges.get(r.id, [])) for r in css}
|
||
|
||
|
||
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()]
|
||
# A location naming a FILE and no symbol is a whole-file record: it makes
|
||
# no claim about a definition inside (`location_covers`), but it is
|
||
# exactly the claim a `file` row asks about.
|
||
whole_files = {(p or "").strip(): nid for nid, p, s in recorded
|
||
if (p or "").strip() and not (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 None and row.kind == "file":
|
||
covering = whole_files.get(row.path)
|
||
if covering is not None and row.status in _MECHANICAL_TODO:
|
||
await _judge(session, row, status="canonical", snippet_id=covering,
|
||
by="mechanical", reason=None, at=now)
|
||
elif (
|
||
covering is None
|
||
and row.status == "canonical"
|
||
and row.classified_by == "mechanical"
|
||
):
|
||
await _judge(session, row, status="unclassified", snippet_id=None,
|
||
by=None, reason=None, at=now)
|
||
await session.commit()
|
||
|
||
|
||
async def live_rows_for(user_id: int, project_id: int) -> list[CodeShape]:
|
||
"""`live_rows` behind the project read gate — for callers that arrive from
|
||
outside (an MCP tool, a route) rather than from a job that already
|
||
established who is asking. Empty for a project the caller cannot read,
|
||
never a partial answer."""
|
||
# Deferred, like every other access import in this module: services/access
|
||
# reaches back here, and a module-level import closes the cycle.
|
||
from grimoire.services import access
|
||
|
||
if not await access.can_read_project(user_id, project_id):
|
||
return []
|
||
return await live_rows(project_id)
|
||
|
||
|
||
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: 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, so a caller never learns 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 (or idea_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"
|
||
)
|
||
code = (item.get("reason_code") or "").strip()
|
||
if code and code not in REASON_CODES:
|
||
return (
|
||
f"classifications[{i}]: unknown reason_code {code!r} "
|
||
f"(one of: {', '.join(REASON_CODES)})"
|
||
)
|
||
uses = item.get("uses")
|
||
if uses is not None and (
|
||
not isinstance(uses, list) or not all(isinstance(u, int) and u > 0 for u in uses)
|
||
):
|
||
return f"classifications[{i}]: uses must be a list of snippet ids"
|
||
return None
|
||
|
||
|
||
async def classify_shapes(
|
||
user_id: int,
|
||
project_id: int,
|
||
classifications: list[dict],
|
||
*,
|
||
via: str = "agent",
|
||
repo: str = "",
|
||
) -> 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.
|
||
|
||
``repo`` — a repo bound to the project — lets a judgment land on a shape
|
||
the ledger has not synced yet: the shape the agent wrote
|
||
this turn. Such an item becomes a PROVISIONAL row under that repo, the
|
||
same row the write-path hook has always created, with no seen commit;
|
||
the next sync confirms it, or marks it vanished until the code reaches
|
||
the bound ref, and either way the judgment rides along (a vanished row
|
||
that reappears keeps its status). Without ``repo`` an unknown shape is
|
||
`unmatched`, exactly as before. An unbound ``repo`` is refused rather
|
||
than ignored: a verdict silently dropped reads as one recorded.
|
||
"""
|
||
from grimoire.services import repo_bindings as repo_bindings_svc
|
||
|
||
from grimoire.services import access
|
||
from grimoire.services import snippets as snippets_svc
|
||
|
||
if via not in _CALLER_VIAS:
|
||
raise ValueError(f"via must be one of: {', '.join(_CALLER_VIAS)}")
|
||
classifications = await _resolve_ideas(user_id, classifications)
|
||
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, 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 item in classifications:
|
||
target_ids.update(int(u) for u in (item.get("uses") or []))
|
||
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)")
|
||
repo_key = ""
|
||
if (repo or "").strip():
|
||
repo_key = await repo_bindings_svc.is_bound(project_id, repo)
|
||
if not repo_key:
|
||
raise ValueError(
|
||
f"repo {repo!r} is not bound to project {project_id} — "
|
||
"bind_repo it, or omit repo to judge synced shapes only"
|
||
)
|
||
|
||
now = datetime.now(timezone.utc)
|
||
classified = 0
|
||
provisional = 0
|
||
unmatched: list[dict] = []
|
||
touched: set[int] = set()
|
||
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 and repo_key and item.get("status") != "unclassified":
|
||
path = (item.get("path") or "").strip()
|
||
symbol = (item.get("symbol") or "").strip()
|
||
row = CodeShape(
|
||
project_id=project_id, repo_key=repo_key, path=path,
|
||
symbol=symbol, kind=kind or snippet_kind(symbol, ""),
|
||
)
|
||
session.add(row)
|
||
by_key.setdefault((path, symbol), []).append(row)
|
||
matches = [row]
|
||
provisional += 1
|
||
if not matches:
|
||
unmatched.append({
|
||
"path": item.get("path"), "symbol": item.get("symbol"),
|
||
})
|
||
continue
|
||
status = item["status"]
|
||
for row in matches:
|
||
touched.update(x for x in (row.snippet_id, item.get("snippet_id")) if x)
|
||
await _judge(
|
||
session, row, status=status,
|
||
snippet_id=int(item["snippet_id"]) if status in _NEEDS_TARGET else None,
|
||
by=via, reason=item.get("reason"), at=now,
|
||
reason_code=item.get("reason_code"),
|
||
)
|
||
if item.get("uses"):
|
||
await record_uses(session, row, item["uses"], basis=via,
|
||
evidence=item.get("reason"))
|
||
classified += 1
|
||
await session.commit()
|
||
out = {"classified": classified, "unmatched": unmatched}
|
||
if provisional:
|
||
out["provisional"] = provisional
|
||
moved = await _sync_family(user_id, project_id, touched)
|
||
if moved:
|
||
out["family"] = moved
|
||
return out
|
||
|
||
|
||
async def _resolve_ideas(user_id: int, items: list[dict]) -> list[dict]:
|
||
"""An item may name a family idea (`idea_id`) in place of a snippet: it
|
||
is judged against the idea's reference in the shape's language, or its
|
||
first when there is none in that language — an idea is shared across
|
||
languages. An explicit snippet_id wins."""
|
||
from grimoire.services import family_adoption
|
||
|
||
out = []
|
||
for i, item in enumerate(items or []):
|
||
if (isinstance(item, dict) and item.get("idea_id") and not item.get("snippet_id")
|
||
and item.get("status") in _NEEDS_TARGET):
|
||
try:
|
||
sid = await family_adoption.reference_for(
|
||
user_id, int(item["idea_id"]), (item.get("path") or "").strip())
|
||
except (TypeError, ValueError) as exc:
|
||
raise ValueError(f"classifications[{i}]: {exc}") from None
|
||
item = {**item, "snippet_id": sid}
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
async def _sync_family(user_id: int, project_id: int, snippet_ids: set[int]) -> list[dict]:
|
||
"""The adoption ledger follows the shapes: after a
|
||
judgment commits, the canon ideas whose references it touched — as the
|
||
new target or the one it replaced — are answered from the shapes.
|
||
|
||
A failure is logged and the judgment stands; the next judgment touching
|
||
those references, or the next coverage refresh, brings the two back in
|
||
line."""
|
||
from grimoire.services import family_adoption
|
||
|
||
if not snippet_ids:
|
||
return []
|
||
try:
|
||
return await family_adoption.sync_from_shapes(user_id, project_id, snippet_ids)
|
||
except Exception:
|
||
logger.warning("family adoption sync failed for project %s", project_id, exc_info=True)
|
||
return []
|
||
|
||
|
||
def rule_matches(row: CodeShape, *, path: str, pattern: str, kind: str) -> bool:
|
||
"""Does a ledger row fall under a rule-form classification?
|
||
``path`` is a file or a directory (everything beneath it), ``pattern``
|
||
a shell glob on the symbol (``""`` = every symbol), ``kind`` narrows to
|
||
sym/css. Pure, so the sweep's reach can be tested without a database."""
|
||
import fnmatch
|
||
|
||
clean = (path or "").strip().strip("/")
|
||
if clean and not (row.path == clean or row.path.startswith(clean + "/")):
|
||
return False
|
||
if kind and row.kind != kind:
|
||
return False
|
||
if pattern and not fnmatch.fnmatchcase(_norm_symbol(row.symbol), pattern):
|
||
return False
|
||
return True
|
||
|
||
|
||
async def classify_shapes_where(
|
||
user_id: int,
|
||
project_id: int,
|
||
*,
|
||
path: str,
|
||
status: str,
|
||
pattern: str = "",
|
||
kind: str = "",
|
||
snippet_id: int | None = None,
|
||
reason: str | None = None,
|
||
via: str = "agent",
|
||
include_judged: bool = False,
|
||
reason_code: str | None = None,
|
||
uses: list[int] | None = None,
|
||
) -> dict:
|
||
"""The sweep form of classify_shapes: one judgment applied to
|
||
every live row under ``path`` whose symbol matches ``pattern`` (and
|
||
``kind``). By default only unjudged rows are touched — `unclassified`
|
||
and the sync's mechanical `scoped` stamp — a sweep must never silently
|
||
overwrite a judgment; ``include_judged`` opts in.
|
||
Same gates as the row form (status vocabulary, snippet target, reason
|
||
for variant/exempt, write access); one transaction, so it applies whole
|
||
or not at all. Returns the count and a sample of what it judged."""
|
||
from grimoire.services import access
|
||
from grimoire.services import snippets as snippets_svc
|
||
|
||
if via not in _CALLER_VIAS:
|
||
raise ValueError(f"via must be one of: {', '.join(_CALLER_VIAS)}")
|
||
if not (path or "").strip():
|
||
raise ValueError("path is required — a sweep names the directory it judges")
|
||
if status == "canonical":
|
||
raise ValueError("canonical is the sync's stamp on a snippet's own location — a sweep cannot set it")
|
||
probe = {"path": path, "symbol": "*", "status": status,
|
||
"snippet_id": snippet_id or 0, "reason": reason or "",
|
||
"reason_code": reason_code or "", "uses": uses}
|
||
error = validate_classifications([probe])
|
||
if error:
|
||
raise ValueError(error.replace("classifications[0]", "rule"))
|
||
if not await access.can_write_project(user_id, project_id):
|
||
raise ValueError(f"project {project_id} not found or no write access")
|
||
if status in _NEEDS_TARGET and await snippets_svc.get_snippet(user_id, int(snippet_id)) is None:
|
||
raise ValueError(f"snippet {snippet_id} not found (or not readable)")
|
||
for sid in sorted({int(u) for u in (uses or [])}):
|
||
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)
|
||
judged: list[str] = []
|
||
touched: set[int] = {int(snippet_id)} if snippet_id and status in _NEEDS_TARGET else set()
|
||
async with async_session() as session:
|
||
conds = [CodeShape.project_id == project_id, CodeShape.vanished_at.is_(None)]
|
||
if not include_judged:
|
||
conds.append(CodeShape.status.in_(_MECHANICAL_TODO))
|
||
rows = (await session.execute(select(CodeShape).where(*conds))).scalars().all()
|
||
for row in rows:
|
||
if not rule_matches(row, path=path, pattern=pattern, kind=kind):
|
||
continue
|
||
if row.snippet_id:
|
||
touched.add(row.snippet_id)
|
||
await _judge(
|
||
session, row, status=status,
|
||
snippet_id=int(snippet_id) if status in _NEEDS_TARGET else None,
|
||
by=via, reason=reason, at=now, reason_code=reason_code,
|
||
)
|
||
if uses:
|
||
await record_uses(session, row, uses, basis=via, evidence=reason)
|
||
judged.append(f"{row.path}::{row.symbol}")
|
||
await session.commit()
|
||
out = {"classified": len(judged), "sample": sorted(judged)[:12]}
|
||
moved = await _sync_family(user_id, project_id, touched)
|
||
if moved:
|
||
out["family"] = moved
|
||
return out
|
||
|
||
|
||
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,
|
||
proposal: str = "",
|
||
flag: str = "",
|
||
uses: 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. ``proposal`` narrows
|
||
to rows the proposer has spoken about: "any", "canon" (an instance-of-#N
|
||
suggestion), "derive" (a repeats-with-no-canon group), or one basis
|
||
name (symbol/reference/text/signature/semantic). ``flag`` narrows to
|
||
the readout's asks: "divergence" (new where a canon dominates,
|
||
`diverges_from` names it), "recheck" (a judged shape whose body moved),
|
||
or "unused-css" (a css rule no template names).
|
||
"""
|
||
from sqlalchemy import func, or_
|
||
|
||
from grimoire.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)
|
||
if proposal == "any":
|
||
conds.append(or_(
|
||
CodeShape.proposed_snippet_id.isnot(None),
|
||
CodeShape.proposal_group.isnot(None),
|
||
))
|
||
elif proposal == "canon":
|
||
conds.append(CodeShape.proposed_snippet_id.isnot(None))
|
||
elif proposal == "derive":
|
||
conds.append(CodeShape.proposal_group.isnot(None))
|
||
elif proposal:
|
||
conds.append(CodeShape.proposal_basis == proposal)
|
||
if flag == "divergence":
|
||
conds.append(CodeShape.diverges_from.isnot(None))
|
||
elif flag == "recheck":
|
||
conds.append(CodeShape.recheck_at.isnot(None))
|
||
elif flag == "unused-css":
|
||
# The consumer map's negative space: a live css rule
|
||
# no file's markup names. A candidate for deletion, surfaced — never
|
||
# deleted — because the map reads templates only (a class built at
|
||
# runtime, or used from a script, is invisible to it).
|
||
conds.append(CodeShape.kind == "css")
|
||
conds.append(~CodeShape.id.in_(select(CodeShapeConsumer.shape_id)))
|
||
if uses:
|
||
# Consumers of a canon: rows with a uses edge to it, whatever
|
||
# shape they themselves are.
|
||
conds.append(CodeShape.id.in_(
|
||
select(CodeShapeUse.shape_id).where(CodeShapeUse.snippet_id == uses)
|
||
))
|
||
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: 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"."""
|
||
from grimoire.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()
|
||
# The consumption edges: rows that USE this canon, whatever
|
||
# shape they are themselves — the call-site map.
|
||
using = (
|
||
await session.execute(
|
||
select(CodeShape, CodeShapeUse.basis, CodeShapeUse.evidence)
|
||
.join(CodeShapeUse, CodeShapeUse.shape_id == CodeShape.id)
|
||
.where(CodeShapeUse.snippet_id == note_id, CodeShape.vanished_at.is_(None))
|
||
)
|
||
).all()
|
||
readable: dict[int, bool] = {}
|
||
|
||
async def can_read(pid: int) -> bool:
|
||
if pid not in readable:
|
||
readable[pid] = await access.can_read_project(user_id, pid)
|
||
return readable[pid]
|
||
|
||
out: dict[str, list[dict]] = {"instances": [], "variants": [], "uses": []}
|
||
for row in rows:
|
||
if not await can_read(row.project_id):
|
||
continue
|
||
out["instances" if row.status == "instance" else "variants"].append(
|
||
_consumer_dict(row)
|
||
)
|
||
for row, basis, evidence in using:
|
||
if not await can_read(row.project_id):
|
||
continue
|
||
d = _consumer_dict(row)
|
||
d["basis"] = basis
|
||
if evidence:
|
||
d["evidence"] = evidence
|
||
d.pop("reason", None)
|
||
out["uses"].append(d)
|
||
return out
|
||
|
||
|
||
# --- write-path stamping: hook evidence lands as rows -----------------------
|
||
#
|
||
# The write-path hook (plugin/hooks/grimoire_prior_art.sh) fires on every
|
||
# Write/Edit and already carries the two halves of a consumer-map row: the
|
||
# file being written and the definitions in (or enclosing) the payload. What
|
||
# it could not say on its own is WHICH canon the session is instantiating.
|
||
# The pull stream answers that: a snippet the session opened in full
|
||
# (get_snippet) and is now writing code that references or resembles is being
|
||
# reused — and a reused canon's call site is an `instance`.
|
||
#
|
||
# The rule, deliberately two-sided so it cannot fire on noise:
|
||
# PULLED — a PULLED usage event by this user inside PULL_WINDOW. Offered-
|
||
# but-ignored (surfaced, never opened) stamps nothing.
|
||
# IN PLAY — the payload references the snippet's symbol by name, or the
|
||
# semantic arm scored it above the write-path threshold for this
|
||
# very payload. Either is evidence; the pull alone is not.
|
||
# Both hold → every shape the hook named at that path, of the snippet's kind,
|
||
# carries N as a PROPOSAL. Older rows stamped instance-of-N with
|
||
# classified_by="hook" — a verdict nobody read — are what `stamps_to_review`
|
||
# lists.
|
||
#
|
||
# A proposal is EVIDENCE, not judgment: it lands only on rows nobody has
|
||
# judged, never touches a status, and is what the end-of-turn question shows
|
||
# the agent as "looks like #N". The agent's classify_shapes is the verdict.
|
||
|
||
# "The write path actually pulled it": a working session's reach. The
|
||
# precision comes from the in-play test above, not from this window.
|
||
PULL_WINDOW = timedelta(hours=6)
|
||
|
||
# ── WHAT KIND OF THING a shape is, past `css | sym` ──────────────────────
|
||
#
|
||
# THE PROBLEM THIS EXISTS FOR. `kind` has exactly two values, so a frozen
|
||
# dataclass, a module constant, a sync predicate, a class and an async service
|
||
# function are all siblings of one another. Nothing downstream could possibly
|
||
# discriminate on a distinction the column does not carry, which leaves a
|
||
# directory base rate standing alone as the entire divergence argument. The
|
||
# consequence: a registry module of pure helpers gets prompted to build from
|
||
# an `async_session` service canon, and one snippet can be recorded as the
|
||
# canon for a `class`, a plain `def`, an `async def` and a dozen test
|
||
# functions at once.
|
||
#
|
||
# DERIVED FROM THE SIGNATURE ON READ, not stored. `kind` is part of the row
|
||
# identity — ("project_id", "repo_key", "path", "symbol", "kind") — so
|
||
# widening that column rewrites every row's key and needs a migration plus a
|
||
# re-extract. Reading the form off the signature costs nothing, needs no
|
||
# migration, and is reversible. Widening `kind` remains the principled fix and
|
||
# this does not foreclose it: every caller below asks `shape_form`, so the day
|
||
# the column carries the answer, this function returns it instead.
|
||
#
|
||
# DELIBERATELY COARSE, and it must stay that way. It answers "could these
|
||
# plausibly be the same kind of thing", never "what language construct is
|
||
# this". A finer taxonomy would need per-language parsing and would start
|
||
# disagreeing with itself across the four languages this ledger already holds
|
||
# (Python, Go, TypeScript, Vue SFC) — and a classifier that is wrong in a new
|
||
# way is worse than the coarse one it replaced.
|
||
FORM_UNKNOWN = ""
|
||
|
||
# How much a canon's own instances must agree before it may assert a form.
|
||
# A canon whose rows disagree about what they are cannot tell anyone else what
|
||
# to be — and that disagreement is the SYMPTOM of the bad stamping this same
|
||
# change fixes, so reading it as "no opinion" makes the two halves cooperate:
|
||
# a canon poisoned by loose stamping falls silent instead of flagging.
|
||
_FORM_SHARE = 0.6
|
||
|
||
# How much of a canon's membership may be shapes it cannot account for
|
||
# before the canon stops meaning anything. Not a majority test — see
|
||
# `canon_coherence` for why a majority test goes blind exactly when the
|
||
# ledger is worst.
|
||
_STRANGER_SHARE = 0.2
|
||
|
||
# Leading words that say nothing about the form of the thing being declared.
|
||
_FORM_NOISE = ("export ", "default ", "public ", "private ", "static ", "final ")
|
||
|
||
# A declared TYPE, across the languages in play. Coarse on purpose: a Go
|
||
# struct, a TS interface and a Python class are one bucket because the
|
||
# question is only ever "is the other thing also a type".
|
||
_FORM_TYPE_WORDS = ("class ", "interface ", "type ", "struct ", "enum ")
|
||
_FORM_FN_WORDS = ("def ", "function ", "func ", "fn ", "sub ")
|
||
|
||
# THE DECLARING KEYWORD IS OPTIONAL, because Python has none. A module
|
||
# constant is written `MAX_BUDGET = 10` — no `const`, no `let` — and a
|
||
# pattern that required one classified every Python constant as unreadable,
|
||
# which then read downstream as "do not assert" and quietly excluded a whole
|
||
# form from both checks. Caught by the payload test, not by review.
|
||
_FORM_BINDING = re.compile(r"^(?:(?:const|let|var)\s+)?[\w$]+\s*(?::[^=]+)?=")
|
||
_FORM_ARROW = re.compile(
|
||
r"^(?:(?:const|let|var)\s+)?[\w$]+\s*(?::[^=]+)?=\s*(async\s*)?\("
|
||
)
|
||
|
||
|
||
def shape_form(signature: str, kind: str = "sym") -> str:
|
||
"""The structural form of a shape: css / type / async-fn / fn / binding.
|
||
|
||
Returns FORM_UNKNOWN when the signature does not say, and every caller
|
||
treats that as "do not assert", never as "no match". An unreadable
|
||
signature must make this quieter, not more confident.
|
||
"""
|
||
if kind == "css":
|
||
return "css"
|
||
if kind == "file":
|
||
# A whole file: its own family, comparable only with
|
||
# other files — a component is never told to build from a helper.
|
||
return "file"
|
||
sig = (signature or "").strip()
|
||
if not sig:
|
||
return FORM_UNKNOWN
|
||
changed = True
|
||
while changed:
|
||
changed = False
|
||
for lead in _FORM_NOISE:
|
||
if sig.startswith(lead):
|
||
sig, changed = sig[len(lead):].lstrip(), True
|
||
if sig.startswith(_FORM_TYPE_WORDS):
|
||
return "type"
|
||
if sig.startswith("async "):
|
||
return "async-fn"
|
||
if sig.startswith(_FORM_FN_WORDS):
|
||
return "fn"
|
||
m = _FORM_ARROW.match(sig)
|
||
if m:
|
||
return "async-fn" if m.group(1) else "fn"
|
||
if _FORM_BINDING.match(sig):
|
||
return "binding"
|
||
return FORM_UNKNOWN
|
||
|
||
|
||
def forms_agree(a: str, b: str) -> bool:
|
||
"""Do two forms match well enough to assert a relationship?
|
||
|
||
BOTH must be known. "Unknown equals unknown" would make two shapes nobody
|
||
can read into a confident pair, which is the failure this whole change is
|
||
about — a guess dressed as a finding.
|
||
"""
|
||
return bool(a) and bool(b) and a == b
|
||
|
||
|
||
def forms_conflict(a: str, b: str) -> bool:
|
||
"""Do two KNOWN forms rule each other out?
|
||
|
||
The weaker sibling of `forms_agree`, and the pair exists because the two
|
||
kinds of evidence this ledger acts on deserve different burdens.
|
||
|
||
An explicit by-name reference — the payload literally names the canon's
|
||
symbol — is strong, so it needs only the absence of a contradiction: stamp
|
||
unless the forms are both readable and different. A payload-level
|
||
RESEMBLANCE SCORE is weak, computed against the whole file, so one score
|
||
speaks for every symbol in it; that needs positive agreement before
|
||
asserting anything, which is `forms_agree`.
|
||
|
||
Demanding agreement everywhere would silence the by-name path whenever a
|
||
shape's definition is not in the payload — an Edit rather than a Write —
|
||
turning strong evidence into no evidence for a reason that has nothing to
|
||
do with the code.
|
||
"""
|
||
return bool(a) and bool(b) and a != b
|
||
|
||
|
||
def shape_family(form: str) -> str:
|
||
"""The coarse category a form belongs to, or "" when the form is unknown.
|
||
|
||
WHY THIS IS SEPARATE FROM `shape_form`. The two checks in this module ask
|
||
opposite questions, so they cannot share a burden:
|
||
|
||
STAMPING asserts "this IS that canon". Agreement in form is evidence
|
||
FOR the claim, so `forms_agree` / `forms_conflict` — the precise level,
|
||
where `fn` and `async-fn` are different — is right.
|
||
|
||
DIVERGENCE asserts "this is NOT the canon that dominates here — did you
|
||
mean to?" A form MISMATCH is the premise of that prompt, not an
|
||
objection to it. Gating it on `forms_agree` would invert the check: it
|
||
would go silent on exactly the mismatches it exists to catch, such as a
|
||
hand-rolled sync `confirmDanger` in a directory where an async confirm
|
||
helper is canon.
|
||
|
||
So divergence gates at FAMILY level and only on contradiction. A sync
|
||
function beside an async one is still a fair question. A frozen dataclass
|
||
told to build from an async service function is not a question at all.
|
||
|
||
This does NOT remove every false prompt. A type against a callable canon
|
||
is silenced; an unrelated callable beside a callable canon still asks,
|
||
because at the signature level it is indistinguishable from the
|
||
`confirmDanger` case above. Only widening `kind` past `css | sym` or
|
||
comparing meaning rather than form can separate them. The gap is known,
|
||
not an oversight.
|
||
"""
|
||
if form in ("fn", "async-fn"):
|
||
return "callable"
|
||
if form == "type":
|
||
return "type"
|
||
if form == "binding":
|
||
return "value"
|
||
if form == "css":
|
||
return "css"
|
||
if form == "file":
|
||
return "file"
|
||
return ""
|
||
|
||
|
||
def families_conflict(a: str, b: str) -> bool:
|
||
"""Do two forms belong to categorically different KNOWN families?
|
||
|
||
Like `forms_conflict`, both sides must be readable: an unknown form makes
|
||
this quieter, never more confident.
|
||
"""
|
||
fa, fb = shape_family(a), shape_family(b)
|
||
return bool(fa) and bool(fb) and fa != fb
|
||
|
||
|
||
def canon_form(rows: Iterable, snippet_id: int) -> str:
|
||
"""The form a canon's own judged rows agree on, or FORM_UNKNOWN."""
|
||
forms: dict[str, int] = {}
|
||
for r in rows:
|
||
if r.snippet_id != snippet_id or r.status not in ("canonical", "instance"):
|
||
continue
|
||
if is_weak_stamp(r):
|
||
continue
|
||
f = shape_form(getattr(r, "signature", "") or "", r.kind)
|
||
if f:
|
||
forms[f] = forms.get(f, 0) + 1
|
||
if not forms:
|
||
return FORM_UNKNOWN
|
||
top, n = max(forms.items(), key=lambda kv: kv[1])
|
||
return top if n / sum(forms.values()) >= _FORM_SHARE else FORM_UNKNOWN
|
||
|
||
|
||
# A judgment nobody read before it was written. "hook" is the write-path
|
||
# stamp — a similarity score and no reader. Every other value ("agent",
|
||
# "audit", "import", "mechanical") came from something that examined the
|
||
# shape and said so, which is a different kind of evidence, not a stronger
|
||
# score.
|
||
_UNATTENDED_BY = ("hook",)
|
||
|
||
|
||
def canon_coherence(members: Iterable) -> dict:
|
||
"""Does a canon's membership still agree on what the canon IS?
|
||
|
||
The verdict is NOT "do all the rows share a form". A form-agreement
|
||
check reports sound canons and misses polluted ones, for two reasons, and
|
||
the fix for the first can break the second:
|
||
|
||
1. FAMILY, NOT FORM. A sync loop-starter and the async tick it schedules
|
||
are one shape written two ways; `shape_family` collapses `fn` and
|
||
`async-fn` into `callable` for exactly this reason on the divergence
|
||
side. A canon of four starters and three ticks is coherent, and a
|
||
form count would score it 4/7.
|
||
|
||
2. A METHOD OF A MEMBER IS NOT A STRANGER. A model convention can cover
|
||
the class AND the `to_dict` the class must carry. Thirty-seven classes
|
||
and twenty-five serialisers score 37/62 on form, under the bar, for
|
||
containing exactly what the canon says it contains.
|
||
|
||
THE TRAP, and it is why this is not simply a family histogram: grouping
|
||
by family and keeping a majority test makes the check BLIND. The same
|
||
model canon polluted with 31 unrelated callables — Vue functions, route
|
||
handlers, tests — holds 37 classes and 56 callables; as families that is
|
||
56/93, a clean pass, and the 31 rows that have no business being there
|
||
are never reported. A looser bar in the same shape is not a fix.
|
||
|
||
So the verdict is inverted. Instead of asking whether most rows agree, it
|
||
asks how many rows the canon CANNOT ACCOUNT FOR:
|
||
|
||
* a row in the majority family is accounted for;
|
||
* a callable defined in a FILE that also holds a majority-family `type`
|
||
row is accounted for — it is a method of a member;
|
||
* everything else is a STRANGER, and strangers above `_STRANGER_SHARE`
|
||
of the readable rows make the canon incoherent.
|
||
|
||
The majority vote abstains those methods, so a class's own serialisers
|
||
cannot outvote the classes and turn the members into the strangers. The
|
||
three cases above separate completely: the clean model canon has 0
|
||
strangers in 62, the loop canon 0 in 7, and the polluted one 31 in 93.
|
||
|
||
Scoped to the REVIEW surface on purpose. `canon_form` still answers at
|
||
the precise form level for stamping and divergence, where a sync helper
|
||
beside an async canon is a fair question; nothing here changes what the
|
||
ledger writes.
|
||
|
||
Returns a dict, deliberately. Widening a tuple return breaks every
|
||
consumer that still unpacks the old width, and Python says nothing until
|
||
that line runs.
|
||
"""
|
||
forms: dict[str, int] = {}
|
||
families: dict[str, int] = {}
|
||
per_row: list[tuple[object, str]] = []
|
||
has_type: set[str] = set()
|
||
for r in members:
|
||
f = shape_form(getattr(r, "signature", "") or "", r.kind)
|
||
if not f:
|
||
continue # unreadable: never counts either way
|
||
fam = shape_family(f)
|
||
forms[f] = forms.get(f, 0) + 1
|
||
families[fam] = families.get(fam, 0) + 1
|
||
per_row.append((r, fam))
|
||
if fam == "type":
|
||
has_type.add(r.path)
|
||
|
||
out = {
|
||
"readable": len(per_row),
|
||
"forms": dict(sorted(forms.items(), key=lambda kv: -kv[1])),
|
||
"families": dict(sorted(families.items(), key=lambda kv: -kv[1])),
|
||
"family": FORM_UNKNOWN,
|
||
"attached": 0,
|
||
"strangers": [],
|
||
"coheres": True,
|
||
}
|
||
if not per_row:
|
||
return out # nothing legible: say nothing, not "broken"
|
||
|
||
# The vote, with methods abstaining. A callable sitting in a file that
|
||
# defines a class is presumed to belong to it and does not get to argue
|
||
# that the canon is really about callables.
|
||
def _attachable(row, fam: str) -> bool:
|
||
return fam == "callable" and row.path in has_type
|
||
|
||
votes: dict[str, int] = {}
|
||
for r, fam in per_row:
|
||
if _attachable(r, fam):
|
||
continue
|
||
votes[fam] = votes.get(fam, 0) + 1
|
||
if not votes: # every readable row is a method
|
||
votes = dict(families)
|
||
# Ties go to `type`: a canon that defines a class is about the class.
|
||
family = max(votes, key=lambda k: (votes[k], k == "type"))
|
||
|
||
attached, strangers = 0, []
|
||
for r, fam in per_row:
|
||
if fam == family:
|
||
continue
|
||
if family == "type" and _attachable(r, fam):
|
||
attached += 1
|
||
else:
|
||
strangers.append(r)
|
||
out["family"] = family
|
||
out["attached"] = attached
|
||
out["strangers"] = strangers
|
||
out["coheres"] = len(strangers) / len(per_row) <= _STRANGER_SHARE
|
||
return out
|
||
|
||
|
||
def signature_in(code: str, symbol: str, kind: str) -> str:
|
||
"""The line in ``code`` that DEFINES ``symbol``, or "" if none does.
|
||
|
||
The write-time checks are asked about a shape that may have no ledger row
|
||
yet — it is being written right now — so the payload is the only place its
|
||
signature exists. Finding nothing returns "", which reads downstream as
|
||
FORM_UNKNOWN and therefore as silence.
|
||
"""
|
||
if not code or not symbol:
|
||
return ""
|
||
name = re.escape(symbol.lstrip("."))
|
||
if kind == "css":
|
||
pat = re.compile(r"^\s*\." + name + r"\b")
|
||
else:
|
||
pat = re.compile(
|
||
r"^\s*(?:(?:export|default|public|private|static|final)\s+)*"
|
||
r"(?:"
|
||
r"(?:(?:async\s+)?(?:def|function|func|fn|sub)|class|interface|type"
|
||
r"|struct|enum|const|let|var)\s+" + name + r"\b"
|
||
# A bare binding — `MAX = 10`, `Handler = ...` — which is how
|
||
# Python (and plain JS assignment) declares one.
|
||
r"|" + name + r"\s*(?::[^=\n]+)?=(?!=)"
|
||
r")"
|
||
)
|
||
for line in code.splitlines():
|
||
if pat.match(line):
|
||
return line.strip()
|
||
return ""
|
||
|
||
|
||
def snippet_kind(symbol: str, language: str) -> str:
|
||
"""The ledger kind a snippet's reference belongs to — "css" when its
|
||
symbol is a class selector (or it is a stylesheet with no symbol),
|
||
else "sym"."""
|
||
sym = (symbol or "").strip()
|
||
if sym.startswith("."):
|
||
return "css"
|
||
if not sym and (language or "").strip().lower() in ("css", "scss", "sass", "less"):
|
||
return "css"
|
||
return "sym"
|
||
|
||
|
||
def references_symbol(code: str, symbol: str, kind: str) -> bool:
|
||
"""Does this payload name the snippet's symbol? Word-bounded so `confirm`
|
||
never claims `confirmed`; a CSS class matches as `.btn` or inside a class
|
||
attribute (`btn btn-primary`), dashes counting as part of the name."""
|
||
sym = _norm_symbol(symbol or "")
|
||
if not sym or not code:
|
||
return False
|
||
if kind == "css":
|
||
pattern = rf"(?<![\w-]){re.escape(sym)}(?![\w-])"
|
||
else:
|
||
pattern = rf"(?<![\w$]){re.escape(sym)}(?![\w$])"
|
||
return re.search(pattern, code) is not None
|
||
|
||
|
||
async def recent_pulls(user_id: int, *, window: timedelta = PULL_WINDOW) -> dict[int, datetime]:
|
||
"""{note_id: last pulled at} for every note this user opened in full
|
||
inside ``window`` — the "actually pulled it" half of the stamping rule.
|
||
Reads the usage telemetry table; an unreadable table means no evidence."""
|
||
from sqlalchemy import func
|
||
|
||
from grimoire.models.note_usage import PULLED, NoteUsageEvent
|
||
|
||
since = datetime.now(timezone.utc) - window
|
||
try:
|
||
async with async_session() as session:
|
||
rows = await session.execute(
|
||
select(NoteUsageEvent.note_id, func.max(NoteUsageEvent.created_at))
|
||
.where(
|
||
NoteUsageEvent.user_id == user_id,
|
||
NoteUsageEvent.event == PULLED,
|
||
NoteUsageEvent.created_at >= since,
|
||
)
|
||
.group_by(NoteUsageEvent.note_id)
|
||
)
|
||
return {int(nid): ts for nid, ts in rows.all()}
|
||
except Exception:
|
||
logger.warning("recent_pulls read failed — no hook stamping this write", exc_info=True)
|
||
return {}
|
||
|
||
|
||
# THE HOOK'S EVIDENCE, WRITTEN ONCE AND READ BACK BY ONE PARSER.
|
||
#
|
||
# The stamp's score is stored in a fixed format so something can ask "how
|
||
# strong was the evidence for this row?" without re-deriving it. A score that
|
||
# is only printed is never queried, and weak rows sit unexamined. Format and
|
||
# parser live beside each other so a change to one that breaks the other is a
|
||
# visible edit rather than a silent drift.
|
||
_REFERENCE_REASON = "hook: pulled #{sid}; payload references `{symbol}`"
|
||
_RESEMBLE_REASON = "hook: pulled #{sid}; payload resembles it ({score:.2f})"
|
||
_RESEMBLE_READ = re.compile(
|
||
r"^hook: pulled #(\d+); payload resembles it \(([0-9]*\.?[0-9]+)\)$"
|
||
)
|
||
|
||
|
||
def stamp_score(reason: str | None) -> float | None:
|
||
"""The resemblance a hook stamp recorded, or None if it was not one.
|
||
|
||
None means "this row was not written on a similarity score" — a by-name
|
||
reference, an agent judgment, an audit — NOT "the score was zero". Every
|
||
caller treats the two differently, because a row a person judged is not
|
||
weak evidence, it is a different kind of evidence entirely.
|
||
"""
|
||
m = _RESEMBLE_READ.match((reason or "").strip())
|
||
return float(m.group(2)) if m else None
|
||
|
||
|
||
def _stamp_allowed(rank: int, mine: str, canon: str) -> bool:
|
||
"""May this evidence assert that a shape of form ``mine`` IS ``canon``?
|
||
|
||
Named rather than inlined because the asymmetry is the decision, not an
|
||
implementation detail: rank 2 (the payload names the canon's symbol) has
|
||
only to avoid contradicting, rank 1 (a whole-file similarity score) has to
|
||
positively agree. See `forms_conflict` for why both exist.
|
||
"""
|
||
if rank >= 2:
|
||
return not forms_conflict(mine, canon)
|
||
return forms_agree(mine, canon)
|
||
|
||
|
||
async def suggest_write_path_instances(
|
||
user_id: int,
|
||
project_id: int,
|
||
*,
|
||
path: str,
|
||
shapes: list[tuple[str, str]],
|
||
code: str,
|
||
pulled: dict[int, datetime],
|
||
resembles: dict[int, float] | None = None,
|
||
repo_key: str = "",
|
||
) -> list[dict]:
|
||
"""Offer hook evidence as a PROPOSAL on the shapes being written.
|
||
|
||
Evidence never lands as an `instance` row: that would be a permanent
|
||
classification written with nobody reading it. The agent that wrote the
|
||
code says what it is, at the end of the turn, and this evidence is what
|
||
it is shown when it does. So the row keeps its status; the candidate
|
||
canon goes into the
|
||
proposal fields (`proposed_snippet_id`, `proposal_basis` "reference" for a
|
||
by-name hit or "semantic" for a resemblance, `proposal_score`), where the
|
||
end-of-turn question reads it as "looks like #N". A judged row is left
|
||
alone entirely.
|
||
|
||
``shapes`` is the hook's (kind, name) list for ``path``; ``pulled`` is
|
||
recent_pulls(); ``resembles`` maps snippet ids the semantic arm scored
|
||
for this payload to their score. Returns the suggestions, each
|
||
{path, symbol, kind, snippet_id, reason} — empty in the common case.
|
||
|
||
A shape the ledger has no live row for yet (it is being written right
|
||
now) gets a PROVISIONAL unclassified row under ``repo_key`` — first/
|
||
last-seen empty — so the suggestion survives to the end of the turn and
|
||
the next sync either confirms the shape or stamps it vanished. No repo
|
||
key → only existing rows carry a suggestion.
|
||
|
||
Every pulled canon the payload NAMES is still recorded as a `uses` edge:
|
||
that is a fact about the code, not a verdict about what it is.
|
||
|
||
When more than one pulled snippet is in play for a shape, a by-name
|
||
reference beats resemblance and the most recent pull breaks ties.
|
||
"""
|
||
from grimoire.services import access
|
||
from grimoire.services import snippets as snippets_svc
|
||
from grimoire.services.snippets import snippet_fields
|
||
|
||
resembles = resembles or {}
|
||
path = (path or "").strip()
|
||
wanted = [(k, n.strip()) for k, n in shapes if k in ("css", "sym") and n.strip()]
|
||
if not project_id or not path or not wanted or not pulled:
|
||
return []
|
||
if not await access.can_write_project(user_id, project_id):
|
||
return []
|
||
|
||
# Which pulled canons are in play for this payload, by kind, ranked.
|
||
#
|
||
# THE FORM OF THE CANON TRAVELS WITH IT, because what gets written
|
||
# here is an ASSERTION — "this shape IS that canon" — and it is permanent
|
||
# until something re-judges it. The evidence below is payload-level: a
|
||
# resemblance score is computed against the whole file, so without a
|
||
# per-shape test every symbol in that file inherits one verdict. Measured
|
||
# on another project: a single write stamped `class SessionAbsent`,
|
||
# `async def attach` and `_run_control_client` as instances of one snippet.
|
||
in_play: dict[str, list[tuple[int, datetime, int, str, str]]] = {}
|
||
for sid, pulled_at in pulled.items():
|
||
note = await snippets_svc.get_snippet(user_id, sid)
|
||
if note is None:
|
||
continue
|
||
fields = snippet_fields(note)
|
||
symbol = fields.get("symbol") or ""
|
||
kind = snippet_kind(symbol, fields.get("language") or "")
|
||
if references_symbol(code, symbol, kind):
|
||
rank, why = 2, _REFERENCE_REASON.format(
|
||
sid=sid, symbol=_norm_symbol(symbol)
|
||
)
|
||
elif resembles.get(sid, 0.0) >= _RESEMBLE_MIN:
|
||
rank, why = 1, _RESEMBLE_REASON.format(sid=sid, score=resembles[sid])
|
||
else:
|
||
# A SCORE BELOW THE FLOOR IS NOT WEAK EVIDENCE, IT IS NONE.
|
||
# Taking any score at all (`elif sid in resembles`) would count
|
||
# 0.69 exactly as much as 0.95, and a row written on that basis is
|
||
# indistinguishable afterwards from one written on real evidence.
|
||
continue
|
||
in_play.setdefault(kind, []).append(
|
||
(rank, pulled_at, sid, why, shape_form(fields.get("signature") or "", kind))
|
||
)
|
||
if not in_play:
|
||
return []
|
||
for bucket in in_play.values():
|
||
bucket.sort(key=lambda t: (t[0], t[1]), reverse=True)
|
||
|
||
suggested: list[dict] = []
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.path == path,
|
||
CodeShape.vanished_at.is_(None),
|
||
)
|
||
)
|
||
).scalars().all()
|
||
by_key = {(r.symbol, r.kind): r for r in rows}
|
||
for kind, name in wanted:
|
||
bucket = in_play.get(kind)
|
||
if not bucket:
|
||
continue
|
||
row = by_key.get((name, kind))
|
||
# PER SHAPE, NOT PER FILE. The best candidate whose FORM
|
||
# matches this shape's — not simply the best candidate. A class and
|
||
# a function in one file are never handed the same canon because
|
||
# the file as a whole resembled it.
|
||
#
|
||
# The signature comes from the payload first: a shape being written
|
||
# right now may have no ledger row yet, and then the code is the
|
||
# only place it exists. No readable signature on either side means
|
||
# no stamp — `forms_agree` requires both to be known, so an
|
||
# unreadable shape falls silent instead of matching everything.
|
||
mine = shape_form(
|
||
signature_in(code, name, kind) or getattr(row, "signature", "") or "", kind
|
||
)
|
||
# THE BURDEN SCALES WITH THE EVIDENCE. Rank 2 is an explicit
|
||
# by-name reference to the canon in this very payload; it stands
|
||
# unless the forms actively contradict. Rank 1 is a similarity
|
||
# score over the whole file — one number that would otherwise
|
||
# speak for every symbol in it — so it must positively agree.
|
||
cand = next(
|
||
(t for t in bucket if _stamp_allowed(t[0], mine, t[4])), None
|
||
)
|
||
if cand is None:
|
||
continue
|
||
_rank, _at, sid, why, _cform = cand
|
||
if row is None:
|
||
if not repo_key:
|
||
continue
|
||
row = CodeShape(
|
||
project_id=project_id, repo_key=repo_key,
|
||
path=path, symbol=name, kind=kind,
|
||
)
|
||
session.add(row)
|
||
by_key[(name, kind)] = row
|
||
elif not (row.status in _MECHANICAL_TODO or row.classified_by == "hook"):
|
||
continue # a judgment — or the canon itself — stands
|
||
# A proposal, never a status: the writer decides.
|
||
row.proposed_snippet_id = sid
|
||
row.proposal_basis = "reference" if _rank >= 2 else "semantic"
|
||
row.proposal_score = resembles.get(sid) if _rank < 2 else 1.0
|
||
row.proposal_group = None
|
||
suggested.append({
|
||
"path": path, "symbol": name, "kind": kind,
|
||
"snippet_id": sid, "reason": why,
|
||
})
|
||
# Every pulled canon the payload NAMES is a uses edge — the
|
||
# call-site fact, independent of which one the row is judged to be.
|
||
await record_uses(
|
||
session, row,
|
||
# Indexed, not destructured: the bucket tuples carry the
|
||
# canon's form as a fifth field, and a positional unpack breaks
|
||
# silently whenever they widen. The unit lane never reaches
|
||
# `record_uses`; only the integration lane would catch it.
|
||
[t[2] for t in bucket if t[0] == 2],
|
||
basis="hook", evidence="write path: pulled the snippet, payload names its symbol",
|
||
)
|
||
if suggested:
|
||
await session.commit()
|
||
return suggested
|
||
|
||
|
||
# --- the mechanical proposer: the machine proposes, judgment classifies
|
||
#
|
||
# Runs inside the coverage refresh — the one moment the shape BODIES exist
|
||
# (the archive is in memory; the ledger stores fingerprints, never code).
|
||
# Over every live unclassified row whose content changed since it was last
|
||
# examined, it tries, strongest first:
|
||
#
|
||
# symbol the shape bears a recorded canon's symbol at another location
|
||
# (a second definition of the canon's name — an instance or a
|
||
# duplicate to consolidate; either way, it answers to #N);
|
||
# text whitespace-insensitive containment either way between the
|
||
# body and the canon's recorded code (git-grep, in effect —
|
||
# a verbatim copy, or the canon's own call-site example);
|
||
# reference the body calls/uses a canon's symbol — the call-site shape;
|
||
# signature a sym whose definition line, name blanked, resembles the
|
||
# canon's (the move_*/resequence family shape);
|
||
# semantic the widest net: the body (concept-queried, as the write-path
|
||
# arm does) scores above the write-path threshold against a
|
||
# canon — capped per refresh, so the cost is bounded and a big
|
||
# ledger is worked through across refreshes.
|
||
#
|
||
# A hit is a PROPOSAL on the row (proposed_snippet_id/basis/score), never a
|
||
# classification: an agent confirms in batches (confirm_proposals) or judges
|
||
# otherwise (classify_shapes clears it). Rows with no canon hit are grouped
|
||
# by the derive-first rule: the same body fingerprint in ≥2
|
||
# places, or the same name defined in ≥3 files, is "a repeating shape with
|
||
# no canon — derive one first", carried as proposal_basis="derive" + a group
|
||
# key so sessions see the consolidation candidates as one thing.
|
||
|
||
# Derive-first floors. Identical bodies twice is already a copy; a bare name
|
||
# needs more repetition before it reads as a family (setup/handler/register
|
||
# recur by convention, not by duplication).
|
||
_DERIVE_MIN_DUP = 2
|
||
_DERIVE_MIN_NAME = 3
|
||
# CSS is never grouped by body: classes for different purposes
|
||
# share declarations because the style system makes them alike — `.text-muted`
|
||
# and `.pin-badge-auto` carrying the same `color: var(--fs-text-tertiary)` are
|
||
# two meanings, not two copies. A CSS family is a NAME defined in more than
|
||
# one file: that is a recipe living in several places, and two is already
|
||
# the signal (a class name is deliberate in a way `setup`/`load` are not).
|
||
_DERIVE_MIN_NAME_CSS = 2
|
||
# Semantic checks per repo per refresh — an embedding each (local fastembed),
|
||
# bounded so a 4,000-row ledger is worked through over refreshes, not in one.
|
||
_SEMANTIC_CAP = 150
|
||
# The semantic arm has NO floor of its own: it uses the write-path surface's,
|
||
# because it asks the write-path hint's question — "which canon does this body
|
||
# resemble?" — of the same snippet documents.
|
||
#
|
||
# A higher floor (0.8) would be sized for a proposal nobody reads, and that is
|
||
# not how proposals are consumed: a judge reviews each semantic one
|
||
# (confirm_proposals refuses an unnamed batch, and the tool says "semantic
|
||
# deserves a look"). Judged instances of a canon typically score 0.68–0.71
|
||
# against it, so 0.8 proposes nothing. A wrong proposal costs the judge a
|
||
# look; a floor that yields nothing costs the signal. Noise in this band is
|
||
# expected — the score rides on the proposal so the judge can weigh it.
|
||
#
|
||
# How many hits the arm scans for an ALLOWED canon. Wider than the hint's
|
||
# budget because disallowed snippets (other language families) share the
|
||
# ranking: a true instance can sit 4th, behind three it could not use.
|
||
_SEMANTIC_LIMIT = 8
|
||
# A MISS FROM THIS ARM IS NOT EVIDENCE, and nothing may read it as one.
|
||
# Letting "compared, nothing cleared the floor" withdraw a divergence prompt
|
||
# silences real divergences exactly as it silences helpers: many judged
|
||
# instances of a canon score below any floor that keeps noise out (0.66 and
|
||
# under), and helpers score 0.66–0.75 against snippets they have nothing to do
|
||
# with. A code body against a prose-forward snippet document measures the
|
||
# wrong field; no floor separates the two bands. The arm PROPOSES on a hit;
|
||
# its silence says nothing.
|
||
#
|
||
# THE FAMILY ARM. The arm above is held to the shape's own project and
|
||
# language family, because across projects it is noise. A family idea is the
|
||
# exception: an idea shared across projects ON PURPOSE, whose reference may be
|
||
# in another language (a Python throttle is an instance of a Go one). So the
|
||
# arm also proposes the reference of a canon idea that reaches the shape's
|
||
# project through a shared platform — any language, any project — under a
|
||
# stricter rule of its own, calibrated on cross-project pairs:
|
||
#
|
||
# true pairs A py→go 0.728 (next 0.703) B py→go 0.715 (next 0.675)
|
||
# D kt→kt 0.816 (next 0.734) E go→go 0.610 (next 0.604)
|
||
# no reference N1 0.666 N2 0.623 N3 0.626 N4 0.715 N5 0.614 (best hit,
|
||
# none of them a family reference); unrelated hits to 0.774
|
||
#
|
||
# No floor separates those bands — the false band reaches 0.715–0.774, a
|
||
# true pair sits at 0.610. RANK does: every true pair was the single best
|
||
# snippet the user can read. So a family reference is proposed only when it
|
||
# is the TOP hit over everything readable (not merely the first allowed one,
|
||
# as above) and clears _FAMILY_FLOOR: A, B, D proposed; E missed; no
|
||
# negative proposed. A miss is still not evidence — the judge can classify
|
||
# against the idea by idea_id.
|
||
_FAMILY_FLOOR = 0.70
|
||
|
||
|
||
def _semantic_priority(row) -> tuple:
|
||
"""Order for the capped semantic pass: new shapes first.
|
||
|
||
The pass is capped (`_SEMANTIC_CAP`), and `flag_divergence` acts only on
|
||
shapes NEW since the previous refresh — and skips one the proposer matched
|
||
to the dominant canon. In plain row order the new shapes are the highest
|
||
ids, the back of the queue, so whenever the backlog exceeds the cap (every
|
||
`_PROPOSER_VERSION` bump queues the whole todo at once) they would be
|
||
flagged before the arm could propose for them.
|
||
|
||
So: the human todo (`unclassified`) before `scoped`, newest first within
|
||
each. A row with no creation time sorts last in its group.
|
||
"""
|
||
created = row.created_at.timestamp() if row.created_at is not None else float("-inf")
|
||
return (row.status != "unclassified", -created)
|
||
# Bump when a basis's rule changes: rows remember the (body, ruleset) they
|
||
# were examined under, so a tightened rule re-examines everything once. A
|
||
# change that does not alter what is proposed needs no bump. A stale "miss"
|
||
# basis on an old row is inert: a miss is not evidence (see _SEMANTIC_LIMIT).
|
||
_PROPOSER_VERSION = 6
|
||
# Signature resemblance floor, name blanked (difflib ratio) — and a length
|
||
# floor, because `def NAME():` resembles `def NAME(x):` at 0.95 while saying
|
||
# nothing; a family shape has parameters to resemble.
|
||
#
|
||
# A proposal is read before it is confirmed, so this admits the band where
|
||
# true instances live rather than the band where nothing is wrong. Over
|
||
# judged instance rows against every same-family canon, true instances score
|
||
# 1.0 or ~0.795 (a model class whose mixin list differs from the canon's), and
|
||
# 0.8 cuts the whole second band. At 0.75 all
|
||
# of them clear; the price is `async def NAME(user_id: int, …) -> dict`
|
||
# look-alikes (35 wrong-canon pairings vs 14 at 0.8), which read as wrong at
|
||
# a glance. Below 0.75 the next true row is at 0.6 behind 150+ wrong ones.
|
||
_SIGNATURE_FLOOR = 0.75
|
||
_SIGNATURE_MIN_LEN = 30
|
||
# Textual containment needs enough substance to mean anything.
|
||
_TEXT_FLOOR = 40
|
||
_BASIS_ORDER = ("symbol", "text", "reference", "signature")
|
||
|
||
|
||
class Canon(NamedTuple):
|
||
snippet_id: int
|
||
kind: str
|
||
symbol: str
|
||
locations: tuple[tuple[str, str], ...]
|
||
signature: str
|
||
code_norm: str
|
||
project_id: int = 0
|
||
language: str = "" # the snippet's recorded language; "" = unknown, no gate
|
||
|
||
|
||
# Language families: the sym bases only propose within one. Cross-language
|
||
# hits are wrong — a Python tool-module canon named `register` would be
|
||
# proposed for Vue `handleSubmit`s that call `authStore.register()`, a TS
|
||
# store's `register` would match it by symbol, and TS canon would be proposed
|
||
# for Python bodies by resemblance. CSS is its own kind and is not gated here.
|
||
_FAMILY_BY_LANG = {
|
||
"python": "py", "py": "py",
|
||
"typescript": "js", "ts": "js", "tsx": "js", "javascript": "js", "js": "js",
|
||
"jsx": "js", "vue": "js", "mjs": "js", "cjs": "js",
|
||
"css": "css", "scss": "css", "sass": "css", "less": "css",
|
||
"bash": "sh", "sh": "sh", "shell": "sh", "zsh": "sh",
|
||
"sql": "sql",
|
||
}
|
||
_FAMILY_BY_EXT = {
|
||
".py": "py", ".pyi": "py",
|
||
".ts": "js", ".tsx": "js", ".js": "js", ".jsx": "js", ".vue": "js", ".mjs": "js", ".cjs": "js",
|
||
".css": "css", ".scss": "css", ".sass": "css", ".less": "css",
|
||
".sh": "sh", ".bash": "sh", ".zsh": "sh",
|
||
".sql": "sql",
|
||
}
|
||
|
||
|
||
def language_family(language: str) -> str:
|
||
"""The family a recorded snippet language belongs to ("" when unknown)."""
|
||
return _FAMILY_BY_LANG.get((language or "").strip().lower(), "")
|
||
|
||
|
||
def path_family(path: str) -> str:
|
||
"""The family a file path belongs to, by extension ("" when unknown)."""
|
||
p = (path or "").lower()
|
||
for ext, fam in _FAMILY_BY_EXT.items():
|
||
if p.endswith(ext):
|
||
return fam
|
||
return ""
|
||
|
||
|
||
def same_family(path: str, canon_language: str) -> bool:
|
||
"""A sym basis may propose this canon for this path: both families known
|
||
and equal, or either unknown (no evidence either way → no gate)."""
|
||
a = path_family(path)
|
||
b = language_family(canon_language)
|
||
return not a or not b or a == b
|
||
|
||
|
||
# Reference basis: generic verbs name too many unrelated things to count a
|
||
# bare mention as a call site of THIS canon (`register`, `load`, `save` …).
|
||
# The symbol basis still catches a second definition of such a name.
|
||
_REFERENCE_STOPLIST = frozenset({
|
||
"get", "set", "put", "post", "load", "save", "run", "main", "init", "setup",
|
||
"register", "restore", "reset", "toggle", "close", "open", "submit", "handler",
|
||
"update", "create", "delete", "remove", "add", "start", "stop", "send",
|
||
"receive", "render", "mount", "dispatch", "call", "apply", "execute",
|
||
})
|
||
|
||
|
||
def _norm_text(text: str) -> str:
|
||
return " ".join((text or "").split())
|
||
|
||
|
||
def signature_similarity(sig_a: str, name_a: str, sig_b: str, name_b: str) -> float:
|
||
"""How alike two definition lines are once their own names are blanked —
|
||
`def move_event(project_id, event_id, after_id)` against
|
||
`def move_beat(project_id, beat_id, after_id)` reads high."""
|
||
a = _norm_text(sig_a)
|
||
b = _norm_text(sig_b)
|
||
if name_a:
|
||
a = a.replace(name_a, "NAME")
|
||
if name_b:
|
||
b = b.replace(name_b, "NAME")
|
||
if len(a) < _SIGNATURE_MIN_LEN or len(b) < _SIGNATURE_MIN_LEN:
|
||
return 0.0
|
||
return difflib.SequenceMatcher(None, a, b).ratio()
|
||
|
||
|
||
def text_contains(body: str, code: str) -> bool:
|
||
"""Whitespace-insensitive containment, either way, above a substance
|
||
floor — the proposer's git-grep."""
|
||
a = _norm_text(body)
|
||
b = _norm_text(code)
|
||
if len(a) < _TEXT_FLOOR or len(b) < _TEXT_FLOOR:
|
||
return False
|
||
return a in b or b in a
|
||
|
||
|
||
def reference_canons(kind: str, path: str, symbol: str, body: str, canons: Iterable[Canon]) -> list[int]:
|
||
"""Every canon this body NAMES — the uses edges the proposer can
|
||
write mechanically: same kind, same language family, symbol not in the
|
||
generic-verb stoplist, and not the shape's own name."""
|
||
norm_sym = _norm_symbol(symbol)
|
||
out: list[int] = []
|
||
for c in canons:
|
||
if c.kind != kind or not c.symbol:
|
||
continue
|
||
if kind == "sym" and not same_family(path, c.language):
|
||
continue
|
||
if _norm_symbol(c.symbol) == norm_sym:
|
||
continue
|
||
if _norm_symbol(c.symbol).lower() in _REFERENCE_STOPLIST:
|
||
continue
|
||
if references_symbol(body, c.symbol, kind):
|
||
out.append(c.snippet_id)
|
||
return out
|
||
|
||
|
||
def match_canon(
|
||
kind: str, path: str, symbol: str, signature: str, body: str,
|
||
canons: Iterable[Canon], *, project_id: int = 0,
|
||
) -> tuple[int, str, float] | None:
|
||
"""The strongest (snippet_id, basis, score) a shape earns against the
|
||
canon catalog, by the non-semantic bases — or None. Strongest by
|
||
basis order, then by score within the basis; on a tie, a canon recorded
|
||
in the shape's own project beats family canon from another (the same
|
||
helper recorded in two projects is the shape's own first)."""
|
||
best: dict[str, tuple[float, bool, int]] = {}
|
||
|
||
def offer(basis: str, score: float, canon: Canon) -> None:
|
||
cur = best.get(basis)
|
||
same = bool(project_id) and canon.project_id == project_id
|
||
if cur is None or (score, same) > (cur[0], cur[1]):
|
||
best[basis] = (score, same, canon.snippet_id)
|
||
|
||
norm_sym = _norm_symbol(symbol)
|
||
for c in canons:
|
||
if c.kind != kind:
|
||
continue
|
||
if kind == "sym" and not same_family(path, c.language):
|
||
continue # a Python canon says nothing about a Vue body, and vice versa
|
||
if c.symbol and _norm_symbol(c.symbol) == norm_sym:
|
||
if not any(location_covers(lp, ls, path, symbol) for lp, ls in c.locations):
|
||
offer("symbol", 1.0, c)
|
||
continue # its own location is canonical territory, not a proposal
|
||
if (
|
||
c.symbol
|
||
and _norm_symbol(c.symbol).lower() not in _REFERENCE_STOPLIST
|
||
and references_symbol(body, c.symbol, kind)
|
||
):
|
||
offer("reference", 0.9, c)
|
||
if c.code_norm and text_contains(body, c.code_norm):
|
||
offer("text", 0.95, c)
|
||
if kind == "sym" and c.signature:
|
||
ratio = signature_similarity(signature, symbol, c.signature, c.symbol)
|
||
if ratio >= _SIGNATURE_FLOOR:
|
||
offer("signature", round(ratio, 3), c)
|
||
for basis in _BASIS_ORDER:
|
||
if basis in best:
|
||
score, _same, sid = best[basis]
|
||
return (sid, basis, score)
|
||
return None
|
||
|
||
|
||
async def canon_catalog(user_id: int) -> list[Canon]:
|
||
"""Every snippet this user can browse, as matchable canon — own projects
|
||
and shared ones alike, because family canon counts."""
|
||
from grimoire.models.note import Note
|
||
from grimoire.services.access import browsable_notes_clause
|
||
from grimoire.services.coverage import extract_definitions
|
||
from grimoire.services.snippets import SNIPPET_NOTE_TYPE, snippet_fields
|
||
|
||
async with async_session() as session:
|
||
notes = (
|
||
await session.execute(
|
||
select(Note).where(
|
||
browsable_notes_clause(user_id),
|
||
Note.note_type == SNIPPET_NOTE_TYPE,
|
||
Note.deleted_at.is_(None),
|
||
)
|
||
)
|
||
).scalars().all()
|
||
out: list[Canon] = []
|
||
for note in notes:
|
||
fields = snippet_fields(note)
|
||
symbol = (fields.get("symbol") or "").strip()
|
||
kind = snippet_kind(symbol, fields.get("language") or "")
|
||
code = fields.get("code") or ""
|
||
# The signature basis matches against the canon's OWN definition
|
||
# line — never against an example in its code. A snippet whose code
|
||
# opens with a call-site example (confirmed()'s does) would otherwise
|
||
# make every `async function x(): Promise<void>` resemble it.
|
||
own = next(
|
||
(d for d in extract_definitions(code)
|
||
if symbol and _norm_symbol(d.name) == _norm_symbol(symbol)),
|
||
None,
|
||
)
|
||
signature = own.signature if own else ""
|
||
out.append(Canon(
|
||
int(note.id), kind, symbol,
|
||
tuple(
|
||
((loc.get("path") or ""), (loc.get("symbol") or ""))
|
||
for loc in fields.get("locations") or []
|
||
),
|
||
signature, _norm_text(code), int(note.project_id or 0),
|
||
(fields.get("language") or "").strip().lower(),
|
||
))
|
||
return out
|
||
|
||
|
||
def _clear_proposal(row: CodeShape, *, reexamine: bool = False) -> None:
|
||
row.proposed_snippet_id = None
|
||
row.proposal_basis = None
|
||
row.proposal_score = None
|
||
row.proposal_group = None
|
||
if reexamine:
|
||
row.proposed_at = None
|
||
row.proposed_sha = ""
|
||
|
||
|
||
def _substance(text: str) -> int:
|
||
return len("".join((text or "").split()))
|
||
|
||
|
||
def pick_semantic(
|
||
hits: list[tuple[float, int]], allowed: set[int], family: set[int], *, floor: float,
|
||
) -> tuple[int, float, str] | None:
|
||
"""Which ranked hit the semantic arm proposes, as (snippet_id, score,
|
||
basis). Pure; ``hits`` are (score, snippet_id) over everything the user
|
||
can read, best first. The TOP hit may be a family reference (basis
|
||
"family", at _FAMILY_FLOOR); any hit at ``floor`` may be an own-project
|
||
canon (basis "semantic")."""
|
||
for rank, (score, sid) in enumerate(hits):
|
||
if rank == 0 and sid in family and score >= _FAMILY_FLOOR:
|
||
return sid, round(float(score), 3), "family"
|
||
if sid in allowed and score >= floor:
|
||
return sid, round(float(score), 3), "semantic"
|
||
return None
|
||
|
||
|
||
async def _semantic_canon(
|
||
user_id: int, body: str, allowed: set[int], family: set[int] | None = None,
|
||
) -> tuple[int, float, str] | None:
|
||
"""The canon this body MEANS, as (snippet_id, score, basis), or None.
|
||
None is "no proposal", never "not the canon" — see the note above
|
||
`_SEMANTIC_LIMIT`; the family arm's rule is the note above
|
||
`_FAMILY_FLOOR`."""
|
||
from grimoire.services.embeddings import semantic_search_notes
|
||
from grimoire.services.plugin_context import (
|
||
WRITEPATH_DEFAULT_THRESHOLD, WRITEPATH_MIN_CODE_CHARS, concept_query,
|
||
)
|
||
|
||
family = family or set()
|
||
if _substance(body) < WRITEPATH_MIN_CODE_CHARS or not (allowed or family):
|
||
return None
|
||
query = concept_query(body) or body
|
||
hits = await semantic_search_notes(
|
||
user_id, query, limit=_SEMANTIC_LIMIT,
|
||
threshold=min(WRITEPATH_DEFAULT_THRESHOLD, _FAMILY_FLOOR),
|
||
note_type="snippet", scope="browse",
|
||
)
|
||
return pick_semantic(
|
||
[(float(score), int(note.id)) for score, note in hits], allowed, family,
|
||
floor=WRITEPATH_DEFAULT_THRESHOLD,
|
||
)
|
||
|
||
|
||
async def propose_for_repo(
|
||
user_id: int,
|
||
project_id: int,
|
||
repo_key: str,
|
||
definitions: list,
|
||
*,
|
||
canons: list[Canon] | None = None,
|
||
semantic_cap: int = _SEMANTIC_CAP,
|
||
) -> dict:
|
||
"""Examine one repo's live unclassified rows against canon and record
|
||
proposals. ``definitions`` are the ArchiveShape records the sync just
|
||
upserted (their bodies are the matching material). Rows whose fingerprint
|
||
is unchanged since their last examination are skipped; rows that only
|
||
the capped semantic pass could not reach stay unexamined, so the next
|
||
refresh reaches the next slice. Returns counts."""
|
||
if canons is None:
|
||
canons = await canon_catalog(user_id)
|
||
by_key = {(d[0], d[1], d[2]): d for d in definitions}
|
||
# The semantic arm is the widest net and, across projects, is noise: it
|
||
# is held to the shape's own project and language family. The precise
|
||
# bases (symbol/text) still reach family canon in other projects.
|
||
sym_canons = [c for c in canons if c.kind == "sym" and c.project_id == project_id]
|
||
|
||
def semantic_allowed(path: str) -> set[int]:
|
||
return {c.snippet_id for c in sym_canons if same_family(path, c.language)}
|
||
# The family arm: references of canon ideas on a platform this project
|
||
# shares, in any language (see _FAMILY_FLOOR). A project on no shared
|
||
# platform gets none, so nothing is proposed across to it.
|
||
try:
|
||
from grimoire.services import family_adoption
|
||
|
||
family_refs = await family_adoption.family_reference_ids(project_id)
|
||
except Exception:
|
||
logger.warning("family references unreadable for project %s", project_id, exc_info=True)
|
||
family_refs = set()
|
||
now = datetime.now(timezone.utc)
|
||
examined = proposed = checked = 0
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.repo_key == repo_key,
|
||
CodeShape.vanished_at.is_(None),
|
||
)
|
||
)
|
||
).scalars().all()
|
||
semantic_todo: list[tuple[CodeShape, object]] = []
|
||
for row in rows:
|
||
d = by_key.get((row.path, row.kind, row.symbol))
|
||
if d is None:
|
||
continue
|
||
signature, body_sha, body = d[3], d[4], d[5]
|
||
examined_as = f"{body_sha}@{_PROPOSER_VERSION}"
|
||
if row.proposed_at is not None and row.proposed_sha == examined_as:
|
||
continue
|
||
if row.status not in _MECHANICAL_TODO:
|
||
# A judged row gets no proposal — but its uses edges
|
||
# are a fact about the body, judged or not: the consumer map
|
||
# of a canon must include the call sites someone already
|
||
# classified. Mark it examined so the scan runs once per body.
|
||
used = reference_canons(row.kind, row.path, row.symbol, body, canons)
|
||
if used:
|
||
await record_uses(session, row, used, basis="reference",
|
||
evidence="proposer: body names the canon's symbol")
|
||
row.proposed_at = now
|
||
row.proposed_sha = examined_as
|
||
continue
|
||
examined += 1
|
||
group = row.proposal_group # derive grouping is reassigned below
|
||
hit = match_canon(
|
||
row.kind, row.path, row.symbol, signature, body, canons,
|
||
project_id=project_id,
|
||
)
|
||
_clear_proposal(row)
|
||
row.proposal_group = group
|
||
row.proposed_at = now
|
||
row.proposed_sha = examined_as
|
||
# Consumption is recorded for every canon the body names,
|
||
# whatever the row is then judged to be.
|
||
used = reference_canons(row.kind, row.path, row.symbol, body, canons)
|
||
if used:
|
||
await record_uses(session, row, used, basis="reference",
|
||
evidence="proposer: body names the canon's symbol")
|
||
if hit:
|
||
row.proposed_snippet_id, row.proposal_basis, row.proposal_score = hit
|
||
row.proposal_group = None
|
||
proposed += 1
|
||
elif row.kind == "sym":
|
||
semantic_todo.append((row, d))
|
||
semantic_todo.sort(key=lambda item: _semantic_priority(item[0]))
|
||
for i, (row, d) in enumerate(semantic_todo):
|
||
if i >= semantic_cap:
|
||
# Not reached this refresh: leave it unexamined so the next
|
||
# refresh picks it up, rather than stamping a false "nothing".
|
||
row.proposed_at = None
|
||
row.proposed_sha = ""
|
||
continue
|
||
checked += 1
|
||
try:
|
||
found = await _semantic_canon(
|
||
user_id, d[5], semantic_allowed(row.path), family_refs)
|
||
except Exception:
|
||
logger.warning("semantic proposal failed", exc_info=True)
|
||
found = None
|
||
if found:
|
||
row.proposed_snippet_id, row.proposal_score, row.proposal_basis = found
|
||
row.proposal_group = None
|
||
proposed += 1
|
||
await session.commit()
|
||
return {"examined": examined, "proposed": proposed, "semantic_checked": checked}
|
||
|
||
|
||
def derive_groups(
|
||
rows: Iterable[tuple[str, str, str, str]]
|
||
) -> dict[tuple[str, str, str], str]:
|
||
"""The derive-first grouping over (path, kind, symbol, body_sha) rows
|
||
that matched no canon: {(path, kind, symbol): group_key}. For code
|
||
(kind `sym`) identical bodies in ≥2 places group as `dup:<sha>` and the
|
||
same name defined in ≥3 files groups as `name:sym:<symbol>`, the copy
|
||
before the name. CSS groups by name only — the same class defined in
|
||
≥2 files is `name:css:<symbol>`; its body never groups it."""
|
||
by_sha: dict[str, list[tuple[str, str, str]]] = {}
|
||
by_name: dict[tuple[str, str], list[tuple[str, str, str]]] = {}
|
||
for path, kind, symbol, sha in rows:
|
||
key = (path, kind, symbol)
|
||
if sha and kind != "css":
|
||
by_sha.setdefault(sha, []).append(key)
|
||
by_name.setdefault((kind, _norm_symbol(symbol)), []).append(key)
|
||
out: dict[tuple[str, str, str], str] = {}
|
||
for sha, keys in by_sha.items():
|
||
if len(set(keys)) >= _DERIVE_MIN_DUP:
|
||
for key in keys:
|
||
out.setdefault(key, f"dup:{sha}")
|
||
for (kind, symbol), keys in by_name.items():
|
||
floor = _DERIVE_MIN_NAME_CSS if kind == "css" else _DERIVE_MIN_NAME
|
||
if len({k[0] for k in keys}) >= floor:
|
||
for key in keys:
|
||
out.setdefault(key, f"name:{kind}:{symbol}")
|
||
return out
|
||
|
||
|
||
async def apply_derive_groups(project_id: int) -> int:
|
||
"""Recompute derive-first groups over the project's live unclassified
|
||
rows that carry no canon proposal; returns how many rows are grouped."""
|
||
now = datetime.now(timezone.utc)
|
||
grouped = 0
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.status.in_(_MECHANICAL_TODO),
|
||
CodeShape.vanished_at.is_(None),
|
||
CodeShape.proposed_snippet_id.is_(None),
|
||
# A file row is named by its stem, and `index`, `+page`
|
||
# and `__init__` repeat by convention, not by copying —
|
||
# a derive family of them would be pure noise.
|
||
CodeShape.kind != "file",
|
||
)
|
||
)
|
||
).scalars().all()
|
||
groups = derive_groups(
|
||
(r.path, r.kind, r.symbol, r.body_sha or "") for r in rows
|
||
)
|
||
sizes: dict[str, int] = {}
|
||
for g in groups.values():
|
||
sizes[g] = sizes.get(g, 0) + 1
|
||
for row in rows:
|
||
key = groups.get((row.path, row.kind, row.symbol))
|
||
if key:
|
||
row.proposal_basis = "derive"
|
||
row.proposal_group = key
|
||
row.proposal_score = float(sizes[key])
|
||
if row.proposed_at is None:
|
||
row.proposed_at = now
|
||
grouped += 1
|
||
elif row.proposal_group:
|
||
row.proposal_basis = None
|
||
row.proposal_group = None
|
||
row.proposal_score = None
|
||
await session.commit()
|
||
return grouped
|
||
|
||
|
||
def proposal_summary(
|
||
rows: Iterable[CodeShape], *, top: int = 8,
|
||
consumer_paths: dict[int, list[str]] | None = None,
|
||
) -> dict:
|
||
"""The readout's view of the proposer's standing: how many canon
|
||
proposals await confirmation, and the largest derive-first groups.
|
||
``consumer_paths`` (shape_id → files whose markup names it, milestone
|
||
302) puts `consumers` on each group — the family's distinct consumer
|
||
files across its members, the datum that separates a shared recipe
|
||
from a scoped convention."""
|
||
proposed = 0
|
||
by_canon: dict[int, int] = {}
|
||
groups: dict[str, dict] = {}
|
||
files: dict[str, set[str]] = {}
|
||
consumers: dict[str, set[str]] = {}
|
||
for row in rows:
|
||
if row.status not in _MECHANICAL_TODO:
|
||
continue
|
||
if row.proposed_snippet_id is not None:
|
||
proposed += 1
|
||
by_canon[row.proposed_snippet_id] = by_canon.get(row.proposed_snippet_id, 0) + 1
|
||
elif row.proposal_group:
|
||
dup = not row.proposal_group.startswith("name:")
|
||
g = groups.setdefault(row.proposal_group, {
|
||
"group": row.proposal_group, "kind": row.kind,
|
||
"label": (
|
||
f"{row.symbol} (identical body)" if dup
|
||
else ("." if row.kind == "css" else "") + row.symbol
|
||
),
|
||
"size": 0, "files": 0, "paths": [],
|
||
})
|
||
g["size"] += 1
|
||
files.setdefault(row.proposal_group, set()).add(row.path)
|
||
if len(g["paths"]) < 3:
|
||
g["paths"].append(row.path)
|
||
if consumer_paths is not None and row.kind == "css":
|
||
consumers.setdefault(row.proposal_group, set()).update(
|
||
consumer_paths.get(row.id) or ()
|
||
)
|
||
for key, g in groups.items():
|
||
g["files"] = len(files[key])
|
||
if key in consumers:
|
||
g["consumers"] = consumer_summary(consumers[key])
|
||
# Body-identical groups first: the things an audit actually
|
||
# consolidated were identical bodies under different names/files; a
|
||
# name repeated across modules is usually convention. Within a tier,
|
||
# the group spread over more files is the bigger copy.
|
||
ranked = sorted(
|
||
groups.values(),
|
||
key=lambda g: (g["group"].startswith("name:"), -g["files"], -g["size"], g["group"]),
|
||
)
|
||
top_canon = None
|
||
if by_canon:
|
||
sid, n = max(by_canon.items(), key=lambda kv: (kv[1], -kv[0]))
|
||
top_canon = {"snippet_id": sid, "count": n}
|
||
return {"proposed": proposed, "derive_groups": ranked[:top], "top_canon": top_canon}
|
||
|
||
|
||
def derive_new_summary(
|
||
rows: Iterable[CodeShape], *, since: datetime | None, top: int = 3
|
||
) -> dict:
|
||
"""The arrival-moment drift signal: derive-grouped rows FIRST
|
||
SEEN after ``since`` — the previous refresh's stamp, the same one
|
||
flag_divergence uses. "Since the last refresh, N more copies joined a
|
||
duplicate family" is the sentence that makes the derive queue a thing
|
||
you notice on entering, not a thing an audit finds. ``since`` None (a
|
||
first seed) means nothing is new. Judged rows never count."""
|
||
if since is None:
|
||
return {"count": 0, "examples": []}
|
||
fresh = [
|
||
r for r in rows
|
||
if r.proposal_basis == "derive" and r.proposal_group
|
||
and r.status in _MECHANICAL_TODO and r.vanished_at is None
|
||
and r.created_at is not None and r.created_at > since
|
||
]
|
||
fresh.sort(key=lambda r: r.created_at, reverse=True)
|
||
return {
|
||
"count": len(fresh),
|
||
"examples": [
|
||
{"label": ("." if r.kind == "css" else "") + r.symbol,
|
||
"path": r.path, "group": r.proposal_group}
|
||
for r in fresh[:top]
|
||
],
|
||
}
|
||
|
||
|
||
async def confirm_proposals(
|
||
user_id: int,
|
||
project_id: int,
|
||
*,
|
||
snippet_id: int = 0,
|
||
path: str = "",
|
||
basis: str = "",
|
||
min_score: float = 0.0,
|
||
) -> dict:
|
||
"""Turn reviewed canon proposals into `instance` rows, in one batch.
|
||
|
||
At least one of snippet_id / path / basis must narrow the batch — "confirm
|
||
everything proposed" without having looked is not a judgment. Each row
|
||
becomes instance-of-its-proposed-snippet, classified_by="agent", reason
|
||
naming the basis and score; the proposal is retired. Returns
|
||
{"confirmed": N}."""
|
||
from grimoire.services import access
|
||
|
||
if not (snippet_id or path.strip() or basis.strip()):
|
||
raise ValueError(
|
||
"name what you reviewed: confirm by snippet_id, path, and/or basis"
|
||
)
|
||
if not await access.can_write_project(user_id, project_id):
|
||
raise ValueError(f"project {project_id} not found or no write access")
|
||
from sqlalchemy import or_
|
||
|
||
conds = [
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.status.in_(_MECHANICAL_TODO),
|
||
CodeShape.vanished_at.is_(None),
|
||
CodeShape.proposed_snippet_id.isnot(None),
|
||
]
|
||
if snippet_id:
|
||
conds.append(CodeShape.proposed_snippet_id == snippet_id)
|
||
if path.strip():
|
||
clean = path.strip().strip("/")
|
||
conds.append(or_(CodeShape.path == clean, CodeShape.path.like(clean + "/%")))
|
||
if basis.strip():
|
||
conds.append(CodeShape.proposal_basis == basis.strip())
|
||
now = datetime.now(timezone.utc)
|
||
confirmed = 0
|
||
touched: set[int] = set()
|
||
async with async_session() as session:
|
||
rows = (await session.execute(select(CodeShape).where(*conds))).scalars().all()
|
||
for row in rows:
|
||
if (row.proposal_score or 0.0) < min_score:
|
||
continue
|
||
touched.add(row.proposed_snippet_id)
|
||
await _judge(
|
||
session, row, status="instance", snippet_id=row.proposed_snippet_id,
|
||
by="agent", at=now,
|
||
reason=(
|
||
f"confirmed {row.proposal_basis} proposal"
|
||
f" ({(row.proposal_score or 0.0):.2f})"
|
||
),
|
||
)
|
||
confirmed += 1
|
||
await session.commit()
|
||
out = {"confirmed": confirmed}
|
||
moved = await _sync_family(user_id, project_id, touched)
|
||
if moved:
|
||
out["family"] = moved
|
||
return out
|
||
|
||
|
||
# --- the divergence readout: button B where button A is canon --------------
|
||
#
|
||
# Three answers the ledger can now give mechanically:
|
||
# DIVERGENCE 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 → `diverges_from=#N`.
|
||
# Read: "button B appeared where button A is canon — divergence
|
||
# or variant? classify it." Surfaced in the coverage readout and
|
||
# in-band at write time (the prior-art hook names the shapes).
|
||
# HISTORY every judgment / vanish / reappearance / drift is an event;
|
||
# shape_history answers "what was used here, when, and why".
|
||
# RECHECK an instance/variant whose body moved since it was judged is
|
||
# flagged recheck_at (sync) — the judgment stands, re-confirm it.
|
||
|
||
# How much a payload must resemble a canon before the hook SUGGESTS it as the
|
||
# shape's canon — and the bar below which a stored hook stamp (an `instance`
|
||
# row written unattended, classified_by="hook") is weak (`is_weak_stamp`).
|
||
# The hook itself never asserts a classification.
|
||
#
|
||
# WHY A FLOOR AT ALL. Without one, any score the semantic arm produced would
|
||
# count, 0.69 as confidently as 0.95. A stamp feeds `dominant_canon`, which
|
||
# tells the next writer in that directory what to build from, so a thin stamp
|
||
# does not stay thin, it compounds.
|
||
#
|
||
# 0.80 rather than the retrieval floors near 0.70, deliberately. Those bars
|
||
# decide whether to SHOW someone a record, where being wrong costs a glance.
|
||
# This one weighs a claim about the codebase, where being wrong costs a wrong
|
||
# instruction to everyone who writes in that directory afterwards. A claim
|
||
# should need more evidence than a suggestion to read, not the same.
|
||
_RESEMBLE_MIN = 0.80
|
||
|
||
|
||
def is_weak_stamp(row) -> bool:
|
||
"""Was this row written by the hook, unattended, on a resemblance below
|
||
the floor it would have to clear today?
|
||
|
||
ONE PREDICATE FOR "WEAK", read by the review surface that lists these rows
|
||
and by every reader that would otherwise take them as evidence. Rows
|
||
stamped below the floor stay stored, and until an agent re-judges them
|
||
they must not speak. They are not withdrawn here — `stamps_to_review`
|
||
lists them for judgment — they are simply not counted, which is the
|
||
module's rule for evidence it cannot trust: quieter, never more confident.
|
||
"""
|
||
if getattr(row, "classified_by", None) not in _UNATTENDED_BY:
|
||
return False
|
||
score = stamp_score(getattr(row, "reason", None))
|
||
return score is not None and score < _RESEMBLE_MIN
|
||
|
||
|
||
# A canon dominates a directory+kind when at least this many siblings are
|
||
# judged (canonical/instance) and this share of them answer to one snippet.
|
||
_DENSITY_MIN_JUDGED = 3
|
||
_DENSITY_SHARE = 0.6
|
||
|
||
|
||
def comparable_siblings(rows: Iterable, form: str) -> list:
|
||
"""The siblings a candidate of ``form`` can honestly be counted against.
|
||
|
||
Excludes only a KNOWN family contradiction, via `families_conflict` — the
|
||
same predicate the divergence gate uses, so the denominator and the gate
|
||
cannot drift into disagreeing about what "comparable" means.
|
||
|
||
`form` is a FORM (`fn`, `async-fn`, `type`, …), never a family:
|
||
`families_conflict` coarsens both sides itself, and handing it a family
|
||
makes `shape_family("callable")` return "" so nothing is excluded — a
|
||
narrowing that silently becomes a no-op while still reading as applied.
|
||
|
||
AN UNREADABLE SIBLING STAYS COUNTED, and that direction is the point.
|
||
Dropping it would shrink `judged`, raise the dominant canon's share, and
|
||
make the check fire MORE on the directories it can read least. Every
|
||
unknown-form decision in this module goes the same way: quieter, never
|
||
more confident.
|
||
"""
|
||
if not form:
|
||
return list(rows)
|
||
return [
|
||
r for r in rows
|
||
if not families_conflict(
|
||
shape_form(getattr(r, "signature", "") or "", getattr(r, "kind", "sym")),
|
||
form,
|
||
)
|
||
]
|
||
|
||
|
||
def dominant_canon(rows: Iterable[CodeShape]) -> tuple[int, int, int] | None:
|
||
"""(snippet_id, its_count, judged_count) when one canon dominates these
|
||
sibling rows (same directory + kind), else None."""
|
||
counts: dict[int, int] = {}
|
||
judged = 0
|
||
for r in rows:
|
||
# A weak stamp is not a judgment: counted, it could let a YAML CI
|
||
# snippet "dominate" a Svelte directory, and every write there would
|
||
# be told it diverged from it.
|
||
if is_weak_stamp(r):
|
||
continue
|
||
if r.status in ("canonical", "instance") and r.snippet_id is not None:
|
||
judged += 1
|
||
counts[r.snippet_id] = counts.get(r.snippet_id, 0) + 1
|
||
if judged < _DENSITY_MIN_JUDGED or not counts:
|
||
return None
|
||
sid, n = max(counts.items(), key=lambda kv: (kv[1], -kv[0]))
|
||
if n / judged < _DENSITY_SHARE:
|
||
return None
|
||
return sid, n, judged
|
||
|
||
|
||
def _dir_of(path: str) -> str:
|
||
return path.rsplit("/", 1)[0] if "/" in path else ""
|
||
|
||
|
||
async def canon_density(
|
||
project_id: int, path: str, kind: str, form: str = ""
|
||
) -> tuple[int, int, int, str] | None:
|
||
"""The dominant canon for the directory ``path`` sits in, for ``kind`` —
|
||
the write-time question "is this a canon-dense place?".
|
||
|
||
``form`` is the candidate's own FORM — `fn`, `async-fn`, `type`, … as
|
||
`shape_form` returns it, NOT a family. `families_conflict` coarsens both
|
||
sides itself, and handing it a family makes `shape_family("callable")`
|
||
return "" so nothing is ever excluded: the narrowing silently becomes a
|
||
no-op that still reads as applied. It narrows the DENOMINATOR. Without
|
||
it the base rate is computed over every code symbol in the directory as
|
||
one bucket, counting a dataclass, a constant and an async service unit as
|
||
three comparable things, and the share that comes out is a statement about
|
||
a population nobody asked a question about.
|
||
|
||
EXCLUDES ONLY A KNOWN CONFLICT, using `families_conflict` — the same
|
||
predicate the divergence gate uses, so the two cannot drift apart. A
|
||
sibling whose form is unreadable STAYS COUNTED. That direction is
|
||
deliberate and it is the one that matters: dropping unknown rows would
|
||
shrink `judged`, raise the share, and make the check fire MORE on exactly
|
||
the directories it can read least. Every other unknown-form decision in
|
||
this module goes the same way — quieter, never more confident.
|
||
|
||
NOT narrowed to the candidate's exact form, for the reason `shape_family`
|
||
gives at length: `fn` beside `async-fn` is a fair question, not noise.
|
||
Bucketing the denominator by form would take the async canon out of a
|
||
sync candidate's count and silence that flag — the inversion a form gate
|
||
makes, one layer down.
|
||
"""
|
||
directory = _dir_of(path)
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.kind == kind,
|
||
CodeShape.vanished_at.is_(None),
|
||
CodeShape.path.like(directory + "/%") if directory
|
||
else CodeShape.path.notlike("%/%"),
|
||
)
|
||
)
|
||
).scalars().all()
|
||
siblings = [r for r in rows if _dir_of(r.path) == directory]
|
||
siblings = comparable_siblings(siblings, form)
|
||
dom = dominant_canon(siblings)
|
||
if dom is None:
|
||
return None
|
||
# The canon's form travels with the count, read from its OWN judged rows.
|
||
# `dominant_canon` keeps its three-value shape: it answers "what dominates
|
||
# here", which is still a true and separately useful question, and its
|
||
# test pins that arithmetic. Callers act on the count together with the
|
||
# canon's form, never on the count alone.
|
||
return dom[0], dom[1], dom[2], canon_form(siblings, dom[0])
|
||
|
||
|
||
async def write_time_divergence(
|
||
project_id: int, path: str, shapes: list[tuple[str, str]], suggested: list[dict],
|
||
code: str = "",
|
||
) -> list[dict]:
|
||
"""The in-band check for the shapes the hook named at ``path``: for each
|
||
kind whose directory has a dominant canon, the named shapes that are
|
||
not that canon's instance/canonical (judged, or just suggested as it) — new or
|
||
unclassified rows only; a judged shape is not re-litigated at every
|
||
edit. Returns [{symbol, kind, canon_snippet_id, instances, judged}]."""
|
||
just_stamped = {(s["symbol"], s["kind"]): s["snippet_id"] for s in suggested}
|
||
out: list[dict] = []
|
||
if not shapes:
|
||
return out
|
||
# DENSITY IS PER CANDIDATE, not per kind: the denominator
|
||
# excludes siblings whose family contradicts what is being written, so it
|
||
# cannot be computed until the candidate's own form is known. Cached on
|
||
# (kind, family) — a write names a handful of shapes and they collapse to
|
||
# one or two buckets, so this stays one or two queries.
|
||
#
|
||
# The cost is that the row load below does not sit behind an early exit
|
||
# on "nothing is dense here". That is one indexed lookup on
|
||
# (project_id, path), and it has to happen first regardless: the
|
||
# candidate's signature comes from its stored row when the payload does
|
||
# not carry one.
|
||
_density: dict[tuple[str, str], tuple[int, int, int, str] | None] = {}
|
||
|
||
async def density_for(kind: str, form: str):
|
||
# Keyed and passed as a FORM, not a family — see `canon_density`.
|
||
key = (kind, form)
|
||
if key not in _density:
|
||
_density[key] = await canon_density(project_id, path, kind, form)
|
||
return _density[key]
|
||
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.path == path,
|
||
CodeShape.vanished_at.is_(None),
|
||
)
|
||
)
|
||
).scalars().all()
|
||
by_key = {(r.symbol, r.kind): r for r in rows}
|
||
for kind, name in shapes:
|
||
row = by_key.get((name, kind))
|
||
# The candidate's own form, read before anything is counted. Signature
|
||
# from the payload first — a shape being written now may have no row
|
||
# yet — falling back to the stored row's.
|
||
mine = shape_form(
|
||
signature_in(code, name, kind) or getattr(row, "signature", "") or "", kind
|
||
)
|
||
dom = await density_for(kind, mine)
|
||
if not dom:
|
||
continue
|
||
sid, n, judged, cform = dom
|
||
if just_stamped.get((name, kind)) == sid:
|
||
continue
|
||
if row is not None and (
|
||
row.status != "unclassified" or row.proposed_snippet_id == sid
|
||
):
|
||
continue
|
||
# IS THIS EVEN THE SAME CATEGORY OF THING AS THE CANON?
|
||
# Without this the line was a pure base rate: "most things here are X,
|
||
# so be X", with the candidate never examined at all. It told a frozen
|
||
# dataclass to build from the `async_session` service canon.
|
||
#
|
||
# FAMILY, not form, and only on contradiction — see `shape_family`. A
|
||
# divergence prompt is ABOUT a mismatch, so requiring the candidate to
|
||
# match would silence the check precisely where it belongs.
|
||
#
|
||
# `mine` was read above, before the denominator was counted — the
|
||
# same value serves both, and they must not be able to disagree.
|
||
# An unreadable signature produces a fair question rather than a
|
||
# guess, because `families_conflict` needs both sides.
|
||
if families_conflict(mine, cform):
|
||
continue
|
||
out.append({"symbol": name, "kind": kind, "canon_snippet_id": sid,
|
||
"instances": n, "judged": judged})
|
||
return out
|
||
|
||
|
||
# How many other files a family line names before "…" — enough to go look,
|
||
# not a wall.
|
||
_DERIVE_FILES_SHOWN = 4
|
||
|
||
|
||
async def write_time_derive(
|
||
project_id: int, path: str, shapes: list[tuple[str, str]]
|
||
) -> list[dict]:
|
||
"""The in-band DERIVE check: for each (kind, name) the hook
|
||
named at ``path``, what the ledger already knows about that name
|
||
elsewhere in the project —
|
||
|
||
family the name sits in a derive-first group (code: identical body
|
||
in N files or the same name in ≥3; CSS: the same class in
|
||
≥2 files): "this is a known family with no canon
|
||
— derive it now, don't add a copy";
|
||
canon a `canonical` row of that name at another path: "this is
|
||
canon #N at <path> — reuse, don't redefine".
|
||
|
||
Only for shapes not yet judged at ``path`` (a judged shape is not
|
||
re-litigated at every edit), never for the canon's own file. Returns
|
||
[{symbol, kind, key, family?|canon?}] — `key` is the dedup token the
|
||
hook keeps per session (the group id, or canon:<snippet_id>)."""
|
||
wanted = {(k, _norm_symbol(n)): n for k, n in shapes if n}
|
||
if not wanted:
|
||
return []
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.vanished_at.is_(None),
|
||
CodeShape.symbol.in_({norm for (_k, norm) in wanted}),
|
||
)
|
||
)
|
||
).scalars().all()
|
||
out: list[dict] = []
|
||
for (kind, norm), name in wanted.items():
|
||
same = [r for r in rows if r.kind == kind and _norm_symbol(r.symbol) == norm]
|
||
here = next((r for r in same if r.path == path), None)
|
||
if here is not None and here.status not in _MECHANICAL_TODO:
|
||
continue # judged here (or this IS the canon): nothing to say
|
||
others = [r for r in same if r.path != path]
|
||
label = ("." if kind == "css" else "") + name
|
||
canon = next((r for r in others if r.status == "canonical" and r.snippet_id), None)
|
||
if canon is not None:
|
||
out.append({"symbol": name, "kind": kind, "key": f"canon:{canon.snippet_id}",
|
||
"canon": {"snippet_id": canon.snippet_id, "path": canon.path,
|
||
"label": label}})
|
||
continue
|
||
grouped = [r for r in others if r.proposal_group and r.status in _MECHANICAL_TODO]
|
||
if here is not None and here.proposal_group:
|
||
grouped = [r for r in grouped if r.proposal_group == here.proposal_group] or grouped
|
||
if not grouped:
|
||
continue
|
||
group = grouped[0].proposal_group
|
||
members = [r for r in grouped if r.proposal_group == group]
|
||
files = sorted({r.path for r in members})
|
||
family = {
|
||
"group": group, "label": label,
|
||
"identical": not group.startswith("name:"),
|
||
"files": files[:_DERIVE_FILES_SHOWN], "file_count": len(files),
|
||
"size": len(members) + (1 if here is not None else 0),
|
||
}
|
||
if kind == "css":
|
||
# What renders the family: the members' consumer
|
||
# files, the row at `path` included when it already exists.
|
||
ids = [r.id for r in members] + ([here.id] if here is not None else [])
|
||
edges = await consumers_of(ids)
|
||
family["consumers"] = consumer_summary(
|
||
e.path for es in edges.values() for e in es
|
||
)
|
||
out.append({"symbol": name, "kind": kind, "key": group, "family": family})
|
||
return out
|
||
|
||
|
||
async def new_shapes_at(
|
||
project_id: int, path: str, shapes: list[tuple[str, str]]
|
||
) -> list[tuple[str, str]]:
|
||
"""The (kind, name) shapes the hook named at ``path`` that the ledger has
|
||
no live row for there — definitions this write introduces, as opposed to
|
||
ones it edits inside. Order follows ``shapes``."""
|
||
wanted = [(k, n) for k, n in shapes if n]
|
||
if not wanted:
|
||
return []
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape.kind, CodeShape.symbol).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.path == path,
|
||
CodeShape.vanished_at.is_(None),
|
||
)
|
||
)
|
||
).all()
|
||
here = {(k, _norm_symbol(s)) for k, s in rows}
|
||
return [(k, n) for k, n in wanted if (k, _norm_symbol(n)) not in here]
|
||
|
||
|
||
async def flag_divergence(project_id: int, *, since: datetime | None) -> int:
|
||
"""Flag shapes created after ``since`` (the previous refresh) that sit
|
||
where a canon dominates and were not proposed as that canon. With no
|
||
previous refresh (first seed) nothing is new, nothing is flagged.
|
||
Standing flags persist until judged. Returns how many are flagged."""
|
||
if since is None:
|
||
return 0
|
||
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_dir: dict[tuple[str, str], list[CodeShape]] = {}
|
||
for r in rows:
|
||
by_dir.setdefault((_dir_of(r.path), r.kind), []).append(r)
|
||
flagged = 0
|
||
for siblings in by_dir.values():
|
||
dom = dominant_canon(siblings)
|
||
# Computed once per directory rather than per row: the canon's
|
||
# form is a property of the canon, not of who is being judged.
|
||
cform = canon_form(siblings, dom[0]) if dom else FORM_UNKNOWN
|
||
for r in siblings:
|
||
if r.status not in _MECHANICAL_TODO:
|
||
continue
|
||
if r.diverges_from is not None:
|
||
# A flag is a mechanical prompt resting on one premise —
|
||
# "this canon dominates here". When the premise no longer
|
||
# holds (the rows that elected it were weak stamps, or
|
||
# were re-judged away) the prompt is false, and a false
|
||
# prompt on every write is a flag nobody reads.
|
||
# Withdrawing it judges nothing: the row stays
|
||
# unclassified, and a judgment is never touched here.
|
||
if dom is None or dom[0] != r.diverges_from:
|
||
r.diverges_from = None
|
||
continue
|
||
flagged += 1
|
||
continue
|
||
if dom is None or r.created_at is None or r.created_at <= since:
|
||
continue
|
||
if r.proposed_snippet_id == dom[0]:
|
||
continue # the proposer already says "instance of the canon"
|
||
# NOT the converse. A semantic MISS does not mean "not the
|
||
# canon": true instances of a canon often miss the arm's floor
|
||
# too (see the note above _SEMANTIC_LIMIT). A helper beside a
|
||
# canon and a genuine divergence — a sync `confirmDanger`
|
||
# beside an async confirm canon — are the same case to every
|
||
# signal this ledger holds, so the prompt asks, and the judge
|
||
# answers it (`exempt` with a reason retires a false one for
|
||
# good).
|
||
# The same structural test the write-time check applies. The
|
||
# sweep and the hook must agree about what counts
|
||
# as divergence, or an audit contradicts the line the writer
|
||
# was shown at the keyboard.
|
||
if families_conflict(shape_form(r.signature or "", r.kind), cform):
|
||
continue
|
||
r.diverges_from = dom[0]
|
||
flagged += 1
|
||
await session.commit()
|
||
return flagged
|
||
|
||
|
||
# --- the end-of-turn question: what did this turn write that
|
||
# nobody has judged? ---------------------------------------------------------
|
||
#
|
||
# The write hooks keep a session ledger of the definitions each turn wrote; at
|
||
# the end of the turn the client asks this, and the agent that wrote them
|
||
# answers. "Judged" means a verdict SOMEONE READ: an agent's, an audit's, an
|
||
# import's. The sync's `scoped` stamp, `unclassified`, a hook stamp and no row
|
||
# at all are all unjudged — each is a machine's guess or nothing, and the
|
||
# point of the question is that the writer, who knows, says.
|
||
|
||
def _is_judged(row) -> bool:
|
||
return (
|
||
row.status not in _MECHANICAL_TODO
|
||
and row.classified_by not in _UNATTENDED_BY
|
||
)
|
||
|
||
|
||
async def unjudged_shapes(
|
||
project_id: int, written: list[tuple[str, str, str]], *, repo_key: str = "",
|
||
) -> list[dict]:
|
||
"""The (path, kind, name) shapes in ``written`` that carry no judgment.
|
||
|
||
Each comes back as {path, kind, symbol, status, evidence} — status is the
|
||
row's, or "new" when the ledger has none yet; evidence is what the
|
||
machinery holds that the writer may want to weigh: the proposer's
|
||
candidate canon, a hook's stamp, a divergence flag. Evidence, never a
|
||
verdict — the writer is asked because the machine's guesses were the
|
||
thing that poisoned two ledgers. Order follows ``written``; duplicates
|
||
collapse. The caller has already resolved the project and checked access.
|
||
"""
|
||
wanted: list[tuple[str, str, str]] = []
|
||
for path, kind, name in written:
|
||
key = ((path or "").strip(), (kind or "").strip(), (name or "").strip())
|
||
if all(key) and key not in wanted:
|
||
wanted.append(key)
|
||
if not project_id or not wanted:
|
||
return []
|
||
paths = sorted({p for p, _k, _n in wanted})
|
||
query = select(CodeShape).where(
|
||
CodeShape.project_id == project_id,
|
||
CodeShape.vanished_at.is_(None),
|
||
CodeShape.path.in_(paths),
|
||
)
|
||
if repo_key:
|
||
query = query.where(CodeShape.repo_key == repo_key)
|
||
async with async_session() as session:
|
||
rows = (await session.execute(query)).scalars().all()
|
||
by_key = {(r.path, r.kind, r.symbol): r for r in rows}
|
||
out: list[dict] = []
|
||
for path, kind, name in wanted:
|
||
row = by_key.get((path, kind, name))
|
||
if row is not None and _is_judged(row):
|
||
continue
|
||
evidence: dict = {}
|
||
if row is not None:
|
||
if row.proposed_snippet_id:
|
||
evidence["looks_like"] = {
|
||
"snippet_id": row.proposed_snippet_id,
|
||
"basis": row.proposal_basis,
|
||
"score": row.proposal_score,
|
||
}
|
||
if row.classified_by in _UNATTENDED_BY and row.snippet_id:
|
||
evidence["stamped"] = {
|
||
"snippet_id": row.snippet_id, "score": stamp_score(row.reason),
|
||
}
|
||
if row.proposal_group:
|
||
evidence["copies"] = row.proposal_group
|
||
if row.diverges_from:
|
||
evidence["diverges_from"] = row.diverges_from
|
||
out.append({
|
||
"path": path, "kind": kind, "symbol": name,
|
||
"status": row.status if row is not None else "new",
|
||
"evidence": evidence,
|
||
})
|
||
return out
|
||
|
||
|
||
# How many of a canon's rows a review listing shows before "…" — enough to
|
||
# judge from, not the whole table.
|
||
_REVIEW_ROWS_SHOWN = 12
|
||
|
||
|
||
def stamps_to_review(rows: Iterable[CodeShape], *, top: int = 10) -> dict:
|
||
"""Judged rows whose EVIDENCE no longer meets the bar the ledger now
|
||
holds, and canons whose own rows no longer agree what they are.
|
||
|
||
THIS CHANGES NOTHING, AND THAT IS THE POINT. Every `weak` row was written
|
||
unattended by the write-path hook on a similarity score, and the harm in
|
||
that is not a score being wrong once — it is a machine asserting a
|
||
permanent classification with nobody reading it. A sweep that silently
|
||
un-asserted them would be the identical mistake with a larger blast
|
||
radius and a cleaner conscience. So this reports; an agent reads the
|
||
evidence, decides, and records the decision through `classify_shapes`
|
||
under its own name. The ledger should be able to say "these look wrong"
|
||
without being able to act on it alone.
|
||
|
||
Two lists, because they are different questions:
|
||
|
||
`weak` — rows stamped on a resemblance below `_RESEMBLE_MIN`. A guard at
|
||
the point of classification does not undo classifications already stored,
|
||
so rows written before the floor need reading. Rows a
|
||
person or an audit judged are never listed, however old: a human judgment
|
||
is not weak evidence, it is a different kind of evidence.
|
||
|
||
`incoherent` — canons whose membership no longer agrees what the canon
|
||
IS. A canon is a claim that some set of shapes are the same sort of
|
||
thing; when its own members are a class, three getters and a dozen tests,
|
||
that claim has stopped being true, and every base-rate reading built on
|
||
it is reading noise. `canon_form` already makes such a canon fall silent
|
||
— this is what makes it VISIBLE.
|
||
|
||
What counts as disagreement is `canon_coherence`, and it is deliberately
|
||
looser than "one form": it compares FAMILIES, and it does not hold a
|
||
method against the class it is defined beside. A review surface whose
|
||
output is noise stops being read, which costs more than the check is
|
||
worth.
|
||
|
||
`strangers` names the rows that do not fit, not the first dozen members:
|
||
the reader's question is which ones are wrong. `unattended` counts rows
|
||
written by the hook with nobody reading — an incoherence made entirely of
|
||
judged rows is far more likely to be this check being too strict than a
|
||
ledger full of junk, and the reader should be able to see that without
|
||
opening the canon.
|
||
"""
|
||
live = [r for r in rows if r.vanished_at is None]
|
||
weak = []
|
||
for r in live:
|
||
if r.status not in _NEEDS_TARGET or not is_weak_stamp(r):
|
||
continue
|
||
score = stamp_score(r.reason)
|
||
weak.append({
|
||
"path": r.path, "symbol": r.symbol, "kind": r.kind,
|
||
"status": r.status, "snippet_id": r.snippet_id,
|
||
"score": score, "signature": r.signature or "",
|
||
"form": shape_form(r.signature or "", r.kind),
|
||
"classified_at": r.classified_at.isoformat() if r.classified_at else None,
|
||
})
|
||
weak.sort(key=lambda d: (d["score"], d["path"], d["symbol"]))
|
||
|
||
by_canon: dict[int, list[CodeShape]] = {}
|
||
for r in live:
|
||
if r.status in _NEEDS_TARGET and r.snippet_id:
|
||
by_canon.setdefault(int(r.snippet_id), []).append(r)
|
||
incoherent = []
|
||
for sid, members in by_canon.items():
|
||
coh = canon_coherence(members)
|
||
if coh["readable"] < _DENSITY_MIN_JUDGED:
|
||
continue # too few to say anything either way
|
||
if coh["coheres"]:
|
||
continue
|
||
unattended = sum(1 for r in members if r.classified_by in _UNATTENDED_BY)
|
||
incoherent.append({
|
||
"snippet_id": sid,
|
||
"judged": len(members),
|
||
"forms": coh["forms"],
|
||
"families": coh["families"],
|
||
"family": coh["family"],
|
||
"attached": coh["attached"],
|
||
"unattended": unattended,
|
||
# The listing below is capped; without this you cannot tell a
|
||
# canon with twelve strangers from one with three hundred.
|
||
"stranger_count": len(coh["strangers"]),
|
||
"weak_rows": sum(1 for r in members if is_weak_stamp(r)),
|
||
# The rows that do NOT fit — not the first dozen members. The
|
||
# reader's question is "which ones are wrong", and a sample of
|
||
# the majority cannot answer it.
|
||
"strangers": [
|
||
{"path": r.path, "symbol": r.symbol,
|
||
"signature": r.signature or "", "by": r.classified_by,
|
||
"form": shape_form(r.signature or "", r.kind)}
|
||
for r in coh["strangers"][:_REVIEW_ROWS_SHOWN]
|
||
],
|
||
})
|
||
incoherent.sort(key=lambda d: (-d["weak_rows"], -d["unattended"], -d["judged"]))
|
||
return {
|
||
"weak_count": len(weak),
|
||
"weak": weak[:top],
|
||
"incoherent_count": len(incoherent),
|
||
"incoherent": incoherent[:top],
|
||
"floor": _RESEMBLE_MIN,
|
||
}
|
||
|
||
|
||
def divergence_summary(rows: Iterable[CodeShape], *, top: int = 10) -> dict:
|
||
"""Readout view: flagged shapes (newest first) and the recheck count."""
|
||
flagged = [r for r in rows if r.diverges_from is not None and r.status in _MECHANICAL_TODO]
|
||
flagged.sort(key=lambda r: (r.created_at or datetime.min.replace(tzinfo=timezone.utc)), reverse=True)
|
||
recheck = sum(1 for r in rows if r.recheck_at is not None and r.vanished_at is None)
|
||
return {
|
||
"divergent": len(flagged),
|
||
"divergence": [
|
||
{"path": r.path, "symbol": r.symbol, "kind": r.kind,
|
||
"canon_snippet_id": r.diverges_from}
|
||
for r in flagged[:top]
|
||
],
|
||
"recheck": recheck,
|
||
}
|
||
|
||
|
||
async def shape_history(
|
||
user_id: int, project_id: int, path: str, *, symbol: str = "", limit: int = 200
|
||
) -> dict:
|
||
"""What was used at ``path`` (a file or directory), when, and why: the
|
||
current rows plus their events, oldest first. Read-gated like every
|
||
other ledger read; {} when the caller cannot read the project."""
|
||
from sqlalchemy import or_
|
||
|
||
from grimoire.services import access
|
||
|
||
if not await access.can_read_project(user_id, project_id):
|
||
return {}
|
||
clean = (path or "").strip().strip("/")
|
||
conds = [CodeShape.project_id == project_id]
|
||
if clean:
|
||
conds.append(or_(CodeShape.path == clean, CodeShape.path.like(clean + "/%")))
|
||
if symbol.strip():
|
||
conds.append(CodeShape.symbol == symbol.strip())
|
||
async with async_session() as session:
|
||
rows = (
|
||
await session.execute(
|
||
select(CodeShape).where(*conds)
|
||
.order_by(CodeShape.path, CodeShape.symbol, CodeShape.kind)
|
||
.limit(500)
|
||
)
|
||
).scalars().all()
|
||
ids = [r.id for r in rows]
|
||
events = (
|
||
await session.execute(
|
||
select(CodeShapeEvent).where(CodeShapeEvent.shape_id.in_(ids))
|
||
.order_by(CodeShapeEvent.at.asc(), CodeShapeEvent.id.asc())
|
||
.limit(max(1, min(limit, 1000)))
|
||
)
|
||
).scalars().all() if ids else []
|
||
return {
|
||
"shapes": [
|
||
{
|
||
"path": r.path, "symbol": r.symbol, "kind": r.kind,
|
||
"status": r.status, "snippet_id": r.snippet_id,
|
||
"classified_by": r.classified_by, "reason": r.reason,
|
||
"first_seen_commit": r.first_seen_commit,
|
||
"last_seen_commit": r.last_seen_commit,
|
||
"first_seen_at": iso(r.created_at),
|
||
"vanished_at": iso(r.vanished_at),
|
||
"recheck_at": iso(r.recheck_at),
|
||
"diverges_from": r.diverges_from,
|
||
}
|
||
for r in rows
|
||
],
|
||
"events": [e.to_dict() for e in events],
|
||
}
|