src/scribe/services/shape_ledger.py

@@ -2135,6 +2135,35 @@ _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."""
@@ -2156,9 +2185,37 @@ 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) -> tuple[int, int, int] | None:
+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?"."""
+    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
+    (#4208). Without it the base rate was computed over every code symbol in
+    the directory as one bucket: "372 judged siblings" counted a dataclass, a
+    CSS-less constant and an async service unit as three comparable things,
+    and the share that came out of that was a statement about a population
+    nobody had 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 the acceptance case of
+    milestone #2793, not noise. Bucketing the denominator by form would take
+    the async canon out of a sync candidate's count and silence that flag —
+    the same inversion the first form gate made, one layer down.
+    """
     directory = _dir_of(path)
     async with async_session() as session:
         rows = (
@@ -2173,6 +2230,7 @@ async def canon_density(project_id: int, path: str, kind: str) -> tuple[int, int
             )
         ).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
@@ -2195,10 +2253,28 @@ async def write_time_divergence(
     edit. Returns [{symbol, kind, canon_snippet_id, instances, judged}]."""
     just_stamped = {(s["symbol"], s["kind"]): s["snippet_id"] for s in stamped}
     out: list[dict] = []
-    kinds = {k for k, _n in shapes}
+    if not shapes:
-    density = {k: await canon_density(project_id, path, k) for k in kinds}
-    if not any(density.values()):
         return out
+    # DENSITY IS NOW PER CANDIDATE, not per kind (#4208): 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 is the same one-or-two queries as before.
+    #
+    # The cost is that the row load below no longer sits 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(
@@ -2211,13 +2287,19 @@ async def write_time_divergence(
         ).scalars().all()
     by_key = {(r.symbol, r.kind): r for r in rows}
     for kind, name in shapes:
-        dom = density.get(kind)
+        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
-        row = by_key.get((name, kind))
         if row is not None and (
             row.status != "unclassified" or row.proposed_snippet_id == sid
         ):
@@ -2231,12 +2313,10 @@ async def write_time_divergence(
         # divergence prompt is ABOUT a mismatch, so requiring the candidate to
         # match would silence the check precisely where it belongs.
         #
-        # Signature from the payload first — a shape being written now may
+        # `mine` was read above, before the denominator was counted — the
-        # have no row yet — and an unreadable one produces a fair question
+        # same value serves both, and they must not be able to disagree.
-        # rather than a guess, because `families_conflict` needs both sides.
+        # An unreadable signature produces a fair question rather than a
-        mine = shape_form(
+        # guess, because `families_conflict` needs both sides.
-            signature_in(code, name, kind) or getattr(row, "signature", "") or "", kind
-        )
         if families_conflict(mine, cform):
             continue
         out.append({"symbol": name, "kind": kind, "canon_snippet_id": sid,

tests/test_shape_form_gate.py

@@ -26,8 +26,8 @@ from __future__ import annotations
 import pytest
 
 from scribe.services.shape_ledger import (
-    FORM_UNKNOWN, canon_form, families_conflict, forms_agree, shape_family,
+    FORM_UNKNOWN, canon_form, comparable_siblings, dominant_canon,
-    shape_form, signature_in,
+    families_conflict, forms_agree, shape_family, shape_form, signature_in,
 )
 
 
@@ -103,8 +103,17 @@ def test_how_many_of_the_five_the_divergence_gate_actually_silences() -> None:
     That is not a shortcoming of the gate, it is the limit of the signature:
     at this level those four are indistinguishable from #2793's acceptance
     case, where a sync `confirmDanger` beside an async confirm helper SHOULD
-    be flagged. Separating them needs #4204 option 2 (widen `kind`) or a
+    be flagged.
-    comparison of meaning."""
+
+    CORRECTED 2026-09-21 (#4208). This used to say separating them needs
+    "widen `kind` or a comparison of meaning", offering the two as
+    alternatives. Widening `kind` does not separate them: it buys the silence
+    by bucketing `fn` apart from `async-fn`, which takes the async canon out
+    of a sync candidate's denominator and silences #2793's acceptance case by
+    the identical mechanism, one layer down. Whatever separates these four
+    has to distinguish a helper that does the canon's JOB from one that does
+    not, and no signature carries that. A comparison of meaning is the only
+    lever, not one of two."""
     canon = shape_form("async def create_note(user_id: int, ...):", "sym")
     silenced = {
         sig: families_conflict(shape_form(sig, "sym"), canon)
@@ -323,3 +332,112 @@ def test_the_resemblance_floor_is_above_the_retrieval_floors() -> None:
     from scribe.services.shape_ledger import _RESEMBLE_MIN
 
     assert _RESEMBLE_MIN >= 0.80
+
+
+# ── The denominator, not the gate (#4208) ─────────────────────────────────
+#
+# `dominant_canon` is a base rate, and a base rate is only a statement about
+# something if its denominator is a population somebody asked a question
+# about. It counted every code symbol in a directory as one bucket: a frozen
+# dataclass, a module constant and an async service unit were three
+# comparable things, and "372 judged siblings" was the authority the
+# divergence line spoke with.
+#
+# `comparable_siblings` narrows it using the SAME predicate as the gate, so
+# the count and the verdict cannot drift into disagreeing about what
+# comparable means.
+
+
+# `_Row` above is reused rather than redefined. A second class of the same
+# name here shadowed the first — same fields, different `snippet_id` default —
+# and silently broke three `canon_form` tests that had been passing, which is
+# a neater demonstration of this file's subject than anything it asserts.
+SERVICE = "async def get_note(user_id: int, note_id: int) -> Note | None:"
+HELPER = "def is_registered(source: str) -> bool:"
+
+
+def test_a_type_is_not_counted_against_a_directory_of_callables() -> None:
+    rows = [_Row(SERVICE) for _ in range(6)] + [_Row("class Point:", snippet_id=9)]
+    assert len(comparable_siblings(rows, shape_form("class Point:"))) == 1
+
+
+def test_a_callable_is_not_counted_against_a_dataclass() -> None:
+    """The honest-denominator half, and the one that changes reported numbers:
+    `judged` stops overstating how much of the directory was ever comparable."""
+    rows = [_Row(SERVICE) for _ in range(6)] + [_Row("class Point:", snippet_id=9)]
+    assert len(comparable_siblings(rows, shape_form(HELPER))) == 6
+
+
+def test_the_async_canon_stays_in_a_sync_candidates_denominator() -> None:
+    """THE ACCEPTANCE CASE OF #2793, and the reason this narrows by family
+    rather than by form.
+
+    A hand-rolled sync `confirmDanger` in a directory where an async confirm
+    helper is canon must still be flagged. Bucketing the denominator by exact
+    form — which is what widening `kind` to carry `fn` vs `async-fn` amounts
+    to — would take the canon out of this count and silence it.
+    """
+    rows = [_Row("async def confirmDanger(message: str) -> bool:") for _ in range(5)]
+    kept = comparable_siblings(rows, shape_form("function confirmDanger(message) {"))
+    assert len(kept) == 5
+    assert dominant_canon(kept) is not None
+
+
+def test_an_unreadable_sibling_stays_counted() -> None:
+    """Quieter, never louder. Dropping unknown rows shrinks `judged`, raises
+    the dominant canon's share, and fires the check MORE on exactly the
+    directories it can read least."""
+    rows = [_Row(SERVICE) for _ in range(4)] + [_Row("")]
+    assert len(comparable_siblings(rows, shape_form(SERVICE))) == 5
+
+
+def test_an_unreadable_candidate_narrows_nothing() -> None:
+    """The other side of the same discipline: a candidate whose own signature
+    says nothing gets the full denominator, not a guessed one."""
+    rows = [_Row(SERVICE), _Row("class Point:")]
+    assert comparable_siblings(rows, FORM_UNKNOWN) == rows
+
+
+def test_a_family_passed_where_a_form_belongs_would_be_a_silent_no_op() -> None:
+    """A REGRESSION GUARD ON A BUG THIS CHANGE ACTUALLY HAD.
+
+    `families_conflict` coarsens both sides itself, so handing it a family
+    makes `shape_family("callable")` return "" and the whole narrowing becomes
+    a no-op — while every call site still reads as though it applied. Pinned
+    because the wrong value is the right TYPE and the failure is silent.
+    """
+    # A CALLABLE candidate, deliberately: `fn`'s family is `callable`, a word
+    # that is not itself a form, so `shape_family("callable")` is "" and the
+    # exclusion never fires. Picking `type` here would prove nothing — `type`
+    # is both a form and a family name, so passing the family still narrows
+    # and the bug hides. That near-miss is why this test exists at all.
+    rows = [_Row(SERVICE) for _ in range(6)] + [_Row("class Point:", snippet_id=9)]
+    form = shape_form("def helper(x) -> bool:")
+    assert shape_family(form) != form, "this test needs a form whose family differs"
+    by_form = comparable_siblings(rows, form)
+    by_family = comparable_siblings(rows, shape_family(form))
+    assert len(by_form) == 6, "the dataclass is not comparable to a function"
+    assert len(by_family) == len(rows), "a family narrows nothing — that is the bug"
+    assert by_form != by_family
+
+
+def test_the_four_survivors_still_prompt() -> None:
+    """HONEST ACCOUNTING, matching the sibling test above.
+
+    The narrowed denominator does not silence #4204's four `def` helpers, and
+    was never going to: they are callables, the canon is a callable, so
+    nothing is excluded from their count. What changes is that the count is
+    now over comparable things. Asserted so the claim cannot quietly rot into
+    "this fixed it".
+    """
+    rows = [_Row(SERVICE) for _ in range(6)] + [_Row("class Point:", snippet_id=9)]
+    for sig in (
+        "def _p(source, kind, what, **kw) -> tuple[str, Point]:",
+        "def get_point(source: str) -> Point | None:",
+        "def is_registered(source: str) -> bool:",
+        "def sources_expected_to_emit() -> list[str]:",
+    ):
+        kept = comparable_siblings(rows, shape_form(sig))
+        dom = dominant_canon(kept)
+        assert dom is not None, sig
+        assert not families_conflict(shape_form(sig), canon_form(kept, dom[0])), sig