alembic/versions/0086_code_shape_consumers.py

@@ -0,0 +1,35 @@
+"""code_shape_consumers — the CSS consumer map (milestone 302, note 2917)
+
+Revision ID: 0086
+Revises: 0085
+Create Date: 2026-08-23
+
+CSS is watched by name, by recipe, by token and by WHAT USES IT. This table
+holds the fourth: CSS shape → the file whose markup names its class, with how
+many times. Mechanical and recomputed by every coverage sync from the repo
+archive; the analogue of code_shape_uses for styling. Cascades with the shape.
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0086"
+down_revision = "0085"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "code_shape_consumers",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("shape_id", sa.Integer(), sa.ForeignKey("code_shapes.id", ondelete="CASCADE"), nullable=False),
+        sa.Column("path", sa.Text(), nullable=False),
+        sa.Column("count", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("basis", sa.Text(), nullable=False, server_default="template"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.UniqueConstraint("shape_id", "path", name="uq_code_shape_consumers_shape_path"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("code_shape_consumers")

docs/api-reference.md

@@ -173,7 +173,7 @@ endpoint at `/mcp`, not these REST routes.
 | GET | `/api/plugin/retrieve` | Title-first knowledge-injection candidates |
 | GET | `/api/plugin/processes` | Stored Processes for skill-stub sync |
 | GET | `/api/plugin/prior-art` | Write-path hint for the plugin hooks (params: `path`, `code`, `repo`, `shapes`, `exclude_ids`, `exclude_sync_ids`, `exclude_derive`); returns `context`, `note_ids`, `sync_note_ids`, `stamped`, `divergence`, `derive`, `derive_keys` |
-| GET / POST | `/api/projects/<id>/coverage`, `…/coverage/refresh` | Shape-ledger accounting (`pattern_coverage` line, counts, `derive_groups`, `derive_new`, `divergence`, `recheck`) |
+| GET / POST | `/api/projects/<id>/coverage`, `…/coverage/refresh` | Shape-ledger accounting (`pattern_coverage` line, counts, `derive_groups` — css groups carry `consumers`, `derive_new`, `unused_css`, `divergence`, `recheck`) |
 | GET / PUT | `/api/plugin/marketplace-url` | Read / set the plugin marketplace URL |
 
 ## Dashboard, Export, Trash, Users

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
-  "version": "0.1.41",
+  "version": "0.1.44",
   "author": { "name": "Bryan Van Deusen" },
   "mcpServers": {
     "scribe": {

plugin/README.md

@@ -55,15 +55,21 @@ On install you'll be asked for:
   the edit" — each with its own once-per-session dedup. A third, ledger-fed
   line names a duplicate family (no canon) or a canon recorded elsewhere for
   the names being written (its own dedup channel, `exclude_derive`).
-  Toggle in **Settings → Knowledge auto-inject**.
+  Fail-open but not fail-silent: a configured instance that does not answer
+  in time is said, once per outage ("Scribe did not answer … this write went
+  UNCHECKED"), so a session can tell "checked, nothing there" from "never
+  checked"; an answer clears the marker. The local by-name arm needs no
+  server and always runs. Toggle in **Settings → Knowledge auto-inject**.
 - `hooks/hooks.json` → PostToolUse hook on `Bash`
   (`hooks/scribe_after_write.sh`): code written through sed/heredocs/scripts
   never reaches the PreToolUse hook, so this one diffs the working tree after
   every Bash call (per-session path+blob snapshot; one `git status` when
   nothing changed) and runs the same arms on the definitions just written,
   through the same endpoint and the same dedup channels. `additionalContext`
-  only; silent on any failure. The extractor, the prose/data skip list and the
-  local by-name duplicate arm are shared in `hooks/scribe_defs.sh`.
+  only; never blocks, and shares the pre-write hook's once-per-outage "did not
+  answer" line (8 s budget here — it runs after the tool, so it gates
+  nothing). The extractor, the prose/data skip list, the local by-name
+  duplicate arm and the outage line are shared in `hooks/scribe_defs.sh`.
 - `skills/` → the universal process-skills, surfaced by description match.
 - `hooks/scribe_sync_processes.sh` (a 2nd SessionStart hook) + the `/scribe:sync`
   command → generate `~/.claude/skills/scribe-proc-*` stubs from your Scribe

plugin/hooks/scribe_after_write.sh

@@ -154,6 +154,8 @@ while IFS= read -r rel_path; do
 
   context=""
   body=""
+  reached=""      # "" unconfigured (no call owed) · 1 answered · 0 did not
+  unreached_context=""
   if [ -n "$url" ] && [ -n "$token" ]; then
     q=$(printf '%s' "$code" | head -c 1200)
     path_enc=$(printf '%s' "$rel_path" | jq -sRr '@uri' 2>/dev/null) || path_enc=""
@@ -177,9 +179,23 @@ while IFS= read -r rel_path; do
       [ -n "$derive_seen" ] && derive_exclude_q="&exclude_derive=${derive_seen}"
     fi
     if [ -n "$path_enc" ]; then
-      body=$(curl -fsS --max-time 4 \
+      # 8s, not the pre-write hook's 5: this hook runs AFTER the tool, so it
+      # gates nothing the session is waiting on, and the first prior-art call
+      # after a redeploy is a cold start (embedding warm-up, ~4.6s observed)
+      # that a 4s cap turned into a silent fail-open — the one write a
+      # session most wants the ledger's word on lost it.
+      reached=1
+      body=$(curl -fsS --max-time 8 \
         -H "Authorization: Bearer ${token}" \
-        "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${derive_exclude_q}${shapes_q}" 2>/dev/null) || body=""
+        "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${derive_exclude_q}${shapes_q}" 2>/dev/null) || { body=""; reached=0; }
+      # A call that was owed and didn't come back is said, once per outage
+      # (#2932) — shared marker with the pre-write hook, so one outage is one
+      # line however the code was written.
+      if [ "$reached" = 1 ]; then
+        scribe_reached "$state_dir" "$safe_sid"
+      else
+        unreached_context=$(scribe_unreached "$state_dir" "$safe_sid" 8 "$rel_path")
+      fi
     fi
     if [ -n "$body" ]; then
       context=$(printf '%s' "$body" | jq -r '.context // empty' 2>/dev/null) || context=""
@@ -197,8 +213,10 @@ while IFS= read -r rel_path; do
   fi
 
   # The record nudge (#2664), same gate as the pre-write hook: duplication
-  # demonstrated locally AND nothing recorded for it.
-  if [ -n "$local_lines" ]; then
+  # demonstrated locally AND nothing recorded for it — and (#2932) never on a
+  # call that did not answer; "nothing recorded" is a claim only an answer
+  # can back.
+  if [ -n "$local_lines" ] && [ "$reached" != 0 ]; then
     n_recorded=$(printf '%s' "$body" | jq -r '.note_ids | length' 2>/dev/null) || n_recorded=0
     if [ "${n_recorded:-0}" = "0" ] || [ "$n_recorded" = "" ]; then
       local_context="${local_context}"$'\n'"> None of those existing copies is recorded in Scribe. If the version just written is the canonical one — or this edit is consolidating the copies — record it now with create_snippet so the next session is offered it instead of writing another copy."
@@ -210,6 +228,10 @@ while IFS= read -r rel_path; do
     [ -n "$part" ] && part="${part}"$'\n'
     part="${part}${context}"
   fi
+  if [ -n "$unreached_context" ]; then
+    [ -n "$part" ] && part="${part}"$'\n'
+    part="${part}${unreached_context}"
+  fi
   [ -n "$part" ] || continue
   [ -n "$combined" ] && combined="${combined}"$'\n'
   combined="${combined}${part}"

plugin/hooks/scribe_defs.sh

@@ -11,6 +11,9 @@
 #   scribe_defs                stdin code → "kind<TAB>name" per definition
 #   scribe_local_dups ROOT REL "kind<TAB>name" lines on stdin → the by-name
 #                              local-duplicate lines (ARM 1, #2280)
+#   scribe_unreached STATE SID SECS REL  the "Scribe didn't answer" line, once
+#                              per outage (#2932) — or nothing, if said lately
+#   scribe_reached STATE SID   the server answered: the next outage speaks again
 #
 # Sourced, not executed: `. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"`.
 
@@ -114,3 +117,31 @@ scribe_local_dups() {
     printf '> - `%s` is already defined in %s other file(s): %s\n' "$label" "$count" "$files"
   done
 }
+
+# ---------------------------------------------------------------------------
+# The blind spot made visible (#2932). Both write-path hooks fail OPEN when the
+# instance is slow or down — right for noise, wrong for silence: a session
+# cannot tell "the ledger checked and found nothing" from "the ledger never
+# answered", and a self-surfacing system cannot afford an invisible miss (the
+# first write after a redeploy lost its derive line to a 4s cold start and
+# nobody knew). So a failed call says so — ONCE per outage: the marker holds
+# the time it last spoke; within ten minutes of that it stays quiet, and a
+# successful call clears it so the next outage announces itself afresh.
+# Unconfigured installs never reach this: no URL/token means no call was owed.
+_SCRIBE_UNREACHED_QUIET=600
+
+scribe_unreached() {
+  local marker="$1/$2.unreached" now last
+  now=$(date +%s 2>/dev/null) || now=0
+  if [ -f "$marker" ]; then
+    last=$(cat "$marker" 2>/dev/null) || last=0
+    case "$last" in ''|*[!0-9]*) last=0 ;; esac
+    [ $((now - last)) -lt "$_SCRIBE_UNREACHED_QUIET" ] && return 0
+  fi
+  printf '%s' "$now" > "$marker" 2>/dev/null || true
+  printf '> Scribe did not answer the prior-art check for `%s` within %ss — this write went UNCHECKED against the record and the shape ledger (the local by-name arm, if it spoke above, needed no server). If the name matters, check it yourself: `search` for the concept, `list_shapes(project_id, path=…)` for the ledger. Said once per outage; if it keeps happening the instance is slow or down.' "$4" "$3"
+}
+
+scribe_reached() {
+  rm -f "$1/$2.unreached" 2>/dev/null || true
+}

plugin/hooks/scribe_prior_art.sh

@@ -203,11 +203,20 @@ if [ -n "$session_id" ]; then
   fi
 fi
 
-# `|| true`, not `|| exit 0`: an unreachable instance must not discard a local
-# finding that needed no instance to produce.
+# Not `|| exit 0`: an unreachable instance must not discard a local finding
+# that needed no instance to produce. And not silence either (#2932): a call
+# that was owed and didn't come back is said, once per outage, so the session
+# knows this write went unchecked.
+reached=1
 body=$(curl -fsS --max-time 5 \
   -H "Authorization: Bearer ${token}" \
-  "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${derive_exclude_q}${shapes_q}" 2>/dev/null) || body=""
+  "${url%/}/api/plugin/prior-art?path=${path_enc}&code=${code_enc}${repo_q}${exclude_q}${sync_exclude_q}${derive_exclude_q}${shapes_q}" 2>/dev/null) || { body=""; reached=0; }
+unreached_context=""
+if [ "$reached" = 1 ]; then
+  scribe_reached "$state_dir" "${safe_sid:-nosession}"
+else
+  unreached_context=$(scribe_unreached "$state_dir" "${safe_sid:-nosession}" 5 "$rel_path")
+fi
 
 context=""
 if [ -n "$body" ]; then
@@ -234,9 +243,10 @@ fi
 # noise: the duplication is demonstrated, not guessed. Gated on BOTH sides so
 # an ordinary new helper (no other copies) and an already-recorded one (the
 # server spoke) stay nudge-free — a reflex that fires on everything is one
-# that gets skipped. An unreachable server counts as "nothing recorded": the
-# local finding needed no server, and the nudge fails open with it.
-if [ -n "$local_lines" ]; then
+# that gets skipped. A server that did not ANSWER earns no nudge (#2932): "none
+# of those copies is recorded" is a claim only an answer can back — the
+# unreached line says what actually happened instead.
+if [ -n "$local_lines" ] && [ "$reached" = 1 ]; then
   n_recorded=$(printf '%s' "$body" | jq -r '.note_ids | length' 2>/dev/null) || n_recorded=0
   if [ "${n_recorded:-0}" = "0" ] || [ "$n_recorded" = "" ]; then
     local_context="${local_context}"$'\n'"> None of those existing copies is recorded in Scribe. If the version being written is the canonical one — or this edit is consolidating the copies — record it now with create_snippet (name, code, when-to-reach-for-it, location) so the next session is offered it instead of writing another copy."
@@ -251,6 +261,10 @@ if [ -n "$context" ]; then
   [ -n "$combined" ] && combined="${combined}"$'\n'
   combined="${combined}${context}"
 fi
+if [ -n "$unreached_context" ]; then
+  [ -n "$combined" ] && combined="${combined}"$'\n'
+  combined="${combined}${unreached_context}"
+fi
 [ -n "$combined" ] || exit 0
 
 # No permissionDecision: this is a nudge, not a gate. The write goes ahead.

plugin/skills/shape-accounting/SKILL.md

@@ -118,6 +118,14 @@ the last sweep left it. Three surfaces say so without anyone running an audit
   under different names are never a family. Derive a CSS family by moving
   the recipe to the shared sheet and recording it; a class name reused for
   genuinely different things is dismissed with `reason_code="scoped-css"`.
+  The datum that decides between the two is **what renders it**: every css
+  row carries `used_by` (the files whose markup names the class — the CSS
+  consumer map, milestone 302), a derive group carries the family's
+  `consumers`, and the write-path line says "used by N template(s)". Many
+  templates, one recipe → derive; one template each, different purposes →
+  dismiss. `list_shapes(flag="unused-css")` is the map's negative space —
+  css rules no template names, a deletion candidate to look at, never
+  auto-deleted (a class built at runtime is invisible to the map).
 
 After the one-time pay-down the derive queue reads empty; anything in it
 afterwards is drift of the moment, and the hint already said so at the write.

scripts/check_plugin.py

@@ -172,15 +172,24 @@ def check_shellcheck() -> None:
 # --- the fail-open contract ------------------------------------------------
 
 # Every hook promises never to break the operator's session: unconfigured or
-# unreachable, it exits 0. Three of them additionally promise SILENCE, because
-# they are pure enrichment. scribe_session_context.sh is the exception by
-# design — it always emits a static behavioural floor that needs no credentials
-# and no network, so "silent" would be the wrong assertion for it.
+# unreachable, it exits 0. Unconfigured, the enrichment hooks are SILENT — no
+# call was owed. scribe_session_context.sh is the exception by design — it
+# always emits a static behavioural floor that needs no credentials and no
+# network, so "silent" would be the wrong assertion for it.
+#
+# UNREACHABLE is different for the two write-path hooks since #2932: a call
+# that was owed and did not come back is SAID, once per outage ("> Scribe did
+# not answer …"), so a session can tell "checked, nothing there" from "never
+# checked". That line — or silence, when the once-per-outage marker in
+# ${TMPDIR:-/tmp}/scribe-priorart/ was set by a run in the last ten minutes —
+# is the only output allowed with no working instance; anything else is a hook
+# speaking on data it cannot have.
 #
 # This is the contract that made #2198 invisible for weeks, so it is worth
-# pinning: the bug and the healthy no-results case look identical from outside.
-# Pinning it does NOT make the failure visible; it makes sure the fail-open
-# behaviour is deliberate rather than accidental.
+# pinning: the bug and the healthy no-results case looked identical from
+# outside. #2932 is what finally makes the failure visible at the write; this
+# check makes sure the fail-open behaviour stays deliberate rather than
+# accidental.
 # A symbol that exists nowhere, ASSEMBLED rather than written literally.
 # The prior-art hook's local arm (#2280) fires with no credentials, so the
 # silence assertion below needs a name the repo genuinely lacks. Two traps,
@@ -218,6 +227,9 @@ SMOKE_EVENTS: dict[str, str] = {
 
 # The one hook that legitimately produces output with no credentials.
 STATIC_FLOOR = "scribe_session_context.sh"
+# The hooks that say so when a configured instance does not answer (#2932).
+OUTAGE_SPEAKERS = {"scribe_prior_art.sh", "scribe_after_write.sh"}
+OUTAGE_LINE = "> Scribe did not answer the prior-art check"
 
 
 def _run_hook(script: Path, event: str, env_extra: dict[str, str]) -> subprocess.CompletedProcess:
@@ -269,6 +281,17 @@ def check_fail_open() -> None:
                          f"behavioural floor must survive having no credentials")
                 else:
                     ok(f"{rel} [{label}]: exit 0, static floor present")
+            elif out and label == "unreachable" and script.name in OUTAGE_SPEAKERS:
+                # The only thing allowed here is the outage line itself.
+                try:
+                    ctx = json.loads(out)["hookSpecificOutput"]["additionalContext"]
+                except (ValueError, KeyError, TypeError):
+                    ctx = ""
+                if ctx.startswith(OUTAGE_LINE):
+                    ok(f"{rel} [{label}]: exit 0, says the instance did not answer")
+                else:
+                    fail(f"{rel} [{label}]: emitted output with no working instance "
+                         f"that is not the outage line:\n        {out[:200]}")
             elif out:
                 fail(f"{rel} [{label}]: emitted output with no working instance:\n"
                      f"        {out[:200]}")

src/scribe/mcp/tools/shapes.py

@@ -116,10 +116,17 @@ async def list_shapes(
             classify it: instance if it should use the canon, variant with
             the why if deliberate); "recheck": judged instances/variants
             whose body changed since judged (the judgment stands; confirm
-            it again with classify_shapes, or re-judge).
+            it again with classify_shapes, or re-judge); "unused-css"
+            (milestone 302): live css rules no file's markup names — a
+            deletion candidate to look at, never auto-deleted (the map reads
+            templates only; a class built at runtime is invisible to it).
 
     Returns {"shapes": [...], "total": N} — total counts every match, not
-    just this page. Each row's `classified_by` says who judged: agent /
+    just this page. Every css row carries `used_by` {count, paths} — the
+    files whose markup names its class (milestone 302, the CSS consumer
+    map: a scoped rule is used by its own template; a shared recipe by
+    many; a count of 0 is "no template names it"). Each row's
+    `classified_by` says who judged: agent /
     audit / import are judgments; `mechanical` is the canonical stamp the
     sync applies; `hook` is write-path EVIDENCE (#2791) — the session pulled
     a snippet and then wrote code referencing/resembling it, so the shape
@@ -145,10 +152,12 @@ async def list_shapes(
         include_vanished=include_vanished, limit=limit, offset=offset,
         proposal=proposal, flag=flag, uses=uses,
     )
-    return {
-        "shapes": [r.to_compact() if compact else r.to_dict() for r in rows],
-        "total": total,
-    }
+    shapes = [r.to_compact() if compact else r.to_dict() for r in rows]
+    used_by = await shape_ledger_svc.used_by_map(rows)
+    for row, out in zip(rows, shapes):
+        if row.id in used_by:
+            out["used_by"] = used_by[row.id]
+    return {"shapes": shapes, "total": total}
 
 
 async def classify_shapes_by_rule(
@@ -295,7 +304,10 @@ async def refresh_pattern_coverage(project_id: int) -> dict:
     Returns the accounting payload — total, accounted, counts by status,
     unclassified, repos, largest_gaps, `proposed` (canon proposals awaiting
     confirmation), `derive_groups` (the biggest repeats-with-no-canon
-    families), `derive_new` (copies that joined a family since the previous
+    families, each css one with `consumers` — the files whose markup
+    render it, milestone 302), `unused_css` (css rules no template names;
+    None where the map has no evidence of templates), `derive_new` (copies
+    that joined a family since the previous
     refresh — the drift to act on now: derive the canon, don't queue an
     audit), `proposer` (what this refresh examined) — plus
     `pattern_coverage`, the same one-line summary enter_project carries.

src/scribe/models/__init__.py

@@ -44,6 +44,6 @@ from scribe.models.rulebook import (  # noqa: E402, F401
 )
 from scribe.models.repo_binding import RepoBinding  # noqa: E402, F401
 from scribe.models.forge_connection import ForgeConnection  # noqa: E402, F401
-from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse  # noqa: E402, F401
+from scribe.models.code_shape import CodeShape, CodeShapeConsumer, CodeShapeEvent, CodeShapeUse  # noqa: E402, F401
 from scribe.models.system import System, RecordSystem  # noqa: E402, F401
 from scribe.models.design_system import DesignSystem, DesignToken  # noqa: E402, F401

src/scribe/models/code_shape.py

@@ -265,6 +265,52 @@ class CodeShapeUse(Base):
         }
 
 
+# How a consumer edge was established (milestone 302). `template` is the
+# sync's mechanical read of a file's markup (class= / :class= / className=);
+# the vocabulary is a list so a later basis (a stylesheet `@apply`, a script's
+# classList) has a name without a schema change.
+CONSUMER_BASES = ("template",)
+
+
+class CodeShapeConsumer(Base):
+    """One consumer edge: CSS shape → the file whose markup names its class
+    (milestone 302; note 2917 — CSS is watched by name, by recipe, by token
+    and by WHAT USES IT). The analogue of CodeShapeUse for styling: `uses`
+    says what a shape calls, this says who renders a class. Rows, not prose,
+    so "is this recipe shared or scoped?" is a count, not a guess.
+
+    Mechanical and fully recomputable: every coverage sync rebuilds a repo's
+    edges from its archive, so the table is not backed up (see
+    services/backup._NOT_INCLUDED). Cascades with the shape.
+    """
+
+    __tablename__ = "code_shape_consumers"
+    __table_args__ = (
+        UniqueConstraint("shape_id", "path", name="uq_code_shape_consumers_shape_path"),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    shape_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("code_shapes.id", ondelete="CASCADE"), nullable=False
+    )
+    path: Mapped[str] = mapped_column(Text, nullable=False)
+    count: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    basis: Mapped[str] = mapped_column(Text, nullable=False, default="template")
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "shape_id": self.shape_id,
+            "path": self.path,
+            "count": self.count,
+            "basis": self.basis,
+            "created_at": iso(self.created_at),
+        }
+
+
 # What a shape's history records (#2793). Not "appeared" — first_seen and
 # created_at already say that on the row; history is for what CHANGED:
 SHAPE_EVENTS = ("classified", "vanished", "reappeared", "drifted")

src/scribe/services/backup.py

@@ -92,6 +92,10 @@ _NOT_INCLUDED = [
     # deliberately not exported either, so restored projects fall back to
     # keyring-by-host resolution — the documented unpinned behavior (#2778).
     "forge_connections",
+    # Derived, like note_embeddings: the CSS consumer map (milestone 302) is
+    # rebuilt from the repo archive by every coverage sync, and carries no
+    # judgment — the first refresh after a restore recreates it exactly.
+    "code_shape_consumers",
 ]
 
 

src/scribe/services/coverage.py

@@ -269,6 +269,91 @@ def scoped_definitions(path: str, text: str, defs: list[Definition]) -> set[tupl
     return out
 
 
+# --- template class references: the CSS consumer map (milestone 302) ---------
+
+# Files whose MARKUP can consume a class. Styling consumers are templates —
+# `querySelector('.x')` / classList in scripts are deliberately not read in
+# v1 (note 2917: watch CSS by name, by recipe, by token and by what uses it;
+# "what uses it" is the template).
+_TEMPLATE_SUFFIXES = (
+    ".vue", ".html", ".htm", ".jsx", ".tsx", ".js", ".ts", ".svelte", ".astro",
+)
+_CLASS_TOKEN_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_-]*$")
+# Static: class="a b" / class='a b' / className="a b". The lookbehind keeps
+# `:class=`, `v-bind:class=`, `data-class=` and `headerClass=` out of the
+# static form (the Vue/React dynamic forms are read below; the others are
+# not class attributes).
+_STATIC_CLASS_RE = re.compile(
+    r"""(?<![:\w.-])(?:class|className)\s*=\s*(?:"([^"]*)"|'([^']*)')"""
+)
+# Dynamic: Vue `:class="…"` / `v-bind:class="…"`, React `className={…}` (one
+# level of nested braces — an object literal inside the expression).
+_DYNAMIC_CLASS_RE = re.compile(
+    r""":class\s*=\s*(?:"([^"]*)"|'([^']*)')"""
+    r"""|(?<![:\w.-])className\s*=\s*\{((?:[^{}]|\{[^{}]*\})*)\}"""
+)
+# Svelte's directive form: class:active={cond}.
+_SVELTE_CLASS_RE = re.compile(r"(?<![:\w.-])class:([A-Za-z_][A-Za-z0-9_-]*)\s*=")
+# Inside a dynamic expression: string literals (ternary arms, array items,
+# quoted object keys) and the bare keys of object literals.
+_STR_LIT_RE = re.compile(r"""'([^'\\]*)'|"([^"\\]*)"|`([^`]*)`""")
+_OBJ_SPAN_RE = re.compile(r"\{([^{}]*)\}")
+_OBJ_KEY_RE = re.compile(r"(?:^|[{,\s])([A-Za-z_][A-Za-z0-9_-]*)\s*:(?!:)")
+_TEMPLATE_HOLE_RE = re.compile(r"\$\{[^}]*\}")
+# A server-side / mustache interpolation inside a static value (`{{ cls }}`,
+# `{% if %}`): unknowable at read time, contributes no token.
+_MUSTACHE_RE = re.compile(r"\{[{%][^}]*[}%]\}")
+
+
+def _class_tokens(value: str) -> list[str]:
+    """The class tokens of a static attribute value: whitespace-split, only
+    well-formed names (an interpolation like `{{ cls }}` contributes none)."""
+    return [t for t in _MUSTACHE_RE.sub(" ", value).split() if _CLASS_TOKEN_RE.match(t)]
+
+
+def _dynamic_class_tokens(expr: str) -> list[str]:
+    """Class tokens named by a dynamic class expression: every string
+    literal's tokens (a template literal's static text only — its `${…}`
+    holes are unknowable) and the bare keys of object literals. Bare
+    identifiers elsewhere (`cond ? clsA : clsB`) are variables, not names."""
+    out: list[str] = []
+    for m in _STR_LIT_RE.finditer(expr):
+        literal = m.group(1) if m.group(1) is not None else (
+            m.group(2) if m.group(2) is not None else m.group(3)
+        )
+        if m.group(3) is not None:
+            literal = _TEMPLATE_HOLE_RE.sub(" ", literal)
+        out.extend(_class_tokens(literal))
+    for span in _OBJ_SPAN_RE.finditer(expr):
+        # Quoted keys were read as literals above; bare keys here.
+        body = _STR_LIT_RE.sub(" ", span.group(1))
+        out.extend(k for k in _OBJ_KEY_RE.findall(body) if _CLASS_TOKEN_RE.match(k))
+    return out
+
+
+def class_references(path: str, text: str) -> dict[str, int]:
+    """class token → how many times this file's markup names it. Empty for
+    files that carry no markup (by suffix). Reads the static `class=` /
+    `className=` attributes, the Vue and React dynamic forms and Svelte's
+    `class:x` directive; never a CSS selector (`.x {` is a definition, read
+    by extract_definitions) and never a script's `querySelector('.x')`."""
+    if not (path or "").lower().endswith(_TEMPLATE_SUFFIXES):
+        return {}
+    counts: dict[str, int] = {}
+
+    def bump(tokens: list[str]) -> None:
+        for t in tokens:
+            counts[t] = counts.get(t, 0) + 1
+
+    for m in _STATIC_CLASS_RE.finditer(text):
+        bump(_class_tokens(m.group(1) if m.group(1) is not None else m.group(2)))
+    for m in _DYNAMIC_CLASS_RE.finditer(text):
+        expr = next((g for g in m.groups() if g is not None), "")
+        bump(_dynamic_class_tokens(expr))
+    bump([m.group(1) for m in _SVELTE_CLASS_RE.finditer(text)])
+    return counts
+
+
 def extract_shapes(text: str) -> list[tuple[str, str]]:
     """Every (kind, name) this text DEFINES — kind is "css" or "sym".
 
@@ -307,14 +392,31 @@ def shapes_from_archive(blob: bytes) -> list[tuple[str, str, str]]:
     return [(d.path, d.kind, d.name) for d in definitions_from_archive(blob)]
 
 
+class ArchiveScan(NamedTuple):
+    """One walk of a repo tarball: what each file DEFINES (the ledger rows)
+    and which class names each file's markup REFERENCES (the CSS consumer
+    map, milestone 302) — read together because the bodies are in hand once."""
+
+    definitions: list[ArchiveShape]
+    references: dict[str, dict[str, int]]   # path → class token → count
+
+
 def definitions_from_archive(blob: bytes) -> list[ArchiveShape]:
-    """Every definition in a repo tarball, with its fingerprint and body.
+    """Every definition in a repo tarball, with its fingerprint and body —
+    the definitions half of scan_archive."""
+    return scan_archive(blob).definitions
+
+
+def scan_archive(blob: bytes) -> ArchiveScan:
+    """Every definition in a repo tarball, with its fingerprint and body,
+    plus each template-bearing file's class references.
 
     Forge archives wrap content in a single top-level directory (repo-ref/);
     that component is stripped so paths match recorded snippet locations,
     which are repo-relative. Non-UTF-8 files are binaries and skipped.
     """
     shapes: list[ArchiveShape] = []
