feat(rules): a session is told when its rules move under it (#3244, milestone 323 step 5)

The rules payload carries a marker; the write-path hook hands it back; the
server says which rules moved. Nothing is said when nothing moved.

THE COUNT IS NOT DECORATION. max(updated_at) alone cannot see a DELETED rule
— it moves no timestamp — and that is the single change that takes an
instruction OUT of force, which is the one a session most needs to hear
about. The marker is `<max updated_at>|<count>`, and a deletion is reported
through the count because there is no row left to name.

THE HOOK IS THE CARRIER because it already fires before a write, which is the
moment acting on a stale rule costs something. One comparison, no payload,
no extra round trip.

WHERE THE MARKER IS CAPTURED, and it could not be anywhere else: the
SessionStart hook, from /api/plugin/context. The model also receives one from
list_always_on_rules, but a hook cannot see an MCP tool's result — so the
value the write path compares has to be stored where a shell script can
reach it. Keyed by session id in the state dir the prior-art hook already
uses, so "changed since" means since THIS session loaded its rules.

NOT ON rules_payload, against the task's letter. Those are applicable_rules —
a different, subscription-derived set. One key name over two sets is how a
comparison starts reporting phantom changes, and the write path compares
against the always-on set.

WHAT IT CANNOT SEE is stated in both the service and the write-path arm as a
table, because a reader who finds an etag will assume it covers staleness
generally:

  another session edits a rule mid-flight             | caught
  the session is misremembering a rule read hours ago | caught
  compaction summarised the rules out of context      | NOT caught

The third is the most common, and the marker 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. A test asserts
both modules still explain that.

Instance-agnostic (rule 115): an install with no rules produces a stable
marker rather than an error, and "no rules" reads as a state rather than as a
change. An unreadable or absent marker reports nothing — a signal that cries
wolf is worse than none, because it trains a reader to skip the line that
will one day be true. The arm fails open like every other arm on this hook.

The delivery is tested through the real build_write_path_hint rather than the
helper alone: the feature IS a line arriving in a session, and the arithmetic
being right proves nothing about that.

Live acceptance is deploy-gated and not yet recorded on the task.

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

tests/test_rules_etag.py

@@ -0,0 +1,226 @@
+"""The staleness marker on the rules payload (milestone 323 step 5).
+
+WHAT THE MARKER IS FOR: telling a session that the rules it is holding have
+MOVED since it loaded them. Not general staleness — the limitation is stated
+in services/rulebooks.py and in the write-path arm, and two tests here pin the
+cases that would otherwise be quietly lost.
+
+The two that matter most are both about NOT crying wolf. A marker that reports
+a change when nothing changed gets ignored within a day, and an ignored
+staleness signal is worse than none: it trains a reader to skip the line that
+will one day be true.
+"""
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scribe.services import rulebooks as svc
+
+NOW = datetime(2026, 8, 30, 12, 0, tzinfo=timezone.utc)
+LATER = NOW + timedelta(hours=1)
+
+
+def _rule(updated_at=NOW, rid=1, title="a rule"):
+    """A Rule-shaped stand-in. The marker only ever reads three attributes,
+    and a real model would need a session to build."""
+    return SimpleNamespace(id=rid, title=title, updated_at=updated_at)
+
+
+def test_the_same_set_produces_the_same_marker():
+    """The whole mechanism rests on this. If the marker moved on its own, every
+    write would report a change and the line would be noise by lunchtime."""
+    rules = [_rule(rid=1), _rule(rid=2, updated_at=LATER)]
+    assert svc.rules_etag(rules) == svc.rules_etag(list(reversed(rules))), (
+        "the marker depends on the ORDER rules come back in, so any query "
+        "whose sort changes would look like an edit"
+    )
+
+
+def test_an_edit_moves_the_marker():
+    before = svc.rules_etag([_rule(rid=1), _rule(rid=2)])
+    after = svc.rules_etag([_rule(rid=1), _rule(rid=2, updated_at=LATER)])
+    assert before != after
+
+
+def test_a_DELETED_rule_moves_the_marker():
+    """THE CASE max(updated_at) ALONE CANNOT SEE, and the reason the count is
+    in there. Deleting a rule moves no timestamp — and it is the single change
+    that takes an instruction OUT of force, which is the one a session most
+    needs to hear about."""
+    before = svc.rules_etag([_rule(rid=1), _rule(rid=2)])
+    after = svc.rules_etag([_rule(rid=1)])
+    assert before != after, (
+        "a deleted rule left the marker unchanged — the count is missing, and "
+        "the session would keep obeying an instruction that no longer exists"
+    )
+
+
+def test_no_rules_is_a_state_not_a_change():
+    """Rule 115: this has to behave on an install with no rules at all. `max()`
+    over an empty set raises; a marker that raised would take the whole write
+    path's hint down, and one that varied would tell every session on a fresh
+    install that its rules had changed."""
+    assert svc.rules_etag([]) == svc.rules_etag([])
+    assert svc.rules_etag([]) != svc.rules_etag([_rule()])
+
+
+def test_moved_since_names_only_what_actually_moved():
+    held = svc.rules_etag([_rule(rid=1), _rule(rid=2)])
+    current = [_rule(rid=1), _rule(rid=2, updated_at=LATER, title="reworded")]
+    moved = svc.rules_moved_since(current, held)
+    assert [r.id for r in moved] == [2]
+
+
+def test_a_matching_marker_names_nothing():
+    rules = [_rule(rid=1), _rule(rid=2)]
+    assert svc.rules_moved_since(rules, svc.rules_etag(rules)) == []
+
+
+def test_an_unreadable_marker_reports_no_change():
+    """A caller cannot act on "something differs but I cannot say what", and a
+    garbled marker must never be rendered as a change — that is the shape of a
+    signal that gets ignored."""
+    assert svc.rules_moved_since([_rule(updated_at=LATER)], "not-an-etag") == []
+    assert svc.rules_moved_since([_rule(updated_at=LATER)], "") == []
+    assert svc.etag_count("garbled") is None
+
+
+def test_the_empty_marker_reports_no_change():
+    """An install that had no rules and now has some: the count says so, and
+    this function has no timestamp to reason from. Silence here, not a claim."""
+    assert svc.rules_moved_since([_rule()], svc.rules_etag([])) == []
+
+
+def test_the_count_survives_the_round_trip():
+    assert svc.etag_count(svc.rules_etag([_rule(rid=1), _rule(rid=2)])) == 2
+    assert svc.etag_count(svc.rules_etag([])) == 0
+
+
+def test_the_limitation_is_stated_where_a_reader_will_be():
+    """A future reader who finds an etag will assume it covers staleness
+    generally. It does not — it is blind to compaction, which is the most
+    common case — and the SessionStart nudge is that case's only mechanism.
+
+    Pinned because the plausible mistake is retiring a nudge that works on the
+    strength of a signal that does not cover it, and the comment is the only
+    thing standing in the way.
+    """
+    import inspect
+
+    from scribe.services import plugin_context
+
+    for module in (svc, plugin_context):
+        src = inspect.getsource(module).lower()
+        assert "compaction" in src and "etag" in src, (
+            f"{module.__name__} no longer explains what the rules marker "
+            f"cannot see. Without it the next reader will treat an etag as a "
+            f"general staleness check and soften the SessionStart nudge."
+        )
+
+
+# ── The arm that delivers the message (milestone 323 step 5) ───────────
+#
+# The marker is worth nothing until a session is actually TOLD. These drive
+# the real `build_write_path_hint`, because the feature IS a line arriving in
+# a hook's output — a test of the helper alone would prove the arithmetic and
+# nothing about the delivery.
+
+
+def _quiet_write_path(pc, rules):
+    """Every other arm stubbed to silent, so the only line that can appear is
+    the one under test."""
+    return (
+        patch.object(pc, "get_writepath_config",
+                     AsyncMock(return_value={"enabled": True, "threshold": 0.6,
+                                             "top_k": 3})),
+        patch.object(pc.snippets_svc, "list_snippets",
+                     AsyncMock(return_value=([], 0))),
+        patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])),
+        patch.object(pc, "semantic_search_rules", AsyncMock(return_value=[])),
+        patch.object(pc, "record_retrieval", MagicMock()),
+        patch.object(pc, "record_surfaced", MagicMock()),
+        patch.object(pc, "owner_names_for", AsyncMock(return_value={})),
+        patch.object(pc.rulebooks_svc, "list_always_on_rules",
+                     AsyncMock(return_value=rules)),
+    )
+
+
+async def _hint(rules, held_etag):
+    from scribe.services import plugin_context as pc
+
+    patches = _quiet_write_path(pc, rules)
+    for p in patches:
+        p.start()
+    try:
+        out = await pc.build_write_path_hint(
+            1, "src/scribe/services/rulebooks.py", code="x" * 400,
+            rules_etag=held_etag,
+        )
+    finally:
+        for p in patches:
+            p.stop()
+    return out["context"]
+
+
+@pytest.mark.asyncio
+async def test_a_session_is_told_which_rule_moved():
+    """The delivery, end to end through the real hint builder. Naming the rule
+    is the point — "something changed" sends the reader to re-read everything,
+    which is the cost the marker was meant to avoid."""
+    held = svc.rules_etag([_rule(rid=1), _rule(rid=2, title="dev is home")])
+    current = [_rule(rid=1), _rule(rid=2, title="dev is home", updated_at=LATER)]
+
+    ctx = await _hint(current, held)
+    assert "changed since this session started" in ctx
+    assert "#2" in ctx and "dev is home" in ctx
+
+
+@pytest.mark.asyncio
+async def test_a_session_holding_the_current_rules_is_told_nothing():
+    """The one that keeps the signal worth reading. A line on every write is a
+    line nobody reads."""
+    rules = [_rule(rid=1), _rule(rid=2)]
+    ctx = await _hint(rules, svc.rules_etag(rules))
+    assert "changed since this session started" not in ctx
+
+
+@pytest.mark.asyncio
+async def test_a_session_that_sent_no_marker_is_told_nothing():
+    """An install whose hook never reached /api/plugin/context has nothing
+    stored. Absent must read as silence, not as a mismatch — otherwise the
+    first thing a new install hears is that its rules changed."""
+    ctx = await _hint([_rule(updated_at=LATER)], "")
+    assert "changed since this session started" not in ctx
+
+
+@pytest.mark.asyncio
+async def test_a_DELETED_rule_is_reported_even_though_it_has_no_row():
+    """The count arm. A deleted rule leaves nothing to name, and it is the
+    change that takes an instruction OUT of force — so "no longer in force"
+    has to be sayable without a row to say it about."""
+    held = svc.rules_etag([_rule(rid=1), _rule(rid=2)])
+    ctx = await _hint([_rule(rid=1)], held)
+    assert "no longer in force" in ctx
+
+
+@pytest.mark.asyncio
+async def test_the_arm_fails_open():
+    """A staleness hint must never break a write. Every other arm here fails
+    open for the same reason, and this one runs a query that can fail."""
+    from scribe.services import plugin_context as pc
+
+    patches = _quiet_write_path(pc, [])
+    for p in patches:
+        p.start()
+    try:
+        with patch.object(pc.rulebooks_svc, "list_always_on_rules",
+                          AsyncMock(side_effect=RuntimeError("database down"))):
+            out = await pc.build_write_path_hint(
+                1, "src/x.py", code="x" * 400, rules_etag="2026-01-01T00:00:00+00:00|3",
+            )
+    finally:
+        for p in patches:
+            p.stop()
+    assert "changed since this session started" not in out["context"]