diff --git a/alembic/versions/0083_code_shape_reason_code.py b/alembic/versions/0083_code_shape_reason_code.py new file mode 100644 index 0000000..6798ae7 --- /dev/null +++ b/alembic/versions/0083_code_shape_reason_code.py @@ -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") diff --git a/src/scribe/mcp/tools/shapes.py b/src/scribe/mcp/tools/shapes.py index 1d094a8..c5e9349 100644 --- a/src/scribe/mcp/tools/shapes.py +++ b/src/scribe/mcp/tools/shapes.py @@ -36,10 +36,14 @@ async def classify_shapes( Args: project_id: The project whose ledger is being judged. classifications: Objects of {path, symbol, status, kind?, snippet_id?, - reason?}. path+symbol name the shape exactly as list_shapes shows - it; kind ("sym"/"css") narrows when one file defines both. - snippet_id is required for canonical/instance/variant; reason is - required for variant/exempt. + reason?, reason_code?}. path+symbol name the shape exactly as + list_shapes shows it; kind ("sym"/"css") narrows when one file + defines both. snippet_id is required for canonical/instance/ + 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 "import" (carrying maps recorded elsewhere). @@ -145,6 +149,7 @@ async def classify_shapes_by_rule( reason: str = "", via: str = "agent", include_judged: bool = False, + reason_code: str = "", ) -> dict: """The sweep form of classify_shapes: ONE judgment applied to every 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. reason: Required for variant/exempt — the why, recorded on every row. 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 — `unclassified` and the sync's mechanical `scoped` stamp — a 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, kind=kind, snippet_id=snippet_id or None, reason=reason or None, via=via, include_judged=include_judged, + reason_code=reason_code or None, ) except ValueError as exc: return {"error": str(exc)} diff --git a/src/scribe/models/code_shape.py b/src/scribe/models/code_shape.py index a3b522b..4d3010a 100644 --- a/src/scribe/models/code_shape.py +++ b/src/scribe/models/code_shape.py @@ -26,6 +26,20 @@ from scribe.models.base import TimestampMixin, iso SHAPE_STATUSES = ("canonical", "instance", "variant", "exempt", "scoped", "unclassified") SHAPE_CLASSIFIERS = ("agent", "audit", "hook", "mechanical", "import") +# The reason catalogue (#2874): an OPTIONAL code beside the prose reason on +# variant/exempt rows, so the ledger can be filtered and aggregated by kind +# of one-off. The prose remains the record; the code is the index. +REASON_CODES = ( + "scoped-css", # a scoped rule styling one element (pre-#2869 rows) + "one-off-handler", # a view/component handler or loader, one per surface + "test-helper", # a test module's stub, driver or fixture data + "convention-plumbing", # registration, wiring, app factory — one of each + "pure-helper", # a sync module-private helper with no session + "generated", # generated source (theme.css, protos, bundles) + "script", # a standalone dev/CI script + "typed-record", # a NamedTuple / dataclass / error class — one each +) + # How the mechanical proposer (#2792) arrived at a proposal, strongest first. # `derive` is the odd one out: not "this is an instance of #N" but "this # shape repeats with NO canon — derive one first" (note 2786's derive-first @@ -105,6 +119,7 @@ class CodeShape(Base, TimestampMixin): BigInteger, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True ) reason: Mapped[str | None] = mapped_column(Text, nullable=True) + reason_code: Mapped[str | None] = mapped_column(Text, nullable=True) # REASON_CODES (#2874) classified_by: Mapped[str | None] = mapped_column(Text, nullable=True) classified_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True @@ -158,6 +173,7 @@ class CodeShape(Base, TimestampMixin): "status": self.status, "snippet_id": self.snippet_id, "reason": self.reason, + "reason_code": self.reason_code, "classified_by": self.classified_by, "classified_at": iso(self.classified_at), "first_seen_commit": self.first_seen_commit, @@ -189,6 +205,8 @@ class CodeShape(Base, TimestampMixin): out["snippet_id"] = self.snippet_id if self.classified_by: out["by"] = self.classified_by + if self.reason_code: + out["reason_code"] = self.reason_code proposal = self.proposal if proposal: out["proposal"] = proposal diff --git a/src/scribe/services/coverage.py b/src/scribe/services/coverage.py index 2d1db2d..89e12b0 100644 --- a/src/scribe/services/coverage.py +++ b/src/scribe/services/coverage.py @@ -488,6 +488,7 @@ async def compute_coverage( # confirm, the largest derive-first groups, and what this refresh did. "proposed": proposals["proposed"], "derive_groups": proposals["derive_groups"], + "top_canon": proposals.get("top_canon"), "proposer": proposer_stats, # The divergence readout (#2793): button B where button A is canon, # 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 ''}") if coverage.get("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: line += f" ({', '.join(standing)})" gaps = [g["dir"] for g in coverage.get("largest_gaps") or []] diff --git a/src/scribe/services/knowledge.py b/src/scribe/services/knowledge.py index e0a74f2..b0c2229 100644 --- a/src/scribe/services/knowledge.py +++ b/src/scribe/services/knowledge.py @@ -16,6 +16,7 @@ endorsed, so a one-off direct share has to be searched for rather than arriving in your ambient lists. """ import json +import re import logging from sqlalchemy import and_, func, or_, select @@ -76,6 +77,10 @@ def location_matches(data: dict | None, parts: dict[str, str]) -> bool: if all( _path_matches((loc.get(key) or "").strip(), want) 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 for key, want in parts.items() ): @@ -99,6 +104,11 @@ def location_jsonpath(parts: dict[str, str]) -> str: if key == "path": prefix = json.dumps(want.rstrip("/") + "/") 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: filters.append(f"@.{key} == {literal}") return f"$.locations[*] ? ({' && '.join(filters)})" diff --git a/src/scribe/services/shape_ledger.py b/src/scribe/services/shape_ledger.py index e0dc983..806c0e0 100644 --- a/src/scribe/services/shape_ledger.py +++ b/src/scribe/services/shape_ledger.py @@ -30,7 +30,7 @@ from typing import Iterable, NamedTuple from sqlalchemy import select 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 logger = logging.getLogger(__name__) @@ -183,7 +183,7 @@ def _event(row: CodeShape, event: str, at: datetime, *, commit: str = "") -> Cod async def _judge( 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: """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 @@ -194,6 +194,7 @@ async def _judge( 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 "" @@ -314,6 +315,12 @@ def validate_classifications(items: list[dict]) -> str | None: f"classifications[{i}]: status {status!r} needs a reason — " "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 @@ -389,6 +396,7 @@ async def classify_shapes( 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"), ) classified += 1 await session.commit() @@ -424,6 +432,7 @@ async def classify_shapes_where( reason: str | None = None, via: str = "agent", include_judged: bool = False, + reason_code: str | None = None, ) -> dict: """The sweep form of classify_shapes (#2868): one judgment applied to every live row under ``path`` whose symbol matches ``pattern`` (and @@ -443,7 +452,8 @@ async def classify_shapes_where( 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 ""} + "snippet_id": snippet_id or 0, "reason": reason or "", + "reason_code": reason_code or ""} error = validate_classifications([probe]) if error: raise ValueError(error.replace("classifications[0]", "rule")) @@ -465,7 +475,7 @@ async def classify_shapes_where( await _judge( session, row, status=status, 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}") 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 proposals await confirmation, and the largest derive-first groups.""" proposed = 0 + by_canon: dict[int, int] = {} groups: dict[str, dict] = {} files: dict[str, set[str]] = {} for row in rows: @@ -1221,6 +1232,7 @@ def proposal_summary(rows: Iterable[CodeShape], *, top: int = 8) -> dict: 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, { @@ -1245,7 +1257,11 @@ def proposal_summary(rows: Iterable[CodeShape], *, top: int = 8) -> dict: groups.values(), 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( diff --git a/tests/test_pattern_coverage.py b/tests/test_pattern_coverage.py index 0d1c0cb..c603fc7 100644 --- a/tests/test_pattern_coverage.py +++ b/tests/test_pattern_coverage.py @@ -497,6 +497,12 @@ def test_coverage_line_names_the_proposers_standing(): assert "; 90 unclassified (40 proposed, 2 derive groups), largest: src" in line line = coverage_line({**base, "proposed": 0, "derive_groups": [{"group": "a"}]}) 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(): diff --git a/tests/test_shape_ledger.py b/tests/test_shape_ledger.py index c0f923f..fd9a8c1 100644 --- a/tests/test_shape_ledger.py +++ b/tests/test_shape_ledger.py @@ -391,6 +391,22 @@ def test_compact_row_carries_identity_standing_and_the_proposers_word_only(): 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(): from scribe.models.code_shape import CodeShape from scribe.services.shape_ledger import rule_matches diff --git a/tests/test_snippet_location_filter.py b/tests/test_snippet_location_filter.py index 882c4dc..29617a6 100644 --- a/tests/test_snippet_location_filter.py +++ b/tests/test_snippet_location_filter.py @@ -56,6 +56,7 @@ def test_no_parts_matches_everything(): def test_matches_exact_repo_path_and_symbol(): data = _data({"repo": "Scribe", "path": "src/scribe/x.py", "symbol": "helper"}) 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, {"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(): expr = location_jsonpath({"repo": "Scribe", "symbol": "helper"}) 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 " && " 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.""" nasty = 'we"ird' 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 json.dumps(nasty) in location_jsonpath({"symbol": nasty}) def test_jsonpath_emits_parts_in_a_fixed_key_order():