diff --git a/ops/derive_leoclean_runtime_query_contract.py b/ops/derive_leoclean_runtime_query_contract.py
index df86c77..045a012 100644
--- a/ops/derive_leoclean_runtime_query_contract.py
+++ b/ops/derive_leoclean_runtime_query_contract.py
@@ -12,6 +12,7 @@ from __future__ import annotations
 
 import argparse
 import ast
+import hashlib
 import json
 import re
 from collections import defaultdict
@@ -47,6 +48,7 @@ DYNAMIC_SQL_VALUE = "__teleo_dynamic_sql_value__"
 UNREVIEWED_DYNAMIC_SQL = "__teleo_unreviewed_dynamic_sql__"
 REVIEWED_SQL_INTERPOLATION_CALLS = frozenset({"sql_array", "sql_integer", "sql_json_array", "sql_literal"})
 QUERY_RUNNERS = frozenset({"run_psql", "psql_json_lines"})
+REVIEWED_RUNTIME_SOURCE_SHA256 = "5f98218622803e0a1bbedc1e23e538d1f152ecb516e31a387a5eb39080aaf2cd"
 DYNAMIC_CALL_BUILTINS = frozenset({"eval", "exec", "globals", "locals"})
 READ_CAPABILITY = "read"
 STAGE_PROPOSAL_CAPABILITY = "stage-proposal"
@@ -83,6 +85,15 @@ class QueryContractError(ValueError):
     """The runtime SQL cannot be safely reduced to an exact read contract."""
 
 
+def _assert_exact_runtime_source(source: str) -> None:
+    observed = hashlib.sha256(source.encode("utf-8")).hexdigest()
+    if observed != REVIEWED_RUNTIME_SOURCE_SHA256:
+        raise QueryContractError(
+            "runtime query surface differs from the exact reviewed runtime source: "
+            f"expected={REVIEWED_RUNTIME_SOURCE_SHA256}, observed={observed}"
+        )
+
+
 def _is_reviewed_sql_interpolation(node: ast.expr) -> bool:
     return (
         isinstance(node, ast.Call)
@@ -216,7 +227,11 @@ def _is_reviewed_psql_json_lines_forwarder(
     )
 
 
-def extract_runtime_query_literals(source: str) -> tuple[tuple[int, str, str], ...]:
+def extract_runtime_query_literals(
+    source: str,
+    *,
+    enforce_source_pin: bool = True,
+) -> tuple[tuple[int, str, str], ...]:
     """Return every statically rendered SQL statement and its capability."""
 
     tree = ast.parse(source)
@@ -319,14 +334,20 @@ def extract_runtime_query_literals(source: str) -> tuple[tuple[int, str, str], .
 
     if not queries:
         raise QueryContractError("runtime source contains no SQL query literals")
+    if enforce_source_pin:
+        _assert_exact_runtime_source(source)
     return tuple(sorted(queries, key=lambda item: item[0]))
 
 
-def extract_canonical_query_literals(source: str) -> tuple[tuple[int, str], ...]:
+def extract_canonical_query_literals(
+    source: str,
+    *,
+    enforce_source_pin: bool = True,
+) -> tuple[tuple[int, str], ...]:
     """Return canonical SQL literals actually passed to the runtime query runners."""
 
     queries: list[tuple[int, str]] = []
-    for line, sql, _capability in extract_runtime_query_literals(source):
+    for line, sql, _capability in extract_runtime_query_literals(source, enforce_source_pin=enforce_source_pin):
         normalized = _normalize_sql_identifier_syntax(sql, line)
         if CANONICAL_RELATION.search(normalized):
             queries.append((line, normalized))
@@ -564,7 +585,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[
     """Mechanically derive canonical relation columns from runtime SQL."""
 
     observed: dict[tuple[str, str], set[str]] = defaultdict(set)
-    for line, sql in extract_canonical_query_literals(source):
+    for line, sql in extract_canonical_query_literals(source, enforce_source_pin=False):
         aliases: dict[str, tuple[str, str]] = {}
         explicit_starts: set[int] = set()
         for match in ALIASED_CANONICAL_RELATION.finditer(sql):
@@ -600,6 +621,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[
             joined = ", ".join(sorted(unqualified))
             raise QueryContractError(f"line {line}: unqualified SQL identifiers prevent exact derivation: {joined}")
 
+    _assert_exact_runtime_source(source)
     return {relation: frozenset(columns) for relation, columns in observed.items()}
 
 
diff --git a/tests/test_gcp_leoclean_runtime_permissions.py b/tests/test_gcp_leoclean_runtime_permissions.py
index 9a16eaa..c270f52 100644
--- a/tests/test_gcp_leoclean_runtime_permissions.py
+++ b/tests/test_gcp_leoclean_runtime_permissions.py
@@ -672,6 +672,39 @@ def test_runtime_query_contract_rejects_a_new_unallowlisted_column() -> None:
         query_contract.verify_runtime_read_columns(mutated, expected)
 
 
+def test_runtime_query_contract_pins_the_exact_reviewed_runtime_source() -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+
+    assert hashlib.sha256(runtime_source.encode("utf-8")).hexdigest() == query_contract.REVIEWED_RUNTIME_SOURCE_SHA256
+
+
+def test_runtime_query_contract_rejects_recursive_proposal_execution_cardinality_mutation() -> None:
+    runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
+    runtime_source = runtime_path.read_text(encoding="utf-8")
+    recursive_sql = '''    sql = """
+with recursive spam as (
+  select 1 as n, null::jsonb as proposal
+  union all
+  select n + 1, kb_stage.stage_leoclean_proposal(
+    'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb
+  )
+  from spam
+  where n < 10
+)
+select proposal from spam;
+"""
+'''
+    function_start = runtime_source.index("def propose_core_change")
+    sql_start = runtime_source.index('    sql = f"""\n', function_start)
+    sql_end = runtime_source.index('"""\n    rows = psql_json_lines(', sql_start) + len('"""\n')
+    mutated = runtime_source[:sql_start] + recursive_sql + runtime_source[sql_end:]
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="exact reviewed runtime source"):
+        query_contract.extract_runtime_query_literals(mutated)
+
+
 @pytest.mark.parametrize(
     "query",
     (