bvandeusenandClaude Opus 5.5
47b52197aa
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Failing after 15s
CI & Build / TypeScript typecheck (push) Successful in 55s
CI & Build / integration (push) Successful in 1m15s
CI & Build / Python tests (push) Successful in 2m9s
CI & Build / Build & push image (push) Successful in 30s
Ledger rows are keyed by path, so after a directory moves the next sync marks every judged row vanished and lists the same shapes again, unjudged. move_shapes(project_id, from_path, to_path) re-keys the rows under one path to the other, in every repo of the project, keeping each row's judgment, uses edges and history and adding a `moved` event that records the path it left. A row already at the target is replaced when nobody judged it; a judged one stands, and the old row is reported in `kept`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3029 lines
132 KiB
Python
3029 lines
132 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()
|
||
|
||
|
||
def _under(path: str, prefix: str) -> bool:
|
||
return path == prefix or path.startswith(prefix + "/")
|
||
|
||
|
||
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
|
||
|
||
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 _under(old, new) or _under(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 _under(r.path, old)
|
||
]
|
||
if not sources:
|
||
raise ValueError(f"project {project_id} has no shapes under `{old}`")
|
||
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 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],
|
||
}
|