Fix the 10 defects surfaced by the end-to-end claims audit

.github/workflows/ci.yml

@@ -96,3 +96,14 @@ jobs:
           cd conduit-ui
           npm install
           npm test
+
+  gitleaks:
+    name: Secret Scan
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - uses: gitleaks/gitleaks-action@v2
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.gitleaks.toml

@@ -0,0 +1,9 @@
+[extend]
+useDefault = true
+
+[[allowlists]]
+description = "Documentation placeholders and test fixtures, not real credentials"
+regexes = [
+  '''your-api-key''',
+  '''cdt_abc123def456abc123def456abc123''',
+]

README.md

@@ -2,7 +2,7 @@
 
 **A Rust-native data pipeline orchestrator.**
 
-Conduit is not "Airflow but faster." It solves problems that Airflow architecturally *cannot* solve — virtual pipeline environments, time-travel debugging, compile-time DAG validation, and plan/apply deployments — all in a single binary with zero external dependencies.
+Conduit is not "Airflow but faster." It solves problems that Airflow architecturally *cannot* solve — virtual pipeline environments, time-travel debugging, compile-time DAG validation, and plan/apply deployments — all in a single binary with no external services (no database, no message broker; task execution shells out to your `python3`/`bash`).
 
 ![Conduit Web UI](conduit-demo-final.gif)
 
@@ -81,8 +81,9 @@ conduit/
   conduit-executor/     Process-based task runtime with timeout, retry, and sensor polling
   conduit-planner/      Fingerprint diffing, impact analysis, and plan/apply deployment workflow
   conduit-lineage/      Column-level SQL lineage via sqlparser-rs AST + TableCatalog
-  conduit-providers/    Data source adapters: Postgres, MySQL, SQLite, CockroachDB, Redshift,
-                        TimescaleDB, HTTP/REST, and 26 additional provider stubs
+  conduit-providers/    Data source adapters: 12 implemented (Postgres, MySQL, SQLite,
+                        CockroachDB, Redshift, TimescaleDB, BigQuery, Snowflake, S3, GCS,
+                        DuckDB, HTTP/REST) plus 20 provider stubs
   conduit-api/          REST + WebSocket API with live run dispatch
   conduit-distributed/  Distributed executor for multi-node task dispatch
   conduit-ui/           React web UI (DAG visualization, run monitoring, log streaming)
@@ -99,18 +100,19 @@ conduit/
 Tree-sitter parses Python `@dag`/`@task` definitions **without executing Python**. Extracts schedules, tags, retry policies, pools, timeouts, and data-flow dependencies from call chains. Kahn's algorithm detects cycles, duplicates, and unknown references at compile time. Includes Criterion benchmarks for 10–1,000 DAG workloads.
 
 ### Event-Driven Scheduler (`conduit-scheduler`)
-Fully async tokio-channel scheduler (no database polling). Manages DAG run state machines (Pending → Queued → Running → Success/Failed/Skipped/Retrying). Evaluates trigger rules (AllSuccess, AllDone, OneSuccess, OneFailed, NoDeps), enforces named resource pools, and parses 5-field cron expressions.
+Fully async tokio-channel scheduler (no database polling). Manages task state machines (Pending → Queued → Success/Failed/Skipped/Retrying). Evaluates trigger rules (AllSuccess, AllDone, OneSuccess, OneFailed, NoDeps), enforces named resource pools declared in `conduit.yaml` (slot-limited, shared across runs), retries failed tasks with fixed or exponential backoff, and fires 5-field cron schedules (`conduit serve` ticks the scheduler every minute).
 
 ### Task Executor (`conduit-executor`)
-Process-isolated task execution with stdin/stdout protocol. Supports Python, Bash, SQL, Sensor, and generic Executable task types. Enforces timeouts via `tokio::time::timeout`, implements fixed and exponential backoff retry policies, and parses structured protocol messages (XCOM, LOG, PROGRESS, METRIC) from task output. Sensor tasks poll at configurable `poke_interval` until success or timeout. Tasks are dispatched concurrently via `tokio::spawn`.
+Process-isolated task execution with a stdout protocol (`CONDUIT::` lines) and env/file-based XCom injection. Supports Python, Bash, SQL, Sensor, and generic Executable task types. Enforces timeouts via `tokio::time::timeout` and parses structured protocol messages (XCOM, LOG, PROGRESS, METRIC) from task output. Retries are owned by the scheduler: fixed delay by default, exponential when a task sets `retry_backoff` (e.g. `@task(retries=3, retry_delay="30s", retry_backoff=2.0)`), and the scheduler re-dispatches the task itself when the delay elapses. Sensor tasks poll at configurable `poke_interval` until success or timeout. Tasks are dispatched concurrently via `tokio::spawn`.
 
 ### Data Providers (`conduit-providers`)
