fix(sources): paginate github issues on the gateway raw page length (#176)

fix(sources): paginate github issues on the gateway's raw page length

The OC list_repository_issues action filters pull requests out AFTER
paginating, so the filtered issues array's length is not a termination
signal: short-page termination silently truncated the scan on the first
PR-bearing page, and even empty-page termination fails on 100
consecutive PRs. Upstream now reports the raw page length
(pageInfo.fetched, oomol-lab/open-connector#228); this change consumes
it.

Engine: PageNumber gains raw_page_size_path, mirroring
total_pages_path — when declared, the scan continues while the RAW page
was full regardless of how short (or empty) the filtered rows are; a
missing signal propagates as RowPathNotFound and a non-integer one
fails as the new PaginationRawPageSizeInvalid (kind-only), never a
silent truncation. The two authoritative signals are mutually exclusive
at validate time. Loader exposes raw_page_size_path on page_number
pagination; the issues table declares $.pageInfo.fetched.

Contract re-captured from a live gateway carrying the upstream fix
(outputSchema now declares pageInfo) and the fingerprint re-pinned, so
older gateways fail issues registration at the fingerprint gate instead
of truncating. Live-verified: all 11 tables register with the new pin
and the issues scan reaches the credential wall.

Tests: engine units (continue on full raw page with short/empty rows,
terminate on short raw page, missing/invalid signal failures, total/raw
mutual exclusion) and a pack e2e driving a 3-page scan whose middle
page is ALL pull requests — the case that defeats every filtered-count
heuristic. Stubs and the demo stub gateway now emit pageInfo. Docs and
spec updated with the minimum-gateway note. 246 open_connector / 803
lib tests green.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: bakey

crates/skardi/src/sources/providers/open_connector/error.rs

@@ -230,6 +230,19 @@ pub enum OpenConnectorError {
         found: String,
     },
 
+    /// The declared raw-page-size signal was present but not a non-negative
+    /// integer. Treating it as anything else would either truncate the scan
+    /// or loop it; carries the JSON *kind* only, never the value.
+    #[error(
+        "Open Connector pagination raw page size at '{path}' on page {page} is {found}, \
+         expected a non-negative integer"
+    )]
+    PaginationRawPageSizeInvalid {
+        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

crates/skardi/src/sources/providers/open_connector/packs/fixtures/github/contracts/list_repository_issues.json

@@ -123,7 +123,22 @@
         },
         "additionalProperties": true
       }
+    },
+    "pageInfo": {
+      "type": "object",
+      "properties": {
+        "fetched": {
+          "type": "integer",
+          "minimum": 0,
+          "description": "Number of items GitHub returned on this page before filtering. Continue paginating while this equals perPage, which defaults to 30."
+        }
+      },
+      "additionalProperties": false,
+      "required": [
+        "fetched"
+      ],
+      "description": "Pagination signals from the raw GitHub page, before pull requests are filtered out."
     }
   },
   "additionalProperties": false
-}
+}

crates/skardi/src/sources/providers/open_connector/packs/github.rs

@@ -7,8 +7,15 @@
 //! - **Page-number pagination everywhere** (`page`/`perPage`, 100 per page
 //!   — GitHub's maximum; the camelCase keys are Open Connector's strict
 //!   action-input contract, not GitHub's raw REST parameters). A short or
-//!   empty page terminates the scan, which is GitHub's documented
-//!   end-of-collection signal.
+//!   empty page terminates the scan — except for `issues`, whose OC action
+//!   filters pull requests out AFTER paginating, so a filtered page's
+//!   length is not a termination signal: the table declares
+//!   `raw_page_size_path: $.pageInfo.fetched` (upstream #228) and the scan
+//!   continues while the RAW page was full, even when the filtered rows
+//!   come back short or empty. Requires a gateway with
+//!   oomol-lab/open-connector#228 (older gateways fail the fingerprint
+//!   gate at registration, and their responses would fail the scan loudly
+//!   with a missing `$.pageInfo.fetched` — never a silent truncation).
 //! - **Filters are allowlisted only where faithful — and every string-enum
 //!   push is Inexact.** `issues.state` / `pull_requests.state` narrow the
 //!   fetch but DataFusion re-applies them: the translation is faithful only
@@ -115,6 +122,49 @@ mod tests {
         MockResponse::ok(&discovery_ok("{}", output_schema, true, None))
     }
 
+    #[tokio::test]
+    async fn filtered_issue_pages_do_not_truncate_the_scan() {
+        // The OC action filters pull requests out AFTER paginating, so a
+        // filtered page can be short — or entirely empty — while more pages
+        // exist. The raw signal (pageInfo.fetched, upstream #228) must
+        // drive continuation: page 1 returns 2 issues of a full raw page,
+        // page 2 is ALL pull requests (0 issues, raw full), page 3 is the
+        // genuine final page. Short-page termination would stop after page
+        // 1 and lose everything after it.
+        let gateway = MockGateway::start(|req| {
+            if req.method == "GET" && req.path == "/v1/health" {
+                return MockResponse::ok("{}");
+            }
+            if req.method == "GET" && req.path.starts_with("/v1/actions/") {
+                return github_discovery(&req.path);
+            }
+            if req.method == "POST" && req.path == "/v1/actions/github.list_repository_issues" {
+                let body: Value = serde_json::from_str(&req.body).unwrap_or_default();
+                let page = body["input"]["page"].as_u64().unwrap_or(1);
+                let response = match page {
+                    1 => json!({"issues": [issue(1, "open", "2026-01-01T00:00:00Z"),
+                                             issue(2, "open", "2026-01-01T00:00:00Z")],
+                                 "pageInfo": {"fetched": 100}}),
+                    2 => json!({"issues": [], "pageInfo": {"fetched": 100}}),
+                    3 => json!({"issues": [issue(3, "open", "2026-01-02T00:00:00Z")],
+                                 "pageInfo": {"fetched": 1}}),
+                    other => panic!("unexpected page {other}"),
+                };
+                return MockResponse::ok(&envelope_ok(&response.to_string()));
+            }
+            MockResponse::new(404, "{}")
+        })
+        .await;
+        let (_gw, ctx) = setup_with_gateway(gateway, "SKARDI_TEST_OC_GITHUB_RAW_PAGE").await;
+
+        let batches = collect(&ctx, "SELECT number FROM saas.gh.issues ORDER BY number").await;
+        assert_eq!(
+            rows_of(&batches),
+            3,
+            "all three pages were scanned: the short and the all-PR page did not terminate"
+        );
+    }
+
     #[test]
     fn fingerprint_coverage_gap_is_pinned() {
         // The gate protects only what upstream DECLARES; these mapped
@@ -623,16 +673,24 @@ bindings:
                 .and_then(Value::as_str)
                 .unwrap_or("open")
                 .to_string();
-            let slice: Vec<_> = rows
+            let matching: Vec<_> = rows
                 .iter()
                 .filter(|row| {
                     state == "all" || row.get("state").and_then(Value::as_str) == Some(&state)
                 })
+                .collect();
+            let slice: Vec<_> = matching
+                .iter()
                 .skip((page - 1) * per_page)
                 .take(per_page)
-                .cloned()
+                .map(|row| (*row).clone())
                 .collect();
-            return MockResponse::ok(&envelope_ok(&json!({"issues": slice}).to_string()));
+            // The raw page length before any post-pagination filtering; the
+            // stub does none, so it equals the slice length.
+            let fetched = slice.len();
+            return MockResponse::ok(&envelope_ok(
+                &json!({"issues": slice, "pageInfo": {"fetched": fetched}}).to_string(),
+            ));
         }
         MockResponse::new(404, "{}")
     }
@@ -658,6 +716,14 @@ bindings:
             let served = std::sync::Arc::clone(&served);
             MockGateway::start(move |req| issues_handler(req, &served)).await
         };
+        setup_with_gateway(gateway, token_env).await
+    }
+
+    /// Register the issues binding against an arbitrary gateway stub.
+    async fn setup_with_gateway(
+        gateway: MockGateway,
+        token_env: &str,
+    ) -> (MockGateway, SessionContext) {
         unsafe {
             std::env::set_var(token_env, "test-token");
         }
@@ -1356,7 +1422,9 @@ bindings:
                 return MockResponse::ok(&envelope_ok(r#"{"repositories": []}"#));
             }
             if req.method == "POST" && req.path == "/v1/actions/github.list_repository_issues" {
-                return MockResponse::ok(&envelope_ok(r#"{"issues": []}"#));
+                return MockResponse::ok(&envelope_ok(
+                    r#"{"issues": [], "pageInfo": {"fetched": 0}}"#,
+                ));
             }
             MockResponse::new(404, "{}")
         })

crates/skardi/src/sources/providers/open_connector/packs/github.yaml

@@ -35,13 +35,17 @@ tables:
 
   issues:
     action: github.list_repository_issues
-    fingerprint: 45fbef8622d90702e536b9b2e0ade6bcc7da61f7cb654d62a90b2e36df13c64b
+    fingerprint: 0530ff93d700f3aba2754160a76cf69919a7753fe14b26835463fdb88cfad8e3
     row_path: "$.issues"
     pagination:
       strategy: page_number
       page_input: page
       page_size_input: perPage
       page_size: 100
+      # The OC action filters pull requests out AFTER paginating; the raw
+      # page length (oomol-lab/open-connector#228) is the only sound
+      # termination signal — a short or empty issues array is not.
+      raw_page_size_path: "$.pageInfo.fetched"
     resources:
       required: [owner, repo]
     fixed_inputs:

crates/skardi/src/sources/providers/open_connector/packs/loader.rs

@@ -400,6 +400,8 @@ enum PaginationDoc {
         page_size: u32,
         #[serde(default)]
         total_pages_path: Option<String>,
+        #[serde(default)]
+        raw_page_size_path: Option<String>,
     },
     Cursor {
         cursor_input: String,
@@ -418,11 +420,13 @@ impl PaginationDoc {
                 page_size_input,
                 page_size,
                 total_pages_path,
+                raw_page_size_path,
             } => PaginationStrategy::PageNumber {
                 page_param: leak_str(page_input),
                 per_page_param: leak_str(page_size_input),
                 per_page: page_size,
                 total_pages_path: total_pages_path.map(leak_str),
+                raw_page_size_path: raw_page_size_path.map(leak_str),
             },
             Self::Cursor {
                 cursor_input,