diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index bd29c9f..9bdf9e4 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -29,10 +29,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install -r requirements.txt
-          # ruff is PINNED: it gates the build, and its default rule set moves
-          # between minor releases -- 0.16 flagged 392 findings across untouched
-          # files. Bump deliberately, with the findings, never as a side effect.
-          pip install "ruff==0.15.12" pip-audit mypy pytest-cov
+          pip install ruff pip-audit mypy pytest-cov
 
       - name: Audit pinned deps for known CVEs
         run: pip-audit -r requirements.txt
@@ -126,16 +123,6 @@ jobs:
           AUGUR_NATS_URL: nats://127.0.0.1:4223
         run: pytest tests/integration/ -m "not slow" -v
 
-      # The same suite with session-provenance enforcement on. Enforcement is
-      # off by default in production, so nothing else would catch a learned
-      # write that loses its provenance and starts failing closed in silence.
-      - name: Run fast integration tests with provenance enforcement
-        env:
-          AUGUR_REDIS_URL: redis://127.0.0.1:6379/1
-          AUGUR_NATS_URL: nats://127.0.0.1:4223
-          AUGUR_PROVENANCE_MODE: enforce
-        run: pytest tests/integration/ -m "not slow" -q
-
       - name: Dump service logs on failure
         if: failure()
         run: |
diff --git a/consilium/advisor.py b/consilium/advisor.py
index e06b424..5f61c6f 100644
--- a/consilium/advisor.py
+++ b/consilium/advisor.py
@@ -1322,7 +1322,6 @@ async def process_message(
             now,
             decision=decision,
             tier=1,
-            ctx=pm.resolve_learn_context(payload.get("session_id")),
         )
         return
 
diff --git a/disciplina/reflection_engine.py b/disciplina/reflection_engine.py
index 6f4690a..fbd446d 100644
--- a/disciplina/reflection_engine.py
+++ b/disciplina/reflection_engine.py
@@ -1217,12 +1217,9 @@ def analyze_gate(
     for r in gate_rows:
         r["_arm"] = "withheld"
     reliability_audit = _behavioral_audit_per_arm(advice_rows + gate_rows, config)
-    # CL11: the readout tunes the gate, so it reads the LEARNING view of the
-    # logs — a synthetic driver's emissions/silences are excluded under ENFORCE
-    # (the online arms still read them unfiltered).
     mrt = _mrt_ipw_readout(
-        pm.load_emissions(limit=100, learnable_only=True),
-        pm.load_silence_records(limit=100, learnable_only=True),
+        pm.load_emissions(limit=100),
+        pm.load_silence_records(limit=100),
         advice_rows,
         gate_rows,
     )
diff --git a/docker-compose.deploy.yml b/docker-compose.deploy.yml
index c1311c6..d713210 100644
--- a/docker-compose.deploy.yml
+++ b/docker-compose.deploy.yml
@@ -10,7 +10,6 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
     restart: unless-stopped
 
   nexus:
@@ -24,7 +23,6 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
     restart: unless-stopped
 
   consilium:
@@ -40,7 +38,6 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
       AUGUR_OLLAMA_URL: http://host.docker.internal:11434
     restart: unless-stopped
 
@@ -55,7 +52,6 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
     restart: unless-stopped
 
   disciplina:
@@ -69,7 +65,6 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
       AUGUR_OLLAMA_URL: http://host.docker.internal:11434
     restart: unless-stopped
 
@@ -84,7 +79,6 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
     restart: unless-stopped
 
   praefectus:
@@ -98,7 +92,6 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
     restart: unless-stopped
 
   imperator:
@@ -111,7 +104,6 @@ services:
         condition: service_healthy
     environment:
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
       AUGUR_NATS_URL: nats://nats:4222
     restart: unless-stopped
 
@@ -126,7 +118,6 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
       AUGUR_OLLAMA_URL: http://host.docker.internal:11434
     extra_hosts:
       - "host.docker.internal:host-gateway"
@@ -143,5 +134,4 @@ services:
     environment:
       AUGUR_NATS_URL: nats://nats:4222
       AUGUR_REDIS_URL: redis://redis:6379
-      AUGUR_PROVENANCE_MODE: ${AUGUR_PROVENANCE_MODE:-report}
     restart: unless-stopped
diff --git a/imperator/apply.py b/imperator/apply.py
index fa63b7e..6179b46 100644
--- a/imperator/apply.py
+++ b/imperator/apply.py
@@ -27,15 +27,10 @@
 _DIRECTIVE_ACTIONS = ("suppress", "downgrade")
 
 
-def _arm_gate(pm, p: dict, *, cfg, ctx) -> bool:
+def _arm_gate(pm, p: dict, *, cfg, ctx=None) -> bool:
     """Write the durable applied-marker that arms the one-move-per-(kind,target)
     anti-thrash gate. Returns True if armed, False if the write failed.
 
-    ``ctx`` is REQUIRED, not defaulted: the marker is a learned write, so a
-    caller that forgets it makes this fail closed under ENFORCE — and because
-    the failure is caught below, the whole armed-apply path would die behind a
-    single log line. A missing keyword must break loudly at the call instead.
-
     The marker is the ONLY thing closing the gate, so it is written BEFORE the
     primary (matrix/prompt) write: a marker failure must abort the apply rather
     than leave a committed change behind an open gate, where a DIFFERENT-text
@@ -173,7 +168,7 @@ def _apply_prompt_strategy(pm, p: dict, *, cfg, ctx=None) -> bool:
     current = pm.load_prompt(domain)
     if not is_prompt_acceptable(text, cfg) or current is None:
         return False
-    if not _arm_gate(pm, p, cfg=cfg, ctx=ctx):
+    if not _arm_gate(pm, p, cfg=cfg):
         return False
     if current != text:
         pm.save_prompt(domain, text, ctx=ctx)
@@ -212,7 +207,7 @@ def _apply_sigma(pm, p: dict, *, cfg, ctx=None) -> bool:
     if not math.isfinite(sigma) or not (sigma_min <= sigma <= sigma_max):
         return False
     current = pm.load_thresholds(domain) or {}
-    if not _arm_gate(pm, p, cfg=cfg, ctx=ctx):
+    if not _arm_gate(pm, p, cfg=cfg):
         return False
     a["prior_sigma"] = current.get("sigma_threshold")
     pm.save_thresholds(domain, {**current, "sigma_threshold": sigma}, ctx=ctx)
@@ -244,7 +239,7 @@ def _apply_gate_calibration(pm, p: dict, *, cfg, ctx=None) -> bool:
     op, sk = a.get("op"), a.get("state_key", p["target"])
     if op in ("self_tolerance_add", "self_tolerance_remove"):
         prior = pm.is_self_tolerant(sk)
-        if not _arm_gate(pm, p, cfg=cfg, ctx=ctx):
+        if not _arm_gate(pm, p, cfg=cfg):
             return False
         if op == "self_tolerance_add":
             pm.add_self_tolerance(sk, ctx=ctx)
@@ -267,7 +262,7 @@ def _apply_gate_calibration(pm, p: dict, *, cfg, ctx=None) -> bool:
         if not math.isfinite(value) or not (_FLOOR_MIN <= value <= _FLOOR_MAX):
             return False
         prior_entry = pm.load_habituation_floor(sk) or {}
-        if not _arm_gate(pm, p, cfg=cfg, ctx=ctx):
+        if not _arm_gate(pm, p, cfg=cfg):
             return False
         new_entry = {**prior_entry, "floor": value, "last_ts": time.time()}
         pm.save_gate_tuning_state(floors={sk: new_entry}, ctx=ctx)
@@ -381,7 +376,7 @@ def _apply_context_directive(pm, p: dict, *, cfg, ctx=None) -> bool:
         if not directive_id:
             return False
         prior = pm.get_dialogue_directive(directive_id)  # rollback anchor, pre-delete
-        if not _arm_gate(pm, p, cfg=cfg, ctx=ctx):
+        if not _arm_gate(pm, p, cfg=cfg):
             return False
         pm.remove_dialogue_directive(directive_id, ctx=ctx)
         a["prior_directive"] = prior
@@ -404,7 +399,7 @@ def _apply_context_directive(pm, p: dict, *, cfg, ctx=None) -> bool:
         "rationale": a.get("rationale", p.get("rationale", "")),
     }
     prior = pm.get_dialogue_directive(directive_id)  # rollback anchor, pre-write
-    if not _arm_gate(pm, p, cfg=cfg, ctx=ctx):
+    if not _arm_gate(pm, p, cfg=cfg):
         return False
     # Propagate the write's bool (True = written, False = NEW id refused at cap);
     # record the rollback anchor only when the directive was actually stored.
@@ -449,16 +444,19 @@ def _apply_semantic_fact(pm, p: dict, *, cfg, session_id: str | None) -> bool:
     other _dispatch_confirmed handler's contract.
     """
     a = p.get("action") or {}
-    ctx = pm.resolve_learn_context(session_id)
     if a.get("op") == "remove":
         mid = a.get("memory_id")
         if not mid:
             return False
         prior = pm.load_memory_state(mid)  # rollback anchor, read before the flip
-        if not _arm_gate(pm, p, cfg=cfg, ctx=ctx):
+        if not _arm_gate(pm, p, cfg=cfg):
             return False
         if prior is not None:
-            pm.save_memory_state(mid, {**prior, "status": "archived"}, ctx=ctx)
+            pm.save_memory_state(
+                mid,
+                {**prior, "status": "archived"},
+                ctx=pm.resolve_learn_context(session_id),
+            )
         a["prior_fact"] = prior
         return True
     pattern = a.get("pattern")
@@ -466,7 +464,7 @@ def _apply_semantic_fact(pm, p: dict, *, cfg, session_id: str | None) -> bool:
         return False
     mid = make_memory_id(pattern)
     prior = pm.load_memory_state(mid)  # rollback anchor, read before the write
-    if not _arm_gate(pm, p, cfg=cfg, ctx=ctx):
+    if not _arm_gate(pm, p, cfg=cfg):
         return False
     # Rationale precedence: action-level first (set by the undo-inverse
     # builder restoring a prior fact's own rationale), then proposal-level
diff --git a/imperator/improver.py b/imperator/improver.py
index 9fe8ee2..5decf10 100644
--- a/imperator/improver.py
+++ b/imperator/improver.py
@@ -23,12 +23,10 @@
 from tabula.connections import connect_redis  # noqa: E402
 from tabula.heartbeat import start_heartbeat  # noqa: E402
 from tabula.persistence import PersistenceManager  # noqa: E402
-from tabula.provenance import ProvenanceMode, get_provenance_mode  # noqa: E402
 from imperator import proposals as P, reasoner, apply as A  # noqa: E402
 
 log = logging.getLogger("imperator.improver")
-_REFLECTION_SUBJECT = "augur.disciplina.complete"
-_CONSUMED = frozenset({_REFLECTION_SUBJECT, "augur.imperator.ii.trigger"})
+_CONSUMED = frozenset({"augur.disciplina.complete", "augur.imperator.ii.trigger"})
 
 
 def consumed(subject: str) -> bool:
@@ -174,20 +172,6 @@ async def on_msg(msg):
         except (json.JSONDecodeError, UnicodeDecodeError):
             log.warning("imperator II trigger payload undecodable; skipping")
             return
-        # §4.3e second layer: filtering the read-model keeps a non-learnable
-        # reflection out of the self-model, but the trigger would still spend a
-        # cycle reasoning about it. Only the reflection subject is gated — the
-        # dialogue trigger is a direct user action carrying no session.
-        if (
-            msg.subject == _REFLECTION_SUBJECT
-            and get_provenance_mode() is ProvenanceMode.ENFORCE
-            and not pm.is_learnable_session(payload.get("session_id"))
-        ):
-            log.info(
-                "imperator II trigger dropped: session %s is not learnable",
-                payload.get("session_id"),
-            )
-            return
         # Take the lock synchronously HERE (not inside the task): this closes the
         # window where two back-to-back callbacks both pass lock.locked()==False
         # before either task gets scheduled, which would queue the second cycle
diff --git a/imperator/sources.py b/imperator/sources.py
index 2267e52..330aa30 100644
--- a/imperator/sources.py
+++ b/imperator/sources.py
@@ -110,20 +110,16 @@ def resolve_reception(pm, last_advice) -> dict | None:
 def windowed_rates(pm, now: float, window_s: float) -> dict:
     """True windowed suppression rate + advice volume. Read cap-sized (default 100 < 2000 cap).
     Exclude probe / audit_only emissions from 'genuine delivered'.
-
-    CL11/§4.3e: these rates feed the self-model's blind spots, which Imperator II
-    reasons over — so they read the LEARNING view of the gate logs and a
-    synthetic driver's burst is excluded under ENFORCE.
     """
     lo = now - window_s
     silences = [
         s
-        for s in pm.load_silence_records(limit=_GATE_LOG_CAP, learnable_only=True)
+        for s in pm.load_silence_records(limit=_GATE_LOG_CAP)
         if float(s.get("ts", 0.0)) >= lo
     ]
     emissions = [
         e
-        for e in pm.load_emissions(limit=_GATE_LOG_CAP, learnable_only=True)
+        for e in pm.load_emissions(limit=_GATE_LOG_CAP)
         if float(e.get("ts", 0.0)) >= lo
         and not e.get("probe")
         and not e.get("audit_only")
diff --git a/infrastructure/run_augur.sh b/infrastructure/run_augur.sh
index eceaf37..8c0cad6 100755
--- a/infrastructure/run_augur.sh
+++ b/infrastructure/run_augur.sh
@@ -18,14 +18,6 @@
 #   AUGUR_REDIS_URL=redis://127.0.0.1:6379/1 \
 #   AUGUR_NATS_URL=nats://127.0.0.1:4223 \
 #   AUGUR_TEST_STARTUP_WAIT_S=12 .venv/bin/pytest tests/integration/
-#
-# Every component reads AUGUR_PROVENANCE_MODE (off|report|enforce) at import,
-# so exporting it here covers the whole pipeline:
-#   AUGUR_PROVENANCE_MODE=report bash infrastructure/run_augur.sh
-# report logs what enforcement WOULD withhold without withholding it; enforce
-# stops non-learnable sessions from training the system. The deploy stack
-# defaults to report; this launcher inherits whatever the shell exports (off
-# when unset). Watch the report logs across a few real sessions before enforce.
 
 set -euo pipefail
 
diff --git a/limen/gate.py b/limen/gate.py
index c9eb05f..0bae83a 100644
--- a/limen/gate.py
+++ b/limen/gate.py
@@ -1201,17 +1201,13 @@ def record_delivery_success(
         """
         if audit_only:
             pm.save_emission(
-                self._emission_record(
-                    signature, decision, now, tier, ctx=ctx, audit_only=True
-                )
+                self._emission_record(signature, decision, now, tier, audit_only=True)
             )
             return
 
         if decision.probe:
             pm.save_emission(
-                self._emission_record(
-                    signature, decision, now, tier, ctx=ctx, audit_only=False
-                )
+                self._emission_record(signature, decision, now, tier, audit_only=False)
             )
             return
 
@@ -1221,17 +1217,13 @@ def record_delivery_success(
             # state (no observed/h/advice-rate/channel_stats/cost_tier on a bogus
             # single:{domain}:? key).
             pm.save_emission(
-                self._emission_record(
-                    signature, decision, now, tier, ctx=ctx, audit_only=False
-                )
+                self._emission_record(signature, decision, now, tier, audit_only=False)
             )
             return
 
         # ── Normal delivery: advance all online state ──
         pm.save_emission(
-            self._emission_record(
-                signature, decision, now, tier, ctx=ctx, audit_only=False
-            )
+            self._emission_record(signature, decision, now, tier, audit_only=False)
         )
         pm.save_observed(
             {
@@ -1268,7 +1260,6 @@ def record_suppression(
         record = {
             "ts": now,
             "decision_id": decision.id,
-            "session_id": ctx.session_id if ctx else None,
             "state_key": signature.state_key,
             "domain": signature.domain,
             "entity": signature.entity,
@@ -1368,7 +1359,6 @@ def _emission_record(
         tier: int | None,
         *,
         audit_only: bool,
-        ctx=None,
     ) -> dict[str, Any]:
         """Build a gate emission record (spec §6 emissions schema).
 
@@ -1376,17 +1366,10 @@ def _emission_record(
         normal delivery carries ``audit_only=False``.  Gating-visible readers
         (refractory/pressure/duplicate/habituation) ignore any row whose
         ``probe`` or ``audit_only`` is True.
-
-        ``session_id`` comes from the event's ``LearnContext`` so the offline
-        learning readers can exclude a non-learnable session's rows (CL11); the
-        online arms read the log unfiltered, since a synthetic burst was still
-        really delivered.  ``None`` when no context was in hand — fail-closed,
-        so an unprovenanced row never trains anything.
         """
         return {
             "ts": now,
             "decision_id": decision.id,
-            "session_id": ctx.session_id if ctx else None,
             "state_key": signature.state_key,
             "severity": signature.severity,
             "tier": tier,
diff --git a/requirements.txt b/requirements.txt
index b136013..4009474 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -7,6 +7,6 @@ httpx
 keyboard
 pytest
 pytest-asyncio
-fakeredis>=2.36.2
+fakeredis>=2.37.0
 fastmcp
 networkx
diff --git a/tabula/persistence.py b/tabula/persistence.py
index 0a34f5b..634732e 100644
--- a/tabula/persistence.py
+++ b/tabula/persistence.py
@@ -5,7 +5,6 @@
 
 import json
 import logging
-import time
 from datetime import datetime, timezone
 from typing import Any, cast
 
@@ -87,13 +86,6 @@ def _to_epoch(ts: Any) -> float:
 
 PROVENANCE_TTL_S = SESSION_META_TTL_S
 
-# Provenance-cache bounds (spec §4.3.1). Under ENFORCE every event on the
-# Vigil/Limen hot paths resolves its session, which would otherwise be a Redis
-# GET per event. Only LEARNABLE contexts are cached, and never longer than this
-# — nor longer than the metadata's own remaining TTL, whichever is shorter.
-LEARN_CONTEXT_CACHE_TTL_S = 60.0
-MAX_LEARN_CONTEXT_CACHE = 256
-
 # TTL for the correlation-tuning idempotency marker. Long enough to survive
 # manual reflect-trigger replays of a recent session, short enough to prevent
 # the key from lingering indefinitely.
@@ -105,8 +97,6 @@ class PersistenceManager:
 
     def __init__(self, r: redis.Redis) -> None:
         self._r = r
-        # session_id -> (monotonic expiry, LearnContext). Positives only.
-        self._learn_ctx_cache: dict[str, tuple[float, LearnContext]] = {}
 
     # -- JSON get/set helpers ------------------------------------------------
     # Collapse the uniform "set(key, json.dumps(x))" / "raw = get(key); return
@@ -152,22 +142,11 @@ def resolve_learn_context(self, session_id: str | None) -> LearnContext:
         session id is not evidence; only a durable record this system wrote makes
         a session learnable. Unknown / missing / corrupt / expired provenance, or
         any Redis error, yields a non-learnable ``"unknown"`` context.
-
-        Learnable results are cached (§4.3.1, see :meth:`_cache_learn_context`);
-        a non-learnable one never is, so this stays a Redis read per event for
-        exactly the sessions that are not training anything.
         """
         if not session_id:
             return LearnContext.unknown(session_id)
-        cached = self._learn_ctx_cache.get(session_id)
-        if cached is not None:
-            expires_at, ctx = cached
-            if time.monotonic() < expires_at:
-                return ctx
-            del self._learn_ctx_cache[session_id]
-        key = REDIS_KEY_META.format(sid=session_id)
         try:
-            raw = self._r.get(key)
+            raw = self._r.get(REDIS_KEY_META.format(sid=session_id))
             if raw is None:
                 return LearnContext.unknown(session_id)
             data = json.loads(raw)
@@ -176,73 +155,9 @@ def resolve_learn_context(self, session_id: str | None) -> LearnContext:
             origin = data.get("origin")
             if not isinstance(origin, str) or not origin:
                 origin = "unknown"
-            ctx = LearnContext(session_id, data.get("learnable") is True, origin)
+            return LearnContext(session_id, data.get("learnable") is True, origin)
         except Exception:
             return LearnContext.unknown(session_id)
-        if ctx.learnable:
-            self._cache_learn_context(session_id, key, ctx)
-        return ctx
-
-    def _cache_learn_context(
-        self, session_id: str, meta_key: str, ctx: LearnContext
-    ) -> None:
-        """Cache a LEARNABLE context, never past the metadata's own expiry (§4.3.1).
-
-        Both halves matter. Caching a negative would let a lookup that raced
-        metadata creation drop a real session's learning for the life of the
-        process. Letting a positive outlive its key would train on a session
-        whose provenance has expired — and expired means non-learnable — so the
-        entry's lifetime is capped by the key's remaining ``PTTL``.
-
-        Best-effort throughout: a Redis error here costs a cache entry, never a
-        wrong answer, since every miss re-reads the truth.
-        """
-        try:
-            pttl_ms = int(self._r.pttl(meta_key))
-        except Exception:
-            return  # unreadable TTL (or a stubbed client): resolve every time
-        if pttl_ms == -2:  # key vanished between the GET and here
-            return
-        ttl = LEARN_CONTEXT_CACHE_TTL_S
-        if pttl_ms >= 0:
-            ttl = min(ttl, pttl_ms / 1000.0)
-        if ttl <= 0:
-            return
-        if len(self._learn_ctx_cache) >= MAX_LEARN_CONTEXT_CACHE:
-            now = time.monotonic()
-            self._learn_ctx_cache = {
-                sid: entry
-                for sid, entry in self._learn_ctx_cache.items()
-                if entry[0] > now
-            }
-            if len(self._learn_ctx_cache) >= MAX_LEARN_CONTEXT_CACHE:
-                return  # still full of live entries: skip rather than thrash
-        self._learn_ctx_cache[session_id] = (time.monotonic() + ttl, ctx)
-
-    def filter_learnable_records(self, records: list[dict]) -> list[dict]:
-        """Drop records whose session may not train (CL11) — ENFORCE only.
-
-        For logs that are written unconditionally but read by BOTH a runtime
-        consumer and a learning one: the writer stays ``@non_learning_write`` (so
-        the online path always sees what really happened) and only the learning
-        reader opts in here. Provenance is resolved once per distinct session,
-        not once per record — a gate-log window is thousands of rows over a
-        handful of sessions. A record with no ``session_id`` is fail-closed:
-        no evidence it may train, so it is excluded.
-        """
-        from tabula.provenance import ProvenanceMode, get_provenance_mode
-
-        if get_provenance_mode() is not ProvenanceMode.ENFORCE:
-            return records
-        learnable: dict[Any, bool] = {}
-        out: list[dict] = []
-        for rec in records:
-            sid = rec.get("session_id")
-            if sid not in learnable:
-                learnable[sid] = self.resolve_learn_context(sid).learnable
-            if learnable[sid]:
-                out.append(rec)
-        return out
 
     def is_learnable_session(self, session_id: str | None) -> bool:
         """Return True only for a session recorded as learnable. Fails CLOSED.
@@ -1066,31 +981,25 @@ def is_tuning_applied(
     def save_silence_record(self, record: dict) -> None:
         """Append a gate suppression record to augur:limen:silences (capped).
 
-        Schema: {ts, decision_id, session_id, state_key, domain, entity,
-                 severity, arm, reason, metrics, mrt_eligible, p_withhold}
+        Schema: {ts, decision_id, state_key, domain, entity, severity, arm,
+                 reason, metrics, mrt_eligible, p_withhold}
         """
         key = "augur:limen:silences"
         self._r.lpush(key, json.dumps(record))
         self._r.ltrim(key, 0, MAX_GATE_SILENCES - 1)
 
-    def load_silence_records(
-        self, *, limit: int = 100, learnable_only: bool = False
-    ) -> list[dict]:
+    def load_silence_records(self, *, limit: int = 100) -> list[dict]:
         """Return up to *limit* recent silence records, newest first.
 
         Returns [] if the list is absent or any entry is corrupt.
-        ``learnable_only`` is the CL11 opt-in for the offline learning readers
-        (see :meth:`filter_learnable_records`); the online arms leave it False,
-        because a suppression that really happened must still count.
         """
         key = "augur:limen:silences"
         raw_list = cast(list[Any], self._r.lrange(key, 0, limit - 1))
         try:
-            records = [json.loads(entry) for entry in raw_list]
+            return [json.loads(entry) for entry in raw_list]
         except (json.JSONDecodeError, TypeError, UnicodeDecodeError):
             log.warning("augur:limen:silences contained a corrupt entry; returning []")
             return []
-        return self.filter_learnable_records(records) if learnable_only else records
 
     @non_learning_write(
         reason="gate log; excluded from tuning at read in analyze_gate (CL11)"
@@ -1098,31 +1007,25 @@ def load_silence_records(
     def save_emission(self, record: dict) -> None:
         """Append a gate emission record to augur:limen:emissions (capped).
 
-        Schema: {ts, decision_id, session_id, state_key, severity, tier, probe,
+        Schema: {ts, decision_id, state_key, severity, tier, probe,
                  audit_only, withheld_reason, mrt_eligible, p_fire}
         """
         key = "augur:limen:emissions"
         self._r.lpush(key, json.dumps(record))
         self._r.ltrim(key, 0, MAX_GATE_EMISSIONS - 1)
 
-    def load_emissions(
-        self, *, limit: int = 100, learnable_only: bool = False
-    ) -> list[dict]:
+    def load_emissions(self, *, limit: int = 100) -> list[dict]:
         """Return up to *limit* recent emission records, newest first.
 
         Returns [] if the list is absent or any entry is corrupt.
-        ``learnable_only`` is the CL11 opt-in for the offline learning readers
-        (see :meth:`filter_learnable_records`); the online arms leave it False,
-        because a delivery that really happened must still refract.
         """
         key = "augur:limen:emissions"
         raw_list = cast(list[Any], self._r.lrange(key, 0, limit - 1))
         try:
-            records = [json.loads(entry) for entry in raw_list]
+            return [json.loads(entry) for entry in raw_list]
         except (json.JSONDecodeError, TypeError, UnicodeDecodeError):
             log.warning("augur:limen:emissions contained a corrupt entry; returning []")
             return []
-        return self.filter_learnable_records(records) if learnable_only else records
 
     @non_learning_write(
         reason="gate log; excluded from tuning at read in analyze_gate (CL11)"
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 1db909a..454e3a6 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -134,13 +134,8 @@ async def nats_conn() -> AsyncIterator[nats_client.NATS]:
 
 @pytest.fixture
 def session_id() -> str:
-    """Fresh session identifier, with its provenance recorded.
-
-    A live sensor mints provenance before its first event; the harness does the
-    same, so anything driven by this fixture can train (see
-    :func:`mint_session_provenance`).
-    """
-    return learnable_session()
+    """Fresh UUID session identifier."""
+    return str(uuid4())
 
 
 @pytest_asyncio.fixture
@@ -239,43 +234,6 @@ async def pipeline(
 # ---------------------------------------------------------------------------
 
 
-_provenance_pm: PersistenceManager | None = None
-
-
-def mint_session_provenance(session_id: str, *, origin: str = "real") -> None:
-    """Record a test-cell session's provenance (spec §4.4).
-
-    Sessions in the test cell are real *within that cell* — nothing is withheld
-    there — but "real" still has to be RECORDED, because provenance is only ever
-    a durable record this system wrote, never an inference from a session id. So
-    the harness mints it exactly like a live sensor does, rather than the
-    provenance layer learning about cells.
-
-    Without this, ENFORCE correctly withholds every learned write the suite
-    triggers and the tuning assertions fail. Pass ``origin="synthetic"`` to
-    exercise the withholding path on purpose.
-    """
-    global _provenance_pm
-    if _provenance_pm is None:
-        _provenance_pm = PersistenceManager(
-            redis.Redis.from_url(_config.redis_url, decode_responses=True)
-        )
-    _provenance_pm.save_session_meta(
-        session_id, origin=origin, created_by="integration"
-    )
-
-
-def learnable_session(session_id: str | None = None, *, origin: str = "real") -> str:
-    """Mint a session id whose provenance is recorded, and return it.
-
-    The one-liner replacement for a bare ``str(uuid4())`` in a test that then
-    drives a learned write.
-    """
-    sid = session_id or str(uuid4())
-    mint_session_provenance(sid, origin=origin)
-    return sid
-
-
 async def inject_perception_event(
     nc: nats_client.NATS,
     domain: str,
@@ -285,15 +243,8 @@ async def inject_perception_event(
     unit: str,
     context: dict,
     session_id: str,
-    origin: str = "real",
 ) -> PerceptionEvent:
-    """Create a PerceptionEvent and publish it to the NATS perception subject.
-
-    Mints the session's provenance first (see :func:`mint_session_provenance`),
-    so the components that consume the event can resolve it — the live sensors
-    do the same thing before their first event.
-    """
-    mint_session_provenance(session_id, origin=origin)
+    """Create a PerceptionEvent and publish it to the NATS perception subject."""
     event = PerceptionEvent(
         domain=domain,
         stream_id=f"{domain}_stream",
diff --git a/tests/integration/test_app_descriptor_pipeline.py b/tests/integration/test_app_descriptor_pipeline.py
index 7615215..d54f6ac 100644
--- a/tests/integration/test_app_descriptor_pipeline.py
+++ b/tests/integration/test_app_descriptor_pipeline.py
@@ -11,7 +11,6 @@
 from tabula.persistence import PersistenceManager
 from consilium.app_descriptor import ClassifierLane
 from consilium.advisor import enrich_activity_descriptor
-from tests.integration.conftest import learnable_session
 
 
 @pytest.fixture
@@ -29,9 +28,6 @@ def test_os_identity_cached_and_injected(pm):
     anomaly = {
         "domain": "activity_intensity",
         "entity": "alpha_app",
-        # A real vigil anomaly carries its session (anomaly_detector.py), which
-        # is where the descriptor cache write gets its provenance.
-        "session_id": learnable_session(),
         "context": {"app_identity": "Alpha Browser"},
     }
     enrich_activity_descriptor(pm, lane, anomaly)
@@ -43,7 +39,6 @@ def test_os_identity_cached_and_injected(pm):
 
 
 def test_llm_fallback_does_not_clobber_os_identity(pm):
-    ctx = pm.resolve_learn_context(learnable_session())
-    pm.save_app_descriptor("alpha_app", "Alpha Browser", overwrite=True, ctx=ctx)
-    pm.save_app_descriptor("alpha_app", "some llm guess", overwrite=False, ctx=ctx)
+    pm.save_app_descriptor("alpha_app", "Alpha Browser", overwrite=True)
+    pm.save_app_descriptor("alpha_app", "some llm guess", overwrite=False)
     assert pm.load_app_descriptor("alpha_app") == "Alpha Browser"
diff --git a/tests/integration/test_conscientia_live.py b/tests/integration/test_conscientia_live.py
index 8f9eb9f..7455939 100644
--- a/tests/integration/test_conscientia_live.py
+++ b/tests/integration/test_conscientia_live.py
@@ -19,12 +19,11 @@
 import pytest
 
 from tabula.persistence import PersistenceManager
-from tests.integration.conftest import learnable_session
 
 # The gated proposal's target sits outside charter.PROTECTED_SURFACES, so the
 # rubric's recommendation is "needs_human" (not "reject").
 _PROPOSAL_ID = "conscientia-live-g1"
-_SESSION_ID = "conscientia-live"  # provenance minted in the test body
+_SESSION_ID = "conscientia-live"
 
 
 @pytest.mark.parametrize("pipeline", [["disciplina"]], indirect=True)
@@ -50,7 +49,6 @@ async def _capture(msg) -> None:  # type: ignore[no-untyped-def]
     # Seed one gated proposal and matching feedback. The feedback is required:
     # on_trigger bails before run_reflection if get_feedback(session_id) is None,
     # so without it the conscientia pass never runs.
-    learnable_session(_SESSION_ID)
     pm.save_proposal(
         {
             "proposal_id": _PROPOSAL_ID,
@@ -61,8 +59,7 @@ async def _capture(msg) -> None:  # type: ignore[no-untyped-def]
             "ts": 1.0,
             "action": {"patch": "x"},
             "status": "logged",
-        },
-        ctx=pm.resolve_learn_context(_SESSION_ID),
+        }
     )
     pm.save_feedback(_SESSION_ID, {"session_id": _SESSION_ID, "advice_events": []})
 
diff --git a/tests/integration/test_correlator_graph_persistence.py b/tests/integration/test_correlator_graph_persistence.py
index 8325aff..05b1aed 100644
--- a/tests/integration/test_correlator_graph_persistence.py
+++ b/tests/integration/test_correlator_graph_persistence.py
@@ -11,12 +11,11 @@
 
 import asyncio
 import json
+import uuid
 from datetime import datetime, timezone
 
 import pytest
 
-from tests.integration.conftest import learnable_session
-
 pytestmark = pytest.mark.asyncio
 
 
@@ -52,7 +51,7 @@ async def test_session_end_flushes_graph_to_redis(
     nats_conn,
 ) -> None:
     """Correlation event + session.end → graph in Redis."""
-    sid = learnable_session()
+    sid = str(uuid.uuid4())
 
     # Warm chess baseline. Must be long enough that the detector's
     # min_observations gate flips to trained — pad to (gate + 5) so the test
@@ -192,7 +191,7 @@ async def test_session_end_with_no_correlations_still_saves_empty_graph(
     """A session with no cross-domain correlations still persists an
     empty graph so consumers can distinguish 'session ended cleanly with
     zero correlations' from 'session never existed'."""
-    sid = learnable_session()
+    sid = str(uuid.uuid4())
 
     # Publish session.end immediately — no perception events at all
     await nats_conn.publish(
diff --git a/tests/integration/test_dialogue_showcase.py b/tests/integration/test_dialogue_showcase.py
index d85dfda..66d7bea 100644
--- a/tests/integration/test_dialogue_showcase.py
+++ b/tests/integration/test_dialogue_showcase.py
@@ -49,9 +49,7 @@
 from imperator.dialogue import router as R
 from limen import gate as G
 from memoria.tiers import plan_sweep
-from tabula.provenance import LearnContext
 from tests.conftest import CORRELATION_MEDIUM, SINGLE_MEDIUM, SINGLE_MEDIUM_TYPING
-from tests.integration.conftest import learnable_session
 from tests.test_advisor_gate_flow import _run, _scheduler
 
 pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
@@ -64,15 +62,6 @@ def _sig(payload: dict) -> G.Signature:
     return G.build_signature(payload)
 
 
-def _showcase_ctx(pm):
-    """The showcase session's provenance, as the advisor would resolve it.
-
-    The gate's writers take the EVENT's context; these scenarios drive the gate
-    directly, so the harness supplies the same thing ``process_message`` would.
-    """
-    return pm.resolve_learn_context(learnable_session("showcase"))
-
-
 def _focus_app(pm, app: str, ts: str | None = None) -> None:
     """Seed the real activity-focus stream ``load_focused_app`` reads (spec
     §7.2) — mirrors ``tests/test_dialogue_gate_directive.py``'s
@@ -91,7 +80,6 @@ def _focus_app(pm, app: str, ts: str | None = None) -> None:
 
     if ts is None:
         ts = datetime.now(timezone.utc).isoformat()
-    ctx = _showcase_ctx(pm)
     pm.append_event(
         PerceptionEvent(
             domain="activity_focus",
@@ -102,9 +90,8 @@ def _focus_app(pm, app: str, ts: str | None = None) -> None:
             unit="none",
             context={"new_app": app},
             timestamp=ts,
-            session_id=ctx.session_id,
-        ),
-        ctx=ctx,
+            session_id="showcase",
+        )
     )
 
 
@@ -134,7 +121,7 @@ async def teach(prompt, system, client, cfg):
     _focus_app(real_pm, "appX")
 
     t1 = await E.handle_turn(
-        learnable_session("sc1"),
+        "sc1",
         "stay quiet in appX",
         pm=real_pm,
         nc=real_nc,
@@ -145,7 +132,7 @@ async def teach(prompt, system, client, cfg):
     assert t1.pending is not None and t1.applied is None
 
     t2 = await E.handle_turn(
-        learnable_session("sc1"),
+        "sc1",
         "yes",
         pm=real_pm,
         nc=real_nc,
@@ -167,7 +154,7 @@ async def undo(prompt, system, client, cfg):
         )
 
     t3 = await E.handle_turn(
-        learnable_session("sc1"),
+        "sc1",
         "undo that",
         pm=real_pm,
         nc=real_nc,
@@ -181,7 +168,7 @@ async def undo(prompt, system, client, cfg):
     assert t3.pending is not None and t3.applied is None
 
     t4 = await E.handle_turn(
-        learnable_session("sc1"),
+        "sc1",
         "yes",
         pm=real_pm,
         nc=real_nc,
@@ -208,7 +195,7 @@ async def test_correct_silence_reverses_arm_and_fires(real_pm, real_nc, dialogue
     sig = _sig(SINGLE_MEDIUM_TYPING)
     gate = G.Gate()
 
-    real_pm.add_self_tolerance(state_key, ctx=LearnContext.system())
+    real_pm.add_self_tolerance(state_key)
     before = gate.evaluate(sig, real_pm, dialogue_cfg, now=200.0)
     assert before.action == "suppress"
     assert before.deciding_arm == "central_tolerance"
@@ -216,10 +203,7 @@ async def test_correct_silence_reverses_arm_and_fires(real_pm, real_nc, dialogue
     # Authoritative silence write (invariant A) — this is what
     # imperator/dialogue/router.py's _arm_for_silence reads via
     # ctx.recent_suppressions to pick which arm to reverse.
-    assert (
-        gate.record_suppression(before, sig, real_pm, 200.0, ctx=_showcase_ctx(real_pm))
-        is True
-    )
+    assert gate.record_suppression(before, sig, real_pm, 200.0) is True
 
     async def correct(prompt, system, client, cfg):
         return (
@@ -230,7 +214,7 @@ async def correct(prompt, system, client, cfg):
         )
 
     await E.handle_turn(
-        learnable_session("sc2"),
+        "sc2",
         "you should've spoken up about typing",
         pm=real_pm,
         nc=real_nc,
@@ -239,7 +223,7 @@ async def correct(prompt, system, client, cfg):
         query_fn=correct,
     )
     t2 = await E.handle_turn(
-        learnable_session("sc2"),
+        "sc2",
         "yes",
         pm=real_pm,
         nc=real_nc,
@@ -262,7 +246,6 @@ async def correct(prompt, system, client, cfg):
             "last_ts": 201.0,
             "suppressing": False,
         },
-        ctx=LearnContext.system(),
     )
     after = G.Gate().evaluate(sig, real_pm, dialogue_cfg, now=201.0)
     assert after.action != "suppress"
@@ -289,7 +272,6 @@ async def test_correct_noise_raises_tolerance_and_suppresses(
             "last_ts": 300.0,
             "suppressing": False,
         },
-        ctx=LearnContext.system(),
     )
     before = G.Gate().evaluate(sig, real_pm, dialogue_cfg, now=300.0)
     assert before.action != "suppress"
@@ -303,7 +285,7 @@ async def correct(prompt, system, client, cfg):
         )
 
     await E.handle_turn(
-        learnable_session("sc3"),
+        "sc3",
         "stop flagging my chess moves",
         pm=real_pm,
         nc=real_nc,
@@ -312,7 +294,7 @@ async def correct(prompt, system, client, cfg):
         query_fn=correct,
     )
     t2 = await E.handle_turn(
-        learnable_session("sc3"),
+        "sc3",
         "yes",
         pm=real_pm,
         nc=real_nc,
@@ -346,7 +328,7 @@ async def teach(prompt, system, client, cfg):
         )
 
     await E.handle_turn(
-        learnable_session("sc4"),
+        "sc4",
         "when chess and typing spike together that means stress",
         pm=real_pm,
         nc=real_nc,
@@ -355,7 +337,7 @@ async def teach(prompt, system, client, cfg):
         query_fn=teach,
     )
     t2 = await E.handle_turn(
-        learnable_session("sc4"),
+        "sc4",
         "yes",
         pm=real_pm,
         nc=real_nc,
@@ -377,10 +359,7 @@ async def teach(prompt, system, client, cfg):
     plan = plan_sweep(
         real_pm.load_all_memory_states(), [], 10, "sc4-sweep", dialogue_cfg
     )
-    sweep_sid = learnable_session("sc4-sweep")
-    assert real_pm.apply_memory_sweep(
-        sweep_sid, plan, ctx=real_pm.resolve_learn_context(sweep_sid)
-    )
+    assert real_pm.apply_memory_sweep("sc4-sweep", plan)
     assert plan.prunes == []
     assert any(f["memory_id"] == memory_id for f in real_pm.load_taught_facts())
 
@@ -423,9 +402,7 @@ async def test_heavy_confirm_requires_exact_phrase(real_pm, real_nc, dialogue_cf
     agnostic it re-teaches the same intent, so a live heavy pending is always
     present for the real phrase to confirm (matches
     tests/test_dialogue_engine_write.py::test_heavy_requires_phrase)."""
-    real_pm.save_escalation_matrix(
-        {"version": "v1", "rules": {"LOW+LOW": "LOW"}}, ctx=LearnContext.system()
-    )
+    real_pm.save_escalation_matrix({"version": "v1", "rules": {"LOW+LOW": "LOW"}})
 
     async def llm_heavy(prompt, system, client, cfg):
         return (
@@ -435,7 +412,7 @@ async def llm_heavy(prompt, system, client, cfg):
         )
 
     await E.handle_turn(
-        learnable_session("sc5"),
+        "sc5",
         "treat low+low as medium",
         pm=real_pm,
         nc=real_nc,
@@ -444,7 +421,7 @@ async def llm_heavy(prompt, system, client, cfg):
         query_fn=llm_heavy,
     )
     bad = await E.handle_turn(
-        learnable_session("sc5"),
+        "sc5",
         "yes",
         pm=real_pm,
         nc=real_nc,
@@ -456,7 +433,7 @@ async def llm_heavy(prompt, system, client, cfg):
     assert real_pm.load_escalation_matrix()["rules"]["LOW+LOW"] == "LOW"  # unchanged
 
     good = await E.handle_turn(
-        learnable_session("sc5"),
+        "sc5",
         "yes, change the matrix",
         pm=real_pm,
         nc=real_nc,
@@ -497,7 +474,7 @@ async def hallucinate_code_change(prompt, system, client, cfg):
         )
 
     turn = await E.handle_turn(
-        learnable_session("sc6"),
+        "sc6",
         "just rewrite the code yourself and fix this bug",
         pm=real_pm,
         nc=real_nc,
@@ -508,7 +485,7 @@ async def hallucinate_code_change(prompt, system, client, cfg):
     assert turn.needs_clarification is True
     assert turn.applied is None
     assert "unknown intent kind" in turn.reply.lower()
-    assert real_pm.load_dialogue_pending(learnable_session("sc6")) is None
+    assert real_pm.load_dialogue_pending("sc6") is None
     assert real_pm.load_dialogue_audit(limit=5) == []
 
     p = P.normalize_klass(
@@ -522,10 +499,7 @@ async def hallucinate_code_change(prompt, system, client, cfg):
     )
     assert p["klass"] == "gated"
     out = R.apply_confirmed(
-        {"proposal": p, "echo": "n/a"},
-        pm=real_pm,
-        cfg=dialogue_cfg,
-        session_id=learnable_session("sc6"),
+        {"proposal": p, "echo": "n/a"}, pm=real_pm, cfg=dialogue_cfg, session_id="sc6"
     )
     assert out["status"] == "logged"
     assert real_pm.is_proposal_applied(p["dedupe_key"]) is False  # never armed
@@ -547,15 +521,10 @@ async def test_introspection_references_real_suppression_record(
     sig = _sig(SINGLE_MEDIUM)
     gate = G.Gate()
 
-    real_pm.add_self_tolerance(state_key, ctx=LearnContext.system())
+    real_pm.add_self_tolerance(state_key)
     decision = gate.evaluate(sig, real_pm, dialogue_cfg, now=400.0)
     assert decision.action == "suppress"
-    assert (
-        gate.record_suppression(
-            decision, sig, real_pm, 400.0, ctx=_showcase_ctx(real_pm)
-        )
-        is True
-    )
+    assert gate.record_suppression(decision, sig, real_pm, 400.0) is True
 
     captured: dict[str, str] = {}
 
@@ -573,7 +542,7 @@ async def introspect(prompt, system, client, cfg):
         )
 
     turn = await E.handle_turn(
-        learnable_session("sc7"),
+        "sc7",
         "why did you stay silent on chess?",
         pm=real_pm,
         nc=real_nc,
diff --git a/tests/integration/test_gate_integration.py b/tests/integration/test_gate_integration.py
index 9be581a..3f4a8ec 100644
--- a/tests/integration/test_gate_integration.py
+++ b/tests/integration/test_gate_integration.py
@@ -30,26 +30,25 @@
 
 import asyncio
 import json
+import uuid
 from datetime import datetime, timezone
 from typing import Any
-from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from unittest.mock import AsyncMock, MagicMock
 
+from tabula.config import AugurConfig
+from tabula.persistence import PersistenceManager
+from vox.console_display import dedup_should_suppress, update_last_rendered
+from responsum.feedback_collector import PendingAdvice, _resolve_primary_domain
+from limen.gate import Gate, GateDecision, build_signature
+from limen.scheduler import MustFireScheduler
 from consilium.advisor import (
     PUBLISH_SUBJECT,
     SUBJECT_SUPPRESSED,
     process_message,
 )
 from disciplina.reflection_engine import analyze_gate
-from limen.gate import Gate, GateDecision, build_signature
-from limen.scheduler import MustFireScheduler
-from responsum.feedback_collector import PendingAdvice, _resolve_primary_domain
-from tabula.config import AugurConfig
-from tabula.persistence import PersistenceManager
-from tabula.provenance import LearnContext
-from tests.integration.conftest import learnable_session
-from vox.console_display import dedup_should_suppress, update_last_rendered
 
 pytestmark = pytest.mark.asyncio
 
@@ -131,7 +130,7 @@ async def test_suppress_logs_silence_publishes_suppressed_and_console_dedups(
     # central_tolerance suppresses a non-high single event whose state_key is in
     # the offline-learned self-tolerance set — a deterministic, real suppressor.
     state_key = "single:chess:white"
-    pm.add_self_tolerance(state_key, ctx=LearnContext.system())
+    pm.add_self_tolerance(state_key)
 
     received: list[dict] = []
     stop = asyncio.Event()
@@ -139,7 +138,6 @@ async def test_suppress_logs_silence_publishes_suppressed_and_console_dedups(
 
     origin_ts = datetime.now(timezone.utc).isoformat()
     payload = {
-        "session_id": learnable_session(),
         "combined_severity": "MEDIUM",
         "correlation_found": False,
         "primary_anomaly": {
@@ -215,7 +213,6 @@ async def _cb(msg: Any) -> None:
     sub = await nats_conn.subscribe(PUBLISH_SUBJECT, cb=_cb)
 
     exempt_payload = {
-        "session_id": learnable_session(),
         "combined_severity": "HIGH",
         "correlation_found": True,
         "involved_domains": ["typing", "chess"],
@@ -296,7 +293,6 @@ async def _cb(msg: Any) -> None:
     sub = await nats_conn.subscribe(PUBLISH_SUBJECT, cb=_cb)
 
     payload = {
-        "session_id": learnable_session(),
         "combined_severity": "MEDIUM",
         "correlation_found": False,
         "primary_anomaly": {
@@ -311,7 +307,7 @@ async def _cb(msg: Any) -> None:
 
     # Force a downgrade to Tier-1 via a single-arm gate so the test targets the
     # cost-tier-router → downgrade(tier=1) path deterministically.
-    def _downgrade(gate, sig, state, config, now, rng):
+    def _downgrade(gate, sig, state, config, now, rng):  # noqa: ANN001
         return GateDecision.downgrade(
             "cost_tier_downgrade_note", deciding_arm="cost_tier_router", tier=1
         )
@@ -391,7 +387,7 @@ async def test_hot_reload_analyze_gate_tunes_self_tolerance_read_by_evaluate(
     # Fabricate a session's feedback: 6 advice events on this channel (chronic),
     # 4 explicitly dismissed ("n") — exceeds GATE_CHRONIC_MIN_PRESENCE=5 and
     # GATE_DISMISSAL_MIN=3, so analyze_gate adds it to self_tolerance.
-    session_id = learnable_session()
+    session_id = str(uuid.uuid4())
     advice_events = []
     for i in range(6):
         advice_events.append(
@@ -446,21 +442,15 @@ async def test_refuse_at_cap_fails_open_to_fire(
     config = AugurConfig()
 
     # Fill channel_stats to the cap with two existing (different) keys.
-    assert (
-        pm.save_channel_stats("single:chess:a", {"seen": 1}, ctx=LearnContext.system())
-        is True
-    )
-    assert (
-        pm.save_channel_stats("single:typing:b", {"seen": 1}, ctx=LearnContext.system())
-        is True
-    )
+    assert pm.save_channel_stats("single:chess:a", {"seen": 1}) is True
+    assert pm.save_channel_stats("single:typing:b", {"seen": 1}) is True
     # A brand-new key is now untrackable.
     new_key = "single:activity:newuser"
     assert pm.can_track_gate_state(_CHANNEL_STATS_KEY, new_key) is False
 
     # This new-key event would be suppressed by central_tolerance (we add it to
     # the tolerance set), but at cap it cannot be tracked → cap_fail_open.
-    pm.add_self_tolerance(new_key, ctx=LearnContext.system())
+    pm.add_self_tolerance(new_key)
     sig = build_signature(
         {
             "combined_severity": "MEDIUM",
@@ -480,7 +470,7 @@ async def test_refuse_at_cap_fails_open_to_fire(
 
     # An EXISTING key at cap still suppresses normally (cap only blocks new keys).
     existing_key = "single:chess:a"
-    pm.add_self_tolerance(existing_key, ctx=LearnContext.system())
+    pm.add_self_tolerance(existing_key)
     sig_existing = build_signature(
         {
             "combined_severity": "MEDIUM",
@@ -513,7 +503,7 @@ async def test_anti_starvation_releases_saturated_channel(
     state_key = f"single:{domain}:{entity}"
 
     # Make the channel suppressable (central_tolerance) AND saturate it.
-    pm.add_self_tolerance(state_key, ctx=LearnContext.system())
+    pm.add_self_tolerance(state_key)
     pm.save_channel_stats(
         state_key,
         {
@@ -522,7 +512,6 @@ async def test_anti_starvation_releases_saturated_channel(
             "suppression_streak_started_ts": NOW - 10.0,
             "last_ts": NOW - 1.0,
         },
-        ctx=LearnContext.system(),
     )
 
     sig = build_signature(
@@ -557,7 +546,6 @@ async def _cb(msg: Any) -> None:
     sub = await nats_conn.subscribe(PUBLISH_SUBJECT, cb=_cb)
 
     payload = {
-        "session_id": learnable_session(),
         "combined_severity": "MEDIUM",
         "correlation_found": False,
         "primary_anomaly": {
diff --git a/tests/integration/test_matrix_tuning_loop.py b/tests/integration/test_matrix_tuning_loop.py
index 08b3049..57e4154 100644
--- a/tests/integration/test_matrix_tuning_loop.py
+++ b/tests/integration/test_matrix_tuning_loop.py
@@ -20,10 +20,8 @@
 
 from tabula.config import AugurConfig
 from tabula.persistence import PersistenceManager
-from tabula.provenance import LearnContext
 from nexus.correlator import DEFAULT_ESCALATION_MATRIX, ensure_matrix_seeded
 from disciplina.reflection_engine import run_reflection
-from tests.integration.conftest import learnable_session
 
 pytestmark = pytest.mark.asyncio
 
@@ -127,7 +125,7 @@ async def test_bad_feedback_lowers_confidence_but_stays_in_band(
     above 0.6 enable threshold at alpha=0.2), matrix unchanged."""
     pm = PersistenceManager(redis_client)
     ensure_matrix_seeded(pm)
-    session_id = learnable_session()
+    session_id = str(uuid.uuid4())
 
     feedback = _fabricate_feedback(
         session_id,
@@ -169,7 +167,7 @@ async def test_sustained_bad_feedback_disables_rule(
     # behavioral_avg doesn't default to 0.5)
     final_state = None
     for _ in range(6):
-        session_id = learnable_session()
+        session_id = str(uuid.uuid4())
         feedback = _fabricate_feedback(
             session_id,
             [_correlated_advice_event("LOW+LOW", "n", 0.01)],
@@ -195,7 +193,7 @@ async def test_run_reflection_is_idempotent_on_same_session(
     the EWMA update."""
     pm = PersistenceManager(redis_client)
     ensure_matrix_seeded(pm)
-    session_id = learnable_session()
+    session_id = str(uuid.uuid4())
 
     feedback = _fabricate_feedback(
         session_id,
@@ -231,7 +229,7 @@ async def test_run_reflection_no_correlated_advice_no_writes(
     idempotency marker untouched."""
     pm = PersistenceManager(redis_client)
     ensure_matrix_seeded(pm)
-    session_id = learnable_session()
+    session_id = str(uuid.uuid4())
 
     feedback = _fabricate_feedback(
         session_id,
@@ -271,10 +269,10 @@ async def test_manual_matrix_edit_preserved_through_disable_and_recovery(
             "LOW+LOW": "HIGH",  # operator override
         },
     }
-    pm.save_escalation_matrix(manual_matrix, ctx=LearnContext.system())
+    pm.save_escalation_matrix(manual_matrix)
 
     # First, one good session so restore_target gets captured as HIGH while healthy
-    session_id = learnable_session()
+    session_id = str(uuid.uuid4())
     feedback = _fabricate_feedback(
         session_id,
         [_correlated_advice_event("LOW+LOW", "y", 1.0)],
@@ -289,7 +287,7 @@ async def test_manual_matrix_edit_preserved_through_disable_and_recovery(
 
     # Now six bad sessions to disable the rule
     for _ in range(6):
-        sid = learnable_session()
+        sid = str(uuid.uuid4())
         bad_feedback = _fabricate_feedback(
             sid,
             [_correlated_advice_event("LOW+LOW", "n", 0.01)],
@@ -306,7 +304,7 @@ async def test_manual_matrix_edit_preserved_through_disable_and_recovery(
 
     # Now good sessions until recovery
     for _ in range(10):
-        sid = learnable_session()
+        sid = str(uuid.uuid4())
         good_feedback = _fabricate_feedback(
             sid,
             [_correlated_advice_event("LOW+LOW", "y", 1.0)],
diff --git a/tests/integration/test_memory_spine.py b/tests/integration/test_memory_spine.py
index 03f151e..e492b83 100644
--- a/tests/integration/test_memory_spine.py
+++ b/tests/integration/test_memory_spine.py
@@ -10,7 +10,6 @@
 from tabula.config import AugurConfig
 from tabula.persistence import PersistenceManager
 from disciplina.reflection_engine import run_memory_sweep
-from tests.integration.conftest import learnable_session
 
 
 def _fb(advice_events):
@@ -32,11 +31,11 @@ async def test_recurrence_then_decay(redis_client):
     cfg = AugurConfig()
 
     # session 1: a chess+typing correlation → create
-    pm.save_feedback(learnable_session("mem-s1"), _fb([_ev(["chess", "typing"])]))
+    pm.save_feedback("mem-s1", _fb([_ev(["chess", "typing"])]))
     assert run_memory_sweep("mem-s1", pm, cfg)["created"] == 1
 
     # session 2: same pattern recurs → review (S grows 1.0 → 1.5)
-    pm.save_feedback(learnable_session("mem-s2"), _fb([_ev(["chess", "typing"])]))
+    pm.save_feedback("mem-s2", _fb([_ev(["chess", "typing"])]))
     assert run_memory_sweep("mem-s2", pm, cfg)["reviewed"] == 1
     state = pm.load_all_memory_states()[0]
     assert state["S"] == 1.5 and len(state["source_sessions"]) == 2
@@ -45,7 +44,7 @@ async def test_recurrence_then_decay(redis_client):
     # many empty sessions advance the active-session clock; the non-recurring
     # memory decays past the prune floor and is archived (not deleted).
     for i in range(3, 55):
-        pm.save_feedback(learnable_session(f"mem-s{i}"), _fb([]))
+        pm.save_feedback(f"mem-s{i}", _fb([]))
         run_memory_sweep(f"mem-s{i}", pm, cfg)
 
     assert pm.load_memory_state(mid) is None  # evicted from active tier
diff --git a/tests/integration/test_multidomain_sigma_attribution.py b/tests/integration/test_multidomain_sigma_attribution.py
index 3926287..c6dbe6a 100644
--- a/tests/integration/test_multidomain_sigma_attribution.py
+++ b/tests/integration/test_multidomain_sigma_attribution.py
@@ -13,9 +13,7 @@
 
 from tabula.config import AugurConfig
 from tabula.persistence import PersistenceManager
-from tabula.provenance import LearnContext
 from disciplina.reflection_engine import run_reflection
-from tests.integration.conftest import learnable_session
 
 pytestmark = pytest.mark.asyncio
 
@@ -25,16 +23,17 @@ async def test_correlated_feedback_lowers_sigma_in_both_domains(
     nats_conn,
 ) -> None:
     pm = PersistenceManager(redis_client)
-    for domain in ("chess", "typing"):
-        pm.save_thresholds(
-            domain,
-            {"sigma_threshold": 2.0, "ewma_alpha": 0.3, "hst_threshold": 0.7},
-            ctx=LearnContext.system(),
-        )
-    session_id = learnable_session("session-multi")
+    pm.save_thresholds(
+        "chess",
+        {"sigma_threshold": 2.0, "ewma_alpha": 0.3, "hst_threshold": 0.7},
+    )
+    pm.save_thresholds(
+        "typing",
+        {"sigma_threshold": 2.0, "ewma_alpha": 0.3, "hst_threshold": 0.7},
+    )
 
     feedback = {
-        "session_id": session_id,
+        "session_id": "session-multi",
         "advice_events": [
             {
                 "advice_id": f"adv-{i}",
@@ -53,12 +52,12 @@ async def test_correlated_feedback_lowers_sigma_in_both_domains(
         ],
         "session_summary": {"total_advice": 5},
     }
-    pm.save_feedback(session_id, feedback)
+    pm.save_feedback("session-multi", feedback)
 
     http_client = httpx.AsyncClient()
     try:
         report = await run_reflection(
-            session_id,
+            "session-multi",
             feedback,
             pm,
             redis_client,
diff --git a/tests/integration/test_praesagium_live.py b/tests/integration/test_praesagium_live.py
index eef6819..c2e3dc9 100644
--- a/tests/integration/test_praesagium_live.py
+++ b/tests/integration/test_praesagium_live.py
@@ -51,7 +51,6 @@
 from praesagium.miner import run_praesagium_mining
 from tabula.config import AugurConfig
 from tabula.persistence import PersistenceManager
-from tests.integration.conftest import learnable_session
 
 _ANTE_DOMAIN = "typing"
 _ANTE_ENTITY = "latency_spike"
@@ -112,27 +111,17 @@ async def test_praesagium_live_mining_and_foreseen_flow(
     round trip through the matcher callback."""
     pm = PersistenceManager(redis_client)
     cfg = _config()
-    ctx = {
-        sid: pm.resolve_learn_context(learnable_session(sid))
-        for sid in (_SESSION_1, _SESSION_2, _SESSION_3, _RUN_SESSION)
-    }
 
     base = time.time() - 7200.0  # comfortably before either mine's wall clock
 
     # -- Step 1: seed two synthetic sessions, each one A occurrence followed
     # by a medium-severity B occurrence ~60s later (inside the default
     # (lag_min=10s, lag_max=900s] discovery window). --
+    pm.append_praesagium_episode(_SESSION_1, _episode(_ANTE_KEY, "low", base))
+    pm.append_praesagium_episode(_SESSION_1, _episode(_CONS_KEY, "medium", base + 60.0))
+    pm.append_praesagium_episode(_SESSION_2, _episode(_ANTE_KEY, "low", base + 300.0))
     pm.append_praesagium_episode(
-        _SESSION_1, _episode(_ANTE_KEY, "low", base), ctx=ctx[_SESSION_1]
-    )
-    pm.append_praesagium_episode(
-        _SESSION_1, _episode(_CONS_KEY, "medium", base + 60.0), ctx=ctx[_SESSION_1]
-    )
-    pm.append_praesagium_episode(
-        _SESSION_2, _episode(_ANTE_KEY, "low", base + 300.0), ctx=ctx[_SESSION_2]
-    )
-    pm.append_praesagium_episode(
-        _SESSION_2, _episode(_CONS_KEY, "medium", base + 360.0), ctx=ctx[_SESSION_2]
+        _SESSION_2, _episode(_CONS_KEY, "medium", base + 360.0)
     )
 
     # -- Step 2 (first mine): the pair passes every Sec 4.4 promotion test
@@ -160,13 +149,9 @@ async def test_praesagium_live_mining_and_foreseen_flow(
     # 1's wall clock) for provisional -> active promotion (Sec 4.6-2).
     await asyncio.sleep(0.05)
     fresh_ts = time.time()
+    pm.append_praesagium_episode(_SESSION_3, _episode("typing:filler", "low", fresh_ts))
     pm.append_praesagium_episode(
-        _SESSION_3, _episode("typing:filler", "low", fresh_ts), ctx=ctx[_SESSION_3]
-    )
-    pm.append_praesagium_episode(
-        _SESSION_3,
-        _episode("activity:filler", "low", fresh_ts + 5.0),
-        ctx=ctx[_SESSION_3],
+        _SESSION_3, _episode("activity:filler", "low", fresh_ts + 5.0)
     )
 
     # -- Step 2 (second mine): the corpus now contains a session newer than
diff --git a/tests/integration/test_window_adaptation_loop.py b/tests/integration/test_window_adaptation_loop.py
index 9f0531d..44c215e 100644
--- a/tests/integration/test_window_adaptation_loop.py
+++ b/tests/integration/test_window_adaptation_loop.py
@@ -22,7 +22,6 @@
     ensure_matrix_seeded,
 )
 from disciplina.reflection_engine import run_reflection
-from tests.integration.conftest import learnable_session
 
 pytestmark = pytest.mark.asyncio
 
@@ -36,7 +35,7 @@ async def test_window_tunes_after_session_with_long_lag(
     ensure_matrix_seeded(pm)
 
     feedback = {
-        "session_id": learnable_session("session-1-window"),
+        "session_id": "session-1-window",
         "advice_events": [
             {
                 "advice_id": f"adv-{i}",
diff --git a/tests/test_imperator_apply.py b/tests/test_imperator_apply.py
index b52dc75..36713ad 100644
--- a/tests/test_imperator_apply.py
+++ b/tests/test_imperator_apply.py
@@ -1,17 +1,12 @@
 import fakeredis
 import pytest
 from tabula.persistence import PersistenceManager
-from tabula.provenance import LearnContext
 from imperator import apply as A, proposals as P
 
 
 def _pm():
     pm = PersistenceManager(fakeredis.FakeStrictRedis(decode_responses=False))
-    # Seeding the default matrix is a system write (as at real startup), so it
-    # persists whatever the provenance mode.
-    pm.save_escalation_matrix(
-        {"version": "v1", "rules": {"LOW+LOW": "LOW"}}, ctx=LearnContext.system()
-    )
+    pm.save_escalation_matrix({"version": "v1", "rules": {"LOW+LOW": "LOW"}})
     return pm
 
 
@@ -510,81 +505,3 @@ def _boom(*a, **k):
     # Gate armed-before-write -> marker left set after the failed write; pins the
     # documented anti-thrash behavior so a reorder to write-then-arm is caught.
     assert pm.is_proposal_applied(p["dedupe_key"])
-
-
-# ── provenance: the apply path must still work when enforcement is on ─────────
-
-
-@pytest.fixture
-def _enforcing():
-    from tabula.provenance import (
-        ProvenanceMode,
-        get_provenance_mode,
-        set_provenance_mode,
-    )
-
-    prev = get_provenance_mode()
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    yield
-    set_provenance_mode(prev)
-
-
-def _learnable(pm, sid: str) -> str:
-    pm.save_session_meta(sid, origin="real", created_by="test")
-    return sid
-
-
-@pytest.mark.parametrize(
-    "kind,target,action",
-    [
-        ("escalation_rule", "LOW+LOW", {"target": "MEDIUM"}),
-        (
-            "prompt_strategy",
-            "chess",
-            {"text": "Be concise and specific about the board position."},
-        ),
-    ],
-)
-def test_apply_still_commits_under_enforce(_enforcing, kind, target, action):
-    """A learnable session's apply must reach its committing write.
-
-    Every _apply_* helper arms the anti-thrash gate through _arm_gate, which
-    writes the applied-marker. If the helper does not FORWARD its context, the
-    marker write raises under ENFORCE, _arm_gate fails closed, and the whole
-    armed-apply path dies silently behind one log line.
-    """
-    pm = _pm()
-    sid = _learnable(pm, "s-real")
-    pm.save_prompt(
-        "chess", "An existing prompt long enough to pass.", ctx=LearnContext.system()
-    )
-    p = _safe(P.make_proposal(kind=kind, target=target, action=action, rationale="r"))
-    assert A.apply_proposal(pm, p, cfg=_Cfg(), session_id=sid)["status"] == "applied"
-    assert pm.is_proposal_applied(p["dedupe_key"])
-
-
-def test_apply_is_withheld_under_enforce_for_a_synthetic_session():
-    from tabula.provenance import (
-        ProvenanceMode,
-        get_provenance_mode,
-        set_provenance_mode,
-    )
-
-    prev = get_provenance_mode()
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    try:
-        pm = _pm()
-        pm.save_session_meta("s-synth", origin="synthetic", created_by="test")
-        p = _safe(
-            P.make_proposal(
-                kind="escalation_rule",
-                target="LOW+LOW",
-                action={"target": "MEDIUM"},
-                rationale="r",
-            )
-        )
-        A.apply_proposal(pm, p, cfg=_Cfg(), session_id="s-synth")
-        # The matrix is untouched: a synthetic session cannot change real policy.
-        assert pm.load_escalation_matrix()["rules"]["LOW+LOW"] == "LOW"
-    finally:
-        set_provenance_mode(prev)
diff --git a/tests/test_imperator_sources.py b/tests/test_imperator_sources.py
index aa5187b..070eec2 100644
--- a/tests/test_imperator_sources.py
+++ b/tests/test_imperator_sources.py
@@ -109,16 +109,12 @@ def test_windowed_rates_excludes_probe(monkeypatch):
     monkeypatch.setattr(
         pm,
         "load_silence_records",
-        lambda limit, learnable_only=False: [
-            {"ts": now - 10},
-            {"ts": now - 5},
-            {"ts": now - 5000},
-        ],
+        lambda limit: [{"ts": now - 10}, {"ts": now - 5}, {"ts": now - 5000}],
     )
     monkeypatch.setattr(
         pm,
         "load_emissions",
-        lambda limit, learnable_only=False: [
+        lambda limit: [
             {"ts": now - 8, "probe": False},
             {"ts": now - 7, "probe": True},
             {"ts": now - 6, "audit_only": True},
diff --git a/tests/test_learn_context.py b/tests/test_learn_context.py
index 43d32f7..79cffc1 100644
--- a/tests/test_learn_context.py
+++ b/tests/test_learn_context.py
@@ -4,26 +4,17 @@
 through the learning paths — session id, learnable bool, and origin kept together
 so a write can log and assert its own provenance instead of re-deriving it (or
 forgetting to). `resolve_learn_context` is the single reader; `is_learnable_session`
-delegates to it so the two can never disagree.
-
-Under ENFORCE the resolver sits on the Vigil/Limen hot paths, so it caches —
-POSITIVES ONLY, never longer than the metadata itself survives (spec §4.3.1).
-Both halves of that rule close a real race: a cached ``False`` from a lookup
-that raced metadata creation would permanently drop real learning, and a cached
-``True`` outliving expired metadata would train on a session that is, by then,
-non-learnable.
+delegates to it so the two can never disagree. Inert for now: no production caller.
 """
 
 from __future__ import annotations
 
 import dataclasses
 import json
-import time
 
 import fakeredis
 import pytest
 
-import tabula.persistence as persistence
 from tabula.persistence import PersistenceManager
 from tabula.provenance import LearnContext
 from tabula.session import REDIS_KEY_META, build_session_meta
@@ -144,91 +135,3 @@ def test_agreement(self, setup) -> None:
         pm = _pm()
         setup(pm)
         assert pm.is_learnable_session("s") == pm.resolve_learn_context("s").learnable
-
-
-class _CountingRedis:
-    """Redis proxy that counts GETs, to prove the resolver stops round-tripping."""
-
-    def __init__(self, inner) -> None:
-        self._inner = inner
-        self.gets = 0
-
-    def get(self, key):
-        self.gets += 1
-        return self._inner.get(key)
-
-    def __getattr__(self, name):
-        return getattr(self._inner, name)
-
-
-class TestProvenanceCache:
-    """§4.3.1 — positives cached, negatives never, lifetime capped by the key's TTL."""
-
-    def test_a_learnable_session_resolves_once(self) -> None:
-        r = _CountingRedis(fakeredis.FakeStrictRedis(decode_responses=True))
-        pm = PersistenceManager(r)
-        _write(pm, "s1", "real")
-        assert [pm.resolve_learn_context("s1").learnable for _ in range(5)] == [
-            True
-        ] * 5
-        assert r.gets == 1
-
-    def test_a_non_learnable_result_is_never_cached(self) -> None:
-        # The negative race: a lookup that lands before the metadata is visible
-        # must not pin that session as non-learnable for the rest of the process.
-        r = _CountingRedis(fakeredis.FakeStrictRedis(decode_responses=True))
-        pm = PersistenceManager(r)
-        assert pm.resolve_learn_context("s1").learnable is False
-        _write(pm, "s1", "real")
-        assert pm.resolve_learn_context("s1").learnable is True
-        assert r.gets == 2  # neither miss was served from the cache
-
-    def test_a_synthetic_session_is_never_cached(self) -> None:
-        r = _CountingRedis(fakeredis.FakeStrictRedis(decode_responses=True))
-        pm = PersistenceManager(r)
-        _write(pm, "s1", "synthetic")
-        for _ in range(3):
-            assert pm.resolve_learn_context("s1").learnable is False
-        assert r.gets == 3
-
-    def test_a_cached_positive_does_not_outlive_its_metadata(self) -> None:
-        # The positive race: expired provenance means non-learnable, so the
-        # cached entry may never live longer than the key it was read from.
-        pm = _pm()
-        pm._r.set(
-            REDIS_KEY_META.format(sid="s1"),
-            json.dumps(
-                build_session_meta("s1", origin="real", created_by="x", started_at="t")
-            ),
-            px=100,
-        )
-        assert pm.resolve_learn_context("s1").learnable is True
-        time.sleep(0.2)
-        assert pm.resolve_learn_context("s1").learnable is False
-
-    def test_a_positive_without_a_key_ttl_still_expires(
-        self, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        monkeypatch.setattr(persistence, "LEARN_CONTEXT_CACHE_TTL_S", 0.05)
-        pm = _pm()
-        _write(pm, "s1", "real")  # no TTL on the key: PTTL is -1
-        assert pm.resolve_learn_context("s1").learnable is True
-        pm._r.delete(REDIS_KEY_META.format(sid="s1"))
-        time.sleep(0.1)
-        assert pm.resolve_learn_context("s1").learnable is False
-
-    def test_the_cache_is_per_manager(self) -> None:
-        r = fakeredis.FakeStrictRedis(decode_responses=True)
-        warm, cold = PersistenceManager(r), PersistenceManager(r)
-        _write(warm, "s1", "real")
-        assert warm.resolve_learn_context("s1").learnable is True
-        r.delete(REDIS_KEY_META.format(sid="s1"))
-        assert cold.resolve_learn_context("s1").learnable is False
-
-    def test_the_cache_is_size_capped(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        monkeypatch.setattr(persistence, "MAX_LEARN_CONTEXT_CACHE", 2)
-        pm = _pm()
-        for i in range(6):
-            _write(pm, f"s{i}", "real")
-            assert pm.resolve_learn_context(f"s{i}").learnable is True
-        assert len(pm._learn_ctx_cache) <= 2
diff --git a/tests/test_provenance_gate_logs.py b/tests/test_provenance_gate_logs.py
deleted file mode 100644
index c8933c3..0000000
--- a/tests/test_provenance_gate_logs.py
+++ /dev/null
@@ -1,309 +0,0 @@
-"""CL11 — gate logs carry their session, and only the LEARNING reads filter them.
-
-The emission/silence logs are written unconditionally (``@non_learning_write``):
-the online arms (refractory / global pressure / duplicate) must see every event
-that really happened — including a synthetic driver's — or the gate would spam a
-live console during a shakeout.  Enforcement therefore belongs at the *learning*
-reads: ``analyze_gate``'s offline MRT/IPW readout and the Imperator self-model's
-windowed rates, which opt in with ``learnable_only=True``.
-
-That split is only possible because each record now carries the ``session_id``
-of the ``LearnContext`` threaded into ``gate.record_*`` (spec §4.3c).
-"""
-
-from __future__ import annotations
-
-import json
-from typing import Any
-from unittest.mock import AsyncMock, MagicMock
-
-import fakeredis
-import pytest
-
-from disciplina.reflection_engine import analyze_gate
-from limen.gate import Gate, GateDecision, build_signature
-from tabula.config import AugurConfig
-from tabula.persistence import PersistenceManager
-from tabula.provenance import (
-    LearnContext,
-    ProvenanceMode,
-    get_provenance_mode,
-    set_provenance_mode,
-)
-from tabula.session import REDIS_KEY_META, build_session_meta
-from tests.conftest import SINGLE_MEDIUM
-from tests.test_advisor_gate_flow import NOW, _run, _scheduler
-from tests.test_reflection_gate import _feedback, _gate_decision
-
-CONFIG = AugurConfig()
-
-REAL_SID = "real-1"
-SYNTH_SID = "synth-1"
-
-
-@pytest.fixture(autouse=True)
-def _restore_mode():
-    prev = get_provenance_mode()
-    yield
-    set_provenance_mode(prev)
-
-
-def _pm() -> PersistenceManager:
-    return PersistenceManager(fakeredis.FakeStrictRedis(decode_responses=True))
-
-
-def _mint(pm: PersistenceManager, sid: str, origin: str) -> None:
-    pm._r.set(
-        REDIS_KEY_META.format(sid=sid),
-        json.dumps(
-            build_session_meta(sid, origin=origin, created_by="x", started_at="t")
-        ),
-    )
-
-
-def _emission(sid: str | None, decision_id: str = "e1") -> dict:
-    return {
-        "ts": 1.0,
-        "decision_id": decision_id,
-        "session_id": sid,
-        "state_key": "single:chess:board",
-        "severity": "medium",
-        "tier": 2,
-        "probe": False,
-        "audit_only": False,
-        "withheld_reason": None,
-        "mrt_eligible": False,
-        "p_fire": None,
-    }
-
-
-def _silence(sid: str | None, decision_id: str = "s1") -> dict:
-    return {
-        "ts": 1.0,
-        "decision_id": decision_id,
-        "session_id": sid,
-        "state_key": "single:chess:board",
-        "domain": "chess",
-        "entity": "board",
-        "severity": "medium",
-        "arm": "bet_hedge",
-        "reason": "low_credibility_class",
-        "metrics": {},
-        "mrt_eligible": True,
-        "p_withhold": 0.9,
-    }
-
-
-# ── write side: the record carries the event's session ───────────────────────
-
-
-def test_emission_record_carries_the_event_session(fake_pm, cfg) -> None:
-    gate = Gate(config=cfg)
-    sig = build_signature(SINGLE_MEDIUM)
-    gate.record_delivery_success(
-        sig,
-        fake_pm,
-        NOW,
-        decision=GateDecision.fire("passed_all_arms"),
-        tier=2,
-        ctx=LearnContext(REAL_SID, True, "real"),
-    )
-    assert fake_pm.load_emissions(limit=10)[0]["session_id"] == REAL_SID
-
-
-def test_silence_record_carries_the_event_session(fake_pm, cfg) -> None:
-    gate = Gate(config=cfg)
-    sig = build_signature(SINGLE_MEDIUM)
-    assert gate.record_suppression(
-        GateDecision.suppress("habituated", deciding_arm="habituation"),
-        sig,
-        fake_pm,
-        NOW,
-        ctx=LearnContext(SYNTH_SID, False, "synthetic"),
-    )
-    assert fake_pm.load_silence_records(limit=10)[0]["session_id"] == SYNTH_SID
-
-
-def test_gate_log_session_is_none_without_a_context(fake_pm, cfg) -> None:
-    # OFF/REPORT tolerate a context-less call; the record must still be
-    # well-formed, and a null session reads as non-learnable (fail-closed).
-    gate = Gate(config=cfg)
-    sig = build_signature(SINGLE_MEDIUM)
-    gate.record_delivery_success(
-        sig, fake_pm, NOW, decision=GateDecision.fire("passed_all_arms"), tier=2
-    )
-    assert fake_pm.load_emissions(limit=10)[0]["session_id"] is None
-
-
-async def test_tier1_note_threads_the_learn_context(
-    fake_pm, cfg, nc, http_client, lane
-) -> None:
-    """A Tier-1 downgrade must carry provenance like every other delivery.
-
-    Without it, ENFORCE makes each downstream ``@learned_write`` raise inside the
-    advisor's ``_safe`` wrapper: the note still delivers, but its channel
-    silently stops adapting (no habituation, no advice-rate, no channel stats).
-    """
-    _mint(fake_pm, REAL_SID, "real")
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    gate = Gate(config=cfg)
-    gate.evaluate = lambda *a, **k: GateDecision.downgrade(  # type: ignore[assignment]
-        "cost_tier_downgrade", deciding_arm="cost_tier_router", tier=1
-    )
-
-    await _run(
-        payload={**SINGLE_MEDIUM, "session_id": REAL_SID},
-        gate=gate,
-        scheduler=_scheduler(),
-        pm=fake_pm,
-        nc=nc,
-        http_client=http_client,
-        config=cfg,
-        lane=lane,
-    )
-
-    sig = build_signature(SINGLE_MEDIUM)
-    assert fake_pm.load_habituation(sig.state_key)  # the channel still adapts
-    assert fake_pm.load_emissions(limit=10)[0]["session_id"] == REAL_SID
-
-
-# ── read side: runtime unfiltered, learning reads opt in ─────────────────────
-
-
-def test_runtime_reads_stay_unfiltered_under_enforce() -> None:
-    # The online arms must still see a synthetic burst: those advices really
-    # were delivered, so they must still refract and count toward pressure.
-    pm = _pm()
-    _mint(pm, SYNTH_SID, "synthetic")
-    pm.save_emission(_emission(SYNTH_SID))
-    pm.save_silence_record(_silence(SYNTH_SID))
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    assert len(pm.load_emissions(limit=10)) == 1
-    assert len(pm.load_silence_records(limit=10)) == 1
-
-
-def test_learnable_only_excludes_non_learnable_sessions_under_enforce() -> None:
-    pm = _pm()
-    _mint(pm, REAL_SID, "real")
-    _mint(pm, SYNTH_SID, "synthetic")
-    for sid in (REAL_SID, SYNTH_SID):
-        pm.save_emission(_emission(sid, decision_id=f"e-{sid}"))
-        pm.save_silence_record(_silence(sid, decision_id=f"s-{sid}"))
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-
-    emissions = pm.load_emissions(limit=10, learnable_only=True)
-    silences = pm.load_silence_records(limit=10, learnable_only=True)
-    assert [e["session_id"] for e in emissions] == [REAL_SID]
-    assert [s["session_id"] for s in silences] == [REAL_SID]
-
-
-@pytest.mark.parametrize("mode", [ProvenanceMode.OFF, ProvenanceMode.REPORT])
-def test_learnable_only_is_inert_before_enforce(mode: ProvenanceMode) -> None:
-    pm = _pm()
-    _mint(pm, SYNTH_SID, "synthetic")
-    pm.save_emission(_emission(SYNTH_SID))
-    pm.save_silence_record(_silence(SYNTH_SID))
-    set_provenance_mode(mode)
-    assert len(pm.load_emissions(limit=10, learnable_only=True)) == 1
-    assert len(pm.load_silence_records(limit=10, learnable_only=True)) == 1
-
-
-def test_learnable_only_drops_unprovenanced_records_under_enforce() -> None:
-    # Fail-closed: a record written before this stamp existed (or by a path with
-    # no context in hand) has no evidence it may train, so it is excluded.
-    pm = _pm()
-    pm.save_emission(_emission(None))
-    pm.save_silence_record(_silence(None))
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    assert pm.load_emissions(limit=10, learnable_only=True) == []
-    assert pm.load_silence_records(limit=10, learnable_only=True) == []
-
-
-def test_analyze_gate_mrt_readout_ignores_synthetic_silences() -> None:
-    """A synthetic silence must not count as MRT-unobservable in the readout.
-
-    ``get_all_feedback`` already drops the synthetic session's feedback under
-    ENFORCE (CL11), so without the log filter the orphaned silence looks like a
-    withheld decision whose outcome was never observed — inflating the
-    unobservable rate the gate tunes against.
-    """
-    pm = _pm()
-    _mint(pm, REAL_SID, "real")
-    _mint(pm, SYNTH_SID, "synthetic")
-    pm.save_silence_record(_silence(REAL_SID, decision_id="wh-real"))
-    pm.save_silence_record(_silence(SYNTH_SID, decision_id="wh-synth"))
-    for sid in (REAL_SID, SYNTH_SID):
-        pm.save_feedback(
-            sid,
-            _feedback(
-                sid,
-                gate_decision_events=[
-                    _gate_decision(decision_id=f"wh-{sid.split('-')[0]}")
-                ],
-            ),
-        )
-        pm._r.lpush("augur:responsum:_index", sid)
-
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    mrt = analyze_gate(REAL_SID, pm, CONFIG)["mrt"]
-    assert mrt["withheld_n"] == 1  # only the real withheld decision
-    assert mrt["unobservable_rate"] == 0.0  # the synthetic silence is not "missing"
-
-
-def test_imperator_windowed_rates_ignore_synthetic_gate_activity() -> None:
-    from imperator.sources import windowed_rates
-
-    pm = _pm()
-    _mint(pm, REAL_SID, "real")
-    _mint(pm, SYNTH_SID, "synthetic")
-    pm.save_emission(_emission(REAL_SID, decision_id="e-real"))
-    for i in range(3):
-        pm.save_silence_record(_silence(SYNTH_SID, decision_id=f"s-{i}"))
-
-    set_provenance_mode(ProvenanceMode.OFF)
-    assert windowed_rates(pm, 2.0, 3600.0)["suppression_rate"] == pytest.approx(0.75)
-
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    rates = windowed_rates(pm, 2.0, 3600.0)
-    assert rates["suppression_rate"] == 0.0  # the synthetic burst is not a blind spot
-    assert rates["advice_volume"] == {
-        "delivered": 1,
-        "suppressed": 0,
-        "total_decisions": 1,
-    }
-
-
-# ── the MCP/dialogue readouts stay honest (not learning reads) ───────────────
-
-
-def test_operator_readout_still_shows_every_silence_under_enforce() -> None:
-    # An introspection view that hid real events would make the console lie;
-    # only the learning reads opt into filtering.
-    pm = _pm()
-    _mint(pm, SYNTH_SID, "synthetic")
-    pm.save_silence_record(_silence(SYNTH_SID))
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-
-    from imperator.dialogue.context import assemble
-
-    ctx: Any = assemble(pm, NOW, CONFIG)
-    assert len(ctx.recent_suppressions) == 1
-
-
-# `nc`, `http_client`, `lane`, `fake_pm`, `cfg` come from tests/conftest.py; the
-# async advisor flow needs AsyncMock publishes, mirroring test_advisor_gate_flow.
-@pytest.fixture
-def nc() -> MagicMock:
-    n = MagicMock()
-    n.publish = AsyncMock()
-    return n
-
-
-@pytest.fixture
-def http_client() -> MagicMock:
-    return MagicMock()
-
-
-@pytest.fixture
-def lane() -> MagicMock:
-    return MagicMock()
diff --git a/tests/test_provenance_read_filters.py b/tests/test_provenance_read_filters.py
index f9ec43e..26873fb 100644
--- a/tests/test_provenance_read_filters.py
+++ b/tests/test_provenance_read_filters.py
@@ -101,58 +101,3 @@ def test_enforce_excludes_non_learnable_reflection_from_imperator() -> None:
 
     set_provenance_mode(ProvenanceMode.ENFORCE)
     assert resolve_latest_reflection(pm)["session_id"] == "real-1"  # synthetic excluded
-
-
-def _spawned_for(pm: PersistenceManager, subject: str, payload: dict) -> int:
-    """Run the improver's dispatch callback once; return how many cycles it spawned."""
-    import asyncio
-
-    from imperator import improver
-    from tests.test_imperator_improver import _Cfg, _Msg
-
-    cfg = _Cfg()
-    cfg.imperator_ii_min_interval_s = 0.0
-    spawned: list = []
-
-    async def scenario() -> None:
-        on_msg = improver.make_on_msg(
-            pm,
-            cfg,
-            None,
-            lock=asyncio.Lock(),
-            last_run=[0.0],
-            spawn=lambda coro: (spawned.append(coro), coro.close()),
-            publish=lambda s, d: None,
-        )
-        await on_msg(_Msg(subject, payload))
-
-    asyncio.run(scenario())
-    return len(spawned)
-
-
-def test_enforce_skips_a_reflection_trigger_from_a_non_learnable_session() -> None:
-    # §4.3e second layer: filtering the read-model keeps a synthetic reflection
-    # out of the self-model, but the trigger itself would still spend an LLM
-    # cycle reasoning about it. Drop it at the dispatch path too.
-    pm = _pm()
-    _seed_reflection(pm, "synth-1", "synthetic", "2026-07-17T13:00:00+00:00")
-    payload = {"session_id": "synth-1", "timestamp": "2030-01-01T00:00:00+00:00"}
-
-    set_provenance_mode(ProvenanceMode.OFF)
-    assert _spawned_for(pm, "augur.disciplina.complete", payload) == 1
-
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    assert _spawned_for(pm, "augur.disciplina.complete", payload) == 0
-
-
-def test_enforce_still_runs_a_user_driven_dialogue_trigger() -> None:
-    # The dialogue trigger is a direct user action carrying no session; it is not
-    # perception learning and must never be gated on a session's provenance.
-    pm = _pm()
-    set_provenance_mode(ProvenanceMode.ENFORCE)
-    assert (
-        _spawned_for(
-            pm, "augur.imperator.ii.trigger", {"reason": "dialogue", "ts": 1.0}
-        )
-        == 1
-    )