fix: consolidated bot review follow-up sweep (7 findings across 6 merged PRs)

.github/workflows/develop-post-merge.yml

@@ -85,16 +85,26 @@ jobs:
       - name: Emit fast-test signature
         # Bare python is intentional: the signature builder is stdlib-only.
         # Falls back to a job-failure descriptor if pytest crashed before
-        # writing junit XML at all (e.g. a collection-time crash), or left a
-        # truncated/unparseable file behind (killed mid-write) - a signature
-        # artifact must ALWAYS upload, or the auto-revert diff would compare
-        # against a partial current signature and could wrongly suppress.
+        # writing junit XML at all (e.g. a collection-time crash), left a
+        # truncated/unparseable file behind (killed mid-write), or failed for
+        # a reason junit never records (e.g. pytest-cov's --cov-fail-under
+        # gate, where every testcase passes but the step still fails) - a
+        # signature artifact must ALWAYS upload and must never look clean
+        # when the job actually failed, or the auto-revert diff would
+        # compare against a partial/false-clean current signature and could
+        # wrongly suppress.
         if: always()
         shell: bash
         run: |
           set -euo pipefail
+          JOB_FAILED_FLAG=""
+          if [ "${{ job.status }}" = "failure" ]; then
+            JOB_FAILED_FLAG="--job-failed"
+          fi
           if [ -f fast-test-junit.xml ]; then
-            python scripts/post_merge_signature.py build --job fast-test --junit fast-test-junit.xml --out signature.json \
+            python scripts/post_merge_signature.py build --job fast-test --junit fast-test-junit.xml \
+              --failed-step "Run CI fast-test mirror (job failed with no testcase-level failures, e.g. coverage threshold)" \
+              $JOB_FAILED_FLAG --out signature.json \
               || python scripts/post_merge_signature.py build --job fast-test --failed-step "Run CI fast-test mirror (junit xml unparseable)" --out signature.json
           else
             python scripts/post_merge_signature.py build --job fast-test --failed-step "Run CI fast-test mirror (no junit xml produced)" --out signature.json
@@ -209,15 +219,23 @@ jobs:
 
       - name: Emit medium-test signature
         # Bare python is intentional: the signature builder is stdlib-only.
-        # Same missing/unparseable-junit fallback as fast-test: a signature
-        # artifact must always upload so the auto-revert diff never runs on
-        # a partial current signature.
+        # Same missing/unparseable-junit AND job-failed-but-junit-is-clean
+        # fallbacks as fast-test: a signature artifact must always upload and
+        # must never look clean when the job actually failed, or the
+        # auto-revert diff could run on a partial/false-clean current
+        # signature.
         if: always()
         shell: bash
         run: |
           set -euo pipefail
+          JOB_FAILED_FLAG=""
+          if [ "${{ job.status }}" = "failure" ]; then
+            JOB_FAILED_FLAG="--job-failed"
+          fi
           if [ -f medium-test-junit.xml ]; then
-            python scripts/post_merge_signature.py build --job medium-test --junit medium-test-junit.xml --out signature.json \
+            python scripts/post_merge_signature.py build --job medium-test --junit medium-test-junit.xml \
+              --failed-step "Run medium speed tier (job failed with no testcase-level failures)" \
+              $JOB_FAILED_FLAG --out signature.json \
               || python scripts/post_merge_signature.py build --job medium-test --failed-step "Run medium speed tier (junit xml unparseable)" --out signature.json
           else
             python scripts/post_merge_signature.py build --job medium-test --failed-step "Run medium speed tier (no junit xml produced)" --out signature.json
@@ -413,14 +431,25 @@ jobs:
 
           # Merge the four gate jobs' per-job signature artifacts (one
           # subdirectory per artifact, downloaded above) into a single
-          # combined signature for this run.
-          if compgen -G "current-signatures/*/signature.json" > /dev/null; then
-            if ! jq -s '{failure_ids: ([.[].failure_ids] | add | unique | sort)}' \
-              current-signatures/*/signature.json > current-combined.json; then
-              fail_open_reason="Could not combine this run's signature artifacts (malformed signature JSON)."
+          # combined signature for this run. Require ALL FOUR expected gate
+          # artifacts, not just "at least one": if one gate crashes/times out
+          # before its always()-run signature step can upload (or the upload
+          # itself fails), the other three still leave signature.json files
+          # behind, so a bare compgen glob check would silently diff a
+          # partial current signature and could suppress a revert the missing
+          # gate's real failure warranted.
+          EXPECTED_SIGNATURE_JOBS=(lint fast-test explorer-tokens medium-test)
+          missing_signatures=()
+          for job in "${EXPECTED_SIGNATURE_JOBS[@]}"; do
+            if [ ! -f "current-signatures/signature-${job}/signature.json" ]; then
+              missing_signatures+=("${job}")
             fi
-          else
-            fail_open_reason="No current-run signature artifacts were found."
+          done
+          if [ "${#missing_signatures[@]}" -gt 0 ]; then
+            fail_open_reason="Missing current-run signature artifact(s) for: ${missing_signatures[*]} (gate job crashed, timed out, or was cancelled before its signature could upload)."
+          elif ! jq -s '{failure_ids: ([.[].failure_ids] | add | unique | sort)}' \
+            current-signatures/*/signature.json > current-combined.json; then
+            fail_open_reason="Could not combine this run's signature artifacts (malformed signature JSON)."
           fi
 
           if [ -z "${fail_open_reason}" ]; then