+    references: dict[str, dict[str, int]] = {}
     with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tar:
         for member in tar:
             if not member.isfile() or "/" not in member.name:
@@ -338,7 +440,10 @@ def definitions_from_archive(blob: bytes) -> list[ArchiveShape]:
                 )
                 for d in defs
             )
-    return shapes
+            refs = class_references(path, text)
+            if refs:
+                references[path] = refs
+    return ArchiveScan(shapes, references)
 
 
 # --- matching shapes against recorded locations ------------------------------
@@ -450,7 +555,8 @@ async def compute_coverage(
         # The binding's own ref when it names one (#2873: a dev-first project
         # has its ledger follow dev), else the forge's default branch.
         ref = binding.ref or await forge.default_branch(api_repo)
-        definitions = definitions_from_archive(await forge.archive(api_repo, ref))
+        scan = scan_archive(await forge.archive(api_repo, ref))
+        definitions = scan.definitions
         # The head commit is provenance sugar on the ledger rows; failing to
         # learn it must not fail the sync — the ref names the point well
         # enough and the row timestamps carry the when.
@@ -462,6 +568,13 @@ async def compute_coverage(
             project_id, key, definitions, seen_marker=marker
         )
         served.append((key, ref))
+        # The CSS consumer map (milestone 302) rides the same archive: which
+        # files' markup names each class. Mechanical and recomputable, so it
+        # must not be able to fail the refresh either.
+        try:
+            await shape_ledger.sync_repo_consumers(project_id, key, scan.references)
+        except Exception:
+            logger.warning("consumer map sync failed for %s", key, exc_info=True)
         # Propose while the bodies are in hand — the one moment they exist.
         # Canonical marking below only touches rows the proposer leaves
         # alone (a canon's own location never gets a proposal), so the order
@@ -516,7 +629,23 @@ async def compute_coverage(
         agg["accounted"] += row.status != "unclassified"
 
     unclassified = counts.pop("unclassified")
-    proposals = shape_ledger.proposal_summary(rows)
+    # The CSS consumer map's readout (milestone 302): which files render each
+    # css row — on the derive groups (a shared recipe vs a scoped one is a
+    # count), and the negative space: css rules no template names. "Unused"
+    # is measured only where the map has evidence of templates at all (one
+    # edge somewhere); a repo of bare stylesheets is "not measured", not
+    # "all unused".
+    css_rows = [r for r in rows if r.kind == "css"]
+    consumer_paths: dict[int, list[str]] = {}
+    unused_css = None
+    try:
+        edges = await shape_ledger.consumers_of([r.id for r in css_rows])
+        consumer_paths = {sid: [e.path for e in es] for sid, es in edges.items()}
+        if consumer_paths:
+            unused_css = sum(1 for r in css_rows if r.id not in consumer_paths)
+    except Exception:
+        logger.warning("consumer map read failed", exc_info=True)
+    proposals = shape_ledger.proposal_summary(rows, consumer_paths=consumer_paths)
     divergence = shape_ledger.divergence_summary(rows)
     derive_new = shape_ledger.derive_new_summary(rows, since=since)
     return {
@@ -533,6 +662,9 @@ async def compute_coverage(
         # duplicate family — what the arrival line names so drift is noticed
         # on entering, not found by an audit.
         "derive_new": derive_new,
+        # The consumer map's negative space (milestone 302): live css rules
+        # no template names — None when the map has no evidence of templates.
+        "unused_css": unused_css,
         "proposer": proposer_stats,
         # The divergence readout (#2793): button B where button A is canon,
         # and judged shapes whose bodies moved since they were judged.
@@ -710,7 +842,16 @@ def coverage_line(coverage: dict) -> str:
         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")
+        top_copy = f"top copy {first['label']} ×{first['files']} files"
+        # A css family says what renders it (milestone 302): the count that
+        # tells a shared recipe from a scoped convention.
+        if "consumers" in first:
+            n_t = (first.get("consumers") or {}).get("count", 0)
+            top_copy += f" · used by {n_t} template{'s' if n_t != 1 else ''}"
+        standing.append(top_copy)
+    if coverage.get("unused_css"):
+        n_u = coverage["unused_css"]
+        standing.append(f"{n_u} unused class{'es' if n_u != 1 else ''}")
     if unclassified:
         line += f"; {unclassified} unclassified"
         if standing:

src/scribe/services/plugin_context.py

@@ -1073,6 +1073,18 @@ def _derive_line(path: str, derive: list[dict]) -> str:
             # files. CSS is only ever grouped this way (note 2917) — a class
             # is a recipe, and the recipe is what gets derived or dismissed.
             what = f"is a repeated name with no canon — defined in {n} other file(s)"
+        # What renders a css family (milestone 302): the consumer count is
+        # the datum that separates a shared recipe from a scoped convention.
+        cons = f.get("consumers")
+        if cons is not None:
+            n_t = cons.get("count", 0)
+            used = f"; used by {n_t} template{'s' if n_t != 1 else ''}"
+            if cons.get("paths"):
+                used += ": " + ", ".join(f"`{x}`" for x in cons["paths"])
+                extra = n_t - len(cons["paths"])
+                if extra > 0:
+                    used += f" +{extra} more"
+            files += used
         # The dismissal reason the family most likely earns: a class name
         # reused for different purposes is scoped styling; a code name reused
         # across modules is convention plumbing.

src/scribe/services/shape_ledger.py

@@ -30,7 +30,9 @@ from typing import Iterable, NamedTuple
 from sqlalchemy import select
 
 from scribe.models import async_session
-from scribe.models.code_shape import REASON_CODES, CodeShape, CodeShapeEvent, CodeShapeUse
+from scribe.models.code_shape import (
+    REASON_CODES, CodeShape, CodeShapeConsumer, CodeShapeEvent, CodeShapeUse,
+)
 from scribe.models.base import iso
 
 logger = logging.getLogger(__name__)
@@ -262,6 +264,118 @@ async def uses_of(shape_ids) -> dict[int, list[CodeShapeUse]]:
     return out
 
 
+# --- the CSS consumer map (milestone 302) ------------------------------------
+
+
+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 (note 2917): 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."""
+    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] = {}
+    for consumer, tokens in references.items():
+        for token, count in tokens.items():
+            rows = by_symbol.get(token)
+            if not rows:
+                continue
+            own = [sid for sid, path in rows if path == consumer]
+            targets = own or [sid for sid, _path in rows]
+            for sid in targets:
+                out[(sid, consumer)] = out.get((sid, consumer), 0) + int(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:
@@ -574,7 +688,8 @@ async def list_project_shapes(
     suggestion), "derive" (a repeats-with-no-canon group), or one basis
     name (symbol/reference/text/signature/semantic). ``flag`` narrows to
     the readout's asks (#2793): "divergence" (new where a canon dominates,
-    `diverges_from` names it) or "recheck" (a judged shape whose body moved).
+    `diverges_from` names it), "recheck" (a judged shape whose body moved),
+    or "unused-css" (milestone 302: a css rule no template names).
     """
     from sqlalchemy import func, or_
 
@@ -609,6 +724,13 @@ async def list_project_shapes(
         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 (milestone 302): 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 (#2870): rows with a uses edge to it, whatever
         # shape they themselves are.
@@ -1369,13 +1491,21 @@ async def apply_derive_groups(project_id: int) -> int:
     return grouped
 
 
-def proposal_summary(rows: Iterable[CodeShape], *, top: int = 8) -> dict:
+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."""
+    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
@@ -1396,8 +1526,14 @@ def proposal_summary(rows: Iterable[CodeShape], *, top: int = 8) -> dict:
             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 (#2872): the things an audit actually
     # consolidated were identical bodies under different names/files; a
     # name repeated across modules is usually convention. Within a tier,
@@ -1661,15 +1797,21 @@ async def write_time_derive(
         group = grouped[0].proposal_group
         members = [r for r in grouped if r.proposal_group == group]
         files = sorted({r.path for r in members})
-        out.append({
-            "symbol": name, "kind": kind, "key": group,
-            "family": {
+        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 (milestone 302): 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
 
 

tests/test_after_write_hook.py

@@ -102,20 +102,49 @@ def test_after_write_is_silent_where_it_has_nothing_to_say(tmp_path):
     (loose / "a.css").write_text(".x {\n  color: red;\n}\n")
     assert _run(loose, env, session="s-loose") == ""
     # A change that defines nothing (prose, a call-site edit) → nothing, even
-    # with the server unreachable (port 9 refuses): no definitions, no arms.
+    # with the server unreachable (port 9 refuses): no definitions, no call
+    # owed, so not even the #2932 outage line. (a.css above is removed first:
+    # it DOES define a shape, and an unanswered call for it would rightly speak.)
+    (repo / "a.css").unlink()
     (repo / "README.md").write_text("# notes\n")
     (repo / "b.py").write_text("def one():\n    return one_more()\n")
     assert _run(repo, env, session="s-quiet") == ""
 
 
-def test_after_write_local_arm_and_record_nudge_work_without_a_server(tmp_path):
-    """The local by-name arm needs no instance (#2280) and the record nudge
-    (#2664) fails open with it — a refused connection stands in for the
-    instance."""
+def test_after_write_local_arm_works_without_a_server_and_says_the_server_did_not_answer(tmp_path):
+    """The local by-name arm needs no instance (#2280). A configured instance
+    that does not ANSWER (a refused connection stands in for it) is said, once
+    per outage (#2932) — and the record nudge, which claims "nothing recorded",
+    is withheld: no answer backs that claim."""
     env = _env(tmp_path)
     repo = _repo(tmp_path, env)
     (repo / "d.py").write_text("def slug(t):\n    return t.lower()\n")
     out = _run(repo, env, session="s-local")
     ctx = json.loads(out)["hookSpecificOutput"]["additionalContext"]
     assert "`slug` is already defined in 1 other file(s): c.py" in ctx
+    assert "Scribe did not answer the prior-art check for `d.py` within 8s" in ctx
+    assert "UNCHECKED" in ctx
+    assert "None of those existing copies is recorded" not in ctx
+    marker = tmp_path / "scribe-priorart" / "s-local.unreached"
+    assert marker.is_file() and marker.read_text().isdigit()
+    # Still down a moment later: the local arm speaks, the outage line does not
+    # repeat (once per outage, not once per write).
+    (repo / "e.py").write_text("def slug(t):\n    return t.upper()\n")
+    out = _run(repo, env, session="s-local")
+    ctx = json.loads(out)["hookSpecificOutput"]["additionalContext"]
+    assert "`slug` is already defined in" in ctx
+    assert "did not answer" not in ctx
+
+
+def test_after_write_unconfigured_install_owes_no_call_and_keeps_the_record_nudge(tmp_path):
+    """No URL/token → no call was owed, so nothing is "unreached"; the local
+    arm and the record nudge (#2664) stand on their own, as before."""
+    env = {k: v for k, v in _env(tmp_path).items() if k not in ("SCRIBE_URL", "SCRIBE_TOKEN")}
+    repo = _repo(tmp_path, env)
+    (repo / "d.py").write_text("def slug(t):\n    return t.lower()\n")
+    out = _run(repo, env, session="s-unconf")
+    ctx = json.loads(out)["hookSpecificOutput"]["additionalContext"]
+    assert "`slug` is already defined in 1 other file(s): c.py" in ctx
     assert "create_snippet" in ctx
+    assert "did not answer" not in ctx
+    assert not (tmp_path / "scribe-priorart" / "s-unconf.unreached").exists()

tests/test_integration_shape_classify.py

@@ -683,6 +683,68 @@ async def test_derive_new_names_the_copy_that_joined_a_family_since_the_stamp(se
     assert derive_new_summary(rows, since=None)["count"] == 0
 
 
+@pytest.mark.integration
+async def test_consumer_map_syncs_edges_from_template_references(seeded):
+    """Milestone 302: the consumer edges follow the archive — own-file
+    resolution for a scoped class, fan-out to the shared sheet for a class a
+    template does not define, counts refreshed and stale edges removed on
+    the next sync, and a vanished row's edges gone with it."""
+    from scribe.services.shape_ledger import consumers_of, live_rows, sync_repo_consumers
+
+    pid = seeded["pid"]
+    defs = _defs(
+        ("v/A.vue", "css", "error-msg", ".error-msg {", ".error-msg { color: red }"),
+        ("v/B.vue", "css", "error-msg", ".error-msg {", ".error-msg { color: blue }"),
+        ("assets/components.css", "css", "btn-primary", ".btn-primary {", ".btn-primary { x: 1 }"),
+        ("assets/orphan.css", "css", "orphan", ".orphan {", ".orphan { y: 2 }"),   # no template names it
+    )
+    await sync_repo_shapes(pid, REPO, defs, seen_marker="m1")
+    refs = {
+        "v/A.vue": {"error-msg": 2, "btn-primary": 1},
+        "v/B.vue": {"error-msg": 1},
+        "v/C.vue": {"error-msg": 1, "btn-primary": 4},
+    }
+    # A and B consume their OWN error-msg; C defines none, so its use fans
+    # out to both rows; btn-primary resolves to the shared sheet from A and C.
+    assert await sync_repo_consumers(pid, REPO, refs) == 6
+    rows = {(r.path, r.symbol): r.id for r in await live_rows(pid) if r.kind == "css"}
+    edges = await consumers_of(rows.values())
+    view = {(p, s): [(e.path, e.count) for e in edges.get(i, [])] for (p, s), i in rows.items()}
+    assert view[("v/A.vue", "error-msg")] == [("v/A.vue", 2), ("v/C.vue", 1)]
+    assert view[("v/B.vue", "error-msg")] == [("v/B.vue", 1), ("v/C.vue", 1)]
+    assert view[("assets/components.css", "btn-primary")] == [("v/A.vue", 1), ("v/C.vue", 4)]
+    assert view[("assets/orphan.css", "orphan")] == []
+
+    # The next tree: C stops using error-msg, A uses btn-primary twice now.
+    refs2 = {"v/A.vue": {"error-msg": 2, "btn-primary": 2}, "v/B.vue": {"error-msg": 1}}
+    assert await sync_repo_consumers(pid, REPO, refs2) == 3
+    edges = await consumers_of(rows.values())
+    assert [(e.path, e.count) for e in edges[rows[("v/B.vue", "error-msg")]]] == [("v/B.vue", 1)]
+    assert [(e.path, e.count) for e in edges[rows[("assets/components.css", "btn-primary")]]] == [("v/A.vue", 2)]
+
+    # The readout side: used_by per css row, the unused-css flag, and the
+    # family's consumers on the write-path check.
+    from scribe.services.shape_ledger import (
+        apply_derive_groups, used_by_map, write_time_derive,
+    )
+    owner = seeded["owner"]
+    live = [r for r in await live_rows(pid) if r.kind == "css"]
+    used = await used_by_map(live)
+    assert used[rows[("v/B.vue", "error-msg")]] == {"count": 1, "paths": ["v/B.vue"]}
+    assert used[rows[("assets/orphan.css", "orphan")]] == {"count": 0, "paths": []}
+    unused, n = await list_project_shapes(owner, pid, flag="unused-css")
+    assert n == 1 and [(r.path, r.symbol) for r in unused] == [("assets/orphan.css", "orphan")]
+    await apply_derive_groups(pid)
+    out = await write_time_derive(pid, "v/New.vue", [("css", "error-msg")])
+    assert out and out[0]["family"]["consumers"] == {"count": 2, "paths": ["v/A.vue", "v/B.vue"]}
+
+    # B's rule vanishes from the tree → its edges go with the pass.
+    await sync_repo_shapes(pid, REPO, [d for d in defs if d[0] != "v/B.vue"], seen_marker="m2")
+    await sync_repo_consumers(pid, REPO, refs2)
+    edges = await consumers_of(rows.values())
+    assert rows[("v/B.vue", "error-msg")] not in edges
+
+
 # --- #2793: the divergence readout against real rows -------------------------
 
 

tests/test_pattern_coverage.py

@@ -16,7 +16,10 @@ import pytest
 import pytest_asyncio
 
 from scribe.services.coverage import (
+    ArchiveScan,
+    class_references,
     coverage_line,
+    scan_archive,
     extract_shapes,
     largest_gaps,
     scannable,
@@ -118,6 +121,80 @@ def test_shapes_from_archive_strips_the_wrapper_and_gates_files():
     assert shapes_from_archive(_tarball(TREE)) == TREE_SHAPES
 
 
+# --- unit: template class references — the CSS consumer map (milestone 302) --
+
+
+def test_class_references_reads_vue_static_and_dynamic_forms_only():
+    """A template's class attributes name the classes it consumes: the static
+    `class=`, the Vue dynamic object/array/ternary forms (string literals and
+    bare object keys), never a selector in <style>, a `class Foo` in
+    <script>, a `querySelector('.x')`, or a look-alike attribute."""
+    vue = (
+        "<template>\n"
+        '  <div class="card card--wide" :class="{ active: isOpen, \'is-error\': err }">\n'
+        '    <span :class="[ \'pill\', cond ? \'pill-on\' : \'pill-off\', other ]" />\n'
+        '    <p class="card" v-bind:class="open ? openCls : \'closed\'">{{ t }}</p>\n'
+        '    <i data-class="nope" headerClass="nope2" />\n'
+        "  </div>\n"
+        "</template>\n"
+        '<script setup lang="ts">\n'
+        "class Foo {}\n"
+        "const el = document.querySelector('.zap')\n"
+        "</script>\n"
+        "<style scoped>\n"
+        ".card { color: red; }\n"
+        ".zap { color: blue; }\n"
+        "</style>\n"
+    )
+    assert class_references("a/B.vue", vue) == {
+        "card": 2, "card--wide": 1, "active": 1, "is-error": 1,
+        "pill": 1, "pill-on": 1, "pill-off": 1, "closed": 1,
+    }
+
+
+def test_class_references_reads_react_svelte_and_server_templates():
+    tsx = (
+        "export function X({ on }: { on: boolean }) {\n"
+        '  return <button className="btn btn-primary" data-x="y">\n'
+        "    <i className={on ? 'tab tab-on' : 'tab'} />\n"
+        "    <b className={`chip ${on ? 'chip-on' : ''} chip-sm`} />\n"
+        "    <u className={cn({ pill: on, 'pill-off': !on })} />\n"
+        "  </button>\n"
+        "}\n"
+    )
+    # A template literal's static text counts; its `${…}` hole is unknowable
+    # (chip-on sits inside the hole's own ternary and is NOT claimed).
+    assert class_references("a/x.tsx", tsx) == {
+        "btn": 1, "btn-primary": 1, "tab": 2, "tab-on": 1,
+        "chip": 1, "chip-sm": 1, "pill": 1, "pill-off": 1,
+    }
+    assert class_references("a/y.svelte", '<div class:active={on} class="row">') == {
+        "row": 1, "active": 1,
+    }
+    # A server-side interpolation contributes no token; a literal class inside
+    # a template conditional still does.
+    html = '<div class="row {{ cls }} col-2 {% if x %}y{% endif %}">'
+    assert class_references("t/p.html", html) == {"row": 1, "col-2": 1, "y": 1}
+    # Not a template-bearing file: nothing, however it reads.
+    assert class_references("a/z.py", 'html = \'<div class="row">\'') == {}
+
+
+def test_scan_archive_returns_definitions_and_references_from_one_walk():
+    tree = dict(TREE)
+    tree["web/Card.vue"] = (
+        b'<template><div class="btn card">x</div></template>\n'
+        b"<style scoped>\n.card {\n  color: red;\n}\n</style>\n"
+    )
+    scan = scan_archive(_tarball(tree))
+    assert isinstance(scan, ArchiveScan)
+    assert [(d.path, d.kind, d.name) for d in scan.definitions] == TREE_SHAPES + [
+        ("web/Card.vue", "css", "card"),
+    ]
+    # Only files whose markup names a class appear; the .py/.css files don't.
+    assert scan.references == {"web/Card.vue": {"btn": 1, "card": 1}}
+    assert shapes_from_archive(_tarball(tree)) == [(d.path, d.kind, d.name) for d in scan.definitions]
+
+
 # --- unit: the covering predicate (lives with the ledger since #2788) --------
 
 
@@ -527,6 +604,17 @@ 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
+    # Milestone 302: a css top copy says what renders it; unused classes
+    # join the standing block only when measured (None = no evidence).
+    line = coverage_line({**base, "unclassified": 0, "proposed": 0, "derive_groups": [
+        {"group": "name:css:error-msg", "label": ".error-msg", "files": 6,
+         "consumers": {"count": 6, "paths": ["a.vue"]}}], "unused_css": 3})
+    assert "top copy .error-msg ×6 files · used by 6 templates" in line
+    assert "3 unused classes" in line
+    line = coverage_line({**base, "unclassified": 0, "proposed": 0, "derive_groups": [
+        {"group": "name:css:x", "label": ".x", "files": 2,
+         "consumers": {"count": 1, "paths": ["a.vue"]}}], "unused_css": None})
+    assert "top copy .x ×2 files · used by 1 template" in line and "unused" not 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},

tests/test_shape_ledger.py

@@ -380,6 +380,17 @@ def test_proposal_summary_ranks_body_identical_groups_first_and_sees_scoped_rows
         row("v/J.vue", "closed-msg", "dup:abc", status="exempt"),
     ]
     out = proposal_summary(rows)
+    # Milestone 302: with consumer paths in hand, each css group says what
+    # renders it — distinct files across the members; absent otherwise.
+    assert "consumers" not in out["derive_groups"][0]
+    for i, r in enumerate(rows):      # unsaved rows have no id; give them one
+        r.id = i + 1
+    cpaths = {rows[0].id: ["v/0.vue", "v/Z.vue"], rows[1].id: ["v/1.vue"], rows[2].id: ["v/0.vue"]}
+    with_c = proposal_summary(rows, consumer_paths=cpaths)
+    badge = next(g for g in with_c["derive_groups"] if g["group"] == "name:css:status-badge")
+    assert badge["consumers"] == {"count": 3, "paths": ["v/0.vue", "v/1.vue", "v/Z.vue"]}
+    dup = next(g for g in with_c["derive_groups"] if g["group"] == "dup:abc")
+    assert dup["consumers"] == {"count": 0, "paths": []}
     assert [g["group"] for g in out["derive_groups"]] == ["dup:abc", "dup:def", "name:css:status-badge"]
     assert out["derive_groups"][0]["files"] == 3 and out["derive_groups"][0]["size"] == 3
     assert out["derive_groups"][0]["label"] == "closed-msg (identical body)"
@@ -433,6 +444,44 @@ def test_compact_row_carries_identity_standing_and_the_proposers_word_only():
         assert noisy not in compact
 
 
+def test_resolve_consumers_prefers_the_own_file_and_fans_out_for_shared_names():
+    """Milestone 302: a class named in a template resolves to that file's
+    OWN row when it defines the class (a scoped rule, consumed by its own
+    markup); otherwise to every other definition of the name — one shared
+    sheet, or all of several (the map fans out rather than guessing)."""
+    from scribe.services.shape_ledger import resolve_consumers
+    css_rows = [
+        (1, "v/A.vue", "error-msg"),        # scoped, defined + used in A
+        (2, "v/B.vue", "error-msg"),        # scoped, defined in B, used in B and C
+        (3, "assets/components.css", "btn-primary"),   # the shared sheet
+        (4, "assets/a.css", "pill"), (5, "assets/b.css", "pill"),  # two shared defs
+        (6, "assets/c.css", "unused"),
+    ]
+    refs = {
+        "v/A.vue": {"error-msg": 2, "btn-primary": 1, "nothing-defined": 1},
+        "v/B.vue": {"error-msg": 1},
+        "v/C.vue": {"error-msg": 1, "pill": 3},
+    }
+    assert resolve_consumers(css_rows, refs) == {
+        (1, "v/A.vue"): 2,            # own row, not B's
+        (3, "v/A.vue"): 1,            # the shared sheet
+        (2, "v/B.vue"): 1,            # own row
+        (1, "v/C.vue"): 1, (2, "v/C.vue"): 1,   # C defines none → every other definition
+        (4, "v/C.vue"): 3, (5, "v/C.vue"): 3,   # ambiguous: both, not a guess
+    }
+    # Unknown tokens and an unreferenced row leave no trace.
+    assert all(sid != 6 for sid, _ in resolve_consumers(css_rows, refs))
+
+
+def test_consumer_edges_table_cascades_with_the_shape():
+    from scribe.models import Base
+    from scribe.models.code_shape import CONSUMER_BASES, CodeShapeConsumer
+    assert "code_shape_consumers" in Base.metadata.tables
+    cols = CodeShapeConsumer.__table__.c
+    assert next(iter(cols.shape_id.foreign_keys)).ondelete == "CASCADE"
+    assert CONSUMER_BASES == ("template",)
+
+
 def test_uses_edges_table_and_validation():
     """#2870: consumption is its own relation — a table that cascades with
     both ends, and `uses` on a classification must be a list of ids."""

tests/test_write_path_trigger.py

@@ -909,29 +909,35 @@ def _hook_runtime_env():
             "SCRIBE_URL": "http://127.0.0.1:9", "SCRIBE_TOKEN": "t"}
 
 
-def test_hook_nudges_recording_when_copies_exist_but_nothing_is_recorded(tmp_path):
-    """#2664: the local arm proves duplication; when Scribe has no record of it,
-    the same context block must ask for create_snippet — the one moment the
-    recording nudge is earned rather than noise. An unreachable server counts
-    as "nothing recorded": the local finding needed no server, and the nudge
-    fails open with it (here: a refused connection stands in for the instance)."""
-    env = _hook_runtime_env()
+def _dup_repo(tmp_path, env):
     repo = tmp_path / "repo"
     repo.mkdir()
     subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
     (repo / "a.py").write_text("def debounce(fn):\n    return fn\n")
     # git grep searches the index, so the existing copy must be staged.
     subprocess.run(["git", "add", "."], cwd=repo, check=True, env=env)
-    out = subprocess.run(
-        ["bash", str(HOOK)],
-        input=json.dumps({
-            "session_id": "s-nudge", "cwd": str(repo), "tool_name": "Write",
+    return repo
+
+
+def _write_event(repo, session="s-nudge"):
+    return json.dumps({
+        "session_id": session, "cwd": str(repo), "tool_name": "Write",
         "tool_input": {"file_path": str(repo / "b.py"),
                        "content": "def debounce(fn):\n    return fn\n"},
-        }),
-        capture_output=True, text=True, env=env,
-    )
+    })
+
+
+def test_hook_nudges_recording_when_copies_exist_but_nothing_is_recorded(tmp_path):
+    """#2664: the local arm proves duplication; when Scribe ANSWERS that it has
+    no record of it, the same context block must ask for create_snippet — the
+    one moment the recording nudge is earned rather than noise."""
+    with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, seen):
+        env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}")
+        repo = _dup_repo(tmp_path, env)
+        out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo),
+                             capture_output=True, text=True, env=env)
     assert out.returncode == 0
+    assert seen and seen[0]["path"] == ["b.py"]
     assert out.stdout.strip(), (
         "hook produced no output — the local arm should have found the "
         "staged duplicate and nudged"
@@ -939,6 +945,51 @@ def test_hook_nudges_recording_when_copies_exist_but_nothing_is_recorded(tmp_pat
     ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
     assert "already defined" in ctx          # the duplication finding
     assert "create_snippet" in ctx           # the recording ask riding it
+    assert "did not answer" not in ctx
+
+
+def test_hook_says_when_scribe_did_not_answer_once_per_outage(tmp_path):
+    """#2932: a configured instance that does not answer (refused connection)
+    is SAID — the write went unchecked — instead of the hook failing open in
+    silence; the record nudge's "nothing recorded" claim is withheld. Once per
+    outage: a second miss is quiet, an answer clears the marker, and the next
+    miss speaks again. The marker is shared with the after-write hook."""
+    env = _hook_runtime_env()                 # SCRIBE_URL → a refused port
+    repo = _dup_repo(tmp_path, env)
+    marker = tmp_path / "scribe-priorart" / "s-out.unreached"
+    env["TMPDIR"] = str(tmp_path)
+    out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
+                         capture_output=True, text=True, env=env)
+    assert out.returncode == 0
+    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
+    assert "already defined" in ctx
+    assert "Scribe did not answer the prior-art check for `b.py` within 5s" in ctx
+    assert "UNCHECKED" in ctx and "list_shapes" in ctx
+    assert "None of those existing copies is recorded" not in ctx
+    assert marker.is_file()
+    # Second miss inside the quiet window: local arm only.
+    out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
+                         capture_output=True, text=True, env=env)
+    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
+    assert "already defined" in ctx and "did not answer" not in ctx
+    # An answer clears the marker …
+    with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, _seen):
+        up = dict(env, SCRIBE_URL=f"http://127.0.0.1:{port}")
+        subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
+                       capture_output=True, text=True, env=up)
+    assert not marker.exists()
+    # … so the next outage is announced afresh.
+    out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
+                         capture_output=True, text=True, env=env)
+    assert "did not answer" in json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
+    # A write the hook had nothing local to say about still carries the line
+    # (the line is the whole message then): a fresh session, no duplicate.
+    (repo / "a.py").unlink()
+    subprocess.run(["git", "add", "-A"], cwd=repo, check=True, env=env)
+    out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out-2"),
+                         capture_output=True, text=True, env=env)
+    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
+    assert ctx.startswith("> Scribe did not answer")
 
 
 def test_hook_stays_quiet_about_recording_when_nothing_is_duplicated(tmp_path):
@@ -982,8 +1033,10 @@ def test_local_arm_finds_duplicates_in_every_language_family(
     every Go/Kotlin/Rust project, which is exactly where the operator observed
     recording never happening. Each case stages an existing copy and writes the
     same definition to a second file; the hook must prove the duplication and
-    ask for the record."""
-    env = _hook_runtime_env()
+    ask for the record (the instance ANSWERS "nothing recorded" — since #2932
+    an unanswered call withholds the nudge, so a sink stands in for it)."""
+    with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, _seen):
+        env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}")
         repo = tmp_path / "repo"
         repo.mkdir()
         subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
@@ -1253,7 +1306,8 @@ async def test_the_write_time_derive_check_names_a_family_or_a_canon_in_band():
         {"symbol": "log-empty", "kind": "css", "key": "name:css:log-empty",
          "family": {"group": "name:css:log-empty", "label": ".log-empty", "identical": False,
                     "files": ["a/TaskLogSection.vue", "a/WorkspaceTaskPanel.vue"],
-                    "file_count": 5, "size": 6}},
+                    "file_count": 5, "size": 6,
+                    "consumers": {"count": 6, "paths": ["a/TaskLogSection.vue", "a/V.vue"]}}},
         {"symbol": "btn-primary", "kind": "css", "key": "canon:2855",
          "canon": {"snippet_id": 2855, "path": "frontend/src/assets/components.css",
                    "label": ".btn-primary"}},
@@ -1291,8 +1345,10 @@ async def test_the_write_time_derive_check_names_a_family_or_a_canon_in_band():
     assert "Shape ledger at `frontend/src/components/New.vue`" in ctx
     # A CSS family is a repeated NAME (note 2917) and its dismissal is scoped-css;
     # a code dup family is an identical body and dismisses as convention-plumbing.
+    # A css family says what renders it (milestone 302) before the ask.
     assert "`.log-empty` is a repeated name with no canon — defined in 5 other file(s): " \
-           "`a/TaskLogSection.vue`, `a/WorkspaceTaskPanel.vue` +3 more; derive it now" in ctx
+           "`a/TaskLogSection.vue`, `a/WorkspaceTaskPanel.vue` +3 more; used by 6 templates: " \
+           "`a/TaskLogSection.vue`, `a/V.vue` +4 more; derive it now" in ctx
     assert "`slugify` is a duplicate family with no canon — identical body in 2 other file(s): " \
            "`a/x.py`, `a/y.py`; derive it now" in ctx
     assert "`.btn-primary` is canon — snippet #2855 at `frontend/src/assets/components.css`" in ctx