Add Leo production preflight collector

docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.json

@@ -5,6 +5,7 @@
     "all_phase_files_non_empty": true,
     "apply_order_matches_manifest": true,
     "post_apply_regression_commands_present": true,
+    "pre_apply_validation_commands_present": true,
     "production_apply_receipt_schema_present": true,
     "readiness_passes": true,
     "readiness_says_not_authorized": true,
@@ -26,7 +27,7 @@
     "rio_strategy_nodes": 2,
     "sources": 28
   },
-  "generated_at_utc": "2026-07-10T01:25:52.995962+00:00",
+  "generated_at_utc": "2026-07-10T01:39:08.583122+00:00",
   "mode": "operator_runbook_generation_only",
   "operator_environment": {
     "operator_commands_run_sql_if_executed": true,
@@ -333,6 +334,16 @@
     "Run the post-apply regression command bundle and retain every artifact.",
     "Record production apply receipt, row counts, service MainPID/NRestarts, and cleanup readback."
   ],
+  "pre_apply_validation_commands": [
+    {
+      "applies_production_packet": false,
+      "command_template": ".venv/bin/python scripts/collect_working_leo_production_preflight.py --execute --out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.json --markdown-out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md",
+      "description": "Collect a production baseline for every preflight SQL file through a read-only transaction wrapper. During an authorized apply window, still run each phase preflight again immediately before its phase.",
+      "expected_artifact": "docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md",
+      "id": "readonly_production_preflight_collector",
+      "mutates_production_db": false
+    }
+  ],
   "production_apply_authorization_present": false,
   "production_apply_executed": false,
   "production_apply_receipt_required_fields": [

docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.md

@@ -1,6 +1,6 @@
 # Working Leo Authorized Apply Runbook
 
-Generated UTC: `2026-07-10T01:25:52.995962+00:00`
+Generated UTC: `2026-07-10T01:39:08.583122+00:00`
 Mode: `operator_runbook_generation_only`
 Status: `ready_not_executed`
 Ready for authorized operator: `True`
@@ -22,6 +22,7 @@ Do not run apply commands unless the operator explicitly authorizes production D
 - `all_phase_files_non_empty`: `True`
 - `all_apply_and_rollback_sql_have_transactions`: `True`
 - `post_apply_regression_commands_present`: `True`
+- `pre_apply_validation_commands_present`: `True`
 - `production_apply_receipt_schema_present`: `True`
 
 ## Preflight Order
@@ -91,6 +92,14 @@ Do not run apply commands unless the operator explicitly authorizes production D
 - `rio_strategy_nodes`: `2`
 - `sources`: `28`
 
+## Pre-Apply Validation Commands
+
+- `readonly_production_preflight_collector`: Collect a production baseline for every preflight SQL file through a read-only transaction wrapper. During an authorized apply window, still run each phase preflight again immediately before its phase.
+  - command: `.venv/bin/python scripts/collect_working_leo_production_preflight.py --execute --out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.json --markdown-out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md`
+  - artifact: `docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md`
+  - mutates production DB: `False`
+  - applies production packet: `False`
+
 ## Post-Apply Required Actions
 
 - Run every postflight SQL file and retain output.

scripts/build_working_leo_authorized_apply_runbook.py

@@ -133,6 +133,24 @@ PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS: list[str] = [
     "known_residual_risks",
 ]
 
+PRE_APPLY_VALIDATION_COMMANDS: list[dict[str, Any]] = [
+    {
+        "id": "readonly_production_preflight_collector",
+        "description": (
+            "Collect a production baseline for every preflight SQL file through a read-only transaction wrapper. "
+            "During an authorized apply window, still run each phase preflight again immediately before its phase."
+        ),
+        "command_template": (
+            ".venv/bin/python scripts/collect_working_leo_production_preflight.py --execute "
+            "--out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.json "
+            "--markdown-out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md"
+        ),
+        "expected_artifact": str(REPO_REPORT_DIR / "working-leo-production-preflight-current.md"),
+        "mutates_production_db": False,
+        "applies_production_packet": False,
+    },
+]
+
 
 def _sha256(path: Path) -> str:
     return hashlib.sha256(path.read_bytes()).hexdigest()
@@ -194,6 +212,9 @@ def build_runbook(report_dir: Path = REPO_REPORT_DIR) -> dict[str, Any]:
         "post_apply_regression_commands_present": all(
             item.get("command_template") and item.get("expected_artifact") for item in POST_APPLY_REGRESSION_COMMANDS
         ),
+        "pre_apply_validation_commands_present": all(
+            item.get("command_template") and item.get("expected_artifact") for item in PRE_APPLY_VALIDATION_COMMANDS
+        ),
         "production_apply_receipt_schema_present": all(PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS),
     }
     ready = all(checks.values())
