Fail closed on leoclean runtime SQL

hermes-agent/leoclean-bin/cloudsql_memory_tool.py

@@ -170,6 +170,57 @@ def runtime_cloudsdk_config() -> str:
 
 
 CLONE_READ_ONLY_PGOPTIONS = "-c default_transaction_read_only=on"
+RUNTIME_STANDARD_STRINGS_PGOPTIONS = "-c standard_conforming_strings=on"
+RUNTIME_READ_ONLY_PGOPTIONS = (
+    f"{RUNTIME_STANDARD_STRINGS_PGOPTIONS} {CLONE_READ_ONLY_PGOPTIONS}"
+)
+RUNTIME_SQL_READ_CAPABILITY = "read"
+RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY = "stage-proposal"
+RUNTIME_SQL_CAPABILITIES = frozenset(
+    {RUNTIME_SQL_READ_CAPABILITY, RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY}
+)
+MAX_RUNTIME_SQL_BYTES = 262_144
+RUNTIME_SQL_FORBIDDEN_KEYWORD = re.compile(
+    r"\b(?:alter|analyze|begin|call|checkpoint|cluster|comment|commit|copy|create|"
+    r"deallocate|delete|discard|do|drop|execute|grant|insert|into|listen|load|lock|merge|"
+    r"notify|prepare|refresh|reindex|release|reset|revoke|rollback|savepoint|security|"
+    r"set|start|truncate|update|vacuum)\b",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_LARGE_OBJECT_MUTATOR = re.compile(
+    r"(?<![a-z0-9_])(?:(?:pg_catalog)\s*\.\s*)?"
+    r"(?:lo_creat|lo_create|lo_export|lo_from_bytea|lo_import|lo_open|lo_put|"
+    r"lo_truncate64|lo_truncate|lo_unlink|lowrite)\s*\(",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_FORBIDDEN_FUNCTION = re.compile(
+    r"(?<![a-z0-9_])(?:(?:pg_catalog)\s*\.\s*)?"
+    r"(?:set_config|nextval|setval|pg_notify|"
+    r"pg_(?:try_)?advisory_(?:xact_)?lock(?:_shared)?|"
+    r"pg_advisory_unlock(?:_all|_shared)?|pg_cancel_backend|"
+    r"pg_terminate_backend|pg_logical_emit_message)\s*\(",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_APPLICATION_FUNCTION = re.compile(
+    r"(?<![a-z0-9_])(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)\s*\(",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_FUNCTION_CALL = re.compile(
+    r"(?<![a-z0-9_.])(?:(?P<schema>[a-z_][a-z0-9_]*)\s*\.\s*)?"
+    r"(?P<name>[a-z_][a-z0-9_]*)\s*\(",
+    re.IGNORECASE,
+)
+RUNTIME_SQL_NONFUNCTION_PAREN_KEYWORDS = frozenset(
+    {"as", "case", "exists", "filter", "in", "over", "select", "values", "when", "where", "with"}
+)
+RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION = ("kb_stage", "stage_leoclean_proposal")
+RUNTIME_SQL_STAGE_PROPOSAL_ALLOWED_FUNCTIONS = frozenset(
+    {
+        (None, "jsonb_build_object"),
+        (None, "nullif"),
+        RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION,
+    }
+)
 RECEIPTED_READ_COMMANDS = frozenset(
     {
         "search",
@@ -596,7 +647,180 @@ def password(args: argparse.Namespace) -> str:
     raise SystemExit("Clone credential mode requires an inherited test credential.")
 
 
-def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str:
+def _runtime_sql_code(sql: str) -> str:
+    """Return SQL code with literals/comments removed, rejecting inexact syntax."""
+
+    if not isinstance(sql, str) or not sql or "\x00" in sql:
+        raise SystemExit("Runtime SQL was rejected: expected a non-empty text statement.")
+    if "\r" in sql:
+        raise SystemExit("Runtime SQL was rejected: carriage returns are not allowed.")
+    if len(sql.encode("utf-8")) > MAX_RUNTIME_SQL_BYTES:
+        raise SystemExit("Runtime SQL was rejected: statement exceeds the reviewed size bound.")
+
+    code: list[str] = []
+    index = 0
+    block_depth = 0
+    state = "code"
+    while index < len(sql):
+        char = sql[index]
+        next_char = sql[index + 1] if index + 1 < len(sql) else ""
+
+        if state == "line-comment":
+            if char == "\n":
+                code.append("\n")
+                state = "code"
+            else:
+                code.append(" ")
+            index += 1
+            continue
+
+        if state == "block-comment":
+            if char == "/" and next_char == "*":
+                block_depth += 1
+                code.extend((" ", " "))
+                index += 2
+            elif char == "*" and next_char == "/":
+                block_depth -= 1
+                code.extend((" ", " "))
+                index += 2
+                if block_depth == 0:
+                    state = "code"
+            else:
+                code.append("\n" if char == "\n" else " ")
+                index += 1
+            continue
+
+        if state in {"single-quote", "escape-string"}:
+            if state == "escape-string" and char == "\\":
+                code.append(" ")
+                if next_char:
+                    code.append("\n" if next_char == "\n" else " ")
+                    index += 2
+                else:
+                    index += 1
+            elif char == "'" and next_char == "'":
+                code.extend((" ", " "))
+                index += 2
+            elif char == "'":
+                code.append(" ")
+                state = "code"
+                index += 1
+            else:
+                code.append("\n" if char == "\n" else " ")
+                index += 1
+            continue
+
+        if state == "double-quote":
+            if char == '"' and next_char == '"':
+                code.append("_")
+                index += 2
+            elif char == '"':
+                code.append(" ")
+                state = "code"
+                index += 1
+            else:
+                code.append(char if (char.isalnum() or char == "_") else "_")
+                index += 1
+            continue
+
+        if char == "-" and next_char == "-":
+            code.extend((" ", " "))
+            state = "line-comment"
+            index += 2
+        elif char == "/" and next_char == "*":
+            code.extend((" ", " "))
+            block_depth = 1
+            state = "block-comment"
+            index += 2
+        elif char == "'":
+            escape_string = index > 0 and sql[index - 1] in "eE" and (
+                index == 1 or not (sql[index - 2].isalnum() or sql[index - 2] == "_")
+            )
+            code.append(" ")
+            state = "escape-string" if escape_string else "single-quote"
+            index += 1
+        elif char == '"':
+            if index >= 2 and sql[index - 2 : index].lower() == "u&":
+                raise SystemExit("Runtime SQL was rejected: Unicode-escape identifiers are not allowed.")
+            code.append(" ")
+            state = "double-quote"
+            index += 1
+        elif char == "\\":
+            raise SystemExit("Runtime SQL was rejected: psql meta-commands are not allowed.")
+        elif char == "$" and re.match(r"\$(?:[a-z_][a-z0-9_]*)?\$", sql[index:], re.IGNORECASE):
+            raise SystemExit("Runtime SQL was rejected: dollar-quoted bodies are not allowed.")
+        else:
+            code.append(char)
+            index += 1
+
+    if state == "block-comment":
+        raise SystemExit("Runtime SQL was rejected: unterminated block comment.")
+    if state in {"single-quote", "escape-string", "double-quote"}:
+        raise SystemExit("Runtime SQL was rejected: unterminated quoted value.")
+    return "".join(code)
+
+
+def validate_runtime_sql(sql: str, *, capability: str = RUNTIME_SQL_READ_CAPABILITY) -> None:
+    """Fail closed unless rendered SQL matches one reviewed runtime capability."""
+
+    if capability not in RUNTIME_SQL_CAPABILITIES:
+        raise SystemExit("Runtime SQL was rejected: unknown execution capability.")
+    code = _runtime_sql_code(sql)
+    if code.count(";") > 1:
+        raise SystemExit("Runtime SQL was rejected: exactly one statement is required.")
+    if ";" in code:
+        statement, remainder = code.split(";", 1)
+        if remainder.strip():
+            raise SystemExit("Runtime SQL was rejected: multiple statements are not allowed.")
+    else:
+        statement = code
+    first_token = re.match(r"\s*([a-z_][a-z0-9_]*)", statement, re.IGNORECASE)
+    if first_token is None or first_token.group(1).lower() not in {"select", "with"}:
+        raise SystemExit("Runtime SQL was rejected: only SELECT or WITH statements are allowed.")
+    if RUNTIME_SQL_FORBIDDEN_KEYWORD.search(statement):
+        raise SystemExit("Runtime SQL was rejected: mutating or session-control SQL is not allowed.")
+    if RUNTIME_SQL_LARGE_OBJECT_MUTATOR.search(statement):
+        raise SystemExit("Runtime SQL was rejected: large-object mutators are not allowed.")
+    if RUNTIME_SQL_FORBIDDEN_FUNCTION.search(statement):
+        raise SystemExit("Runtime SQL was rejected: side-effecting system functions are not allowed.")
+
+    application_functions = {
+        (match.group(1).lower(), match.group(2).lower())
+        for match in RUNTIME_SQL_APPLICATION_FUNCTION.finditer(statement)
+    }
+    if capability == RUNTIME_SQL_READ_CAPABILITY:
+        if application_functions:
+            raise SystemExit("Runtime SQL was rejected: application functions are not allowed for reads.")
+    else:
+        function_calls = [
+            (
+                match.group("schema").lower() if match.group("schema") else None,
+                match.group("name").lower(),
+            )
+            for match in RUNTIME_SQL_FUNCTION_CALL.finditer(statement)
+            if not (
+                match.group("schema") is None
+                and match.group("name").lower() in RUNTIME_SQL_NONFUNCTION_PAREN_KEYWORDS
+            )
+        ]
+        if function_calls.count(RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION) != 1 or any(
+            function not in RUNTIME_SQL_STAGE_PROPOSAL_ALLOWED_FUNCTIONS for function in function_calls
+        ):
+            raise SystemExit(
+                "Runtime SQL was rejected: proposal execution exceeds the reviewed staging function surface."
+            )
+
+
+def run_psql(
+    args: argparse.Namespace,
+    sql: str,
+    db: str | None = None,
+    *,
+    capability: str = RUNTIME_SQL_READ_CAPABILITY,
+) -> str:
+    validate_runtime_sql(sql, capability=capability)
+    if capability == RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY and args.credential_mode != "runtime":
+        raise SystemExit("Proposal execution requires the scoped runtime credential mode.")
     credential = password(args)
     if args.credential_mode == "runtime":
         env = dict(RUNTIME_PSQL_ENV_BASE)
@@ -612,8 +836,11 @@ def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str:
             "PGPASSWORD": credential,
         }
     )
-    if args.credential_mode == "clone-readonly":
+    env["PGOPTIONS"] = (
-        env["PGOPTIONS"] = CLONE_READ_ONLY_PGOPTIONS
+        RUNTIME_READ_ONLY_PGOPTIONS
+        if capability == RUNTIME_SQL_READ_CAPABILITY
+        else RUNTIME_STANDARD_STRINGS_PGOPTIONS
+    )
     result = subprocess.run(
         [RUNTIME_PSQL_BIN, "-X", "-At", "-q", "-v", "ON_ERROR_STOP=1"],
         text=True,
@@ -627,9 +854,15 @@ def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str:
     return result.stdout
 
 
-def psql_json_lines(args: argparse.Namespace, sql: str, db: str | None = None) -> list[dict[str, Any]]:
+def psql_json_lines(
+    args: argparse.Namespace,
+    sql: str,
+    db: str | None = None,
+    *,
+    capability: str = RUNTIME_SQL_READ_CAPABILITY,
+) -> list[dict[str, Any]]:
     rows: list[dict[str, Any]] = []
-    for line in run_psql(args, sql, db=db).splitlines():
+    for line in run_psql(args, sql, db=db, capability=capability).splitlines():
         line = line.strip()
         if line:
             rows.append(json.loads(line))
@@ -1440,7 +1673,12 @@ select jsonb_build_object(
 )::text
 from proposer;
 """
-    rows = psql_json_lines(args, sql, db=args.canonical_db)
+    rows = psql_json_lines(
+        args,
+        sql,
+        db=args.canonical_db,
+        capability=RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+    )
     if not rows:
         raise SystemExit("No core-change proposal inserted.")
     return rows[0]
@@ -1513,7 +1751,12 @@ select jsonb_build_object(
 )::text
 from staged;
 """
-    rows = psql_json_lines(args, sql, db=args.canonical_db)
+    rows = psql_json_lines(
+        args,
+        sql,
+        db=args.canonical_db,
+        capability=RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+    )
     if not rows:
         raise SystemExit("No edge proposal inserted. Check that both claim ids exist and edge_type is valid.")
     return rows[0]

ops/derive_leoclean_runtime_query_contract.py

@@ -18,11 +18,12 @@ from collections import defaultdict
 from pathlib import Path
 
 CANONICAL_RELATION = re.compile(
-    r"\b(?:from|join)\s+(public|kb_stage)\.([a-z_][a-z0-9_]*)",
+    r"\b(?:from|join)\s+(?:only\s+)?(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)",
     re.IGNORECASE,
 )
 ALIASED_CANONICAL_RELATION = re.compile(
-    r"\b(?:from|join)\s+(public|kb_stage)\.([a-z_][a-z0-9_]*)\s+as\s+([a-z_][a-z0-9_]*)",
+    r"\b(?:from|join)\s+(?:only\s+)?(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)"
+    r"\s+as\s+([a-z_][a-z0-9_]*)",
     re.IGNORECASE,
 )
 MUTATING_SQL_KEYWORD = re.compile(
@@ -41,17 +42,26 @@ LARGE_OBJECT_MUTATOR_REFERENCE = re.compile(
 )
 PSQL_META_COMMAND = re.compile(r"\\")
 CANONICAL_NAMESPACE_REFERENCE = re.compile(r"\b(?:public|kb_stage)\s*\.", re.IGNORECASE)
+QUOTED_SQL_IDENTIFIER = re.compile(r'"(?:""|[^"])*"')
 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"})
+DYNAMIC_CALL_BUILTINS = frozenset({"eval", "exec", "globals", "locals"})
+READ_CAPABILITY = "read"
+STAGE_PROPOSAL_CAPABILITY = "stage-proposal"
 SQL_IDENTIFIER = re.compile(r"\b[a-z_][a-z0-9_]*\b", re.IGNORECASE)
+SQL_STRUCTURE_TOKEN = re.compile(r"[a-z_][a-z0-9_]*|[(),.;]", re.IGNORECASE)
+FROM_CLAUSE_END_KEYWORDS = frozenset(
+    {"except", "fetch", "for", "group", "having", "intersect", "limit", "offset", "order", "returning", "union", "where", "window"}
+)
 SQL_KEYWORDS = frozenset(
     """
     all and any array as asc between bigint bool boolean by case cast cross
     current_database current_user desc distinct else end exists false filter
     first following for from full group having ilike in inner interval is join
     json jsonb last lateral left like limit not null nulls offset on or order
-    outer over partition preceding range recursive right row rows select some
+    only outer over partition preceding range recursive right row rows select some
     symmetric table text then ties time timestamp timestamptz to true union
     unknown using uuid values varchar when where window with zone
     """.split()  # noqa: SIM905 - the compact vocabulary is easier to audit as SQL.
@@ -66,6 +76,7 @@ NONCANONICAL_QUERY_IDENTIFIERS = frozenset(
         "table_type",
     }
 )
+UNQUALIFIED_NONCANONICAL_RELATIONS = frozenset({"pg_extension", "pg_tables"})
 
 
 class QueryContractError(ValueError):
@@ -98,38 +109,198 @@ def _render_sql_expression(node: ast.expr) -> str | None:
     return None
 
 
-def extract_canonical_query_literals(source: str) -> tuple[tuple[int, str], ...]:
+def _statement_path(
-    """Return canonical SQL literals actually passed to the runtime query runners."""
+    node: ast.AST,
+    function: ast.FunctionDef | ast.AsyncFunctionDef,
+    parents: dict[ast.AST, ast.AST],
+) -> tuple[tuple[ast.AST, str, int, ast.stmt], ...]:
+    """Return the ordered statement blocks containing ``node``."""
+
+    path: list[tuple[ast.AST, str, int, ast.stmt]] = []
+    current = node
+    while current is not function:
+        parent = parents.get(current)
+        if parent is None:
+            break
+        if isinstance(current, ast.stmt):
+            for field, value in ast.iter_fields(parent):
+                if isinstance(value, list) and current in value:
+                    path.append((parent, field, value.index(current), current))
+                    break
+        current = parent
+    return tuple(reversed(path))
+
+
+def _ancestor_chain(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> tuple[ast.AST, ...]:
+    ancestors: list[ast.AST] = []
+    current = node
+    while current in parents:
+        current = parents[current]
+        ancestors.append(current)
+    return tuple(ancestors)
+
+
+def _assignment_dominates_call(
+    assignment: ast.stmt,
+    call: ast.Call,
+    function: ast.FunctionDef | ast.AsyncFunctionDef,
+    parents: dict[ast.AST, ast.AST],
+) -> bool:
+    """Return whether one unconditional earlier assignment reaches ``call``."""
+
+    assignment_path = _statement_path(assignment, function, parents)
+    call_path = _statement_path(call, function, parents)
+    for index, assignment_entry in enumerate(assignment_path):
+        if index >= len(call_path):
+            return False
+        assignment_owner, assignment_field, assignment_index, assignment_statement = assignment_entry
+        call_owner, call_field, call_index, call_statement = call_path[index]
+        if assignment_owner is not call_owner or assignment_field != call_field:
+            return False
+        if assignment_index < call_index:
+            # An assignment hidden inside an earlier conditional does not
+            # dominate code after that conditional. The assignment itself must
+            # be the direct statement that precedes the call's branch.
+            return index == len(assignment_path) - 1
+        if assignment_index > call_index or assignment_statement is not call_statement:
+            return False
+    return False
+
+
+def _reject_query_runner_indirection(tree: ast.AST, parents: dict[ast.AST, ast.AST]) -> None:
+    """Require every runtime query runner reference to be a direct call."""
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Call) and isinstance(node.func, (ast.Call, ast.Lambda, ast.Subscript)):
+            raise QueryContractError(f"line {node.lineno}: indirect call expressions are not auditable")
+        if isinstance(node, ast.Name) and node.id in DYNAMIC_CALL_BUILTINS and isinstance(node.ctx, ast.Load):
+            raise QueryContractError(f"line {node.lineno}: dynamic call indirection is not auditable")
+        if isinstance(node, ast.Attribute) and node.attr in DYNAMIC_CALL_BUILTINS:
+            raise QueryContractError(f"line {node.lineno}: dynamic call indirection is not auditable")
+        if isinstance(node, ast.Name) and node.id in QUERY_RUNNERS and isinstance(node.ctx, ast.Load):
+            parent = parents.get(node)
+            if not (isinstance(parent, ast.Call) and parent.func is node):
+                raise QueryContractError(
+                    f"line {node.lineno}: query runner {node.id!r} must be called directly"
+                )
+        elif isinstance(node, ast.Attribute) and node.attr in QUERY_RUNNERS:
+            raise QueryContractError(
+                f"line {node.lineno}: query runner {node.attr!r} must be called directly"
+            )
+        elif isinstance(node, ast.Constant) and node.value in QUERY_RUNNERS:
+            raise QueryContractError(
+                f"line {node.lineno}: query runner {node.value!r} cannot be resolved through indirection"
+            )
+
+
+def _is_reviewed_psql_json_lines_forwarder(
+    call: ast.Call,
+    function: ast.FunctionDef | ast.AsyncFunctionDef,
+) -> bool:
+    """Recognize the one query-transparent ``psql_json_lines`` forwarding call."""
+
+    if function.name != "psql_json_lines":
+        return False
+    if not (isinstance(call.func, ast.Name) and call.func.id == "run_psql"):
+        return False
+    if len(call.args) != 2 or not all(isinstance(argument, ast.Name) for argument in call.args):
+        return False
+    if [argument.id for argument in call.args] != ["args", "sql"]:
+        return False
+    keywords = {keyword.arg: keyword.value for keyword in call.keywords if keyword.arg is not None}
+    if len(keywords) != len(call.keywords) or set(keywords) != {"db", "capability"}:
+        return False
+    return all(
+        isinstance(keywords[name], ast.Name) and keywords[name].id == name
+        for name in ("db", "capability")
+    )
+
+
+def extract_runtime_query_literals(source: str) -> tuple[tuple[int, str, str], ...]:
+    """Return every statically rendered SQL statement and its capability."""
 
     tree = ast.parse(source)
-    queries: set[tuple[int, str]] = set()
+    parents = {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)}
+    _reject_query_runner_indirection(tree, parents)
+    queries: set[tuple[int, str, str]] = set()
     for function in (node for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))):
-        assignments: dict[str, tuple[int, str | None]] = {}
+        function_calls = [node for node in ast.walk(function) if isinstance(node, ast.Call)]
+        reviewed_forwarders = [
+            call for call in function_calls if _is_reviewed_psql_json_lines_forwarder(call, function)
+        ]
+        if function.name == "psql_json_lines" and len(reviewed_forwarders) != 1:
+            raise QueryContractError(
+                f"line {function.lineno}: psql_json_lines must contain exactly one reviewed run_psql forwarder"
+            )
+        assignments: dict[str, list[tuple[ast.stmt, str | None]]] = defaultdict(list)
         for node in ast.walk(function):
             if isinstance(node, ast.Assign):
                 rendered = _render_sql_expression(node.value)
                 for target in node.targets:
                     if isinstance(target, ast.Name):
-                        assignments[target.id] = (node.lineno, rendered)
+                        assignments[target.id].append((node, rendered))
             elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
-                assignments[node.target.id] = (node.lineno, _render_sql_expression(node.value))
+                assignments[node.target.id].append((node, _render_sql_expression(node.value)))
+            elif isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name):
+                assignments[node.target.id].append((node, None))
+            elif isinstance(node, ast.NamedExpr) and isinstance(node.target, ast.Name):
+                # Named expressions are deliberately not an accepted SQL
+                # construction surface; record them as an inexact write.
+                statement = next(
+                    (ancestor for ancestor in _ancestor_chain(node, parents) if isinstance(ancestor, ast.stmt)),
+                    function,
+                )
+                assignments[node.target.id].append((statement, None))
 
-        for call in (node for node in ast.walk(function) if isinstance(node, ast.Call)):
+        for call in sorted(
+            function_calls,
+            key=lambda node: (node.lineno, node.col_offset),
+        ):
             runner = call.func.id if isinstance(call.func, ast.Name) else None
-            if runner not in {"run_psql", "psql_json_lines"}:
+            if runner not in QUERY_RUNNERS:
                 continue
-            if function.name == "psql_json_lines" and runner == "run_psql":
+            if _is_reviewed_psql_json_lines_forwarder(call, function):
                 continue
             if len(call.args) < 2:
                 raise QueryContractError(f"line {call.lineno}: {runner} requires a statically inspectable SQL argument")
+            capability_keywords = [keyword for keyword in call.keywords if keyword.arg == "capability"]
+            if len(capability_keywords) > 1:
+                raise QueryContractError(f"line {call.lineno}: {runner} has ambiguous SQL capabilities")
+            capability = READ_CAPABILITY
+            if capability_keywords:
+                capability_expression = capability_keywords[0].value
+                if (
+                    isinstance(capability_expression, ast.Name)
+                    and capability_expression.id == "RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY"
+                ) or (
+                    isinstance(capability_expression, ast.Constant)
+                    and capability_expression.value == STAGE_PROPOSAL_CAPABILITY
+                ):
+                    capability = STAGE_PROPOSAL_CAPABILITY
+                else:
+                    raise QueryContractError(
+                        f"line {call.lineno}: {runner} SQL capability must remain statically reviewed"
+                    )
             expression = call.args[1]
             if isinstance(expression, ast.Name):
-                line, sql = assignments.get(expression.id, (call.lineno, None))
+                writes = assignments.get(expression.id, [])
+                if len(writes) != 1:
+                    raise QueryContractError(
+                        f"line {call.lineno}: {runner} SQL variable {expression.id!r} must have exactly one assignment"
+                    )
+                assignment, sql = writes[0]
+                if not _assignment_dominates_call(assignment, call, function, parents):
+                    raise QueryContractError(
+                        f"line {call.lineno}: {runner} SQL assignment must unambiguously precede the call"
+                    )
+                line = assignment.lineno
             else:
                 line, sql = call.lineno, _render_sql_expression(expression)
             if sql is None:
                 raise QueryContractError(f"line {call.lineno}: {runner} SQL must remain a local string literal")
-            scrubbed = _strip_sql_strings_and_comments(sql)
+            scrubbed = _normalize_sql_identifier_syntax(sql, line)
+            _reject_unqualified_relation_references(scrubbed, line)
+            _reject_comma_separated_base_relations(scrubbed, line)
             if UNREVIEWED_DYNAMIC_SQL in sql:
                 raise QueryContractError(
                     f"line {line}: dynamic SQL interpolation is not part of the reviewed runtime query surface"
@@ -144,20 +315,146 @@ def extract_canonical_query_literals(source: str) -> tuple[tuple[int, str], ...]
                 raise QueryContractError(
                     f"line {line}: runtime query invokes a provider-owned large-object mutator"
                 )
-            if CANONICAL_RELATION.search(sql):
+            queries.add((line, sql, capability))
-                queries.add((line, sql))
 
     if not queries:
-        raise QueryContractError("runtime source contains no canonical SQL query literals")
+        raise QueryContractError("runtime source contains no SQL query literals")
     return tuple(sorted(queries, key=lambda item: item[0]))
 
 
+def extract_canonical_query_literals(source: str) -> 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):
+        normalized = _normalize_sql_identifier_syntax(sql, line)
+        if CANONICAL_RELATION.search(normalized):
+            queries.append((line, normalized))
+    if not queries:
+        raise QueryContractError("runtime source contains no canonical SQL query literals")
+    return tuple(queries)
+
+
 def _strip_sql_strings_and_comments(sql: str) -> str:
     without_strings = re.sub(r"(?:[eE])?'(?:''|[^'])*'", " ", sql)
     without_block_comments = re.sub(r"/\*.*?\*/", " ", without_strings, flags=re.DOTALL)
     return re.sub(r"--[^\n]*", " ", without_block_comments)
 
 
+def _normalize_sql_identifier_syntax(sql: str, line: int) -> str:
+    """Return structural SQL with exact, lowercase quoted identifiers normalized."""
+
+    scrubbed = _strip_sql_strings_and_comments(sql)
+
+    def normalize_quoted_identifier(match: re.Match[str]) -> str:
+        identifier = match.group(0)[1:-1].replace('""', '"')
+        if (
+            not SQL_IDENTIFIER.fullmatch(identifier)
+            or identifier != identifier.lower()
+            or identifier in SQL_KEYWORDS
+        ):
+            raise QueryContractError(
+                f"line {line}: quoted SQL identifier {identifier!r} cannot be reduced exactly"
+            )
+        return identifier
+
+    normalized = QUOTED_SQL_IDENTIFIER.sub(normalize_quoted_identifier, scrubbed)
+    if '"' in normalized:
+        raise QueryContractError(f"line {line}: unterminated quoted SQL identifier")
+    return normalized
+
+
+def _reject_unqualified_relation_references(sql: str, line: int) -> None:
+    """Reject search-path-dependent base relations while allowing local CTEs/functions."""
+
+    cte_names = {
+        match.group(1).lower()
+        for match in re.finditer(
+            r"(?:\bwith\s+(?:recursive\s+)?|,)\s*([a-z_][a-z0-9_]*)"
+            r"\s*(?:\([^)]*\))?\s+as\s+(?:(?:not\s+)?materialized\s+)?\(",
+            sql,
+            re.IGNORECASE,
+        )
+    }
+    for match in re.finditer(
+        r"\b(?:from|join)\s+(?:only\s+)?([a-z_][a-z0-9_]*)",
+        sql,
+        re.IGNORECASE,
+    ):
+        relation = match.group(1).lower()
+        remainder = sql[match.end() :].lstrip()
+        if remainder.startswith((".", "(")):
+            continue
+        if relation in cte_names or relation in UNQUALIFIED_NONCANONICAL_RELATIONS:
+            continue
+        raise QueryContractError(
+            f"line {line}: unqualified relation {relation!r} is search-path dependent"
+        )
+
+
+def _reject_comma_separated_base_relations(sql: str, line: int) -> None:
+    """Require explicit JOIN syntax for base relations after a FROM-list comma."""
+
+    cte_names = {
+        match.group(1).lower()
+        for match in re.finditer(
+            r"(?:\bwith\s+(?:recursive\s+)?|,)\s*([a-z_][a-z0-9_]*)"
+            r"\s*(?:\([^)]*\))?\s+as\s+(?:(?:not\s+)?materialized\s+)?\(",
+            sql,
+            re.IGNORECASE,
+        )
+    }
+    tokens = [(match.group(0).lower(), match.start()) for match in SQL_STRUCTURE_TOKEN.finditer(sql)]
+    depth = 0
+    from_depths: set[int] = set()
+    for index, (token, _position) in enumerate(tokens):
+        if token == "(":
+            depth += 1
+            continue
+        if token == ")":
+            from_depths.discard(depth)
+            depth = max(0, depth - 1)
+            continue
+        if token == "from":
+            from_depths.add(depth)
+            continue
+        if token == ";" or token in FROM_CLAUSE_END_KEYWORDS:
+            from_depths.discard(depth)
+            continue
+        if token != "," or depth not in from_depths:
+            continue
+
+        item_index = index + 1
+        while item_index < len(tokens) and tokens[item_index][0] in {"lateral", "only"}:
+            item_index += 1
+        if item_index >= len(tokens):
+            raise QueryContractError(f"line {line}: incomplete comma-separated FROM item")
+        relation = tokens[item_index][0]
+        if relation == "(":
+            continue
+        if not SQL_IDENTIFIER.fullmatch(relation):
+            raise QueryContractError(f"line {line}: comma-separated FROM item cannot be reduced exactly")
+
+        next_token = tokens[item_index + 1][0] if item_index + 1 < len(tokens) else ""
+        if next_token == "(":
+            continue
+        if next_token == ".":
+            function_token = tokens[item_index + 3][0] if item_index + 3 < len(tokens) else ""
+            if function_token == "(":
+                continue
+            qualified = relation
+            if item_index + 2 < len(tokens):
+                qualified = f"{relation}.{tokens[item_index + 2][0]}"
+            raise QueryContractError(
+                f"line {line}: comma-separated base relation {qualified!r} must use explicit JOIN"
+            )
+        if relation in cte_names:
+            continue
+        raise QueryContractError(
+            f"line {line}: comma-separated base relation {relation!r} must use explicit JOIN"
+        )
+
+
 def _unqualified_identifier_candidates(sql: str) -> set[str]:
     """Find identifiers that are neither qualified columns nor query structure."""
 
@@ -292,7 +589,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[
             observed[relation].update(
                 column.lower()
                 for column in re.findall(
-                    rf"(?<![a-z0-9_]){re.escape(alias)}\.([a-z_][a-z0-9_]*)",
+                    rf"(?<![a-z0-9_]){re.escape(alias)}\s*\.\s*([a-z_][a-z0-9_]*)",
                     sql,
                     re.IGNORECASE,
                 )

scripts/run_gcp_generated_db_direct_claim_suite.py

@@ -53,7 +53,7 @@ DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password"
 DEFAULT_SSL_ROOT_CERT = Path("/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem")
 DEFAULT_SYSTEM_IDENTIFIER = "7659718422914359312"
 DEFAULT_MANIFEST_SQL = HERE.parent / "ops" / "postgres_parity_manifest.sql"
-REVIEWED_CLOUDSQL_TOOL_SHA256 = "cc435231adab4325489936ae18a156224ab0d4a9cb158dad46b4491511b64ea3"
+REVIEWED_CLOUDSQL_TOOL_SHA256 = "5f98218622803e0a1bbedc1e23e538d1f152ecb516e31a387a5eb39080aaf2cd"
 COUNT_READBACK_RE = re.compile(
     r"DB readback:\s*claims:\s*`?(?P<claims>\d+)`?;\s*"
     r"sources:\s*`?(?P<sources>\d+)`?;\s*"

tests/test_gcp_leoclean_runtime_permissions.py

@@ -672,6 +672,110 @@ def test_runtime_query_contract_rejects_a_new_unallowlisted_column() -> None:
         query_contract.verify_runtime_read_columns(mutated, expected)
 
 
+@pytest.mark.parametrize(
+    "query",
+    (
+        'select c.future_private_text from "public"."claims" as c',
+        "select c . future_private_text from public.claims as c",
+        "select c /* qualifier */ . /* column */ future_private_text "
+        "from public /* schema */ . /* relation */ claims as c",
+    ),
+    ids=("quoted-identifiers", "spaced-qualification", "commented-qualification"),
+)
+def test_runtime_query_contract_normalizes_valid_postgres_identifier_qualification(query: str) -> 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")
+    expected = {
+        (schema, relation): frozenset(columns)
+        for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST
+    }
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        f"    run_psql(args, {query!r})\n" '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"):
+        query_contract.verify_runtime_read_columns(mutated, expected)
+
+
+def test_runtime_query_contract_rejects_search_path_dependent_base_relation() -> 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")
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        "    run_psql(args, 'select c.future_private_text from claims as c')\n"
+        '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="unqualified relation 'claims'"):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
+def test_runtime_query_contract_tracks_postgres_only_relation_syntax() -> 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")
+    expected = {
+        (schema, relation): frozenset(columns)
+        for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST
+    }
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        "    run_psql(args, 'select c.future_private_text from ONLY public.claims as c')\n"
+        '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"):
+        query_contract.verify_runtime_read_columns(mutated, expected)
+
+
+@pytest.mark.parametrize(
+    "query",
+    (
+        "select c.future_private_text from pg_catalog.pg_tables as p, public.claims as c",
+        "select c.future_private_text from (values (1)) as v(x), claims as c",
+        "select c.id, s.future_private_text from public.claims as c, public.sources as s",
+    ),
+    ids=("qualified-after-table", "unqualified-after-subquery", "qualified-after-canonical"),
+)
+def test_runtime_query_contract_rejects_comma_separated_base_relations(query: str) -> 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")
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        f"    run_psql(args, {query!r})\n" '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="comma-separated base relation"):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
+@pytest.mark.parametrize("alias", ("where", "order"))
+def test_runtime_query_contract_rejects_quoted_structural_keyword_aliases(alias: str) -> 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")
+    query = (
+        f'select "{alias}".id, s.future_private_text '
+        f'from public.claims as "{alias}", public.sources as s'
+    )
+    mutated = runtime_source.replace(
+        '    claims_sql = f"""\n',
+        f"    run_psql(args, {query!r})\n" '    claims_sql = f"""\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="quoted SQL identifier"):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
 def test_runtime_query_contract_rejects_unqualified_canonical_columns() -> 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")
@@ -746,6 +850,100 @@ def test_runtime_query_contract_fails_closed_for_adversarial_query_mutations(
         query_contract.verify_runtime_read_columns(mutated, expected)
 
 
+@pytest.mark.parametrize(
+    ("replacement", "error"),
+    (
+        (
+            '    claims_sql = "select " + "pg_catalog.lo_from_bytea(0, \'\'::bytea)"\n'
+            "    run_psql(args, claims_sql)\n"
+            '    claims_sql = f"""\n',
+            "must have exactly one assignment",
+        ),
+        (
+            '    claims_sql = f"""\n',
+            "placeholder",
+        ),
+        (
+            "    runner = run_psql\n"
+            "    runner(args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n"
+            '    claims_sql = f"""\n',
+            "must be called directly",
+        ),
+        (
+            "    globals()['run_psql'](args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n"
+            '    claims_sql = f"""\n',
+            "indirect call expressions",
+        ),
+        (
+            "    globals()['run_' + 'psql'](args, \"select c.future_private_text "
+            "from public.claims as c\")\n"
+            '    claims_sql = f"""\n',
+            "indirect call expressions",
+        ),
+        (
+            "    get_namespace = globals\n"
+            "    namespace = get_namespace()\n"
+            "    runner = namespace['run_' + 'psql']\n"
+            "    runner(args, \"select c.future_private_text from public.claims as c\")\n"
+            '    claims_sql = f"""\n',
+            "dynamic call indirection",
+        ),
+        (
+            "    {'query': run_psql}['query'](args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n"
+            '    claims_sql = f"""\n',
+            "indirect call expressions",
+        ),
+    ),
+    ids=(
+        "late-reassignment",
+        "augmented-assignment",
+        "runner-alias",
+        "globals-indirection",
+        "computed-globals-indirection",
+        "aliased-globals-indirection",
+        "dict-indirection",
+    ),
+)
+def test_runtime_query_contract_rejects_query_runner_flow_bypasses(replacement: str, error: str) -> 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")
+    if error == "placeholder":
+        mutated = runtime_source.replace(
+            "    claims = psql_json_lines(args, claims_sql, db=args.canonical_db)\n",
+            "    claims_sql += \"\\nselect pg_catalog.lo_from_bytea(0, ''::bytea)\"\n"
+            "    claims = psql_json_lines(args, claims_sql, db=args.canonical_db)\n",
+            1,
+        )
+        error = "must have exactly one assignment"
+    else:
+        mutated = runtime_source.replace('    claims_sql = f"""\n', replacement, 1)
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match=error):
+        query_contract.derive_runtime_read_columns(mutated)
+
+
+def test_runtime_query_contract_inspects_extra_calls_inside_psql_json_lines_wrapper() -> 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")
+    anchor = "    rows: list[dict[str, Any]] = []\n"
+    mutated = runtime_source.replace(
+        anchor,
+        anchor + '    run_psql(args, "select c.future_private_text from public.claims as c")\n',
+        1,
+    )
+    assert mutated != runtime_source
+
+    with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"):
+        query_contract.verify_runtime_read_columns(
+            mutated,
+            {
+                (schema, relation): frozenset(columns)
+                for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST
+            },
+        )
+
+
 @pytest.mark.parametrize(
     "routine",
     (

tests/test_gcp_leoclean_runtime_role_postgres.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import importlib.util
 import json
 import os
 import re
@@ -8,6 +9,7 @@ import subprocess
 import time
 import uuid
 from pathlib import Path
+from types import ModuleType
 
 import pytest
 import yaml
@@ -16,6 +18,7 @@ from ops import verify_gcp_leoclean_runtime_permissions as verifier
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
 ROLE_SQL = REPO_ROOT / "ops" / "gcp_leoclean_runtime_role.sql"
+RUNTIME_TOOL = REPO_ROOT / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py"
 RUN_ENV = "TELEO_RUN_LEOCLEAN_RUNTIME_POSTGRES_CANARY"
 POSTGRES_IMAGE = "postgres@sha256:eb4759788a2182f08257135e61a34f2cfc3c2914079f3465d64ee62350f4d081"
 POSTGRES_VERSION_NUM = "160014"
@@ -25,6 +28,14 @@ LOCAL_RUNTIME_PASSWORD = "disposable-r2-runtime-password"
 IDENTIFIER_RE = re.compile(r"[a-z_][a-z0-9_]*\Z")
 
 
+def _load_runtime_tool() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("leoclean_runtime_canary_contract", RUNTIME_TOOL)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
 def test_ci_runs_digest_pinned_permission_canary_and_always_cleans_up() -> None:
     workflow_path = REPO_ROOT / ".github" / "workflows" / "ci.yml"
     workflow = yaml.safe_load(workflow_path.read_text(encoding="utf-8"))
@@ -83,12 +94,13 @@ def _psql(
     *,
     role: str = "postgres",
     database: str = DATABASE,
+    pgoptions: str | None = None,
 ) -> subprocess.CompletedProcess[str]:
-    return _run(
+    command = ["docker", "exec", "-i"]
+    if pgoptions is not None:
+        command.extend(("--env", f"PGOPTIONS={pgoptions}"))
+    command.extend(
         [
-            "docker",
-            "exec",
-            "-i",
             container,
             "psql",
             "--no-psqlrc",
@@ -101,9 +113,9 @@ def _psql(
             role,
             "--dbname",
             database,
-        ],
+        ]
-        input_text=sql,
     )
+    return _run(command, input_text=sql)
 
 
 def _assert_denied(container: str, sql: str, sqlstate: str = "42501") -> None:
@@ -615,30 +627,47 @@ def _assert_exact_read_contract(container: str) -> None:
         _assert_denied(container, f"select {_quote_identifier(denied_column)} from {qualified} limit 0;")
 
 
-def _assert_proposal_function_and_rollback(container: str) -> None:
+def _assert_proposal_function_and_rollback(container: str, runtime_tool: ModuleType) -> None:
+    crafted_rationale = "permission canary\\'; select pg_catalog.lo_from_bytea(0, ''::bytea); --"
+    crafted_rationale_sql = runtime_tool.sql_literal(crafted_rationale)
     calls = []
     for proposal_type in ("revise_claim", "revise_strategy", "add_edge"):
+        rationale = crafted_rationale_sql if proposal_type == "revise_claim" else "'permission canary'"
         calls.append(
             "select concat_ws('|', staged->>'proposal_type', staged->>'status', "
             "staged->>'proposed_by_handle', staged->>'channel') "
             "from (select kb_stage.stage_leoclean_proposal("
-            f"'{proposal_type}', '', 'r2:{proposal_type}', 'permission canary', "
+            f"'{proposal_type}', '', 'r2:{proposal_type}', {rationale}, "
             "'{\"canary\": true}'::jsonb) as staged) probe;"
         )
     sql = "\n".join(
         [
             "begin;",
+            "select 'strings=' || current_setting('standard_conforming_strings');",
+            "select 'readonly=' || current_setting('transaction_read_only');",
             *calls,
+            f"select 'crafted=' || count(*) from kb_stage.kb_proposals where rationale = {crafted_rationale_sql};",
             "select 'inside=' || count(id) from kb_stage.kb_proposals;",
             "rollback;",
             "select 'after=' || count(id) from kb_stage.kb_proposals;",
         ]
     )
-    output = _require_success(_psql(container, sql, role=RUNTIME_ROLE), "proposal function rollback")
+    output = _require_success(
+        _psql(
+            container,
+            sql,
+            role=RUNTIME_ROLE,
+            pgoptions=runtime_tool.RUNTIME_STANDARD_STRINGS_PGOPTIONS,
+        ),
+        "proposal function rollback",
+    )
     lines = [line.strip() for line in output.splitlines() if line.strip()]
+    assert "strings=on" in lines
+    assert "readonly=off" in lines
     assert "revise_claim|pending_review|leo|telegram" in lines
     assert "revise_strategy|pending_review|leo|telegram" in lines
     assert "add_edge|pending_review|leo|telegram" in lines
+    assert "crafted=1" in lines
     assert "inside=3" in lines
     assert "after=0" in lines
 
@@ -735,9 +764,62 @@ select 'after=' || count(*)
     assert lines == ["before=0", "created=true", "inside=1", "after=0"]
 
 
+def _assert_runtime_read_only_guard(container: str, runtime_tool: ModuleType) -> None:
+    read_only_options = runtime_tool.RUNTIME_READ_ONLY_PGOPTIONS
+    owner_count_sql = f"""
+select count(*)
+  from pg_catalog.pg_largeobject_metadata metadata
+ where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}');
+"""
+    assert _require_success(_psql(container, owner_count_sql), "large-object count before read guard") == "0"
+    assert (
+        _require_success(
+            _psql(
+                container,
+                "show standard_conforming_strings;",
+                role=RUNTIME_ROLE,
+                pgoptions=read_only_options,
+            ),
+            "runtime string parsing guard",
+        )
+        == "on"
+    )
+    assert (
+        _require_success(
+            _psql(
+                container,
+                "show transaction_read_only;",
+                role=RUNTIME_ROLE,
+                pgoptions=read_only_options,
+            ),
+            "runtime read-only transaction guard",
+        )
+        == "on"
+    )
+    _require_success(
+        _psql(
+            container,
+            "select id, text from public.claims limit 0;",
+            role=RUNTIME_ROLE,
+            pgoptions=read_only_options,
+        ),
+        "allowed read through runtime read-only guard",
+    )
+    denied = _psql(
+        container,
+        "select pg_catalog.lo_from_bytea(0, ''::bytea);",
+        role=RUNTIME_ROLE,
+        pgoptions=read_only_options,
+    )
+    assert denied.returncode != 0
+    assert re.search(r"ERROR:\s+25006:", denied.stderr), denied.stderr
+    assert _require_success(_psql(container, owner_count_sql), "large-object count after read guard") == "0"
+
+
 @pytest.mark.skipif(os.environ.get(RUN_ENV) != "1", reason=f"set {RUN_ENV}=1 for the local PG16 canary")
 def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_path: Path) -> None:
     assert shutil.which("docker") is not None, "Docker is required for the disposable permission canary"
+    runtime_tool = _load_runtime_tool()
 
     image_output = _require_success(
         _run(["docker", "image", "inspect", "--format", "{{.Id}}|{{json .RepoDigests}}", POSTGRES_IMAGE]),
@@ -930,7 +1012,8 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p
         }
 
         _assert_exact_read_contract(container)
-        _assert_proposal_function_and_rollback(container)
+        _assert_runtime_read_only_guard(container, runtime_tool)
+        _assert_proposal_function_and_rollback(container, runtime_tool)
         _assert_write_and_escalation_denials(container)
 
         _require_success(

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -17,6 +17,7 @@ from types import SimpleNamespace
 
 import pytest
 
+from ops import derive_leoclean_runtime_query_contract as query_contract
 from scripts.working_leo_open_ended_benchmark import (
     M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS,
     score_reply,
@@ -657,9 +658,195 @@ def test_cloudsql_psql_uses_only_explicit_pg_environment_fields(monkeypatch: pyt
         "PGSSLROOTCERT": module.RUNTIME_SSL_ROOT_CERT,
         "PGCONNECT_TIMEOUT": "8",
         "PGPASSWORD": "scoped-runtime-password",
+        "PGOPTIONS": module.RUNTIME_READ_ONLY_PGOPTIONS,
     }
 
 
+def test_cloudsql_runtime_rejects_dynamic_large_object_sql_before_credentials_or_subprocess(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    args = _runtime_connection_args(module)
+    sql = "select " + "pg_catalog.lo_from_bytea(0, ''::bytea);"
+    monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired"))
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_args, **_kwargs: pytest.fail("psql must not be launched"),
+    )
+
+    with pytest.raises(SystemExit, match="large-object mutators"):
+        module.run_psql(args, sql)
+
+
+@pytest.mark.parametrize(
+    "sql",
+    (
+        "select 1; select 2;",
+        r"\copy public.claims to '/tmp/probe'",
+        "with changed as (update public.claims set status = 'closed' returning id) select id from changed;",
+        "copy public.claims to stdout;",
+        "select 1 into temporary runtime_probe;",
+        "call public.apply_proposal();",
+        "do $$ begin perform pg_catalog.lo_create(0); end $$;",
+        "select pg_catalog.set_config('default_transaction_read_only', 'off', false);",
+        "select pg_catalog.nextval('runtime_probe_sequence');",
+        "select pg_catalog.pg_notify('runtime_probe', 'payload');",
+        "select pg_catalog.pg_try_advisory_lock_shared(42);",
+        "select pg_catalog.pg_advisory_unlock_all();",
+        r'''select pg_catalog.U&"lo\005ffrom\005fbytea"(0, ''::bytea);''',
+        "select 1 /* unterminated",
+        "select 'unterminated",
+    ),
+)
+def test_cloudsql_runtime_sql_boundary_rejects_unreviewed_statement_shapes(sql: str) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+
+    with pytest.raises(SystemExit, match="Runtime SQL was rejected"):
+        module.validate_runtime_sql(sql)
+
+
+def test_cloudsql_runtime_sql_boundary_ignores_keywords_inside_literals_and_comments() -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+
+    module.validate_runtime_sql(
+        "select 'update; \\lo_unlink pg_catalog.lo_from_bytea(0)'::text /* delete public.claims */;"
+    )
+
+
+def test_cloudsql_sql_literal_with_backslash_quote_remains_one_validated_value() -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    payload = "receipt\\'; select pg_catalog.lo_from_bytea(0, ''::bytea); --"
+    sql = f"select {module.sql_literal(payload)}::text;"
+
+    module.validate_runtime_sql(sql)
+    assert sql == "select 'receipt\\''; select pg_catalog.lo_from_bytea(0, ''''::bytea); --'::text;"
+
+
+def test_cloudsql_runtime_rejects_carriage_return_statement_smuggling_before_execution(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    args = _runtime_connection_args(module)
+    sql = (
+        "select kb_stage.stage_leoclean_proposal("
+        "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb) "
+        "-- hidden from an LF-only lexer\r; select pg_catalog.lo_from_bytea(0, ''::bytea);"
+    )
+    monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired"))
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_args, **_kwargs: pytest.fail("psql must not be launched"),
+    )
+
+    with pytest.raises(SystemExit, match="carriage returns"):
+        module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY)
+
+
+def test_cloudsql_runtime_sql_boundary_separates_reads_from_the_one_proposal_function() -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    proposal_sql = """
+with staged as (
+  select kb_stage.stage_leoclean_proposal(
+    'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb
+  ) as proposal
+)
+select proposal from staged;
+"""
+
+    module.validate_runtime_sql(sql=proposal_sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY)
+    with pytest.raises(SystemExit, match="application functions are not allowed for reads"):
+        module.validate_runtime_sql(proposal_sql)
+    with pytest.raises(SystemExit, match="large-object mutators"):
+        module.validate_runtime_sql(
+            proposal_sql.replace(
+                "select proposal from staged;",
+                "select proposal, pg_catalog.lo_from_bytea(0, ''::bytea) from staged;",
+            ),
+            capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+        )
+    with pytest.raises(SystemExit, match="exceeds the reviewed staging function surface"):
+        module.validate_runtime_sql(
+            proposal_sql.replace(
+                "select proposal from staged;",
+                "select proposal, kb_stage.stage_leoclean_proposal("
+                "'revise_claim', null, null, 'again', '{}'::jsonb) from staged;",
+            ),
+            capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+        )
+    with pytest.raises(SystemExit, match="side-effecting system functions"):
+        module.validate_runtime_sql(
+            proposal_sql.replace(
+                "select proposal from staged;",
+                "select proposal, pg_catalog.pg_notify('attacker', 'payload') from staged;",
+            ),
+            capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
+        )
+
+
+def test_every_mechanically_extracted_runtime_query_passes_its_exact_capability() -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    runtime_source = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text(encoding="utf-8")
+    queries = query_contract.extract_runtime_query_literals(runtime_source)
+
+    assert len(queries) == 18
+    assert [capability for _line, _sql, capability in queries].count(
+        module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY
+    ) == 2
+    for _line, sql, capability in queries:
+        module.validate_runtime_sql(sql, capability=capability)
+
+
+def test_cloudsql_stage_proposal_capability_reaches_psql_without_read_only_mode(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    args = _runtime_connection_args(module)
+    _configure_runtime_environment(monkeypatch)
+    captured: dict[str, object] = {}
+    sql = (
+        "select kb_stage.stage_leoclean_proposal("
+        "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb);"
+    )
+
+    def fake_run(command, **kwargs):
+        captured["command"] = command
+        captured["kwargs"] = kwargs
+        return SimpleNamespace(returncode=0, stdout='{"id":"proposal-id"}\n', stderr="")
+
+    monkeypatch.setattr(module, "runtime_password_from_metadata", lambda: "scoped-runtime-password")
+    monkeypatch.setattr(module.subprocess, "run", fake_run)
+
+    module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY)
+
+    assert captured["command"] == ["/usr/bin/psql", "-X", "-At", "-q", "-v", "ON_ERROR_STOP=1"]
+    kwargs = captured["kwargs"]
+    assert isinstance(kwargs, dict)
+    assert kwargs["env"]["PGOPTIONS"] == module.RUNTIME_STANDARD_STRINGS_PGOPTIONS
+
+
+def test_cloudsql_clone_mode_rejects_proposal_capability_before_credentials_or_subprocess(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
+    args = _runtime_connection_args(module)
+    args.credential_mode = "clone-readonly"
+    sql = (
+        "select kb_stage.stage_leoclean_proposal("
+        "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb);"
+    )
+    monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired"))
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_args, **_kwargs: pytest.fail("psql must not be launched"),
+    )
+
+    with pytest.raises(SystemExit, match="requires the scoped runtime credential mode"):
+        module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY)
+
+
 def test_cloudsql_runtime_metadata_and_secret_requests_are_exact_and_silent(
     monkeypatch: pytest.MonkeyPatch,
     capsys: pytest.CaptureFixture[str],
@@ -927,10 +1114,11 @@ def test_cloudsql_runtime_read_marker_rejects_wrong_live_identity(
 
 def test_cloudsql_proposal_calls_leave_leo_identity_to_the_database(monkeypatch: pytest.MonkeyPatch) -> None:
     module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
-    captured: list[str] = []
+    captured: list[tuple[str, str]] = []
 
-    def fake_psql_json_lines(_args, sql, db=None):
+    def fake_psql_json_lines(_args, sql, db=None, *, capability="read"):
-        captured.append(sql)
+        module.validate_runtime_sql(sql, capability=capability)
+        captured.append((sql, capability))
         return [{"id": "proposal-id", "database": db}]
 
     monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines)
@@ -961,12 +1149,16 @@ def test_cloudsql_proposal_calls_leave_leo_identity_to_the_database(monkeypatch:
     module.propose_core_change(core_args)
     module.propose_edge(edge_args)
 
-    assert "p.proposed_by_handle" not in captured[0]
+    assert [capability for _sql, capability in captured] == [
-    assert "as proposed_by_handle" not in captured[0]
+        module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
-    assert "p.proposed_by_handle" not in captured[1]
+        module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY,
-    assert "as proposed_by_handle" not in captured[1]
+    ]
-    assert "stage_leoclean_proposal(\n    p.proposal_type,\n    p.channel," in captured[0]
+    assert "p.proposed_by_handle" not in captured[0][0]
-    assert "stage_leoclean_proposal(\n    'add_edge',\n    p.channel," in captured[1]
+    assert "as proposed_by_handle" not in captured[0][0]
+    assert "p.proposed_by_handle" not in captured[1][0]
+    assert "as proposed_by_handle" not in captured[1][0]
+    assert "stage_leoclean_proposal(\n    p.proposal_type,\n    p.channel," in captured[0][0]
+    assert "stage_leoclean_proposal(\n    'add_edge',\n    p.channel," in captured[1][0]
 
 
 def test_gcp_runtime_role_uses_one_narrow_staging_function() -> None: