From 5e8cf6be18e4bfc2dafd822eb05149b0cf10c86b Mon Sep 17 00:00:00 2001
From: Eddie A Tejeda <669988+eddietejeda@users.noreply.github.com>
Date: Tue, 4 Aug 2026 12:05:09 -0700
Subject: [PATCH 1/2] feat(ingest): --continuous flag on new-datasource
 (buckets)

Marks a bucket datasource for native continuous sync: the worker's scheduler
re-runs it incrementally, appending only newly-arrived objects. Adds
IngestRequest.continuous (serialized only when set) + the --continuous flag,
wired into the flag-driven filesystem builder and the wizard-skip check.
Server ignores it for non-filesystem families.

Test: --continuous rides through to the request body; default off is omitted
from the JSON.
---
 skills/hotdata/SKILL.md |  2 ++
 src/client/ingest.rs    |  4 ++++
 src/commands/ingest.rs  | 29 +++++++++++++++++++++++++++++
 3 files changed, 35 insertions(+)

diff --git a/skills/hotdata/SKILL.md b/skills/hotdata/SKILL.md
index 515998c..ef6d5ab 100644
--- a/skills/hotdata/SKILL.md
+++ b/skills/hotdata/SKILL.md
@@ -256,6 +256,8 @@ hotdata ingest new-datasource --service buckets --bucket-url s3://bucket/prefix
 # Files in S3/GCS/Azure buckets (csv, jsonl, parquet); --glob narrows the match.
 # Public buckets need no credentials; private ones take --config @creds.json
 # ({"aws_access_key_id": …, "aws_secret_access_key": …, "endpoint_url": …}).
+# --continuous keeps a bucket datasource synced: it's re-run incrementally on a
+# schedule, appending only newly-arrived objects (no re-read of the whole bucket).
 
 hotdata ingest new-datasource --service iceberg --config @catalog.json --table ns.orders
 # Iceberg via a REST catalog. --table is REQUIRED (repeatable, namespace.table).
diff --git a/src/client/ingest.rs b/src/client/ingest.rs
index 397da4f..c11e7b7 100644
--- a/src/client/ingest.rs
+++ b/src/client/ingest.rs
@@ -428,6 +428,10 @@ pub struct IngestRequest {
     pub tables: Vec<String>,
     #[serde(skip_serializing_if = "std::ops::Not::not")]
     pub validate_only: bool,
+    /// filesystem only: keep this datasource continuously synced — the scheduler
+    /// re-runs it incrementally (append only new objects). Ignored otherwise.
+    #[serde(skip_serializing_if = "std::ops::Not::not")]
+    pub continuous: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub database_id: Option<String>,
 }
diff --git a/src/commands/ingest.rs b/src/commands/ingest.rs
index 0a74de1..075a027 100644
--- a/src/commands/ingest.rs
+++ b/src/commands/ingest.rs
@@ -693,6 +693,11 @@ pub struct CreateArgs {
     #[arg(long)]
     glob: Option<String>,
 
+    /// Keep this datasource continuously synced — refreshed incrementally on a
+    /// schedule, appending only newly-arrived objects (buckets only)
+    #[arg(long)]
+    continuous: bool,
+
     /// Catalog type, e.g. rest (iceberg)
     #[arg(long = "catalog-type")]
     catalog_type: Option<String>,
@@ -713,6 +718,7 @@ impl CreateArgs {
             || self.bucket_url.is_some()
             || self.format.is_some()
             || self.glob.is_some()
+            || self.continuous
             || self.catalog_type.is_some()
             || self.database_id.is_some()
     }
@@ -779,6 +785,7 @@ fn build_create_request(
             bucket_url: Some(args.bucket_url.ok_or("buckets connectors need --bucket-url")?),
             file_glob: args.glob,
             file_format: args.format,
+            continuous: args.continuous,
             ..Default::default()
         },
         "iceberg" => IngestRequest {
@@ -1837,6 +1844,7 @@ mod tests {
             bucket_url: None,
             format: None,
             glob: None,
+            continuous: false,
             catalog_type: None,
             database_id: None,
         }
@@ -1869,6 +1877,27 @@ mod tests {
         assert_eq!(req.database_id.as_deref(), Some("db_1"));
     }
 
+    #[test]
+    fn create_request_filesystem_carries_continuous_flag() {
+        let e = entry("buckets", "filesystem");
+        let mut args = create_args();
+        args.bucket_url = Some("s3://b/prefix".into());
+        args.format = Some("jsonl".into());
+        args.continuous = true;
+        let req = build_create_request(&e, args, None).unwrap();
+        assert_eq!(req.family, "filesystem");
+        assert_eq!(req.bucket_url.as_deref(), Some("s3://b/prefix"));
+        assert!(req.continuous); // --continuous rides through to the request body
+
+        // Default is off, and it serializes only when true (skip_serializing_if).
+        let mut off = create_args();
+        off.bucket_url = Some("s3://b".into());
+        off.format = Some("jsonl".into());
+        let req_off = build_create_request(&e, off, None).unwrap();
+        assert!(!req_off.continuous);
+        assert!(!serde_json::to_string(&req_off).unwrap().contains("continuous"));
+    }
+
     #[test]
     fn create_request_rejects_invalid_names() {
         let e = entry("postgres", "sql");

From c38859cc0b4d39e681bfdcaa7124e7d3edff456c Mon Sep 17 00:00:00 2001
From: Eddie A Tejeda <669988+eddietejeda@users.noreply.github.com>
Date: Tue, 4 Aug 2026 13:16:02 -0700
Subject: [PATCH 2/2] style: cargo fmt build_create_request continuous test
 assertion

---
 src/commands/ingest.rs | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/src/commands/ingest.rs b/src/commands/ingest.rs
index 075a027..2b9069e 100644
--- a/src/commands/ingest.rs
+++ b/src/commands/ingest.rs
@@ -1895,7 +1895,11 @@ mod tests {
         off.format = Some("jsonl".into());
         let req_off = build_create_request(&e, off, None).unwrap();
         assert!(!req_off.continuous);
-        assert!(!serde_json::to_string(&req_off).unwrap().contains("continuous"));
+        assert!(
+            !serde_json::to_string(&req_off)
+                .unwrap()
+                .contains("continuous")
+        );
     }
 
     #[test]