-7 fully implemented providers with real database/HTTP connections:
-- **SQL**: PostgreSQL, MySQL, SQLite, CockroachDB, TimescaleDB, Redshift (via sqlx connection pools)
+12 implemented providers with real database/API connections:
+- **SQL**: PostgreSQL, MySQL, SQLite, CockroachDB, TimescaleDB, Redshift (via sqlx connection pools), DuckDB (embedded)
+- **Cloud**: BigQuery and Snowflake (REST APIs), S3 and GCS (object storage)
 - **HTTP**: REST API provider (via reqwest)
-- 26 additional providers stubbed with the trait interface ready for implementation
+- 20 additional providers stubbed with the trait interface ready for implementation (they report `NotImplemented` rather than pretending to work)
 
-All SQL providers use lazy connection pooling, parameterized queries, and percent-encoded credentials.
+All sqlx providers use lazy connection pooling and percent-encoded credentials. Task SQL supports named `:param` placeholders, rewritten to native placeholders and bound as real query parameters (never string-spliced).
 
 ### SQL Lineage (`conduit-lineage`)
 Column-level lineage via `sqlparser-rs` AST walking (not regex). Handles SELECT, JOINs, CTEs, UNIONs, subqueries, window functions, INSERT...SELECT, and CREATE TABLE AS SELECT. Optional `TableCatalog` integration enables bare column resolution, `SELECT *` expansion, CTE column propagation, and view column registration. OpenLineage RunEvent generation emits output `columnLineage` facets. Lineage is currently labeled **beta** — known limitations include semantic dbt/Jinja resolution and dialect-specific constructs such as BigQuery `SELECT * EXCEPT`.
@@ -184,7 +186,7 @@ cargo bench -p conduit-compiler
 3. **Virtual environments** (not physical copies) — snapshot pointers, not data duplication
 4. **Event-driven scheduling** (not polling) — react to state changes via tokio channels
 5. **Content-addressable snapshots** — fingerprint-based reuse skips unchanged tasks automatically
-6. **Process isolation** — tasks run as child processes with cgroup resource limits
+6. **Process isolation** — tasks run as child processes with enforced timeouts (declared CPU/memory limits are carried in the task model but not yet enforced)
 
 **What makes this architecturally different from Airflow/Dagster/Prefect**:
 - They poll a database to find ready tasks; Conduit reacts to events in microseconds.

conduit-api/src/handlers/runs.rs

@@ -24,6 +24,9 @@ pub struct ListRunsQuery {
     pub status: Option<String>,
     /// Filter to runs that targeted this environment (e.g. "production", "staging").
     pub environment: Option<String>,
+    /// Filter to runs of one DAG (accepts `dag_id` or `dagId`).
+    #[serde(alias = "dagId")]
+    pub dag_id: Option<String>,
 }
 
 /// Request body for triggering a DAG run.
@@ -96,6 +99,7 @@ pub async fn trigger_run(
         started_at: now,
         finished_at: None,
         task_states: task_states.clone(),
+        task_logs: HashMap::new(),
         triggered_by: "api".to_string(),
         environment: environment.clone(),
     };
@@ -181,6 +185,7 @@ pub async fn get_run(
         "startedAt": run.started_at.to_rfc3339(),
         "endedAt": run.finished_at.map(|t| t.to_rfc3339()),
         "taskStates": run.task_states,
+        "taskLogs": run.task_logs,
         "triggeredBy": run.triggered_by,
         "environment": run.environment,
     })))
