test(table-catalog): record live conformance evidence (#4606)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

scripts/table-catalog/README.md

@@ -27,6 +27,34 @@ python3 scripts/table-catalog/pyiceberg_smoke.py \
   --cleanup
 ```
 
+To persist the successful PyIceberg run as live conformance evidence, provide
+an output path and the operator-recorded deployment metadata:
+
+```bash
+python3 scripts/table-catalog/pyiceberg_smoke.py \
+  --endpoint http://127.0.0.1:9000 \
+  --access-key rustfsadmin \
+  --secret-key rustfsadmin \
+  --bucket rustfs-s3table-smoke \
+  --replace \
+  --cleanup \
+  --rustfs-build rustfs-v1.0.0-beta.8 \
+  --git-sha "$(git rev-parse HEAD)" \
+  --catalog-backing durable-strong \
+  --live-evidence-output /tmp/rustfs-pyiceberg-live-evidence.json
+```
+
+The evidence file uses the schema printed by:
+
+```bash
+python3 scripts/table-catalog/engine_compatibility.py --print-live-evidence-schema
+```
+
+The smoke helper validates the record before writing it. Missing deployment
+metadata, mismatched expected and observed status, invalid row counts, or
+claim promotion beyond the client boundary are treated as evidence failures
+instead of support-matrix proof.
+
 The smoke test covers:
 
 - create or reuse the S3 bucket
@@ -131,6 +159,7 @@ python3 scripts/table-catalog/pyiceberg_smoke.py --print-production-failure-cove
 python3 scripts/table-catalog/pyiceberg_smoke.py --print-vendor-profiles
 python3 scripts/table-catalog/pyiceberg_smoke.py --print-unsupported-inventory
 python3 scripts/table-catalog/pyiceberg_smoke.py --print-production-readiness
+python3 scripts/table-catalog/engine_compatibility.py --print-live-evidence-schema
 ```
 
 Use these outputs when updating release notes, PR descriptions, or follow-up

scripts/table-catalog/engine_compatibility.py

@@ -22,6 +22,48 @@ DEFAULT_SNOWFLAKE_CLIENT_VERSION = "operator-recorded"
 DEFAULT_DATABEND_VERSION = "operator-recorded"
 DEFAULT_TRINO_SERVER = "http://127.0.0.1:8080"
 DEFAULT_DATABEND_DSN = "databend://root:@127.0.0.1:8000/default"
+LIVE_EVIDENCE_SCHEMA = "rustfs-s3tables-live-conformance-evidence"
+LIVE_EVIDENCE_SCHEMA_VERSION = 1
+LIVE_EVIDENCE_REQUIRED_FIELDS = [
+    "schema",
+    "schema_version",
+    "client_name",
+    "client_version",
+    "scenario",
+    "rustfs_build",
+    "git_sha",
+    "catalog_backing",
+    "endpoint",
+    "warehouse",
+    "rest_path",
+    "namespace",
+    "table",
+    "metadata_location",
+    "run_timestamp_utc",
+    "operator",
+    "expected_status",
+    "observed_status",
+    "row_count",
+    "cleanup_result",
+    "claim",
+    "command",
+]
+LIVE_EVIDENCE_ALLOWED_CLAIMS = OrderedDict(
+    [
+        ("PyIceberg", ["automated-smoke"]),
+        ("Spark Iceberg REST catalog", ["manual-live-verified"]),
+        ("Trino Iceberg REST catalog", ["manual-live-read-verified"]),
+        ("DuckDB Iceberg", ["manual-live-read-verified"]),
+        ("Databend", ["manual-live-s3-stage-verified"]),
+        ("Snowflake Open Catalog / Iceberg integrations", ["reference-only"]),
+    ]
+)
+LIVE_EVIDENCE_ROW_COUNT_CLIENTS = {
+    "PyIceberg",
+    "Spark Iceberg REST catalog",
+    "Trino Iceberg REST catalog",
+    "DuckDB Iceberg",
+}
 
 VENDOR_SPARK_PROFILES: dict[str, dict[str, str]] = {
     "rustfs": {
@@ -634,6 +676,159 @@ def live_conformance_evidence(
     )
 
 
+def live_conformance_evidence_schema() -> OrderedDict[str, Any]:
+    return OrderedDict(
+        [
+            ("schema", LIVE_EVIDENCE_SCHEMA),
+            ("schema_version", LIVE_EVIDENCE_SCHEMA_VERSION),
+            ("required_fields", LIVE_EVIDENCE_REQUIRED_FIELDS.copy()),
+            ("claim_promotion", OrderedDict((client, claims.copy()) for client, claims in LIVE_EVIDENCE_ALLOWED_CLAIMS.items())),
+            (
+                "fail_closed_rules",
+                [
+                    "missing required field rejects the evidence record",
+                    "rustfs_build, git_sha, catalog_backing, and client_version must be recorded before claim promotion",
+                    "observed_status must match expected_status before any claim can be promoted",
+                    "expected_status must be pass before any claim can be promoted",
+                    "read-only probes cannot promote write compatibility",
+                    "PyIceberg, Spark, Trino, and DuckDB pass evidence must record row_count=2",
+                    "reference-only providers remain reference-only without a repeatable live run",
+                ],
+            ),
+        ]
+    )
+
+
+def live_conformance_evidence_record(
+    *,
+    client_name: str,
+    client_version: str,
+    scenario: str,
+    rustfs_build: str,
+    git_sha: str,
+    catalog_backing: str,
+    endpoint: str,
+    warehouse: str,
+    rest_path: str,
+    namespace: str,
+    table: str,
+    metadata_location: str,
+    run_timestamp_utc: str,
+    operator: str,
+    expected_status: str,
+    observed_status: str,
+    row_count: int,
+    cleanup_result: str,
+    claim: str,
+    command: str,
+) -> OrderedDict[str, Any]:
+    return OrderedDict(
+        [
+            ("schema", LIVE_EVIDENCE_SCHEMA),
+            ("schema_version", LIVE_EVIDENCE_SCHEMA_VERSION),
+            ("client_name", client_name),
+            ("client_version", client_version),
+            ("scenario", scenario),
+            ("rustfs_build", rustfs_build),
+            ("git_sha", git_sha),
+            ("catalog_backing", catalog_backing),
+            ("endpoint", normalized_endpoint(endpoint)),
+            ("warehouse", warehouse),
+            ("rest_path", normalized_rest_path(rest_path)),
+            ("namespace", namespace),
+            ("table", table),
+            ("metadata_location", metadata_location),
+            ("run_timestamp_utc", run_timestamp_utc),
+            ("operator", operator),
+            ("expected_status", expected_status),
+            ("observed_status", observed_status),
+            ("row_count", row_count),
+            ("cleanup_result", cleanup_result),
+            ("claim", claim),
+            ("command", command),
+        ]
+    )
+
+
+def require_non_empty_string(record: dict[str, Any], field: str) -> str:
+    value = record.get(field)
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"live conformance evidence field {field} must be a non-empty string")
+    return value
+
+
+def require_operator_recorded_value(record: dict[str, Any], field: str) -> str:
+    value = require_non_empty_string(record, field)
+    if value == "operator-recorded":
+        raise ValueError(f"live conformance evidence field {field} must be recorded for claim promotion")
+    return value
+
+
+def validate_live_conformance_evidence(record: dict[str, Any]) -> OrderedDict[str, Any]:
+    for field in LIVE_EVIDENCE_REQUIRED_FIELDS:
+        if field not in record:
+            raise ValueError(f"live conformance evidence is missing required field {field}")
+    if record["schema"] != LIVE_EVIDENCE_SCHEMA:
+        raise ValueError("live conformance evidence schema does not match RustFS S3 Tables evidence schema")
+    if record["schema_version"] != LIVE_EVIDENCE_SCHEMA_VERSION:
+        raise ValueError("live conformance evidence schema_version is not supported")
+
+    client_name = require_non_empty_string(record, "client_name")
+    claim = require_non_empty_string(record, "claim")
+    allowed_claims = LIVE_EVIDENCE_ALLOWED_CLAIMS.get(client_name)
+    if allowed_claims is None:
+        raise ValueError(f"live conformance evidence client_name is not recognized: {client_name}")
+    if claim not in allowed_claims:
+        raise ValueError(f"live conformance evidence claim {claim} is not allowed for {client_name}")
+
+    expected_status = require_non_empty_string(record, "expected_status")
+    observed_status = require_non_empty_string(record, "observed_status")
+    if expected_status != "pass":
+        raise ValueError("live conformance evidence expected_status must be pass for claim promotion")
+    if observed_status != expected_status:
+        raise ValueError("live conformance evidence observed_status does not match expected_status")
+
+    for field in [
+        "client_version",
+        "rustfs_build",
+        "git_sha",
+        "catalog_backing",
+    ]:
+        require_operator_recorded_value(record, field)
+
+    for field in [
+        "scenario",
+        "endpoint",
+        "warehouse",
+        "rest_path",
+        "namespace",
+        "table",
+        "metadata_location",
+        "run_timestamp_utc",
+        "operator",
+        "cleanup_result",
+        "command",
+    ]:
+        require_non_empty_string(record, field)
+
+    row_count = record["row_count"]
+    if not isinstance(row_count, int) or row_count < 0:
+        raise ValueError("live conformance evidence row_count must be a non-negative integer")
+    if expected_status == "pass" and client_name in LIVE_EVIDENCE_ROW_COUNT_CLIENTS and row_count != 2:
+        raise ValueError("live conformance evidence row_count must be 2 for successful table smoke/read probes")
+
+    return OrderedDict(
+        [
+            ("status", "accepted"),
+            ("client_name", client_name),
+            ("scenario", record["scenario"]),
+            ("claim", claim),
+            ("metadata_location", record["metadata_location"]),
+            ("row_count", row_count),
+        ]
+    )
+
+
 def production_operations_guide(
     *,
     endpoint: str,
@@ -1427,6 +1622,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--print-engine-matrix", action="store_true")
     parser.add_argument("--print-vendor-audit", action="store_true")
     parser.add_argument("--print-live-conformance", action="store_true")
+    parser.add_argument("--print-live-evidence-schema", action="store_true")
     parser.add_argument("--print-operations-guide", action="store_true")
     parser.add_argument("--print-spark-config", action="store_true")
     parser.add_argument("--print-spark-sql", action="store_true")
@@ -1508,6 +1704,9 @@ def run(args: argparse.Namespace, output: StringIO | None = None) -> None:
             output,
         )
         printed = True
+    if args.print_live_evidence_schema:
+        print_json({"live_conformance_evidence_schema": live_conformance_evidence_schema()}, output)
+        printed = True
     if args.print_operations_guide:
         print_json(
             {

scripts/table-catalog/pyiceberg_smoke.py

@@ -5,8 +5,10 @@ from __future__ import annotations
 
 import argparse
 import hashlib
+import importlib.metadata
 import json
 import os
+import shlex
 import ssl
 import sys
 import time
@@ -14,6 +16,8 @@ import urllib.error
 import urllib.parse
 import urllib.request
 from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
 from typing import Any
 
 import engine_compatibility
@@ -27,6 +31,11 @@ REQUIRED_STORAGE_CREDENTIAL_KEYS = (
     "s3.secret-access-key",
     "s3.session-token",
 )
+SENSITIVE_COMMAND_FLAGS = {
+    "--access-key",
+    "--secret-key",
+    "--session-token",
+}
 TABLE_MAINTENANCE_CONFIG_VERSION = 1
 IDENTIFIER_SEGMENT_MAX_LEN = 64
 
@@ -264,6 +273,14 @@ class StorageCredential:
     config: dict[str, str]
 
 
+@dataclass(frozen=True)
+class SmokeResult:
+    metadata_location: str
+    row_count: int
+    cleanup_result: str
+    table_warehouse_location: str
+
+
 class RestRequestError(RuntimeError):
     def __init__(self, method: str, path: str, status_code: int, response_body: str) -> None:
         super().__init__(f"{method} {path} failed with HTTP {status_code}: {response_body}")
@@ -404,6 +421,16 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--timeout", type=float, default=float(os.getenv("RUSTFS_TABLE_SMOKE_TIMEOUT", "20")))
     parser.add_argument("--cleanup", action="store_true", help="Drop the smoke table and namespace before exiting.")
     parser.add_argument("--replace", action="store_true", help="Drop an existing table with the same identifier first.")
+    parser.add_argument(
+        "--live-evidence-output",
+        help="Write a validated live conformance evidence JSON record after a successful PyIceberg smoke run.",
+    )
+    parser.add_argument("--rustfs-build", default=os.getenv("RUSTFS_BUILD", "operator-recorded"))
+    parser.add_argument("--git-sha", default=os.getenv("RUSTFS_GIT_SHA", "operator-recorded"))
+    parser.add_argument("--catalog-backing", default=os.getenv("RUSTFS_TABLE_CATALOG_BACKING", "operator-recorded"))
+    parser.add_argument("--operator", default=os.getenv("USER", "operator-recorded"))
+    parser.add_argument("--run-timestamp-utc", default=env_or_none("RUSTFS_TABLE_LIVE_RUN_TIMESTAMP_UTC"))
+    parser.add_argument("--client-version", default=env_or_none("RUSTFS_TABLE_CLIENT_VERSION"))
     parser.add_argument(
         "--skip-catalog-api-probes",
         action="store_true",
@@ -701,6 +728,21 @@ def table_warehouse_location(table: Any) -> str:
     raise RuntimeError("created table did not expose a warehouse location")
 
 
+def table_metadata_location(table: Any) -> str | None:
+    for source in (table, getattr(table, "metadata", None)):
+        if callable(source):
+            source = source()
+        if source is None:
+            continue
+        if isinstance(source, dict):
+            location = string_value(source.get("metadata-location") or source.get("metadata_location"))
+        else:
+            location = string_value(getattr(source, "metadata_location", None) or getattr(source, "metadataLocation", None))
+        if location is not None:
+            return location
+    return None
+
+
 def safe_probe_segment(namespace: str, table: str) -> str:
     source = f"{namespace}-{table}"
     return "".join(char if char.isalnum() or char in {"-", "_", "."} else "-" for char in source)
@@ -864,6 +906,98 @@ def printed_metadata(args: argparse.Namespace) -> bool:
     return printed
 
 
+def current_utc_timestamp() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def pyiceberg_client_version(args: argparse.Namespace) -> str:
+    if args.client_version:
+        return args.client_version
+    try:
+        return importlib.metadata.version("pyiceberg")
+    except importlib.metadata.PackageNotFoundError:
+        return "operator-recorded"
+
+
+def redacted_command(argv: list[str]) -> str:
+    redacted: list[str] = []
+    redact_next = False
+    for arg in argv:
+        if redact_next:
+            redacted.append("<redacted>")
+            redact_next = False
+            continue
+        flag, separator, value = arg.partition("=")
+        if flag in SENSITIVE_COMMAND_FLAGS:
+            if separator:
+                redacted.append(f"{flag}=<redacted>")
+            else:
+                redacted.append(flag)
+                redact_next = True
+            continue
+        redacted.append(arg)
+    return shlex.join(redacted)
+
+
+def pyiceberg_live_evidence_record(
+    args: argparse.Namespace,
+    result: SmokeResult,
+    *,
+    client_version: str,
+    rustfs_build: str,
+    git_sha: str,
+    catalog_backing: str,
+    run_timestamp_utc: str,
+    operator: str,
+    command: str,
+) -> dict[str, Any]:
+    return engine_compatibility.live_conformance_evidence_record(
+        client_name="PyIceberg",
+        client_version=client_version,
+        scenario="create-append-reload-scan-direct-rest-probes",
+        rustfs_build=rustfs_build,
+        git_sha=git_sha,
+        catalog_backing=catalog_backing,
+        endpoint=args.endpoint,
+        warehouse=profile_warehouse(args),
+        rest_path=args.rest_path,
+        namespace=args.namespace,
+        table=args.table,
+        metadata_location=result.metadata_location,
+        run_timestamp_utc=run_timestamp_utc,
+        operator=operator,
+        expected_status="pass",
+        observed_status="pass",
+        row_count=result.row_count,
+        cleanup_result=result.cleanup_result,
+        claim="automated-smoke",
+        command=command,
+    )
+
+
+def write_live_evidence(args: argparse.Namespace, result: SmokeResult) -> None:
+    if not args.live_evidence_output:
+        return
+    record = pyiceberg_live_evidence_record(
+        args,
+        result,
+        client_version=pyiceberg_client_version(args),
+        rustfs_build=args.rustfs_build,
+        git_sha=args.git_sha,
+        catalog_backing=args.catalog_backing,
+        run_timestamp_utc=args.run_timestamp_utc or current_utc_timestamp(),
+        operator=args.operator,
+        command=redacted_command(sys.argv),
+    )
+    validation = engine_compatibility.validate_live_conformance_evidence(record)
+    document = {
+        "live_conformance_evidence": record,
+        "validation": validation,
+    }
+    output_path = Path(args.live_evidence_output)
+    output_path.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
 def install_rustfs_rest_sigv4_adapter(catalog: Any, args: argparse.Namespace, deps: RuntimeDeps) -> None:
     from urllib import parse
 
@@ -1120,7 +1254,7 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
         raise RuntimeError("maintenance worker endpoint did not return audit events")
 
 
-def run_catalog_api_probes(args: argparse.Namespace, deps: RuntimeDeps) -> None:
+def run_catalog_api_probes(args: argparse.Namespace, deps: RuntimeDeps) -> dict[str, Any]:
     table_response = signed_rest_request(args, deps, "GET", table_endpoint_path(args))
     snapshot_id = current_snapshot_id_from_table_response(table_response)
     run_metadata_location_probe(args, deps, table_response)
@@ -1130,9 +1264,10 @@ def run_catalog_api_probes(args: argparse.Namespace, deps: RuntimeDeps) -> None:
     diagnostics = signed_rest_request(args, deps, "GET", table_endpoint_path(args, "/catalog/diagnostics"))
     if not isinstance(diagnostics, dict) or not diagnostics:
         raise RuntimeError("catalog diagnostics endpoint returned an empty response")
+    return table_response
 
 
-def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None:
+def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> SmokeResult:
     endpoint = normalized_endpoint(args.endpoint)
     ensure_local_proxy_bypass(endpoint)
     ensure_aws_env(args.access_key, args.secret_key, args.region)
@@ -1197,18 +1332,37 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None:
     scanned = loaded.scan().to_arrow()
     if scanned.num_rows != 2:
         raise RuntimeError(f"expected 2 rows after append, got {scanned.num_rows}")
+    loaded_table_location = table_warehouse_location(loaded)
+    loaded_metadata_location = table_metadata_location(loaded)
+    table_response = None
 
     if args.skip_catalog_api_probes:
         print("[10/10] skipping direct REST catalog API probes")
+        if args.live_evidence_output and loaded_metadata_location is None:
+            table_response = signed_rest_request(args, deps, "GET", table_endpoint_path(args))
     else:
         print("[10/10] probing direct REST catalog APIs")
-        run_catalog_api_probes(args, deps)
+        table_response = run_catalog_api_probes(args, deps)
+
+    metadata_location = loaded_metadata_location
+    if isinstance(table_response, dict):
+        response_metadata_location = table_response.get("metadata-location")
+        if isinstance(response_metadata_location, str) and response_metadata_location:
+            metadata_location = response_metadata_location
+    if args.live_evidence_output and metadata_location is None:
+        raise RuntimeError("live evidence output requires a metadata location from PyIceberg or loadTable")
 
     if args.cleanup:
         print(f"cleanup: dropping table and namespace {'.'.join(identifier)}")
         cleanup_catalog(catalog, identifier)
 
     print(f"PASS: PyIceberg smoke test completed for {'.'.join(identifier)}")
+    return SmokeResult(
+        metadata_location=metadata_location or "operator-recorded",
+        row_count=scanned.num_rows,
+        cleanup_result="dropped-table-and-namespace" if args.cleanup else "not-requested",
+        table_warehouse_location=loaded_table_location,
+    )
 
 
 def main() -> int:
@@ -1217,7 +1371,8 @@ def main() -> int:
         if printed_metadata(args):
             return 0
         deps = load_runtime_deps()
-        run_smoke(args, deps)
+        result = run_smoke(args, deps)
+        write_live_evidence(args, result)
         return 0
     except Exception as error:
         print(f"ERROR: {error}", file=sys.stderr)

scripts/table-catalog/test_engine_compatibility.py

@@ -410,6 +410,92 @@ class EngineCompatibilityTest(unittest.TestCase):
         self.assertIn("manual-live", " ".join(evidence["promotion_rules"]))
         self.assertIn("not-claimed", " ".join(evidence["promotion_rules"]))
 
+    def test_live_conformance_evidence_record_validates_claim_promotion(self) -> None:
+        record = engine_compatibility.live_conformance_evidence_record(
+            client_name="PyIceberg",
+            client_version="0.10.0",
+            scenario="create-append-reload-scan-direct-rest-probes",
+            rustfs_build="rustfs-test",
+            git_sha="abc123",
+            catalog_backing="durable-strong",
+            endpoint="http://127.0.0.1:9000",
+            warehouse="rustfs-s3table-smoke",
+            rest_path="/iceberg",
+            namespace="smoke",
+            table="events",
+            metadata_location="s3://rustfs-s3table-smoke/tables/table-id/metadata/v2.metadata.json",
+            run_timestamp_utc="2026-07-09T00:00:00Z",
+            operator="ci",
+            expected_status="pass",
+            observed_status="pass",
+            row_count=2,
+            cleanup_result="dropped-table-and-namespace",
+            claim="automated-smoke",
+            command="python3 scripts/table-catalog/pyiceberg_smoke.py --cleanup",
+        )
+
+        validation = engine_compatibility.validate_live_conformance_evidence(record)
+
+        self.assertEqual(record["schema"], "rustfs-s3tables-live-conformance-evidence")
+        self.assertEqual(record["schema_version"], 1)
+        self.assertEqual(validation["status"], "accepted")
+        self.assertEqual(validation["claim"], "automated-smoke")
+        self.assertEqual(validation["client_name"], "PyIceberg")
+
+    def test_live_conformance_evidence_rejects_missing_or_mismatched_results(self) -> None:
+        record = engine_compatibility.live_conformance_evidence_record(
+            client_name="Spark Iceberg REST catalog",
+            client_version="3.5.4/iceberg-1.7.1",
+            scenario="create-namespace-create-table-append-refresh-count-cleanup",
+            rustfs_build="rustfs-test",
+            git_sha="abc123",
+            catalog_backing="durable-strong",
+            endpoint="http://127.0.0.1:9000",
+            warehouse="rustfs-s3table-smoke",
+            rest_path="/iceberg",
+            namespace="smoke",
+            table="events",
+            metadata_location="s3://rustfs-s3table-smoke/tables/table-id/metadata/v2.metadata.json",
+            run_timestamp_utc="2026-07-09T00:00:00Z",
+            operator="ci",
+            expected_status="pass",
+            observed_status="fail",
+            row_count=0,
+            cleanup_result="not-run",
+            claim="manual-live-verified",
+            command="spark-sql -f /tmp/rustfs-s3tables-smoke-events-spark.sql",
+        )
+
+        with self.assertRaisesRegex(ValueError, "observed_status"):
+            engine_compatibility.validate_live_conformance_evidence(record)
+
+        passing = dict(record, observed_status="pass", row_count=2)
+        del passing["metadata_location"]
+        with self.assertRaisesRegex(ValueError, "metadata_location"):
+            engine_compatibility.validate_live_conformance_evidence(passing)
+
+        overclaim = dict(record, observed_status="pass", row_count=2, client_name="Trino Iceberg REST catalog", claim="manual-live-write-verified")
+        with self.assertRaisesRegex(ValueError, "claim"):
+            engine_compatibility.validate_live_conformance_evidence(overclaim)
+
+        unrecorded = dict(record, expected_status="pass", observed_status="pass", row_count=2, rustfs_build="operator-recorded")
+        with self.assertRaisesRegex(ValueError, "rustfs_build"):
+            engine_compatibility.validate_live_conformance_evidence(unrecorded)
+
+        failed_claim = dict(record, expected_status="fail", observed_status="fail", row_count=0)
+        with self.assertRaisesRegex(ValueError, "expected_status"):
+            engine_compatibility.validate_live_conformance_evidence(failed_claim)
+
+    def test_cli_prints_live_conformance_evidence_schema(self) -> None:
+        payload = engine_compatibility.cli_json(["--print-live-evidence-schema"])
+        document = json.loads(payload)
+
+        schema = document["live_conformance_evidence_schema"]
+        self.assertEqual(schema["schema"], "rustfs-s3tables-live-conformance-evidence")
+        self.assertIn("metadata_location", schema["required_fields"])
+        self.assertIn("claim", schema["required_fields"])
+        self.assertEqual(schema["claim_promotion"]["Trino Iceberg REST catalog"], ["manual-live-read-verified"])
+
     def test_production_operations_guide_covers_release_boundaries(self) -> None:
         guide = engine_compatibility.production_operations_guide(
             endpoint="http://127.0.0.1:9000",

scripts/table-catalog/test_pyiceberg_smoke.py

@@ -914,6 +914,66 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
         self.assertEqual(selected["catalog_uri"], "http://127.0.0.1:9000/iceberg")
         self.assertEqual(selected["warehouse"], "rustfs-s3table-smoke")
 
+    def test_pyiceberg_live_evidence_record_uses_runtime_smoke_result(self) -> None:
+        args = self.parse_with_args([
+            "--endpoint",
+            "http://127.0.0.1:9000",
+            "--bucket",
+            "lake",
+            "--namespace",
+            "smoke",
+            "--table",
+            "events",
+            "--live-evidence-output",
+            "/tmp/rustfs-live-evidence.json",
+        ])
+        result = pyiceberg_smoke.SmokeResult(
+            metadata_location="s3://lake/tables/table-id/metadata/v2.metadata.json",
+            row_count=2,
+            cleanup_result="not-requested",
+            table_warehouse_location="s3://lake/tables/table-id",
+        )
+
+        record = pyiceberg_smoke.pyiceberg_live_evidence_record(
+            args,
+            result,
+            client_version="0.10.0",
+            rustfs_build="rustfs-test",
+            git_sha="abc123",
+            catalog_backing="durable-strong",
+            run_timestamp_utc="2026-07-09T00:00:00Z",
+            operator="ci",
+            command="python3 scripts/table-catalog/pyiceberg_smoke.py --live-evidence-output /tmp/rustfs-live-evidence.json",
+        )
+
+        self.assertEqual(record["client_name"], "PyIceberg")
+        self.assertEqual(record["client_version"], "0.10.0")
+        self.assertEqual(record["warehouse"], "lake")
+        self.assertEqual(record["metadata_location"], "s3://lake/tables/table-id/metadata/v2.metadata.json")
+        self.assertEqual(record["row_count"], 2)
+        self.assertEqual(record["claim"], "automated-smoke")
+
+    def test_live_evidence_command_redacts_cli_secrets(self) -> None:
+        command = pyiceberg_smoke.redacted_command(
+            [
+                "pyiceberg_smoke.py",
+                "--endpoint",
+                "http://127.0.0.1:9000",
+                "--access-key",
+                "root-access",
+                "--secret-key=root-secret",
+                "--bucket",
+                "lake",
+            ]
+        )
+
+        self.assertIn("--access-key '<redacted>'", command)
+        self.assertIn("--secret-key", command)
+        self.assertIn("<redacted>", command)
+        self.assertNotIn("root-access", command)
+        self.assertNotIn("root-secret", command)
+        self.assertIn("--bucket lake", command)
+
     def test_print_vendor_profiles_renders_selected_aws_profile(self) -> None:
         args = self.parse_with_args(
             [