feat(tuning): record physical sort rewrites as attested evidence

CHANGELOG.md

@@ -339,7 +339,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   runs with an actionable error instead.
 
 - Result metadata records genuine engine versions (`engine_version`, `engine_version_source: sql_query`) for DuckDB, SQLite and DataFusion queried directly from the active connection rather than package versions.
-
+- A successful physical sort rewrite (CTAS or constraint-preserving in-place sort)
+  is recorded as an executed post-load ledger statement and shown in the run
+  receipt as attested (`Recorded; not catalog-checkable`). Attested evidence never
+  satisfies catalog-verifiable intents and never promotes a run to verified on
+  its own.
 - ClickHouse tuning introspection bounds table queries in SQL by ledger tables and measures snapshot truncation directly on raw query results.
 - The Atom feed uses canonical `/blog/` self and entry links while retaining stable entry IDs for existing subscribers.
 - Deprecated option aliases now show their warning. The Polars `streaming`

benchbox/core/tuning/introspection.py

@@ -27,6 +27,7 @@
 TRANSIENT = "transient"
 MAINTENANCE = "maintenance"
 UNVERIFIABLE = "unverifiable"
+ATTESTED = "attested"
 
 _VERIFIABLE_VERDICTS = frozenset({CORROBORATED, ABSENT, MISMATCH, UNVERIFIABLE})
 
@@ -728,6 +729,30 @@ def _match_object(intent: _Intent, state: IntrospectedState) -> tuple[str, Intro
     return MISMATCH, named or same_table[0]
 
 
+def _attested_sort_columns(statement_text: str) -> tuple[str, ...]:
+    from benchbox.core.tuning.applied_ledger import _split_sql_script
+
+    segments = [segment for segment in _split_sql_script(statement_text) if segment.strip()]
+    for segment in segments:
+        order = _ORDER_BY_RE.search(segment)
+        if order:
+            return normalize_columns(order.group("cols"))
+    for segment in segments:
+        keyword = _ORDER_BY_KEYWORD_RE.search(segment)
+        if keyword is None:
+            continue
+        tail = segment[keyword.end() :]
+        columns: list[str] = []
+        for chunk in tail.split(","):
+            token = re.sub(r"(?i)\b(asc|desc|nulls\s+first|nulls\s+last)\b", " ", chunk).strip()
+            token = token.strip().strip(";")
+            if token:
+                columns.append(token)
+        if columns:
+            return normalize_columns(columns)
+    return ()
+
+
 def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedState | None) -> IntrospectionReceipt:
     platform = introspected_state.platform if introspected_state else "unknown"
     state_error = None
@@ -758,6 +783,20 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta
             )
             continue
 
+        if stmt.phase == PHASE_POST_LOAD and stmt.mechanism == "sorted_ingestion":
+            entries.append(
+                ReceiptEntry(
+                    statement=stmt.statement,
+                    phase=stmt.phase,
+                    verdict=ATTESTED,
+                    kind=KIND_SORT_KEY,
+                    table=stmt.table,
+                    expected_columns=_attested_sort_columns(str(stmt.statement or "")),
+                    reason="physical sort executed; no catalog footprint",
+                )
+            )
+            continue
+
         klass, intents = _classify(stmt)
 
         if klass == TRANSIENT:
@@ -857,6 +896,7 @@ def corroborate(ledger: AppliedTuningLedger, introspected_state: IntrospectedSta
 
 __all__ = [
     "ABSENT",
+    "ATTESTED",
     "CONSTRAINT_FOREIGN_KEY",
     "CONSTRAINT_PRIMARY_KEY",
     "CONSTRAINT_UNIQUE",

benchbox/platforms/base/sorted_ingestion.py

@@ -206,6 +206,34 @@ def _recording_ctas_connection(self, connection: Any) -> Any:
             execute_verbs=("sql",),
         )
 
+    def _attach_sorted_ingestion_evidence(
+        self, statements: list[str], ledger_statement: str, validated_table: str
+    ) -> None:
+        from benchbox.core.tuning.applied_ledger import EXECUTED, PHASE_POST_LOAD
+
+        ledger = getattr(self, "_applied_tuning_ledger", None)
+        if ledger is None:
+            return
+        upgraded = False
+        for statement in statements:
+            for recorded in ledger.statements:
+                if (
+                    recorded.status == EXECUTED
+                    and recorded.phase == PHASE_POST_LOAD
+                    and recorded.mechanism is None
+                    and recorded.statement == statement
+                ):
+                    recorded.mechanism = "sorted_ingestion"
+                    recorded.table = validated_table
+                    upgraded = True
+        if not upgraded:
+            ledger.record(
+                ledger_statement,
+                PHASE_POST_LOAD,
+                mechanism="sorted_ingestion",
+                table=validated_table,
+            )
+
     def _execute_ctas_sort(
         self,
         table_name: str,
@@ -285,6 +313,7 @@ def _execute_ctas_sort(
             raise
         self._sorted_ingestion_applied_tables.append(validated_table)
         self._sorted_ingestion_total_apply_seconds += elapsed_seconds(apply_start)
+        self._attach_sorted_ingestion_evidence(statements, ledger_statement, validated_table)
         sorted_col_names = ", ".join(column.name for column in sorted_columns)
         self.log_verbose(f"Applied CTAS sorting to {table_name}: {sorted_col_names}")
         return True

results-explorer/src/components/RunReceipt.tsx

@@ -457,7 +457,9 @@ function AppliedReceiptDrilldown({ raw }: { raw?: string | null }) {
 }
 
 function AppliedReceiptEntryBody({ entry }: { entry: AppliedReceiptEntry }) {
-  const verdict = receiptText(entry.verdict);
+  const rawVerdict = receiptText(entry.verdict);
+  const verdict =
+    rawVerdict === "attested" ? "attested — Recorded; not catalog-checkable" : rawVerdict;
   const statement = receiptText(entry.statement);
   const fields: { label: string; value: string }[] = [
     { label: "Kind", value: receiptText(entry.kind) },

results-explorer/src/components/__tests__/RunReceipt.test.tsx

@@ -535,6 +535,34 @@ describe("RunReceipt applied-tuning receipt drill-down", () => {
     expect(within(divergent).getByText("catalog reports a different index definition")).toBeTruthy();
   });
 
+  it("labels an attested sort entry as recorded but not catalog-checkable", () => {
+    const receipt = JSON.stringify({
+      platform: "duckdb",
+      corroborated: false,
+      summary: { attested: 1 },
+      entries: [
+        {
+          statement: "CREATE OR REPLACE TABLE lineitem AS SELECT * FROM lineitem ORDER BY l_shipdate",
+          phase: "post_load",
+          verdict: "attested",
+          kind: "sort_key",
+          table: "lineitem",
+          expected_columns: ["l_shipdate"],
+          reason: "physical sort executed; no catalog footprint",
+        },
+      ],
+    });
+    render(<RunReceipt detail={makeDetail({ applied_receipt: receipt })} />);
+
+    const region = screen.getByRole("region", { name: "Run receipt" });
+    const drilldown = within(region).getByTestId("applied-receipt-drilldown");
+    const entries = within(drilldown).getAllByTestId("applied-receipt-entry");
+    expect(entries.length).toBe(1);
+    const [attested] = entries as [HTMLElement];
+    expect(within(attested).getByText("attested — Recorded; not catalog-checkable")).toBeTruthy();
+    expect(within(attested).getByText("physical sort executed; no catalog footprint")).toBeTruthy();
+  });
+
   it("omits fields the receipt did not record rather than inventing them", () => {
     const sparse = JSON.stringify({ entries: [{ verdict: "corroborated" }] });
     render(<RunReceipt detail={makeDetail({ applied_receipt: sparse })} />);

tests/unit/core/tuning/test_sort_evidence_attested.py

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from benchbox.core.tuning.applied_ledger import PHASE_POST_LOAD, AppliedTuningLedger
+from benchbox.core.tuning.introspection import (
+    ATTESTED,
+    KIND_INDEX,
+    IntrospectedObject,
+    IntrospectedState,
+    _attested_sort_columns,
+    corroborate,
+)
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.fast,
+]
+
+
+SORT_STATEMENT = "CREATE OR REPLACE TABLE lineitem AS SELECT * FROM lineitem ORDER BY l_shipdate, l_orderkey"
+INDEX_STATEMENT = 'CREATE INDEX idx_lineitem_sort ON "lineitem" ("l_shipdate")'
+ATTESTED_REASON = "physical sort executed; no catalog footprint"
+
+
+def _index_state() -> IntrospectedState:
+    return IntrospectedState(
+        platform="duckdb",
+        objects=[IntrospectedObject(kind=KIND_INDEX, table="lineitem", columns=("l_shipdate",))],
+    )
+
+
+def _sort_ledger() -> AppliedTuningLedger:
+    ledger = AppliedTuningLedger()
+    ledger.record(INDEX_STATEMENT, PHASE_POST_LOAD)
+    ledger.record(SORT_STATEMENT, PHASE_POST_LOAD, mechanism="sorted_ingestion", table="lineitem")
+    return ledger
+
+
+class TestAttestedVerdict:
+    def test_executed_rewrite_is_attested_not_catalog_checked(self) -> None:
+        receipt = corroborate(_sort_ledger(), _index_state())
+
+        attested = [entry for entry in receipt.entries if entry.verdict == ATTESTED]
+        assert len(attested) == 1
+        assert attested[0].reason == ATTESTED_REASON
+        assert attested[0].kind == "sort_key"
+        assert attested[0].table == "lineitem"
+        assert list(attested[0].expected_columns) == ["l_shipdate", "l_orderkey"]
+
+    def test_index_corroboration_plus_rewrite_is_applied_verified(self) -> None:
+        receipt = corroborate(_sort_ledger(), _index_state())
+
+        assert receipt.corroborated is True
+        assert any(entry.verdict == "corroborated" for entry in receipt.entries)
+        assert any(entry.verdict == ATTESTED for entry in receipt.entries)
+
+    def test_attested_only_never_reaches_verified(self) -> None:
+        ledger = AppliedTuningLedger()
+        ledger.record(SORT_STATEMENT, PHASE_POST_LOAD, mechanism="sorted_ingestion", table="lineitem")
+
+        receipt = corroborate(ledger, _index_state())
+
+        assert receipt.corroborated is False
+        assert [entry.verdict for entry in receipt.entries] == [ATTESTED]
+
+    def test_requested_sort_without_rewrite_is_dropped_and_blocks(self) -> None:
+        ledger = AppliedTuningLedger()
+        ledger.record(INDEX_STATEMENT, PHASE_POST_LOAD)
+        ledger.record_dropped("sorting:LINEITEM (L_SHIPDATE)", "adapter rendered no statement")
+
+        receipt = corroborate(ledger, _index_state())
+
+        assert receipt.corroborated is False
+        assert [entry.verdict for entry in receipt.entries if entry.verdict == ATTESTED] == []
+        assert len(ledger.dropped) == 1
+
+    def test_failed_rewrite_blocks_corroboration(self) -> None:
+        from benchbox.core.tuning.applied_ledger import STATEMENT_FAILED
+
+        ledger = AppliedTuningLedger()
+        ledger.record(
+            SORT_STATEMENT,
+            PHASE_POST_LOAD,
+            status=STATEMENT_FAILED,
+            mechanism="sorted_ingestion",
+            table="lineitem",
+            error=RuntimeError("denied"),
+        )
+
+        receipt = corroborate(ledger, _index_state())
+
+        assert receipt.corroborated is False
+        assert all(entry.verdict != ATTESTED for entry in receipt.entries)
+
+    def test_attested_verdict_is_excluded_from_gate_totals(self) -> None:
+        receipt = corroborate(_sort_ledger(), _index_state())
+
+        assert receipt.summary["gate_relevant_total"] == 1
+        assert receipt.summary[ATTESTED] == 1
+
+
+class TestAttestedSortColumns:
+    def test_parenthesized_order_by(self) -> None:
+        assert _attested_sort_columns("CREATE TABLE t (a INT) ORDER BY (a, b)") == ("a", "b")
+
+    def test_bare_order_by_with_directions(self) -> None:
+        assert _attested_sort_columns("CREATE TABLE t AS SELECT * FROM t ORDER BY a DESC, b ASC") == ("a", "b")
+
+    def test_missing_order_by_is_empty(self) -> None:
+        assert _attested_sort_columns("CREATE TABLE t (a INT)") == ()
+
+
+class TestAttestedPromotion:
+    def _adapter(self, ledger: AppliedTuningLedger) -> MagicMock:
+        adapter = MagicMock()
+        adapter._applied_tuning_ledger = ledger
+        introspector = MagicMock()
+        introspector.introspect.return_value = _index_state()
+        adapter.get_tuning_introspector.return_value = introspector
+        return adapter
+
+    def test_rewrite_plus_index_promotes_to_verified(self) -> None:
+        from benchbox.platforms.base.tuning_trust import corroborate_applied_ledger
+
+        status, payload = corroborate_applied_ledger(self._adapter(_sort_ledger()), MagicMock(), "applied_unverified")
+
+        assert status == "applied_verified"
+        assert payload is not None
+
+    def test_attested_only_stays_unverified(self) -> None:
+        from benchbox.platforms.base.tuning_trust import corroborate_applied_ledger
+
+        ledger = AppliedTuningLedger()
+        ledger.record(SORT_STATEMENT, PHASE_POST_LOAD, mechanism="sorted_ingestion", table="lineitem")
+        status, _payload = corroborate_applied_ledger(self._adapter(ledger), MagicMock(), "applied_unverified")
+
+        assert status == "applied_unverified"
+
+
+class TestCtasSortRecordsLedgerStatement:
+    @pytest.fixture(autouse=True)
+    def _require_duckdb(self):
+        pytest.importorskip("duckdb")
+
+    def test_successful_rewrite_records_post_load_statement(self) -> None:
+        import duckdb
+
+        from benchbox.platforms.duckdb import DuckDBAdapter
+        from tests.unit.platforms.test_duckdb_sorted_loading import (
+            _make_table_tuning_with_sort,
+            _make_tuning_config,
+        )
+
+        conn = duckdb.connect(":memory:")
+        conn.execute("CREATE TABLE lineitem (l_orderkey INTEGER, l_shipdate DATE)")
+        conn.execute("INSERT INTO lineitem VALUES (2, '2024-02-01'), (1, '2024-01-01')")
+
+        with patch("benchbox.platforms.duckdb.duckdb"):
+            adapter = DuckDBAdapter()
+        adapter._applied_tuning_ledger = AppliedTuningLedger()
+
+        result = adapter.apply_ctas_sort(
+            "lineitem", _make_tuning_config({"lineitem": _make_table_tuning_with_sort(["l_shipdate"])}), conn
+        )
+
+        assert result is True
+        recorded = [
+            statement
+            for statement in adapter._applied_tuning_ledger.statements
+            if statement.mechanism == "sorted_ingestion" and statement.phase == PHASE_POST_LOAD
+        ]
+        assert len(recorded) == 1
+        assert recorded[0].table == "lineitem"
+        post_load_executed = [
+            statement
+            for statement in adapter._applied_tuning_ledger.statements
+            if statement.phase == PHASE_POST_LOAD and statement.status == "executed"
+        ]
+        assert len(post_load_executed) == 1
+
+        from benchbox.core.tuning.introspection import ATTESTED, IntrospectedState, corroborate
+
+        receipt = corroborate(adapter._applied_tuning_ledger, IntrospectedState(platform="duckdb", objects=[]))
+        assert [entry.verdict for entry in receipt.entries] == [ATTESTED]
+        assert receipt.entries[0].reason == "physical sort executed; no catalog footprint"
+        assert list(receipt.entries[0].expected_columns) == ["l_shipdate"]
+        assert receipt.corroborated is False