feat(engine): secret redaction mode, superseded detail, delta recall, enhanced auto routing

engraphis/core/engine.py

@@ -68,7 +68,7 @@
     ResolutionOp,
     resolve,
 )
-from engraphis.core.secrets import reject_secrets
+from engraphis.core.secrets import redact_secrets as _redact_secrets, reject_secrets
 from engraphis.core.store import (
     Store,
     _is_memory_database_path,
@@ -829,7 +829,8 @@ def remember(self, content: str, *, workspace_id: str, repo_id: Optional[str] =
                  valid_from: Optional[float] = None, resolve_conflicts: bool = True,
                  candidate_k: int = 5, subject_key: str = "", claim_kind: str = "",
                  _trusted_graph_keys: Optional[frozenset] = None,
-                 _transactional_finalizer: Optional[Callable[[str], None]] = None) -> str:
+                 _transactional_finalizer: Optional[Callable[[str], None]] = None,
+                 redact_secrets: bool = False) -> str:
         """Store one memory. Returns the resulting record id: a new id for ADD/
         INVALIDATE/quarantine, or the existing memory's id if this was resolved as a
         NOOP (near-duplicate). See ``remember_with_resolution`` for decision detail.
@@ -842,6 +843,7 @@ def remember(self, content: str, *, workspace_id: str, repo_id: Optional[str] =
             candidate_k=candidate_k, subject_key=subject_key, claim_kind=claim_kind,
             _trusted_graph_keys=_trusted_graph_keys,
             _transactional_finalizer=_transactional_finalizer,
+            redact_secrets=redact_secrets,
         )["id"]
 
     def remember_with_resolution(self, content: str, *, workspace_id: str,
@@ -855,7 +857,8 @@ def remember_with_resolution(self, content: str, *, workspace_id: str,
                  _trusted_graph_keys: Optional[frozenset] = None,
                  _approval_override: bool = False,
                  _transactional_finalizer: Optional[Callable[[str], None]] = None,
-                 extra_neighbors: Optional[list] = None) -> dict:
+                 extra_neighbors: Optional[list] = None,
+                 redact_secrets: bool = False) -> dict:
         """Store one memory with deterministic conflict resolution.
 
         Returns ``{"id", "op", ...}`` where ``op`` is one of:
