fix(rules): the staleness signal must not wait for prior art to match (#3244)

CI caught two things, and the first is the feature not working rather than a
test being wrong.

THE SIGNAL WAS GATED ON A COINCIDENCE. build_write_path_hint returns early
when no prior art, stamp, divergence or derive matched, and that guard sat
ABOVE the new arm — so a session whose rules had changed was told only if the
file it happened to be editing also matched something else. A staleness
signal that fires on that coincidence is not a staleness signal. The arm now
runs above the guard, collecting into its own list that `lines` is seeded
from, and the guard accounts for it.

The standing-rule arm (milestone 307) is deliberately LEFT below that guard,
and this is a finding rather than a fix: it has the same gating and probably
should not, but it runs a SEMANTIC search, so lifting it would put an
embedding query on every write in every session. That is a cost decision, not
a bug fix, and not this task's to make.

THE MARKER MUST NOT BREAK THE PAYLOAD IT DECORATES. rules_etag is computed on
the SessionStart path, where `max()` raising costs the whole context payload
— every rule title, the project, all of it — to save a hint. A row with no
usable timestamp is now skipped and a set with none degrades to a count-only
marker, which still catches a rule added or deleted and only loses edits.
That is the right way round to lose information. CI found it because
build_session_context's tests pass MagicMock rules and `max()` over those
raises TypeError.

Also: list_always_on_rules on an install with no always-on rulebooks returns
`rules_etag: "empty|0"`. Its exact-dict test is updated rather than loosened
— the key being present on an empty install is the behaviour, not noise.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/scribe/mcp/tools/milestones.py

@@ -137,11 +137,17 @@ async def delete_milestone(milestone_id: int) -> dict:
     """Move a milestone to the trash (recoverable). Its tasks go with it as one batch.
     Restore via restore(batch_id)."""
     uid = current_user_id()
+    # Read the title BEFORE the delete: afterwards the row is trashed and the
+    # confirmation could only echo the number back. A deletion the operator
+    # cannot recognise is one they cannot tell was the wrong one.
+    doomed = await milestones_svc.get_milestone(uid, milestone_id)
+    title = getattr(doomed, "title", "") if doomed else ""
     batch = await trash_svc.delete(uid, "milestone", milestone_id)
     if batch is None:
         raise ValueError(f"milestone {milestone_id} not found")
-    return {"deleted_batch_id": batch,
-            "message": f"Milestone {milestone_id} + its tasks moved to trash. Restore with restore('{batch}')."}
+    return {"deleted": milestone_id, "title": title, "deleted_batch_id": batch,
+            "message": f'Milestone {milestone_id} ("{title}") and its tasks '
+                       f"moved to trash. Restore with restore('{batch}')."}
 
 
 def register(mcp) -> None:

src/scribe/mcp/tools/notes.py

@@ -324,11 +324,17 @@ async def find_duplicate_records(kind: str = "note", threshold: float = 0.0) ->
 async def delete_note(note_id: int) -> dict:
     """Move a Scribe note to the trash (recoverable). Restore via restore(batch_id)."""
     uid = current_user_id()
+    # Read the title BEFORE the delete: afterwards the row is trashed and the
+    # confirmation could only echo the number back. A deletion the operator
+    # cannot recognise is one they cannot tell was the wrong one.
+    loaded = await notes_svc.get_note_for_user(uid, note_id)
+    title = getattr(loaded[0], "title", "") if loaded else ""
     batch = await trash_svc.delete(uid, "note", note_id)
     if batch is None:
         raise ValueError(f"note {note_id} not found")
-    return {"deleted_batch_id": batch,
-            "message": f"Note {note_id} moved to trash. Restore with restore('{batch}')."}
+    return {"deleted": note_id, "title": title, "deleted_batch_id": batch,
+            "message": f'Note {note_id} ("{title}") moved to trash. '
+                       f"Restore with restore('{batch}')."}
 
 
 async def notes_due_for_verification(

src/scribe/mcp/tools/rulebooks.py

@@ -122,8 +122,9 @@ async def delete_rulebook(rulebook_id: int, confirmed: bool = False) -> dict:
             "confirmed_required": True,
         }
     batch = await trash_svc.delete(uid, "rulebook", rulebook_id)
-    return {"deleted": rulebook_id, "deleted_batch_id": batch,
-            "message": f"Moved to trash. Restore with restore('{batch}')."}
+    return {"deleted": rulebook_id, "title": rb.title, "deleted_batch_id": batch,
+            "message": f'Rulebook {rulebook_id} ("{rb.title}") moved to trash. '
+                       f"Restore with restore('{batch}')."}
 
 
 # ── Topic CRUD ─────────────────────────────────────────────────────────
@@ -192,8 +193,9 @@ async def delete_topic(topic_id: int, confirmed: bool = False) -> dict:
             "confirmed_required": True,
         }
     batch = await trash_svc.delete(uid, "topic", topic_id)
-    return {"deleted": topic_id, "deleted_batch_id": batch,
-            "message": f"Moved to trash. Restore with restore('{batch}')."}
+    return {"deleted": topic_id, "title": topic.title, "deleted_batch_id": batch,
+            "message": f'Topic {topic_id} ("{topic.title}") moved to trash. '
+                       f"Restore with restore('{batch}')."}
 
 
 # ── Rule CRUD ──────────────────────────────────────────────────────────
@@ -602,8 +604,10 @@ async def rule_history(rule_id: int, version_id: int = 0) -> dict:
     versions = await rulebooks_svc.list_rule_versions(rule_id, uid)
     if versions is None:
         raise ValueError(f"rule {rule_id} not found")
