feat(write-gate): shadow/enforce modes, per-intent auto-capture mode, and gate_decision telemetry

src/surreal_memory/engine/gate_telemetry.py

@@ -0,0 +1,65 @@
+"""Write-gate telemetry — append engine-side gate decisions to the
+`gate_decision` table (the same SCHEMALESS table the Hermes plugin writes to),
+so scripts/gate_stats.py sees BOTH the plugin and the engine (smem
+auto-capture / stop-hook) decisions in one report.
+
+Fire-and-forget: logging must never break a write.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+def _lang_hint(s: str) -> str:
+    """'cjk' if the text contains CJK / Kana / Hangul, else 'latin' (no lib)."""
+    for ch in s:
+        o = ord(ch)
+        if (
+            0x4E00 <= o <= 0x9FFF
+            or 0x3400 <= o <= 0x4DBF
+            or 0x3040 <= o <= 0x30FF
+            or 0xAC00 <= o <= 0xD7A3
+        ):
+            return "cjk"
+    return "latin"
+
+
+async def log_gate_decision(
+    storage: Any,
+    *,
+    intent: str,
+    accepted: bool,
+    reason: str,
+    score: int | None,
+    mode: str,
+    content: str,
+    agent_id: str = "",
+) -> None:
+    """Best-effort insert of one gate decision (both SHADOW and ENFORCE log)."""
+    try:
+        from surreal_memory.utils.timeutils import utcnow
+
+        conn = storage._ensure_conn()
+        s = (content or "").strip()
+        await conn.insert(
+            "gate_decision",
+            {
+                "ts": utcnow(),
+                "intent": intent,
+                "accepted": accepted,
+                "score": score,
+                "reason": reason,
+                "preview": s[:80],
+                "length": len(s),
+                "lang_hint": _lang_hint(s),
+                "agent_id": agent_id,
+                "mode": mode,
+                "source": "engine",
+            },
+        )
+    except Exception:
+        logger.debug("gate_decision log failed (non-fatal)", exc_info=True)

src/surreal_memory/hooks/pre_compact.py

@@ -182,10 +182,44 @@ async def flush_text(text: str, project_name: str | None = None) -> dict[str, An
         auto_redact_severity = config.safety.auto_redact_min_severity
         saved: list[str] = []
 
+        # Write gate for the compaction flush (intent=auto). This path encoded
+        # with NO gate at all, so junk auto-captures bypassed both telemetry
+        # and enforcement — same contract as the stop-hook auto block now.
+        write_gate_cfg = config.write_gate
+        gate_mode = write_gate_cfg.effective_auto_mode
+
         for item in boosted:
             try:
                 # Auto-redact sensitive content
                 content = item["content"]
+
+                if gate_mode != "off":
+                    from surreal_memory.engine.gate_telemetry import log_gate_decision
+                    from surreal_memory.engine.quality_scorer import check_write_gate
+
+                    gate_result = check_write_gate(
+                        content,
+                        gate_config=write_gate_cfg,
+                        is_auto_capture=True,
+                        memory_type=item.get("type"),
+                        tags=list(base_tags),
+                    )
+                    await log_gate_decision(
+                        storage,
+                        intent="auto",
+                        accepted=not gate_result.rejected,
+                        reason=gate_result.rejection_reason or "accept",
+                        score=gate_result.score,
+                        mode=gate_mode,
+                        content=content,
+                    )
+                    if gate_mode == "enforce" and gate_result.rejected:
+                        logger.debug(
+                            "Pre-compact write gate rejected: %s",
+                            gate_result.rejection_reason,
+                        )
+                        continue
+
                 redacted_content, matches, _ = auto_redact_content(
                     content, min_severity=auto_redact_severity
                 )

src/surreal_memory/hooks/stop.py

@@ -376,25 +376,38 @@ async def capture_text(text: str, project_name: str | None = None) -> dict[str,
         auto_redact_severity = config.safety.auto_redact_min_severity
         saved: list[str] = []
 
-        # Write gate check for stop hook (auto-capture path)
+        # Write gate check for stop hook (auto-capture path). SHADOW logs the
+        # decision to gate_decision without blocking; ENFORCE rejects. The
+        # stop hook is intent=auto, so it resolves auto_capture_mode.
         write_gate_cfg = config.write_gate
-        gate_enabled = write_gate_cfg.enabled
+        gate_mode = write_gate_cfg.effective_auto_mode
 
         for item in eligible:
             try:
                 content = item["content"]
 
-                # Apply write gate if enabled (uses auto_capture threshold)
-                if gate_enabled:
+                # Apply write gate unless off (uses auto_capture threshold)
+                if gate_mode != "off":
+                    from surreal_memory.engine.gate_telemetry import log_gate_decision
                     from surreal_memory.engine.quality_scorer import check_write_gate
 
                     gate_result = check_write_gate(
                         content,
                         gate_config=write_gate_cfg,
                         is_auto_capture=True,
                         memory_type=item.get("type"),
+                        tags=list(base_tags),
                     )
-                    if gate_result.rejected:
+                    await log_gate_decision(
+                        storage,
+                        intent="auto",
+                        accepted=not gate_result.rejected,
+                        reason=gate_result.rejection_reason or "accept",
+                        score=gate_result.score,
+                        mode=gate_mode,
+                        content=content,
+                    )
+                    if gate_mode == "enforce" and gate_result.rejected:
                         logger.debug(
                             "Stop hook write gate rejected: %s",
                             gate_result.rejection_reason,
@@ -440,16 +453,28 @@ async def capture_text(text: str, project_name: str | None = None) -> dict[str,
             summary = _extract_session_summary(text)
             if summary and len(summary) > 30:
                 # Apply write gate to session summary too
-                if gate_enabled:
+                if gate_mode != "off":
+                    from surreal_memory.engine.gate_telemetry import log_gate_decision
                     from surreal_memory.engine.quality_scorer import check_write_gate
 
+                    summary_gate_tags = list(set(base_tags) | {"session_summary"})
                     gate_result = check_write_gate(
                         summary,
                         gate_config=write_gate_cfg,
                         is_auto_capture=True,
                         memory_type="context",
+                        tags=summary_gate_tags,
+                    )
+                    await log_gate_decision(
+                        storage,
+                        intent="auto",
+                        accepted=not gate_result.rejected,
+                        reason=gate_result.rejection_reason or "accept",
+                        score=gate_result.score,
+                        mode=gate_mode,
+                        content=summary,
                     )
-                    if gate_result.rejected:
+                    if gate_mode == "enforce" and gate_result.rejected:
                         logger.debug(
                             "Stop hook session summary rejected: %s",
                             gate_result.rejection_reason,

src/surreal_memory/mcp/remember_handler.py

@@ -218,9 +218,16 @@ async def _remember(self, args: dict[str, Any]) -> dict[str, Any]:
                     logger.error("Encryption failed, refusing to store plaintext", exc_info=True)
                     return {"error": "Encryption failed — memory not stored. Check encryption key."}
 
-        # Write gate: reject low-quality content before encoding
+        # Write gate: score content; SHADOW logs the decision to gate_decision
+        # without blocking, ENFORCE rejects. Auto-captures resolve their own
+        # mode (auto_capture_mode) so junk can be enforced while manual writes
+        # stay in shadow.
         write_gate_cfg = self.config.write_gate
-        if write_gate_cfg.enabled is True:
+        gate_mode = (
+            write_gate_cfg.effective_auto_mode if is_auto_capture else write_gate_cfg.effective_mode
+        )
+        if gate_mode != "off":
+            from surreal_memory.engine.gate_telemetry import log_gate_decision
             from surreal_memory.engine.quality_scorer import check_write_gate
 
             gate_result = check_write_gate(
@@ -231,9 +238,18 @@ async def _remember(self, args: dict[str, Any]) -> dict[str, Any]:
                 tags=args.get("tags"),
                 context=args.get("context") if isinstance(args.get("context"), dict) else None,
             )
-            if gate_result.rejected:
+            await log_gate_decision(
+                storage,
+                intent="auto" if is_auto_capture else "manual",
+                accepted=not gate_result.rejected,
+                reason=gate_result.rejection_reason or "accept",
+                score=gate_result.score,
+                mode=gate_mode,
+                content=content,
+            )
+            if gate_mode == "enforce" and gate_result.rejected:
                 logger.debug(
-                    "Write gate rejected: %s (score=%d)",
+                    "Write gate rejected: %s (score=%s)",
                     gate_result.rejection_reason,
                     gate_result.score,
                 )

src/surreal_memory/unified_config.py

@@ -441,16 +441,44 @@ class WriteGateConfig:
     Addresses GitHub Issue #95: write-gate to improve brain purity.
     """
 