@@ -871,6 +874,10 @@ def remember_with_resolution(self, content: str, *, workspace_id: str,
         * ``"quarantined"`` — an explicitly untrusted payload matched the deterministic
           poisoning policy; retained only for governed historical inspection.
         """
+        if redact_secrets:
+            content = _redact_secrets(content)
+            if title:
+                title = _redact_secrets(title)
         # Reject credentials before embedding, conflict resolution, graph extraction, or
         # any SQLite mirror sees them. Store.add_memory repeats this for direct callers.
         reject_secrets((("title", title), ("content", content), ("keywords", keywords),
@@ -1576,6 +1583,7 @@ def _resolve_and_store(self, content: str, *, text: str, vec: Optional[np.ndarra
             embedding=None if poisoning.quarantined else vec,
         )
         invalidating = decision is not None and decision.op == ResolutionOp.INVALIDATE
+        predecessor: Optional[MemoryRecord] = None
         try:
             # A replacement and the predecessor interval it closes are one authoritative
             # state transition. Portable vector/FTS mirrors participate in the same
@@ -1725,6 +1733,13 @@ def _resolve_and_store(self, content: str, *, text: str, vec: Optional[np.ndarra
             ) if trusted_write else []
             out = {"id": mid, "op": "invalidate", "superseded": [decision.target_id],
                    "reason": decision.reason}
+            if predecessor is not None:
+                out["superseded_detail"] = {
+                    "id": predecessor.id,
+                    "content_preview": predecessor.content[:120],
+                    "stability_days": predecessor.stability,
+                    "access_count": predecessor.access_count,
+                }
             if linked:
                 out["linked"] = linked
             return out

engraphis/core/interfaces.py

@@ -245,11 +245,13 @@ class SearchFilter:
     # Appended after every 1.x field so positional construction remains compatible.
     valid_at: Optional[float] = None
     known_at: Optional[float] = None
+    modified_since: Optional[float] = None
 
     def __post_init__(self) -> None:
         self.as_of = _finite_timestamp(self.as_of, "as_of")
         self.valid_at = _finite_timestamp(self.valid_at, "valid_at")
         self.known_at = _finite_timestamp(self.known_at, "known_at")
+        self.modified_since = _finite_timestamp(self.modified_since, "modified_since")
         if self.as_of is not None and self.valid_at is not None:
             if self.as_of != self.valid_at:
                 raise ValueError("as_of and valid_at must match when both are supplied")

engraphis/core/retrieval_policy.py

@@ -17,12 +17,12 @@
 CANDIDATE_DEPTH_MODES = frozenset({"fixed", "adaptive"})
 
 _CODE_RE = re.compile(
-    r"(?:\w+[./\\])+\w+|::|->|\b(?:class|def|function|import|module)\b|"
+    r"(?:\w+[./\\])+\w+|::|->|\b(?:class|def|function|import|interface|module|struct)\b|"
     r"\b[A-Za-z_]\w*\([^)]*\)",
     re.IGNORECASE,
 )
 _GRAPH_RE = re.compile(
-    r"\b(?:calls?|causes?|depends?|impact|path|related|relationship|why)\b",
+    r"\b(?:calls?|causes?|connected|connections?|depends?|impact|path|related|relationship|superseded|supersedes|why)\b",
     re.IGNORECASE,
 )
 _LEXICAL_RE = re.compile(

engraphis/core/secrets.py

@@ -29,8 +29,65 @@ def __init__(self, field: str, kind: str) -> None:
 # safe to block without a caller-supplied label.  Assignment detection below catches
 # generic credentials (including private deployment tokens) only when the field name
 # explicitly says that it is a credential.
-_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
-    ("private key", re.compile(r"-----BEGIN(?: [A-Z0-9]+)? PRIVATE KEY-----", re.I)),
+_PEM_HEADER = "-----begin private key-----"
+_PEM_HEADER_ALT = "-----begin "
+
+
+def _contains_pem_header(value: str) -> bool:
+    """O(N) PEM-header detection without regex backtracking.
+
+    Equivalent to ``re.search(r"-----BEGIN(?: [A-Z0-9]+)? PRIVATE KEY-----", re.I)``
+    but implemented as a case-normalized anchor scan: polynomial-ReDoS analysis
+    of the optional ``[A-Z0-9]+`` group is moot when no regex engine is involved.
+    Recognizes the canonical header and ``-----BEGIN X PRIVATE KEY-----`` variants
+    (RSA/EC/OPENSSH/ENCRYPTED/…).
+    """
+    lowered = value.casefold()
+    start = 0
+    while True:
+        index = lowered.find(_PEM_HEADER, start)
+        if index != -1:
+            return True
+        anchor = lowered.find(_PEM_HEADER_ALT, start)
+        if anchor == -1:
+            return False
+        tail = lowered[anchor + len(_PEM_HEADER_ALT):]
+        for token in ("rsa ", "ec ", "dsa ", "openssh ", "encrypted ", "pgp "):
+            if tail.startswith(token) and tail[len(token):].startswith(
+                    "private key-----"):
+                return True
+        start = anchor + len(_PEM_HEADER_ALT)
+
+
+class _PEMHeaderPattern:
+    """Minimal stand-in for the removed PEM regex: a linear-time ``search`` and ``sub``."""
+
+    @staticmethod
+    def search(value: str) -> _PEMMatch | None:
+        return _PEMMatch(True) if _contains_pem_header(value) else None
+
+    @staticmethod
+    def sub(repl: str, string: str, count: int = 0) -> str:
+        return _redact_pem(string)
+
+
+class _PEMMatch:
+    """Truthiness-compatible stand-in for a regex match object."""
+
+    __slots__ = ("_ok",)
+
+    def __init__(self, ok: bool) -> None:
+        self._ok = ok
+
+    def __bool__(self) -> bool:
+        return self._ok
+
+    def group(self, *args: object) -> str:
+        return _PEM_HEADER
+
+
+_PATTERNS: tuple[tuple[str, Any], ...] = (
+    ("private key", _PEMHeaderPattern()),
     ("AWS access key", re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b")),
     ("GitHub token", re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b")),
     ("GitLab token", re.compile(r"\bglpat-[A-Za-z0-9_-]{20,}\b")),
@@ -73,14 +130,47 @@ def __init__(self, field: str, kind: str) -> None:
     """,
 )
 _REDACTION = re.compile(r"^<?(?:redacted|removed|withheld|not[_ -]?set)>?$", re.I)
-_PEM_BLOCK = re.compile(
-    r"-----BEGIN(?: [A-Z0-9]+)? PRIVATE KEY-----[\s\S]*?"
-    r"-----END(?: [A-Z0-9]+)? PRIVATE KEY-----",
-    re.I,
-)
 _REDACTED = "<redacted>"
 
 