@@ -192,7 +197,7 @@ pub async fn list_all_runs(
     Query(params): Query<ListRunsQuery>,
 ) -> Json<Value> {
     let limit = params.limit.unwrap_or(100);
-    let mut runs = state.get_runs(None);
+    let mut runs = state.get_runs(params.dag_id.as_deref());
 
     if let Some(ref status) = params.status {
         runs.retain(|r| r.status == *status);

conduit-api/src/lib.rs

@@ -19,7 +19,7 @@ pub mod routes;
 pub mod state;
 pub mod websocket;
 
-pub use state::AppState;
+pub use state::{AppState, DagRunInfo};
 
 use std::net::SocketAddr;
 use std::sync::Arc;

conduit-api/src/state.rs

@@ -37,6 +37,10 @@ pub struct DagRunInfo {
     #[serde(rename = "endedAt")]
     pub finished_at: Option<DateTime<Utc>>,
     pub task_states: HashMap<String, String>,
+    /// Captured stdout/stderr per task (truncated), for post-hoc debugging
+    /// of completed runs. Live streaming still happens over the WebSocket.
+    #[serde(default)]
+    pub task_logs: HashMap<String, String>,
     pub triggered_by: String,
     /// Virtual environment this run targets. Persisted JSON written before
     /// this field existed deserializes as "production".
@@ -131,6 +135,22 @@ impl AppState {
             }
         }
 
+        // Open (creating if missing) the persistent event store. Failure is
+        // non-fatal: the events API reports empty and lifecycle events are
+        // simply not recorded. The CLI's replay command reads the same DB.
+        let events_dir = state_dir.join("events");
+        let event_store = match conduit_state::EventStore::open(&events_dir) {
+            Ok(store) => Some(Arc::new(store)),
+            Err(e) => {
+                tracing::warn!(
+                    path = %events_dir.display(),
+                    error = %e,
+                    "Failed to open event store; events will not be recorded",
+                );
+                None
+            }
+        };
+
         // Attach env history store so promote/rollback record versions on disk.
         let env_manager = EnvironmentManager::new();
         let env_manager = match conduit_state::EnvHistoryStore::open(state_dir.join("env_history"))
@@ -140,30 +160,14 @@ impl AppState {
         };
         let snapshot_store = Arc::new(SnapshotStore::new());
         let env_manager = env_manager.with_snapshot_store(Arc::clone(&snapshot_store));
+        let env_manager = match &event_store {
+            Some(store) => env_manager.with_event_store(Arc::clone(store)),
+            None => env_manager,
+        };
 
         let external_lineage = Arc::new(open_external_lineage_store(&state_dir));
         let plan_cache = Arc::new(PlanCache::new(dags_path.clone()));
 
-        // Open the persistent event store if it exists. Failure is non-fatal:
-        // the events API just reports an empty result with a clear note. The
-        // CLI's audit/replay commands open the same DB at the same location.
-        let events_dir = state_dir.join("events");
-        let event_store = if events_dir.exists() {
-            match conduit_state::EventStore::open(&events_dir) {
-                Ok(store) => Some(Arc::new(store)),
-                Err(e) => {
-                    tracing::warn!(
-                        path = %events_dir.display(),
-                        error = %e,
-                        "Failed to open event store; /events API will return empty",
-                    );
-                    None
-                }
-            }
-        } else {
-            None
-        };
-
         Arc::new(Self {
             dags_path,
             state_dir,
@@ -309,8 +313,10 @@ impl AppState {
             .unwrap_or_default()
     }
 
-    /// Seed the state with realistic demo run history.
-    /// Called at startup when `--demo` flag is passed.
+    /// Seed the state with fabricated demo run history so the UI has data
+    /// to display. Only called when `conduit serve --demo` is passed; all
+    /// seeded runs carry `triggered_by: "demo"` so they are distinguishable
+    /// from real runs.
     pub fn seed_demo_data(&self) {
         let now = Utc::now();
 
@@ -365,7 +371,8 @@ impl AppState {
                 started_at: started,
                 finished_at: ended,
                 task_states,
-                triggered_by: "scheduler".to_string(),
+                task_logs: HashMap::new(),
+                triggered_by: "demo".to_string(),
                 environment: "production".to_string(),
             });
         }
@@ -402,7 +409,8 @@ impl AppState {
                 started_at: started,
                 finished_at: Some(started + Duration::hours(1) + Duration::minutes(23)),
                 task_states,
-                triggered_by: "scheduler".to_string(),
+                task_logs: HashMap::new(),
+                triggered_by: "demo".to_string(),
                 environment: "production".to_string(),
             });
         }
@@ -465,11 +473,8 @@ impl AppState {
                 started_at: started,
                 finished_at: ended,
                 task_states,
-                triggered_by: if day == 3 {
-                    "api".to_string()
-                } else {
-                    "scheduler".to_string()
-                },
+                task_logs: HashMap::new(),
+                triggered_by: "demo".to_string(),
                 environment: if day == 4 {
                     "staging".to_string()
                 } else {
@@ -504,7 +509,8 @@ impl AppState {
                 started_at: started,
                 finished_at: Some(started + Duration::minutes(28 + day * 3)),
                 task_states,
-                triggered_by: "scheduler".to_string(),
+                task_logs: HashMap::new(),
+                triggered_by: "demo".to_string(),
                 environment: if day < 2 {
                     "staging".to_string()
                 } else {

conduit-api/tests/handler_tests.rs

@@ -579,6 +579,7 @@ async fn runs_list_with_seeded_data() {
             started_at: chrono::Utc::now(),
             finished_at: Some(chrono::Utc::now()),
             task_states: HashMap::new(),
+            task_logs: HashMap::new(),
             triggered_by: "test".to_string(),
             environment: "production".to_string(),
         });
@@ -609,6 +610,7 @@ async fn runs_list_respects_limit_param() {
             started_at: chrono::Utc::now(),
             finished_at: Some(chrono::Utc::now()),
             task_states: HashMap::new(),
+            task_logs: HashMap::new(),
             triggered_by: "test".to_string(),
             environment: "production".to_string(),
         });
@@ -633,6 +635,7 @@ async fn runs_list_filters_by_status() {
             started_at: chrono::Utc::now(),
             finished_at: None,
             task_states: HashMap::new(),
+            task_logs: HashMap::new(),
             triggered_by: "test".to_string(),
             environment: "production".to_string(),
         });
@@ -658,6 +661,7 @@ async fn runs_list_filters_by_environment() {
             started_at: chrono::Utc::now(),
             finished_at: None,
             task_states: HashMap::new(),
+            task_logs: HashMap::new(),
             triggered_by: "test".to_string(),
             environment: env.to_string(),
         });
@@ -730,6 +734,7 @@ async fn run_get_by_id() {
             ("task_a".into(), "success".into()),
             ("task_b".into(), "success".into()),
         ]),