_project/DONE/auto-revert-incident/active/fix-session-isolation-placeholder-rows.yaml

@@ -124,5 +124,12 @@ verification:
 scope_limit:
   only_modify:
   - "tests/integration/test_throughput_session_isolation.py"
+  # #1130 review: the primary path (w0/w1, actually taken) lands #1117 as-is
+  # rather than the fallback test-only rewrite; #1117 changed these two
+  # paths. Recorded post-hoc for an accurate scope_limit on this completed
+  # item's audit trail - both were already merged via #1117 before this
+  # correction, not a re-opening of scope.
+  - "benchbox/platforms/base/connection_wrappers.py"
+  - "tests/unit/platforms/base/test_connection_wrappers.py"
 deps:
   needs: []

_project/DONE/uat-hardening/planning/uat-execute-path-unification.yaml

@@ -218,7 +218,7 @@ verification:
     command: "uv run -- python -m pytest tests/uat/ -m fast -q"
     expected_output: "all pass including the new parity test"
   - description: "Dead-code removal is complete"
-    command: "grep -rn 'env_without_pythonpath\\|iter_argv_for_log' tests/uat/ && echo FOUND || echo CLEAN"
+    command: "grep -rn 'env_without_pythonpath\\|iter_argv_for_log' tests/uat/ && { echo FOUND; exit 1; } || echo CLEAN"
     expected_output: "CLEAN"
 
 scope_limit:

_project/DONE/uat-hardening/planning/uat-resume-retirement-artifact-durability.yaml

@@ -169,10 +169,10 @@ anti_patterns:
 
 verification:
   - description: "No resume references remain in Python code"
-    command: "grep -rni 'resume' tests/uat/ --include='*.py' | grep -vi 'resumable\\|datagen'"
+    command: "! (grep -rni 'resume' tests/uat/ --include='*.py' | grep -vi 'resumable\\|datagen')"
     expected_output: "no hits"
   - description: "No resume references remain in Makefile, spec, or operator docs"
-    command: "grep -ni 'resume' Makefile _project/specs/uat-framework.md docs/operations/uat-framework.md tests/uat/README.md | grep -vi 'resumable\\|datagen'"
+    command: "! (grep -ni 'resume' Makefile _project/specs/uat-framework.md docs/operations/uat-framework.md tests/uat/README.md | grep -vi 'resumable\\|datagen')"
     expected_output: "no hits (or only prose explaining the removal)"
   - description: "UAT unit suite green, including new atomicity/ordering/collision tests"
     command: "uv run -- python -m pytest tests/uat/ -m fast -q"

benchbox/core/throughput/runner.py

@@ -46,23 +46,31 @@
 Non-blocking shutdown (return without joining a leaked thread)
 ----------------------------------------------------------------
 ``execute()`` manages the ``ThreadPoolExecutor`` manually (no ``with``
-block) so it can call ``executor.shutdown(wait=False)`` in a ``finally``
-instead of relying on the context manager's implicit
+block) so it can call ``executor.shutdown(wait=False, cancel_futures=True)``
+in a ``finally`` instead of relying on the context manager's implicit
 ``shutdown(wait=True)``. This means ``execute()`` returns as soon as its
 own timeout/accounting window closes -- bounded by ``config.stream_timeout``
 plus classification overhead -- rather than blocking until every submitted
 thread finishes.
 
 On the healthy path (no timeout fires), every future is already ``done()``