+def _redact_pem(text: str) -> str:
+    """Safely redact PEM private key blocks in O(N) linear time without regex backtracking."""
+    out: list[str] = []
+    pos = 0
+    upper = text.upper()
+    while True:
+        begin = upper.find("-----BEGIN", pos)
+        if begin == -1:
+            out.append(text[pos:])
+            break
+        header_end = upper.find("-----", begin + 10)
+        if header_end == -1:
+            out.append(text[pos:])
+            break
+        header = upper[begin:header_end + 5]
+        if "PRIVATE KEY" not in header:
+            out.append(text[pos:header_end + 5])
+            pos = header_end + 5
+            continue
+        end = upper.find("-----END", header_end + 5)
+        if end == -1:
+            out.append(text[pos:])
+            break
+        footer_end = upper.find("-----", end + 8)
+        if footer_end == -1:
+            out.append(text[pos:])
+            break
+        footer = upper[end:footer_end + 5]
+        if "PRIVATE KEY" not in footer:
+            out.append(text[pos:footer_end + 5])
+            pos = footer_end + 5
+            continue
+        out.append(text[pos:begin])
+        out.append(_REDACTED)
+        pos = footer_end + 5
+    return "".join(out)
+
+
 def _text(value: Any) -> str:
     if isinstance(value, str):
         return value
