wip(394): steps 6+7 — backend path and instruction surfaces

2026-09-11 15:15:33 -04:00
parent 690ca0306e
commit c149ef31a3
28 changed files with 260 additions and 738 deletions
@@ -89,7 +89,7 @@ table here. The tools are grouped by family:
 | Projects / Milestones | `enter_project`, `get_project`, `create_milestone`, … | Containers and outcomes |
 | Search / Recall | `search`, `get_recent`, `list_tags`, `retrieval_telemetry` | Semantic + structured recall, and the readout its thresholds are tuned from |
 | Systems | `create_system`, `list_systems`, `list_system_records` | Reusable per-project subsystems/areas |
-| Rulebooks | `list_always_on_rules`, `list_rules`, `create_rule`, `create_project_rule`, `subscribe_project_to_rulebook`, … | Engineering/workflow rules |
+| Rulebooks | `list_rules`, `create_rule`, `create_project_rule`, `subscribe_project_to_rulebook`, … | Engineering/workflow rules |
 | Processes | `list_processes`, `get_process`, `create_process` | Saved prompts/workflows |
 | Trash | `list_trash`, `restore`, `purge_trash` | Recoverable deletes |
 | Admin | `get_app_logs` (write/admin key) | Diagnostics |
@@ -77,7 +77,7 @@ endpoint at `/mcp`, not these REST routes.
 |--------|------|-------------|
 | GET / POST | `/api/projects` | List (owned + shared) / create |
 | GET / PATCH / DELETE | `/api/projects/:id` | Read (with `milestone_summary`, `inception`) / update / delete |
-| POST | `/api/projects/:id/inception` | Record what the project inherits `{choices: {exclude_always_on_rulebooks, subscribe_rulebooks, design_system_id, seed_systems}}` (owner-only; `POST /api/projects` accepts the same under `inception`) |
+| POST | `/api/projects/:id/inception` | Record what the project inherits `{choices: {subscribe_rulebooks, design_system_id, seed_systems}}` (owner-only; `POST /api/projects` accepts the same under `inception`) |
 | GET | `/api/projects/:id/inception/defaults` | What binds if nobody decides — the inception card's payload |
 | GET | `/api/projects/:id/notes` | Notes + tasks in this project |
 | GET / POST | `/api/projects/:id/milestones` | List / create milestones |
@@ -120,7 +120,6 @@ endpoint at `/mcp`, not these REST routes.
 | POST | `/api/projects/:id/rules` | Create a project-scoped rule |
 | POST / DELETE | `/api/projects/:id/suppressions/rules/:rid` | Suppress / unsuppress a rule |
 | POST / DELETE | `/api/projects/:id/suppressions/topics/:tid` | Suppress / unsuppress a topic |
 | POST / DELETE | `/api/projects/:id/exclusions/rulebooks/:rid` | Exclude / include an always-on rulebook for this project (inception) |
 ## Sharing
@@ -206,6 +205,6 @@ endpoint at `/mcp`, not these REST routes.
 Claude clients connect to the built-in MCP server at `POST /mcp` (streamable HTTP,
 Bearer auth with an `fmcp_` key), served by `src/scribe/mcp/`. It is not a REST
 surface — it exposes the same data as typed tools (`create_note`, `create_task`,
-`start_planning`, `search`, `enter_project`, `list_always_on_rules`, …) with
+`start_planning`, `search`, `enter_project`, …) with
 server-level usage guidance delivered in the MCP `instructions` block. See
 [API Keys & MCP](api-keys-and-mcp.md).
@@ -60,8 +60,10 @@ Scribe stores the operator's engineering and workflow **rules** so Claude follow
 across sessions.
 - **Rulebooks → topics → rules** — Rules are grouped by topic inside a rulebook.
- **Always-on rules** — A rulebook can be flagged always-on; its rules load at the
+- **Rules arrive by retrieval** — Nothing is preloaded. A rule reaches a
-  start of every session through the plugin's push channel.
+  session when what the agent is about to do matches its trigger: a command,
  a file being written, or the operator's own message. `when_to_apply` is
  therefore the field that decides whether a rule is ever seen.
 - **Per-project scope** — A project subscribes to rulebooks, and can add
  project-scoped rules or suppress individual inherited rules/topics.