-by the time ``as_completed()`` returns, so ``shutdown(wait=False)`` is a
-no-op: there is nothing left to join, and default-timeout behavior for
-healthy runs is unchanged. On the hung path, any future still not
-``done()`` when the deadline elapses is *abandoned*, not killed -- Python
-cannot forcibly stop a running thread. ``shutdown(wait=False)`` only stops
-the executor from accepting new submissions; it does not touch threads
-already running, so the leaked thread keeps executing in the background
-until ``stream_fn`` itself returns (naturally, or via cooperative
-cancellation if enabled) and exits.
+by the time ``as_completed()`` returns, so this call is a no-op: there is
+nothing left to join or cancel, and default-timeout behavior for healthy
+runs is unchanged. On the hung path, two distinct futures can remain
+outstanding at the deadline:
+
+* **Running** (already dispatched to a worker thread): *abandoned*, not
+  killed -- Python cannot forcibly stop a running thread. ``cancel_futures``
+  never touches these (Python's documented semantics only cancel futures
+  that have not started running); the leaked thread keeps executing in the
+  background until ``stream_fn`` itself returns (naturally, or via
+  cooperative cancellation if enabled) and exits.
+* **Queued** (submitted, but ``max_workers < num_streams`` left it still
+  waiting for a worker): ``cancel_futures=True`` cancels these outright, so
+  they never start at all. Without this, a queued stream would begin
+  executing only *after* ``execute()`` has already returned and counted it
+  as timed-out -- extra DB work that can overlap with whatever phase runs
+  next.
 
 **Zombie connection/accounting semantics** (see also
 ``docs/development/throughput-result-alignment.md``): a leaked stream's