@@ -157,8 +247,12 @@ def redact_secrets(text: str) -> str:
     """
     if not isinstance(text, str) or not text:
         return text
-    safe = _PEM_BLOCK.sub(_REDACTED, text)
+    safe = _redact_pem(text)
+    # PEM blocks are handled by the linear-time ``_redact_pem`` above; the
+    # detection entry in ``_PATTERNS`` is find-based and has no ``sub``.
     for _kind, pattern in _PATTERNS:
+        if isinstance(pattern, _PEMHeaderPattern):
+            continue
         safe = pattern.sub(_REDACTED, safe)
     safe = _DSN.sub(_REDACTED, safe)
     safe = _ASSIGNMENT.sub(_REDACTED, safe)

engraphis/core/store.py

@@ -837,6 +837,10 @@ def memory_matches_filter(rec: MemoryRecord, flt: Optional[SearchFilter], *,
             return False
         if flt.mtypes is not None and rec.mtype not in flt.mtypes:
             return False
+        if flt.modified_since is not None:
+            mod_ts = max(rec.ingested_at or 0.0, getattr(rec, "updated_at", 0.0) or 0.0)
+            if mod_ts < flt.modified_since:
+                return False
     if include_invalid:
         return True
     valid_at, known_at = _temporal_anchors(flt, valid_at=at)
@@ -9471,6 +9475,9 @@ def _where(self, flt: Optional[SearchFilter], include_invalid: bool,
                     marks = ",".join("?" for _ in flt.mtypes)
                     where.append(f"{p}mtype IN ({marks})")
                     params.extend(_enum(m) for m in flt.mtypes)
+            if flt.modified_since is not None:
+                where.append(f"{p}ingested_at>=?")
+                params.append(flt.modified_since)
         if not include_invalid:
             valid_at, known_at = _temporal_anchors(flt)
             where.append(f"({p}valid_from IS NULL OR {p}valid_from<=?)")

engraphis/service.py

@@ -67,7 +67,7 @@
 )
 from engraphis.core.query_planner import PLANNING_MODES
 from engraphis.core.retrieval_policy import CANDIDATE_DEPTH_MODES, RETRIEVAL_PROFILES
-from engraphis.core.secrets import SecretDetectedError, reject_secrets
+from engraphis.core.secrets import SecretDetectedError, redact_secrets as _redact_secrets, reject_secrets
 from engraphis.core.store import (
     _loads,
     _merge_edge_provenance,
@@ -1718,13 +1718,18 @@ def remember(self, content: str, *, workspace: str, repo: Optional[str] = None,
                  subject_key: str = "", claim_kind: str = "",
                  _local_cli_operator: bool = False,
                  _local_agent_operator: bool = False,
-                 _ingress: str = "service") -> dict:
+                 _ingress: str = "service",
+                 redact_secrets: bool = False) -> dict:
         """Store one memory. Returns its id, resolved scope, and the resolution
         outcome (``op``: add/noop/invalidate/relate — see
         ``MemoryEngine.remember_with_resolution``).
         """
         content = _clean_text(content, field="content", max_chars=MAX_CONTENT_CHARS)
         title = _clean_text(title, field="title", max_chars=MAX_TITLE_CHARS, required=False)
+        if redact_secrets:
+            content = _redact_secrets(content)
+            if title:
+                title = _redact_secrets(title)
         _reject_secret_capture((
             ("content", content), ("title", title), ("keywords", keywords),
             ("metadata", metadata), ("subject_key", subject_key), ("claim_kind", claim_kind),
@@ -1805,6 +1810,7 @@ def remember(self, content: str, *, workspace: str, repo: Optional[str] = None,
                 valid_from=valid_from,
                 subject_key=subject_key, claim_kind=claim_kind,
                 resolve_conflicts=bool(resolve_conflicts),
+                redact_secrets=redact_secrets,
             )
         except ValueError as exc:
             if str(exc).startswith("valid_from "):
@@ -1824,6 +1830,8 @@ def remember(self, content: str, *, workspace: str, repo: Optional[str] = None,
             out["resolution"] = result.get("reason", "")
         if result["op"] == "invalidate":
             out["superseded"] = result["superseded"]
+            if "superseded_detail" in result:
+                out["superseded_detail"] = result["superseded_detail"]
         if result["op"] == "relate":
             out["related_to"] = result.get("related_to")
         if result["op"] == "quarantined":

tests/test_bitemporal_recall.py

@@ -307,8 +307,28 @@ def test_as_of_is_the_valid_at_compatibility_alias_and_conflicts_are_rejected():
         SearchFilter(as_of=100.0, valid_at=101.0)
 
 
-@pytest.mark.parametrize("field", ["as_of", "valid_at", "known_at"])
+@pytest.mark.parametrize("field", ["as_of", "valid_at", "known_at", "modified_since"])
 def test_temporal_filter_anchors_must_be_finite(field):
     for invalid in (float("nan"), True):
         with pytest.raises(ValueError, match=field + " must be a finite timestamp"):
             SearchFilter(**{field: invalid})
+
+
+def test_modified_since_filters_memories_for_delta_recall():
+    engine, workspace_id, repo_id, _ = _engine_with_historical_memory()
+    m1 = engine.remember("First memory created at t1", workspace_id=workspace_id, repo_id=repo_id)
+    rec1 = engine.store.get_memory(m1)
+    assert rec1 is not None
+
+    # artificially set ingested_at for deterministic separation
+    engine.store.conn.execute("UPDATE memories SET ingested_at=100.0 WHERE id=?", (m1,))
+    m2 = engine.remember("Second memory created at t2", workspace_id=workspace_id, repo_id=repo_id)
+    engine.store.conn.execute("UPDATE memories SET ingested_at=200.0 WHERE id=?", (m2,))
+    engine.store.conn.commit()
+
+    delta_filter = SearchFilter(workspace_id=workspace_id, repo_id=repo_id, modified_since=150.0)
+    memories = engine.store.list_memories(delta_filter)
+    ids = [m.id for m in memories]
+    assert m2 in ids
+    assert m1 not in ids
+

tests/test_release_write_path.py

@@ -88,3 +88,27 @@ def test_keyed_number_change_invalidates_predecessor():
         for record in eng.store.list_memories(SearchFilter(workspace_id=wid))
     }
     assert old_id not in live and new_id in live
+
+
+def test_remember_with_resolution_provides_superseded_detail():
+    eng, wid, _rid = _engine()
+    key = {"subject_key": "api.rate_limit", "claim_kind": "configured_value"}
+    old_id = eng.remember(
+        "The API rate limit is 100 requests per minute per API key.",
+        workspace_id=wid, **key,
+    )
+    eng.store.reinforce(old_id)
+
+    res = eng.remember_with_resolution(
+        "The API rate limit is now 500 requests per minute per API key.",
+        workspace_id=wid, **key,
+    )
+    assert res["op"] == "invalidate"
+    assert res["superseded"] == [old_id]
+    assert "superseded_detail" in res
+    detail = res["superseded_detail"]
+    assert detail["id"] == old_id
+    assert "100 requests" in detail["content_preview"]
+    assert detail["access_count"] >= 1
+    assert detail["stability_days"] > 0
+

tests/test_retrieval_policy.py

@@ -63,6 +63,9 @@ def test_fast_profile_skips_graph_traversal_for_latency_sensitive_recalls() -> N
         ('Find the exact "RATE_LIMIT" identifier.', "lexical"),
         ("What did we decide for the launch?", "balanced"),
         ("Why does Handler.handle() call the API_KEY module?", "code"),
+        ("How is auth connected to billing?", "graph"),
+        ("Find the struct definition", "code"),
+        ("What was superseded by the v2 migration?", "graph"),
     ],
 )
 def test_auto_routing_is_deterministic_and_uses_specific_signals_first(

tests/test_secret_hygiene.py

@@ -513,3 +513,21 @@ def test_sync_drops_secret_bearing_rows_before_store_upsert():
     assert workspace
     assert report["rejected"] == 1
     assert store.count_memories() == 0
+
+
+def test_service_and_engine_redact_secrets_when_opted_in():
+    service = MemoryService.create(":memory:")
+    # By default, storing an API key raises ValidationError
+    with pytest.raises(ValidationError, match="potential OpenAI API key"):
+        service.remember(f"Debugging log with key {_LEAK}", workspace="acme")
+
+    # With redact_secrets=True, it succeeds and masks the credential safely
+    result = service.remember(
+        f"Debugging log with key {_LEAK}", workspace="acme", redact_secrets=True
+    )
+    assert result["stored"] is True
+    mem = service.store.get_memory(result["id"])
+    assert mem is not None
+    assert _LEAK not in mem.content
+    assert "<redacted>" in mem.content
+