feat(ingest): add cancel subcommand and handle cancelled status (#223)

POST /jobs/{id}/cancel marks an import cancelled without requiring a
rerun. The server uses a non-terminal 'cancelled' status to keep the
rerun in-flight guard active while the dltHub drain finishes.

- Add IngestCancelAck struct and cancel() client method
- Add `hotdata ingest cancel <id>` subcommand
- normalize_status: add 'cancelled' arm (was falling through to running)
- status_exit_code / poll_ingest: treat cancelled as terminal (exit 1)
  so `status --wait` terminates cleanly instead of spinning forever

Co-authored-by: Eddie A Tejeda <669988+eddietejeda@users.noreply.github.com>

Author: eddietejeda

src/client/ingest.rs

@@ -285,6 +285,17 @@ impl IngestClient {
         )
     }
 
+    /// Cancel a running or pending ingest. The dltHub drain may still be
+    /// running (we can't interrupt it remotely); the server marks the status
+    /// as `cancelled` — a non-terminal state that keeps the /rerun in-flight
+    /// guard active. A 409 from /rerun means the drain hasn't settled yet.
+    pub fn cancel(&self, ingest_id: &str) -> Result<IngestCancelAck, IngestError> {
+        self.send(
+            self.authed(reqwest::Method::POST, &format!("/jobs/{ingest_id}/cancel")),
+            None,
+        )
+    }
+
     // --- read endpoints --------------------------------------------------
 
     /// The connector catalog. REST entries carry a ready-to-edit `template`
@@ -460,6 +471,19 @@ pub struct JobStatus {
     pub updated_at: Option<String>,
 }
 
+/// `POST /jobs/{id}/cancel` body.
+#[derive(Debug, Deserialize, Serialize)]
+pub struct IngestCancelAck {
+    pub ingest_id: String,
+    pub status: String,
+    #[serde(default)]
+    pub detail: Option<String>,
+    #[serde(default)]
+    pub name: Option<String>,
+    #[serde(default)]
+    pub database_id: Option<String>,
+}
+
 /// `DELETE /sources/{id}` body.
 #[derive(Debug, Deserialize, Serialize)]
 pub struct DeleteSourceAck {

src/commands/ingest.rs

@@ -190,6 +190,17 @@ pub enum IngestCommands {
         #[command(flatten)]
         poll: PollArgs,
     },
+
+    /// Cancel a running or pending import
+    ///
+    /// Marks the import cancelled immediately. The dltHub drain may still be
+    /// running and will finish on its own — use `status <id> --wait` to see
+    /// when it settles. To retry after cancellation, use `trigger-import <id>`
+    /// once the status reaches `failed`.
+    Cancel {
+        /// Import or datasource id to cancel
+        id: String,
+    },
 }
 
 /// Wait flags shared by the non-blocking commands (`new-import`,
@@ -246,6 +257,7 @@ pub fn dispatch(workspace_id: &str, output: &str, command: IngestCommands) {
             trigger_import(workspace_id, output, &id, &poll)
         }
         IngestCommands::Status { id, poll } => status(workspace_id, output, &id, &poll),
+        IngestCommands::Cancel { id } => cancel_import(workspace_id, output, &id),
     }
 }
 
@@ -261,6 +273,7 @@ fn normalize_status(raw: &str) -> (&'static str, Option<&str>) {
         "running" => ("running", None),
         "done" => ("done", None),
         "failed" => ("failed", None),
+        "cancelled" => ("cancelled", None),
         stage => ("running", Some(stage)),
     }
 }
@@ -986,7 +999,7 @@ fn trigger_import(workspace_id: &str, output: &str, id: &str, poll: &PollArgs) {
 fn status_exit_code(status: &str) -> i32 {
     match normalize_status(status).0 {
         "done" => 0,
-        "failed" => 1,
+        "failed" | "cancelled" => 1,
         _ => 2,
     }
 }
@@ -1030,6 +1043,36 @@ fn status(workspace_id: &str, output: &str, id: &str, poll: &PollArgs) {
     std::process::exit(status_exit_code(&st.status));
 }
 
+// --- cancel ----------------------------------------------------------------
+
+fn cancel_import(workspace_id: &str, output: &str, id: &str) {
+    let client = IngestClient::new(workspace_id);
+    let ack = with_spinner("cancelling…", || client.cancel(id));
+    let machine = serde_json::json!({
+        "ingest_id": ack.ingest_id,
+        "status": ack.status,
+        "detail": ack.detail,
+        "database_id": ack.database_id,
+    });
+    render(output, &machine, || {
+        use crossterm::style::Stylize;
+        let label = |l: &str| format!("{:<14}", l).dark_grey().to_string();
+        println!("{}{}", label("status:"), util::color_status(&ack.status));
+        if let Some(db) = ack.database_id.as_deref() {
+            println!("{}{}", label("database:"), db);
+        }
+        println!(
+            "{}",
+            format!(
+                "Drain may still be running — track with: hotdata ingest status {} --wait",
+                ack.ingest_id
+            )
+            .dark_grey()
+        );
+    });
+    std::process::exit(1);
+}
+
 // --- list-datasources ------------------------------------------------------
 
 fn list_datasources(workspace_id: &str, output: &str, all: bool) {
@@ -1280,14 +1323,15 @@ fn poll_ingest(
                 spinner.finish_and_clear();
                 return st;
             }
-            "failed" => {
+            "failed" | "cancelled" => {
                 spinner.finish_and_clear();
                 // Machine formats get the projected status object (with the
                 // server's detail) on stdout, matching one-shot `status`.
                 render(output, &status_json(&st), || {
                     use crossterm::style::Stylize;
+                    let label = st.status.as_str();
                     let detail = st.detail.as_deref().unwrap_or("unknown error");
-                    eprintln!("{}", format!("failed: {detail}").red());
+                    eprintln!("{}", format!("{label}: {detail}").red());
                     if detail.contains("Forbidden") {
                         eprintln!(
                             "{}",