fix(smem-hygiene): remember the gate's refusals, not only its saves

src/surreal_memory/hooks/capture_state.py

@@ -83,6 +83,26 @@ def content_key(content: str) -> str:
     return hashlib.md5(norm.encode("utf-8")).hexdigest()
 
 
+def rejected_key(ck: str, min_score: int) -> str:
+    """Key marking a candidate the write gate REJECTED at ``min_score``.
+
+    Rejected candidates were never marked as seen, so every Stop/PreCompact
+    re-submitted them to the gate for as long as the transcript tail carried
+    them. Measured 2026-08-08: 499 auto decisions in 24h held only 134 distinct
+    contents (one fragment judged 36 times), inflating the gate's denominator
+    ~3.7x and making the observed accept rate look several times worse than it
+    was (0.39% on rows vs ~1.4% on distinct content).
+
+    Namespaced by the threshold **on purpose**: marking a rejection as a plain
+    seen-key would silence that content for the rest of the session even if the
+    threshold were lowered afterwards -- trading duplicate noise for silent
+    loss, which is the worse failure. Because the key embeds the score it was
+    judged against, changing ``auto_capture_min_score`` stops matching these
+    keys and every previously-rejected candidate gets a fresh hearing.
+    """
+    return f"rej{min_score}:{ck}"
+
+
 def _load_all() -> dict[str, Any]:
     path = _state_path()
     if not path.exists():

src/surreal_memory/hooks/pre_compact.py

