SkardiLabs
diff --git a/‎README.md‎
Lines changed: 1 addition & 1 deletion b/‎README.md‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎crates/cli/src/main.rs‎
Lines changed: 26 additions & 16 deletions b/‎crates/cli/src/main.rs‎
Lines changed: 26 additions & 16 deletions
diff --git a/‎crates/server/src/optimizer_registry.rs‎
Lines changed: 1 addition & 1 deletion b/‎crates/server/src/optimizer_registry.rs‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎crates/skardi/src/sources/providers/open_connector/action_registry.rs‎
Lines changed: 1 addition & 1 deletion b/‎crates/skardi/src/sources/providers/open_connector/action_registry.rs‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎crates/skardi/src/sources/providers/open_connector/error.rs‎
Lines changed: 46 additions & 0 deletions b/‎crates/skardi/src/sources/providers/open_connector/error.rs‎
Lines changed: 46 additions & 0 deletions
diff --git a/‎crates/skardi/src/sources/providers/open_connector/exec.rs‎
Lines changed: 46 additions & 6 deletions b/‎crates/skardi/src/sources/providers/open_connector/exec.rs‎
Lines changed: 46 additions & 6 deletions
@@ -330,7 +330,7 @@ For end-to-end walkthroughs — RAG, recommendations, an agent-native wiki, a si
 | S3 / GCS / Azure | Read | No | CSV, Parquet, Lance from object stores | [docs/S3_USAGE.md](docs/S3_USAGE.md) |
 | Apache Iceberg | Read | No | Schema evolution, partition pruning | [docs/iceberg/](docs/iceberg/) |
 | InfluxDB 3 | Read | No | Time-series measurements over Arrow Flight SQL | [docs/influxdb/](docs/influxdb/) |
