feat(ledger): reason codes, case-insensitive repo filter, next action on the coverage line (#2874, milestone 294)
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / integration (push) Failing after 25s
CI & Build / TypeScript typecheck (push) Canceled after 32s
CI & Build / Python tests (push) Canceled after 31s
CI & Build / Build & push image (push) Canceled after 0s

- code_shapes.reason_code (migration 0083): an optional code from a fixed
  catalogue (scoped-css, one-off-handler, test-helper, convention-plumbing,
  pure-helper, generated, script, typed-record) beside the prose reason, so
  the ledger can be filtered/aggregated by kind of one-off; validated in
  classify_shapes and classify_shapes_by_rule; on to_dict/to_compact.
- Snippet location lookups match repo case-insensitively in both dialects
  (location_matches / location_jsonpath via like_regex flag "i") — "Scribe"
  vs "FabledScribe" vs "fabledscribe" recorded free-form hid half the canon
  from list_snippets(repo=, path=).
- coverage line names the next action: "top canon #N ×k" (biggest proposal
  queue) and "top copy <label> ×files" (widest body-identical group).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-21 15:14:34 -04:00
co-authored by Claude Fable 5
parent d01201539b
commit 1a8e5787e8
9 changed files with 123 additions and 11 deletions
@@ -0,0 +1,26 @@
"""Exempt/variant reason codes — a small fixed catalogue beside the prose (#2874, milestone 294)
Revision ID: 0083
Revises: 0082
Create Date: 2026-08-21
The 2026-08 audit wrote the same free-text reason thousands of times
("scoped rule — styles one element of this view"); a judgment's WHY stays
prose, but an optional code from a fixed catalogue makes the ledger
filterable and aggregable ("how many pure helpers, how many test helpers").
"""
import sqlalchemy as sa
from alembic import op
revision = "0083"
down_revision = "0082"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column("code_shapes", sa.Column("reason_code", sa.Text(), nullable=True))
def downgrade() -> None:
op.drop_column("code_shapes", "reason_code")
+12 -4
View File
@@ -36,10 +36,14 @@ async def classify_shapes(
Args: Args:
project_id: The project whose ledger is being judged. project_id: The project whose ledger is being judged.
classifications: Objects of {path, symbol, status, kind?, snippet_id?, classifications: Objects of {path, symbol, status, kind?, snippet_id?,
reason?}. path+symbol name the shape exactly as list_shapes shows reason?, reason_code?}. path+symbol name the shape exactly as
it; kind ("sym"/"css") narrows when one file defines both. list_shapes shows it; kind ("sym"/"css") narrows when one file
snippet_id is required for canonical/instance/variant; reason is defines both. snippet_id is required for canonical/instance/
required for variant/exempt. variant; reason is required for variant/exempt. reason_code is
an OPTIONAL index beside the prose (one of: scoped-css,
one-off-handler, test-helper, convention-plumbing, pure-helper,
generated, script, typed-record) so the ledger can be filtered
and aggregated by kind of one-off — the prose stays the record.
via: Who is judging — "agent" (default), "audit" (a sweep), or via: Who is judging — "agent" (default), "audit" (a sweep), or
"import" (carrying maps recorded elsewhere). "import" (carrying maps recorded elsewhere).
@@ -145,6 +149,7 @@ async def classify_shapes_by_rule(
reason: str = "", reason: str = "",
via: str = "agent", via: str = "agent",
include_judged: bool = False, include_judged: bool = False,
reason_code: str = "",
) -> dict: ) -> dict:
"""The sweep form of classify_shapes: ONE judgment applied to every """The sweep form of classify_shapes: ONE judgment applied to every
unclassified shape under a directory whose symbol matches a glob. unclassified shape under a directory whose symbol matches a glob.
@@ -167,6 +172,8 @@ async def classify_shapes_by_rule(
snippet_id: Required for instance/variant — the canon judged against. snippet_id: Required for instance/variant — the canon judged against.
reason: Required for variant/exempt — the why, recorded on every row. reason: Required for variant/exempt — the why, recorded on every row.
via: "agent" (default) | "audit" | "import". via: "agent" (default) | "audit" | "import".
reason_code: Optional catalogue code beside the reason (see
classify_shapes) — a sweep is exactly where one applies.
include_judged: By default only unjudged rows are touched — include_judged: By default only unjudged rows are touched —
`unclassified` and the sync's mechanical `scoped` stamp — a `unclassified` and the sync's mechanical `scoped` stamp — a
sweep never silently overwrites a judgment. True re-judges every sweep never silently overwrites a judgment. True re-judges every
@@ -184,6 +191,7 @@ async def classify_shapes_by_rule(
uid, project_id, path=path, status=status, pattern=pattern, uid, project_id, path=path, status=status, pattern=pattern,
kind=kind, snippet_id=snippet_id or None, reason=reason or None, kind=kind, snippet_id=snippet_id or None, reason=reason or None,
via=via, include_judged=include_judged, via=via, include_judged=include_judged,
reason_code=reason_code or None,
) )
except ValueError as exc: except ValueError as exc:
return {"error": str(exc)} return {"error": str(exc)}
+18
View File
@@ -26,6 +26,20 @@ from scribe.models.base import TimestampMixin, iso
SHAPE_STATUSES = ("canonical", "instance", "variant", "exempt", "scoped", "unclassified") SHAPE_STATUSES = ("canonical", "instance", "variant", "exempt", "scoped", "unclassified")
SHAPE_CLASSIFIERS = ("agent", "audit", "hook", "mechanical", "import") SHAPE_CLASSIFIERS = ("agent", "audit", "hook", "mechanical", "import")
# The reason catalogue (#2874): an OPTIONAL code beside the prose reason on
# variant/exempt rows, so the ledger can be filtered and aggregated by kind
# of one-off. The prose remains the record; the code is the index.
REASON_CODES = (
"scoped-css", # a scoped rule styling one element (pre-#2869 rows)
"one-off-handler", # a view/component handler or loader, one per surface
"test-helper", # a test module's stub, driver or fixture data
"convention-plumbing", # registration, wiring, app factory — one of each
"pure-helper", # a sync module-private helper with no session
"generated", # generated source (theme.css, protos, bundles)
"script", # a standalone dev/CI script
"typed-record", # a NamedTuple / dataclass / error class — one each
)
# How the mechanical proposer (#2792) arrived at a proposal, strongest first. # How the mechanical proposer (#2792) arrived at a proposal, strongest first.
# `derive` is the odd one out: not "this is an instance of #N" but "this # `derive` is the odd one out: not "this is an instance of #N" but "this
# shape repeats with NO canon — derive one first" (note 2786's derive-first # shape repeats with NO canon — derive one first" (note 2786's derive-first
@@ -105,6 +119,7 @@ class CodeShape(Base, TimestampMixin):
BigInteger, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True BigInteger, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
) )
reason: Mapped[str | None] = mapped_column(Text, nullable=True) reason: Mapped[str | None] = mapped_column(Text, nullable=True)
reason_code: Mapped[str | None] = mapped_column(Text, nullable=True) # REASON_CODES (#2874)
classified_by: Mapped[str | None] = mapped_column(Text, nullable=True) classified_by: Mapped[str | None] = mapped_column(Text, nullable=True)
classified_at: Mapped[datetime | None] = mapped_column( classified_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True DateTime(timezone=True), nullable=True
@@ -158,6 +173,7 @@ class CodeShape(Base, TimestampMixin):
"status": self.status, "status": self.status,
"snippet_id": self.snippet_id, "snippet_id": self.snippet_id,
"reason": self.reason, "reason": self.reason,
"reason_code": self.reason_code,
"classified_by": self.classified_by, "classified_by": self.classified_by,
"classified_at": iso(self.classified_at), "classified_at": iso(self.classified_at),
"first_seen_commit": self.first_seen_commit, "first_seen_commit": self.first_seen_commit,
@@ -189,6 +205,8 @@ class CodeShape(Base, TimestampMixin):
out["snippet_id"] = self.snippet_id out["snippet_id"] = self.snippet_id
if self.classified_by: if self.classified_by:
out["by"] = self.classified_by out["by"] = self.classified_by
if self.reason_code:
out["reason_code"] = self.reason_code
proposal = self.proposal proposal = self.proposal
if proposal: if proposal:
out["proposal"] = proposal out["proposal"] = proposal
+9
View File
@@ -488,6 +488,7 @@ async def compute_coverage(
# confirm, the largest derive-first groups, and what this refresh did. # confirm, the largest derive-first groups, and what this refresh did.
"proposed": proposals["proposed"], "proposed": proposals["proposed"],
"derive_groups": proposals["derive_groups"], "derive_groups": proposals["derive_groups"],
"top_canon": proposals.get("top_canon"),
"proposer": proposer_stats, "proposer": proposer_stats,
# The divergence readout (#2793): button B where button A is canon, # The divergence readout (#2793): button B where button A is canon,
# and judged shapes whose bodies moved since they were judged. # and judged shapes whose bodies moved since they were judged.
@@ -645,6 +646,14 @@ def coverage_line(coverage: dict) -> str:
standing.append(f"{n_groups} derive group{'s' if n_groups != 1 else ''}") standing.append(f"{n_groups} derive group{'s' if n_groups != 1 else ''}")
if coverage.get("divergent"): if coverage.get("divergent"):
standing.append(f"{coverage['divergent']} DIVERGENT") standing.append(f"{coverage['divergent']} DIVERGENT")
# The next action, on the line (#2874): the canon with the biggest
# queue to confirm, and the widest body-identical copy to consolidate.
top = coverage.get("top_canon") or {}
if top.get("snippet_id"):
standing.append(f"top canon #{top['snippet_id']} ×{top.get('count', 0)}")
first = (coverage.get("derive_groups") or [{}])[0]
if first.get("label") and first.get("files"):
standing.append(f"top copy {first['label']} ×{first['files']} files")
if standing: if standing:
line += f" ({', '.join(standing)})" line += f" ({', '.join(standing)})"
gaps = [g["dir"] for g in coverage.get("largest_gaps") or []] gaps = [g["dir"] for g in coverage.get("largest_gaps") or []]
+10
View File
@@ -16,6 +16,7 @@ endorsed, so a one-off direct share has to be searched for rather than arriving
in your ambient lists. in your ambient lists.
""" """
import json import json
import re
import logging import logging
from sqlalchemy import and_, func, or_, select from sqlalchemy import and_, func, or_, select
@@ -76,6 +77,10 @@ def location_matches(data: dict | None, parts: dict[str, str]) -> bool:
if all( if all(
_path_matches((loc.get(key) or "").strip(), want) _path_matches((loc.get(key) or "").strip(), want)
if key == "path" if key == "path"
# Repo names are recorded free-form ("Scribe" / "FabledScribe" /
# "fabledscribe") — case is never the distinguishing thing (#2874).
else (loc.get(key) or "").strip().lower() == want.lower()
if key == "repo"
else (loc.get(key) or "").strip() == want else (loc.get(key) or "").strip() == want
for key, want in parts.items() for key, want in parts.items()
): ):
@@ -99,6 +104,11 @@ def location_jsonpath(parts: dict[str, str]) -> str:
if key == "path": if key == "path":
prefix = json.dumps(want.rstrip("/") + "/") prefix = json.dumps(want.rstrip("/") + "/")
filters.append(f"(@.path == {literal} || @.path starts with {prefix})") filters.append(f"(@.path == {literal} || @.path starts with {prefix})")
elif key == "repo":
# Case-insensitive, anchored, regex-escaped (#2874) — mirrors the
# Python dialect's .lower() compare.
pattern = json.dumps("^" + re.escape(want) + "$")
filters.append(f'(@.repo like_regex {pattern} flag "i")')
else: else:
filters.append(f"@.{key} == {literal}") filters.append(f"@.{key} == {literal}")
return f"$.locations[*] ? ({' && '.join(filters)})" return f"$.locations[*] ? ({' && '.join(filters)})"
+21 -5
View File
@@ -30,7 +30,7 @@ from typing import Iterable, NamedTuple
from sqlalchemy import select from sqlalchemy import select
from scribe.models import async_session from scribe.models import async_session
from scribe.models.code_shape import CodeShape, CodeShapeEvent from scribe.models.code_shape import REASON_CODES, CodeShape, CodeShapeEvent
from scribe.models.base import iso from scribe.models.base import iso
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -183,7 +183,7 @@ def _event(row: CodeShape, event: str, at: datetime, *, commit: str = "") -> Cod
async def _judge( async def _judge(
session, row: CodeShape, *, status: str, snippet_id: int | None, session, row: CodeShape, *, status: str, snippet_id: int | None,
by: str | None, reason: str | None, at: datetime, by: str | None, reason: str | None, at: datetime, reason_code: str | None = None,
) -> None: ) -> None:
"""Apply a judgment to a row — the ONE place a status is set — and write """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 its history. Clears what a judgment settles: the standing proposal, the
@@ -194,6 +194,7 @@ async def _judge(
row.status = status row.status = status
row.snippet_id = snippet_id if status in _NEEDS_TARGET else None row.snippet_id = snippet_id if status in _NEEDS_TARGET else None
row.reason = (reason or "").strip() or 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_by = by if status != "unclassified" else None
row.classified_at = at 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.classified_sha = row.body_sha if status != "unclassified" else ""
@@ -314,6 +315,12 @@ def validate_classifications(items: list[dict]) -> str | None:
f"classifications[{i}]: status {status!r} needs a reason — " f"classifications[{i}]: status {status!r} needs a reason — "
"the WHY is the record (note 2786)" "the WHY is the record (note 2786)"
) )
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)})"
)
return None return None
@@ -389,6 +396,7 @@ async def classify_shapes(
session, row, status=status, session, row, status=status,
snippet_id=int(item["snippet_id"]) if status in _NEEDS_TARGET else None, snippet_id=int(item["snippet_id"]) if status in _NEEDS_TARGET else None,
by=via, reason=item.get("reason"), at=now, by=via, reason=item.get("reason"), at=now,
reason_code=item.get("reason_code"),
) )
classified += 1 classified += 1
await session.commit() await session.commit()
@@ -424,6 +432,7 @@ async def classify_shapes_where(
reason: str | None = None, reason: str | None = None,
via: str = "agent", via: str = "agent",
include_judged: bool = False, include_judged: bool = False,
reason_code: str | None = None,
) -> dict: ) -> dict:
"""The sweep form of classify_shapes (#2868): one judgment applied to """The sweep form of classify_shapes (#2868): one judgment applied to
every live row under ``path`` whose symbol matches ``pattern`` (and every live row under ``path`` whose symbol matches ``pattern`` (and
@@ -443,7 +452,8 @@ async def classify_shapes_where(
if status == "canonical": if status == "canonical":
raise ValueError("canonical is the sync's stamp on a snippet's own location — a sweep cannot set it") 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, probe = {"path": path, "symbol": "*", "status": status,
"snippet_id": snippet_id or 0, "reason": reason or ""} "snippet_id": snippet_id or 0, "reason": reason or "",
"reason_code": reason_code or ""}
error = validate_classifications([probe]) error = validate_classifications([probe])
if error: if error:
raise ValueError(error.replace("classifications[0]", "rule")) raise ValueError(error.replace("classifications[0]", "rule"))
@@ -465,7 +475,7 @@ async def classify_shapes_where(
await _judge( await _judge(
session, row, status=status, session, row, status=status,
snippet_id=int(snippet_id) if status in _NEEDS_TARGET else None, snippet_id=int(snippet_id) if status in _NEEDS_TARGET else None,
by=via, reason=reason, at=now, by=via, reason=reason, at=now, reason_code=reason_code,
) )
judged.append(f"{row.path}::{row.symbol}") judged.append(f"{row.path}::{row.symbol}")
await session.commit() await session.commit()
@@ -1214,6 +1224,7 @@ def proposal_summary(rows: Iterable[CodeShape], *, top: int = 8) -> dict:
"""The readout's view of the proposer's standing: how many canon """The readout's view of the proposer's standing: how many canon
proposals await confirmation, and the largest derive-first groups.""" proposals await confirmation, and the largest derive-first groups."""
proposed = 0 proposed = 0
by_canon: dict[int, int] = {}
groups: dict[str, dict] = {} groups: dict[str, dict] = {}
files: dict[str, set[str]] = {} files: dict[str, set[str]] = {}
for row in rows: for row in rows:
@@ -1221,6 +1232,7 @@ def proposal_summary(rows: Iterable[CodeShape], *, top: int = 8) -> dict:
continue continue
if row.proposed_snippet_id is not None: if row.proposed_snippet_id is not None:
proposed += 1 proposed += 1
by_canon[row.proposed_snippet_id] = by_canon.get(row.proposed_snippet_id, 0) + 1
elif row.proposal_group: elif row.proposal_group:
dup = not row.proposal_group.startswith("name:") dup = not row.proposal_group.startswith("name:")
g = groups.setdefault(row.proposal_group, { g = groups.setdefault(row.proposal_group, {
@@ -1245,7 +1257,11 @@ def proposal_summary(rows: Iterable[CodeShape], *, top: int = 8) -> dict:
groups.values(), groups.values(),
key=lambda g: (g["group"].startswith("name:"), -g["files"], -g["size"], g["group"]), key=lambda g: (g["group"].startswith("name:"), -g["files"], -g["size"], g["group"]),
) )
return {"proposed": proposed, "derive_groups": ranked[:top]} 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}
async def confirm_proposals( async def confirm_proposals(
+6
View File
@@ -497,6 +497,12 @@ def test_coverage_line_names_the_proposers_standing():
assert "; 90 unclassified (40 proposed, 2 derive groups), largest: src" in line assert "; 90 unclassified (40 proposed, 2 derive groups), largest: src" in line
line = coverage_line({**base, "proposed": 0, "derive_groups": [{"group": "a"}]}) line = coverage_line({**base, "proposed": 0, "derive_groups": [{"group": "a"}]})
assert "(1 derive group)" in line assert "(1 derive group)" in line
# #2874: the next action on the line — biggest canon queue, widest copy.
line = coverage_line({
**base, "proposed": 40, "top_canon": {"snippet_id": 2844, "count": 78},
"derive_groups": [{"group": "dup:abc", "label": "closed-msg (identical body)", "files": 3}],
})
assert "top canon #2844 ×78" in line and "top copy closed-msg (identical body) ×3 files" in line
def test_coverage_line_names_divergence_and_recheck(): def test_coverage_line_names_divergence_and_recheck():
+16
View File
@@ -391,6 +391,22 @@ def test_compact_row_carries_identity_standing_and_the_proposers_word_only():
assert noisy not in compact assert noisy not in compact
def test_reason_codes_are_a_fixed_catalogue_and_validated():
"""#2874: an optional index beside the prose reason; unknown codes are a
structural error (the batch applies nothing)."""
from scribe.models.code_shape import REASON_CODES
from scribe.services.shape_ledger import validate_classifications
assert set(REASON_CODES) == {
"scoped-css", "one-off-handler", "test-helper", "convention-plumbing",
"pure-helper", "generated", "script", "typed-record",
}
assert "reason_code" in CodeShape.__table__.c
ok = [{"path": "a.py", "symbol": "f", "status": "exempt", "reason": "x", "reason_code": "pure-helper"}]
assert validate_classifications(ok) is None
bad = [{"path": "a.py", "symbol": "f", "status": "exempt", "reason": "x", "reason_code": "nope"}]
assert "unknown reason_code" in validate_classifications(bad)
def test_rule_matches_is_directory_glob_and_kind_aware(): def test_rule_matches_is_directory_glob_and_kind_aware():
from scribe.models.code_shape import CodeShape from scribe.models.code_shape import CodeShape
from scribe.services.shape_ledger import rule_matches from scribe.services.shape_ledger import rule_matches
+5 -2
View File
@@ -56,6 +56,7 @@ def test_no_parts_matches_everything():
def test_matches_exact_repo_path_and_symbol(): def test_matches_exact_repo_path_and_symbol():
data = _data({"repo": "Scribe", "path": "src/scribe/x.py", "symbol": "helper"}) data = _data({"repo": "Scribe", "path": "src/scribe/x.py", "symbol": "helper"})
assert location_matches(data, {"repo": "Scribe"}) assert location_matches(data, {"repo": "Scribe"})
assert location_matches(data, {"repo": "scribe"}) # repo names: case never distinguishes (#2874)
assert location_matches(data, {"path": "src/scribe/x.py"}) assert location_matches(data, {"path": "src/scribe/x.py"})
assert location_matches(data, {"symbol": "helper"}) assert location_matches(data, {"symbol": "helper"})
assert location_matches(data, {"repo": "Scribe", "symbol": "helper"}) assert location_matches(data, {"repo": "Scribe", "symbol": "helper"})
@@ -101,7 +102,8 @@ def test_blank_recorded_part_does_not_match_a_requested_one():
def test_jsonpath_filters_within_one_locations_entry(): def test_jsonpath_filters_within_one_locations_entry():
expr = location_jsonpath({"repo": "Scribe", "symbol": "helper"}) expr = location_jsonpath({"repo": "Scribe", "symbol": "helper"})
assert expr.startswith("$.locations[*] ? (") assert expr.startswith("$.locations[*] ? (")
assert '@.repo == "Scribe"' in expr # repo: anchored, case-insensitive (#2874) — mirrors location_matches.
assert '@.repo like_regex "^Scribe$" flag "i"' in expr
assert '@.symbol == "helper"' in expr assert '@.symbol == "helper"' in expr
assert " && " in expr assert " && " in expr
@@ -121,8 +123,9 @@ def test_jsonpath_quotes_values_as_json_literals():
"""A quote in a repo name must stay inside the literal, not end it.""" """A quote in a repo name must stay inside the literal, not end it."""
nasty = 'we"ird' nasty = 'we"ird'
expr = location_jsonpath({"repo": nasty}) expr = location_jsonpath({"repo": nasty})
assert json.dumps(nasty) in expr assert json.dumps("^" + nasty + "$") in expr # the regex is a JSON literal too
assert '\\"' in expr assert '\\"' in expr
assert json.dumps(nasty) in location_jsonpath({"symbol": nasty})
def test_jsonpath_emits_parts_in_a_fixed_key_order(): def test_jsonpath_emits_parts_in_a_fixed_key_order():