jayhere1
diff --git a/‎Cargo.lock‎
Lines changed: 1 addition & 0 deletions b/‎Cargo.lock‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎conduit-api/Cargo.toml‎
Lines changed: 1 addition & 1 deletion b/‎conduit-api/Cargo.toml‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎conduit-api/src/error.rs‎
Lines changed: 4 additions & 0 deletions b/‎conduit-api/src/error.rs‎
Lines changed: 4 additions & 0 deletions
diff --git a/‎conduit-api/src/handlers/envs.rs‎
Lines changed: 5 additions & 0 deletions b/‎conduit-api/src/handlers/envs.rs‎
Lines changed: 5 additions & 0 deletions
diff --git a/‎conduit-api/src/handlers/plan.rs‎
Lines changed: 218 additions & 33 deletions b/‎conduit-api/src/handlers/plan.rs‎
Lines changed: 218 additions & 33 deletions
diff --git a/‎conduit-api/src/handlers/runs.rs‎
Lines changed: 10 additions & 1 deletion b/‎conduit-api/src/handlers/runs.rs‎
Lines changed: 10 additions & 1 deletion
@@ -14,6 +14,7 @@ conduit-scheduler = { path = "../conduit-scheduler" }
 conduit-planner = { path = "../conduit-planner" }
 conduit-lineage = { path = "../conduit-lineage" }
 conduit-providers = { path = "../conduit-providers" }
+conduit-executor = { path = "../conduit-executor" }
 axum = { workspace = true, features = ["ws"] }
 tower = { workspace = true }
 tower-http = { workspace = true }
@@ -42,4 +43,3 @@ conduit-common = { path = "../conduit-common" }
 conduit-compiler = { path = "../conduit-compiler" }
 conduit-state = { path = "../conduit-state" }
 conduit-scheduler = { path = "../conduit-scheduler" }