-| Open Connector | Read | Yes | SaaS resources as stable SQL tables via a self-hosted [Open Connector](https://github.com/oomol-lab/open-connector) gateway; GitHub pack (repos, issues, PRs, reviews, commits, workflow runs, releases — [guide](docs/open-connector-github.md)), `open_connector_query` / `open_connector_scan` UDTFs, filter + limit pushdown, bounded TTL cache (more provider packs rolling out) | [docs/open-connector.md](docs/open-connector.md), [demo](docs/open-connector/) |
+| Open Connector | Read | Yes | SaaS resources as stable SQL tables via a self-hosted [Open Connector](https://github.com/oomol-lab/open-connector) gateway; GitHub pack (repos, issues, PRs, reviews, commits, workflow runs, releases — [guide](docs/open-connector-github.md)), Slack pack (conversations, users, files — [guide](docs/open-connector-slack.md)), `open_connector_query` / `open_connector_scan` UDTFs, filter + limit pushdown, bounded TTL cache (more provider packs rolling out) | [docs/open-connector.md](docs/open-connector.md), [demo](docs/open-connector/) |
 | Documents | Read | No | PDF/Office/ODF/image -> per-page markdown, tables, images (local directories; `documents` feature) | [docs/documents.md](docs/documents.md) |
 
 ---
 
@@ -483,7 +483,7 @@ impl UrlTableFactory for SkardiUrlTableFactory {
 
 /// Create a new SessionContext with custom URL table support (built-in files + Lance)
 /// and the `lance_knn` / `pg_knn` / Open Connector UDTFs registered.
-fn new_session_context() -> (SessionContext, DatasetRegistry, OpenConnectorGateways) {
+fn new_session_context() -> Result<(SessionContext, DatasetRegistry, OpenConnectorGateways)> {
     let dataset_registry: DatasetRegistry = Arc::new(RwLock::new(HashMap::new()));
     let open_connector_gateways = OpenConnectorGateways::default();
     let session_store = SessionStore::new();
@@ -522,7 +522,7 @@ fn new_session_context() -> (SessionContext, DatasetRegistry, OpenConnectorGatew
     register_vec_to_binary_udf(&mut ctx);
     // Open Connector UDTFs plan against the gateway state that
     // register_open_connector_tables fills in during ctx registration.
-    register_open_connector_udtfs(&ctx, Arc::clone(&open_connector_gateways));
+    register_open_connector_udtfs(&ctx, Arc::clone(&open_connector_gateways))?;
 
     // Embedding UDFs (gated by feature flags, lazy model loading on first call).
     #[cfg(feature = "onnx")]
@@ -551,7 +551,7 @@ fn new_session_context() -> (SessionContext, DatasetRegistry, OpenConnectorGatew
         registry.register_chunk_udf(&mut ctx);
     }
 
-    (ctx, dataset_registry, open_connector_gateways)
+    Ok((ctx, dataset_registry, open_connector_gateways))
 }
 
 /// Resolve a path string: if relative (and not remote), resolve against cwd.
@@ -1237,7 +1237,7 @@ async fn show_schema(
     table_filter: Option<&str>,
     out: &mut dyn Write,
 ) -> Result<()> {
-    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context();
+    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context()?;
     let config =
         load_and_register_all(ctx_path, &mut session_ctx, &dataset_registry, &oc_gateways).await?;
 
@@ -1384,7 +1384,7 @@ fn source_name_for<'a>(
 /// data sources first. If no context file is found, run the query in a bare session with
 /// URL table support (allowing direct file/lance paths in SQL).
 async fn run_query(ctx_override: Option<PathBuf>, sql: &str) -> Result<()> {
-    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context();
+    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context()?;
 
     // Try to load context file, but don't fail if not found when no explicit --ctx was given
     match resolve_ctx_path(ctx_override.as_deref()) {
@@ -1542,7 +1542,7 @@ async fn run_pipeline_with_params(
         .with_context(|| format!("Failed to render SQL for pipeline '{}'", pipeline_name))?;
 
     // 3. Build a SessionContext with ctx data sources registered, then execute.
-    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context();
+    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context()?;
     if let Some(p) = &ctx_path_for_load {
         load_and_register_all(p, &mut session_ctx, &dataset_registry, &oc_gateways).await?;
     }
@@ -2896,7 +2896,8 @@ spec:
 
         #[tokio::test]
         async fn errors_without_connection_string() {
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let err = register_source(
                 &mut session_ctx,
                 &dynamodb_source(None),
@@ -2914,7 +2915,8 @@ spec:
 
         #[tokio::test]
         async fn errors_without_options() {
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let source = dynamodb_source(Some("http://localhost:8000"));
             let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
@@ -2949,7 +2951,8 @@ spec:
 
         #[tokio::test]
         async fn errors_without_connection_string() {
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let err = register_source(
                 &mut session_ctx,
                 &clickhouse_source(None),
@@ -2970,7 +2973,8 @@ spec:
             // The provider is the single enforcement point for the read-only
             // invariant — the CLI must reject read_write exactly like the
             // server's UnsupportedWriteMode.
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let mut source = clickhouse_source(Some("http://127.0.0.1:1"));
             source.access_mode = Some("read_write".to_string());
             let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
@@ -3015,7 +3019,8 @@ bindings:
 
         #[tokio::test]
         async fn errors_without_connection_string() {
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let source = open_connector_source(None, Some(VALID_CONFIG));
             let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
@@ -3031,7 +3036,8 @@ bindings:
         async fn errors_with_table_hierarchy() {
             // hierarchy_level defaults to Table; the CLI must reject it with
             // a clear message, not the provider's wrapped error.
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let mut source =
                 open_connector_source(Some("http://localhost:3000"), Some(VALID_CONFIG));
             source.hierarchy_level = HierarchyLevel::Table;
@@ -3047,7 +3053,8 @@ bindings:
 
         #[tokio::test]
         async fn errors_without_typed_config() {
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let source = open_connector_source(Some("http://localhost:3000"), None);
             let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
@@ -3065,7 +3072,8 @@ bindings:
             // The provider is the single enforcement point for the
             // read-only invariant — the CLI must reject read_write exactly
             // like the server's UnsupportedWriteMode.
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let mut source =
                 open_connector_source(Some("http://localhost:3000"), Some(VALID_CONFIG));
             source.access_mode = Some("read_write".to_string());
@@ -3078,7 +3086,8 @@ bindings:
 
         #[tokio::test]
         async fn errors_when_typed_config_on_wrong_type() {
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let mut source =
                 open_connector_source(Some("http://localhost:3000"), Some(VALID_CONFIG));
             source.source_type = "csv".to_string();
@@ -3096,7 +3105,8 @@ bindings:
         async fn errors_when_token_env_missing() {
             // With the config valid, the next failure is the unset runtime
             // token — before any network call to the (unroutable) gateway.
-            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) =
+                new_session_context().expect("session context");
             let config =
                 VALID_CONFIG.replace("OPEN_CONNECTOR_TOKEN", "SKARDI_CLI_TEST_OC_TOKEN_UNSET");
             let source = open_connector_source(Some("http://127.0.0.1:1"), Some(config.as_str()));
 
@@ -120,7 +120,7 @@ impl OptimizerRegistry {
         // Register Open Connector table functions
         if source_types.contains(&DataSourceType::OpenConnector) {
             tracing::info!("Registering Open Connector table functions");
-            register_open_connector_udtfs(ctx, self.open_connector_gateways());
+            register_open_connector_udtfs(ctx, self.open_connector_gateways())?;
             tracing::info!("✓ Registered open_connector_query and open_connector_scan");
         }
 
 
@@ -170,7 +170,7 @@ impl ActionRegistry {
 /// The schema is canonicalized first (object keys sorted recursively, so two
 /// semantically identical schemas with different key orders fingerprint
 /// equally), then hashed with BLAKE3 and hex-encoded.
-fn fingerprint_schema(output_schema: Option<&Value>) -> String {
+pub(crate) fn fingerprint_schema(output_schema: Option<&Value>) -> String {
     let canonical = match output_schema {
         Some(schema) => canonical_json(schema),
         None => "null".to_string(),
 
@@ -204,6 +204,46 @@ pub enum OpenConnectorError {
     )]
     NonIdempotentAmbiguousFailure { operation: String, reason: String },
 
+    /// The provider reported an in-band error inside an otherwise
+    /// successful response envelope — Slack's HTTP-200 `ok: false` +
+    /// `error` pattern. The code is a short provider-authored identifier
+    /// (`missing_scope`, `not_authed`), bounded before display.
+    #[error(
+        "Open Connector action '{action_id}' page {page}: the provider reported \
+         error '{code}'"
+    )]
+    ProviderReportedError {
+        action_id: String,
+        page: usize,
+        code: String,
+    },
+
+    /// A declared total-pages location resolved to a non-numeric value, so
+    /// the scan cannot know when the collection ends.
+    #[error(
+        "Open Connector pagination total at '{path}' on page {page} is {found}, \
+         expected a non-negative integer"
+    )]
+    PaginationTotalInvalid {
+        path: String,
+        page: usize,
+        found: String,
+    },
+
+    /// A continuation cursor was present at the declared path but was not a
+    /// string. Treating it as end-of-collection would silently truncate the
+    /// scan, so it fails instead. Carries the JSON *kind* only, never the
+    /// value.
+    #[error(
+        "Open Connector pagination cursor at '{path}' on page {page} is {found}, \
+         not a string; refusing to treat it as end-of-collection"
+    )]
+    PaginationCursorInvalid {
+        path: String,
+        page: usize,
+        found: String,
+    },
+
     /// Pagination failed to advance: the gateway returned an already-seen
     /// cursor, which would loop the scan forever.
     #[error(
@@ -354,6 +394,12 @@ pub enum OpenConnectorError {
         found: String,
     },
 
+    /// An embedded source-pack asset failed to parse or validate. A build
+    /// defect (assets ship inside the binary), surfaced as a registration /
+    /// UDTF-setup diagnostic instead of a panic.
+    #[error("embedded source pack asset '{asset}' is invalid: {reason}")]
+    SourcePackAssetInvalid { asset: String, reason: String },
+
     /// A response body grew past the configured decoding bound.
     #[error("Open Connector {operation} response exceeded the {limit_bytes}-byte bound")]
     ResponseTooLarge {
 
@@ -47,6 +47,9 @@ pub struct ScanTarget {
     pub action_id: Arc<str>,
     /// Pagination contract.
     pub pagination: PaginationStrategy,
+    /// In-band provider-error location (see `SourcePackTable::error_path`);
+    /// `None` for raw scans and packs whose providers error at HTTP level.
+    pub error_path: Option<&'static str>,
     /// Fixed action inputs sent with every request (see
     /// [`SourcePackTable::fixed_inputs`]); empty for raw scans, whose whole
     /// input is caller-supplied.
@@ -63,6 +66,7 @@ impl ScanTarget {
             table_id: Arc::from(table.id),
             action_id: Arc::from(table.action_id),
             pagination: table.pagination,
+            error_path: table.error_path,
             fixed_inputs: table.fixed_inputs,
             source_pack_version,
         }
@@ -274,6 +278,9 @@ struct ScanState {
     scan_timeout: Duration,
     deadline: Instant,
     pagination: Pagination,
+    /// Pre-parsed in-band provider-error path, checked before each page's
+    /// row extraction.
+    error_path: Option<RowPath>,
     rows_emitted: u64,
     /// Cached batches to replay (non-empty only on a cache hit).
     replay: VecDeque<RecordBatch>,
@@ -358,6 +365,7 @@ impl ScanState {
             scan_timeout: exec.scan_timeout,
             deadline: Instant::now() + exec.scan_timeout,
             pagination: Pagination::new(exec.target.pagination)?,
+            error_path: exec.target.error_path.map(RowPath::parse).transpose()?,
             rows_emitted: 0,
             replay,
             fetched: Vec::new(),
@@ -474,6 +482,30 @@ impl ScanState {
         if Instant::now() >= self.deadline {
             return Err(self.timeout_error());
         }
+        // Some gateways forward a provider's in-band application errors
+        // unchanged (Slack-style HTTP 200, `ok: false` + `error`). Packs
+        // targeting such a gateway declare `error_path` so the provider's
+        // own code surfaces instead of the misleading row-path error the
+        // missing row array would raise. (Open Connector's own executors
+        // consume Slack's `ok:false` and return a failure envelope, so its
+        // slack pack declares none — the mock pack models this mechanism.)
+        if let Some(error_path) = &self.error_path
+            && let Ok(code) = error_path.extract(&envelope, page)
+            && !code.is_null()
+        {
+            let code = match code.as_str() {
+                Some(text) => text.chars().take(128).collect(),
+                None => format!(
+                    "<{}>",
+                    crate::sources::providers::open_connector::row_path::json_kind(code)
+                ),
+            };
+            return Err(OpenConnectorError::ProviderReportedError {
+                action_id: self.target.action_id.to_string(),
+                page,
+                code,
+            });
+        }
         let rows = self.row_path.rows(&envelope, page)?;
         let batch = self.converter.convert(rows, page)?;
         // Conversion is synchronous, so it cannot be preempted by Tokio; do
@@ -529,10 +561,18 @@ impl ScanState {
             self.store_cache();
         }
 
-        let more = self.pagination.advance(&envelope, rows.len())?;
-        if !more {
-            self.done = true;
-            self.store_cache();
+        // Pagination advances only while the scan is still going. After a
+        // LIMIT-satisfied page there is no next request to prepare — and
+        // advance() also parses and validates continuation state, so a
+        // repeated cursor or a missing/malformed page total on that final
+        // page would fail a scan whose result is already complete for its
+        // key.
+        if !self.done {
+            let more = self.pagination.advance(&envelope, rows.len())?;
+            if !more {
+                self.done = true;
+                self.store_cache();
+            }
         }
 
         // A terminal empty page is completion, not output.
@@ -557,7 +597,7 @@ impl ScanState {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::sources::providers::open_connector::packs::mock::MOCK_PACK;
+    use crate::sources::providers::open_connector::packs::mock;
     use crate::sources::providers::open_connector::testutil::{
         CapturedEvent, MockGateway, MockResponse, RecordedRequest, capture_events, envelope_ok,
     };
@@ -571,7 +611,7 @@ mod tests {
         limit: Option<usize>,
         source_pack_version: u32,
     ) -> OpenConnectorExec {
-        let table = &MOCK_PACK.tables[0];
+        let table = &mock::pack().expect("embedded asset parses").tables[0];
         OpenConnectorExec::new(
             client,
             cache,
Original file line number	Diff line number	Diff line change
`@@ -120,7 +120,7 @@ impl OptimizerRegistry {`
`120`	`120`	`// Register Open Connector table functions`
`121`	`121`	`if source_types.contains(&DataSourceType::OpenConnector) {`
`122`	`122`	`tracing::info!("Registering Open Connector table functions");`
`123`		`- register_open_connector_udtfs(ctx, self.open_connector_gateways());`
	`123`	`+ register_open_connector_udtfs(ctx, self.open_connector_gateways())?;`
`124`	`124`	`tracing::info!("✓ Registered open_connector_query and open_connector_scan");`
`125`	`125`	`}`
`126`	`126`