Merge branch 'main' into kenli/cp-shape-optimization

Author: kyli87

.github/workflows/ai-review.yml

@@ -377,9 +377,9 @@ jobs:
           import json, os, pathlib
           gw = os.environ['GATEWAY_BASE_URL'].rstrip('/')
           model = os.environ.get('MODEL') or 'claude-opus-4-1'
-          claude_maintainer_model = os.environ.get('CLAUDE_MAINTAINER_MODEL') or 'claude-opus-4-8'
+          claude_maintainer_model = os.environ.get('CLAUDE_MAINTAINER_MODEL') or model
           codex_maintainer_model = os.environ.get('CODEX_MAINTAINER_MODEL') or 'gpt-5-codex'
-          disprove_model = os.environ.get('DISPROVE_MODEL') or 'gpt-5.6-sol'
+          disprove_model = os.environ.get('DISPROVE_MODEL') or 'gpt-5-6-sol'
           cfg = {
               'providers': {
                   'gateway': {
@@ -407,6 +407,39 @@ jobs:
           pathlib.Path.home().joinpath('.omnigent', 'config.yaml').write_text(json.dumps(cfg, indent=2))
           "
 
+      - name: Materialize reviewer model config
+        if: steps.creds.outputs.available == 'true'
+        env:
+          CLAUDE_MAINTAINER_MODEL: ${{ vars.CLAUDE_MAINTAINER_MODEL }}
+          CODEX_MAINTAINER_MODEL: ${{ vars.CODEX_MAINTAINER_MODEL }}
+          DISPROVE_MODEL: ${{ vars.DISPROVE_MODEL }}
+          MODEL: ${{ vars.MODEL }}
+        run: |
+          set -euo pipefail
+          python3 -u <<'PYEOF'
+          import os
+          import pathlib
+          import yaml
+
+          model = os.environ.get("MODEL") or "claude-opus-4-1"
+          replacements = {
+              "default": model,
+              "maintainer-claude": os.environ.get("CLAUDE_MAINTAINER_MODEL") or model,
+              "maintainer-codex": os.environ.get("CODEX_MAINTAINER_MODEL") or "gpt-5-codex",
+              "disprove": os.environ.get("DISPROVE_MODEL") or "gpt-5-6-sol",
+          }
+
+          for path in pathlib.Path(".github/omnigent/reviewer").rglob("config.yaml"):
+              cfg = yaml.safe_load(path.read_text())
+              executor = cfg.get("executor", {})
+              old_model = executor.get("model")
+              if old_model in replacements:
+                  executor["model"] = replacements[old_model]
+                  path.write_text(yaml.safe_dump(cfg, sort_keys=False))
+
+          print("Materialized AI reviewer model aliases for CI runtime.")
+          PYEOF
+
       - name: Collect PR context and build review prompt
         if: steps.creds.outputs.available == 'true'
         env:
@@ -498,17 +531,38 @@ jobs:
         if: steps.creds.outputs.available == 'true'
         id: review
         env:
+          GATEWAY_BASE_URL: ${{ secrets.GATEWAY_BASE_URL }}
           LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
+          MODEL: ${{ vars.MODEL }}
         run: |
           set -euo pipefail
           prompt=$(cat /tmp/review_prompt.txt)
-
+          model="${MODEL:-claude-opus-4-1}"
+
+          export ANTHROPIC_AUTH_TOKEN="$LLM_API_KEY"
+          export ANTHROPIC_BASE_URL="${GATEWAY_BASE_URL%/}/anthropic"
+          export ANTHROPIC_MODEL="$model"
+          export ANTHROPIC_DEFAULT_MODEL="$model"
+          export ANTHROPIC_DEFAULT_OPUS_MODEL="$model"
+          export ANTHROPIC_DEFAULT_SONNET_MODEL="$model"
+          export OPENAI_API_KEY="$LLM_API_KEY"
+          export OPENAI_BASE_URL="${GATEWAY_BASE_URL%/}/openai"
+
+          set +e
           omnigent run .github/omnigent/reviewer/ \
             -p "$prompt" \
             --no-session \
             2>review-stderr.log \
-            | tee /tmp/review_output.txt \
-            || { echo "::warning::AI review exited non-zero"; cat review-stderr.log; }
+            | tee /tmp/review_output.txt
+          review_status="${PIPESTATUS[0]}"
+          set -e
+
+          if [ "$review_status" -ne 0 ]; then
+            echo "::warning::AI review exited with status ${review_status}."
+            if [ -s review-stderr.log ]; then
+              cat review-stderr.log
+            fi
+          fi
 
           python3 -c "
           import pathlib
@@ -528,9 +582,10 @@ jobs:
           echo "review_text<<${delim}" >> "$GITHUB_OUTPUT"
           cat /tmp/review_output.txt >> "$GITHUB_OUTPUT"
           echo "${delim}" >> "$GITHUB_OUTPUT"
+          echo "review_status=${review_status}" >> "$GITHUB_OUTPUT"
 
       - name: Scan review output for secrets before publishing
-        if: steps.creds.outputs.available == 'true'
+        if: always() && steps.creds.outputs.available == 'true'
         env:
           LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
         run: |
@@ -540,6 +595,21 @@ jobs:
             exit 1
           fi
 
+      - name: Validate review output
+        if: always() && steps.creds.outputs.available == 'true'
+        env:
+          REVIEW_STATUS: ${{ steps.review.outputs.review_status }}
+        run: |
+          set -euo pipefail
+          if [ "${REVIEW_STATUS:-1}" -ne 0 ]; then
+            echo "::error::AI review exited with status ${REVIEW_STATUS:-unknown}."
+            exit 1
+          fi
+          if [ ! -s /tmp/review_output.txt ]; then
+            echo "::error::AI review produced no publishable output."
+            exit 1
+          fi
+
       - name: Post review comment
         if: >-
           steps.review.outputs.review_text != '' &&

default-engine/src/executor.rs

@@ -155,11 +155,9 @@ pub mod tokio {
 
             let fut = Box::pin(async move {
                 let task_output = task.await;
-                tokio::task::spawn_blocking(move || {
-                    sender.send(task_output).ok();
-                })
-                .await
-                .unwrap();
+                // `std::sync::mpsc::Sender::send` never waits for channel capacity, so this
+                // synchronous handoff is safe to do directly from the async task.
+                sender.send(task_output).ok();
             });
 
             self.send_future(fut);
@@ -255,9 +253,6 @@ pub mod tokio {
     impl TaskExecutor for TokioMultiThreadExecutor {
         type Guard<'a> = EnterGuard<'a>;
 
-        // `block_on` uses `block_in_place`; If concurrent `block_on` calls exceed Tokio's
-        // `max_blocking_threads`, this can deadlock See:
-        // https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html#method.max_blocking_threads
         fn block_on<T>(&self, task: T) -> T::Output
         where
             T: Future + Send + 'static,
@@ -270,11 +265,9 @@ pub mod tokio {
 
             let fut = Box::pin(async move {
                 let task_output = task.await;
-                tokio::task::spawn_blocking(move || {
-                    sender.send(task_output).ok();
-                })
-                .await
-                .unwrap();
+                // `std::sync::mpsc::Sender::send` never waits for channel capacity, so this
+                // synchronous handoff is safe to do directly from the async task.
+                sender.send(task_output).ok();
             });
 
             // We throw away the handle, but it should continue on.
@@ -414,47 +407,51 @@ pub mod tokio {
         }
 
         #[test]
-        fn test_owned_runtime_small_pool_nested_block_on_deadlocks() {
+        fn test_owned_runtime_small_blocking_pool_completes_many_block_on_calls() {
             use std::sync::Arc;
             use std::time::Duration;
 
-            // Create a small pool
             let executor = Arc::new(
-                TokioMultiThreadExecutor::new_owned_runtime(Some(1), Some(1))
+                TokioMultiThreadExecutor::new_owned_runtime(Some(2), Some(1))
                     .expect("Failed to create executor"),
             );
-            let e1 = executor.clone();
-            let e2 = executor.clone();
-            let e3 = executor.clone();
 
             let (tx, rx) = channel::<i32>();
-
-            // Spawn a thread to do deeply nested block_on calls
-            std::thread::spawn(move || {
-                let result = executor.block_on(async move {
-                    e1.block_on(async move {
-                        e2.block_on(async move {
-                            e3.block_on(async {
+            let handles = executor.block_on({
+                let executor = executor.clone();
+                let tx = tx.clone();
+                async move {
+                    let mut handles = Vec::new();
+                    for value in 0..8 {
+                        let executor = executor.clone();
+                        let tx = tx.clone();
+                        handles.push(tokio::task::spawn_blocking(move || {
+                            let result = executor.block_on(async move {
                                 tokio::time::sleep(Duration::from_millis(1)).await;
-                                42
-                            })
-                        })
-                    })
-                });
-                tx.send(result).ok();
+                                value
+                            });
+                            tx.send(result).ok();
+                        }));
+                    }
+                    handles
+                }
             });
+            drop(tx);
+
+            let mut results = Vec::new();
+            for _ in 0..handles.len() {
+                results.push(
+                    rx.recv_timeout(Duration::from_secs(5))
+                        .expect("Timeout - likely deadlock in TokioMultiThreadExecutor::block_on"),
+                );
+            }
 
-            // With 1 worker thread, 1 blocking thread and 4 nested block_on calls, this should
-            // deadlock
-            let timeout = Duration::from_millis(500);
-            let result = rx.recv_timeout(timeout);
+            for handle in handles {
+                executor.block_on(handle).expect("blocking task panicked");
+            }
 
-            // Test passes if we got a timeout (deadlock occurred as expected)
-            // Test fails if we got a result (no deadlock - unexpected)
-            assert!(
-                result.is_err(),
-                "Expected deadlock with 1 worker thread, 1 blocking thread and 4 nested block_on calls",
-            );
+            results.sort_unstable();
+            assert_eq!(results, (0..8).collect::<Vec<_>>());
         }
 
         #[test]

ffi/src/delta_kernel_unity_catalog.rs

@@ -103,7 +103,8 @@ impl UpdateTableClient for FfiUCCommitClient {
 
             match (self.commit_callback)(self.context, c_commit_request) {
                 OptionalValue::Some(e) => {
-                    let boxed_str = unsafe { e.into_inner() }; // get the string back into Box<String>
+                    let boxed_str = unsafe { e.into_inner() }; // get the string back into
+                                                               // Box<String>
                     let s: String = *boxed_str; // move back onto the stack
                     Err(unity_catalog_delta_client_api::Error::Generic(s))
                 }

ffi/src/ffi_tracing.rs

@@ -925,8 +925,8 @@ mod tests {
     }
 
     #[test]
-    #[ignore] // We cannot run this test if test_enable_log_line_tracing was run before - see comment there,
-              // however this test works if run individually.
+    #[ignore] // We cannot run this test if test_enable_log_line_tracing was run before - see
+              // comment there, however this test works if run individually.
     fn test_enable_event_tracing() {
         let _lock = TEST_LOCK.lock().unwrap();
         setup_events();

kernel/src/action_reconciliation/log_replay.rs

@@ -1033,7 +1033,8 @@ mod tests {
         let (results, actions_count, add_actions) = run_action_reconciliation_test(input_batches)?;
 
         // Verify results
-        assert_eq!(results.len(), 2); // The third batch should be filtered out since there are no selected actions
+        assert_eq!(results.len(), 2); // The third batch should be filtered out since there are no
+                                      // selected actions
         assert_eq!(results[0].selection_vector(), &vec![true]);
         assert_eq!(results[1].selection_vector(), &vec![false, true]);
         assert_eq!(actions_count, 2);

kernel/src/actions/mod.rs

@@ -325,8 +325,7 @@ impl Default for Format {
     Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, IntoStructData,
 )]
 #[serde(rename_all = "camelCase")]
-#[internal_api]
-pub(crate) struct Metadata {
+pub struct Metadata {
     /// Unique identifier for this table
     id: String,
     /// User-provided identifier for this table
@@ -567,10 +566,9 @@ impl IntoEngineData for Metadata {
 // validated by `try_new`, like the JSON-replay path. Otherwise a CRC file could load a malformed
 // feature shape that log replay would reject.
 #[serde(rename_all = "camelCase", try_from = "ProtocolRaw")]
-#[internal_api]
 // TODO move to another module so that we disallow constructing this struct without using the
 // try_new function.
-pub(crate) struct Protocol {
+pub struct Protocol {
     /// The minimum version of the Delta read protocol that a client must implement
     /// in order to correctly read this table
     min_reader_version: i32,
@@ -1538,8 +1536,7 @@ pub(crate) struct CheckpointMetadata {
 #[derive(
     Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, IntoStructData, IntoEngineData,
 )]
-#[internal_api]
-pub(crate) struct DomainMetadata {
+pub struct DomainMetadata {
     domain: String,
     configuration: String,
     removed: bool,
@@ -1572,18 +1569,16 @@ impl DomainMetadata {
         self.domain.starts_with(INTERNAL_DOMAIN_PREFIX)
     }
 
-    #[internal_api]
-    pub(crate) fn domain(&self) -> &str {
+    pub fn domain(&self) -> &str {
         &self.domain
     }
 
-    #[internal_api]
-    pub(crate) fn configuration(&self) -> &str {
+    pub fn configuration(&self) -> &str {
         &self.configuration
     }
 
     /// Returns `true` if this action is a tombstone (marking domain removal).
-    pub(crate) fn is_removed(&self) -> bool {
+    pub fn is_removed(&self) -> bool {
         self.removed
     }
 }