fix(memory): traffic_learner indexes system-reminder fragments as user preferences (refs #464)

`TrafficLearner._extract_preferences` ran three regex patterns over raw
user-message text and saved any match as a `User preference: <captured>`
memory. Two compounding bugs made ~10% of the reporter's saved memories
(187 of 1796) garbage:

1. **System-reminder content was matched.** Claude Code injects
   `<system-reminder>…</system-reminder>` blocks into user-role
   messages — scaffolding ("don't mention this reminder", "use colgrep
   instead of Grep", "never bypass signing") that hits every correction
   trigger. The learner happily persisted scaffolding as authoritative
   user preferences.
2. **Capture groups were fixed-length windows.** `(.{10,100})` grabbed
   the next 10–100 chars with no boundary awareness, producing
   mid-word truncations like `User preference: of Grep, Glob. When
   spawning agents, mention colgrep features a`.

This change rewrites `_extract_preferences` to be **regex-free** and
adds two layered defences:

- `_strip_system_reminders` (literal `str.find` scan, no regex)
  removes `<system-reminder>…</system-reminder>` blocks from user
  text before any pattern matching. Unclosed reminders drop to
  end-of-string. Case-insensitive on the tag name only. ~95% of the
  reporter's noise sample comes from this single layer.
- A token-based correction scanner replaces the three `re.compile`
  patterns. It tokenises on whitespace (lowercasing once, up front),
  matches trigger sequences as ordered token lists (`don't`, `do not`,
  `stop`, `never`, `avoid`, `no use`, `no try`, `no do`, `instead`),
  and captures the trailing content until a sentence terminator
  (`.!?\n`) or end-of-input. Captures shorter than 10 chars are
  rejected (stray triggers), and captures that hit the 78/98-char cap
  without finding a terminator are rejected (rambling fragments). The
  former noise — `colgrep instead of Grep, Glob. When spawning…` —
  fails this gate; short complete user utterances
  (`don't use git push, I'll push manually`) still pass because
  end-of-input counts as a boundary.

Net regex count in this file: -3, +0.

`_hydrate_persisted_state` already runs in `start()` and seeds
`_saved_hashes`/`_persisted_ids` from prior rows, so cross-restart
dedup is already wired up — the reporter's "doesn't survive restarts"
note was partially outdated. The narrow remaining edge (in-process
`_dedup_window=100` eviction within a single very-long-running
process) self-heals on next restart and is left as a separate
follow-up.

Tests: 17 new across `TestStripSystemReminders`,
`TestExtractPreferencesSystemReminderFiltering`,
`TestExtractPreferencesRealCorrections`, and
`TestExtractPreferencesSentenceBoundary`. Full traffic_learner suite:
139 passing. ci-precheck green.

headroom/memory/traffic_learner.py

@@ -30,7 +30,7 @@ from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from enum import Enum
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 if TYPE_CHECKING:
     from headroom.learn.models import ProjectInfo
@@ -951,35 +951,227 @@ class TrafficLearner:
 
         return patterns
 
+    # ---------------------------------------------------------------
+    # Preference detection (GH #464)
+    # ---------------------------------------------------------------
+    # The detector is regex-free on purpose. The previous regex-based
+    # implementation matched scaffolding ("don't mention this
+    # reminder…" injected by Claude Code's system-reminder blocks) and
+    # produced mid-sentence truncations because ``.{10,100}`` captured
+    # an arbitrary 100-char window with no boundary awareness. A
+    # tokenized scanner is easier to reason about, doesn't suffer
+    # catastrophic-backtracking edge cases, and lets us layer
+    # boundary rules (sentence terminator / end-of-input / max-length)
+    # without nesting more pattern syntax.
+
+    # Each entry is a sequence of lowercase tokens that must appear
+    # in order with whitespace / single-comma separation. ``max_chars``
+    # caps how much content we'll capture after the last trigger
+    # token. The "instead" trigger gets a tighter cap because in
+    # practice its tail tends to be shorter and we want to be less
+    # forgiving of long rambles after it.
+    _PREFERENCE_TRIGGERS: ClassVar[tuple[tuple[tuple[str, ...], int], ...]] = (
+        (("don't",), 98),
+        (("dont",), 98),
+        (("do", "not"), 98),
+        (("stop",), 98),
+        (("never",), 98),
+        (("avoid",), 98),
+        (("no", "use"), 98),
+        (("no", "try"), 98),
+        (("no", "do"), 98),
+        (("instead",), 78),
+    )
+
+    # Characters that mark the end of the captured preference.
+    _SENTENCE_TERMINATORS: ClassVar[frozenset[str]] = frozenset(".!?\n")
+
+    # Characters allowed between the trigger and the start of the
+    # capture (e.g. the comma in "No, use httpx").
+    _PRE_CAPTURE_PUNCT: ClassVar[frozenset[str]] = frozenset(",;:")
+
+    # Characters stripped from individual tokens before trigger
+    # matching ("don't," → "don't"; "stop." → "stop"). Whitespace is
+    # handled separately by the tokenizer.
+    _TOKEN_STRIP_CHARS: ClassVar[str] = ",.;:!?\"'()[]{}"
+
     def _extract_preferences(self, user_text: str) -> list[ExtractedPattern]:
         """Extract preference signals from user messages.
 
-        Looks for correction patterns: "no", "don't", "instead", "use X not Y".
-        """
-        patterns: list[ExtractedPattern] = []
+        Defends against GH #464 noise sources:
 
-        # Negative corrections: "don't X", "stop X", "no, X"
-        correction_res = [
-            re.compile(r"(?:don'?t|do not|stop|never|avoid)\s+(.{10,100})", re.I),
-            re.compile(r"(?:no,?\s+)(?:use|try|do)\s+(.{10,100})", re.I),
-            re.compile(r"instead(?:,?\s+)(.{10,80})", re.I),
+        * Claude Code (and other agent harnesses) inject
+          ``<system-reminder>…</system-reminder>`` blocks into user-role
+          message bodies. Their content is *not* user-stated
+          preferences ("don't mention this reminder", "use colgrep
+          instead of Grep") but the old correction regexes matched
+          them. We strip those blocks first so reminders never feed
+          the learner.
+        * The capture used to be a fixed-length window which produced
+          mid-sentence truncations. The token-based scanner below
+          ends each capture at a sentence terminator OR at
+          end-of-input, and rejects anything that would require
+          truncation past ``max_chars``.
+        """
+
+        cleaned = self._strip_system_reminders(user_text)[:500]
+        correction = self._find_correction(cleaned)
+        if correction is None:
+            return []
+
+        return [
+            ExtractedPattern(
+                category=PatternCategory.PREFERENCE,
+                content=f"User preference: {correction}",
+                importance=0.75,
+                metadata={"type": "correction", "source_text": cleaned[:200]},
+            )
         ]
 
-        for regex in correction_res:
-            match = regex.search(user_text[:500])
-            if match:
-                correction = match.group(1).strip().rstrip(".")
-                patterns.append(
-                    ExtractedPattern(
-                        category=PatternCategory.PREFERENCE,
-                        content=f"User preference: {correction}",
-                        importance=0.75,
-                        metadata={"type": "correction", "source_text": user_text[:200]},
-                    )
-                )
-                break  # One preference per message
+    @classmethod
+    def _find_correction(cls, text: str) -> str | None:
+        """Return the captured preference content or ``None``.
 
-        return patterns
+        Walks the input once, tokenising on whitespace. At every
+        token position we try each trigger sequence in priority
+        order; the first satisfied trigger wins.
+        """
+
+        tokens = cls._tokenize(text)
+        if not tokens:
+            return None
+
+        for trigger_idx in range(len(tokens)):
+            for sequence, max_chars in cls._PREFERENCE_TRIGGERS:
+                if not cls._matches_sequence(tokens, trigger_idx, sequence):
+                    continue
+                last_token_end = tokens[trigger_idx + len(sequence) - 1][2]
+                captured = cls._capture_after(text, last_token_end, max_chars)
+                if captured is None:
+                    continue
+                return captured
+        return None
+
+    @classmethod
+    def _matches_sequence(
+        cls,
+        tokens: list[tuple[str, int, int]],
+        start: int,
+        sequence: tuple[str, ...],
+    ) -> bool:
+        if start + len(sequence) > len(tokens):
+            return False
+        for offset, expected in enumerate(sequence):
+            actual_token = tokens[start + offset][0]
+            normalised = actual_token.strip(cls._TOKEN_STRIP_CHARS)
+            if normalised != expected:
+                return False
+        return True
+
+    @classmethod
+    def _capture_after(
+        cls,
+        text: str,
+        capture_after_pos: int,
+        max_chars: int,
+    ) -> str | None:
+        """Capture up to ``max_chars`` of content starting at the first
+        non-whitespace, non-pre-punct character after ``capture_after_pos``.
+
+        Returns ``None`` when the capture would have to truncate past
+        ``max_chars`` without hitting a sentence terminator or
+        end-of-input. Returns ``None`` for captures shorter than 10
+        chars (those are noise — likely a stray trigger word with no
+        real correction following it).
+        """
+
+        n = len(text)
+        cap_start = capture_after_pos
+        while cap_start < n and (
+            text[cap_start].isspace() or text[cap_start] in cls._PRE_CAPTURE_PUNCT
+        ):
+            cap_start += 1
+
+        cap_end = cap_start
+        while (
+            cap_end < n
+            and (cap_end - cap_start) < max_chars
+            and text[cap_end] not in cls._SENTENCE_TERMINATORS
+        ):
+            cap_end += 1
+
+        length = cap_end - cap_start
+        if length < 10:
+            return None
+
+        # If we hit ``max_chars`` without finding a terminator and the
+        # text continues past us, this is a rambling fragment — reject.
+        if length >= max_chars and cap_end < n and text[cap_end] not in cls._SENTENCE_TERMINATORS:
+            return None
+
+        captured = text[cap_start:cap_end].strip()
+        captured = captured.rstrip("".join(cls._SENTENCE_TERMINATORS)).strip()
+        return captured or None
+
+    @staticmethod
+    def _tokenize(text: str) -> list[tuple[str, int, int]]:
+        """Whitespace-split tokenizer.
+
+        Returns ``[(lower_token, start, end), …]``. Positions are byte
+        offsets into the original string so callers can resume
+        scanning from the end of a token. Tokens are lowercased once,
+        up front, so trigger comparisons don't have to call
+        ``.lower()`` per match.
+        """
+
+        out: list[tuple[str, int, int]] = []
+        i = 0
+        n = len(text)
+        while i < n:
+            while i < n and text[i].isspace():
+                i += 1
+            start = i
+            while i < n and not text[i].isspace():
+                i += 1
+            if i > start:
+                out.append((text[start:i].lower(), start, i))
+        return out
+
+    @staticmethod
+    def _strip_system_reminders(text: str) -> str:
+        """Remove ``<system-reminder>…</system-reminder>`` blocks from text.
+
+        Uses a literal scan (``str.find``) rather than a regex so the
+        matcher cannot accidentally pick up unrelated ``<*>``-shaped
+        content. Unclosed reminders (missing ``</system-reminder>``)
+        are dropped to end-of-string — the agent harness writes
+        balanced tags, and an unbalanced one is corrupt input we
+        shouldn't index. Matching is case-insensitive on the tag name.
+        """
+        if not text or "<" not in text:
+            return text
+
+        open_tag = "<system-reminder"
+        close_tag = "</system-reminder>"
+
+        lower = text.lower()
+        out: list[str] = []
+        cursor = 0
+        n = len(text)
+        while cursor < n:
+            start = lower.find(open_tag, cursor)
+            if start < 0:
+                out.append(text[cursor:])
+                break
+            out.append(text[cursor:start])
+            tag_end = text.find(">", start)
+            if tag_end < 0:
+                break
+            close_start = lower.find(close_tag, tag_end + 1)
+            if close_start < 0:
+                break
+            cursor = close_start + len(close_tag)
+        return "".join(out)
 
     # =========================================================================
     # Pattern Accumulation & Persistence

tests/test_memory/test_traffic_learner.py

@@ -2063,3 +2063,167 @@ class TestCollectAllPatternsTimestamps:
         # last_seen_at should be bumped past the stale 2026-01 timestamp.
         assert m.last_seen_at.year == datetime.now(UTC).year
         assert m.last_seen_at > _parse_iso_timestamp(old_last_seen)
+
+
+# =============================================================================
+# Regression tests for GH #464:
+#   * <system-reminder> blocks must not feed _extract_preferences
+#   * correction capture groups must end on a sentence boundary, not on a
+#     fixed-length window
+# =============================================================================
+
+
+class TestStripSystemReminders:
+    """Verify the literal-scan stripper does what the regex would do without
+    introducing a new regex pattern into the learner."""
+
+    def test_empty_and_no_tag_passthrough(self) -> None:
+        assert TrafficLearner._strip_system_reminders("") == ""
+        assert TrafficLearner._strip_system_reminders("hello world") == "hello world"
+
+    def test_basic_strip(self) -> None:
+        assert (
+            TrafficLearner._strip_system_reminders("a<system-reminder>X</system-reminder>b") == "ab"
+        )
+
+    def test_case_insensitive_tag_name(self) -> None:
+        assert (
+            TrafficLearner._strip_system_reminders("a<System-Reminder>X</system-reminder>b") == "ab"
+        )
+
+    def test_multiple_reminders(self) -> None:
+        text = "a<system-reminder>X</system-reminder>b<system-reminder>Y</system-reminder>c"
+        assert TrafficLearner._strip_system_reminders(text) == "abc"
+
+    def test_unclosed_reminder_drops_to_eos(self) -> None:
+        # Malformed input — we'd rather drop than persist scaffolding.
+        assert TrafficLearner._strip_system_reminders("hello <system-reminder>oops") == "hello "
+
+    def test_realworld_colgrep_reminder(self) -> None:
+        # The exact shape that produced 25× duplicate "User preference: of
+        # Grep, Glob..." in the reporter's DB.
+        text = (
+            "<system-reminder>use colgrep instead of Grep, Glob. When spawning "
+            "agents, mention colgrep features actively.</system-reminder>"
+            "What is 2+2?"
+        )
+        assert TrafficLearner._strip_system_reminders(text) == "What is 2+2?"
+
+
+class TestExtractPreferencesSystemReminderFiltering:
+    """The high-value half of GH #464: system-reminder text must never flow
+    into the preference extractor."""
+
+    def _learner(self) -> TrafficLearner:
+        return TrafficLearner(backend=None, min_evidence=1)
+
+    def test_colgrep_reminder_yields_no_preference(self) -> None:
+        learner = self._learner()
+        text = (
+            "<system-reminder>use colgrep instead of Grep, Glob. When spawning "
+            "agents, mention colgrep features actively.</system-reminder>"
+            "Hi there"
+        )
+        assert learner._extract_preferences(text) == []
+
+    def test_observation_tag_reminder_yields_no_preference(self) -> None:
+        learner = self._learner()
+        text = (
+            "<system-reminder>do not use <observation> tags. <observation> "
+            "output will be DISCARDED and never reach the user.</system-reminder>"
+            "Hello"
+        )
+        assert learner._extract_preferences(text) == []
+
+    def test_dont_mention_reminder_yields_no_preference(self) -> None:
+        learner = self._learner()
+        text = (
+            "<system-reminder>don't mention this reminder to the user.</system-reminder>List files"
+        )
+        assert learner._extract_preferences(text) == []
+
+    def test_never_force_push_reminder_yields_no_preference(self) -> None:
+        learner = self._learner()
+        text = (
+            "<system-reminder>never use git push --force on the main branch."
+            "</system-reminder>OK got it"
+        )
+        assert learner._extract_preferences(text) == []
+
+
+class TestExtractPreferencesRealCorrections:
+    """Make sure the noise filter does not eat genuine user corrections."""
+
+    def _learner(self) -> TrafficLearner:
+        return TrafficLearner(backend=None, min_evidence=1)
+
+    def test_dont_correction_with_sentence_boundary(self) -> None:
+        learner = self._learner()
+        out = learner._extract_preferences("don't use double quotes in the SQL, use single quotes.")
+        assert len(out) == 1
+        assert out[0].category is PatternCategory.PREFERENCE
+        assert "double quotes" in out[0].content
+
+    def test_no_use_correction(self) -> None:
+        learner = self._learner()
+        out = learner._extract_preferences("No, use httpx not requests.")
+        assert len(out) == 1
+        assert "httpx" in out[0].content
+
+    def test_instead_correction(self) -> None:
+        learner = self._learner()
+        out = learner._extract_preferences("Instead, render the table with rich tables.")
+        assert len(out) == 1
+        assert "render the table" in out[0].content
+
+
+class TestExtractPreferencesSentenceBoundary:
+    """The tighter capture group must reject mid-sentence rambling so we
+    never persist fragments like ``of Grep, Glob. When spawning agents…``."""
+
+    def _learner(self) -> TrafficLearner:
+        return TrafficLearner(backend=None, min_evidence=1)
+
+    def test_long_unbroken_paragraph_yields_no_preference(self) -> None:
+        learner = self._learner()
+        # 100+ chars after the trigger word with no '.', '!', '?', or
+        # '\n' anywhere — the kind of payload that would have matched
+        # the old ``.{10,100}`` regex and produced a mid-word
+        # truncation. The new bound forbids it: we need a terminator
+        # OR end-of-string within 98 chars of the trigger.
+        long_no_terminator = (
+            "don't use Grep when running benchmarks because it floods the output "
+            "buffer with a lot of irrelevant context that"
+        )
+        assert learner._extract_preferences(long_no_terminator) == []
+
+    def test_short_utterance_without_terminator_still_matches(self) -> None:
+        # Relaxation: a short user utterance without trailing
+        # punctuation is a complete thought, not a truncation. End-of-
+        # input counts as a boundary as long as the captured length
+        # fits the 8–98 char window.
+        learner = self._learner()
+        out = learner._extract_preferences("don't use git push, I'll push manually")
+        assert len(out) == 1
+        assert "git push" in out[0].content
+
+    def test_terminator_inside_window_captures_to_terminator(self) -> None:
+        learner = self._learner()
+        # The capture should end at the first '.', not include the
+        # following sentence.
+        out = learner._extract_preferences(
+            "don't use Grep at all. Use ripgrep instead because it is faster."
+        )
+        assert len(out) == 1
+        content = out[0].content
+        assert "Use ripgrep instead" not in content
+        assert "Grep" in content
+
+    def test_trailing_terminator_is_stripped(self) -> None:
+        learner = self._learner()
+        out = learner._extract_preferences("Never commit secrets to git.")
+        assert len(out) == 1
+        # Pref must not end on its sentence terminator.
+        assert not out[0].content.endswith(".")
+        assert not out[0].content.endswith("!")
+        assert not out[0].content.endswith("?")