@@ -306,11 +314,24 @@ def _record_completed_future(
             # (abandoned, not killed) thread. On the healthy path every
             # future is already done() by this point, so this is a no-op --
             # nothing to join, default timeout behavior for healthy runs is
-            # unchanged. On the hung path, any leaked thread keeps running
-            # to completion in the background; its connection is closed by
-            # its own stream_fn's finally whenever that eventually happens.
-            # See the module docstring "Non-blocking shutdown" section.
-            executor.shutdown(wait=False)
+            # unchanged. On the hung path, any already-RUNNING leaked thread
+            # keeps running to completion in the background; its connection
+            # is closed by its own stream_fn's finally whenever that
+            # eventually happens.
+            #
+            # cancel_futures=True: when max_workers < num_streams, a future
+            # can still be QUEUED (never dispatched to a worker) at the
+            # deadline -- distinct from a leaked RUNNING future. Without
+            # this, such a future would start executing only AFTER
+            # execute() has already returned and counted it as timed-out,
+            # creating extra DB work that can overlap with whatever phase
+            # runs next. cancel_futures only cancels futures that have not
+            # started running (Python's documented semantics); an
+            # already-running future is never touched by it, so the
+            # can't-forcibly-cancel-a-running-thread invariant above is
+            # unaffected. See the module docstring "Non-blocking shutdown"
+            # section.
+            executor.shutdown(wait=False, cancel_futures=True)
 
     @staticmethod
     def compute_metrics(

benchbox/core/tpcds/power_test.py

@@ -315,7 +315,6 @@ def _execute_one_query(
                 if hasattr(connection, "set_query_context"):
                     connection.set_query_context(query_display_id, stream_id=self.config.stream_id)
                 cursor = connection.execute(query_text)
-                rows = cursor.fetchall() if hasattr(cursor, "fetchall") else []
 
                 if hasattr(cursor, "platform_result"):
                     result_dict = cursor.platform_result
@@ -330,6 +329,14 @@ def _execute_one_query(
                     # fallback in _attach_captured_plans.
                     propagate_plan_capture_fields(result_dict, query_result)
 
+                # Count BEFORE commit (#1144 review): when rows_returned
+                # isn't available, _query_result_count falls back to
+                # cursor.fetchall(). Some raw DB-API drivers with unbuffered
+                # SELECT results can reject/invalidate commit() while rows
+                # are still unread, so draining must happen first - matches
+                # the pre-#1137 ordering.
+                result_count = self._query_result_count(cursor)
+
                 if hasattr(connection, "commit"):
                     connection.commit()
             finally:
@@ -340,7 +347,7 @@ def _execute_one_query(
                 {
                     "execution_time_seconds": execution_time,
                     "success": True,
-                    "result_count": len(rows),
+                    "result_count": result_count,
                 }
             )
             result.queries_successful += 1
@@ -365,6 +372,27 @@ def _execute_one_query(
         result.query_results.append(query_result)
         result.queries_executed += 1
 
+    @staticmethod
+    def _query_result_count(cursor: Any) -> int:
+        """Return the true result cardinality for adapter cursors when available.
+
+        Checks platform_result["rows_returned"] first - this never
+        materializes the cursor's row list, so a count-only power run never
+        trips PlatformAdapterCursor's placeholder-materialization warning
+        (#1137: that warning exists to catch VALUE-dependent consumers of
+        fabricated placeholder rows, not this count-only path). Only calls
+        cursor.fetchall() as a fallback when no reported count is available
+        (raw DB-API cursors, test doubles).
+        """
+        platform_result = getattr(cursor, "platform_result", None)
+        if isinstance(platform_result, dict):
+            reported = platform_result.get("rows_returned")
+            if isinstance(reported, int) and reported >= 0:
+                return reported
+        if hasattr(cursor, "fetchall"):
+            return len(cursor.fetchall())
+        return 0
+
     def _finalize_power_metrics(self, result: TPCDSPowerTestResult, start_time: float) -> None:
         """Compute Power@Size, total time, success flag - TPC-DS requires >=70% query success."""
         total_execution_time = elapsed_seconds(start_time)
@@ -581,7 +609,16 @@ def _warm_up_database(self) -> None:
         for query in warm_up_queries:
             try:
                 cursor = self.connection.execute(query)
-                cursor.fetchall()
+                # row_count() reads the already-executed platform_result
+                # directly and never materializes/warns (#1137); fall back to
+                # fetchall() to drain a raw DB-API cursor that has no
+                # row_count(). The result is discarded either way - this is
+                # purely a drain/no-op, not a data dependency.
+                counter = getattr(cursor, "row_count", None)
+                if callable(counter):
+                    counter()
+                elif hasattr(cursor, "fetchall"):
+                    cursor.fetchall()
             except Exception:
                 pass  # Ignore warm-up failures
 
@@ -606,8 +643,7 @@ def _execute_query(self, query_id: Any, query_text: str) -> dict[str, Any]:
 
         try:
             cursor = self.connection.execute(query_text)
-            rows = cursor.fetchall()
-            result["result_count"] = len(rows)
+            result["result_count"] = self._query_result_count(cursor)
             self.connection.commit()
         except Exception as e:
             result["status"] = "error"

benchbox/core/tpch/power_test.py

@@ -310,7 +310,6 @@ def run(self) -> TPCHPowerTestResult:
                             self.connection.set_query_context(query_id, stream_id=self.config.stream_id)
 
                         cursor = self.connection.execute(query_text)
-                        rows = cursor.fetchall() if hasattr(cursor, "fetchall") else []
 
                         # Check for validation failures from platform adapter
                         if hasattr(cursor, "platform_result"):
@@ -326,6 +325,15 @@ def run(self) -> TPCHPowerTestResult:
                             # exact key rather than the ambiguous public-id fallback.
                             propagate_plan_capture_fields(result_dict, query_result)
 
+                        # Count BEFORE commit (#1144 review): when
+                        # rows_returned isn't available, _query_result_count
+                        # falls back to cursor.fetchall(). Some raw DB-API
+                        # drivers with unbuffered SELECT results can
+                        # reject/invalidate commit() while rows are still
+                        # unread, so draining must happen first - matches
+                        # the pre-#1137 ordering.
+                        result_count = self._query_result_count(cursor)
+
                         if hasattr(self.connection, "commit"):
                             self.connection.commit()
                     finally:
@@ -338,7 +346,7 @@ def run(self) -> TPCHPowerTestResult:
                         {
                             "execution_time_seconds": execution_time,
                             "success": True,
-                            "result_count": self._query_result_count(cursor, rows),
+                            "result_count": result_count,
                         }
                     )
 
@@ -423,14 +431,25 @@ def _result_digest_from_cursor(cursor: Any) -> str | None:
         return None
 
     @staticmethod
-    def _query_result_count(cursor: Any, rows: list[Any]) -> int:
-        """Return the true result cardinality for adapter cursors when available."""
+    def _query_result_count(cursor: Any) -> int:
+        """Return the true result cardinality for adapter cursors when available.
+
+        Checks platform_result["rows_returned"] first - this never
+        materializes the cursor's row list, so a count-only power run never
+        trips PlatformAdapterCursor's placeholder-materialization warning
+        (#1137: that warning exists to catch VALUE-dependent consumers of
+        fabricated placeholder rows, not this count-only path). Only calls
+        cursor.fetchall() as a fallback when no reported count is available
+        (raw DB-API cursors, test doubles).
+        """
         platform_result = getattr(cursor, "platform_result", None)
         if isinstance(platform_result, dict):
             reported = platform_result.get("rows_returned")
             if isinstance(reported, int) and reported >= 0:
                 return reported
-        return len(rows)
+        if hasattr(cursor, "fetchall"):
+            return len(cursor.fetchall())
+        return 0
 
     def get_all_queries(self) -> dict[str, str]:
         """Get all queries for the power test."""