feat: CI runner MVP

Cargo.lock

@@ -1383,9 +1383,9 @@ dependencies = [
 
 [[package]]
 name = "futures"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876"
+checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -1411,9 +1411,9 @@ dependencies = [
 
 [[package]]
 name = "futures-channel"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10"
+checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -1421,15 +1421,15 @@ dependencies = [
 
 [[package]]
 name = "futures-core"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
+checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
 
 [[package]]
 name = "futures-executor"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d"
 dependencies = [
  "futures-core",
  "futures-task",
@@ -1438,9 +1438,9 @@ dependencies = [
 
 [[package]]
 name = "futures-io"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
+checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
 
 [[package]]
 name = "futures-lite"
@@ -1457,9 +1457,9 @@ dependencies = [
 
 [[package]]
 name = "futures-macro"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
+checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1468,21 +1468,21 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7"
+checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
 
 [[package]]
 name = "futures-task"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988"
+checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393"
 
 [[package]]
 name = "futures-util"
-version = "0.3.31"
+version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
+checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -1492,7 +1492,6 @@ dependencies = [
  "futures-task",
  "memchr",
  "pin-project-lite",
- "pin-utils",
  "slab",
 ]
 
@@ -3786,6 +3785,7 @@ dependencies = [
  "serde",
  "serde_json",
  "swactor",
+ "swactor-ci",
  "tiny_http",
  "tracing",
  "tracing-subscriber",
@@ -4057,6 +4057,19 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_yaml"
+version = "0.9.34+deprecated"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
+dependencies = [
+ "indexmap",
+ "itoa",
+ "ryu",
+ "serde",
+ "unsafe-libyaml",
+]
+
 [[package]]
 name = "sha1_smol"
 version = "1.0.1"
@@ -4178,6 +4191,7 @@ dependencies = [
  "serde_json",
  "simulation",
  "swactor",
+ "swactor-ci",
  "tiny_http",
  "toml",
 ]
@@ -4375,6 +4389,16 @@ dependencies = [
  "wat",
 ]
 
+[[package]]
+name = "swactor-ci"
+version = "0.1.0"
+dependencies = [
+ "serde",
+ "serde_json",
+ "serde_yaml",
+ "swactor",
+]
+
 [[package]]
 name = "swactor-datastore"
 version = "0.1.0"
@@ -4974,6 +4998,12 @@ version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7264e107f553ccae879d21fbea1d6724ac785e8c3bfc762137959b5802826ef3"
 
+[[package]]
+name = "unsafe-libyaml"
+version = "0.2.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861"
+
 [[package]]
 name = "untrusted"
 version = "0.9.0"

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = [".", "crates/python", "crates/wasm", "crates/bin-runner", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/std", "crates/datastore", "tests/docker", "xtask"]
+members = [".", "crates/python", "crates/wasm", "crates/bin-runner", "crates/simulation", "crates/runtime-dashboard", "crates/distribution", "crates/std", "crates/datastore", "tests/docker", "xtask", "crates/ci"]
 exclude = ["tools/depgraph"]
 
 [package]

crates/ci/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "swactor-ci"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+swactor = { path = "../..", features = ["serde"] }
+serde = { version = "1", features = ["derive"] }
+serde_yaml = "0.9"
+serde_json = "1"
+
+[dev-dependencies]

crates/ci/src/coordinator.rs

@@ -0,0 +1,426 @@
+//! Coordinator actor: central brain of the CI system.
+//!
+//! Receives webhook events, manages pipeline lifecycles, dispatches jobs
+//! to the Provisioner and RunnerSupervisor actors.
+
+use std::collections::HashMap;
+
+use swactor::actor::{ActorAddress, ActorInterface, Ctx};
+
+use crate::pipeline::PipelineExecution;
+use crate::yaml::{self, CiYaml};
+use crate::{
+    CiConfig, JobComplete, JobId, JobProgress, JobStatus, PipelineId, ProvisionRequest,
+    ProvisionResponse, StatusUpdate, TerminateRequest, WebhookEvent,
+};
+
+/// Messages the Coordinator can receive.
+#[derive(Debug, Clone)]
+pub enum CoordinatorMsg {
+    /// A webhook event from Forgejo.
+    Webhook(WebhookEvent),
+    /// The CI YAML config to use (loaded externally or from the repo).
+    SetCiYaml(CiYaml),
+    /// Response from the Provisioner.
+    ProvisionResponse(ProvisionResponse),
+    /// Streamed output from a RunnerSupervisor.
+    JobProgress(JobProgress),
+    /// Final result from a RunnerSupervisor.
+    JobComplete(JobComplete),
+    /// Notification that the provisioner is offline (detected via SWIM).
+    ProvisionerOffline,
+    /// Notification that the provisioner is back online.
+    ProvisionerOnline,
+}
+
+/// The Coordinator actor state.
+pub struct Coordinator {
+    config: CiConfig,
+    ci_yaml: Option<CiYaml>,
+    pipelines: HashMap<PipelineId, PipelineExecution>,
+    next_pipeline_id: u64,
+    provisioner_addr: Option<ActorAddress>,
+    provisioner_online: bool,
+    /// Maps job_id → runner supervisor address.
+    runner_addrs: HashMap<JobId, ActorAddress>,
+    /// Captured status updates (for testing/simulation).
+    status_updates: Vec<StatusUpdate>,
+    /// Jobs waiting for the provisioner to come online.
+    queued_provisions: Vec<ProvisionRequest>,
+}
+
+impl Coordinator {
+    pub fn new(config: CiConfig) -> Self {
+        Self {
+            config,
+            ci_yaml: None,
+            pipelines: HashMap::new(),
+            next_pipeline_id: 1,
+            provisioner_addr: None,
+            provisioner_online: false,
+            runner_addrs: HashMap::new(),
+            status_updates: Vec::new(),
+            queued_provisions: Vec::new(),
+        }
+    }
+
+    pub fn with_provisioner(mut self, addr: ActorAddress) -> Self {
+        self.provisioner_addr = Some(addr);
+        self.provisioner_online = true;
+        self
+    }
+
+    pub fn with_ci_yaml(mut self, yaml: CiYaml) -> Self {
+        self.ci_yaml = Some(yaml);
+        self
+    }
+
+    pub fn pipelines(&self) -> &HashMap<PipelineId, PipelineExecution> {
+        &self.pipelines
+    }
+
+    pub fn status_updates(&self) -> &[StatusUpdate] {
+        &self.status_updates
+    }
+
+    fn handle_webhook(&mut self, ctx: &Ctx, event: WebhookEvent) {
+        let ci = match &self.ci_yaml {
+            Some(ci) => ci.clone(),
+            None => return,
+        };
+
+        let matched = yaml::matching_pipelines(&ci, &event);
+        for pipeline_name in matched {
+            let pipeline_def = &ci.pipelines[&pipeline_name];
+            let pipeline_id = PipelineId(self.next_pipeline_id);
+            self.next_pipeline_id += 1;
+
+            // Build job definitions.
+            let job_defs: Vec<_> = pipeline_def
+                .jobs
+                .iter()
+                .map(|(name, def)| yaml::to_job_definition(name, def))
+                .collect();
+
+            let pipeline = PipelineExecution::new(
+                pipeline_id,
+                pipeline_name.clone(),
+                event.repo_owner.clone(),
+                event.repo_name.clone(),
+                event.commit_sha.clone(),
+                event.branch.clone(),
+                job_defs,
+            );
+
+            // Set pending status on Forgejo.
+            self.emit_status_update(StatusUpdate {
+                repo_owner: event.repo_owner.clone(),
+                repo_name: event.repo_name.clone(),
+                commit_sha: event.commit_sha.clone(),
+                state: "pending".into(),
+                context: format!("ci/{pipeline_name}"),
+                description: format!("Pipeline '{pipeline_name}' is pending"),
+            });
+
+            self.pipelines.insert(pipeline_id, pipeline);
+
+            // Start eligible jobs.
+            self.advance_pipeline(ctx, pipeline_id);
+        }
+    }
+
+    fn advance_pipeline(&mut self, ctx: &Ctx, pipeline_id: PipelineId) {
+        let pipeline = match self.pipelines.get(&pipeline_id) {
+            Some(p) => p,
+            None => return,
+        };
+
+        // If pipeline is already terminal, emit final status.
+        if pipeline.status.is_terminal() {
+            let update = StatusUpdate {
+                repo_owner: pipeline.repo_owner.clone(),
+                repo_name: pipeline.repo_name.clone(),
+                commit_sha: pipeline.commit_sha.clone(),
+                state: pipeline.status.forgejo_state().into(),
+                context: format!("ci/{}", pipeline.pipeline_name),
+                description: format!(
+                    "Pipeline '{}' {}",
+                    pipeline.pipeline_name,
+                    pipeline.status.forgejo_state()
+                ),
+            };
+            self.emit_status_update(update);
+            return;
+        }
+
+        let eligible = pipeline.eligible_jobs();
+        let repo_owner = pipeline.repo_owner.clone();
+        let repo_name = pipeline.repo_name.clone();
+        let commit_sha = pipeline.commit_sha.clone();
+
+        for job_name in eligible {
+            let job_id = JobId {
+                pipeline_id,
+                job_name: job_name.clone(),
+            };
+
+            let spec = {
+                let pipeline = self.pipelines.get(&pipeline_id).unwrap();
+                let job = &pipeline.jobs[&job_name];
+                crate::InstanceSpec {
+                    docker_required: job.definition.docker,
+                    ..Default::default()
+                }
+            };
+
+            if self.provisioner_online {
+                // Request provisioning.
+                if let Some(prov_addr) = self.provisioner_addr {
+                    let request = ProvisionRequest {
+                        job_id: job_id.clone(),
+                        instance_spec: spec,
+                    };
+                    let _ = ctx.send(prov_addr, crate::provisioner::ProvisionerMsg::Provision(request));
+                }
+                if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
+                    if let Some(job) = pipeline.jobs.get_mut(&job_name) {
+                        job.status = JobStatus::Provisioning;
+                    }
+                }
+            } else {
+                // Queue for later.
+                let request = ProvisionRequest {
+                    job_id: job_id.clone(),
+                    instance_spec: spec,
+                };
+                self.queued_provisions.push(request);
+                if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
+                    if let Some(job) = pipeline.jobs.get_mut(&job_name) {
+                        job.status = JobStatus::WaitingForProvisioner;
+                    }
+                }
+            }
+
+            // Emit per-job status.
+            self.emit_status_update(StatusUpdate {
+                repo_owner: repo_owner.clone(),
+                repo_name: repo_name.clone(),
+                commit_sha: commit_sha.clone(),
+                state: "pending".into(),
+                context: format!("ci/{job_name}"),
+                description: format!("Job '{job_name}' is provisioning"),
+            });
+        }
+    }
+
+    fn handle_provision_response(&mut self, ctx: &Ctx, response: ProvisionResponse) {
+        let pipeline_id = response.job_id.pipeline_id;
+        let job_name = response.job_id.job_name.clone();
+
+        match response.result {
+            Ok(instance) => {
+                // Store instance_id for cleanup.
+                if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
+                    if let Some(job) = pipeline.jobs.get_mut(&job_name) {
+                        job.instance_id = Some(instance.instance_id.clone());
+                        job.status = JobStatus::Running;
+                    }
+                }
+
+                // Spawn a RunnerSupervisor for this job.
+                let pipeline = &self.pipelines[&pipeline_id];
+                let job = &pipeline.jobs[&job_name];
+
+                let start_job = crate::StartJob {
+                    job_id: response.job_id.clone(),
+                    instance: instance.clone(),
+                    repo_url: format!(
+                        "{}/{}/{}",
+                        self.config.forgejo_url, pipeline.repo_owner, pipeline.repo_name
+                    ),
+                    commit_sha: pipeline.commit_sha.clone(),
+                    job_def: job.definition.clone(),
+                };
+
+                let runner = crate::runner::RunnerSupervisor::new(
+                    ctx.self_addr(),
+                    start_job,
+                );
+
+                match ctx.spawn(runner) {
+                    Ok(runner_addr) => {
+                        self.runner_addrs.insert(response.job_id, runner_addr);
+                    }
+                    Err(_) => {
+                        // Failed to spawn runner — mark job as failed.
+                        if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
+                            pipeline.set_job_status(
+                                &job_name,
+                                JobStatus::Failed {
+                                    reason: "failed to spawn runner".into(),
+                                },
+                            );
+                        }
+                        self.advance_pipeline(ctx, pipeline_id);
+                    }
+                }
+            }
+            Err(err) => {
+                if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
+                    pipeline.set_job_status(
+                        &job_name,
+                        JobStatus::Failed {
+                            reason: err.to_string(),
+                        },
+                    );
+                }
+                self.advance_pipeline(ctx, pipeline_id);
+            }
+        }
+    }
+
+    fn handle_job_complete(&mut self, ctx: &Ctx, complete: JobComplete) {
+        let pipeline_id = complete.job_id.pipeline_id;
+        let job_name = complete.job_id.job_name.clone();
+
+        // Send terminate request for the instance.
+        if let Some(pipeline) = self.pipelines.get(&pipeline_id) {
+            if let Some(job) = pipeline.jobs.get(&job_name) {
+                if let Some(ref instance_id) = job.instance_id {
+                    let terminate = TerminateRequest {
+                        job_id: complete.job_id.clone(),
+                        instance_id: instance_id.clone(),
+                    };
+                    if let Some(prov_addr) = self.provisioner_addr {
+                        let _ = ctx.send(
+                            prov_addr,
+                            crate::provisioner::ProvisionerMsg::Terminate(terminate),
+                        );
+                    }
+                }
+            }
+        }
+
+        // Update job status.
+        let status = match complete.result {
+            Ok(_) => JobStatus::Passed,
+            Err(ref failure) => JobStatus::Failed {
+                reason: failure.to_string(),
+            },
+        };
+
+        let (repo_owner, repo_name, commit_sha) = {
+            let pipeline = match self.pipelines.get(&pipeline_id) {
+                Some(p) => p,
+                None => return,
+            };
+            (
+                pipeline.repo_owner.clone(),
+                pipeline.repo_name.clone(),
+                pipeline.commit_sha.clone(),
+            )
+        };
+
+        // Emit per-job final status.
+        self.emit_status_update(StatusUpdate {
+            repo_owner,
+            repo_name,
+            commit_sha,
+            state: match &status {
+                JobStatus::Passed => "success".into(),
+                _ => "failure".into(),
+            },
+            context: format!("ci/{job_name}"),
+            description: format!("Job '{job_name}' completed"),
+        });
+
+        if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
+            pipeline.set_job_status(&job_name, status);
+        }
+
+        // Remove runner address.
+        self.runner_addrs.remove(&complete.job_id);
+
+        // Advance pipeline to schedule downstream jobs.
+        self.advance_pipeline(ctx, pipeline_id);
+    }
+
+    fn handle_provisioner_offline(&mut self) {
+        self.provisioner_online = false;
+
+        // Mark all provisioning jobs as waiting.
+        for pipeline in self.pipelines.values_mut() {
+            for job in pipeline.jobs.values_mut() {
+                if job.status == JobStatus::Provisioning {
+                    job.status = JobStatus::WaitingForProvisioner;
+                }
+            }
+        }
+    }
+
+    fn handle_provisioner_online(&mut self, ctx: &Ctx) {
+        self.provisioner_online = true;
+
+        // Flush queued provision requests.
+        let queued = std::mem::take(&mut self.queued_provisions);
+        for request in queued {
+            if let Some(prov_addr) = self.provisioner_addr {
+                let _ = ctx.send(prov_addr, crate::provisioner::ProvisionerMsg::Provision(request));
+            }
+        }
+
+        // Re-advance pipelines that have waiting jobs.
+        let pipeline_ids: Vec<PipelineId> = self.pipelines.keys().copied().collect();
+        for pid in pipeline_ids {
+            // Move waiting jobs back to provisioning.
+            if let Some(pipeline) = self.pipelines.get_mut(&pid) {
+                let waiting_jobs: Vec<String> = pipeline
+                    .jobs
+                    .iter()
+                    .filter(|(_, j)| j.status == JobStatus::WaitingForProvisioner)
+                    .map(|(name, _)| name.clone())
+                    .collect();
+
+                for job_name in waiting_jobs {
+                    if let Some(job) = pipeline.jobs.get_mut(&job_name) {
+                        job.status = JobStatus::Pending;
+                    }
+                }
+            }
+            self.advance_pipeline(ctx, pid);
+        }
+    }
+
+    fn emit_status_update(&mut self, update: StatusUpdate) {
+        self.status_updates.push(update);
+    }
+}
+
+impl ActorInterface for Coordinator {
+    type Incoming = CoordinatorMsg;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: CoordinatorMsg) {
+        match msg {
+            CoordinatorMsg::Webhook(event) => self.handle_webhook(ctx, event),
+            CoordinatorMsg::SetCiYaml(yaml) => {
+                self.ci_yaml = Some(yaml);
+            }
+            CoordinatorMsg::ProvisionResponse(resp) => {
+                self.handle_provision_response(ctx, resp);
+            }
+            CoordinatorMsg::JobProgress(progress) => {
+                if let Some(pipeline) = self.pipelines.get_mut(&progress.job_id.pipeline_id) {
+                    if let Some(job) = pipeline.jobs.get_mut(&progress.job_id.job_name) {
+                        job.output_lines.push(progress.output_line);
+                    }
+                }
+            }
+            CoordinatorMsg::JobComplete(complete) => {
+                self.handle_job_complete(ctx, complete);
+            }
+            CoordinatorMsg::ProvisionerOffline => self.handle_provisioner_offline(),
+            CoordinatorMsg::ProvisionerOnline => self.handle_provisioner_online(ctx),
+        }
+    }
+}

crates/ci/src/lib.rs

@@ -0,0 +1,288 @@
+pub mod coordinator;
+pub mod pipeline;
+pub mod provisioner;
+pub mod runner;
+pub mod yaml;
+
+use std::collections::HashMap;
+use std::fmt;
+use std::net::IpAddr;
+
+use serde::{Deserialize, Serialize};
+
+// ─── Core Identifiers ───────────────────────────────────────────────────────
+
+/// Unique identifier for a pipeline execution.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct PipelineId(pub u64);
+
+impl fmt::Display for PipelineId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "pipeline-{}", self.0)
+    }
+}
+
+/// Unique identifier for a job within a pipeline.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct JobId {
+    pub pipeline_id: PipelineId,
+    pub job_name: String,
+}
+
+impl fmt::Display for JobId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}/{}", self.pipeline_id, self.job_name)
+    }
+}
+
+// ─── Job Status ─────────────────────────────────────────────────────────────
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum JobStatus {
+    Pending,
+    WaitingForProvisioner,
+    Provisioning,
+    Running,
+    Passed,
+    Failed { reason: String },
+    Skipped,
+    Interrupted,
+}
+
+impl JobStatus {
+    pub fn is_terminal(&self) -> bool {
+        matches!(
+            self,
+            JobStatus::Passed | JobStatus::Failed { .. } | JobStatus::Skipped | JobStatus::Interrupted
+        )
+    }
+}
+
+// ─── Pipeline Status ────────────────────────────────────────────────────────
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum PipelineStatus {
+    Pending,
+    Running,
+    Passed,
+    Failed,
+    Error { reason: String },
+}
+
+impl PipelineStatus {
+    pub fn is_terminal(&self) -> bool {
+        matches!(
+            self,
+            PipelineStatus::Passed | PipelineStatus::Failed | PipelineStatus::Error { .. }
+        )
+    }
+
+    /// Convert to Forgejo commit status string.
+    pub fn forgejo_state(&self) -> &'static str {
+        match self {
+            PipelineStatus::Pending => "pending",
+            PipelineStatus::Running => "pending",
+            PipelineStatus::Passed => "success",
+            PipelineStatus::Failed => "failure",
+            PipelineStatus::Error { .. } => "error",
+        }
+    }
+}
+
+// ─── Instance Types ─────────────────────────────────────────────────────────
+
+/// Specification for a spot instance.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct InstanceSpec {
+    pub min_cpus: u32,
+    pub min_ram_mb: u32,
+    pub min_disk_gb: u32,
+    pub docker_required: bool,
+    pub region_preferences: Vec<String>,
+}
+
+impl Default for InstanceSpec {
+    fn default() -> Self {
+        Self {
+            min_cpus: 2,
+            min_ram_mb: 2048,
+            min_disk_gb: 20,
+            docker_required: false,
+            region_preferences: Vec::new(),
+        }
+    }
+}
+
+/// Connection details for a provisioned spot instance.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct InstanceReady {
+    pub instance_id: String,
+    pub ip: IpAddr,
+    pub ssh_port: u16,
+    pub ssh_host_key: String,
+}
+
+// ─── Provisioner ↔ Coordinator Messages ─────────────────────────────────────
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProvisionRequest {
+    pub job_id: JobId,
+    pub instance_spec: InstanceSpec,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProvisionResponse {
+    pub job_id: JobId,
+    pub result: Result<InstanceReady, ProvisionError>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum ProvisionError {
+    NoCapacity,
+    ProviderError(String),
+    Timeout,
+    ProvisionerOffline,
+}
+
+impl fmt::Display for ProvisionError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ProvisionError::NoCapacity => write!(f, "no capacity available"),
+            ProvisionError::ProviderError(msg) => write!(f, "provider error: {msg}"),
+            ProvisionError::Timeout => write!(f, "provisioning timed out"),
+            ProvisionError::ProvisionerOffline => write!(f, "provisioner is offline"),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TerminateRequest {
+    pub job_id: JobId,
+    pub instance_id: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TerminateAck {
+    pub job_id: JobId,
+}
+
+// ─── Coordinator ↔ RunnerSupervisor Messages ────────────────────────────────
+
+/// Sent from Coordinator to RunnerSupervisor to begin a job.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct StartJob {
+    pub job_id: JobId,
+    pub instance: InstanceReady,
+    pub repo_url: String,
+    pub commit_sha: String,
+    pub job_def: JobDefinition,
+}
+
+/// Streamed output from RunnerSupervisor back to Coordinator.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobProgress {
+    pub job_id: JobId,
+    pub output_line: String,
+}
+
+/// Final result from RunnerSupervisor.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobComplete {
+    pub job_id: JobId,
+    pub result: Result<JobSuccess, JobFailure>,
+    pub artifacts: Vec<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobSuccess;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum JobFailure {
+    CommandFailed { exit_code: i32, last_lines: Vec<String> },
+    SshError(String),
+    Timeout,
+    Interrupted,
+}
+
+impl fmt::Display for JobFailure {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            JobFailure::CommandFailed { exit_code, .. } => {
+                write!(f, "command exited with code {exit_code}")
+            }
+            JobFailure::SshError(msg) => write!(f, "SSH error: {msg}"),
+            JobFailure::Timeout => write!(f, "job timed out"),
+            JobFailure::Interrupted => write!(f, "spot instance interrupted"),
+        }
+    }
+}
+
+// ─── Job Definition ─────────────────────────────────────────────────────────
+
+/// A parsed job from the .ci.yml file.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobDefinition {
+    pub name: String,
+    pub run: Vec<String>,
+    pub needs: Vec<String>,
+    pub timeout_secs: u64,
+    pub docker: bool,
+    pub artifacts: Vec<String>,
+    pub env: HashMap<String, String>,
+}
+
+// ─── Webhook Types ──────────────────────────────────────────────────────────
+
+/// Parsed webhook event from Forgejo.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WebhookEvent {
+    pub event_type: EventType,
+    pub repo_owner: String,
+    pub repo_name: String,
+    pub branch: String,
+    pub commit_sha: String,
+    pub tag: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum EventType {
+    Push,
+    Tag,
+    Merge,
+}
+
+// ─── Forgejo Status Updates ─────────────────────────────────────────────────
+
+/// A commit status update to send to Forgejo's API.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct StatusUpdate {
+    pub repo_owner: String,
+    pub repo_name: String,
+    pub commit_sha: String,
+    pub state: String,
+    pub context: String,
+    pub description: String,
+}
+
+// ─── Coordinator Config ─────────────────────────────────────────────────────
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CiConfig {
+    pub webhook_port: u16,
+    pub webhook_secret: String,
+    pub forgejo_url: String,
+    pub forgejo_token: String,
+    pub data_dir: String,
+}
+
+impl Default for CiConfig {
+    fn default() -> Self {
+        Self {
+            webhook_port: 8787,
+            webhook_secret: String::new(),
+            forgejo_url: String::new(),
+            forgejo_token: String::new(),
+            data_dir: "./ci-data".into(),
+        }
+    }
+}

crates/ci/src/pipeline.rs

@@ -0,0 +1,391 @@
+//! Pipeline resolution, DAG execution logic, and job ordering.
+
+use std::collections::{HashMap, HashSet, VecDeque};
+
+use crate::{JobDefinition, JobId, JobStatus, PipelineId, PipelineStatus};
+
+/// A pipeline execution: tracks the DAG of jobs and their statuses.
+#[derive(Debug, Clone)]
+pub struct PipelineExecution {
+    pub pipeline_id: PipelineId,
+    pub pipeline_name: String,
+    pub repo_owner: String,
+    pub repo_name: String,
+    pub commit_sha: String,
+    pub branch: String,
+    pub status: PipelineStatus,
+    pub jobs: HashMap<String, JobExecution>,
+}
+
+/// State of a single job within a pipeline execution.
+#[derive(Debug, Clone)]
+pub struct JobExecution {
+    pub job_id: JobId,
+    pub definition: JobDefinition,
+    pub status: JobStatus,
+    pub output_lines: Vec<String>,
+    pub instance_id: Option<String>,
+}
+
+impl PipelineExecution {
+    /// Create a new pipeline execution from a set of job definitions.
+    pub fn new(
+        pipeline_id: PipelineId,
+        pipeline_name: String,
+        repo_owner: String,
+        repo_name: String,
+        commit_sha: String,
+        branch: String,
+        job_defs: Vec<JobDefinition>,
+    ) -> Self {
+        let mut jobs = HashMap::new();
+        for def in job_defs {
+            let job_id = JobId {
+                pipeline_id,
+                job_name: def.name.clone(),
+            };
+            jobs.insert(
+                def.name.clone(),
+                JobExecution {
+                    job_id,
+                    definition: def,
+                    status: JobStatus::Pending,
+                    output_lines: Vec::new(),
+                    instance_id: None,
+                },
+            );
+        }
+        Self {
+            pipeline_id,
+            pipeline_name,
+            repo_owner,
+            repo_name,
+            commit_sha,
+            branch,
+            status: PipelineStatus::Pending,
+            jobs,
+        }
+    }
+
+    /// Return job names that are eligible for execution: Pending with all needs satisfied.
+    pub fn eligible_jobs(&self) -> Vec<String> {
+        self.jobs
+            .values()
+            .filter(|job| {
+                job.status == JobStatus::Pending
+                    && job.definition.needs.iter().all(|dep| {
+                        self.jobs
+                            .get(dep)
+                            .map(|d| d.status == JobStatus::Passed)
+                            .unwrap_or(false)
+                    })
+            })
+            .map(|job| job.definition.name.clone())
+            .collect()
+    }
+
+    /// Mark a job as a given status. If a job fails, propagate skip to dependents.
+    pub fn set_job_status(&mut self, job_name: &str, status: JobStatus) {
+        if let Some(job) = self.jobs.get_mut(job_name) {
+            job.status = status.clone();
+        }
+
+        // If failure or interruption, skip all transitive dependents.
+        if matches!(
+            status,
+            JobStatus::Failed { .. } | JobStatus::Interrupted
+        ) {
+            let to_skip = self.transitive_dependents(job_name);
+            for dep_name in to_skip {
+                if let Some(dep_job) = self.jobs.get_mut(&dep_name) {
+                    if dep_job.status == JobStatus::Pending {
+                        dep_job.status = JobStatus::Skipped;
+                    }
+                }
+            }
+        }
+
+        // Update pipeline status.
+        self.update_pipeline_status();
+    }
+
+    /// Compute the overall pipeline status from individual job statuses.
+    fn update_pipeline_status(&mut self) {
+        let all_terminal = self.jobs.values().all(|j| j.status.is_terminal());
+        let any_running = self.jobs.values().any(|j| {
+            matches!(
+                j.status,
+                JobStatus::Running | JobStatus::Provisioning | JobStatus::WaitingForProvisioner
+            )
+        });
+        let any_failed = self.jobs.values().any(|j| {
+            matches!(
+                j.status,
+                JobStatus::Failed { .. } | JobStatus::Interrupted
+            )
+        });
+
+        if all_terminal {
+            self.status = if any_failed {
+                PipelineStatus::Failed
+            } else {
+                PipelineStatus::Passed
+            };
+        } else if any_running
+            || self
+                .jobs
+                .values()
+                .any(|j| j.status == JobStatus::Passed)
+        {
+            self.status = PipelineStatus::Running;
+        }
+    }
+
+    /// Find all transitive dependents of a job (jobs that directly or indirectly need it).
+    fn transitive_dependents(&self, job_name: &str) -> Vec<String> {
+        let mut result = Vec::new();
+        let mut queue: VecDeque<&str> = VecDeque::new();
+        queue.push_back(job_name);
+        let mut visited = HashSet::new();
+
+        while let Some(current) = queue.pop_front() {
+            for (name, job) in &self.jobs {
+                if job.definition.needs.iter().any(|n| n == current) && visited.insert(name.clone())
+                {
+                    result.push(name.clone());
+                    queue.push_back(name.as_str());
+                }
+            }
+        }
+        result
+    }
+}
+
+// ─── DAG Validation ─────────────────────────────────────────────────────────
+
+/// Topological sort of job definitions. Returns ordered job names or an error if cyclic.
+pub fn topological_sort(jobs: &HashMap<String, JobDefinition>) -> Result<Vec<String>, DagError> {
+    let mut in_degree: HashMap<&str, usize> = HashMap::new();
+    let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new();
+
+    for (name, def) in jobs {
+        in_degree.entry(name.as_str()).or_insert(0);
+        for dep in &def.needs {
+            if !jobs.contains_key(dep) {
+                return Err(DagError::MissingDependency {
+                    job: name.clone(),
+                    dependency: dep.clone(),
+                });
+            }
+            dependents.entry(dep.as_str()).or_default().push(name.as_str());
+            *in_degree.entry(name.as_str()).or_insert(0) += 1;
+        }
+    }
+
+    let mut queue: VecDeque<&str> = in_degree
+        .iter()
+        .filter(|(_, deg)| **deg == 0)
+        .map(|(&name, _)| name)
+        .collect();
+
+    // Sort the initial queue for deterministic ordering.
+    let mut sorted_queue: Vec<&str> = queue.drain(..).collect();
+    sorted_queue.sort();
+    queue.extend(sorted_queue);
+
+    let mut result = Vec::new();
+    while let Some(node) = queue.pop_front() {
+        result.push(node.to_string());
+        if let Some(deps) = dependents.get(node) {
+            let mut next_nodes = Vec::new();
+            for &dep in deps {
+                if let Some(deg) = in_degree.get_mut(dep) {
+                    *deg -= 1;
+                    if *deg == 0 {
+                        next_nodes.push(dep);
+                    }
+                }
+            }
+            next_nodes.sort();
+            queue.extend(next_nodes);
+        }
+    }
+
+    if result.len() != jobs.len() {
+        return Err(DagError::Cycle);
+    }
+    Ok(result)
+}
+
+#[derive(Debug, Clone)]
+pub enum DagError {
+    Cycle,
+    MissingDependency { job: String, dependency: String },
+}
+
+impl std::fmt::Display for DagError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            DagError::Cycle => write!(f, "job dependency cycle detected"),
+            DagError::MissingDependency { job, dependency } => {
+                write!(f, "job '{job}' depends on unknown job '{dependency}'")
+            }
+        }
+    }
+}
+
+impl std::error::Error for DagError {}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use super::*;
+    use crate::JobDefinition;
+
+    fn make_job(name: &str, needs: &[&str]) -> JobDefinition {
+        JobDefinition {
+            name: name.into(),
+            run: vec!["echo test".into()],
+            needs: needs.iter().map(|s| s.to_string()).collect(),
+            timeout_secs: 300,
+            docker: false,
+            artifacts: Vec::new(),
+            env: HashMap::new(),
+        }
+    }
+
+    #[test]
+    fn topological_sort_linear_chain() {
+        let mut jobs = HashMap::new();
+        jobs.insert("a".into(), make_job("a", &[]));
+        jobs.insert("b".into(), make_job("b", &["a"]));
+        jobs.insert("c".into(), make_job("c", &["b"]));
+
+        let order = topological_sort(&jobs).unwrap();
+        assert_eq!(order, vec!["a", "b", "c"]);
+    }
+
+    #[test]
+    fn topological_sort_diamond() {
+        let mut jobs = HashMap::new();
+        jobs.insert("a".into(), make_job("a", &[]));
+        jobs.insert("b".into(), make_job("b", &["a"]));
+        jobs.insert("c".into(), make_job("c", &["a"]));
+        jobs.insert("d".into(), make_job("d", &["b", "c"]));
+
+        let order = topological_sort(&jobs).unwrap();
+        let pos = |name: &str| order.iter().position(|n| n == name).unwrap();
+        assert!(pos("a") < pos("b"));
+        assert!(pos("a") < pos("c"));
+        assert!(pos("b") < pos("d"));
+        assert!(pos("c") < pos("d"));
+    }
+
+    #[test]
+    fn topological_sort_detects_cycle() {
+        let mut jobs = HashMap::new();
+        jobs.insert("a".into(), make_job("a", &["b"]));
+        jobs.insert("b".into(), make_job("b", &["a"]));
+
+        let result = topological_sort(&jobs);
+        assert!(matches!(result, Err(DagError::Cycle)));
+    }
+
+    #[test]
+    fn topological_sort_independent_jobs() {
+        let mut jobs = HashMap::new();
+        jobs.insert("a".into(), make_job("a", &[]));
+        jobs.insert("b".into(), make_job("b", &[]));
+        jobs.insert("c".into(), make_job("c", &[]));
+
+        let order = topological_sort(&jobs).unwrap();
+        // All jobs present, order is alphabetical for independent nodes
+        assert_eq!(order.len(), 3);
+        assert_eq!(order, vec!["a", "b", "c"]);
+    }
+
+    #[test]
+    fn pipeline_eligible_jobs_respects_dag() {
+        let jobs = vec![
+            make_job("fmt", &[]),
+            make_job("clippy", &[]),
+            make_job("test", &["fmt", "clippy"]),
+        ];
+
+        let mut pipeline = PipelineExecution::new(
+            PipelineId(1),
+            "check".into(),
+            "user".into(),
+            "repo".into(),
+            "abc123".into(),
+            "main".into(),
+            jobs,
+        );
+
+        // Initially, fmt and clippy are eligible
+        let mut eligible = pipeline.eligible_jobs();
+        eligible.sort();
+        assert_eq!(eligible, vec!["clippy", "fmt"]);
+
+        // After fmt passes, test is still not eligible (clippy pending)
+        pipeline.set_job_status("fmt", JobStatus::Passed);
+        let eligible = pipeline.eligible_jobs();
+        assert_eq!(eligible, vec!["clippy"]);
+
+        // After clippy passes, test becomes eligible
+        pipeline.set_job_status("clippy", JobStatus::Passed);
+        let eligible = pipeline.eligible_jobs();
+        assert_eq!(eligible, vec!["test"]);
+    }
+
+    #[test]
+    fn pipeline_failure_skips_dependents() {
+        let jobs = vec![
+            make_job("fmt", &[]),
+            make_job("test", &["fmt"]),
+            make_job("bench", &["test"]),
+        ];
+
+        let mut pipeline = PipelineExecution::new(
+            PipelineId(1),
+            "check".into(),
+            "user".into(),
+            "repo".into(),
+            "abc123".into(),
+            "main".into(),
+            jobs,
+        );
+
+        // fmt fails → test and bench should be skipped
+        pipeline.set_job_status(
+            "fmt",
+            JobStatus::Failed {
+                reason: "formatting error".into(),
+            },
+        );
+
+        assert_eq!(pipeline.jobs["test"].status, JobStatus::Skipped);
+        assert_eq!(pipeline.jobs["bench"].status, JobStatus::Skipped);
+        assert_eq!(pipeline.status, PipelineStatus::Failed);
+    }
+
+    #[test]
+    fn pipeline_all_pass_yields_passed() {
+        let jobs = vec![make_job("a", &[]), make_job("b", &["a"])];
+
+        let mut pipeline = PipelineExecution::new(
+            PipelineId(1),
+            "check".into(),
+            "user".into(),
+            "repo".into(),
+            "abc123".into(),
+            "main".into(),
+            jobs,
+        );
+
+        pipeline.set_job_status("a", JobStatus::Passed);
+        pipeline.set_job_status("b", JobStatus::Passed);
+        assert_eq!(pipeline.status, PipelineStatus::Passed);
+    }
+}

crates/ci/src/provisioner.rs

@@ -0,0 +1,86 @@
+//! Provisioner actor: manages spot instance lifecycle via pluggable provider scripts.
+//!
+//! In real deployment, runs on the developer's laptop and calls cloud provider APIs.
+//! In simulation, provisions are driven by the simulation harness.
+
+use swactor::actor::{ActorAddress, ActorInterface, Ctx};
+
+use crate::coordinator::CoordinatorMsg;
+use crate::{
+    InstanceReady, ProvisionError, ProvisionRequest, ProvisionResponse, TerminateRequest,
+};
+
+/// Messages the Provisioner can receive.
+#[derive(Debug, Clone)]
+pub enum ProvisionerMsg {
+    /// Request to provision a new spot instance.
+    Provision(ProvisionRequest),
+    /// Request to terminate a spot instance.
+    Terminate(TerminateRequest),
+    /// Simulated: provisioning result delivered asynchronously.
+    SimProvisionResult {
+        request: ProvisionRequest,
+        result: Result<InstanceReady, ProvisionError>,
+    },
+}
+
+/// Provisioner actor state.
+///
+/// In real deployment, this would invoke provider scripts.
+/// In simulation, the sim harness controls provision outcomes.
+pub struct Provisioner {
+    coordinator_addr: ActorAddress,
+    /// Active instances tracked for cleanup.
+    active_instances: Vec<String>,
+}
+
+impl Provisioner {
+    pub fn new(coordinator_addr: ActorAddress) -> Self {
+        Self {
+            coordinator_addr,
+            active_instances: Vec::new(),
+        }
+    }
+
+    pub fn active_instances(&self) -> &[String] {
+        &self.active_instances
+    }
+}
+
+impl ActorInterface for Provisioner {
+    type Incoming = ProvisionerMsg;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: ProvisionerMsg) {
+        match msg {
+            ProvisionerMsg::Provision(_request) => {
+                // In real deployment: invoke provider script, await result.
+                // In simulation: the sim harness sends SimProvisionResult.
+            }
+            ProvisionerMsg::Terminate(request) => {
+                self.active_instances.retain(|id| id != &request.instance_id);
+                // In real deployment: invoke provider destroy script.
+                // In simulation: just track the termination.
+                let _ = ctx.send(
+                    self.coordinator_addr,
+                    CoordinatorMsg::ProvisionResponse(ProvisionResponse {
+                        job_id: request.job_id.clone(),
+                        result: Err(ProvisionError::NoCapacity), // placeholder, terminate doesn't need response
+                    }),
+                );
+            }
+            ProvisionerMsg::SimProvisionResult { request, result } => {
+                if let Ok(ref instance) = result {
+                    self.active_instances.push(instance.instance_id.clone());
+                }
+                let _ = ctx.send(
+                    self.coordinator_addr,
+                    CoordinatorMsg::ProvisionResponse(ProvisionResponse {
+                        job_id: request.job_id,
+                        result,
+                    }),
+                );
+            }
+        }
+    }
+}

crates/ci/src/runner.rs