-conduit-executor = { path = "../conduit-executor" }
@@ -16,6 +16,8 @@ pub enum ApiError {
     Unauthorized(String),
     Forbidden(String),
     PromotionPolicyViolation(String),
+    Conflict(String),
+    ApplyFailed(String),
 }
 
 impl IntoResponse for ApiError {
@@ -52,6 +54,8 @@ impl IntoResponse for ApiError {
                 "promotion_policy_violation",
                 msg,
             ),
+            ApiError::Conflict(msg) => (StatusCode::CONFLICT, "conflict", msg),
+            ApiError::ApplyFailed(msg) => (StatusCode::UNPROCESSABLE_ENTITY, "apply_failed", msg),
         };
 
         let body = json!({
 
@@ -90,6 +90,7 @@ pub async fn create_environment(
         .env_manager
         .create(&body.name, body.based_on.as_deref())
         .map_err(ApiError::from)?;
+    state.persist_environments();
 
     // Broadcast event
     let event = json!({
@@ -156,6 +157,7 @@ pub async fn delete_environment(
         .env_manager
         .delete(&env_name)
         .map_err(ApiError::from)?;
+    state.persist_environments();
 
     Ok(Json(json!({
         "message": format!("Environment '{}' deleted", env_name),
@@ -174,6 +176,7 @@ pub async fn promote_environment(
         .env_manager
         .promote(&body.source, &body.target)
         .map_err(ApiError::from)?;
+    state.persist_environments();
 
     // Broadcast event
     let event = json!({
@@ -385,6 +388,7 @@ pub async fn update_env_policy(
         .env_manager
         .set_promotion_policy(&env_name, policy)
         .map_err(ApiError::from)?;
+    state.persist_environments();
 
     Ok(Json(json!({
         "environment": env.id,
@@ -413,6 +417,7 @@ pub async fn rollback_environment(
         .env_manager
         .rollback(&env_name, body.to_version)
         .map_err(ApiError::from)?;
+    state.persist_environments();
 
     let event = json!({
         "type": "environment_rolled_back",
 
@@ -61,6 +61,10 @@ pub async fn generate_plan(
     // Generate deployment plan
     let deploy = conduit_planner::DeploymentPlan::generate(&plan, &env, &state.snapshot_store);
 
+    // Cache it so a later POST /apply can look it up by id and apply
+    // exactly what was reviewed here.
+    state.store_plan(&deploy);
+
     // Broadcast plan event
     let event = json!({
         "type": "plan_generated",
@@ -110,66 +114,247 @@ pub async fn generate_plan(
 }
 
 /// POST /api/v1/apply — apply a deployment plan.
+///
+/// Honors `plan_id` when provided: looks the plan up in the in-memory
+/// cache populated by POST /plan, enforces that it targets the requested
+/// environment and that the environment hasn't moved since the plan was
+/// generated (409 on stale), then executes it for real — dispatching each
+/// `Execute` action through the provider registry, validating contracts,
+/// storing snapshots, and recording the environment update with history.
+/// Without `plan_id`, generates a fresh plan against the current
+/// environment state and applies it immediately.
 pub async fn apply_plan(
     auth: RequireAuth,
     State(state): State<Arc<AppState>>,
     Json(body): Json<ApplyRequest>,
 ) -> Result<Json<Value>, ApiError> {
     auth.require(Permission::ApplyPlan)?;
 
-    let env_name = body.environment.as_deref().unwrap_or("production");
+    let env_name = body
+        .environment
+        .as_deref()
+        .unwrap_or("production")
+        .to_string();
 
-    // Compile and generate fresh plan
+    // Compile current DAGs — needed to look up task definitions for execution.
     let (plan, stats) = ConduitPlan::compile(&state.dags_path)
         .map_err(|e| ApiError::CompilationFailed(e.to_string()))?;
-
     if !stats.errors.is_empty() {
         let error_msgs: Vec<String> = stats.errors.iter().map(|e| e.to_string()).collect();
         return Err(ApiError::CompilationFailed(error_msgs.join("; ")));
     }
 
-    let env = state
-        .env_manager
-        .get(env_name)
-        .unwrap_or_else(|_| conduit_common::snapshot::Environment::new(env_name));
-
-    let deploy = conduit_planner::DeploymentPlan::generate(&plan, &env, &state.snapshot_store);
+    let deploy = if let Some(plan_id) = body.plan_id.as_deref() {
+        let stored = state.get_plan(plan_id).ok_or_else(|| {
+            ApiError::NotFound(format!(
+                "plan '{}' not found (plans are cached in-memory; regenerate via POST /api/v1/plan)",
+                plan_id
+            ))
+        })?;
+        if stored.target_environment != env_name {
+            return Err(ApiError::BadRequest(format!(
+                "plan '{}' targets environment '{}', not '{}'",
+                plan_id, stored.target_environment, env_name
+            )));
+        }
+        let current_version = state
+            .env_manager
+            .get(&env_name)
+            .map(|e| e.current_version)
+            .unwrap_or(0);
+        if current_version != stored.base_environment_version {
+            return Err(ApiError::Conflict(format!(
+                "stale plan: environment '{}' is at version {} but plan '{}' was generated against version {} — regenerate the plan",
+                env_name, current_version, plan_id, stored.base_environment_version
+            )));
+        }
+        stored
+    } else {
+        let env = state
+            .env_manager
+            .get(&env_name)
+            .unwrap_or_else(|_| conduit_common::snapshot::Environment::new(&env_name));
+        let deploy = conduit_planner::DeploymentPlan::generate(&plan, &env, &state.snapshot_store);
+        state.store_plan(&deploy);
+        deploy
+    };
 
     if deploy.stats.tasks_to_execute == 0 && deploy.stats.tasks_to_remove == 0 {
         return Ok(Json(json!({
+            "plan_id": deploy.id,
             "message": format!("Nothing to apply. Environment '{}' is up to date.", env_name),
-            "tasks_executed": 0,
-            "tasks_reused": 0,
-            "tasks_removed": 0,
+            "status": "noop",
+            "tasks_executed": 0, "tasks_reused": 0, "tasks_removed": 0,
         })));
     }
 
-    // In production, this would dispatch to the scheduler/executor.
-    // For now, we record the intent and report what would happen.
+    state.broadcast_event(
+        &json!({
+            "type": "apply_started",
+            "plan_id": deploy.id,
+            "environment": env_name,
+            "tasks_to_execute": deploy.stats.tasks_to_execute,
+            "timestamp": Utc::now().to_rfc3339(),
+        })
+        .to_string(),
+    );
 
-    let executable_count = deploy.executable_actions().len();
+    // ── Execute the plan (mirrors CLI cmd_apply) ──
+    use conduit_executor::process_runner::{ProcessRunner, TaskContext};
+    use conduit_planner::ActionKind;
 
-    // Broadcast apply event
-    let event = json!({
-        "type": "apply_started",
-        "plan_id": deploy.id,
-        "environment": env_name,
-        "tasks_to_execute": executable_count,
-        "timestamp": Utc::now().to_rfc3339(),
-    });
-    state.broadcast_event(&event.to_string());
+    let registry = state.provider_registry.read().ok().and_then(|g| g.clone());
+    let contract_index: std::collections::HashMap<
+        (String, String),
+        &conduit_common::contracts::TaskContracts,
+    > = deploy
+        .pending_contracts
+        .iter()
+        .map(|tc| {
+            (
+                (tc.dag_id.clone().unwrap_or_default(), tc.task_id.clone()),
+                tc,
+            )
+        })
+        .collect();
+    let mut contract_results: Vec<conduit_common::contracts::ValidationResult> = Vec::new();
+    let mut new_snapshots: std::collections::HashMap<(String, String), String> =
+        std::collections::HashMap::new();
+    let (mut executed, mut reused, mut removed) = (0usize, 0usize, 0usize);
+    let logical_date = Utc::now();
+    let run_id = format!("apply_{}", Utc::now().format("%Y%m%d%H%M%S"));
+
+    for action in &deploy.actions {
+        match &action.action {
+            ActionKind::Execute => {
+                let task = plan
+                    .dags
+                    .get(&action.dag_id)
+                    .and_then(|dag| dag.tasks.get(&action.task_id))
+                    .ok_or_else(|| {
+                        ApiError::ApplyFailed(format!(
+                            "task {}.{} not found in compiled plan",
+                            action.dag_id, action.task_id
+                        ))
+                    })?;
+
+                let context = TaskContext {
+                    dag_id: action.dag_id.clone(),
+                    run_id: run_id.clone(),
+                    task_id: action.task_id.clone(),
+                    attempt: 1,
+                    logical_date,
+                    environment: env_name.clone(),
+                    params: Default::default(),
+                    extra_env: Vec::new(),
+                };
+
+                let output = ProcessRunner::run_with_providers(task, &context, registry.as_deref())
+                    .await
+                    .map_err(|e| {
+                        ApiError::ApplyFailed(format!(
+                            "task {}.{} execution error: {}",
+                            action.dag_id, action.task_id, e
+                        ))
+                    })?;
+                if output.exit_code != 0 {
+                    return Err(ApiError::ApplyFailed(format!(
+                        "task {}.{} failed with exit code {}: {}",
+                        action.dag_id,
+                        action.task_id,
+                        output.exit_code,
+                        output.stderr.trim()
+                    )));
+                }
+
+                if let Some(tc) =
+                    contract_index.get(&(action.dag_id.clone(), action.task_id.clone()))
+                {
+                    let result = conduit_common::contracts::ContractEvaluator::evaluate(
+                        tc,
+                        &output.evidence,
+                    );
+                    let blocked = !result.passed;
+                    contract_results.push(result);
+                    if blocked {
+                        return Err(ApiError::ApplyFailed(format!(
+                            "contract validation failed for {}.{} — environment not updated",
+                            action.dag_id, action.task_id
+                        )));
+                    }
+                }
+
+                let snap_id = format!(
+                    "snap_{}_{}",
+                    action.task_id,
+                    Utc::now().format("%Y%m%d%H%M%S%3f")
+                );
+                if let Some(ref fp) = action.fingerprint {
+                    let snapshot = conduit_common::snapshot::Snapshot {
+                        id: snap_id.clone(),
+                        fingerprint: fp.clone(),
+                        dag_id: action.dag_id.clone(),
+                        task_id: action.task_id.clone(),
+                        created_at: Utc::now(),
+                        parent_fingerprints: vec![],
+                        metadata: Default::default(),
+                    };
+                    let _ = state.snapshot_store.put(snapshot);
+                }
+                new_snapshots.insert((action.dag_id.clone(), action.task_id.clone()), snap_id);
+                executed += 1;
+            }
+            ActionKind::ReuseSnapshot { .. } => reused += 1,
+            ActionKind::Skip => {}
+            ActionKind::Remove => removed += 1,
+        }
+    }
+
+    // ── Update the environment (history-recorded, rollbackable) ──
+    if state.env_manager.get(&env_name).is_err() {
+        let _ = state.env_manager.create(&env_name, None);
+    }
+    let mut env_snapshot = state
+        .env_manager
+        .get(&env_name)
+        .unwrap_or_else(|_| conduit_common::snapshot::Environment::new(&env_name));
+    deploy.apply_to_environment(&mut env_snapshot, &new_snapshots);
+    let recorded_version = state
+        .env_manager
+        .apply_snapshot_map(
+            &env_name,
+            env_snapshot.snapshot_map.clone(),
+            deploy.id.clone(),
+        )
+        .map_err(|e| ApiError::Internal(e.to_string()))?;
+    state.persist_environments();
+
+    if let Some(store) = &state.event_store {
+        let _ = store.append(conduit_common::event::EventKind::PlanApplied {
+            plan_id: deploy.id.clone(),
+            environment: env_name.clone(),
+            tasks_executed: executed as u32,
+            tasks_skipped: reused as u32,
+        });
+    }
+    state.broadcast_event(
+        &json!({
+            "type": "apply_completed",
+            "plan_id": deploy.id,
+            "environment": env_name,
+            "tasks_executed": executed,
+            "timestamp": Utc::now().to_rfc3339(),
+        })
+        .to_string(),
+    );
 
     Ok(Json(json!({
         "plan_id": deploy.id,
         "environment": env_name,
-        "status": "accepted",
-        "tasks_to_execute": deploy.stats.tasks_to_execute,
-        "tasks_to_reuse": deploy.stats.tasks_to_reuse,
-        "tasks_to_skip": deploy.stats.tasks_to_skip,
-        "tasks_to_remove": deploy.stats.tasks_to_remove,
-        "message": format!(
-            "Apply accepted. {} tasks queued for execution in '{}'.",
-            deploy.stats.tasks_to_execute, env_name
-        ),
+        "status": "applied",
+        "tasks_executed": executed,
+        "tasks_reused": reused,
+        "tasks_removed": removed,
+        "environment_version": recorded_version,
     })))
 }
@@ -68,8 +68,17 @@ pub async fn trigger_run(
         .map(|dt| dt.with_timezone(&Utc))
         .unwrap_or(now);
 
-    let config = body.config.unwrap_or_default();
+    let mut config = body.config.unwrap_or_default();
     let environment = body.environment.unwrap_or_else(|| "production".to_string());
+    // The scheduler reads environment/triggered_by out of the run config
+    // (scheduler.rs handle_dag_run_requested) — without these keys every
+    // API-triggered run is logged as production/scheduler.
+    config
+        .entry("environment".to_string())
+        .or_insert_with(|| environment.clone());
+    config
+        .entry("triggered_by".to_string())
+        .or_insert_with(|| "api".to_string());
 
     let task_states: HashMap<String, String> = dag
         .tasks