+        task_logs: HashMap::new(),
         triggered_by: "api".to_string(),
         environment: "production".to_string(),
     });
@@ -1045,3 +1050,56 @@ async fn cluster_status_has_expected_fields() {
     // Should return an object with cluster info.
     assert!(body.is_object());
 }
+
+#[tokio::test]
+async fn runs_list_filters_by_dag_id_param() {
+    let (router, state) = app(false);
+
+    for (i, dag) in ["alpha", "beta", "alpha"].iter().enumerate() {
+        state.record_run(DagRunInfo {
+            run_id: format!("run-{}", i),
+            dag_id: dag.to_string(),
+            status: "success".to_string(),
+            started_at: chrono::Utc::now(),
+            finished_at: Some(chrono::Utc::now()),
+            task_states: HashMap::new(),
+            task_logs: HashMap::new(),
+            triggered_by: "test".to_string(),
+            environment: "production".to_string(),
+        });
+    }
+
+    // Both snake_case and camelCase spellings must filter.
+    let (_, body) = get(&router, "/api/v1/runs?dag_id=alpha").await;
+    assert_eq!(body["total"], 2, "dag_id filter must apply: {}", body);
+
+    let (_, body) = get(&router, "/api/v1/runs?dagId=beta").await;
+    assert_eq!(body["total"], 1, "dagId filter must apply: {}", body);
+}
+
+#[tokio::test]
+async fn run_detail_exposes_task_logs() {
+    let (router, state) = app(false);
+
+    let mut task_logs = HashMap::new();
+    task_logs.insert("greet".to_string(), "Hello from Conduit!\n".to_string());
+    state.record_run(DagRunInfo {
+        run_id: "run-logs".to_string(),
+        dag_id: "hello".to_string(),
+        status: "success".to_string(),
+        started_at: chrono::Utc::now(),
+        finished_at: Some(chrono::Utc::now()),
+        task_states: HashMap::new(),
+        task_logs,
+        triggered_by: "test".to_string(),
+        environment: "production".to_string(),
+    });
+
+    let (status, body) = get(&router, "/api/v1/runs/run-logs").await;
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(
+        body["taskLogs"]["greet"], "Hello from Conduit!\n",
+        "run detail must expose captured task logs: {}",
+        body
+    );
+}