+    rule = await rulebooks_svc.get_rule(rule_id, uid)
     return {
         "rule_id": rule_id,
+        "title": rule.title if rule else "",
         "versions": [v.to_dict(include_text=False) for v in versions],
         "total": len(versions),
         # Said in-band because an empty list is the ordinary case and reads
@@ -632,8 +636,9 @@ async def delete_rule(rule_id: int, confirmed: bool = False) -> dict:
             "confirmed_required": True,
         }
     batch = await trash_svc.delete(uid, "rule", rule_id)
-    return {"deleted": rule_id, "deleted_batch_id": batch,
-            "message": f"Moved to trash. Restore with restore('{batch}')."}
+    return {"deleted": rule_id, "title": rule.title, "deleted_batch_id": batch,
+            "message": f'Rule {rule_id} ("{rule.title}") moved to trash. '
+                       f"Restore with restore('{batch}')."}
 
 
 # ── Subscriptions ──────────────────────────────────────────────────────

src/scribe/mcp/tools/snippets.py

@@ -494,9 +494,13 @@ async def delete_snippet(snippet_id: int) -> dict:
     that should survive, prefer merge_snippets — that keeps the call sites.
     """
     uid = current_user_id()
+    # Read before deleting so the confirmation can NAME what went — an id
+    # alone leaves the operator unable to tell which snippet this was.
+    doomed = await snippets_svc.get_snippet(uid, snippet_id)
+    title = getattr(doomed, "title", "") if doomed else ""
     if not await snippets_svc.delete_snippet(uid, snippet_id):
         raise ValueError(f"snippet {snippet_id} not found")
-    return {"deleted": True, "id": snippet_id}
+    return {"deleted": True, "id": snippet_id, "title": title}
 
 
 async def merge_snippets(target_id: int, source_ids: list[int]) -> dict:

src/scribe/mcp/tools/tasks.py

@@ -364,11 +364,17 @@ async def delete_task(task_id: int) -> dict:
     """Move a Scribe task (or plan) to the trash (recoverable). Sub-tasks go with it.
     Restore via restore(batch_id)."""
     uid = current_user_id()
+    # Read the title BEFORE the delete: afterwards the row is trashed and the
+    # confirmation could only echo the number back. A deletion the operator
+    # cannot recognise is one they cannot tell was the wrong one.
+    loaded = await notes_svc.get_note_for_user(uid, task_id)
+    title = getattr(loaded[0], "title", "") if loaded else ""
     batch = await trash_svc.delete(uid, "task", task_id)
     if batch is None:
         raise ValueError(f"task {task_id} not found")
-    return {"deleted_batch_id": batch,
-            "message": f"Task {task_id} moved to trash. Restore with restore('{batch}')."}
+    return {"deleted": task_id, "title": title, "deleted_batch_id": batch,
+            "message": f'Task {task_id} ("{title}") moved to trash. '
+                       f"Restore with restore('{batch}')."}
 
 
 def register(mcp) -> None:

src/scribe/services/plugin_context.py

@@ -952,7 +952,74 @@ async def build_write_path_hint(
             derive = [d for d in found if d.get("key") not in skip]
         except Exception:
             logger.warning("write-time derive check failed", exc_info=True)
-    if not synced and not menu and not stamped and not divergence and not derive:
+    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
+    # 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
+    # otherwise:
+    #
+    #   what goes wrong                                    | caught?
+    #   ---------------------------------------------------|--------
+    #   another session edits a rule mid-flight             | yes
+    #   the session is misremembering a rule read hours ago | yes
+    #   compaction summarised the rules out of context      | NO
+    #
+    # The third is the most common and this is blind to it: the etag was in
+    # 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
+    # particular write happened to match no prior art — and this arm is one
+    # indexed query, only when the session actually sent a marker.
+    #
+    # The standing-rule arm further down is deliberately left on the far side
+    # of this guard: that one runs a SEMANTIC search, and moving it here would
+    # run an embedding query on every write in the session. Its gating is a
+    # separate question from this one (see the note on #3244).
+    if not staleness and not synced and not menu and not stamped and not divergence and not derive:
         return empty
 
     owners = await owner_names_for({
@@ -971,7 +1038,9 @@ async def build_write_path_hint(
     for marker, item in menu:
         rendered.append((item, marker, _owner_of(item), _foreign_language(item, target_lang)))
 
-    lines: list[str] = []
+    # Seeded with the staleness line, which is decided above the early
+    # return and so cannot wait for this list to exist.
+    lines: list[str] = list(staleness)
     sync_note_ids: list[int] = []
     if synced:
         # The sync framing (#2708). Deliberately imperative about the record —
@@ -1039,63 +1108,6 @@ async def build_write_path_hint(
     for arm, ids in by_arm.items():
         record_surfaced(user_id=user_id, note_ids=ids, source=arm)
 
-    # ── Have the rules moved under this session? (milestone 323) ───────
-    #
-    # THE CARRIER IS THE POINT. This hook already fires before a write — the
-    # 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
-    # otherwise:
-    #
-    #   what goes wrong                                    | caught?
-    #   ---------------------------------------------------|--------
-    #   another session edits a rule mid-flight             | yes
-    #   the session is misremembering a rule read hours ago | yes
-    #   compaction summarised the rules out of context      | NO
-    #
-    # The third is the most common and this is blind to it: the etag was in
-    # 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:
-                    lines.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)
-
     # ── Standing rules that may apply here (milestone 307) ──────────────
     #
     # A SUGGESTION, not a binding surface, and the distinction is the design

src/scribe/services/rulebooks.py

@@ -1449,8 +1449,20 @@ def rules_etag(rules: list) -> str:
     """
     if not rules:
         return _ETAG_EMPTY
-    newest = max(r.updated_at for r in rules)
-    return f"{newest.isoformat()}|{len(rules)}"
+    # 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: