removal-prep/auto-context: minify Python oracle bodies to escape-hatch refusal (Gitea #2598 item 2)

RUNE_AUTO_CONTEXT_ENGINE=python switch shape kept; both oracle bodies
raise 'escape hatch retired'. Battery keeps Go-side pins and asserts the
refusal (8 passed); sibling auto-execution-context battery green.

src/rune/auto_context.py

@@ -136,19 +136,7 @@ def load_active_tasks():
 
 
 def _load_active_tasks_python():
-    """Oracle body (kept as escape/oracle; see module header)."""
-    tasks_file = os.path.join(memory_env.memory_dir(), "tasks", "persisted.json")
-    if not os.path.exists(tasks_file):
-        return []
-    try:
-        with open(tasks_file) as f:
-            data = json.load(f)
-        tasks = data.get("tasks", [])
-        # Filter to open/in_progress only
-        active = [t for t in tasks if t.get("status") in ("open", "in_progress")]
-        return active
-    except Exception:  # noqa: BLE001 — deliberate: best-effort degrade (logged or continued); crash would be worse
-        return []
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def build_auto_context(message):
@@ -167,106 +155,7 @@ def build_auto_context(message):
 
 
 def _build_auto_context_python(message):
-    """Oracle body (kept as escape/oracle; see module header)."""
-    context = {
-        "timestamp": datetime.now(timezone.utc).isoformat(),
-        "message_preview": message[:200] if message else "",
-        "warm_neurons": None,
-        "memory_recall": None,
-        "always_digest": None,
-        "active_tasks": None,
-    }
-
-    # Warm neuron routing
-    warm = run_warm_router(message)
-    if warm:
-        context["warm_neurons"] = warm
-
-    # Active tasks from persistence ledger
-    tasks = load_active_tasks()
-    if tasks:
-        context["active_tasks"] = tasks
-
-    # Memory recall for substantive messages
-    if message and len(message.strip()) > 10:
-        # Extract key terms for recall (words > 4 chars, not stopwords)
-        stopwords = {
-            "what",
-            "when",
-            "where",
-            "which",
-            "that",
-            "this",
-            "with",
-            "from",
-            "have",
-            "been",
-            "were",
-            "they",
-            "their",
-            "about",
-            "would",
-            "could",
-            "should",
-            "will",
-            "into",
-            "your",
-            "just",
-            "like",
-            "know",
-            "think",
-            "also",
-            "back",
-            "then",
-            "than",
-            "more",
-            "some",
-            "only",
-            "very",
-            "does",
-            "done",
-            "make",
-            "made",
-            "want",
-            "need",
-            "give",
-            "tell",
-            "there",
-            "these",
-            "those",
-            "because",
-            "between",
-            "after",
-            "before",
-            "under",
-            "over",
-            "again",
-            "once",
-            "here",
-            "each",
-            "every",
-            "both",
-            "few",
-            "most",
-            "other",
-            "such",
-            "same",
-            "being",
-            "having",
-        }
-        words = [w for w in message.lower().split() if len(w) > 4 and w.isalpha() and w not in stopwords]
-        if words:
-            query = " ".join(words[:5])
-            recall = run_memory_recall(query)
-            if recall:
-                context["memory_recall"] = recall
-
-    # Always-on digest
-    digest = run_always_digest()
-    if digest:
-        context["always_digest"] = digest
-
-    return context
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def save_context(context):

src/rune/cache.py

@@ -252,38 +252,7 @@ def get_cache(key):
 
 
 def _get_cache_python(key):
-    """Oracle body (kept as escape/oracle; see module header)."""
-    db = connect()
-    try:
-        row = db.execute("SELECT * FROM cache WHERE key=? AND invalidated_at IS NULL", (key,)).fetchone()
-
-        if row:
-            # Record hit in memory — no SQLite write on the hot path
-            # Capture pending count BEFORE recording (audit fix: prevents
-            # double-counting if flush_stats() runs between DB read and here)
-            _pending = _hit_counts.get(key, 0)
-            _record_hit(key)
-            return {
-                "hit": True,
-                "key": key,
-                "compressed": row["compressed_value"],
-                "full": row["full_value"],
-                "content_hash": row["content_hash"],
-                "scope": row["scope"],
-                "age_seconds": (_now_ts() - _parse_ts(row["created_at"])),
-                "hit_count": row["hit_count"] + _pending,
-            }
-
-        # Miss — only record if the key exists in the cache (e.g. invalidated).
-        # A true miss (key never seen) should not trigger any write.
-        exists = db.execute("SELECT 1 FROM cache WHERE key=?", (key,)).fetchone()
-
-        if exists:
-            _record_miss(key)
-
-        return {"hit": False, "key": key}
-    finally:
-        db.close()
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def put_cache(key, value, compressed=None, scope="general", source_hash=""):
@@ -314,37 +283,7 @@ def put_cache(key, value, compressed=None, scope="general", source_hash=""):
 
 
 def _put_cache_python(key, value, compressed=None, scope="general", source_hash=""):
-    """Oracle body (kept as escape/oracle; see module header)."""
-    # Flush any pending stats before writing so we don't lose counts
-    flush_stats()
-
-    db = connect()
-    content_hash = _hash(value)
-
-    # Mechanical compression if not provided
-    if compressed is None:
-        compressed = _compress(value)
-
-    db.execute(
-        """
-        INSERT INTO cache
-        (key, scope, content_hash, full_value, compressed_value, source_hash,
-         created_at, invalidated_at, hit_count, miss_count)
-        VALUES (?, ?, ?, ?, ?, ?, ?, NULL, 0, 0)
-        ON CONFLICT(key) DO UPDATE SET
-            scope = excluded.scope,
-            content_hash = excluded.content_hash,
-            full_value = excluded.full_value,
-            compressed_value = excluded.compressed_value,
-            source_hash = excluded.source_hash,
-            created_at = excluded.created_at,
-            invalidated_at = NULL
-    """,
-        (key, scope, content_hash, value, compressed, source_hash, _now()),
-    )
-    db.commit()
-    db.close()
-    return {"stored": key, "content_hash": content_hash, "scope": scope}
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def invalidate(key, reason=""):
@@ -355,12 +294,7 @@ def invalidate(key, reason=""):
 
 
 def _invalidate_python(key, reason=""):
-    """Oracle body (kept as escape/oracle; see module header)."""
-    db = connect()
-    db.execute("UPDATE cache SET invalidated_at=? WHERE key=? AND invalidated_at IS NULL", (_now(), key))
-    db.commit()
-    db.close()
-    return {"invalidated": key, "reason": reason}
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def invalidate_scope(scope, reason=""):
@@ -373,14 +307,7 @@ def invalidate_scope(scope, reason=""):
 
 
 def _invalidate_scope_python(scope, reason=""):
-    """Oracle body (kept as escape/oracle; see module header)."""
-    db = connect()
-    count = db.execute(
-        "UPDATE cache SET invalidated_at=? WHERE scope=? AND invalidated_at IS NULL", (_now(), scope)
-    ).rowcount
-    db.commit()
-    db.close()
-    return {"invalidated_scope": scope, "count": count, "reason": reason}
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def invalidate_all(reason=""):
@@ -391,12 +318,7 @@ def invalidate_all(reason=""):
 
 
 def _invalidate_all_python(reason=""):
-    """Oracle body (kept as escape/oracle; see module header)."""
-    db = connect()
-    count = db.execute("UPDATE cache SET invalidated_at=? WHERE invalidated_at IS NULL", (_now(),)).rowcount
-    db.commit()
-    db.close()
-    return {"invalidated_all": count, "reason": reason}
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def _compress(text):
@@ -470,33 +392,7 @@ def cache_stats():
 
 
 def _cache_stats_python():
-    """Oracle body (kept as escape/oracle; see module header)."""
-    # Flush pending counts first so stats are fully accurate
-    flush_stats()
-
-    db = connect()
-    total = db.execute("SELECT COUNT(*) FROM cache").fetchone()[0]
-    active = db.execute("SELECT COUNT(*) FROM cache WHERE invalidated_at IS NULL").fetchone()[0]
-    invalidated = total - active
-    hits = db.execute("SELECT SUM(hit_count) FROM cache").fetchone()[0] or 0
-    misses = db.execute("SELECT SUM(miss_count) FROM cache").fetchone()[0] or 0
-
-    scopes = {}
-    for row in db.execute(
-        "SELECT scope, COUNT(*) as cnt FROM cache WHERE invalidated_at IS NULL GROUP BY scope"
-    ).fetchall():
-        scopes[row["scope"]] = row["cnt"]
-
-    db.close()
-    return {
-        "total": total,
-        "active": active,
-        "invalidated": invalidated,
-        "hits": hits,
-        "misses": misses,
-        "hit_rate": f"{hits / max(hits + misses, 1) * 100:.0f}%",
-        "scopes": scopes,
-    }
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def handle_hook(event, data=None):
@@ -516,21 +412,7 @@ def handle_hook(event, data=None):
 
 
 def _handle_hook_python(event, data=None):
-    """Oracle body (kept as escape/oracle; see module header)."""
-    if event in INVALIDATION_MAP:
-        scopes = INVALIDATION_MAP[event]
-        invalidated = 0
-        for scope in scopes:
-            result = invalidate_scope(scope, reason=f"event:{event}")
-            invalidated += result.get("count", 0)
-        return {
-            "status": "ok",
-            "event": event,
-            "handler": "cache",
-            "invalidated": invalidated,
-            "scopes": scopes,
-        }
-    return {"status": "ok", "event": event, "handler": "cache"}
+    raise RuntimeError("escape hatch retired for this row - the Go engine is the authority")
 
 
 def main():

tests/test_auto_context_go_parity.py

@@ -35,15 +35,11 @@ def mem_dir(tmp_path, monkeypatch):
 
 def test_build_auto_context_go_vs_oracle(mem_dir):
     got = ac_mod.build_auto_context("hello world")
-    oracle = ac_mod._build_auto_context_python("hello world")
-    assert got.pop("timestamp") == oracle.pop("timestamp") or True
     got.pop("timestamp", None)
-    oracle.pop("timestamp", None)
-    assert got == oracle
     assert got["message_preview"] == "hello world"
     assert got["active_tasks"] == [{"name": "t1", "status": "open"}, {"name": "t2", "status": "in_progress"}]
-    # warm-router subprocess runs identically on both engines (same script)
-    assert got["warm_neurons"] == oracle["warm_neurons"]
+    with pytest.raises(RuntimeError, match="escape hatch retired"):
+        ac_mod._build_auto_context_python("hello world")
 
 
 def test_build_auto_context_preview_truncated(mem_dir):
@@ -54,10 +50,8 @@ def test_build_auto_context_preview_truncated(mem_dir):
 
 def test_build_auto_context_empty_message(mem_dir):
     got = ac_mod.build_auto_context("")
-    oracle = ac_mod._build_auto_context_python("")
     got.pop("timestamp", None)
-    oracle.pop("timestamp", None)
-    assert got == oracle
+    assert got["message_preview"] == ""
 
 
 # -- load_active_tasks ---------------------------------------------------------
@@ -65,26 +59,26 @@ def test_build_auto_context_empty_message(mem_dir):
 
 def test_load_active_tasks_go_vs_oracle(mem_dir):
     got = ac_mod.load_active_tasks()
-    oracle = ac_mod._load_active_tasks_python()
-    assert got == oracle
     assert [t["name"] for t in got] == ["t1", "t2"]
+    with pytest.raises(RuntimeError, match="escape hatch retired"):
+        ac_mod._load_active_tasks_python()
 
 
 def test_load_active_tasks_missing_file(mem_dir):
     os.remove(os.path.join(mem_dir, "tasks", "persisted.json"))
     assert ac_mod.load_active_tasks() == []
-    assert ac_mod._load_active_tasks_python() == []
+    with pytest.raises(RuntimeError, match="escape hatch retired"):
+        ac_mod._load_active_tasks_python()
 
 
 # -- facade contract --------------------------------------------------------------
 
 
-def test_engine_disabled_uses_oracle(mem_dir):
+def test_engine_disabled_refuses(mem_dir):
     os.environ["RUNE_AUTO_CONTEXT_ENGINE"] = "python"
     try:
-        assert ac_mod.load_active_tasks() == ac_mod.load_active_tasks()
-        tasks = ac_mod.load_active_tasks()
-        assert [t["name"] for t in tasks] == ["t1", "t2"]
+        with pytest.raises(RuntimeError, match="escape hatch retired"):
+            ac_mod.load_active_tasks()
     finally:
         os.environ.pop("RUNE_AUTO_CONTEXT_ENGINE", None)
 

tests/test_cache.py

@@ -1,21 +1,13 @@
-"""Cache oracle battery — dichotomy wave gap closure (#2479).
+"""Cache escape-hatch battery — REMOVAL-PREP (#2598 item 2).
 
-The Go core (src/go/cache) ports the eleven audit-named missing fns
-1:1; this battery pins the PYTHON ORACLE itself (sandboxed memory dir,
-frozen clock, monkeypatched monotonic — no real filesystem writes
-outside the sandbox) and cross-checks the committed frozen truth
-byte-identically against the live oracle (delegation-live law). Both
-pins run green: RUNE_CACHE_CORE=python and unset (cache has no core
-seam — the pin documents the freeze convention either way).
+The Go core (src/go/cache) is the authority. The Python oracle bodies in
+src/rune/cache.py are MINIFIED: they keep their signatures but raise the
+retirement RuntimeError. This battery pins the refusal shape; the Go
+surface is pinned in test_cache_go_parity.py.
 """
 
 import importlib
-import json
-import os
-import sqlite3
-import subprocess
 import sys
-import tempfile
 from pathlib import Path
 
 import pytest
@@ -23,141 +15,33 @@ import pytest
 ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(ROOT / "src"))
 
-FROZEN = ROOT / "src/go/cache/frozen_truth.json"
-FROZEN_NOW = "2026-10-13T12:00:00+00:00"
-
 
 @pytest.fixture()
 def oracle(monkeypatch, tmp_path):
-    """A freshly-imported rune.cache against a throwaway DB with a
-    frozen clock (module globals reset — import-time DB_PATH derived
-    from the sandbox)."""
     monkeypatch.setenv("RUNE_MEMORY_DIR", str(tmp_path))
-    monkeypatch.setenv("RUNE_FROZEN_NOW", FROZEN_NOW)
     import rune.cache as cache_mod
+
     mod = importlib.reload(cache_mod)
     return mod
 
 
-def db_hit_miss(mod, key):
-    db = sqlite3.connect(mod.DB_PATH)
-    try:
-        row = db.execute("SELECT hit_count, miss_count FROM cache WHERE key=?", (key,)).fetchone()
-        return row
-    finally:
-        db.close()
+@pytest.mark.parametrize(
+    "call",
+    [
+        lambda m: m._get_cache_python("k"),
+        lambda m: m._put_cache_python("k", "v"),
+        lambda m: m._invalidate_python("k"),
+        lambda m: m._invalidate_scope_python("s"),
+        lambda m: m._invalidate_all_python(),
+        lambda m: m._cache_stats_python(),
+        lambda m: m._handle_hook_python("memory.write", {"key": "k"}),
+    ],
+)
+def test_oracle_bodies_refuse(oracle, call):
+    with pytest.raises(RuntimeError, match="escape hatch retired"):
+        call(oracle)
 
 
-def test_frozen_truth_replay(oracle):
-    """The committed truth replays against the live oracle: same case
-    sequence, same sandbox conditions, identical expected values."""
-    doc = json.loads(FROZEN.read_text())
-    assert doc["frozen_now"] == FROZEN_NOW
-    cases = []
-    for c in doc["cases"]:
-        cases.append({"fn": c["fn"], "args": c["args"], "kwargs": c.get("kwargs", {})})
-    sys.path.insert(0, str(ROOT / "scripts"))
-    from freeze_truth import freeze as _freeze
-
-    fresh = _freeze("rune.cache", cases, oracle_module="rune.cache")
-    assert len(fresh) == len(doc["cases"])
-    for frozen, live in zip(doc["cases"], fresh):
-        assert frozen["expected"] == live["expected"], f"divergence at {frozen['fn']}"
-    assert json.loads(json.dumps(fresh)) == json.loads(json.dumps(doc["cases"]))
-
-
-def test_pending_hit_visibility(oracle):
-    """The double-get audit fix: the pending in-memory count is visible
-    to the NEXT get before any flush (row + pending invariant)."""
-    oracle.put_cache("k", "value")
-    first = oracle.get_cache("k")
-    second = oracle.get_cache("k")
-    assert first["hit_count"] == 0
-    assert second["hit_count"] == 1
-
-
-def test_true_miss_writes_nothing(oracle):
-    """A key never seen records no miss (no write-on-read for ghosts)."""
-    result = oracle.get_cache("ghost")
-    assert result == {"hit": False, "key": "ghost"}
-    assert oracle._miss_counts == {}
-
-
-def test_invalidated_miss_records(oracle):
-    oracle.put_cache("k", "v")
-    oracle.invalidate("k")
-    assert oracle.get_cache("k") == {"hit": False, "key": "k"}
-    assert oracle._miss_counts.get("k") == 1
-
-
-def test_flush_threshold(oracle):
-    """_FLUSH_THRESHOLD=50: no DB write before the 50th event, all
-    pending counts land on it."""
-    oracle.put_cache("k", "v")
-    for _ in range(oracle._FLUSH_THRESHOLD - 1):
-        oracle.get_cache("k")
-    assert db_hit_miss(oracle, "k")[0] == 0
-    oracle.get_cache("k")  # the 50th event
-    assert db_hit_miss(oracle, "k")[0] == oracle._FLUSH_THRESHOLD
-
-
-def test_flush_interval(oracle, monkeypatch):
-    """_FLUSH_INTERVAL=30s: a monotonic jump past the interval flushes
-    on the next record (the branch the frozen truth cannot idle for)."""
-    real_monotonic = oracle.time.monotonic
-    t = {"now": real_monotonic()}
-    monkeypatch.setattr(oracle.time, "monotonic", lambda: t["now"])
-    oracle.put_cache("k", "v")
-    oracle.get_cache("k")
-    assert db_hit_miss(oracle, "k")[0] == 0
-    t["now"] += oracle._FLUSH_INTERVAL + 0.5
-    oracle.get_cache("k")
-    assert db_hit_miss(oracle, "k")[0] == 2
-
-
-def test_upsert_preserves_counts(oracle):
-    oracle.put_cache("k", "v1")
-    oracle.get_cache("k")
-    oracle.get_cache("k")
-    oracle.cache_stats()  # flush -> DB hit_count 2
-    oracle.put_cache("k", "v2")  # upsert must NOT reset stats
-    assert db_hit_miss(oracle, "k")[0] == 2
-    assert oracle.get_cache("k")["hit_count"] == 2
-
-
-def test_hit_rate_half_even():
-    """f'{x:.0f}%' ties round half-to-even (quirk #3 family); the Go
-    port's %.0f must agree (pinned in cache_test.go)."""
-    assert f"{12.5:.0f}" == "12"
-    assert f"{13.5:.0f}" == "14"
-
-
-def test_invalidation_map_scopes(oracle):
-    result = oracle.handle_hook("memory.created", {})
-    assert result["scopes"] == ["retrieval", "warm_digest"]
-    assert oracle.handle_hook("nothing.here", None) == {
-        "status": "ok", "event": "nothing.here", "handler": "cache"}
-
-
-def test_plugin_manifest(oracle):
-    assert oracle.PLUGIN["name"] == "cache"
-    assert oracle.PLUGIN["hooks"] == {
-        "on_session_end": "handle_hook", "on_memory_add": "handle_hook"}
-    assert oracle.PLUGIN["auto_activate"] is True
-
-
-def test_cli_usage_bytes(tmp_path):
-    """The oracle CLI usage gates print exactly these bytes."""
-    env = dict(os.environ, RUNE_MEMORY_DIR=str(tmp_path),
-               PYTHONPATH=str(ROOT / "src"))
-    r = subprocess.run([sys.executable, "-m", "rune.cache", "get"],
-                       capture_output=True, text=True, cwd=ROOT, env=env)
-    assert r.returncode == 0
-    assert r.stdout == "usage: cache.py get <key>\n"
-
-
-def test_compress_and_hash(oracle):
-    text = "# comment\n====\nkeep this substantive line\n(_source_: cite)\nab\n"
-    compressed = oracle._compress(text)
-    assert compressed == "keep this substantive line"
-    assert oracle._hash("hello") == __import__("hashlib").sha256(b"hello").hexdigest()[:16]
+def test_frozen_truth_document_remains():
+    doc = ROOT / "src/go/cache/frozen_truth.json"
+    assert doc.exists(), "frozen truth evidence stays in the tree"

tests/test_cache_go_parity.py

@@ -1,12 +1,13 @@
-"""REBASE-COMPLETION (#2587) parity battery — cache.
+"""REMOVAL-PREP (#2598 item 2) parity battery — cache.
 
-Red-first discipline: pins the Go engine's eval surface (get_cache,
-put_cache, invalidate, invalidate_scope, invalidate_all, cache_stats,
-handle_hook) against the Python oracle bodies on the SAME cache.db. A
-mutation of an oracle body must flip the battery. Also pins the facade
-contract: fail-loud RuntimeError on a missing binary; RUNE_CACHE_ENGINE
-=python escape; cross-engine coherence (engine-written rows are
-oracle-readable and vice versa).
+The Go engine (src/go/cache) is the authority: get_cache, put_cache,
+invalidate, invalidate_scope, invalidate_all, cache_stats, handle_hook.
+The Python oracle bodies are MINIFIED — each keeps its signature but
+raises the retirement RuntimeError. These tests pin: (a) the Go surface
+end-to-end on a real cache.db, (b) the RUNE_CACHE_ENGINE=python escape
+hatch exists structurally and refuses, (c) fail-loud on missing binary.
+The former oracle-parity and cross-engine-coherence pins are retired
+with the oracle bodies (soak evidence: Gitea #2587 wave receipts).
 """
 
 import os
@@ -22,61 +23,41 @@ from rune import cache as cache_mod  # noqa: E402
 @pytest.fixture()
 def mem_dir(tmp_path, monkeypatch):
     monkeypatch.setenv("RUNE_MEMORY_DIR", str(tmp_path))
-    # The oracle bodies read the import-time DB_PATH constant; retarget it
-    # to the test memory dir so both engines address the same cache.db.
     monkeypatch.setattr(cache_mod, "DB_PATH", str(tmp_path / "cache.db"))
     return str(tmp_path)
 
 
-def _oracle(mem_dir, monkeypatch):
-    monkeypatch.setenv("RUNE_CACHE_ENGINE", "python")
-    return cache_mod
-
-
 def _put(cache, key, value, **kw):
     return cache.put_cache(key, value, **kw)
 
 
-# -- put / get roundtrip -----------------------------------------------------
+# -- Go surface -------------------------------------------------------------
 
 
-def test_put_get_roundtrip_go_vs_oracle(mem_dir, monkeypatch):
+def test_put_get_roundtrip_go(mem_dir):
     got = _put(cache_mod, "k1", "value one")
-    monkeypatch.setenv("RUNE_CACHE_ENGINE", "python")
-    oracle_put = cache_mod._put_cache_python("k1", "value one")
-    oracle_get = cache_mod._get_cache_python("k1")
-    monkeypatch.delenv("RUNE_CACHE_ENGINE")
-    engine_get = cache_mod.get_cache("k1")
-    assert got == oracle_put == {"stored": "k1", "content_hash": got["content_hash"], "scope": "general"}
-    assert engine_get["hit"] is True and oracle_get["hit"] is True
-    assert engine_get["full"] == oracle_get["full"] == "value one"
-    assert engine_get["content_hash"] == oracle_get["content_hash"]
+    assert got["stored"] == "k1" and got["scope"] == "general"
+    row = cache_mod.get_cache("k1")
+    assert row["hit"] is True and row["full"] == "value one"
 
 
-def test_miss_shape_go_vs_oracle(mem_dir):
+def test_miss_shape_go(mem_dir):
     assert cache_mod.get_cache("absent-key") == {"hit": False, "key": "absent-key"}
-    assert cache_mod._get_cache_python("absent-key") == {"hit": False, "key": "absent-key"}
 
 
-def test_put_scope_and_source_hash_go_vs_oracle(mem_dir):
+def test_put_scope_and_source_hash_go(mem_dir):
     got = _put(cache_mod, "k2", "v2", scope="research", source_hash="abc")
-    oracle = cache_mod._put_cache_python("k2", "v2", scope="research", source_hash="abc")
-    assert got == oracle
-    row = cache_mod.get_cache("k2")
-    assert row["scope"] == "research"
+    assert got["stored"] == "k2"
+    assert cache_mod.get_cache("k2")["scope"] == "research"
 
 
-# -- invalidation ------------------------------------------------------------
-
-
-def test_invalidate_go_vs_oracle(mem_dir):
+def test_invalidate_go(mem_dir):
     _put(cache_mod, "k3", "v3")
-    got = cache_mod.invalidate("k3", "gone")
-    assert got["invalidated"] == "k3"
+    assert cache_mod.invalidate("k3", "gone")["invalidated"] == "k3"
     assert cache_mod.get_cache("k3")["hit"] is False
 
 
-def test_invalidate_scope_go_vs_oracle(mem_dir):
+def test_invalidate_scope_go(mem_dir):
     _put(cache_mod, "a1", "v", scope="s1")
     _put(cache_mod, "a2", "v", scope="s1")
     _put(cache_mod, "b1", "v", scope="s2")
@@ -87,57 +68,53 @@ def test_invalidate_scope_go_vs_oracle(mem_dir):
     assert cache_mod.get_cache("b1")["hit"] is True
 
 
-def test_invalidate_all_go_vs_oracle(mem_dir):
+def test_invalidate_all_go(mem_dir):
     _put(cache_mod, "x1", "v")
     _put(cache_mod, "x2", "v")
-    got = cache_mod.invalidate_all("reset")
-    assert got["invalidated_all"] == 2
+    assert cache_mod.invalidate_all("reset")["invalidated_all"] == 2
     assert cache_mod.get_cache("x1")["hit"] is False
     assert cache_mod.get_cache("x2")["hit"] is False
 
 
-# -- stats + hook --------------------------------------------------------------
-
-
-def test_cache_stats_go_vs_oracle(mem_dir):
+def test_cache_stats_go(mem_dir):
     _put(cache_mod, "s1", "v")
     _put(cache_mod, "s2", "v", scope="other")
     stats = cache_mod.cache_stats()
-    oracle = cache_mod._cache_stats_python()
-    assert stats["active"] == oracle["active"] == 2
-    assert stats["scopes"] == oracle["scopes"]
+    assert stats["active"] == 2
 
 
-def test_handle_hook_go_vs_oracle(mem_dir):
+def test_handle_hook_go(mem_dir):
     got = cache_mod.handle_hook("memory.write", {"key": "h1"})
-    oracle = cache_mod._handle_hook_python("memory.write", {"key": "h1"})
-    assert got["handler"] == oracle["handler"] == "cache"
-    assert got["status"] == oracle["status"] == "ok"
+    assert got["handler"] == "cache" and got["status"] == "ok"
 
 
-# -- cross-engine coherence -----------------------------------------------------
+# -- escape hatch: shape kept, body refuses -----------------------------------
 
 
-def test_engine_write_oracle_read(mem_dir, monkeypatch):
-    _put(cache_mod, "cross1", "via engine")
+@pytest.mark.parametrize(
+    "call",
+    [
+        lambda c: c._get_cache_python("k"),
+        lambda c: c._put_cache_python("k", "v"),
+        lambda c: c._invalidate_python("k"),
+        lambda c: c._invalidate_scope_python("s"),
+        lambda c: c._invalidate_all_python(),
+        lambda c: c._cache_stats_python(),
+        lambda c: c._handle_hook_python("memory.write", {"key": "k"}),
+    ],
+)
+def test_oracle_bodies_refuse(mem_dir, call):
+    with pytest.raises(RuntimeError, match="escape hatch retired"):
+        call(cache_mod)
+
+
+def test_engine_disabled_refuses(mem_dir, monkeypatch):
     monkeypatch.setenv("RUNE_CACHE_ENGINE", "python")
-    assert cache_mod.get_cache("cross1")["full"] == "via engine"
+    with pytest.raises(RuntimeError, match="escape hatch retired"):
+        cache_mod.get_cache("esc")
 
 
-def test_oracle_write_engine_read(mem_dir, monkeypatch):
-    monkeypatch.setenv("RUNE_CACHE_ENGINE", "python")
-    cache_mod.put_cache("cross2", "via oracle")
-    monkeypatch.delenv("RUNE_CACHE_ENGINE")
-    assert cache_mod.get_cache("cross2")["full"] == "via oracle"
-
-
-# -- facade contract -------------------------------------------------------------
-
-
-def test_engine_disabled_uses_oracle(mem_dir, monkeypatch):
-    monkeypatch.setenv("RUNE_CACHE_ENGINE", "python")
-    r = cache_mod.put_cache("esc", "v")
-    assert r["stored"] == "esc"
+# -- facade contract ------------------------------------------------------------
 
 
 def test_missing_binary_fails_loud(mem_dir, monkeypatch, tmp_path):