@@ -129,7 +129,13 @@ async def flush_text(text: str, project_name: str | None = None) -> dict[str, An
     from surreal_memory.core.memory_types import MemoryType, Priority, TypedMemory
     from surreal_memory.engine.dedup.factory import build_dedup_pipeline
     from surreal_memory.engine.encoder import MemoryEncoder
-    from surreal_memory.hooks.capture_state import content_key, load_seen, mark_seen, session_key
+    from surreal_memory.hooks.capture_state import (
+        content_key,
+        load_seen,
+        mark_seen,
+        rejected_key,
+        session_key,
+    )
     from surreal_memory.mcp.auto_capture import analyze_text_for_memories
     from surreal_memory.safety.input_firewall import check_content
     from surreal_memory.safety.sensitive import auto_redact_content
@@ -186,9 +192,18 @@ async def flush_text(text: str, project_name: str | None = None) -> dict[str, An
         seen = load_seen(skey)
         captured_keys: list[str] = []
         duplicate_skipped = False
+        # Suppress both what we already stored and what the gate already turned
+        # down at this threshold -- shared seen-set with the Stop hook, so a
+        # rejection recorded there must silence the candidate here too.
+        gate_min_score = config.write_gate.auto_capture_min_score
         if seen:
             before = len(boosted)
-            boosted = [it for it in boosted if content_key(it["content"]) not in seen]
+            boosted = [
+                it
+                for it in boosted
+                if content_key(it["content"]) not in seen
+                and rejected_key(content_key(it["content"]), gate_min_score) not in seen
+            ]
             duplicate_skipped = len(boosted) < before
 
         if not boosted:
@@ -250,6 +265,9 @@ async def flush_text(text: str, project_name: str | None = None) -> dict[str, An
                             gate_result.rejection_reason,
                         )
                         gate_rejected = True
+                        # Remember the refusal, not just the save: without this the
+                        # next compaction re-submits the identical fragment.
+                        captured_keys.append(rejected_key(content_key(content), gate_min_score))
                         continue
 
                 redacted_content, matches, _ = auto_redact_content(

src/surreal_memory/hooks/stop.py

@@ -352,7 +352,13 @@ async def capture_text(text: str, project_name: str | None = None) -> dict[str,
     from surreal_memory.core.memory_types import MemoryType, Priority, TypedMemory
     from surreal_memory.engine.dedup.factory import build_dedup_pipeline
     from surreal_memory.engine.encoder import MemoryEncoder
-    from surreal_memory.hooks.capture_state import content_key, load_seen, mark_seen, session_key
+    from surreal_memory.hooks.capture_state import (
+        content_key,
+        load_seen,
+        mark_seen,
+        rejected_key,
+        session_key,
+    )
     from surreal_memory.mcp.auto_capture import analyze_text_for_memories
     from surreal_memory.safety.input_firewall import check_content
 
@@ -406,9 +412,18 @@ async def capture_text(text: str, project_name: str | None = None) -> dict[str,
         captured_keys: list[str] = []
         had_candidate = bool(eligible)
         duplicate_skipped = False
+        # Suppress both what we already stored and what the gate already turned
+        # down at this threshold -- otherwise a rejected fragment is re-judged on
+        # every turn for as long as it survives in the transcript tail.
+        gate_min_score = config.write_gate.auto_capture_min_score
         if seen and eligible:
             before = len(eligible)
-            eligible = [it for it in eligible if content_key(it["content"]) not in seen]
+            eligible = [
+                it
+                for it in eligible
+                if content_key(it["content"]) not in seen
+                and rejected_key(content_key(it["content"]), gate_min_score) not in seen
+            ]
             duplicate_skipped = len(eligible) < before
         # All originally-detected fragments were duplicates (none survived the
         # filter above) -- nothing left to even attempt the gate/encode loop,
@@ -464,6 +479,9 @@ async def capture_text(text: str, project_name: str | None = None) -> dict[str,
                             gate_result.rejection_reason,
                         )
                         gate_rejected = True
+                        # Remember the refusal, not just the save: without this the
+                        # next turn re-submits the identical fragment to the gate.
+                        captured_keys.append(rejected_key(content_key(content), gate_min_score))
                         continue
 
                 redacted_content, matches, _ = auto_redact_content(
@@ -510,7 +528,10 @@ async def capture_text(text: str, project_name: str | None = None) -> dict[str,
             summary = _extract_session_summary(text)
             if summary:
                 summary_had_candidate = True
-                if content_key(summary) in seen:
+                if (
+                    content_key(summary) in seen
+                    or rejected_key(content_key(summary), gate_min_score) in seen
+                ):
                     duplicate_skipped = True
                     summary = None  # type: ignore[assignment]
             if summary and len(summary) > 30:
@@ -542,6 +563,8 @@ async def capture_text(text: str, project_name: str | None = None) -> dict[str,
                             gate_result.rejection_reason,
                         )
                         gate_rejected = True
+                        # Key must be taken before `summary` is cleared below.
+                        captured_keys.append(rejected_key(content_key(summary), gate_min_score))
                         summary = None  # type: ignore[assignment]
 
                 if summary:

tests/unit/test_capture_state.py

@@ -14,6 +14,7 @@
     content_key,
     load_seen,
     mark_seen,
+    rejected_key,
     session_key,
 )
 
@@ -156,3 +157,56 @@ def test_mark_seen_never_raises_on_unwritable_dir(
         blocker.write_text("x", encoding="utf-8")
         monkeypatch.setenv("SURREAL_MEMORY_DIR", str(blocker / "nested"))
         mark_seen("any-session", ["hash1"])  # must not raise
+
+
+class TestRejectedKey:
+    """Gate refusals are remembered too -- but only for the threshold that refused.
+
+    Rejected candidates used to be left unmarked, so the hooks re-submitted them
+    to the gate on every invocation. Measured on production 2026-08-08: 499 auto
+    decisions in 24h carried only 134 distinct contents (one fragment judged 36
+    times), inflating the gate's denominator ~3.7x.
+    """
+
+    def test_rejected_key_differs_from_plain_key(self) -> None:
+        ck = content_key("Decision: c remains explicitly open and unresolved")
+        assert rejected_key(ck, 5) != ck, (
+            "a refusal must not be recorded as an acceptance -- otherwise lowering "
+            "the threshold could never revive the content"
+        )
+
+    def test_rejected_key_is_threshold_scoped(self) -> None:
+        ck = content_key("Insight: the harvester stalled on shard three")
+        assert rejected_key(ck, 5) != rejected_key(ck, 4)
+
+    def test_rejected_key_is_stable(self) -> None:
+        ck = content_key("Error: flush_batch named the remote file from a stale counter")
+        assert rejected_key(ck, 5) == rejected_key(ck, 5)
+
+    def test_refusal_suppresses_at_same_threshold(self, isolated_state_dir: Path) -> None:
+        ck = content_key("Decision: b is blocked awaiting human authorization")
+        mark_seen("sess-a", [rejected_key(ck, 5)])
+        seen = load_seen("sess-a")
+        assert rejected_key(ck, 5) in seen
+
+    def test_refusal_expires_when_threshold_changes(self, isolated_state_dir: Path) -> None:
+        """The whole point of scoping: a lowered bar must give a second hearing.
+
+        Marking a refusal as a plain seen-key would silence that content for the
+        rest of the session even after the operator lowered the threshold --
+        trading duplicate noise for silent loss, which is the worse failure.
+        """
+        ck = content_key("Decision: b is blocked awaiting human authorization")
+        mark_seen("sess-b", [rejected_key(ck, 5)])
+        seen = load_seen("sess-b")
+        assert ck not in seen
+        assert rejected_key(ck, 4) not in seen, (
+            "after the threshold moved 5->4 the candidate must be judged again"
+        )
+
+    def test_accepted_key_survives_threshold_change(self, isolated_state_dir: Path) -> None:
+        """Acceptance is unconditional -- it must not be revived by a threshold move."""
+        ck = content_key("Insight: rclone token rotation confirmed on every mount")
+        mark_seen("sess-c", [ck])
+        seen = load_seen("sess-c")
+        assert ck in seen