@@ -94,7 +96,7 @@ The whole store is reachable by Claude through a built-in **MCP endpoint at `/mc
 (Bearer-auth with an API key). The **Scribe Claude Code plugin** (shipped in this
 repo) wires it up:
- a `SessionStart` hook that injects the operator's always-on rules + active-project
+- a `SessionStart` hook that injects active-project
  context so Scribe surfaces without being asked (fail-open if Scribe is unreachable);
 - universal process-skills — writing-plans, systematic-debugging, verification,
  brainstorming — that route their output into Scribe;
@@ -1,6 +1,6 @@
 {
  "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.",
+  "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your 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": "2026.09.11.1154",
  "author": {
    "name": "Bryan Van Deusen"
@@ -5,7 +5,7 @@ instance into a first-class Claude Code extension:
 - **MCP tools** over your notes, tasks, projects, milestones, systems, and
  rulebook (the `scribe` server).
- **Session-start push channel** — a `SessionStart` hook injects your always-on
+- **Session-start push channel** — a `SessionStart` hook injects your
  rules + active-project context so Scribe surfaces *without being asked*.
 - **Prior-art recall on writes** — a `PreToolUse` hook on Write/Edit checks the
  file about to be written against your recorded snippets (what's kept at that
@@ -175,16 +175,11 @@ while IFS= read -r rel_path; do
      derive_seen=$(tr '\n' ',' < "$derivefile" 2>/dev/null | sed 's/,$//' | jq -sRr '@uri' 2>/dev/null) || derive_seen=""
      [ -n "$derive_seen" ] && derive_exclude_q="&exclude_derive=${derive_seen}"
    fi
-    # The rules marker the SessionStart hook stored, handed back so the server
+    # The rules marker is gone with the resident set it aged (milestone 394).
-    # can say whether those rules moved since (milestone 323). Nothing stored
+    # A session no longer holds a fixed set of rules from turn zero, so there
-    # means nothing sent, which the server reads as silence rather than as a
+    # is nothing that can have drifted since it loaded them — each rule is
-    # mismatch — an install that never reached /api/plugin/context must not
+    # retrieved at the moment it applies.
    # start claiming its rules changed.
    etag_q=""
    if [ -f "$state_dir/${safe_sid}.rules_etag" ]; then
      held=$(jq -sRr '@uri' < "$state_dir/${safe_sid}.rules_etag" 2>/dev/null) || held=""
      [ -n "$held" ] && etag_q="&rules_etag=${held}"
    fi
    if [ -n "$path_enc" ]; then
      # 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
@@ -8,7 +8,7 @@
 # does not depend on the key or the network.
 #
 # Tier 2 (DYNAMIC, best-effort enrichment): curls the operator's Scribe instance
-# for always-on rules + active-project context and appends it. Config comes from
+# for active-project context and appends it. Config comes from
 # the plugin's userConfig, exported to hooks as:
 #   CLAUDE_PLUGIN_OPTION_API_ENDPOINT  base URL, no trailing slash
 #   CLAUDE_PLUGIN_OPTION_API_TOKEN     fmcp_ API key (sensitive)
@@ -158,7 +158,7 @@ if [ -n "$url" ] && [ -n "$token" ] && command -v curl >/dev/null 2>&1; then
  [ -n "$body" ] && dyn=$(printf '%s' "$body" | jq -r '.context // empty' 2>/dev/null)
  # Stash the rules marker for the write-path hook (milestone 323). THIS is
  # where it has to be captured: the model receives one from
-  # list_always_on_rules too, but a hook cannot see an MCP tool's result. Stored
+  # an MCP tool too, but a hook cannot see a tool's result. Stored
  # under the same state dir the prior-art hook already uses, keyed by session,
  # so "changed since" means since THIS session loaded its rules.
  #
@@ -178,9 +178,9 @@ if [ -n "$url" ] && [ -n "$token" ] && command -v curl >/dev/null 2>&1; then
        && printf '%s' "$etag" > "$etag_dir/${safe_sid}.rules_etag" 2>/dev/null || true
    fi
  fi
-  [ -z "$dyn" ] && status="> ⚠️ Scribe: live rules/project context could not be loaded this session (instance unreachable or request failed). The standing guidance above still applies — pull rules with \`list_always_on_rules()\` and project context with \`enter_project()\` as needed."
+  [ -z "$dyn" ] && status="> ⚠️ Scribe: live project context could not be loaded this session (instance unreachable or request failed). The standing guidance above still applies — ask for rules with \`search(content_type=\"rule\")\` and project context with \`enter_project()\` as needed."
 elif [ -n "$url" ] && [ -z "$token" ]; then
-  status="> ⚠️ Scribe: live context disabled this session — the API key is not configured (Scribe base URL is). Set it with \`/plugin\` → Scribe → configure, or export SCRIBE_TOKEN. Tools still work; pull rules with \`list_always_on_rules()\` and project context with \`enter_project()\`."
+  status="> ⚠️ Scribe: live context disabled this session — the API key is not configured (Scribe base URL is). Set it with \`/plugin\` → Scribe → configure, or export SCRIBE_TOKEN. Tools still work; ask for rules with \`search(content_type=\"rule\")\` and project context with \`enter_project()\`."
 elif [ -z "$url" ] && [ -z "$token" ]; then
  # NEITHER value arrived. Previously this case stayed silent as "an unconfigured
  # install", which made issue #2198 invisible for weeks: a *casing* bug here
@@ -189,7 +189,7 @@ elif [ -z "$url" ] && [ -z "$token" ]; then
  # silently disabled auto-inject and the write-path trigger too. It is not a
  # benign state — the plugin prompts for both values at enable time, so if
  # neither reached the hook, something is wrong. Say so.
-  status="> ⚠️ Scribe: live context disabled this session — neither the Scribe base URL nor the API key reached this hook. Configure the plugin (\`/plugin\` → Scribe), or export SCRIBE_URL + SCRIBE_TOKEN. Note this also disables prompt auto-inject and the write-path prior-art trigger. Tools still work; pull rules with \`list_always_on_rules()\` and project context with \`enter_project()\`."
+  status="> ⚠️ Scribe: live context disabled this session — neither the Scribe base URL nor the API key reached this hook. Configure the plugin (\`/plugin\` → Scribe), or export SCRIBE_URL + SCRIBE_TOKEN. Note this also disables prompt auto-inject and the write-path prior-art trigger. Tools still work; ask for rules with \`search(content_type=\"rule\")\` and project context with \`enter_project()\`."
 fi
 [ -n "$dyn" ]    && append "$dyn"
@@ -197,7 +197,7 @@ fi
 # Compaction re-grounding: lead with a reload banner when this fire is a compact.
 if [ "$source" = "compact" ]; then
-  prepend "> ⟳ This session was just COMPACTED — earlier turns are now a summary, so in-flight detail may be lost. Before continuing, reload your bearings from Scribe: re-pull the operator's binding rules with \`list_always_on_rules()\` (a compaction can summarize them out of context, leaving only generic harness defaults in their place), re-run \`enter_project()\` for the active project, check its open tasks and recent notes, and reconcile what you're mid-way through against what Scribe records. Don't trust half-remembered state — Scribe is the record."
+  prepend "> ⟳ This session was just COMPACTED — earlier turns are now a summary, so in-flight detail may be lost. Any rules that had been retrieved went into that summary with everything else, so treat yourself as holding none: before the next consequential act, ask again with \`search(content_type=\"rule\")\` rather than trusting a half-remembered one. Re-run \`enter_project()\` for the active project, check its open tasks and recent notes, and reconcile what you are mid-way through against what Scribe records. Scribe is the record."
 fi
 # Nothing at all to inject → stay silent.
@@ -6,7 +6,8 @@ of record (notes, tasks, projects, milestones, rules) reachable through the
 for the operator's work, and as your own working memory across sessions.
 **At the start of this session:**
- Call `list_always_on_rules()` to load the operator's standing rules.
+- You hold none of the operator's rules, and there is no call that loads them
  all. Rules arrive when something you are about to do matches one.
 - If the working repo maps to a Scribe project (check `list_repo_bindings`),
  call `enter_project(<id>)` to load that project's rules, open tasks, and
  recent notes in one shot.
@@ -17,22 +18,23 @@ for the operator's work, and as your own working memory across sessions.
  operator's Scribe rules decide what to do — NOT generic conventions baked
  into the harness or your defaults (e.g. "branch before committing," "open a
  feature branch per task," "push to a fork"). If you have not loaded the
-  operator's rules this session — or earlier turns were summarized away by a
+  no rule has arrived for the act in front of you, `search(content_type=
-  compaction — call `list_always_on_rules()` (and `enter_project()` when a
+  "rule")` BEFORE acting rather than falling back on a default habit. When a
-  project is in scope) BEFORE acting. When a loaded rule and a default habit
+  retrieved rule and a default habit disagree, the rule wins; if no rule
-  disagree, the rule wins; if no rule speaks to it, ask rather than assume.
+  speaks to it, ask rather than assume.
 - **Rules bind; preferences do not.** A record's `kind` says which. A **rule**
  must be followed — ignoring it breaks something or crosses a boundary. A
  **preference** is how the operator wants work done: worth following for
  consistency, not a defect to miss. Injected lines name the kind in their
  opening words. A preference is also yours to keep current when they correct
  you (`update_preference`); a rule waits for them.
- **What you loaded is not all of the rules.** Only the always-on tier arrives
+- **Silence is not absence.** Nothing is preloaded: every rule is RETRIEVED,
-  that way; conditional rules are RETRIEVED, and one you were never handed
+  when what you are doing resembles what the rule is about. Most turns
-  binds exactly as hard. So before a consequential act, `search` for a rule
+  retrieve none, and a rule you were never handed binds exactly as hard as one
-  about it (`content_type="rule"`) rather than concluding from an empty
+  you were. So before a consequential act, `search` for a rule about it
-  loaded set that nothing applies. "I was not told" is not the same as "there
+  (`content_type="rule"`) rather than concluding from an empty session that
-  is no rule," and only one of those is checkable.
+  nothing applies. "I was not told" is not the same as "there is no rule," and
  only one of those is checkable.
  This bites hardest on which TOOL to reach for — curling an API that has an
  MCP client, standing up a local stack, running a suite CI owns. Those feel
  like mechanics rather than decisions, so they raise no doubt and generate no
@@ -5,7 +5,7 @@
 # asks "what is recorded about the file being written". This one asks "does a
 # standing rule speak to the command about to be run" — the question nothing
 # could ask before, and the reason every rule about which tool to reach for had
-# to live in the always-on preload instead.
+# to live in the preload instead, back when there was one.
 #
 # WHY A HOOK AND NOT AN INSTRUCTION. A reflex generates no query (note #3089):
 # you reach for `curl` confidently, with no moment of doubt, so a surface that
@@ -1,6 +1,6 @@
 ---
 name: using-scribe
-description: Use at the START of every session, and before answering anything about the operator's work or starting any task — establishes the Scribe-first reflex. FIRST ACTION of a session: call list_always_on_rules() (and enter_project when a repo/project is in scope) to load the operator's binding rules. Then recall before acting, update over duplicate, plan in Scribe not in files.
+description: Use at the START of every session, and before answering anything about the operator's work or starting any task — establishes the Scribe-first reflex. You hold none of the operator's rules: they arrive by retrieval when your work matches one, and search(content_type="rule") is how you ask before a consequential act. Call enter_project when a repo/project is in scope. Then recall before acting, update over duplicate, plan in Scribe not in files.
 ---
 # Using Scribe
@@ -13,12 +13,19 @@ asked for.
 ## Do this first (every session)
-**Pull the standing rules yourself — do not wait for them to be handed to you.**
+**You are not holding the operator's rules, and no call loads them all.**
-At the start of a session, before substantive work, call
+There is no standing set to pull. A rule reaches you when what you are about to
-`list_always_on_rules()` to load the operator's always-on rules. If the working
+do matches it — a command, code you are writing, or what the operator just
-repo maps to a Scribe project (you're in a known repo, or `list_repo_bindings`
+asked for — and on most turns none will. That is the surface working.
-shows a binding), call `enter_project(id)` instead/as-well — it returns the
+
-project plus its applicable rules, open tasks, and recent notes in one shot.
+**So the reflex is to ASK, not to load.** Before a consequential act — anything
 hard to reverse or outward-facing — `search(content_type="rule")` for the thing
 you are about to do. An empty session is not evidence of an empty rulebook.
 If the working repo maps to a Scribe project (you're in a known repo, or
 `list_repo_bindings` shows a binding), call `enter_project(id)` — it returns the
 project plus the rules bound to the areas it works in, open tasks, and recent
 notes in one shot.
 Do this actively. A SessionStart hook *may* also inject a rule index, but treat
 that as a bonus, not a precondition: it can be absent (e.g. when the instance is
@@ -56,11 +63,12 @@ Two constraints on *how* that's achieved:
   re-deriving it or opening a duplicate. When a project is in scope, pass its
   `project_id` so results stay scoped.
-2. **Standing rules are binding — and the ones you were handed are not all of
+2. **Rules are binding, and silence does not mean there are none.** Nothing
-   them.** Load the resident set via `list_always_on_rules()` at session start
+   is preloaded, so "no rule arrived" means "nothing matched" — never "no rule
-   (see "Do this first"). Pull a record's full statement with `get_rule(id)`
+   exists". Ask with `search(content_type="rule")` before a consequential act,
-   when it's about to bite. When a project is in scope, `enter_project(id)`
+   and pull a record's full statement with `get_rule(id)` when it is about to
-   also returns its applicable rules.
+   bite. When a project is in scope, `enter_project(id)` also returns the rules
   bound to its areas.
   **`kind` says how much force a record carries, and it is never something to
   infer.** A **rule** must be followed: ignoring it breaks something or
@@ -222,13 +230,12 @@ bound — confine the session to it:
 ## Starting a project: decide what it inherits
 A project's inheritance is a **decision, not a default**. Before
-`create_project`, ask the operator the four inception questions and pass the
+`create_project`, ask the operator the three inception questions and pass the
 answers — never create a project bare by default:
- which **always-on rulebooks** it should NOT inherit (`list_rulebooks` shows
+- which rulebooks to **subscribe** (`list_rulebooks` shows them; default: none
-  which are always_on; default: inherit them all) →
+  — a rulebook binds a project only when it opts in) →
-  `exclude_always_on_rulebooks=[...]`
+  `subscribe_rulebooks=[...]`
 - which other rulebooks to **subscribe** → `subscribe_rulebooks=[...]`
 - which **design system** its UI is built from (`list_design_systems`; or
  none) → `design_system_id=<id | -1>`
 - whether to **seed the standard starter Systems** so records can be tagged
@@ -246,19 +253,23 @@ inception is the moment they are decided together, and the record of why.
 When codifying a rule, pick its home by **who it should bind** — and keep
 shared homes general:
- **Always-on rulebook** (`create_rule` in an `always_on` rulebook) — universal
+- **Rulebook** (`create_rule` + `subscribe_project_to_rulebook`) — a reusable,
-  norms that bind *every* project. Cross-project standards only.
+  *themed* module of general rules that binds the projects which opt in (e.g. a
- **Subscribed rulebook** (`create_rule` + `subscribe_project_to_rulebook`) — a
+  review checklist → every service). Themed, but project-agnostic.
  reusable, *themed* module of general rules that binds only projects that opt
  in (e.g. a review checklist → every service). Themed, but project-agnostic.
 - **Project rule** (`create_project_rule`) — anything specific to one project
  (its files, paths, quirks).
-Both rulebook tiers are shared, so their rules stay general; they differ in
+There used to be a third home — an `always_on` rulebook that bound every
-**reach** (all vs opt-in), not generality. Names one project's specifics →
+project automatically. It is gone: subscription is the only reach a rulebook
-project rule; a standard a category shares → subscribed rulebook; a universal
+has. Names one project's specifics → project rule; anything a category of
-norm → always-on rulebook. Never put project-specific detail in a shared
+projects shares → rulebook. Never put project-specific detail in a rulebook —
-rulebook — it leaks to every other project that gets it.
+it leaks to every other project that subscribes.
 **Whichever home it gets, a rule needs `when_to_apply`.** It is the only thing
 that decides whether the rule is ever seen: nothing is preloaded, so a rule
 with no trigger is not a quiet rule, it is an unreachable one. Write the moment
 in the words a session actually produces — the command, the error, the
 half-formed ask — not the category it belongs to.
 **First ask whether it's a rule at all.** A rule is prose you have to remember
 and apply; Scribe's other entities are structure a tool can resolve and check.
@@ -38,8 +38,8 @@ from quart import Quart
 # them) was DECLINED a line, deliberately, by the operator — not overlooked.
 # The reasoning, so it is not re-litigated blind: this is a map, and its own
 # closing line says each tool's description carries the full contract. The
-# sweep is a curation act, not a session-start reflex like enter_project or
+# sweep is a curation act, not a session-start reflex like enter_project.
-# list_always_on_rules. Spending the last of the budget on it would leave the
+# Spending the last of the budget on it would leave the
 # map unable to grow for something more central later.
 #
 # The accepted cost: an agent that never opens create_note's docstring never
@@ -59,15 +59,15 @@ from quart import Quart
 #   - What it bought is not per-tool guidance and has nowhere else to live at
 #     session-start altitude. Rules were retrievable only by RESIDENCY: the
 #     always-on preload put them in front of the agent, and nothing told a
-#     session to go looking for one it had not been handed. The tier split is
+#     session to go looking for one it had not been handed. That preload is
-#     therefore load-bearing on ANY install (rule 115): a delivered rule costs
+#     gone (milestone 394), which makes this line LOAD-BEARING rather than
-#     tokens in every session forever, so a rulebook that only delivers cannot
+#     supplementary: retrieval is now the only delivery, and retrieval fires
-#     grow past what one session can hold, and every rule worth keeping has to
+#     only if something asks. A session that waits to be handed a rule is
-#     become resident to bind at all. Retrieval is what lets it keep growing —
+#     handed nothing. A tool-choice reflex asks least of all (#3476, #161).
-#     and retrieval fires only if something asks, which nothing told a session
+#   - It also has to carry what absence MEANS. "No rule arrived" is now the
-#     to do. A tool-choice reflex asks least of all (#3476, #161).
+#     ordinary state rather than the exceptional one, and reading it as
-#   - This states the PULL for conditional rules, exactly as the surrounding
+#     "there is no rule" is the #3720 defect at session scale. Rule 119 makes
-#     line states it for always-on ones. Rule 119 makes these surfaces the
+#     these surfaces the
 #     specification, so the same sentence lands on all three session-start
 #     surfaces, and test_instruction_surfaces_agree pins it.
 _INSTRUCTIONS = """
@@ -77,8 +77,8 @@ in local files (CLAUDE.md, auto-memory); Scribe holds the single copy.
 Hierarchy: Project -> Milestone -> Task/Note. The map, by purpose:
 - ORIENT: enter_project(id) at session start — rules, open tasks, recent
-  notes, Systems, design system. `inception` key: ask what the project
+  notes, Systems, design system. `inception`: ask what the project
-  inherits, decide_project_inception (create_project takes the same).
+  inherits, then decide_project_inception.
 - DO: create_task. Fixed a problem? kind="issue" (symptom -> root cause ->
  fix), never a work-log line on an unrelated task. Log with add_task_log;
  keep status honest — in_progress on start, done on finish.
@@ -88,9 +88,9 @@ Hierarchy: Project -> Milestone -> Task/Note. The map, by purpose:
  active project_id to stay in scope.
 - WHERE work happens: Systems. Tag records with system_ids as you write;
  create_system when the area is unmodelled.
- HOW: rules bind; preferences guide. list_always_on_rules() at start;
+- HOW: rules bind; preferences guide. Nothing preloads — a rule arrives
-  before a consequential act, search(content_type="rule") — the resident
+  when your work matches it. Before a consequential act,
-  set is not all of them.
+  search(content_type="rule"); silence means nothing matched, not none.
 - UI: the project's design system is binding — resolve_design_system /
  get_design_system_stylesheet before hand-writing a value.
 - REUSE: search snippets before writing a helper; record what you build with
@@ -130,7 +130,7 @@ _READ_ONLY_TOOLS = frozenset({
    "get_task", "get_milestone", "get_recent", "enter_project",
    "list_milestones", "list_notes", "list_projects", "list_rulebooks",
    "list_rules", "list_tags", "list_tasks", "list_topics", "list_trash",
-    "list_always_on_rules", "search",
+    "search",
    "get_system", "list_systems", "list_system_records",
    # The global area catalog and its mapping REPORT — propose writes nothing;
    # map_system_to_canonical is the separate, explicitly-called write.
@@ -256,17 +256,16 @@ async def get_project(project_id: int) -> dict:
 def _inception_choices(
-    exclude_always_on_rulebooks, subscribe_rulebooks, design_system_id, seed_systems,
+    subscribe_rulebooks, design_system_id, seed_systems,
 ) -> dict | None:
    """The tool args → an inception choices object, or None when no inception
    arg was given at all (a bare create stays undecided and enter_project
    asks). design_system_id: 0 = not stated, -1 = explicitly none, n = that
    system."""
-    if (exclude_always_on_rulebooks is None and subscribe_rulebooks is None
+    if (subscribe_rulebooks is None
            and not design_system_id and seed_systems is None):
        return None
    return {
        "exclude_always_on_rulebooks": list(exclude_always_on_rulebooks or []),
        "subscribe_rulebooks": list(subscribe_rulebooks or []),
        "design_system_id": None if design_system_id in (0, -1) else design_system_id,
        "seed_systems": bool(seed_systems),
@@ -279,7 +278,6 @@ async def create_project(
    goal: str = "",
    status: str = "active",
    color: str = "",
    exclude_always_on_rulebooks: list[int] | None = None,
    subscribe_rulebooks: list[int] | None = None,
    design_system_id: int = 0,
    seed_systems: bool | None = None,
@@ -299,9 +297,10 @@ async def create_project(
        goal: The desired outcome or definition of done for the project.
        status: one of active (default), paused, completed, archived.
        color: Optional hex colour for the project card (e.g. "#6366f1").
-        exclude_always_on_rulebooks: always-on rulebook ids this project does
+        subscribe_rulebooks: rulebook ids this project opts into. Since
            milestone 394 subscription is the only way a rulebook binds a
            project, so there is no automatic tier left to decline. Was
            NOT inherit ([] = inherit them all). list_rulebooks shows which are
            always_on.
        subscribe_rulebooks: rulebook ids to subscribe (the non-always-on ones).
        design_system_id: the design system this project's UI is built from
            (list_design_systems); -1 = explicitly none; 0 = not stated.
@@ -319,7 +318,7 @@ async def create_project(
    )
    data = project.to_dict()
    choices = _inception_choices(
-        exclude_always_on_rulebooks, subscribe_rulebooks, design_system_id, seed_systems,
+        subscribe_rulebooks, design_system_id, seed_systems,
    )
    if choices is not None:
        decided = await inception_svc.decide(uid, project.id, choices=choices, via="mcp")
@@ -336,7 +335,6 @@ async def create_project(
 async def decide_project_inception(
    project_id: int,
    exclude_always_on_rulebooks: list[int] | None = None,
    subscribe_rulebooks: list[int] | None = None,
    design_system_id: int = 0,
    seed_systems: bool | None = None,
@@ -345,11 +343,11 @@ async def decide_project_inception(
    or re-decide later (milestone 297).
    Owner-only. Applies the effects through the ordinary tools' paths —
-    exclude_always_on_rulebook, subscribe_project_to_rulebook,
+    subscribe_project_to_rulebook,
    set_project_design_system, the standard Systems seed — and writes the
    decision on the project last, so get_project/enter_project can say why
    the project has the rules, design and Systems it has. Re-deciding is
-    additive for exclusions/subscriptions (use include_always_on_rulebook /
+    additive for subscriptions (use
    unsubscribe_project_from_rulebook to undo one), replaces the design
    system, and never re-seeds Systems a project already has.
@@ -359,7 +357,7 @@ async def decide_project_inception(
    """
    uid = current_user_id()
    choices = _inception_choices(
-        exclude_always_on_rulebooks, subscribe_rulebooks, design_system_id, seed_systems,
+        subscribe_rulebooks, design_system_id, seed_systems,
    ) or {}
    decided = await inception_svc.decide(uid, project_id, choices=choices, via="mcp")
    return {"project_id": project_id, **decided}
@@ -48,16 +48,13 @@ async def get_rulebook(rulebook_id: int) -> dict:
 async def create_rulebook(title: str, description: str = "") -> dict:
    """Create a new rulebook (a shared, reusable module of general rules).
-    Two ways a rulebook reaches projects, set by its always_on flag (toggle via
+    A rulebook reaches a project ONE way: the project subscribes to it
-    update_rulebook):
+    (subscribe_project_to_rulebook). There was a second until milestone 394 —
-      - always_on = true  -> binds EVERY one of your projects automatically.
+    an `always_on` flag that bound every project automatically — and it is
-        Use for universal cross-project norms that apply across every
+    gone with the tier it belonged to. Opt-in is now the whole model, so a
-        project, not just one.
+    rulebook binds what asked for it and nothing else.
-      - always_on = false -> binds only projects that subscribe
+
-        (subscribe_project_to_rulebook). Use for a THEMED body of rules a
+    A rulebook is SHARED, so its rules must stay general — agnostic
        category of projects shares (e.g. a design system that visual apps
        opt into).
    Either way a rulebook is SHARED, so its rules must stay general — agnostic
    to any single project. Project-specific rules go in create_project_rule.
    Args:
@@ -73,7 +70,6 @@ async def create_rulebook(title: str, description: str = "") -> dict:
 async def update_rulebook(
    rulebook_id: int, title: str = "", description: str = "",
    always_on: bool | None = None,
 ) -> dict:
    """Update an existing rulebook. Only non-empty fields are changed.
@@ -81,9 +77,6 @@ async def update_rulebook(
        rulebook_id: Rulebook to update.
        title: New title. Empty string leaves unchanged.
        description: New description. Empty string leaves unchanged.
        always_on: When True, rules in this rulebook are loaded at session
            start by list_always_on_rules regardless of project context.
            Pass None to leave unchanged.
    """
    uid = current_user_id()
    fields: dict = {}
@@ -91,8 +84,6 @@ async def update_rulebook(
        fields["title"] = title
    if description:
        fields["description"] = description
    if always_on is not None:
        fields["always_on"] = always_on
    rb = await rulebooks_svc.update_rulebook(rulebook_id, uid, **fields)
    if rb is None:
        raise ValueError(f"rulebook {rulebook_id} not found")
@@ -234,58 +225,6 @@ async def list_rules(
    return {"rules": [_rule_summary(r) for r in rows], "total": len(rows)}
 async def list_always_on_rules(project_id: int = 0) -> dict:
    """Return all rules from rulebooks flagged always_on for the current user.
    Call this at session start. Treat the returned rules as binding for the
    session — they apply regardless of which project (if any) is in scope.
    Returns the ALWAYS-ON tier only (milestone 307). A `conditional` rule is
    still binding when it applies; it just is not resident — it reaches a
    session through enter_project (when the project works in an area the rule
    is tagged to) or through search(content_type="rule"). Nothing here is a
    behaviour change until rules are actually re-tiered: `tier` defaults to
    always_on, so an existing rulebook returns exactly what it always did.
    Pair with get_project(id).applicable_rules when working on a specific
    project to also load that project's subscription-derived rules.
    A rule carrying `last_verified` asserts a FACT about something outside the
    operator's control — a runner's shell, a tool's existence, a setting
    somewhere. It is still binding; the field says how long ago anyone
    confirmed it, and "never" means nobody has. Follow the rule, and if you
    are already standing where the check could be made, make it: get_rule
    gives you its `verify_with`. Most rules have no such field, which means
    they are decisions and there is nothing to check.
    Args:
        project_id: 0 (default) = the user-wide set. Inside a project, pass
            its id: an always-on rulebook the project EXCLUDED at inception
            (see enter_project's `excluded_always_on`) is left out — the
            project decided not to inherit it.
    """
    uid = current_user_id()
    rules = await rulebooks_svc.list_always_on_rules(uid, project_id=project_id)
    # AMBIENT source: the resident set, handed over whole. No ranker chose
    # these, so they must not land in the pull-through numerator's denominator
    # — but they must land SOMEWHERE, or the largest rule surface in the
    # product stays the one surface its own scoreboard cannot see (#3473).
    record_rule_surfaced(
        user_id=uid,
        rule_ids=[r.id for r in rules],
        source="list_always_on_rules",
    )
    return {
        "rules": [_rule_summary(r) for r in rules],
        "total": len(rules),
        # A marker for the set you are now holding. It is not for you to read:
        # the write-path hook carries it back and is told if these rules have
        # moved since. Deliberately NOT on rules_payload's applicable_rules —
        # that is a DIFFERENT set (subscription-derived), and one key name
        # over two sets is how a comparison starts reporting phantom changes.
        "rules_etag": rulebooks_svc.rules_etag(rules),
    }
 async def get_rule(rule_id: int) -> dict:
    """Fetch a rule by id — full statement + why + how_to_apply.
@@ -309,7 +248,6 @@ async def get_rule(rule_id: int) -> dict:
 async def create_rule(
    topic_id: int, title: str, statement: str, when_to_apply: str = "",
    why: str = "", how_to_apply: str = "", order_index: int = 0,
    tier: str = "always_on", system_ids: list[int] | None = None,
    arose_from_id: int = 0, verify_with: str = "", expires_when: str = "",
    force: bool = False,
 ) -> dict:
@@ -357,7 +295,7 @@ async def create_rule(
      * "Approve it AS WRITTEN" — you create it with the statement exactly as
        shown. This is what makes element 1 load-bearing: they approved TEXT,
        so that text is what gets stored, verbatim.
-      * "LET'S TALK ABOUT IT" — the wording, the scope, the tier, whether it
+      * "LET'S TALK ABOUT IT" — the wording, the scope, whether it
        wants to be a rule at all. Most good rules arrive this way, so treat
        this answer as the expected one rather than a setback.
      * "NO" — let it go. If the observation is still worth keeping, it is a
@@ -371,7 +309,7 @@ async def create_rule(
    into existence, which is the thing this whole loop exists to prevent.
    A rulebook rule is shared by every project that gets the rulebook: an
-    always_on rulebook binds ALL your projects; a subscribed rulebook binds the
+    A subscribed rulebook binds the
    projects that opt in. So a rulebook rule must read as a general standard —
    never pin it to one project's files, paths, or quirks. For a rule that
    applies to a single project only, use create_project_rule instead (no
@@ -408,7 +346,7 @@ async def create_rule(
            instruction. State the moment or the material: "before any git
            push", "when adding a value to a CHECK-gated column", "when a
            release is being cut". Write it even though the parameter is
-            optional: it decides the tier below, it is how the rule is found
+            optional: it is how the rule is found
            when it matters, and a rule nobody can place is a rule nobody
            applies.
            This field is also the rule's RETRIEVAL SURFACE — it and the
@@ -428,12 +366,6 @@ async def create_rule(
            category — it produces the command, the error, the half-formed
            ask — so a trigger written that way leaves the embedded
            document to be carried by the title alone.
        tier: "always_on" (default) or "conditional".
            The test: can you name the trigger WITHOUT naming a system, an
            artifact type or a moment? If the honest answer is "whenever you
            are working", it is always_on. If you had to name something, it is
            conditional — and conditional costs nothing when it is irrelevant,
            which is what lets it be as long as it needs to be.
        system_ids: Ids from list_canonical_systems — the global AREAS this
            rule is about. This is what lets a rule reach a project that is
            working in that area, so a CI rule surfaces on a CI change.
@@ -472,7 +404,7 @@ async def create_rule(
    rule = await rulebooks_svc.create_rule(
        topic_id=topic_id, user_id=uid,
        title=title, statement=statement, when_to_apply=when_to_apply,
-        tier=tier, arose_from_id=arose_from_id,
+        arose_from_id=arose_from_id,
        why=why, how_to_apply=how_to_apply, order_index=order_index,
        verify_with=verify_with, expires_when=expires_when,
    )
@@ -482,7 +414,6 @@ async def create_rule(
 async def create_project_rule(
    project_id: int, statement: str, title: str = "", when_to_apply: str = "",
    why: str = "", how_to_apply: str = "", order_index: int = 0,
    tier: str = "always_on", system_ids: list[int] | None = None,
    arose_from_id: int = 0, verify_with: str = "", expires_when: str = "",
    force: bool = False,
 ) -> dict:
@@ -531,25 +462,7 @@ async def create_project_rule(
              RETRIEVES: "the CI job passed locally and fails on the runner
                with a permission error"
              COLLAPSES: "when touching CI config"
-            See create_rule for the full argument. It informs the
+            See create_rule for the full argument.
            tier below rather than deciding it,
            since a project rule's tier turns on area-scope, not on whether
            the trigger can be named.
        tier: "always_on" (default) or "conditional". The SAME two values as
            create_rule, judged against a different cost — do not import that
            tool's test wholesale. There, always_on means every session in
            every project, so the bar is high: the trigger must be nameless
            ("whenever you are working"). Here the rule is already scoped to
            one project by construction, so always_on costs only that
            project's sessions and the bar is correspondingly lower. A
            project rule that names something specific is still ordinarily
            always_on — being specific is what project rules are FOR.
            Reach for conditional when the rule is about one AREA of a large
            project — a CI quirk, a migration gotcha, one subsystem's
            convention — so it arrives with that area instead of resident in
            every session. The failure to avoid is local: forty always-on
            rules on one project reproduces, inside that project, exactly the
            preload bloat that made every rule compete for the same budget.
        system_ids: Ids from list_canonical_systems — the global AREAS this
            rule is about. Worth setting even on a project rule: it is what
            lets a conditional one surface when the project is working in
@@ -583,7 +496,7 @@ async def create_project_rule(
    rule = await rulebooks_svc.create_project_rule(
        project_id=project_id, user_id=uid,
        title=derived_title, statement=statement, when_to_apply=when_to_apply,
-        tier=tier, arose_from_id=arose_from_id,
+        arose_from_id=arose_from_id,
        why=why, how_to_apply=how_to_apply, order_index=order_index,
        verify_with=verify_with, expires_when=expires_when,
    )
@@ -593,7 +506,7 @@ async def create_project_rule(
 async def update_rule(
    rule_id: int, title: str = "", statement: str = "", when_to_apply: str = "",
    why: str = "", how_to_apply: str = "", order_index: int = -1,
-    tier: str = "", system_ids: list[int] | None = None, arose_from_id: int = 0,
+    system_ids: list[int] | None = None, arose_from_id: int = 0,
    verify_with: str = "", expires_when: str = "", kind: str = "",
    clear_fields: list[str] | None = None,
 ) -> dict:
@@ -606,9 +519,10 @@ async def update_rule(
    correct. Ordinary edits to an existing preference belong in
    update_preference, which asks for what taught the change.
-    Adding `when_to_apply` and a `tier` to an existing rule is the ordinary way
+    `when_to_apply` IS HOW A RULE ARRIVES AT ALL. Nothing is preloaded since
-    a rule stops being preloaded into every session and starts arriving when it
+    milestone 394, so a rule with no trigger is not a quiet rule — it is one
-    is relevant. `system_ids` REPLACES the rule's areas (pass [] to clear).
+    no session will ever be shown. `system_ids` REPLACES the rule's areas
    (pass [] to clear), and they decide which PROJECTS a rule binds by area.
    RETROFITTING A TRIGGER HAS ITS OWN TRAP, and it is not the one create_rule
    warns about. There the field is empty and the instruction is "write one".
@@ -662,8 +576,6 @@ async def update_rule(
        fields["statement"] = statement
    if when_to_apply:
        fields["when_to_apply"] = when_to_apply
    if tier:
        fields["tier"] = tier
    if kind:
        fields["kind"] = kind
    if arose_from_id:
@@ -978,7 +890,7 @@ async def subscribe_project_to_rulebook(
 ) -> dict:
    """Subscribe a project to a rulebook — its rules then bind that project.
-    Subscription is the opt-in path for a non-always_on rulebook: a reusable,
+    Subscription is the ONLY path for a rulebook (milestone 394): a reusable,
    themed module of GENERAL rules shared across the projects that subscribe.
    Subscribe a project because it fits the rulebook's theme (e.g. a visual app
    -> the design-system rulebook), not to host rules about this one project —
@@ -1004,34 +916,6 @@ async def unsubscribe_project_from_rulebook(
 # ── Suppressions — project-level mute of rulebook rules / topics ────────
 async def exclude_always_on_rulebook(project_id: int, rulebook_id: int) -> dict:
    """Opt a project OUT of a whole always-on rulebook (milestone 297).
    Always-on rulebooks bind every project implicitly; an inception decision
    can say "not this one, not here". The exclusion is total for that project
    — list_always_on_rules(project_id), enter_project/get_project rules and
    the session-start context all leave it out and name it under
    `excluded_always_on`. Owner-only; the rulebook must be always_on (a
    subscribed rulebook is left with unsubscribe_project_from_rulebook).
    Idempotent; include_always_on_rulebook reverses it. Normally reached via
    decide_project_inception, not by hand.
    """
    uid = current_user_id()
    await rulebooks_svc.exclude_always_on_rulebook_for_project(
        project_id=project_id, rulebook_id=rulebook_id, user_id=uid,
    )
    return {"project_id": project_id, "rulebook_id": rulebook_id, "excluded": True}
 async def include_always_on_rulebook(project_id: int, rulebook_id: int) -> dict:
    """Reverse exclude_always_on_rulebook: the always-on rulebook binds this
    project again. Idempotent."""
    uid = current_user_id()
    await rulebooks_svc.include_always_on_rulebook_for_project(
        project_id=project_id, rulebook_id=rulebook_id, user_id=uid,
    )
    return {"project_id": project_id, "rulebook_id": rulebook_id, "excluded": False}
 async def suppress_rule_for_project(
    project_id: int, rule_id: int,
@@ -1087,8 +971,6 @@ async def unsuppress_topic_for_project(
    return {"project_id": project_id, "topic_id": topic_id, "suppressed": False}
 async def relate_rules(
    from_rule_id: int, to_rule_id: int, kind: str, note: str = "",
 ) -> dict:
@@ -1136,7 +1018,7 @@ async def unrelate_rules(relation_id: int) -> dict:
 # ── The staleness sweep (milestone 312) ────────────────────────────────
 async def rules_due_for_verification(
-    older_than_days: int = 0, tier: str = "", never_only: bool = False,
+    older_than_days: int = 0, never_only: bool = False,
 ) -> dict:
    """Which standing rules assert a FACT that nobody has confirmed lately.
@@ -1163,9 +1045,6 @@ async def rules_due_for_verification(
    Args:
        older_than_days: only rules last verified longer ago than this.
            Never-checked rules always qualify. 0 = no age filter.
        tier: "always_on" or "conditional" to narrow. An always-on constraint
            that has gone false is the expensive kind — it is preloaded into
            every session, so a wrong one is wrong everywhere at once.
        never_only: only rules nobody has ever verified.
    NOT filterable by project, deliberately: a project reaches rules through
@@ -1175,7 +1054,7 @@ async def rules_due_for_verification(
    """
    uid = current_user_id()
    rules = await rulebooks_svc.rules_due_for_verification(
-        uid, older_than_days=older_than_days, tier=tier, never_only=never_only,
+        uid, older_than_days=older_than_days, never_only=never_only,
    )
    return {
        "rules": [rulebooks_svc.verification_row(r) for r in rules],
@@ -1228,14 +1107,13 @@ def register(mcp) -> None:
    for fn in (
        list_rulebooks, get_rulebook, create_rulebook, update_rulebook, delete_rulebook,
        list_topics, create_topic, update_topic, delete_topic,
-        list_rules, list_always_on_rules, get_rule,
+        list_rules, get_rule,
        create_rule, create_project_rule, update_rule, delete_rule,
        create_preference, update_preference,
        relate_rules, unrelate_rules,
        subscribe_project_to_rulebook, unsubscribe_project_from_rulebook,
        suppress_rule_for_project, unsuppress_rule_for_project,
        suppress_topic_for_project, unsuppress_topic_for_project,
        exclude_always_on_rulebook, include_always_on_rulebook,
        rules_due_for_verification, mark_rule_verified,
        rule_history,
    ):
@@ -40,7 +40,6 @@ async def _search_rules(uid: int, q: str, limit: int) -> dict:
                "title": rule.title,
                "statement": rule.statement,
                "when_to_apply": rule.when_to_apply or "",
                "tier": rule.tier,
                "why": rule.why or "",
                "how_to_apply": rule.how_to_apply or "",
                "verify_with": rule.verify_with or "",
@@ -282,7 +281,7 @@ It is an UPPER BOUND per surface: a pull records the door it came
    `surfaced` VS `ambient` IS THE READING THAT MATTERS HERE. `surfaced` counts
    rules a ranker chose — today only the write-path arm — and those are claims
    a pull can settle. `ambient` counts BULK DELIVERIES: the SessionStart
-    preload, `list_always_on_rules`, and the `rules_payload` surfaces
+    preload and the `rules_payload` surfaces
    (`enter_project`, `get_project`, `get_milestone`, `start_planning`,
    `get_task`), which hand over the whole applicable set at once with nobody
    choosing anything. A large `ambient` says the resident set is big and
@@ -39,7 +39,7 @@ class Project(Base, TimestampMixin, SoftDeleteMixin):
    )
    # The inception record (milestone 297): what this project was decided to
    # inherit, when, and through which door — {decided_at, decided_by, via,
-    # choices: {exclude_always_on_rulebooks, subscribe_rulebooks,
+    # choices: {subscribe_rulebooks,
    # design_system_id, seed_systems}}. NULL means nobody has decided yet,
    # and enter_project asks; the effects themselves live in the subscription
    # / exclusion tables, design_system_id and the project's Systems — this is
@@ -58,7 +58,6 @@ class RuleVersion(Base, CreatedAtMixin):
    why: Mapped[str | None] = mapped_column(Text, nullable=True)
    how_to_apply: Mapped[str | None] = mapped_column(Text, nullable=True)
    when_to_apply: Mapped[str | None] = mapped_column(Text, nullable=True)
    tier: Mapped[str | None] = mapped_column(Text, nullable=True)
    # Carried so that a change of FORCE leaves a trace. `record_if_changed`
    # snapshots only the fields a version holds, so a kind omitted here would
    # make "this stopped binding" the one edit with no history behind it.
@@ -90,7 +89,6 @@ class RuleVersion(Base, CreatedAtMixin):
                "why": self.why or "",
                "how_to_apply": self.how_to_apply or "",
                "when_to_apply": self.when_to_apply or "",
                "tier": self.tier or "",
                "kind": self.kind or "",
                "verify_with": self.verify_with or "",
                "expires_when": self.expires_when or "",
@@ -19,9 +19,6 @@ class Rulebook(Base, TimestampMixin, SoftDeleteMixin):
    )
    title: Mapped[str] = mapped_column(Text)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    always_on: Mapped[bool] = mapped_column(
        Boolean, default=False, nullable=False, server_default="false"
    )
    def to_dict(self) -> dict:
        return {
@@ -29,7 +26,6 @@ class Rulebook(Base, TimestampMixin, SoftDeleteMixin):
            "owner_user_id": self.owner_user_id,
            "title": self.title,
            "description": self.description or "",
            "always_on": self.always_on,
            "created_at": iso(self.created_at),
            "updated_at": iso(self.updated_at),
        }
@@ -96,16 +92,15 @@ class Rule(Base, TimestampMixin, SoftDeleteMixin):
    # WHEN this rule applies — the trigger, not the instruction. Required of
    # new rules at the service layer and nullable here, because rules written
    # before migration 0088 have none and a migration cannot invent one.
-    # It carries three jobs at once (note 3026): it is the tier test made
+    # It carries three jobs at once (note 3026): it is the readable form of
-    # concrete, the readable form of the canon tag, and the half of the
+    # the canon tag, the half of the document that makes a rule findable by
-    # document that makes a rule findable by meaning.
+    # meaning, and — since milestone 394 removed the always-on tier — the ONLY
    # thing that decides whether a rule ever reaches a session at all. A rule
    # with no trigger is not a quiet rule, it is an unreachable one.
    when_to_apply: Mapped[str | None] = mapped_column(Text, nullable=True)
-    # always_on = preloaded into every session, as every rule is today.
+    # WHAT KIND of instruction this is. `tier` used to sit beside this and
-    # conditional = reachable, and surfaced when its trigger fires. The
+    # carry delivery; milestone 394 removed it, so kind is now the only axis
-    # default preserves existing behaviour exactly: nothing stops binding
+    # on a rule and delivery belongs entirely to retrieval.
    # because of an upgrade. CHECK ck_rules_tier (migration 0088, rule 36).
    tier: Mapped[str] = mapped_column(Text, default="always_on", server_default="always_on")
    # WHAT KIND of instruction this is — force, where `tier` is delivery.
    # `rule` must be FOLLOWED: ignoring it breaks something or crosses a
    # boundary. `preference` is how this person wants work DONE: ignoring it
    # costs consistency, not correctness.
@@ -161,7 +156,6 @@ class Rule(Base, TimestampMixin, SoftDeleteMixin):
            "title": self.title,
            "statement": self.statement,
            "when_to_apply": self.when_to_apply or "",
            "tier": self.tier,
            # Unconditional, unlike the `if present` keys below. A reader
            # deciding how much force a record carries must never infer it
            # from an ABSENT key: "no kind field" and "kind is rule" would be
@@ -266,7 +260,7 @@ project_rule_suppressions = Table(
 # sibling of the two suppression tables below, one level up. Always-on
 # rulebooks bind every project implicitly; an inception decision can exclude
 # specific ones for this project, and get_applicable_rules /
-# list_always_on_rules(project_id) skip them. FKs CASCADE like the others.
+# get_applicable_rules(project_id) skips them. FKs CASCADE like the others.
 project_rulebook_exclusions = Table(
    "project_rulebook_exclusions",
    Base.metadata,
@@ -202,11 +202,6 @@ async def write_path_prior_art():
                             or `canon:<snippet_id>`) already named this
                             session by the ledger arm (#2900); its own
                             channel, like the two above.
        rules_etag (opt)   — the marker the session was given when it loaded
                             its always-on rules (milestone 323). Sent back
                             so the server can say whether those rules have
                             MOVED since. Absent means the hook has nothing
                             stored, which is silence, not a mismatch.
        shapes (opt)       — comma-separated `kind:name` definitions the hook
                             found in (or enclosing) the payload, kind being
                             css|sym. The shape ledger's write-path feed
@@ -226,7 +221,6 @@ async def write_path_prior_art():
        p.strip() for p in (request.args.get("exclude_derive") or "").split(",") if p.strip()
    ]
    exclude_rule_ids = _int_list(request.args.get("exclude_rule_ids"))
    rules_etag = (request.args.get("rules_etag") or "").strip()
    shapes = _parse_shapes(request.args.get("shapes") or "")
    api_key = getattr(g, "api_key", None)
    may_stamp = api_key is None or getattr(api_key, "scope", "") == "write"
@@ -238,7 +232,6 @@ async def write_path_prior_art():
        repo_key=repo_bindings_svc.normalize_repo_key(repo) if repo else "",
        exclude_derive=exclude_derive,
        exclude_rule_ids=exclude_rule_ids,
        rules_etag=rules_etag,
    )
    return jsonify(result)
@@ -99,7 +99,7 @@ async def create_project_route():
@login_required
 async def decide_inception_route(project_id: int):
    """Record (or re-record) what a project inherits — milestone 297.
-    Body: the choices object {exclude_always_on_rulebooks, subscribe_rulebooks,
+    Body: the choices object {subscribe_rulebooks,
    design_system_id, seed_systems}; owner-only."""
    uid = get_current_user_id()
    data = await request.get_json() or {}
@@ -55,7 +55,7 @@ async def get_rulebook(rulebook_id: int):
@login_required
 async def update_rulebook(rulebook_id: int):
    data = await request.get_json() or {}
-    fields = {k: v for k, v in data.items() if k in ("title", "description", "always_on")}
+    fields = {k: v for k, v in data.items() if k in ("title", "description")}
    rb = await rulebooks_svc.update_rulebook(rulebook_id, get_current_user_id(), **fields)
    if rb is None:
        return jsonify({"error": "rulebook not found"}), 404
@@ -177,7 +177,6 @@ async def create_rule(topic_id: int):
            how_to_apply=data.get("how_to_apply", ""),
            order_index=data.get("order_index", 0),
            when_to_apply=data.get("when_to_apply", ""),
            tier=data.get("tier", "always_on"),
            # The human door carries `kind` too, and without the MCP door's
            # required provenance: an operator editing their own preference
            # owes nobody an explanation. That requirement is about auditing
@@ -217,7 +216,7 @@ async def update_rule(rule_id: int):
    fields = {
        k: v for k, v in data.items()
        if k in ("title", "statement", "why", "how_to_apply", "order_index",
-                 "when_to_apply", "tier", "kind", "arose_from_id",
+                 "when_to_apply", "kind", "arose_from_id",
                 "verify_with", "expires_when")
    }
    # No clear_fields here: a form sends "" for an emptied input, and the
@@ -395,32 +394,6 @@ async def unsuppress_project_topic(project_id: int, topic_id: int):
    return "", 204
@rulebooks_bp.post("/projects/<int:project_id>/exclusions/rulebooks/<int:rulebook_id>")
@login_required
 async def exclude_project_rulebook(project_id: int, rulebook_id: int):
    """Opt the project out of a whole always-on rulebook (milestone 297)."""
    try:
        await rulebooks_svc.exclude_always_on_rulebook_for_project(
            project_id=project_id, rulebook_id=rulebook_id, user_id=get_current_user_id(),
        )
    except ValueError as exc:
        msg = str(exc)
        return jsonify({"error": msg}), (400 if "not always-on" in msg else 404)
    return "", 204
@rulebooks_bp.delete("/projects/<int:project_id>/exclusions/rulebooks/<int:rulebook_id>")
@login_required
 async def include_project_rulebook(project_id: int, rulebook_id: int):
    try:
        await rulebooks_svc.include_always_on_rulebook_for_project(
            project_id=project_id, rulebook_id=rulebook_id, user_id=get_current_user_id(),
        )
    except ValueError as exc:
        return jsonify({"error": str(exc)}), 404
    return "", 204
@rulebooks_bp.post("/projects/<int:project_id>/rules")
@login_required
 async def create_project_rule(project_id: int):
@@ -440,7 +413,6 @@ async def create_project_rule(project_id: int):
            how_to_apply=data.get("how_to_apply", ""),
            order_index=data.get("order_index", 0),
            when_to_apply=data.get("when_to_apply", ""),
            tier=data.get("tier", "always_on"),
            # The human door carries `kind` too, and without the MCP door's
            # required provenance: an operator editing their own preference
            # owes nobody an explanation. That requirement is about auditing
@@ -464,7 +436,7 @@ async def create_project_rule(project_id: int):
 async def rules_due_for_verification():
    """Rules that carry a check, oldest verification first, never-checked top.
-    Query params: older_than_days, tier, never_only. A rule with no
+    Query params: older_than_days, never_only. A rule with no
    `verify_with` never appears — it is a decision, not a fact.
    """
    uid = get_current_user_id()
@@ -477,7 +449,6 @@ async def rules_due_for_verification():
        rules = await rulebooks_svc.rules_due_for_verification(
            uid,
            older_than_days=older,
            tier=args.get("tier", ""),
            never_only=args.get("never_only", "").lower() in ("1", "true", "yes"),
        )
    except ValueError as exc:
@@ -513,7 +513,7 @@ def _rule_version_rows(rows) -> list[dict]:
            "id": rv.id, "rule_id": rv.rule_id, "user_id": rv.user_id,
            "title": rv.title, "statement": rv.statement, "why": rv.why,
            "how_to_apply": rv.how_to_apply, "when_to_apply": rv.when_to_apply,
-            "tier": rv.tier, "kind": rv.kind, "verify_with": rv.verify_with,
+            "kind": rv.kind, "verify_with": rv.verify_with,
            "expires_when": rv.expires_when,
            "created_at": rv.created_at.isoformat(),
        }
@@ -529,7 +529,7 @@ def _rulebook_rows(rows) -> list[dict]:
    return [
        {
            "id": rb.id, "owner_user_id": rb.owner_user_id, "title": rb.title,
-            "description": rb.description, "always_on": rb.always_on,
+            "description": rb.description,
            "created_at": rb.created_at.isoformat(),
            "updated_at": rb.updated_at.isoformat(),
        }
@@ -575,7 +575,7 @@ def _rule_rows(rows) -> list[dict]:
            "id": r.id, "topic_id": r.topic_id, "project_id": r.project_id,
            "title": r.title, "statement": r.statement, "why": r.why,
            "how_to_apply": r.how_to_apply, "order_index": r.order_index,
-            "when_to_apply": r.when_to_apply, "tier": r.tier, "kind": r.kind,
+            "when_to_apply": r.when_to_apply, "kind": r.kind,
            "verify_with": r.verify_with, "expires_when": r.expires_when,
            "verified_at": r.verified_at.isoformat() if r.verified_at else None,
            "arose_from_id": r.arose_from_id,
@@ -1238,7 +1238,6 @@ async def _restore_v2(data: dict) -> dict:
                owner_user_id=mapped_uid,
                title=rb_data.get("title", ""),
                description=rb_data.get("description", ""),
                always_on=rb_data.get("always_on", False),
                created_at=_dt(rb_data.get("created_at")),
                updated_at=_dt(rb_data.get("updated_at")),
            )
@@ -1279,10 +1278,14 @@ async def _restore_v2(data: dict) -> dict:
                why=r_data.get("why") or None,
                how_to_apply=r_data.get("how_to_apply") or None,
                when_to_apply=r_data.get("when_to_apply") or None,
-                # A file written before migration 0088 has no tier. always_on
+                # A file written before milestone 394 carries `tier` and
                # `always_on`; neither is read. Dropping a field the schema
                # no longer has is the tolerant direction — an archive
                # records what WAS, and refusing it because it remembers a
                # deleted column would make every pre-394 backup
                # unrestorable. Previously: always_on
                # is the pre-0088 behaviour, so an old backup restores rules
                # that bind exactly as they did when it was taken.
                tier=r_data.get("tier") or "always_on",
                # Same shape, same reason: a file written before 0098 has no
                # kind, and every rule in it was a rule. Defaulting the other
                # way would restore an old backup with things that had always
@@ -1427,7 +1430,6 @@ async def _restore_v2(data: dict) -> dict:
                why=rv.get("why"),
                how_to_apply=rv.get("how_to_apply"),
                when_to_apply=rv.get("when_to_apply"),
                tier=rv.get("tier"),
                # NOT defaulted, unlike the rule above. A version records what
                # was; absent means nobody wrote it down, and inventing "rule"
                # here would put an artifact where a measurement belongs.
@@ -817,7 +817,6 @@ async def semantic_search_rules(
    query: str,
    limit: int = 5,
    threshold: float = _SIMILARITY_THRESHOLD,
    tier: str | None = None,
    kind: str | None = None,
    report: dict | None = None,
 ) -> list[tuple[float, "Rule"]]:
@@ -844,17 +843,13 @@ async def semantic_search_rules(
    is the surfacing question, and it has its own machinery
    (get_applicable_rules) rather than a second, subtly different copy here.
-    `tier` narrows to one tier, and NONE is the ordinary case. The write-path
+    THERE IS NO TIER TO NARROW BY ANY MORE (milestone 394). This carried a
-    and pre-tool hints deliberately pass nothing: an always-on rule is already
+    `tier` parameter, and the arms deliberately passed nothing: filtering on it
-    in the session, but being in a list from turn zero is not the same as being
+    made a whole class of rules permanently ineligible for the one mechanism
-    in front of the reader when the action it governs is taken, and filtering
+    that surfaces a rule AT the moment it applies. The tier is now gone
-    on tier made a whole class of rules permanently ineligible for the one
+    entirely, so every rule is eligible for every arm and relevance is the
-    mechanism that surfaces a rule AT the moment. Relevance is the threshold's
+    threshold's job alone — see the block above RULEHINT_LIMIT in
-    job; see the block above RULEHINT_LIMIT in services/plugin_context.py for
+    services/plugin_context.py for what those scores are read against.
    the argument and for what the resulting scores are being read against.
    Pass a tier when a caller genuinely wants one class — a listing, an audit,
    a UI that renders the tiers apart. Not to approximate relevance.
    `kind` narrows to `rule` or `preference`, and NONE is likewise the ordinary
    case: a caller asking "what governs this" wants both, because the reader
@@ -905,7 +900,6 @@ async def semantic_search_rules(
                        Rulebook.owner_user_id == user_id,
                        Project.user_id == user_id,
                    ),
                    *( [Rule.tier == tier] if tier else [] ),
                    *( [Rule.kind == kind] if kind else [] ),
                )
                # Overfetch so collapsing chunks to their best row still fills
@@ -7,7 +7,6 @@ A project's inheritance is a decision, not a default. The record lives on
      "decided_at": "<iso>", "decided_by": <user id> | null,
      "via": "mcp" | "ui" | "legacy",
      "choices": {
        "exclude_always_on_rulebooks": [rulebook ids],
        "subscribe_rulebooks": [rulebook ids],
        "design_system_id": <id> | null,
        "seed_systems": bool
@@ -18,9 +17,14 @@ NULL = undecided → enter_project asks. ``legacy`` is the migration's stamp on
 projects that existed before the step did (inherit-all / no design system /
 no seed), so the ask fires only for projects created after this shipped.
 ``exclude_always_on_rulebooks`` was a fourth choice until milestone 394. It
 let a project decline to inherit an always-on rulebook, and with no always-on
 tier there is nothing to decline — a rulebook now reaches a project by
 subscription, which is opt-IN, so declining is expressed by not subscribing.
 The shape and its validator are pure; ``decide`` composes the existing
-services — always-on exclusions, subscriptions, set_project_design_system,
+services — subscriptions, set_project_design_system, the standard Systems
-the standard Systems seed — checks every target BEFORE touching anything,
+seed — checks every target BEFORE touching anything,
 applies the effects (each idempotent), and writes the record LAST, so a
 half-applied decision is re-runnable rather than recorded as done.
 ``current_defaults`` is what the enter_project ask shows: what binds today
@@ -37,7 +41,7 @@ from scribe.models.project import Project
 from scribe.models.rulebook import Rulebook
 INCEPTION_VIAS = ("mcp", "ui", "legacy")
-CHOICE_KEYS = ("exclude_always_on_rulebooks", "subscribe_rulebooks", "design_system_id", "seed_systems")
+CHOICE_KEYS = ("subscribe_rulebooks", "design_system_id", "seed_systems")
 def _is_id_list(value) -> bool:
@@ -60,15 +64,9 @@ def validate_inception(choices) -> str | None:
    unknown = sorted(set(choices) - set(CHOICE_KEYS))
    if unknown:
        return f"unknown inception choice(s): {', '.join(unknown)} (one of: {', '.join(CHOICE_KEYS)})"
    excl = choices.get("exclude_always_on_rulebooks") or []
    subs = choices.get("subscribe_rulebooks") or []
    if not _is_id_list(excl):
        return "exclude_always_on_rulebooks must be a list of rulebook ids"
    if not _is_id_list(subs):
        return "subscribe_rulebooks must be a list of rulebook ids"
    both = sorted(set(excl) & set(subs))
    if both:
        return f"rulebook(s) {both} cannot be both excluded and subscribed"
    ds = choices.get("design_system_id")
    if ds is not None and (isinstance(ds, bool) or not isinstance(ds, int) or ds <= 0):
        return "design_system_id must be a positive id or null"
@@ -79,11 +77,10 @@ def validate_inception(choices) -> str | None:
 def normalize_choices(choices: dict | None) -> dict:
-    """The four keys, always present, in canonical form — what gets stored
+    """The three keys, always present, in canonical form — what gets stored
    and what the UI/agent reads back. Call after validate_inception."""
    choices = choices or {}
    return {
        "exclude_always_on_rulebooks": sorted(set(choices.get("exclude_always_on_rulebooks") or [])),
        "subscribe_rulebooks": sorted(set(choices.get("subscribe_rulebooks") or [])),
        "design_system_id": choices.get("design_system_id"),
        "seed_systems": bool(choices.get("seed_systems", False)),
@@ -98,8 +95,7 @@ def is_decided(project) -> bool:
 async def current_defaults(user_id: int, project_id: int) -> dict:
    """What the project inherits if nobody decides — the ask's payload.
-    {always_on_rulebooks: [{id,title}], other_rulebooks: [{id,title}],
+    {rulebooks: [{id,title}], subscribed_rulebooks: [...],
     excluded_always_on: [...], subscribed_rulebooks: [...],
     design_system_id, design_systems: [{id,title}], systems: <count>}.
    Instance-agnostic: an install with no rulebooks / design systems shows
    empty lists, and the ask says so rather than inventing a default.
@@ -115,7 +111,7 @@ async def current_defaults(user_id: int, project_id: int) -> dict:
    async with async_session() as session:
        rows = (
            await session.execute(
-                select(Rulebook.id, Rulebook.title, Rulebook.always_on)
+                select(Rulebook.id, Rulebook.title)
                .where(Rulebook.owner_user_id == user_id, Rulebook.deleted_at.is_(None))
                .order_by(Rulebook.title)
            )
@@ -124,9 +120,11 @@ async def current_defaults(user_id: int, project_id: int) -> dict:
    designs = await design_systems_svc.list_design_systems(user_id)
    systems = await systems_svc.list_systems(user_id, project_id, include_archived=True)
    return {
-        "always_on_rulebooks": [{"id": i, "title": t} for i, t, on in rows if on],
+        # ONE list since milestone 394. This was split into always-on and
-        "other_rulebooks": [{"id": i, "title": t} for i, t, on in rows if not on],
+        # "other" because the first bound the project whether it asked or not;
-        "excluded_always_on": applicable.get("excluded_always_on", []),
+        # with the tier gone every rulebook is opt-in, so the split named a
        # difference that no longer exists.
        "rulebooks": [{"id": i, "title": t} for i, t in rows],
        "subscribed_rulebooks": applicable.get("subscribed_rulebooks", []),
        "design_system_id": project.design_system_id,
        "design_systems": [{"id": d.id, "title": d.title} for d in designs],
@@ -139,28 +137,22 @@ async def _check_targets(user_id: int, choices: dict) -> None:
    effect lands — a decision applies whole or errors whole."""
    from scribe.services import access
-    wanted = set(choices["exclude_always_on_rulebooks"]) | set(choices["subscribe_rulebooks"])
+    wanted = set(choices["subscribe_rulebooks"])
    if wanted:
        async with async_session() as session:
            rows = (
                await session.execute(
-                    select(Rulebook.id, Rulebook.always_on).where(
+                    select(Rulebook.id).where(
                        Rulebook.id.in_(wanted),
                        Rulebook.owner_user_id == user_id,
                        Rulebook.deleted_at.is_(None),
                    )
                )
            ).all()
-        found = {rid: on for rid, on in rows}
+        found = {rid for (rid,) in rows}
-        missing = sorted(wanted - set(found))
+        missing = sorted(wanted - found)
        if missing:
            raise ValueError(f"rulebook(s) {missing} not found (or not yours)")
        not_always = sorted(r for r in choices["exclude_always_on_rulebooks"] if not found[r])
        if not_always:
            raise ValueError(
                f"rulebook(s) {not_always} are not always-on — only always-on rulebooks "
                "can be excluded; a subscribed rulebook is simply not subscribed"
            )
    ds = choices["design_system_id"]
    if ds is not None and not await access.can_read_design_system(user_id, ds):
        raise ValueError(f"design system {ds} not found (or not readable)")
@@ -176,13 +168,12 @@ async def decide(
    """Record a project's inception decision and apply it (milestone 297).
    Owner-only. Validates the choices (pure) and every target (owned /
-    readable) first; then, each idempotent: exclude the named always-on
+    readable) first; then, each idempotent: subscribe the named rulebooks,
-    rulebooks, subscribe the named rulebooks, point the project at the design
+    point the project at the design system (None = explicitly none), seed the
-    system (None = explicitly none), seed the standard Systems if asked and
+    standard Systems if asked and the project has none; then write
-    the project has none; then write ``projects.inception`` LAST. Re-deciding
+    ``projects.inception`` LAST. Re-deciding is additive for subscriptions
-    is additive for exclusions/subscriptions (nothing is silently dropped —
+    (nothing is silently dropped — unsubscribe is an explicit call), replaces
-    include/unsubscribe are explicit calls), replaces the design system, and
+    the design system, and re-seeds nothing a project already has.
    re-seeds nothing a project already has.
    Returns {"inception": <record>, "effects": {excluded, subscribed,
    design_system_id, systems_seeded}}.
@@ -203,8 +194,6 @@ async def decide(
        raise ValueError(f"project {project_id} not found (or not yours)")
    await _check_targets(user_id, choices)
    for rb in choices["exclude_always_on_rulebooks"]:
        await rulebooks_svc.exclude_always_on_rulebook_for_project(project_id, rb, user_id)
    for rb in choices["subscribe_rulebooks"]:
        await rulebooks_svc.subscribe_project(project_id, rb, user_id)
    if not await design_systems_svc.set_project_design_system(
@@ -230,7 +219,6 @@ async def decide(
    return {
        "inception": record,
        "effects": {
            "excluded": choices["exclude_always_on_rulebooks"],
            "subscribed": choices["subscribe_rulebooks"],
            "design_system_id": choices["design_system_id"],
            "systems_seeded": [sy.name for sy in seeded],
@@ -247,25 +235,24 @@ async def inception_ask(user_id: int, project_id: int) -> dict:
        defaults = await current_defaults(user_id, project_id)
    except Exception:
        return {}
-    always = ", ".join(f"{r['title']} (#{r['id']})" for r in defaults["always_on_rulebooks"]) or "none"
+    books = ", ".join(f"{r['title']} (#{r['id']})" for r in defaults["rulebooks"]) or "none"
    others = ", ".join(f"{r['title']} (#{r['id']})" for r in defaults["other_rulebooks"]) or "none"
    designs = ", ".join(f"{d['title']} (#{d['id']})" for d in defaults["design_systems"]) or "none"
    return {
        "defaults": defaults,
        "ask": (
            "This project has no inception decision: nobody has said what it "
-            f"inherits. Today, by default: always-on rulebooks binding it — {always}; "
+            f"inherits. Rulebooks it could subscribe to — {books}; design system — "
            f"rulebooks it could subscribe to — {others}; design system — "
            f"{'#' + str(defaults['design_system_id']) if defaults['design_system_id'] else 'none'} "
            f"(available: {designs}); Systems — {defaults['systems']}. Ask the operator, "
-            "once: which always-on rulebooks to EXCLUDE here (default: none), which "
+            "once: which rulebooks to subscribe (default: none — a rulebook binds "
-            "rulebooks to subscribe, which design system (or none), and whether to seed "
+            "a project only when it opts in), which design system (or none), and "
            "whether to seed "
            "the standard starter Systems — then record the answers. This ask repeats on "
            "every enter_project until a decision is recorded."
        ),
        "call": (
            f"decide_project_inception(project_id={project_id}, "
-            "exclude_always_on_rulebooks=[...], subscribe_rulebooks=[...], "
+            "subscribe_rulebooks=[...], "
            "design_system_id=<id | -1 for none>, seed_systems=<true|false>)"
        ),
    }
@@ -8,7 +8,7 @@ Design note — altitude: we inject rule *titles* grouped by topic (a compact
 index), NOT every rule's full statement. The 48 always-on statements run well
 past the 10k-char `additionalContext` cap, and the push channel's job is to make
 Claude *aware* the rules exist and *reach* for them — not to dump them. Full
-text stays one `get_rule(id)` / `list_always_on_rules()` call away. Titles are
+text stays one `get_rule(id)` / `search(content_type="rule")` call away. Titles are
 mostly self-describing ("`dev` is home", "No GitHub — Fabled-Git only"), so the
 index alone already steers behavior.
 """
@@ -1400,7 +1400,6 @@ async def build_write_path_hint(
    repo_key: str = "",
    exclude_derive: list[str] | None = None,
    exclude_rule_ids: list[int] | None = None,
    rules_etag: str = "",
 ) -> dict:
    """Prior-art hint for the plugin's PreToolUse hook on Write/Edit.
@@ -1686,60 +1685,22 @@ async def build_write_path_hint(
    staleness: list[str] = []
    # ── Have the rules moved under this session? (milestone 323) ───────
    #
-    # THE CARRIER IS THE POINT. This hook already fires before a write — the
+    # THE RULES-ETAG STALENESS ARM IS GONE (milestone 394).
    # moment acting on a stale rule actually costs something — and the check
    # is one comparison against a marker the session already holds. No
    # payload, no extra round trip, and nothing said when nothing moved.
    #
-    # WHAT THIS CANNOT SEE, and a reader who finds an etag here will assume
+    # It took a marker the session had been given at SessionStart, compared it
-    # otherwise:
+    # against the resident set as it stood now, and said which rules had moved
    # or fallen out of force. That was worth doing while a session held a
    # fixed set of rules from turn zero and could be holding a stale copy of
    # it hours later.
    #
-    #   what goes wrong                                    | caught?
+    # Nothing is resident now. A rule is retrieved at the moment it applies,
-    #   ---------------------------------------------------|--------
+    # so a session cannot be holding an out-of-date one — the next act that
-    #   another session edits a rule mid-flight             | yes
+    # needs it fetches it again. The staleness this arm reported was an
-    #   the session is misremembering a rule read hours ago | yes
+    # artifact of the delivery model rather than a fact about the corpus, and
-    #   compaction summarised the rules out of context      | NO
+    # it goes with the model.
    #
-    # The third is the most common and this is blind to it: the etag was in
+    # `staleness` survives as the list the arms below still append to.
-    # context too and went with the rules. The SessionStart nudge is that
+
    # case's only mechanism and must not be softened because this shipped.
    #
    # Fails open, like every other arm here: a staleness hint must never
    # break a write.
    if rules_etag:
        try:
            current = await rulebooks_svc.list_always_on_rules(
                user_id, project_id=project_id or 0,
            )
            if rulebooks_svc.rules_etag(current) != rules_etag:
                moved = rulebooks_svc.rules_moved_since(current, rules_etag)
                held = rulebooks_svc.etag_count(rules_etag)
                bits = []
                if moved:
                    named = ", ".join(
                        f"#{r.id} \u201c{r.title}\u201d" for r in moved[:3]
                    )
                    more = len(moved) - 3
                    bits.append(
                        f"{named}" + (f", and {more} more" if more > 0 else "")
                    )
                # A DELETED rule moves no timestamp and leaves no row to name,
                # so the count is the only thing that can report the one change
                # that takes an instruction OUT of force.
                if held is not None and held != len(current):
                    delta = len(current) - held
                    bits.append(
                        f"{abs(delta)} rule(s) {'added' if delta > 0 else 'no longer in force'}"
                    )
                if bits:
                    staleness.append(
                        "Your loaded rules have changed since this session "
                        "started — " + "; ".join(bits) + ". Re-read them with "
                        "list_always_on_rules() before relying on the set you "
                        "are holding."
                    )
        except Exception:
            logger.debug("write-path rules-etag arm failed", exc_info=True)
    # The guard sits BELOW the staleness arm on purpose. A rules change is
    # unconditional news — it does not become less true because this
@@ -2208,67 +2169,37 @@ async def build_session_context(
            its normalized key — triggers a one-line "bind this repo" hint so
            the binding is self-healing.
-    Returns {"context": str, "rule_count": int, "project": dict | None,
+    Returns {"context": str, "project": dict | None}.
-    "rules_etag": str}. The etag is for the HOOK, not for the model — the
+
-    hook stores it and hands it back on each write so the server can say
+    It carried `rule_count` and `rules_etag` until milestone 394, when the
-    whether these rules have moved since the session loaded them.
+    preload it described was removed. The etag let the hook hand a marker back
    on each write so the server could say whether the resident rules had
    moved; nothing is resident now, so nothing can have moved, and a rule is
    re-retrieved at the moment it applies rather than held and aged.
    `context` is markdown ready to drop into `additionalContext`; it is capped
    at _MAX_CHARS with an explicit truncation note so the hook can pass it
    through verbatim.
    """
    # Inside a project, the always-on set is the project's: an inception
    # exclusion (milestone 297) takes a rulebook out of this block, and is
    # named below so the departure is visible rather than silent.
    rules = await rulebooks_svc.list_always_on_rules(user_id, project_id=project_id)
    # AMBIENT source, and the one that matters most: this is the preload — the
    # block every session opens with, chosen by nobody, paid for every turn.
    #
    # It emitted nothing until 2026-09-03, which made the resident set's cost
    # certain and its usefulness unfalsifiable at the same time (#3473). Note
    # #3089 is the argument this measurement finally lets someone test: that a
    # rule arriving with thirty others, none of them relevant, is read as
    # preamble rather than as a claim — so presence is not surfacing, and a
    # tier-1 set can grow without anybody noticing it stopped working.
    #
    # Recorded even when the hook truncates the block below: the rules WERE
    # delivered, and counting only the untruncated ones would quietly shrink
    # the denominator exactly where the set is too big to read.
    record_rule_surfaced(
        user_id=user_id,
        rule_ids=[r.id for r in rules],
        source="session_start",
    )
    excluded = (
        await rulebooks_svc.excluded_always_on_rulebooks(user_id, project_id)
        if project_id else []
    )
    topic_map = await _topic_titles({r.topic_id for r in rules if r.topic_id})
    lines: list[str] = [
        "# Scribe — standing session context (auto-injected by the Scribe plugin)",
        "",
-        "You are working with Scribe, the operator's self-hosted second brain. "
+        "You are working with Scribe, the operator's self-hosted second brain.",
        "The always-on rules below are BINDING this session. Titles only — full "
        "text via `list_always_on_rules()` or `get_rule(id)`.",
        "",
-        "## Always-on rules (by topic)",
+        "## You are not holding the operator's rules",
        "",
        "No rule has been loaded into this session, and that is deliberate. "
        "Rules arrive when something you are about to do makes one relevant — "
        "a command you are about to run, code you are writing, or what the "
        "operator just asked for. On most turns none will, and that is the "
        "surface working rather than failing.",
        "",
        "**\"No rule arrived\" means \"nothing matched\" — never \"there is no "
        "rule.\"** Before a consequential act, one that is hard to reverse or "
        "outward-facing, `search(content_type=\"rule\")` is how you ask. "
        "Retrieval runs on its own and is a convenience; asking is what you do "
        "when it matters and nothing has spoken.",
    ]
    # rules already arrive ordered by rulebook/topic/order, so grouping by
    # consecutive topic_id preserves the intended sequence.
    current_topic: int | None = object()  # sentinel distinct from any id/None
    for r in rules:
        if r.topic_id != current_topic:
            current_topic = r.topic_id
            heading = topic_map.get(r.topic_id, "ungrouped") if r.topic_id else "ungrouped"
            lines.append(f"### {heading}")
        lines.append(f"- [{r.id}] {r.title}")
    if excluded:
        names = ", ".join(f"{e['title']} (#{e['id']})" for e in excluded)
        lines += [
            "",
            f"Excluded for this project by its inception decision (not binding here): {names}.",
        ]
    project_dict: dict | None = None
    if project_id:
@@ -2336,14 +2267,10 @@ async def build_session_context(
    context = "\n".join(line for line in lines if line is not None)
    if len(context) > _MAX_CHARS:
-        context = context[:_MAX_CHARS].rstrip() + "\n\n…(truncated — call list_always_on_rules())"
+        context = context[:_MAX_CHARS].rstrip() + \
            "\n\n…(truncated — ask with search(content_type=\"rule\"))"
    return {
        "context": context,
        "rule_count": len(rules),
        "project": project_dict,
        # Computed from the rules THIS payload was built from, not re-queried:
        # the marker has to describe the set the session is actually holding,
        # and a second query could disagree with the first.
        "rules_etag": rulebooks_svc.rules_etag(rules),
    }
@@ -852,7 +852,7 @@ async def retrieval_summary(
    # something else.
    #
    # `ambient` now carries the bulk deliveries — the SessionStart preload,
-    # `list_always_on_rules`, and every `rules_payload` surface (#3473). Before
+    # and every `rules_payload` surface (#3473). Before
    # they emitted, this block had no ambient key and said the absence was a
    # fact about the data. It was, and it was also the thing that made the
    # always-on set impossible to judge: the largest rule surface in the
@@ -46,7 +46,7 @@ AMBIENT VS RANKED. The note twin splits ranked surfacings from ambient ones
 because `enter_project` and the skill sync put records in front of the agent
 without choosing them, and counting those as surfacings makes recency read as
 popularity (#2477). Rules have exactly that shape: the SessionStart preload,
-`list_always_on_rules`, and every `rules_payload` surface hand over the whole
+and every `rules_payload` surface hand over the whole
 applicable set at once, chosen by nobody.
 Until 2026-09-03 those bulk surfaces emitted nothing, and this module said so —
@@ -184,8 +184,8 @@ def record_rule_surfaced(
 def record_rule_pulled(*, user_id: int | None, rule_id: int, source: str) -> None:
    """Fire-and-forget: record that a rule was opened in full.
-    A PULL is somebody choosing to open one record. `list_always_on_rules` and
+    A PULL is somebody choosing to open one record. `enter_project` is NOT a
-    `enter_project` are NOT pulls — they are bulk resident loads that hand over
+    pull — they are bulk resident loads that hand over
    every applicable rule at once, and counting them would swamp the signal
    with the very ambient delivery the ratio exists to distinguish from.
    """
@@ -36,7 +36,7 @@ from scribe.models.rule_version import RuleVersion
 # snapshots would bury the edits somebody is actually looking for.
 SNAPSHOT_FIELDS = (
    "title", "statement", "why", "how_to_apply", "when_to_apply",
-    "tier", "kind", "verify_with", "expires_when",
+    "kind", "verify_with", "expires_when",
 )
@@ -12,7 +12,7 @@ from collections.abc import Iterable
 from datetime import datetime
 from typing import Optional
-from sqlalchemy import and_, delete as sql_delete, insert, or_, select
+from sqlalchemy import and_, delete as sql_delete, false as sa_false, insert, or_, select
 from scribe.models import async_session
 from scribe.models.system import System
@@ -82,7 +82,7 @@ async def update_rulebook(
        rb = result.scalar_one_or_none()
        if rb is None:
            return None
-        allowed = {"title", "description", "always_on"}
+        allowed = {"title", "description"}
        for key, value in fields.items():
            if key in allowed and value is not None:
                setattr(rb, key, value)
@@ -293,7 +293,6 @@ async def _assert_rulebook_rule_owned(session, rule_id: int, user_id: int) -> No
 # The vocabularies migration 0088's CHECK constraints enforce. Named here so
 # a caller can be corrected before the database refuses it (rule 36 keeps the
 # two in step; this keeps the error readable).
 TIERS = ("always_on", "conditional")
 RELATION_KINDS = ("co_surfaces", "overrides", "elaborates")
 # Migration 0098's CHECK. `rule` binds; `preference` is how the operator
 # wants work done — see the model comment for why both live on one table.
@@ -311,21 +310,11 @@ NULLABLE_RULE_TEXT = (
 )
 def _valid_tier(tier: str) -> str:
    """An unrecognised tier falls back to always_on — the SAFE direction.
    Getting this wrong the other way would silently stop a rule binding, which
    is the one failure this whole milestone exists to prevent. A rule that
    preloads when it did not need to costs context; a rule that quietly stops
    preloading costs the behaviour it was written for.
    """
    return tier if tier in TIERS else "always_on"
 def _valid_kind(kind: str) -> str:
    """An unrecognised kind falls back to `rule` — the SAFE direction.
-    Same shape as _valid_tier and the same argument, pointed at force instead
+    The unrecognised value falls back to the binding one — the SAFE
    direction, pointed at force instead
    of delivery. A preference wrongly treated as binding costs a little
    friction: the reader is told something is required that was only
    preferred. A rule wrongly treated as a preference costs the thing the rule
@@ -369,7 +358,6 @@ def rule_brief(rule: Rule, **extra) -> dict:
        "title": rule.title,
        "statement": rule.statement,
        "topic_id": rule.topic_id,
        "tier": rule.tier,
        # Unconditional, and the payload cost is accepted deliberately. Every
        # other optional key below is attached only when present, because an
        # absent key should never read as a capability the record lacks. Force
@@ -503,7 +491,7 @@ async def rule_detail(user_id: int, rule: Rule, system_ids: list[int] | None = N
 async def create_rule(
    topic_id: int, user_id: int, title: str, statement: str,
    why: str = "", how_to_apply: str = "", order_index: int = 0,
-    when_to_apply: str = "", tier: str = "always_on", arose_from_id: int = 0,
+    when_to_apply: str = "", arose_from_id: int = 0,
    verify_with: str = "", expires_when: str = "", kind: str = "rule",
 ) -> Rule:
    async with async_session() as session:
@@ -513,7 +501,6 @@ async def create_rule(
            title=title,
            statement=statement,
            when_to_apply=when_to_apply or None,
            tier=_valid_tier(tier),
            kind=_valid_kind(kind),
            why=why or None,
            how_to_apply=how_to_apply or None,
@@ -532,7 +519,7 @@ async def create_rule(
 async def create_project_rule(
    project_id: int, user_id: int, title: str, statement: str,
    why: str = "", how_to_apply: str = "", order_index: int = 0,
-    when_to_apply: str = "", tier: str = "always_on", arose_from_id: int = 0,
+    when_to_apply: str = "", arose_from_id: int = 0,
    verify_with: str = "", expires_when: str = "", kind: str = "rule",
 ) -> Rule:
    """Create a rule scoped to a single project (no rulebook ceremony).
@@ -548,7 +535,6 @@ async def create_project_rule(
            title=title,
            statement=statement,
            when_to_apply=when_to_apply or None,
            tier=_valid_tier(tier),
            kind=_valid_kind(kind),
            why=why or None,
            how_to_apply=how_to_apply or None,
@@ -632,89 +618,6 @@ async def list_rules(
        return rulebook_rules + list(proj_result.scalars().all())
 def _excluded_rulebook_ids_q(project_id: int):
    """Subquery: the always-on rulebooks this project opted out of at
    inception (milestone 297) — used by every rule-resolution path so an
    exclusion is total, not just cosmetic."""
    from scribe.models.rulebook import project_rulebook_exclusions
    return select(project_rulebook_exclusions.c.rulebook_id).where(
        project_rulebook_exclusions.c.project_id == project_id
    )
 async def excluded_always_on_rulebooks(user_id: int, project_id: int) -> list[dict]:
    """[{id, title}] of the always-on rulebooks excluded for ``project_id``
    (owner-scoped). Empty for an undecided or inherit-all project."""
    from scribe.models.rulebook import project_rulebook_exclusions
    if not project_id:
        return []
    async with async_session() as session:
        rows = (
            await session.execute(
                select(Rulebook.id, Rulebook.title)
                .join(project_rulebook_exclusions,
                      project_rulebook_exclusions.c.rulebook_id == Rulebook.id)
                .where(
                    project_rulebook_exclusions.c.project_id == project_id,
                    Rulebook.owner_user_id == user_id,
                    Rulebook.deleted_at.is_(None),
                )
                .order_by(Rulebook.title)
            )
        ).all()
    return [{"id": rid, "title": title} for rid, title in rows]
 async def list_always_on_rules(
    user_id: int, limit: int = 100, project_id: int = 0,
 ) -> list[Rule]:
    """Return all rules from rulebooks flagged always_on for the user.
    Called by the MCP tool of the same name at session start to load the
    standing rules that apply regardless of which project (if any) is in
    scope. Ordering matches list_rules so results are stable across calls.
    ``project_id`` (milestone 297): inside a project that excluded specific
    always-on rulebooks at inception, those rulebooks' rules are NOT
    returned — the project decided not to inherit them. 0 = the user-wide
    set, which is what a session sees before a project is in scope.
    """
    async with async_session() as session:
        q = (
            select(Rule)
            .join(RulebookTopic, Rule.topic_id == RulebookTopic.id)
            .join(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
            .where(
                Rulebook.owner_user_id == user_id,
                Rulebook.always_on.is_(True),
                Rule.deleted_at.is_(None),
                RulebookTopic.deleted_at.is_(None),
                Rulebook.deleted_at.is_(None),
                # TIER (milestone 307). This is the SESSION-START call, made
                # before any project is in scope — there is no area vocabulary
                # to match a conditional rule against yet, so only the
                # unconditional tier belongs here. A conditional rule reaches a
                # session through enter_project (by area) or search (by
                # meaning), not by being resident.
                #
                # Behaviour is unchanged until rules are actually re-tiered:
                # `tier` defaults to always_on, so every existing rule still
                # arrives exactly as it did.
                Rule.tier == "always_on",
            )
        )
        if project_id:
            q = q.where(Rulebook.id.notin_(_excluded_rulebook_ids_q(project_id)))
        result = await session.execute(
            q.order_by(
                Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
            ).limit(limit)
        )
        return list(result.scalars().all())
 async def _fetch_owned_rule(session, rule_id: int, user_id: int) -> Optional[Rule]:
    """Fetch a rule by id, scoped to user owning either its rulebook
    (via topic) or its project (via project_id). Honors soft-delete.
@@ -780,7 +683,7 @@ async def update_rule(
            return None
        allowed = {
            "title", "statement", "why", "how_to_apply", "order_index",
-            "when_to_apply", "tier", "kind", "arose_from_id",
+            "when_to_apply", "kind", "arose_from_id",
            "verify_with", "expires_when",
        }
        check_before = rule.verify_with
@@ -796,9 +699,7 @@ async def update_rule(
        for key, value in fields.items():
            if key not in allowed or value is None:
                continue
-            if key == "tier":
+            if key == "kind":
                value = _valid_tier(value)
            elif key == "kind":
                value = _valid_kind(value)
            elif key in NULLABLE_RULE_TEXT:
                value = value or None
@@ -808,9 +709,8 @@ async def update_rule(
        # A verification stamp certifies A CHECK, not a rule. Rewrite or
        # remove the check and the old stamp certifies something that no
        # longer exists — so it is dropped, and the rule re-enters the sweep.
-        # The safe direction, for the same reason _valid_tier falls back to
+        # The safe direction: a rule wrongly listed as due costs one look, a
-        # always_on: a rule wrongly listed as due costs one look, a rule
+        # rule wrongly vouched for costs the thing the sweep exists to catch.
        # wrongly vouched for costs the thing the sweep exists to catch.
        if rule.verify_with != check_before:
            rule.verified_at = None
        # Same session as the edit, so the two commit together. The snapshot
@@ -1113,51 +1013,6 @@ async def unsuppress_rule_for_project(
        await session.commit()
 async def exclude_always_on_rulebook_for_project(
    project_id: int, rulebook_id: int, user_id: int,
 ) -> None:
    """Opt one project out of a whole ALWAYS-ON rulebook (milestone 297).
    Owner-only on both sides; the rulebook must be always_on — a subscribed
    rulebook is left by unsubscribing, not excluding. Idempotent."""
    from scribe.models.rulebook import project_rulebook_exclusions
    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await _assert_rulebook_owned(session, rulebook_id, user_id)
        rb = await session.get(Rulebook, rulebook_id)
        if rb is None or not rb.always_on:
            raise ValueError(
                f"rulebook {rulebook_id} is not always-on — it binds only by "
                "subscription; unsubscribe_project_from_rulebook instead"
            )
        try:
            await session.execute(
                insert(project_rulebook_exclusions).values(
                    project_id=project_id, rulebook_id=rulebook_id,
                )
            )
            await session.commit()
        except IntegrityError:
            await session.rollback()  # already excluded — idempotent
 async def include_always_on_rulebook_for_project(
    project_id: int, rulebook_id: int, user_id: int,
 ) -> None:
    """Undo exclude_always_on_rulebook_for_project. Idempotent."""
    from scribe.models.rulebook import project_rulebook_exclusions
    async with async_session() as session:
        await _assert_project_owned(session, project_id, user_id)
        await session.execute(
            sql_delete(project_rulebook_exclusions).where(
                project_rulebook_exclusions.c.project_id == project_id,
                project_rulebook_exclusions.c.rulebook_id == rulebook_id,
            )
        )
        await session.commit()
 async def suppress_topic_for_project(
    project_id: int, topic_id: int, user_id: int,
 ) -> None:
@@ -1326,7 +1181,6 @@ async def get_applicable_rules(
                Rulebook.deleted_at.is_(None),
                # An inception exclusion is total (milestone 297): a rulebook the
                # project opted out of contributes nothing, subscribed or not.
                Rulebook.id.notin_(_excluded_rulebook_ids_q(project_id)),
            )
            .order_by(
                Rulebook.id, RulebookTopic.order_index, Rule.order_index, Rule.title,
@@ -1337,11 +1191,10 @@ async def get_applicable_rules(
            rules_q = rules_q.where(Rule.id.notin_(suppressed_rule_ids))
        if suppressed_topic_ids:
            rules_q = rules_q.where(Rule.topic_id.notin_(suppressed_topic_ids))
-        # TIER (milestone 307). always_on rules are resident, as every rule was
+        # AREA BINDING (milestone 307, narrowed by 394). A rule reaches this
-        # before tiers existed. A conditional rule is REACHABLE, and reaches
+        # project when it is tagged to an area the project actually works in —
-        # this project only when it is tagged to an area this project actually
+        # a deterministic tag match, never a similarity score, so bindingness
-        # works in — a deterministic tag match, never a similarity score, so
+        # never depends on a ranking (D7).
        # bindingness never depends on a ranking (D7).
        #
        # Applied in SQL rather than by filtering afterwards, so `limit` counts
        # the rules that will actually be surfaced instead of counting rules
@@ -1357,10 +1210,21 @@ async def get_applicable_rules(
        reachable = select(rule_systems.c.rule_id).where(
            rule_systems.c.canonical_id.in_(project_area_ids)
        ) if project_area_ids else None
-        tier_clause = (Rule.tier == "always_on")
+        # AREA REACHABILITY IS NOW THE WHOLE TEST (milestone 394). This read
-        if reachable is not None:
+        # `always_on OR reachable`, so a subscribed rulebook's resident rules
-            tier_clause = or_(tier_clause, Rule.id.in_(reachable))
+        # arrived here whatever the project did. The tier is gone, and
-        rules_q = rules_q.where(tier_clause)
+        # dropping its arm rather than the whole clause is the deliberate
        # half: what survives is the DETERMINISTIC one — a rule binds this
        # project because it is tagged to an area the project actually works
        # in (D7), never because a similarity score cleared a bar.
        #
        # A project with no canonical-tagged Systems therefore gets no bulk
        # rules here, and that is the reading rather than a gap: rules still
        # reach it by retrieval, when something it is doing makes one
        # relevant. Handing over every subscribed rule instead would make this
        # payload BIGGER than the preload this milestone exists to remove.
        rules_q = (rules_q.where(Rule.id.in_(reachable)) if reachable is not None
                   else rules_q.where(sa_false()))
        rule_rows = (await session.execute(rules_q)).all()
        truncated = len(rule_rows) > limit
        rules = [
@@ -1381,12 +1245,11 @@ async def get_applicable_rules(
            )
            .order_by(Rule.order_index, Rule.title)
        )
-        if reachable is not None:
+        # A PROJECT'S OWN RULES ARE NOT FILTERED BY AREA, and the asymmetry
-            proj_rules_q = proj_rules_q.where(
+        # with the family query above is the point. A family rule has to earn
-                or_(Rule.tier == "always_on", Rule.id.in_(reachable))
+        # its way into this project; a rule written ON this project is scoped
-            )
+        # to it by construction, and filtering it again would drop rules whose
-        else:
+        # only fault is that nobody tagged them to a System.
            proj_rules_q = proj_rules_q.where(Rule.tier == "always_on")
        proj_rule_rows = (await session.execute(proj_rules_q)).all()
        project_rules = [rule_brief(rule) for (rule,) in proj_rule_rows]
@@ -1422,7 +1285,6 @@ async def get_applicable_rules(
        "suppressed_topics": suppressed_topics,
        "truncated": truncated,
        "subscribed_rulebooks": subscribed_rulebooks,
        "excluded_always_on": await excluded_always_on_rulebooks(user_id, project_id),
    }
@@ -1434,9 +1296,6 @@ def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict
    same seven keys under the same names — so a reader learns them once. One
    place renames `rules` → `applicable_rules` and `truncated` →
    `applicable_rules_truncated`; the tools merge this into their payloads.
    `excluded_always_on` (milestone 297) names the always-on rulebooks this
    project decided NOT to inherit, so the departure is visible wherever the
    rules are.
    IT ALSO RECORDS THE SURFACING, which is why it now takes a caller and a
    source. Every one of those surfaces is a bulk delivery — the applicable set
@@ -1453,7 +1312,7 @@ def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict
    Emitting from here is safe in a way emitting from `get_applicable_rules`
    would not be: this function is only ever called to BUILD A REPLY. The two
    other callers of the rules machinery — the write-path etag arm
-    (`plugin_context`) and `rules_etag_for` — compute a marker and show nobody
+    (`plugin_context`) — computed a marker and showed nobody
    anything, and counting those would put rules in the denominator that no
    agent ever saw.
    """
@@ -1472,7 +1331,6 @@ def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict
        "project_rules": applicable.get("project_rules", []),
        "suppressed_rules": applicable.get("suppressed_rules", []),
        "suppressed_topics": applicable.get("suppressed_topics", []),
        "excluded_always_on": applicable.get("excluded_always_on", []),
    }
@@ -1497,86 +1355,11 @@ def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict
 _ETAG_EMPTY = "empty|0"
 def rules_etag(rules: list) -> str:
    """A marker for "is the set you are holding still the current one?".
    `max(updated_at)` alone is not enough: DELETING a rule moves no timestamp,
    and that is the single change that takes an instruction OUT of force —
    the one a session most needs to hear about. The count catches it.
    Instance-agnostic (rule 115): it knows nothing about any particular
    rulebook, and an install with one rule or none produces a stable marker
    rather than an error. "No rules" must read as a state, not as a change,
    or every session on a fresh install would be told its rules had moved.
    """
    if not rules:
        return _ETAG_EMPTY
    # A decoration must not be able to break what it decorates. This is
    # computed on the SessionStart path, where raising would cost the whole
    # context payload to save a hint — so a row with no usable timestamp is
    # skipped rather than compared, and a set with none degrades to a
    # count-only marker instead of failing. Count-only still catches a rule
    # added or deleted; it just cannot see an edit, which is the right way
    # round to lose information.
    stamps = [
        r.updated_at for r in rules
        if isinstance(getattr(r, "updated_at", None), datetime)
    ]
    if not stamps:
        return f"unknown|{len(rules)}"
    return f"{max(stamps).isoformat()}|{len(rules)}"
 async def rules_etag_for(user_id: int, project_id: int = 0) -> str:
    """The current marker for the set a session at this scope would hold.
    Deliberately built from `list_always_on_rules` rather than from a
    `max()/count()` aggregate. An aggregate would be cheaper, and would have
    to restate that function's definition of the set — the always_on flag,
    the project's inception exclusions, the tier filter. Two definitions of
    "the session's rules" is how the marker starts disagreeing with the
    rules, which is worse than materialising a few dozen rows.
    """
    rules = await list_always_on_rules(user_id, project_id=project_id)
    return rules_etag(rules)
 def rules_moved_since(rules: list, held_etag: str) -> list:
    """The rules whose text changed after `held_etag` was issued.
    Returns [] when the marker matches, is unparseable, or is absent — a
    caller cannot act on "something is different but I cannot say what", and
    a garbled marker must not be reported as a change.
    A count difference is real news that this list cannot show: a rule
    DELETED since the marker was issued has no row left to return. Callers
    compare counts separately.
    """
    if not held_etag or held_etag == _ETAG_EMPTY:
        return []
    stamp, _, _count = held_etag.partition("|")
    try:
        held_at = datetime.fromisoformat(stamp)
    except ValueError:
        return []
    return [r for r in rules if r.updated_at and r.updated_at > held_at]
 def etag_count(held_etag: str) -> int | None:
    """How many rules the holder had. None when the marker cannot be read."""
    _stamp, _, count = (held_etag or "").partition("|")
    try:
        return int(count)
    except ValueError:
        return None
 # ── The staleness sweep (milestone 312) ────────────────────────────────
 async def rules_due_for_verification(
    user_id: int,
    older_than_days: int = 0,
    tier: str = "",
    never_only: bool = False,
 ) -> list[Rule]:
    """Rules that carry a check, oldest verification first, never-checked top.
@@ -1605,20 +1388,12 @@ async def rules_due_for_verification(
        older_than_days: only rules last verified longer ago than this.
            Never-checked rules always qualify — they are the most overdue
            thing there is. 0 = no age filter.
        tier: "always_on" or "conditional" to narrow. Raises on anything else
            rather than falling back: _valid_tier's silent always_on default
            is right for a WRITE (the safe direction is to keep binding), and
            wrong for a FILTER, where it would quietly answer a different
            question than the one asked.
        never_only: only rules that have never been verified.
    """
    from datetime import datetime, timedelta, timezone
    from scribe.models.project import Project
    if tier and tier not in TIERS:
        raise ValueError(f"tier must be one of {TIERS}, got {tier!r}")
    async with async_session() as session:
        stmt = (
            select(Rule)
@@ -1641,8 +1416,6 @@ async def rules_due_for_verification(
                ),
            )
        )
        if tier:
            stmt = stmt.where(Rule.tier == tier)
        if never_only:
            stmt = stmt.where(Rule.verified_at.is_(None))
        elif older_than_days > 0:
@@ -1668,7 +1441,6 @@ def verification_row(rule: Rule) -> dict:
        "id": rule.id,
        "title": rule.title,
        "statement": rule.statement,
        "tier": rule.tier,
        "topic_id": rule.topic_id,
        "project_id": rule.project_id,
        "when_to_apply": rule.when_to_apply or "",