@@ -228,6 +249,7 @@ def build_runbook(report_dir: Path = REPO_REPORT_DIR) -> dict[str, Any]:
             "Run the post-apply regression command bundle and retain every artifact.",
             "Record production apply receipt, row counts, service MainPID/NRestarts, and cleanup readback.",
         ],
+        "pre_apply_validation_commands": PRE_APPLY_VALIDATION_COMMANDS,
         "post_apply_regression_commands": POST_APPLY_REGRESSION_COMMANDS,
         "production_apply_receipt_required_fields": PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS,
         "claim_ceiling": (
@@ -268,6 +290,13 @@ def write_markdown(path: Path, runbook: dict[str, Any]) -> None:
             lines.append(f"   - `{entry['command_template']}`")
     lines.extend(["", "## Expected Counts After Apply", ""])
     lines.extend(f"- `{key}`: `{value}`" for key, value in sorted(runbook["expected_after_apply_counts"].items()))
+    lines.extend(["", "## Pre-Apply Validation Commands", ""])
+    for item in runbook["pre_apply_validation_commands"]:
+        lines.append(f"- `{item['id']}`: {item['description']}")
+        lines.append(f"  - command: `{item['command_template']}`")
+        lines.append(f"  - artifact: `{item['expected_artifact']}`")
+        lines.append(f"  - mutates production DB: `{item['mutates_production_db']}`")
+        lines.append(f"  - applies production packet: `{item['applies_production_packet']}`")
     lines.extend(["", "## Post-Apply Required Actions", ""])
     lines.extend(f"- {item}" for item in runbook["post_apply_required_actions"])
     lines.extend(["", "## Post-Apply Regression Commands", ""])

scripts/collect_working_leo_production_preflight.py

@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+"""Collect live read-only production preflight output for the Working Leo packet.
+
+This tool runs only the preflight SQL files from the authorized apply runbook.
+It does not run apply, postflight, or rollback SQL. Before any SQL is sent to a
+database, each file is statically checked for mutating SQL and then wrapped in a
+Postgres read-only transaction. The intended use is to attach fresh row-level
+preflight evidence to an operator receipt without crossing the production apply
+authorization boundary.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import subprocess
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import build_working_leo_authorized_apply_runbook as runbook
+
+REPO_REPORT_DIR = Path("docs/reports/leo-working-state-20260709")
+DEFAULT_OUT = REPO_REPORT_DIR / "working-leo-production-preflight-current.json"
+DEFAULT_MARKDOWN_OUT = REPO_REPORT_DIR / "working-leo-production-preflight-current.md"
+DEFAULT_SSH_TARGET = "root@77.42.65.182"
+DEFAULT_SSH_KEY = Path.home() / ".ssh/livingip_hetzner_20260604_ed25519"
+
+FORBIDDEN_SQL_TOKENS = {
+    "alter",
+    "analyze",
+    "begin",
+    "call",
+    "cluster",
+    "commit",
+    "copy",
+    "create",
+    "delete",
+    "drop",
+    "execute",
+    "grant",
+    "insert",
+    "merge",
+    "refresh",
+    "reindex",
+    "revoke",
+    "rollback",
+    "truncate",
+    "update",
+    "vacuum",
+}
+
+
+def _sha256_text(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def _strip_sql_for_token_scan(sql: str) -> str:
+    without_block_comments = re.sub(r"/\*.*?\*/", " ", sql, flags=re.S)
+    without_line_comments = re.sub(r"--.*?$", " ", without_block_comments, flags=re.M)
+    without_strings = re.sub(r"'(?:''|[^'])*'", "''", without_line_comments)
+    without_meta = "\n".join(
+        line for line in without_strings.splitlines() if not line.lstrip().startswith("\\")
+    )
+    return without_meta
+
+
+def forbidden_tokens(sql: str) -> list[str]:
+    scan = _strip_sql_for_token_scan(sql)
+    tokens = {match.group(0).lower() for match in re.finditer(r"\b[a-z_]+\b", scan, flags=re.I)}
+    return sorted(tokens & FORBIDDEN_SQL_TOKENS)
+
+
+def readonly_sql(sql: str) -> str:
+    body_lines = [
+        line for line in sql.splitlines() if line.strip().lower() != r"\set on_error_stop on"
+    ]
+    body = "\n".join(body_lines).strip()
+    return f"\\set ON_ERROR_STOP on\nbegin transaction read only;\n{body}\ncommit;\n"
+
+
+def preflight_entries(report_dir: Path = REPO_REPORT_DIR) -> list[dict[str, Any]]:
+    generated = runbook.build_runbook(report_dir)
+    return generated["phases"]["preflight_order"]
+
+
+def build_ssh_command(args: argparse.Namespace) -> list[str]:
+    command = [
+        "ssh",
+        "-i",
+        str(args.ssh_key),
+        "-o",
+        "BatchMode=yes",
+        "-o",
+        "StrictHostKeyChecking=accept-new",
+        "-o",
+        f"ConnectTimeout={args.connect_timeout}",
+        args.ssh_target,
+        (
+            f"docker exec -i {args.container} "
+            f"psql -U {args.db_user} -d {args.db} -v ON_ERROR_STOP=1 -P pager=off"
+        ),
+    ]
+    return command
+
+
+def command_for_report(args: argparse.Namespace) -> str:
+    return (
+        "ssh -i [SSH_KEY] -o BatchMode=yes -o StrictHostKeyChecking=accept-new "
+        f"-o ConnectTimeout={args.connect_timeout} {args.ssh_target} "
+        f"'docker exec -i {args.container} psql -U {args.db_user} -d {args.db} "
+        "-v ON_ERROR_STOP=1 -P pager=off'"
+    )
+
+
+def collect_preflight(args: argparse.Namespace) -> dict[str, Any]:
+    entries = preflight_entries(args.report_dir)
+    results: list[dict[str, Any]] = []
+    mode = "live_vps_readonly_preflight" if args.execute else "validated_not_executed"
+    command = build_ssh_command(args)
+
+    for index, entry in enumerate(entries, start=1):
+        path = args.report_dir / entry["file"]
+        sql = path.read_text(encoding="utf-8")
+        forbidden = forbidden_tokens(sql)
+        wrapped = readonly_sql(sql)
+        result: dict[str, Any] = {
+            "index": index,
+            "packet_id": entry["packet_id"],
+            "preflight_file": entry["file"],
+            "path": str(path),
+            "sql_sha256": _sha256_text(sql),
+            "readonly_sql_sha256": _sha256_text(wrapped),
+            "forbidden_tokens": forbidden,
+            "static_readonly_ok": not forbidden,
+            "executed": False,
+            "ok": False,
+        }
+        if forbidden:
+            result["error"] = "preflight SQL contains forbidden mutating or transaction token(s)"
+            results.append(result)
+            continue
+
+        if args.execute:
+            proc = subprocess.run(
+                command,
+                input=wrapped,
+                text=True,
+                capture_output=True,
+                timeout=args.timeout,
+                check=False,
+            )
+            result.update(
+                {
+                    "executed": True,
+                    "returncode": proc.returncode,
+                    "stdout": proc.stdout,
+                    "stderr": proc.stderr,
+                    "ok": proc.returncode == 0,
+                }
+            )
+        else:
+            result.update({"ok": True, "stdout": "", "stderr": ""})
+        results.append(result)
+
+    status = "pass" if all(item["ok"] for item in results) else "fail"
+    return {
+        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
+        "mode": mode,
+        "status": status,
+        "execute_requested": args.execute,
+        "mutates_production_db": False,
+        "applies_production_packet": False,
+        "production_apply_executed": False,
+        "readonly_transaction_wrapper": True,
+        "ssh_target": args.ssh_target if args.execute else None,
+        "command_redacted": command_for_report(args) if args.execute else None,
+        "preflight_count": len(results),
+        "passes": sum(1 for item in results if item["ok"]),
+        "failures": [item for item in results if not item["ok"]],
+        "results": results,
+        "claim_ceiling": (
+            "This is read-only production preflight evidence. It can prove current row-level readiness or drift before "
+            "an apply window. It does not authorize production apply and does not prove canonical rows were changed."
+        ),
+    }
+
+
+def write_report(path: Path, report: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def write_markdown(path: Path, report: dict[str, Any]) -> None:
+    lines = [
+        "# Working Leo Production Preflight",
+        "",
+        f"Generated UTC: `{report['generated_at_utc']}`",
+        f"Mode: `{report['mode']}`",
+        f"Status: `{report['status']}`",
+        f"Executed: `{report['execute_requested']}`",
+        f"Mutates production DB: `{report['mutates_production_db']}`",
+        f"Applies production packet: `{report['applies_production_packet']}`",
+        f"Production apply executed: `{report['production_apply_executed']}`",
+        f"Read-only transaction wrapper: `{report['readonly_transaction_wrapper']}`",
+        "",
+        "## Results",
+        "",
+    ]
+    for item in report["results"]:
+        lines.append(
+            f"- `{item['packet_id']}` / `{item['preflight_file']}`: "
+            f"`ok={item['ok']}`, `executed={item['executed']}`, "
+            f"`forbidden_tokens={item['forbidden_tokens']}`"
+        )
+        if item.get("returncode") is not None:
+            lines.append(f"  - returncode: `{item['returncode']}`")
+        stdout = (item.get("stdout") or "").strip()
+        stderr = (item.get("stderr") or "").strip()
+        if stdout:
+            lines.extend(["  - stdout:", "", "```text", stdout, "```", ""])
+        if stderr:
+            lines.extend(["  - stderr:", "", "```text", stderr, "```", ""])
+    lines.extend(["", "## Claim Ceiling", "", report["claim_ceiling"], ""])
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--report-dir", type=Path, default=REPO_REPORT_DIR)
+    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
+    parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT)
+    parser.add_argument("--execute", action="store_true", help="run the preflight SQL against the VPS")
+    parser.add_argument("--ssh-target", default=DEFAULT_SSH_TARGET)
+    parser.add_argument("--ssh-key", type=Path, default=DEFAULT_SSH_KEY)
+    parser.add_argument("--connect-timeout", type=int, default=15)
+    parser.add_argument("--timeout", type=int, default=120)
+    parser.add_argument("--container", default="teleo-pg")
+    parser.add_argument("--db-user", default="postgres")
+    parser.add_argument("--db", default="teleo")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    report = collect_preflight(args)
+    write_report(args.out, report)
+    write_markdown(args.markdown_out, report)
+    print(
+        json.dumps(
+            {
+                "out": str(args.out),
+                "markdown_out": str(args.markdown_out),
+                "status": report["status"],
+                "execute_requested": report["execute_requested"],
+                "passes": report["passes"],
+                "preflight_count": report["preflight_count"],
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+    return 0 if report["status"] == "pass" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_build_working_leo_authorized_apply_runbook.py

@@ -32,7 +32,13 @@ def test_build_runbook_is_ready_not_executed():
     assert report["operator_environment"]["operator_commands_run_sql_if_executed"] is True
     assert "does not execute SQL" in report["claim_ceiling"]
     assert report["checks"]["post_apply_regression_commands_present"] is True
+    assert report["checks"]["pre_apply_validation_commands_present"] is True
     assert report["checks"]["production_apply_receipt_schema_present"] is True
+    assert {item["id"] for item in report["pre_apply_validation_commands"]} == {
+        "readonly_production_preflight_collector",
+    }
+    assert all(item["mutates_production_db"] is False for item in report["pre_apply_validation_commands"])
+    assert all(item["applies_production_packet"] is False for item in report["pre_apply_validation_commands"])
     assert {item["id"] for item in report["post_apply_regression_commands"]} == {
         "direct_claim_handler_suite",
         "direct_claim_strict_score",
@@ -65,6 +71,8 @@ def test_write_markdown_includes_authorization_boundary(tmp_path: Path):
     assert "Production apply authorization present: `False`" in text
     assert "Do not run apply commands unless" in text
     assert 'psql "$TELEO_DATABASE_URL"' in text
+    assert "Pre-Apply Validation Commands" in text
+    assert "collect_working_leo_production_preflight.py --execute" in text
     assert "Post-Apply Regression Commands" in text
     assert "scripts/run_leo_direct_claim_handler_suite.py" in text
     assert "direct_claim_score_path" in text

tests/test_collect_working_leo_production_preflight.py

@@ -0,0 +1,102 @@
+"""Tests for scripts/collect_working_leo_production_preflight.py."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+import collect_working_leo_production_preflight as preflight  # noqa: E402
+
+
+def _args(tmp_path: Path, *, execute: bool = False) -> argparse.Namespace:
+    return argparse.Namespace(
+        report_dir=REPO_ROOT / preflight.REPO_REPORT_DIR,
+        out=tmp_path / "preflight.json",
+        markdown_out=tmp_path / "preflight.md",
+        execute=execute,
+        ssh_target="root@example.invalid",
+        ssh_key=tmp_path / "ssh-key",
+        connect_timeout=3,
+        timeout=10,
+        container="teleo-pg",
+        db_user="postgres",
+        db="teleo",
+    )
+
+
+def test_current_preflight_files_validate_without_execution(tmp_path: Path):
+    report = preflight.collect_preflight(_args(tmp_path))
+
+    assert report["status"] == "pass"
+    assert report["execute_requested"] is False
+    assert report["mutates_production_db"] is False
+    assert report["applies_production_packet"] is False
+    assert report["production_apply_executed"] is False
+    assert report["readonly_transaction_wrapper"] is True
+    assert report["preflight_count"] == 5
+    assert report["passes"] == 5
+    assert all(item["static_readonly_ok"] for item in report["results"])
+    assert all(item["executed"] is False for item in report["results"])
+
+
+def test_forbidden_tokens_reject_mutating_sql():
+    sql = """
+    -- update in a comment is ignored
+    select 'delete from public.claims' as harmless_string;
+    update public.claims set status = 'applied';
+    """
+
+    assert preflight.forbidden_tokens(sql) == ["update"]
+
+
+def test_execute_mode_wraps_sql_readonly_and_uses_ssh(monkeypatch, tmp_path: Path):
+    calls = []
+
+    class Completed:
+        returncode = 0
+        stdout = "ok\n"
+        stderr = ""
+
+    def fake_run(command, *, input, text, capture_output, timeout, check):
+        calls.append(
+            {
+                "command": command,
+                "input": input,
+                "text": text,
+                "capture_output": capture_output,
+                "timeout": timeout,
+                "check": check,
+            }
+        )
+        return Completed()
+
+    monkeypatch.setattr(preflight.subprocess, "run", fake_run)
+
+    report = preflight.collect_preflight(_args(tmp_path, execute=True))
+
+    assert report["status"] == "pass"
+    assert report["execute_requested"] is True
+    assert report["passes"] == 5
+    assert len(calls) == 5
+    for call in calls:
+        assert call["command"][0] == "ssh"
+        assert "docker exec -i teleo-pg psql" in call["command"][-1]
+        assert call["input"].startswith("\\set ON_ERROR_STOP on\nbegin transaction read only;\n")
+        assert call["input"].rstrip().endswith("commit;")
+
+
+def test_markdown_preserves_claim_ceiling(tmp_path: Path):
+    report = preflight.collect_preflight(_args(tmp_path))
+    out = tmp_path / "preflight.md"
+
+    preflight.write_markdown(out, report)
+
+    text = out.read_text()
+    assert "Working Leo Production Preflight" in text
+    assert "Mutates production DB: `False`" in text
+    assert "Production apply executed: `False`" in text
+    assert "does not authorize production apply" in text