@@ -0,0 +1,84 @@
+//! RunnerSupervisor actor: manages SSH session and job execution on a spot instance.
+//!
+//! Spawned per-job by the Coordinator. Owns the connection to the spot instance.
+
+use swactor::actor::{ActorAddress, ActorInterface, Ctx};
+
+use crate::coordinator::CoordinatorMsg;
+use crate::{JobComplete, JobFailure, JobProgress, JobSuccess, StartJob};
+
+/// Messages the RunnerSupervisor can receive.
+#[derive(Debug, Clone)]
+pub enum RunnerMsg {
+    /// Begin executing the job (sent immediately after spawn via on_start).
+    Execute,
+    /// Simulated: job command output line.
+    OutputLine(String),
+    /// Simulated: job completed successfully.
+    SimComplete(Result<(), String>),
+}
+
+/// RunnerSupervisor actor state.
+///
+/// In real deployment, this would manage an SSH connection.
+/// In simulation, job execution is driven by external messages.
+pub struct RunnerSupervisor {
+    coordinator_addr: ActorAddress,
+    start_job: StartJob,
+}
+
+impl RunnerSupervisor {
+    pub fn new(coordinator_addr: ActorAddress, start_job: StartJob) -> Self {
+        Self {
+            coordinator_addr,
+            start_job,
+        }
+    }
+}
+
+impl ActorInterface for RunnerSupervisor {
+    type Incoming = RunnerMsg;
+    type Response = ();
+
+    fn on_start(&mut self, ctx: &Ctx) {
+        // In simulation, the sim harness will send SimComplete messages.
+        // In real deployment, this would initiate SSH connection + command execution.
+        let _ = ctx.send(ctx.self_addr(), RunnerMsg::Execute);
+    }
+
+    fn handle(&mut self, ctx: &Ctx, msg: RunnerMsg) {
+        match msg {
+            RunnerMsg::Execute => {
+                // In real mode, we'd SSH into the instance and run commands.
+                // In simulation, this is a no-op; SimComplete drives completion.
+            }
+            RunnerMsg::OutputLine(line) => {
+                let _ = ctx.send(
+                    self.coordinator_addr,
+                    CoordinatorMsg::JobProgress(JobProgress {
+                        job_id: self.start_job.job_id.clone(),
+                        output_line: line,
+                    }),
+                );
+            }
+            RunnerMsg::SimComplete(result) => {
+                let complete = JobComplete {
+                    job_id: self.start_job.job_id.clone(),
+                    result: match result {
+                        Ok(()) => Ok(JobSuccess),
+                        Err(msg) => Err(JobFailure::CommandFailed {
+                            exit_code: 1,
+                            last_lines: vec![msg],
+                        }),
+                    },
+                    artifacts: Vec::new(),
+                };
+                let _ = ctx.send(
+                    self.coordinator_addr,
+                    CoordinatorMsg::JobComplete(complete),
+                );
+                ctx.stop_self();
+            }
+        }
+    }
+}

crates/ci/src/yaml.rs

@@ -0,0 +1,432 @@
+//! Parser for `.ci.yml` pipeline configuration files.
+
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+use crate::{EventType, JobDefinition, WebhookEvent};
+
+/// Root of a `.ci.yml` file.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CiYaml {
+    pub pipelines: HashMap<String, PipelineDef>,
+}
+
+/// A single pipeline definition.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PipelineDef {
+    pub triggers: Vec<TriggerDef>,
+    pub jobs: HashMap<String, JobDef>,
+}
+
+/// A trigger condition.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TriggerDef {
+    pub event: TriggerEvent,
+    #[serde(default)]
+    pub branches: Vec<String>,
+    #[serde(default)]
+    pub exclude: Vec<String>,
+    #[serde(default)]
+    pub pattern: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum TriggerEvent {
+    Push,
+    Tag,
+    Merge,
+}
+
+/// A job definition in YAML form.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobDef {
+    pub run: RunCommand,
+    #[serde(default)]
+    pub needs: Vec<String>,
+    #[serde(default)]
+    pub timeout: Option<u64>,
+    #[serde(default)]
+    pub docker: bool,
+    #[serde(default)]
+    pub artifacts: Vec<String>,
+    #[serde(default)]
+    pub env: HashMap<String, String>,
+}
+
+/// `run` can be a single string or an array of strings.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum RunCommand {
+    Single(String),
+    Multiple(Vec<String>),
+}
+
+impl RunCommand {
+    pub fn into_vec(self) -> Vec<String> {
+        match self {
+            RunCommand::Single(s) => vec![s],
+            RunCommand::Multiple(v) => v,
+        }
+    }
+}
+
+// ─── Parsing ────────────────────────────────────────────────────────────────
+
+/// Parse a `.ci.yml` string into a `CiYaml`.
+pub fn parse_ci_yaml(input: &str) -> Result<CiYaml, ParseError> {
+    let ci: CiYaml = serde_yaml::from_str(input).map_err(ParseError::Yaml)?;
+
+    // Validate: check for unknown job references in `needs`
+    for (pipeline_name, pipeline) in &ci.pipelines {
+        let job_names: Vec<&str> = pipeline.jobs.keys().map(|s| s.as_str()).collect();
+        for (job_name, job) in &pipeline.jobs {
+            for dep in &job.needs {
+                if !job_names.contains(&dep.as_str()) {
+                    return Err(ParseError::UnknownDependency {
+                        pipeline: pipeline_name.clone(),
+                        job: job_name.clone(),
+                        dependency: dep.clone(),
+                    });
+                }
+            }
+        }
+    }
+
+    Ok(ci)
+}
+
+/// Convert a YAML `JobDef` to the runtime `JobDefinition`.
+pub fn to_job_definition(name: &str, def: &JobDef) -> JobDefinition {
+    JobDefinition {
+        name: name.to_string(),
+        run: def.run.clone().into_vec(),
+        needs: def.needs.clone(),
+        timeout_secs: def.timeout.unwrap_or(300),
+        docker: def.docker,
+        artifacts: def.artifacts.clone(),
+        env: def.env.clone(),
+    }
+}
+
+// ─── Trigger Matching ───────────────────────────────────────────────────────
+
+/// Returns the names of pipelines whose triggers match the given webhook event.
+pub fn matching_pipelines(ci: &CiYaml, event: &WebhookEvent) -> Vec<String> {
+    ci.pipelines
+        .iter()
+        .filter(|(_, pipeline)| pipeline.triggers.iter().any(|t| trigger_matches(t, event)))
+        .map(|(name, _)| name.clone())
+        .collect()
+}
+
+/// Check whether a single trigger matches a webhook event.
+fn trigger_matches(trigger: &TriggerDef, event: &WebhookEvent) -> bool {
+    // Event type must match.
+    let event_matches = match (&trigger.event, &event.event_type) {
+        (TriggerEvent::Push, EventType::Push) => true,
+        (TriggerEvent::Tag, EventType::Tag) => true,
+        (TriggerEvent::Merge, EventType::Merge) => true,
+        _ => false,
+    };
+    if !event_matches {
+        return false;
+    }
+
+    // For tag events, check pattern.
+    if trigger.event == TriggerEvent::Tag {
+        if let Some(ref pattern) = trigger.pattern {
+            return glob_matches(pattern, event.tag.as_deref().unwrap_or(""));
+        }
+        return true;
+    }
+
+    // For push/merge, check branch filters.
+    let branch = &event.branch;
+
+    // If exclude patterns are specified and branch matches any, reject.
+    if trigger.exclude.iter().any(|pat| glob_matches(pat, branch)) {
+        return false;
+    }
+
+    // If branch patterns are specified, at least one must match.
+    if trigger.branches.is_empty() {
+        return true;
+    }
+    trigger.branches.iter().any(|pat| glob_matches(pat, branch))
+}
+
+/// Simple glob matching supporting `*` (any chars) and `?` (one char).
+pub fn glob_matches(pattern: &str, text: &str) -> bool {
+    glob_matches_inner(pattern.as_bytes(), text.as_bytes())
+}
+
+fn glob_matches_inner(pat: &[u8], text: &[u8]) -> bool {
+    match (pat.first(), text.first()) {
+        (None, None) => true,
+        (Some(b'*'), _) => {
+            // '*' matches zero or more characters
+            glob_matches_inner(&pat[1..], text)
+                || (!text.is_empty() && glob_matches_inner(pat, &text[1..]))
+        }
+        (Some(b'?'), Some(_)) => glob_matches_inner(&pat[1..], &text[1..]),
+        (Some(a), Some(b)) if a == b => glob_matches_inner(&pat[1..], &text[1..]),
+        _ => false,
+    }
+}
+
+// ─── Errors ─────────────────────────────────────────────────────────────────
+
+#[derive(Debug)]
+pub enum ParseError {
+    Yaml(serde_yaml::Error),
+    UnknownDependency {
+        pipeline: String,
+        job: String,
+        dependency: String,
+    },
+}
+
+impl std::fmt::Display for ParseError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ParseError::Yaml(e) => write!(f, "YAML parse error: {e}"),
+            ParseError::UnknownDependency {
+                pipeline,
+                job,
+                dependency,
+            } => write!(
+                f,
+                "pipeline '{pipeline}', job '{job}': unknown dependency '{dependency}'"
+            ),
+        }
+    }
+}
+
+impl std::error::Error for ParseError {}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_minimal_ci_yaml() {
+        let yaml = r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      test:
+        run: cargo test
+"#;
+        let ci = parse_ci_yaml(yaml).unwrap();
+        assert_eq!(ci.pipelines.len(), 1);
+        assert!(ci.pipelines.contains_key("check"));
+        let check = &ci.pipelines["check"];
+        assert_eq!(check.jobs.len(), 1);
+        assert!(check.jobs.contains_key("test"));
+    }
+
+    #[test]
+    fn parse_full_ci_yaml() {
+        let yaml = r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+        exclude: ["master"]
+    jobs:
+      fmt:
+        run: cargo fmt -- --check
+      clippy:
+        run: cargo clippy --all-features -- -D warnings
+      test:
+        needs: [fmt, clippy]
+        run: cargo test
+  full:
+    triggers:
+      - event: push
+        branches: ["master"]
+    jobs:
+      test:
+        run: cargo test --all-features
+        timeout: 600
+      bench:
+        needs: [test]
+        run: cargo bench -- --output-format json
+        artifacts: ["target/criterion/**"]
+      docker:
+        needs: [test]
+        run: cargo test -p docker-tests -- --ignored
+        docker: true
+  release:
+    triggers:
+      - event: tag
+        pattern: "v*"
+    jobs:
+      build:
+        run: cargo build --release
+        artifacts: ["target/release/swactor-node"]
+"#;
+        let ci = parse_ci_yaml(yaml).unwrap();
+        assert_eq!(ci.pipelines.len(), 3);
+        assert!(ci.pipelines.contains_key("check"));
+        assert!(ci.pipelines.contains_key("full"));
+        assert!(ci.pipelines.contains_key("release"));
+
+        let full = &ci.pipelines["full"];
+        assert_eq!(full.jobs.len(), 3);
+        assert_eq!(full.jobs["bench"].needs, vec!["test"]);
+        assert!(full.jobs["docker"].docker);
+    }
+
+    #[test]
+    fn unknown_dependency_rejected() {
+        let yaml = r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      test:
+        needs: [nonexistent]
+        run: cargo test
+"#;
+        let result = parse_ci_yaml(yaml);
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert!(
+            matches!(err, ParseError::UnknownDependency { .. }),
+            "expected UnknownDependency, got: {err}"
+        );
+    }
+
+    #[test]
+    fn trigger_matching_push_branch() {
+        let yaml = r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+        exclude: ["master"]
+    jobs:
+      test:
+        run: cargo test
+  full:
+    triggers:
+      - event: push
+        branches: ["master"]
+    jobs:
+      test:
+        run: cargo test
+"#;
+        let ci = parse_ci_yaml(yaml).unwrap();
+
+        // Push to feature branch → matches "check" only
+        let event = WebhookEvent {
+            event_type: EventType::Push,
+            repo_owner: "user".into(),
+            repo_name: "repo".into(),
+            branch: "feature-x".into(),
+            commit_sha: "abc123".into(),
+            tag: None,
+        };
+        let mut matched = matching_pipelines(&ci, &event);
+        matched.sort();
+        assert_eq!(matched, vec!["check"]);
+
+        // Push to master → matches "full" only (check excludes master)
+        let event = WebhookEvent {
+            event_type: EventType::Push,
+            repo_owner: "user".into(),
+            repo_name: "repo".into(),
+            branch: "master".into(),
+            commit_sha: "abc123".into(),
+            tag: None,
+        };
+        let mut matched = matching_pipelines(&ci, &event);
+        matched.sort();
+        assert_eq!(matched, vec!["full"]);
+    }
+
+    #[test]
+    fn trigger_matching_tag() {
+        let yaml = r#"
+pipelines:
+  release:
+    triggers:
+      - event: tag
+        pattern: "v*"
+    jobs:
+      build:
+        run: cargo build --release
+"#;
+        let ci = parse_ci_yaml(yaml).unwrap();
+
+        let event = WebhookEvent {
+            event_type: EventType::Tag,
+            repo_owner: "user".into(),
+            repo_name: "repo".into(),
+            branch: "master".into(),
+            commit_sha: "abc123".into(),
+            tag: Some("v1.0.0".into()),
+        };
+        let matched = matching_pipelines(&ci, &event);
+        assert_eq!(matched, vec!["release"]);
+
+        // Non-matching tag
+        let event = WebhookEvent {
+            event_type: EventType::Tag,
+            repo_owner: "user".into(),
+            repo_name: "repo".into(),
+            branch: "master".into(),
+            commit_sha: "abc123".into(),
+            tag: Some("nightly-1".into()),
+        };
+        let matched = matching_pipelines(&ci, &event);
+        assert!(matched.is_empty());
+    }
+
+    #[test]
+    fn glob_matching() {
+        assert!(glob_matches("*", "anything"));
+        assert!(glob_matches("v*", "v1.0.0"));
+        assert!(!glob_matches("v*", "nightly"));
+        assert!(glob_matches("feature-?", "feature-x"));
+        assert!(!glob_matches("feature-?", "feature-xy"));
+        assert!(glob_matches("master", "master"));
+        assert!(!glob_matches("master", "main"));
+    }
+
+    #[test]
+    fn run_command_single_and_multiple() {
+        let yaml = r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      single:
+        run: cargo test
+      multi:
+        run:
+          - cargo fmt -- --check
+          - cargo test
+"#;
+        let ci = parse_ci_yaml(yaml).unwrap();
+        let check = &ci.pipelines["check"];
+
+        let single = to_job_definition("single", &check.jobs["single"]);
+        assert_eq!(single.run, vec!["cargo test"]);
+
+        let multi = to_job_definition("multi", &check.jobs["multi"]);
+        assert_eq!(multi.run, vec!["cargo fmt -- --check", "cargo test"]);
+    }
+}

crates/runtime-dashboard/Cargo.toml

@@ -17,6 +17,7 @@ distribution = { path = "../distribution", optional = true }
 clap = { version = "4", features = ["derive"], optional = true }
 ctrlc = "3"
 iroh = { version = "0.96", optional = true }
+swactor-ci = { path = "../ci", optional = true }
 
 [features]
 default = ["distribution"]
@@ -25,6 +26,7 @@ distribution = ["dep:distribution"]
 node = ["distribution", "dep:clap", "swactor/transport", "tcp"]
 tcp = ["distribution/tcp"]
 iroh = ["distribution/iroh", "dep:iroh"]
+ci = ["dep:swactor-ci"]
 
 [[bin]]
 name = "swactor-tui"

crates/runtime-dashboard/src/ci_collector.rs

@@ -0,0 +1,210 @@
+//! CI Dashboard extension: provides HTTP API endpoints and stats for CI pipelines.
+
+use serde::{Deserialize, Serialize};
+
+use swactor_ci::{JobId, JobStatus, PipelineId, PipelineStatus};
+
+// ─── Stats Provider ─────────────────────────────────────────────────────────
+
+/// Trait for providing CI snapshot data to the dashboard.
+pub trait CiStatsProvider: Send + Sync {
+    fn snapshot(&self) -> CiSnapshot;
+}
+
+/// Point-in-time snapshot of CI system state.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CiSnapshot {
+    pub active_pipelines: Vec<PipelineSnapshot>,
+    pub recent_pipelines: Vec<PipelineSnapshot>,
+    pub provisioner_status: ProvisionerStatus,
+    pub active_instances: usize,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum ProvisionerStatus {
+    Online,
+    Offline,
+    Unknown,
+}
+
+/// Snapshot of a single pipeline execution.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PipelineSnapshot {
+    pub pipeline_id: PipelineId,
+    pub pipeline_name: String,
+    pub repo_owner: String,
+    pub repo_name: String,
+    pub commit_sha: String,
+    pub branch: String,
+    pub status: PipelineStatus,
+    pub jobs: Vec<JobSnapshot>,
+}
+
+/// Snapshot of a single job within a pipeline.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobSnapshot {
+    pub job_id: JobId,
+    pub job_name: String,
+    pub status: JobStatus,
+    pub output_line_count: usize,
+}
+
+// ─── HTTP API Responses ─────────────────────────────────────────────────────
+
+/// Response for GET /api/ci/pipelines
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PipelineListResponse {
+    pub pipelines: Vec<PipelineSnapshot>,
+}
+
+/// Response for GET /api/ci/pipelines/{id}
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PipelineDetailResponse {
+    pub pipeline: PipelineSnapshot,
+}
+
+/// Response for GET /api/ci/status
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SystemStatusResponse {
+    pub provisioner_status: ProvisionerStatus,
+    pub active_pipelines: usize,
+    pub active_instances: usize,
+}
+
+/// Response for GET /api/ci/pipelines/{id}/jobs/{job_id}/log
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobLogResponse {
+    pub job_id: JobId,
+    pub lines: Vec<String>,
+}
+
+// ─── Route Matching ─────────────────────────────────────────────────────────
+
+/// Parsed API route for the CI dashboard.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CiRoute {
+    ListPipelines,
+    PipelineDetail { id: u64 },
+    JobLog { pipeline_id: u64, job_name: String },
+    Artifact { job_id: String, path: String },
+    SystemStatus,
+    NotFound,
+}
+
+/// Parse a request path into a CiRoute.
+pub fn parse_route(path: &str) -> CiRoute {
+    let parts: Vec<&str> = path.trim_start_matches('/').split('/').collect();
+    match parts.as_slice() {
+        ["api", "ci", "pipelines"] => CiRoute::ListPipelines,
+        ["api", "ci", "pipelines", id] => {
+            if let Ok(id) = id.parse() {
+                CiRoute::PipelineDetail { id }
+            } else {
+                CiRoute::NotFound
+            }
+        }
+        ["api", "ci", "pipelines", id, "jobs", job_name, "log"] => {
+            if let Ok(pipeline_id) = id.parse() {
+                CiRoute::JobLog {
+                    pipeline_id,
+                    job_name: job_name.to_string(),
+                }
+            } else {
+                CiRoute::NotFound
+            }
+        }
+        ["api", "ci", "artifacts", job_id, rest @ ..] if !rest.is_empty() => CiRoute::Artifact {
+            job_id: job_id.to_string(),
+            path: rest.join("/"),
+        },
+        ["api", "ci", "status"] => CiRoute::SystemStatus,
+        _ => CiRoute::NotFound,
+    }
+}
+
+/// Render a CiSnapshot into a JSON response for the given route.
+pub fn handle_route(route: &CiRoute, snapshot: &CiSnapshot) -> Option<String> {
+    match route {
+        CiRoute::ListPipelines => {
+            let resp = PipelineListResponse {
+                pipelines: snapshot
+                    .active_pipelines
+                    .iter()
+                    .chain(snapshot.recent_pipelines.iter())
+                    .cloned()
+                    .collect(),
+            };
+            serde_json::to_string(&resp).ok()
+        }
+        CiRoute::PipelineDetail { id } => {
+            let pipeline = snapshot
+                .active_pipelines
+                .iter()
+                .chain(snapshot.recent_pipelines.iter())
+                .find(|p| p.pipeline_id.0 == *id)?;
+            let resp = PipelineDetailResponse {
+                pipeline: pipeline.clone(),
+            };
+            serde_json::to_string(&resp).ok()
+        }
+        CiRoute::SystemStatus => {
+            let resp = SystemStatusResponse {
+                provisioner_status: snapshot.provisioner_status.clone(),
+                active_pipelines: snapshot.active_pipelines.len(),
+                active_instances: snapshot.active_instances,
+            };
+            serde_json::to_string(&resp).ok()
+        }
+        CiRoute::JobLog { .. } | CiRoute::Artifact { .. } => {
+            // These require access to stored data beyond the snapshot.
+            None
+        }
+        CiRoute::NotFound => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn route_parsing() {
+        assert_eq!(parse_route("/api/ci/pipelines"), CiRoute::ListPipelines);
+        assert_eq!(
+            parse_route("/api/ci/pipelines/42"),
+            CiRoute::PipelineDetail { id: 42 }
+        );
+        assert_eq!(
+            parse_route("/api/ci/pipelines/1/jobs/test/log"),
+            CiRoute::JobLog {
+                pipeline_id: 1,
+                job_name: "test".into()
+            }
+        );
+        assert_eq!(
+            parse_route("/api/ci/artifacts/job-1/target/release/bin"),
+            CiRoute::Artifact {
+                job_id: "job-1".into(),
+                path: "target/release/bin".into()
+            }
+        );
+        assert_eq!(parse_route("/api/ci/status"), CiRoute::SystemStatus);
+        assert_eq!(parse_route("/api/ci/unknown"), CiRoute::NotFound);
+    }
+
+    #[test]
+    fn handle_system_status() {
+        let snapshot = CiSnapshot {
+            active_pipelines: vec![],
+            recent_pipelines: vec![],
+            provisioner_status: ProvisionerStatus::Online,
+            active_instances: 2,
+        };
+
+        let route = CiRoute::SystemStatus;
+        let json = handle_route(&route, &snapshot).unwrap();
+        let resp: SystemStatusResponse = serde_json::from_str(&json).unwrap();
+        assert_eq!(resp.provisioner_status, ProvisionerStatus::Online);
+        assert_eq!(resp.active_instances, 2);
+    }
+}

crates/runtime-dashboard/src/lib.rs

@@ -23,6 +23,9 @@ pub mod distribution_collector;
 mod datastore_html;
 pub mod datastore_collector;
 
+#[cfg(feature = "ci")]
+pub mod ci_collector;
+
 use std::io;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
@@ -95,6 +98,8 @@ pub struct DashboardHandle {
     #[cfg(feature = "distribution")]
     distribution: Arc<Mutex<Option<Arc<dyn distribution_collector::DistributionStatsProvider>>>>,
     datastore: Arc<Mutex<Option<Arc<dyn datastore_collector::DatastoreStatsProvider>>>>,
+    #[cfg(feature = "ci")]
+    ci: Arc<Mutex<Option<Arc<dyn ci_collector::CiStatsProvider>>>>,
 }
 
 impl DashboardHandle {
@@ -132,6 +137,12 @@ impl DashboardHandle {
         *self.datastore.lock().unwrap() = Some(provider);
     }
 
+    /// Attach a CI stats provider, enabling the `/api/ci/*` endpoints.
+    #[cfg(feature = "ci")]
+    pub fn set_ci(&self, provider: Arc<dyn ci_collector::CiStatsProvider>) {
+        *self.ci.lock().unwrap() = Some(provider);
+    }
+
     /// Access the time-series history store (for TUI sparklines, etc.).
     pub fn history(&self) -> &Arc<DashboardHistory> {
         &self.history
@@ -190,6 +201,10 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
     let datastore: Arc<Mutex<Option<Arc<dyn datastore_collector::DatastoreStatsProvider>>>> =
         Arc::new(Mutex::new(None));
 
+    #[cfg(feature = "ci")]
+    let ci: Arc<Mutex<Option<Arc<dyn ci_collector::CiStatsProvider>>>> =
+        Arc::new(Mutex::new(None));
+
     server::spawn_http_server(
         Arc::clone(&store),
         Arc::clone(&runtime),
@@ -200,6 +215,8 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
         #[cfg(feature = "distribution")]
         Arc::clone(&distribution),
         Arc::clone(&datastore),
+        #[cfg(feature = "ci")]
+        Arc::clone(&ci),
     );
 
     // Start stats recorder thread when recording is enabled
@@ -243,6 +260,8 @@ pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
         #[cfg(feature = "distribution")]
         distribution,
         datastore,
+        #[cfg(feature = "ci")]
+        ci,
     }
 }
 

crates/runtime-dashboard/src/server.rs

@@ -27,6 +27,9 @@ use crate::distribution_html::DISTRIBUTION_HTML;
 use crate::datastore_collector::DatastoreStatsProvider;
 use crate::datastore_html::DATASTORE_HTML;
 
+#[cfg(feature = "ci")]
+use crate::ci_collector::CiStatsProvider;
+
 /// Format a server-sent event.
 fn format_sse(event: &str, data: &str) -> Vec<u8> {
     format!("event: {event}\ndata: {data}\n\n").into_bytes()
@@ -132,6 +135,8 @@ pub(crate) fn spawn_http_server(
     #[cfg(feature = "distribution")]
     distribution: Arc<Mutex<Option<Arc<dyn DistributionStatsProvider>>>>,
     datastore: Arc<Mutex<Option<Arc<dyn DatastoreStatsProvider>>>>,
+    #[cfg(feature = "ci")]
+    ci: Arc<Mutex<Option<Arc<dyn CiStatsProvider>>>>,
 ) {
     let addr = format!("0.0.0.0:{port}");
     let server = tiny_http::Server::http(&addr).expect("failed to bind HTTP server");
@@ -149,6 +154,8 @@ pub(crate) fn spawn_http_server(
         #[cfg(feature = "distribution")]
         let distribution = Arc::clone(&distribution);
         let datastore = Arc::clone(&datastore);
+        #[cfg(feature = "ci")]
+        let ci = Arc::clone(&ci);
         thread::spawn(move || {
             loop {
                 let request = match server.recv() {
@@ -176,6 +183,8 @@ pub(crate) fn spawn_http_server(
                             #[cfg(feature = "distribution")]
                             Arc::clone(&distribution),
                             Arc::clone(&datastore),
+                            #[cfg(feature = "ci")]
+                            Arc::clone(&ci),
                         );
                     }
                     "/api/stats" => {
@@ -220,6 +229,10 @@ pub(crate) fn spawn_http_server(
                     "/api/logs" => {
                         handle_logs_api(request, &url, Arc::clone(&store));
                     }
+                    #[cfg(feature = "ci")]
+                    _ if path.starts_with("/api/ci/") => {
+                        handle_ci_api(request, path, Arc::clone(&ci));
+                    }
                     _ if path.starts_with("/actor/") => {
                         let hex = &path[7..]; // strip "/actor/"
                         respond_actor_detail(request, hex);
@@ -253,6 +266,8 @@ fn handle_live_sse(
     #[cfg(feature = "distribution")]
     distribution: Arc<Mutex<Option<Arc<dyn DistributionStatsProvider>>>>,
     datastore: Arc<Mutex<Option<Arc<dyn DatastoreStatsProvider>>>>,
+    #[cfg(feature = "ci")]
+    ci: Arc<Mutex<Option<Arc<dyn CiStatsProvider>>>>,
 ) {
     let (tx, rx) = mpsc::channel::<Vec<u8>>();
     let response = make_sse_response(rx);
@@ -335,6 +350,20 @@ fn handle_live_sse(
                 }
             }
 
+            // Send CI snapshot if provider is attached
+            #[cfg(feature = "ci")]
+            {
+                let maybe_ci = ci.lock().unwrap().clone();
+                if let Some(provider) = maybe_ci {
+                    let snapshot = provider.snapshot();
+                    if let Ok(json) = serde_json::to_string(&snapshot) {
+                        if tx.send(format_sse("ci", &json)).is_err() {
+                            return;
+                        }
+                    }
+                }
+            }
+
             // Send new activity events
             let (batch, new_cursor) = store.read_from(cursor);
             if !batch.is_empty() {
@@ -469,6 +498,35 @@ fn handle_datastore_api(
     let _ = request.respond(response);
 }
 
+#[cfg(feature = "ci")]
+fn handle_ci_api(
+    request: tiny_http::Request,
+    path: &str,
+    ci: Arc<Mutex<Option<Arc<dyn CiStatsProvider>>>>,
+) {
+    use crate::ci_collector;
+
+    let route = ci_collector::parse_route(path);
+    let json = match ci.lock().unwrap().as_ref() {
+        Some(provider) => {
+            let snapshot = provider.snapshot();
+            ci_collector::handle_route(&route, &snapshot)
+                .unwrap_or_else(|| r#"{"error":"not found"}"#.to_string())
+        }
+        None => serde_json::json!({
+            "error": "CI provider not attached"
+        })
+        .to_string(),
+    };
+
+    let response = tiny_http::Response::from_string(json).with_header(
+        "Content-Type: application/json"
+            .parse::<tiny_http::Header>()
+            .unwrap(),
+    );
+    let _ = request.respond(response);
+}
+
 fn handle_topology_api(
     request: tiny_http::Request,
     runtime: Arc<Mutex<Option<Arc<Runtime>>>>,

crates/simulation/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2024"
 default = []
 gossip = ["dep:log"]
 dashboard = ["gossip", "dep:tiny_http", "dep:toml"]
+ci = ["dep:swactor-ci"]
 
 [dependencies]
 distribution = { path = "../distribution" }
@@ -17,9 +18,11 @@ getrandom = "0.2"
 log = { version = "0.4", optional = true }
 tiny_http = { version = "0.12", optional = true }
 toml = { version = "0.8", optional = true }
+swactor-ci = { path = "../ci", optional = true }
 
 [dev-dependencies]
-simulation = { path = ".", features = ["gossip"] }
+simulation = { path = ".", features = ["gossip", "ci"] }
+swactor-ci = { path = "../ci" }
 
 [[example]]
 name = "gossip_sim"

crates/simulation/src/ci/mod.rs

@@ -0,0 +1 @@
+pub mod sim;

crates/simulation/src/ci/sim.rs

@@ -0,0 +1,610 @@
+//! CI protocol simulation: deterministic, round-based execution of the CI pipeline
+//! lifecycle without real IO (no SSH, no cloud API, no HTTP).
+//!
+//! Follows the same pattern as `crates/simulation/src/distribution/sim.rs`:
+//! configure → run rounds → collect trace → analyze properties.
+
+use std::collections::HashMap;
+
+use swactor_ci::pipeline::PipelineExecution;
+use swactor_ci::yaml::{self, CiYaml};
+use swactor_ci::{
+    JobId, JobStatus, PipelineId, ProvisionRequest, StatusUpdate, WebhookEvent,
+};
+
+// ─── Simulation Config ──────────────────────────────────────────────────────
+
+/// Configuration for a CI simulation run.
+#[derive(Debug, Clone)]
+pub struct CiSimConfig {
+    pub name: String,
+    pub num_rounds: usize,
+
+    /// CI YAML to use for all simulated repos.
+    pub ci_yaml: String,
+
+    /// Webhook events to inject at specific rounds.
+    pub webhook_schedule: Vec<(usize, WebhookEvent)>,
+
+    /// Rounds of latency for provisioning to complete.
+    pub provision_latency: usize,
+
+    /// Probability that provisioning fails (0.0-1.0).
+    pub provision_failure_rate: f64,
+
+    /// Rounds of latency for a job to complete.
+    pub job_duration: usize,
+
+    /// Force specific jobs to fail: (round, job_name_substring).
+    pub job_failure_schedule: Vec<(usize, String)>,
+
+    /// Rounds during which the provisioner is offline: (start_round, end_round).
+    pub provisioner_offline_schedule: Vec<(usize, usize)>,
+
+    /// Round at which a specific job's instance is interrupted.
+    pub instance_interrupt_schedule: Vec<(usize, String)>,
+}
+
+impl Default for CiSimConfig {
+    fn default() -> Self {
+        Self {
+            name: "ci-sim".into(),
+            num_rounds: 50,
+            ci_yaml: String::new(),
+            webhook_schedule: Vec::new(),
+            provision_latency: 2,
+            provision_failure_rate: 0.0,
+            job_duration: 3,
+            job_failure_schedule: Vec::new(),
+            provisioner_offline_schedule: Vec::new(),
+            instance_interrupt_schedule: Vec::new(),
+        }
+    }
+}
+
+// ─── Simulation Trace ───────────────────────────────────────────────────────
+
+/// Event recorded during simulation.
+#[derive(Debug, Clone)]
+pub enum CiSimEvent {
+    WebhookReceived { commit_sha: String },
+    PipelineCreated { pipeline_id: PipelineId, name: String },
+    ProvisionRequested { job_id: JobId },
+    ProvisionCompleted { job_id: JobId, success: bool },
+    JobStarted { job_id: JobId },
+    JobCompleted { job_id: JobId, passed: bool },
+    JobSkipped { job_id: JobId },
+    InstanceTerminated { instance_id: String },
+    ProvisionerWentOffline,
+    ProvisionerCameOnline,
+    StatusUpdateEmitted(StatusUpdate),
+}
+
+/// Per-round snapshot of simulation state.
+#[derive(Debug, Clone)]
+pub struct CiSimSnapshot {
+    pub active_pipelines: usize,
+    pub completed_pipelines: usize,
+    pub active_provisions: usize,
+    pub active_jobs: usize,
+    pub provisioner_online: bool,
+    pub active_instances: usize,
+}
+
+/// Complete trace output from a CI simulation.
+#[derive(Debug, Clone)]
+pub struct CiSimTrace {
+    pub name: String,
+    pub events: Vec<(usize, CiSimEvent)>,
+    pub snapshots: Vec<CiSimSnapshot>,
+    pub status_updates: Vec<StatusUpdate>,
+    pub num_rounds: usize,
+    /// Final state of all pipelines.
+    pub final_pipelines: Vec<PipelineExecution>,
+    /// Instances that were provisioned.
+    pub provisioned_instances: Vec<String>,
+    /// Instances that were terminated.
+    pub terminated_instances: Vec<String>,
+}
+
+// ─── Simulation State ───────────────────────────────────────────────────────
+
+/// Tracks an in-flight provision request.
+struct PendingProvision {
+    request: ProvisionRequest,
+    started_round: usize,
+}
+
+/// Tracks an in-flight job execution.
+struct RunningJob {
+    job_id: JobId,
+    started_round: usize,
+    instance_id: String,
+}
+
+/// Run a CI simulation and return the trace.
+pub fn run_simulation(config: CiSimConfig) -> CiSimTrace {
+    let ci_yaml: CiYaml = yaml::parse_ci_yaml(&config.ci_yaml)
+        .expect("CiSimConfig.ci_yaml must be valid YAML");
+
+    let mut events: Vec<(usize, CiSimEvent)> = Vec::new();
+    let mut snapshots: Vec<CiSimSnapshot> = Vec::new();
+
+    // Coordinator state (simulated directly, not as actor).
+    let mut pipelines: HashMap<PipelineId, PipelineExecution> = HashMap::new();
+    let mut next_pipeline_id: u64 = 1;
+    let mut all_status_updates: Vec<StatusUpdate> = Vec::new();
+
+    // Provisioner state.
+    let mut provisioner_online = true;
+    let mut pending_provisions: Vec<PendingProvision> = Vec::new();
+    let mut queued_provisions: Vec<ProvisionRequest> = Vec::new();
+    let mut provisioned_instances: Vec<String> = Vec::new();
+    let mut terminated_instances: Vec<String> = Vec::new();
+    let mut active_instances: Vec<String> = Vec::new();
+    let mut next_instance_id: u64 = 1;
+
+    // Runner state.
+    let mut running_jobs: Vec<RunningJob> = Vec::new();
+
+    // Simple deterministic "RNG" for provision failure decisions.
+    let mut rng_counter: u64 = 0x853c49e6748fea9b;
+    let mut det_random = || -> f64 {
+        rng_counter = rng_counter.wrapping_mul(6364136223846793005).wrapping_add(1);
+        (rng_counter >> 33) as f64 / (u32::MAX as f64)
+    };
+
+    for round in 1..=config.num_rounds {
+        // 1. Apply provisioner online/offline schedule.
+        let should_be_offline = config
+            .provisioner_offline_schedule
+            .iter()
+            .any(|(start, end)| round >= *start && round <= *end);
+
+        if should_be_offline && provisioner_online {
+            provisioner_online = false;
+            events.push((round, CiSimEvent::ProvisionerWentOffline));
+
+            // Move pending provisions to queue.
+            for pending in pending_provisions.drain(..) {
+                queued_provisions.push(pending.request);
+            }
+        } else if !should_be_offline && !provisioner_online {
+            provisioner_online = true;
+            events.push((round, CiSimEvent::ProvisionerCameOnline));
+
+            // Flush queued provisions.
+            for request in queued_provisions.drain(..) {
+                pending_provisions.push(PendingProvision {
+                    request,
+                    started_round: round,
+                });
+            }
+
+            // Also resubmit any WaitingForProvisioner jobs.
+            let mut resubmits = Vec::new();
+            for pipeline in pipelines.values_mut() {
+                for job in pipeline.jobs.values_mut() {
+                    if job.status == JobStatus::WaitingForProvisioner {
+                        job.status = JobStatus::Provisioning;
+                        resubmits.push(ProvisionRequest {
+                            job_id: job.job_id.clone(),
+                            instance_spec: swactor_ci::InstanceSpec {
+                                docker_required: job.definition.docker,
+                                ..Default::default()
+                            },
+                        });
+                    }
+                }
+            }
+            for request in resubmits {
+                events.push((round, CiSimEvent::ProvisionRequested { job_id: request.job_id.clone() }));
+                pending_provisions.push(PendingProvision {
+                    request,
+                    started_round: round,
+                });
+            }
+        }
+
+        // 2. Inject webhook events for this round.
+        for (sched_round, event) in &config.webhook_schedule {
+            if *sched_round == round {
+                events.push((
+                    round,
+                    CiSimEvent::WebhookReceived {
+                        commit_sha: event.commit_sha.clone(),
+                    },
+                ));
+
+                let matched = yaml::matching_pipelines(&ci_yaml, event);
+                for pipeline_name in matched {
+                    let pipeline_id = PipelineId(next_pipeline_id);
+                    next_pipeline_id += 1;
+
+                    let pipeline_def = &ci_yaml.pipelines[&pipeline_name];
+                    let job_defs: Vec<_> = pipeline_def
+                        .jobs
+                        .iter()
+                        .map(|(name, def)| yaml::to_job_definition(name, def))
+                        .collect();
+
+                    let pipeline = PipelineExecution::new(
+                        pipeline_id,
+                        pipeline_name.clone(),
+                        event.repo_owner.clone(),
+                        event.repo_name.clone(),
+                        event.commit_sha.clone(),
+                        event.branch.clone(),
+                        job_defs,
+                    );
+
+                    events.push((
+                        round,
+                        CiSimEvent::PipelineCreated {
+                            pipeline_id,
+                            name: pipeline_name.clone(),
+                        },
+                    ));
+
+                    // Emit pending status.
+                    all_status_updates.push(StatusUpdate {
+                        repo_owner: event.repo_owner.clone(),
+                        repo_name: event.repo_name.clone(),
+                        commit_sha: event.commit_sha.clone(),
+                        state: "pending".into(),
+                        context: format!("ci/{pipeline_name}"),
+                        description: format!("Pipeline '{pipeline_name}' is pending"),
+                    });
+
+                    pipelines.insert(pipeline_id, pipeline);
+                }
+            }
+        }
+
+        // 3. Complete provisions that have reached latency.
+        let mut completed_provisions = Vec::new();
+        pending_provisions.retain(|pending| {
+            if round - pending.started_round >= config.provision_latency {
+                completed_provisions.push(pending.request.clone());
+                false
+            } else {
+                true
+            }
+        });
+
+        for request in completed_provisions {
+            let should_fail = det_random() < config.provision_failure_rate;
+
+            if should_fail {
+                events.push((
+                    round,
+                    CiSimEvent::ProvisionCompleted {
+                        job_id: request.job_id.clone(),
+                        success: false,
+                    },
+                ));
+
+                if let Some(pipeline) = pipelines.get_mut(&request.job_id.pipeline_id) {
+                    pipeline.set_job_status(
+                        &request.job_id.job_name,
+                        JobStatus::Failed {
+                            reason: "provision failed".into(),
+                        },
+                    );
+                }
+            } else {
+                let instance_id = format!("instance-{next_instance_id}");
+                next_instance_id += 1;
+                provisioned_instances.push(instance_id.clone());
+                active_instances.push(instance_id.clone());
+
+                events.push((
+                    round,
+                    CiSimEvent::ProvisionCompleted {
+                        job_id: request.job_id.clone(),
+                        success: true,
+                    },
+                ));
+
+                // Mark job as running and record instance.
+                if let Some(pipeline) = pipelines.get_mut(&request.job_id.pipeline_id) {
+                    if let Some(job) = pipeline.jobs.get_mut(&request.job_id.job_name) {
+                        job.status = JobStatus::Running;
+                        job.instance_id = Some(instance_id.clone());
+                    }
+                }
+
+                events.push((
+                    round,
+                    CiSimEvent::JobStarted {
+                        job_id: request.job_id.clone(),
+                    },
+                ));
+
+                running_jobs.push(RunningJob {
+                    job_id: request.job_id,
+                    started_round: round,
+                    instance_id,
+                });
+            }
+        }
+
+        // 4. Apply instance interruptions.
+        for (interrupt_round, job_name_sub) in &config.instance_interrupt_schedule {
+            if *interrupt_round == round {
+                running_jobs.retain(|rj| {
+                    if rj.job_id.job_name.contains(job_name_sub.as_str()) {
+                        // Instance interrupted.
+                        if let Some(pipeline) = pipelines.get_mut(&rj.job_id.pipeline_id) {
+                            pipeline.set_job_status(&rj.job_id.job_name, JobStatus::Interrupted);
+                        }
+                        events.push((
+                            round,
+                            CiSimEvent::JobCompleted {
+                                job_id: rj.job_id.clone(),
+                                passed: false,
+                            },
+                        ));
+                        // Terminate the instance.
+                        active_instances.retain(|id| id != &rj.instance_id);
+                        terminated_instances.push(rj.instance_id.clone());
+                        events.push((
+                            round,
+                            CiSimEvent::InstanceTerminated {
+                                instance_id: rj.instance_id.clone(),
+                            },
+                        ));
+                        false
+                    } else {
+                        true
+                    }
+                });
+            }
+        }
+
+        // 5. Complete jobs that have reached duration.
+        let mut newly_completed = Vec::new();
+        running_jobs.retain(|rj| {
+            if round - rj.started_round >= config.job_duration {
+                newly_completed.push((rj.job_id.clone(), rj.instance_id.clone()));
+                false
+            } else {
+                true
+            }
+        });
+
+        for (job_id, instance_id) in newly_completed {
+            // Check if this job should fail per the schedule.
+            let should_fail = config
+                .job_failure_schedule
+                .iter()
+                .any(|(r, name_sub)| *r <= round && job_id.job_name.contains(name_sub.as_str()));
+
+            let passed = !should_fail;
+
+            if passed {
+                if let Some(pipeline) = pipelines.get_mut(&job_id.pipeline_id) {
+                    pipeline.set_job_status(&job_id.job_name, JobStatus::Passed);
+                }
+            } else {
+                if let Some(pipeline) = pipelines.get_mut(&job_id.pipeline_id) {
+                    pipeline.set_job_status(
+                        &job_id.job_name,
+                        JobStatus::Failed {
+                            reason: "command failed".into(),
+                        },
+                    );
+                }
+            }
+
+            events.push((
+                round,
+                CiSimEvent::JobCompleted {
+                    job_id: job_id.clone(),
+                    passed,
+                },
+            ));
+
+            // Terminate instance.
+            active_instances.retain(|id| id != &instance_id);
+            terminated_instances.push(instance_id.clone());
+            events.push((
+                round,
+                CiSimEvent::InstanceTerminated {
+                    instance_id: instance_id.clone(),
+                },
+            ));
+        }
+
+        // 6. Advance all pipelines: schedule newly-eligible jobs.
+        let pipeline_ids: Vec<PipelineId> = pipelines.keys().copied().collect();
+        for pid in pipeline_ids {
+            let eligible = pipelines[&pid].eligible_jobs();
+            for job_name in eligible {
+                let job_id = JobId {
+                    pipeline_id: pid,
+                    job_name: job_name.clone(),
+                };
+                let docker_required = pipelines[&pid].jobs[&job_name].definition.docker;
+
+                let request = ProvisionRequest {
+                    job_id: job_id.clone(),
+                    instance_spec: swactor_ci::InstanceSpec {
+                        docker_required,
+                        ..Default::default()
+                    },
+                };
+
+                if provisioner_online {
+                    if let Some(pipeline) = pipelines.get_mut(&pid) {
+                        if let Some(job) = pipeline.jobs.get_mut(&job_name) {
+                            job.status = JobStatus::Provisioning;
+                        }
+                    }
+                    events.push((
+                        round,
+                        CiSimEvent::ProvisionRequested { job_id },
+                    ));
+                    pending_provisions.push(PendingProvision {
+                        request,
+                        started_round: round,
+                    });
+                } else {
+                    if let Some(pipeline) = pipelines.get_mut(&pid) {
+                        if let Some(job) = pipeline.jobs.get_mut(&job_name) {
+                            job.status = JobStatus::WaitingForProvisioner;
+                        }
+                    }
+                    queued_provisions.push(request);
+                }
+            }
+
+            // Emit final status updates for terminal pipelines.
+            if let Some(pipeline) = pipelines.get(&pid) {
+                if pipeline.status.is_terminal() {
+                    // Check if we already emitted a terminal status for this pipeline.
+                    let context = format!("ci/{}", pipeline.pipeline_name);
+                    let already_emitted = all_status_updates.iter().any(|u| {
+                        u.context == context
+                            && u.commit_sha == pipeline.commit_sha
+                            && (u.state == "success" || u.state == "failure" || u.state == "error")
+                    });
+                    if !already_emitted {
+                        all_status_updates.push(StatusUpdate {
+                            repo_owner: pipeline.repo_owner.clone(),
+                            repo_name: pipeline.repo_name.clone(),
+                            commit_sha: pipeline.commit_sha.clone(),
+                            state: pipeline.status.forgejo_state().into(),
+                            context,
+                            description: format!(
+                                "Pipeline '{}' {}",
+                                pipeline.pipeline_name,
+                                pipeline.status.forgejo_state()
+                            ),
+                        });
+
+                        // Emit per-job skipped events.
+                        for (_name, job) in &pipeline.jobs {
+                            if job.status == JobStatus::Skipped {
+                                events.push((
+                                    round,
+                                    CiSimEvent::JobSkipped {
+                                        job_id: job.job_id.clone(),
+                                    },
+                                ));
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        // 7. Snapshot.
+        let completed_count = pipelines.values().filter(|p| p.status.is_terminal()).count();
+        snapshots.push(CiSimSnapshot {
+            active_pipelines: pipelines.len() - completed_count,
+            completed_pipelines: completed_count,
+            active_provisions: pending_provisions.len(),
+            active_jobs: running_jobs.len(),
+            provisioner_online,
+            active_instances: active_instances.len(),
+        });
+    }
+
+    CiSimTrace {
+        name: config.name,
+        events,
+        snapshots,
+        status_updates: all_status_updates,
+        num_rounds: config.num_rounds,
+        final_pipelines: pipelines.into_values().collect(),
+        provisioned_instances,
+        terminated_instances,
+    }
+}
+
+// ─── Properties ─────────────────────────────────────────────────────────────
+
+/// Every webhook eventually produces a terminal Forgejo status (success/failure/error).
+pub fn check_all_webhooks_terminate(trace: &CiSimTrace) -> bool {
+    let webhook_commits: Vec<&str> = trace
+        .events
+        .iter()
+        .filter_map(|(_, e)| match e {
+            CiSimEvent::WebhookReceived { commit_sha } => Some(commit_sha.as_str()),
+            _ => None,
+        })
+        .collect();
+
+    for sha in webhook_commits {
+        let has_terminal = trace.status_updates.iter().any(|u| {
+            u.commit_sha == sha && (u.state == "success" || u.state == "failure" || u.state == "error")
+        });
+        if !has_terminal {
+            return false;
+        }
+    }
+    true
+}
+
+/// Job DAG ordering is always respected: no job runs before its `needs`.
+pub fn check_dag_ordering(trace: &CiSimTrace) -> bool {
+    // Build a map of (pipeline_id, job_name) → round when started.
+    let mut started: HashMap<(u64, &str), usize> = HashMap::new();
+    let mut completed: HashMap<(u64, &str), usize> = HashMap::new();
+
+    for (round, event) in &trace.events {
+        match event {
+            CiSimEvent::JobStarted { job_id } => {
+                started.insert(
+                    (job_id.pipeline_id.0, job_id.job_name.as_str()),
+                    *round,
+                );
+            }
+            CiSimEvent::JobCompleted { job_id, .. } => {
+                completed.insert(
+                    (job_id.pipeline_id.0, job_id.job_name.as_str()),
+                    *round,
+                );
+            }
+            _ => {}
+        }
+    }
+
+    // For each pipeline, check that if job B needs job A, then A completed before B started.
+    for pipeline in &trace.final_pipelines {
+        for (name, job) in &pipeline.jobs {
+            if let Some(&start_round) = started.get(&(pipeline.pipeline_id.0, name.as_str())) {
+                for dep in &job.definition.needs {
+                    if let Some(&dep_complete_round) =
+                        completed.get(&(pipeline.pipeline_id.0, dep.as_str()))
+                    {
+                        if dep_complete_round > start_round {
+                            return false;
+                        }
+                    }
+                }
+            }
+        }
+    }
+    true
+}
+
+/// Every provisioned instance is eventually terminated (no resource leaks).
+pub fn check_no_instance_leaks(trace: &CiSimTrace) -> bool {
+    // Every instance that was provisioned should also be terminated.
+    for instance_id in &trace.provisioned_instances {
+        if !trace.terminated_instances.contains(instance_id) {
+            return false;
+        }
+    }
+    true
+}
+
+/// Coordinator state is bounded: active pipeline count doesn't grow unboundedly.
+pub fn check_bounded_state(trace: &CiSimTrace, max_active: usize) -> bool {
+    trace
+        .snapshots
+        .iter()
+        .all(|s| s.active_pipelines <= max_active)
+}

crates/simulation/src/lib.rs

@@ -9,3 +9,6 @@ pub mod gossip;
 
 #[cfg(feature = "dashboard")]
 pub mod dashboard;
+
+#[cfg(feature = "ci")]
+pub mod ci;

crates/simulation/tests/ci_properties.rs

@@ -0,0 +1,275 @@
+//! Property-based tests for the CI simulation.
+//!
+//! These verify invariants that should hold across all possible simulation configurations.
+
+use swactor_ci::{EventType, WebhookEvent};
+use simulation::ci::sim::{self, CiSimConfig};
+
+fn simple_yaml() -> String {
+    r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      fmt:
+        run: cargo fmt -- --check
+      test:
+        needs: [fmt]
+        run: cargo test
+"#
+    .into()
+}
+
+fn push(branch: &str, sha: &str) -> WebhookEvent {
+    WebhookEvent {
+        event_type: EventType::Push,
+        repo_owner: "user".into(),
+        repo_name: "repo".into(),
+        branch: branch.into(),
+        commit_sha: sha.into(),
+        tag: None,
+    }
+}
+
+// ─── Property: Every webhook produces a terminal status ─────────────────────
+
+#[test]
+fn property_all_webhooks_terminate_single() {
+    let config = CiSimConfig {
+        name: "prop-single".into(),
+        num_rounds: 40,
+        ci_yaml: simple_yaml(),
+        webhook_schedule: vec![(1, push("main", "sha-1"))],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+    assert!(
+        sim::check_all_webhooks_terminate(&trace),
+        "single webhook must terminate"
+    );
+}
+
+#[test]
+fn property_all_webhooks_terminate_burst() {
+    // Burst of webhooks all at once.
+    let config = CiSimConfig {
+        name: "prop-burst".into(),
+        num_rounds: 60,
+        ci_yaml: simple_yaml(),
+        webhook_schedule: (1..=5)
+            .map(|i| (1, push("main", &format!("sha-burst-{i}"))))
+            .collect(),
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+    assert!(
+        sim::check_all_webhooks_terminate(&trace),
+        "burst of 5 webhooks must all terminate"
+    );
+}
+
+#[test]
+fn property_all_webhooks_terminate_staggered() {
+    // Webhooks spread across rounds.
+    let config = CiSimConfig {
+        name: "prop-staggered".into(),
+        num_rounds: 60,
+        ci_yaml: simple_yaml(),
+        webhook_schedule: vec![
+            (1, push("main", "sha-s1")),
+            (5, push("main", "sha-s2")),
+            (10, push("main", "sha-s3")),
+            (15, push("main", "sha-s4")),
+        ],
+        provision_latency: 2,
+        job_duration: 3,
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+    assert!(
+        sim::check_all_webhooks_terminate(&trace),
+        "staggered webhooks must all terminate"
+    );
+}
+
+// ─── Property: DAG ordering is always respected ─────────────────────────────
+
+#[test]
+fn property_dag_ordering_always_respected() {
+    // Deep chain: a → b → c → d
+    let yaml = r#"
+pipelines:
+  deep:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      a:
+        run: echo a
+      b:
+        needs: [a]
+        run: echo b
+      c:
+        needs: [b]
+        run: echo c
+      d:
+        needs: [c]
+        run: echo d
+"#;
+    let config = CiSimConfig {
+        name: "prop-dag-deep".into(),
+        num_rounds: 40,
+        ci_yaml: yaml.into(),
+        webhook_schedule: vec![(1, push("main", "sha-dag"))],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+    assert!(
+        sim::check_dag_ordering(&trace),
+        "deep DAG ordering must be respected"
+    );
+}
+
+#[test]
+fn property_dag_ordering_diamond() {
+    let yaml = r#"
+pipelines:
+  diamond:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      root:
+        run: echo root
+      left:
+        needs: [root]
+        run: echo left
+      right:
+        needs: [root]
+        run: echo right
+      merge:
+        needs: [left, right]
+        run: echo merge
+"#;
+    let config = CiSimConfig {
+        name: "prop-dag-diamond".into(),
+        num_rounds: 40,
+        ci_yaml: yaml.into(),
+        webhook_schedule: vec![(1, push("main", "sha-diamond"))],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+    assert!(
+        sim::check_dag_ordering(&trace),
+        "diamond DAG ordering must be respected"
+    );
+}
+
+// ─── Property: No instance leaks ────────────────────────────────────────────
+
+#[test]
+fn property_no_instance_leaks_under_failures() {
+    let config = CiSimConfig {
+        name: "prop-no-leaks-fail".into(),
+        num_rounds: 40,
+        ci_yaml: simple_yaml(),
+        webhook_schedule: vec![
+            (1, push("main", "sha-leak1")),
+            (3, push("main", "sha-leak2")),
+        ],
+        provision_latency: 1,
+        job_duration: 2,
+        job_failure_schedule: vec![(0, "fmt".into())],
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+    assert!(
+        sim::check_no_instance_leaks(&trace),
+        "no instance leaks even when jobs fail"
+    );
+}
+
+#[test]
+fn property_no_instance_leaks_under_interruption() {
+    let yaml = r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      test:
+        run: cargo test
+"#;
+    let config = CiSimConfig {
+        name: "prop-no-leaks-interrupt".into(),
+        num_rounds: 40,
+        ci_yaml: yaml.into(),
+        webhook_schedule: vec![(1, push("main", "sha-int"))],
+        provision_latency: 1,
+        job_duration: 5,
+        instance_interrupt_schedule: vec![(4, "test".into())],
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+    assert!(
+        sim::check_no_instance_leaks(&trace),
+        "interrupted instances must be terminated"
+    );
+}
+
+// ─── Property: Bounded state ────────────────────────────────────────────────
+
+#[test]
+fn property_bounded_state_under_rapid_pushes() {
+    let config = CiSimConfig {
+        name: "prop-bounded".into(),
+        num_rounds: 100,
+        ci_yaml: simple_yaml(),
+        webhook_schedule: (1..=20)
+            .map(|i| (i, push("main", &format!("sha-rapid-{i}"))))
+            .collect(),
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+
+    // With 20 pushes, each triggering 1 pipeline with 2 jobs, we should never
+    // have more than 20 active pipelines at once (and in practice much fewer).
+    assert!(
+        sim::check_bounded_state(&trace, 20),
+        "active pipelines should be bounded"
+    );
+}
+
+// ─── Property: Provisioner offline doesn't lose work ────────────────────────
+
+#[test]
+fn property_provisioner_offline_eventually_resolves() {
+    let config = CiSimConfig {
+        name: "prop-offline-resolve".into(),
+        num_rounds: 60,
+        ci_yaml: simple_yaml(),
+        webhook_schedule: vec![(3, push("main", "sha-offline"))],
+        provision_latency: 1,
+        job_duration: 2,
+        provisioner_offline_schedule: vec![(1, 15)],
+        ..Default::default()
+    };
+    let trace = sim::run_simulation(config);
+    assert!(
+        sim::check_all_webhooks_terminate(&trace),
+        "webhooks during provisioner outage must still terminate"
+    );
+}

crates/simulation/tests/ci_scenarios.rs

@@ -0,0 +1,368 @@
+//! Scenario tests for the CI simulation.
+//!
+//! Each test tells a story: set up a scenario, run the simulation, verify outcomes.
+
+use swactor_ci::{EventType, WebhookEvent};
+use simulation::ci::sim::{self, CiSimConfig, CiSimEvent};
+
+fn basic_ci_yaml() -> String {
+    r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+        exclude: ["master"]
+    jobs:
+      fmt:
+        run: cargo fmt -- --check
+      clippy:
+        run: cargo clippy
+      test:
+        needs: [fmt, clippy]
+        run: cargo test
+  full:
+    triggers:
+      - event: push
+        branches: ["master"]
+    jobs:
+      test:
+        run: cargo test --all-features
+        timeout: 600
+      bench:
+        needs: [test]
+        run: cargo bench
+"#
+    .into()
+}
+
+fn push_event(branch: &str, sha: &str) -> WebhookEvent {
+    WebhookEvent {
+        event_type: EventType::Push,
+        repo_owner: "user".into(),
+        repo_name: "repo".into(),
+        branch: branch.into(),
+        commit_sha: sha.into(),
+        tag: None,
+    }
+}
+
+// ─── Scenario: Single push triggers correct pipeline ────────────────────────
+
+#[test]
+fn single_push_to_feature_branch_triggers_check_pipeline() {
+    let config = CiSimConfig {
+        name: "single-push-feature".into(),
+        num_rounds: 30,
+        ci_yaml: basic_ci_yaml(),
+        webhook_schedule: vec![(1, push_event("feature-x", "abc123"))],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    // A pipeline was created.
+    let pipeline_created = trace
+        .events
+        .iter()
+        .filter(|(_, e)| matches!(e, CiSimEvent::PipelineCreated { .. }))
+        .count();
+    assert_eq!(pipeline_created, 1, "exactly one pipeline should be created");
+
+    // The pipeline name should be "check" (not "full", since branch is not master).
+    let check_created = trace.events.iter().any(|(_, e)| {
+        matches!(e, CiSimEvent::PipelineCreated { name, .. } if name == "check")
+    });
+    assert!(check_created, "pipeline 'check' should be created");
+
+    // All jobs eventually complete (fmt, clippy, test).
+    let completed_jobs: Vec<_> = trace
+        .events
+        .iter()
+        .filter_map(|(_, e)| match e {
+            CiSimEvent::JobCompleted { job_id, passed } => Some((job_id.job_name.clone(), *passed)),
+            _ => None,
+        })
+        .collect();
+
+    assert_eq!(completed_jobs.len(), 3, "all 3 jobs should complete");
+    assert!(
+        completed_jobs.iter().all(|(_, passed)| *passed),
+        "all jobs should pass"
+    );
+
+    // A terminal Forgejo status is emitted.
+    assert!(
+        sim::check_all_webhooks_terminate(&trace),
+        "webhook should produce terminal status"
+    );
+}
+
+// ─── Scenario: Push to master triggers full pipeline ────────────────────────
+
+#[test]
+fn push_to_master_triggers_full_pipeline() {
+    let config = CiSimConfig {
+        name: "push-master".into(),
+        num_rounds: 30,
+        ci_yaml: basic_ci_yaml(),
+        webhook_schedule: vec![(1, push_event("master", "def456"))],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    let full_created = trace.events.iter().any(|(_, e)| {
+        matches!(e, CiSimEvent::PipelineCreated { name, .. } if name == "full")
+    });
+    assert!(full_created, "pipeline 'full' should be created");
+
+    // Both test and bench should eventually complete.
+    let completed: Vec<String> = trace
+        .events
+        .iter()
+        .filter_map(|(_, e)| match e {
+            CiSimEvent::JobCompleted { job_id, .. } => Some(job_id.job_name.clone()),
+            _ => None,
+        })
+        .collect();
+
+    assert!(completed.contains(&"test".to_string()), "test should complete");
+    assert!(completed.contains(&"bench".to_string()), "bench should complete");
+}
+
+// ─── Scenario: Jobs execute in DAG order ────────────────────────────────────
+
+#[test]
+fn jobs_execute_in_dag_order() {
+    let config = CiSimConfig {
+        name: "dag-order".into(),
+        num_rounds: 30,
+        ci_yaml: basic_ci_yaml(),
+        webhook_schedule: vec![(1, push_event("feature-y", "aaa111"))],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    // test should start after fmt and clippy complete.
+    assert!(
+        sim::check_dag_ordering(&trace),
+        "DAG ordering must be respected"
+    );
+}
+
+// ─── Scenario: Job failure skips dependents ─────────────────────────────────
+
+#[test]
+fn job_failure_skips_downstream_dependents() {
+    let config = CiSimConfig {
+        name: "failure-skip".into(),
+        num_rounds: 30,
+        ci_yaml: basic_ci_yaml(),
+        webhook_schedule: vec![(1, push_event("feature-z", "bbb222"))],
+        provision_latency: 1,
+        job_duration: 2,
+        // fmt will fail, so test (which needs fmt) should be skipped.
+        job_failure_schedule: vec![(0, "fmt".into())],
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    // test should be skipped.
+    let test_skipped = trace.events.iter().any(|(_, e)| {
+        matches!(e, CiSimEvent::JobSkipped { job_id } if job_id.job_name == "test")
+    });
+    assert!(test_skipped, "test job should be skipped when fmt fails");
+
+    // Pipeline should be marked as failed.
+    let pipeline_failed = trace.status_updates.iter().any(|u| {
+        u.context == "ci/check" && u.state == "failure"
+    });
+    assert!(pipeline_failed, "pipeline should report failure status");
+}
+
+// ─── Scenario: Parallel pushes execute independently ────────────────────────
+
+#[test]
+fn parallel_pushes_execute_independently() {
+    let config = CiSimConfig {
+        name: "parallel-pushes".into(),
+        num_rounds: 40,
+        ci_yaml: basic_ci_yaml(),
+        webhook_schedule: vec![
+            (1, push_event("feature-a", "ccc333")),
+            (1, push_event("feature-b", "ddd444")),
+        ],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    // Two pipelines should be created.
+    let pipeline_count = trace
+        .events
+        .iter()
+        .filter(|(_, e)| matches!(e, CiSimEvent::PipelineCreated { .. }))
+        .count();
+    assert_eq!(pipeline_count, 2, "two pipelines should be created");
+
+    // Both should have terminal status.
+    assert!(
+        sim::check_all_webhooks_terminate(&trace),
+        "both webhooks should produce terminal statuses"
+    );
+}
+
+// ─── Scenario: Provisioner goes offline, jobs queue and resume ──────────────
+
+#[test]
+fn provisioner_offline_queues_then_resumes() {
+    let config = CiSimConfig {
+        name: "provisioner-offline".into(),
+        num_rounds: 50,
+        ci_yaml: basic_ci_yaml(),
+        // Push at round 3, provisioner offline rounds 1-10.
+        webhook_schedule: vec![(3, push_event("feature-q", "eee555"))],
+        provision_latency: 1,
+        job_duration: 2,
+        provisioner_offline_schedule: vec![(1, 10)],
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    // Provisioner went offline and came back.
+    let went_offline = trace
+        .events
+        .iter()
+        .any(|(_, e)| matches!(e, CiSimEvent::ProvisionerWentOffline));
+    let came_online = trace
+        .events
+        .iter()
+        .any(|(_, e)| matches!(e, CiSimEvent::ProvisionerCameOnline));
+    assert!(went_offline, "provisioner should go offline");
+    assert!(came_online, "provisioner should come back online");
+
+    // Despite the outage, all jobs should eventually complete.
+    assert!(
+        sim::check_all_webhooks_terminate(&trace),
+        "pipeline should complete after provisioner returns"
+    );
+}
+
+// ─── Scenario: Spot instance interrupted mid-job ────────────────────────────
+
+#[test]
+fn spot_instance_interruption_reports_failure() {
+    // Use a simpler pipeline so we have a clear target to interrupt.
+    let yaml = r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      test:
+        run: cargo test
+"#;
+    let config = CiSimConfig {
+        name: "spot-interrupt".into(),
+        num_rounds: 30,
+        ci_yaml: yaml.into(),
+        webhook_schedule: vec![(1, push_event("main", "fff666"))],
+        provision_latency: 1,
+        job_duration: 5,
+        // Interrupt the test job at round 4 (while it's still running).
+        instance_interrupt_schedule: vec![(4, "test".into())],
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    // The pipeline should be terminal (failed due to interruption).
+    let has_failure_status = trace
+        .status_updates
+        .iter()
+        .any(|u| u.state == "failure" || u.state == "error");
+    assert!(
+        has_failure_status,
+        "interrupted job should produce a failure status"
+    );
+
+    // The instance should be terminated.
+    assert!(
+        sim::check_no_instance_leaks(&trace),
+        "interrupted instance should be terminated"
+    );
+}
+
+// ─── Scenario: All jobs pass → pipeline success ─────────────────────────────
+
+#[test]
+fn all_jobs_pass_marks_pipeline_success() {
+    let yaml = r#"
+pipelines:
+  check:
+    triggers:
+      - event: push
+        branches: ["*"]
+    jobs:
+      lint:
+        run: cargo clippy
+      test:
+        needs: [lint]
+        run: cargo test
+"#;
+    let config = CiSimConfig {
+        name: "all-pass".into(),
+        num_rounds: 30,
+        ci_yaml: yaml.into(),
+        webhook_schedule: vec![(1, push_event("main", "ggg777"))],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    let has_success = trace
+        .status_updates
+        .iter()
+        .any(|u| u.state == "success" && u.context == "ci/check");
+    assert!(has_success, "pipeline should be marked as success");
+}
+
+// ─── Scenario: Every provisioned instance is terminated ─────────────────────
+
+#[test]
+fn no_instance_resource_leaks() {
+    let config = CiSimConfig {
+        name: "no-leaks".into(),
+        num_rounds: 40,
+        ci_yaml: basic_ci_yaml(),
+        webhook_schedule: vec![
+            (1, push_event("feature-1", "h1")),
+            (5, push_event("feature-2", "h2")),
+        ],
+        provision_latency: 1,
+        job_duration: 2,
+        ..Default::default()
+    };
+
+    let trace = sim::run_simulation(config);
+
+    assert!(
+        sim::check_no_instance_leaks(&trace),
+        "all provisioned instances must be terminated"
+    );
+}