-    enabled: bool = False  # opt-in, backward compat
+    enabled: bool = False  # opt-in, backward compat (True == enforce)
+    mode: str = "off"  # off | shadow | enforce. Overrides `enabled` when not "off".
+    # Per-intent override for auto-captures (intent=auto ONLY; summaries and
+    # turns keep `mode`). "" = inherit `mode`. Lets junk auto-captures be
+    # ENFORCED while interactive writes stay in shadow — a global enforce is
+    # known to false-reject real turn/summary content.
+    auto_capture_mode: str = ""  # "" (inherit) | off | shadow | enforce
     min_length: int = 30  # reject content shorter than this
     min_quality_score: int = 3  # reject score below this (0-10 scale)
     auto_capture_min_score: int = 5  # stricter threshold for passive captures
     max_content_length: int = 2000  # reject wall-of-text above this
     reject_generic_filler: bool = True  # reject "done", "ok", "completed" etc.
 
+    @property
+    def effective_mode(self) -> str:
+        """Resolve the operating mode. `mode` wins; otherwise fall back to the
+        legacy `enabled` bool (True -> enforce, False -> off)."""
+        m = (self.mode or "off").strip().lower()
+        if m in ("shadow", "enforce"):
+            return m
+        if m == "off" and self.enabled:
+            return "enforce"
+        return "off"
+
+    @property
+    def effective_auto_mode(self) -> str:
+        """Resolve the mode for auto-captures (intent=auto). A valid
+        `auto_capture_mode` wins; otherwise inherit `effective_mode`."""
+        m = (self.auto_capture_mode or "").strip().lower()
+        if m in ("off", "shadow", "enforce"):
+            return m
+        return self.effective_mode
+
     def to_dict(self) -> dict[str, Any]:
         return {
             "enabled": self.enabled,
+            "mode": self.mode,
+            "auto_capture_mode": self.auto_capture_mode,
             "min_length": self.min_length,
             "min_quality_score": self.min_quality_score,
             "auto_capture_min_score": self.auto_capture_min_score,
@@ -462,6 +490,8 @@ def to_dict(self) -> dict[str, Any]:
     def from_dict(cls, data: dict[str, Any]) -> WriteGateConfig:
         return cls(
             enabled=bool(data.get("enabled", False)),
+            mode=str(data.get("mode", "off")),
+            auto_capture_mode=str(data.get("auto_capture_mode", "")),
             min_length=int(data.get("min_length", 30)),
             min_quality_score=int(data.get("min_quality_score", 3)),
             auto_capture_min_score=int(data.get("auto_capture_min_score", 5)),
@@ -1915,6 +1945,8 @@ def save(self) -> None:
             "# Write gate (quality enforcement before storage)",
             "[write_gate]",
             f"enabled = {'true' if self.write_gate.enabled else 'false'}",
+            f'mode = "{self.write_gate.mode}"',
+            f'auto_capture_mode = "{self.write_gate.auto_capture_mode}"',
             f"min_length = {self.write_gate.min_length}",
             f"min_quality_score = {self.write_gate.min_quality_score}",
             f"auto_capture_min_score = {self.write_gate.auto_capture_min_score}",

tests/unit/test_geo_mcp.py

@@ -15,7 +15,7 @@
 
 from surreal_memory.engine.retrieval_types import DepthLevel, RetrievalResult, Subgraph
 from surreal_memory.mcp.server import MCPServer
-from surreal_memory.unified_config import ResponseConfig, ToolTierConfig
+from surreal_memory.unified_config import ResponseConfig, ToolTierConfig, WriteGateConfig
 from surreal_memory.utils.geo import GeoFilter
 
 
@@ -31,7 +31,7 @@ def _make_server() -> MCPServer:
             tool_tier=ToolTierConfig(tier="full"),
             response=ResponseConfig(),
         )
-        cfg.write_gate.enabled = False
+        cfg.write_gate = WriteGateConfig()  # real config (mode="off") so the write-gate is skipped
         cfg.encryption.enabled = False
         cfg.safety.auto_redact_min_severity = 3
         mock_get_config.return_value = cfg