feat(runtime): complete managed Fleet launch and replay

Add explicit local Fleet preparation/start APIs, named role and Workflow metadata, durable privacy-bounded replay/SSE, and per-worker controls. Keep managed launch collision-safe with run-scoped worker identities, effective write-root checks, contention-aware scheduling, and compaction-safe replay epochs. Extend Runtime capabilities, SDK helpers, documentation, and regression coverage while preserving the existing CLI launch path and failing closed on unsupported targets or authority overrides.

crates/protocol/src/fleet.rs

@@ -41,6 +41,23 @@ pub struct FleetRun {
     pub id: FleetRunId,
     pub name: String,
     pub status: FleetRunStatus,
+    /// Explicit execution target selected by the managed client.
+    ///
+    /// Older CLI-created runs predate target selection and therefore omit
+    /// this field. Runtime API creation always persists it and currently
+    /// accepts only [`FleetRuntimeTarget::ThisComputer`].
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub target: Option<FleetRuntimeTarget>,
+    /// Named Workflow descriptor that owns this Fleet run.
+    ///
+    /// The durable task specs below remain the executable source of truth;
+    /// this descriptor keeps the product identity and scheduling policy
+    /// inspectable without smuggling them through labels.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub workflow: Option<FleetWorkflowDescriptor>,
+    /// Canonical named roles declared for the run.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub roles: Vec<String>,
     /// Maximum number of workers the manager may drive concurrently.
     ///
     /// Older ledgers omit this field; callers fall back to the persisted
@@ -62,6 +79,74 @@ pub struct FleetRun {
     pub completed_at: Option<String>,
 }
 
+/// Product-level Runtime target for a managed Fleet run.
+///
+/// The enum intentionally names unsupported targets as contract values so a
+/// client receives a precise capability refusal instead of silently falling
+/// back to local execution.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum FleetRuntimeTarget {
+    ThisComputer,
+    AnotherComputer,
+    Cloud,
+}
+
+/// Scheduling shape currently executable by the durable Fleet manager.
+///
+/// Fleet tasks are independent queue entries today, so only parallel
+/// workflows are advertised. Sequence/pipeline support must not be accepted
+/// until dependencies are durable in the Fleet ledger.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum FleetWorkflowKind {
+    Parallel,
+}
+
+/// Durable identity for the Workflow that coordinates a Fleet run.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct FleetWorkflowDescriptor {
+    pub id: String,
+    pub kind: FleetWorkflowKind,
+}
+
+/// One privacy-bounded durable event exposed to managed Fleet clients.
+///
+/// `cursor` is an opaque stable digest of the underlying ledger transition.
+/// Clients persist it and send it back on reconnect; they must not parse it.
+/// Worker-local sequence numbers remain available separately because they are
+/// monotonic only within one `(worker, task)` lifecycle, not across a run.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct FleetRuntimeEvent {
+    pub cursor: String,
+    pub event: String,
+    pub run_id: FleetRunId,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub worker_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub task_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub timestamp: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub worker_seq: Option<u64>,
+    #[serde(default)]
+    pub payload: Value,
+}
+
+/// Bounded durable replay page.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct FleetEventReplay {
+    pub run_id: FleetRunId,
+    pub events: Vec<FleetRuntimeEvent>,
+    #[serde(default)]
+    pub has_more: bool,
+    /// True when a no-cursor request returned only the newest bounded tail.
+    #[serde(default)]
+    pub history_truncated: bool,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub next_cursor: Option<String>,
+}
+
 /// Lifecycle status for an entire fleet run.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "snake_case")]
@@ -1050,6 +1135,12 @@ mod tests {
             id: FleetRunId::from("run-001"),
             name: "dogfood smoke".to_string(),
             status: FleetRunStatus::Running,
+            target: Some(FleetRuntimeTarget::ThisComputer),
+            workflow: Some(FleetWorkflowDescriptor {
+                id: "release-checks".to_string(),
+                kind: FleetWorkflowKind::Parallel,
+            }),
+            roles: vec!["release-checker".to_string()],
             max_workers: Some(1),
             task_specs: vec![FleetTaskSpec {
                 id: "task-1".to_string(),
@@ -1102,6 +1193,12 @@ mod tests {
         let back: FleetRun = serde_json::from_str(&json).unwrap();
         assert_eq!(back.id, run.id);
         assert_eq!(back.status, FleetRunStatus::Running);
+        assert_eq!(back.target, Some(FleetRuntimeTarget::ThisComputer));
+        assert_eq!(back.roles, vec!["release-checker"]);
+        assert_eq!(
+            back.workflow.as_ref().map(|workflow| workflow.id.as_str()),
+            Some("release-checks")
+        );
         assert_eq!(back.task_specs.len(), 1);
         assert_eq!(
             back.task_specs[0].worker.as_ref().unwrap().role.as_deref(),

crates/protocol/src/runtime/mod.rs

@@ -49,6 +49,16 @@ pub struct RuntimeCapabilities {
     pub external_tools: bool,
     pub environments: bool,
     pub worker_runtime: bool,
+    #[serde(default)]
+    pub fleet_run_create: bool,
+    #[serde(default)]
+    pub fleet_run_start: bool,
+    #[serde(default)]
+    pub fleet_event_replay: bool,
+    #[serde(default)]
+    pub fleet_event_stream: bool,
+    #[serde(default)]
+    pub fleet_local_target: bool,
 }
 
 /// Experimental opt-in flags advertised by `GET /v1/runtime/info`.
@@ -340,6 +350,11 @@ mod tests {
             external_tools: false,
             environments: false,
             worker_runtime: false,
+            fleet_run_create: true,
+            fleet_run_start: true,
+            fleet_event_replay: true,
+            fleet_event_stream: true,
+            fleet_local_target: true,
         };
         let value = serde_json::to_value(&caps).unwrap();
         let obj = value.as_object().unwrap();
@@ -347,6 +362,8 @@ mod tests {
         assert_eq!(obj.get("account_session").unwrap(), &json!(true));
         assert_eq!(obj.get("external_tools").unwrap(), &json!(false));
         assert!(obj.contains_key("worker_runtime"));
+        assert_eq!(obj.get("fleet_run_create").unwrap(), &json!(true));
+        assert_eq!(obj.get("fleet_event_stream").unwrap(), &json!(true));
     }
 
     #[test]

crates/tui/src/fleet/control.rs

@@ -695,6 +695,9 @@ mod tests {
             id: FleetRunId::from(id.to_string()),
             name: "stopship".to_string(),
             status: FleetRunStatus::Running,
+            target: None,
+            workflow: None,
+            roles: Vec::new(),
             max_workers: Some(2),
             task_specs: Vec::new(),
             worker_specs: Vec::new(),

crates/tui/src/fleet/ledger.rs

@@ -15,16 +15,41 @@ use std::path::{Path, PathBuf};
 use anyhow::{Context, Result, bail};
 use codewhale_protocol::fleet::*;
 use serde::{Deserialize, Serialize};
+use serde_json::{Value, json};
 
 const FLEET_DIR: &str = ".codewhale";
 const FLEET_LEDGER_FILE: &str = "fleet.jsonl";
 const FLEET_LEDGER_LOCK_FILE: &str = "fleet.lock";
 const PARTIAL_SUFFIX: &str = ".tmp";
 
+fn inline_secret_assignment_pattern() -> &'static regex::Regex {
+    static PATTERN: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
+    PATTERN.get_or_init(|| {
+        regex::Regex::new(
+            r#"(?i)(?:"|')?\b(api[_-]?key|apikey|secret|token|password|passwd|authorization|auth[_-]?token|access[_-]?key|client[_-]?secret|private[_-]?key)\b(?:"|')?\s*([:=])\s*(?:"[^"]*"|'[^']*'|[^\s,;}\]]+)"#,
+        )
+        .expect("Fleet inline secret redaction pattern must compile")
+    })
+}
+
+fn bearer_secret_pattern() -> &'static regex::Regex {
+    static PATTERN: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
+    PATTERN.get_or_init(|| {
+        regex::Regex::new(r"(?i)\bbearer\s+[A-Za-z0-9._~+/=-]{8,}")
+            .expect("Fleet bearer secret redaction pattern must compile")
+    })
+}
+
 /// A single append-only record in the fleet ledger.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "record", rename_all = "snake_case")]
 pub enum FleetLedgerRecord {
+    /// Replay generation rotated whenever compaction replaces durable history.
+    /// Legacy ledgers omit this record and use the stable `legacy` generation
+    /// until their first compaction.
+    ReplayEpoch {
+        epoch: String,
+    },
     RunCreated {
         // Boxed: FleetRun is by far the largest payload; boxing keeps the enum
         // small (clippy::large_enum_variant). Serde treats Box<T> as T.
@@ -191,6 +216,17 @@ pub(crate) struct FleetEventSequenceOwner {
     task_id: String,
 }
 
+/// Precise failure boundary for managed-client replay.
+#[derive(Debug, thiserror::Error)]
+pub enum FleetEventReplayError {
+    #[error("fleet run {run_id} does not exist")]
+    UnknownRun { run_id: String },
+    #[error("fleet event cursor is no longer available for run {run_id}")]
+    CursorUnavailable { run_id: String },
+    #[error("failed to read fleet event history: {message}")]
+    Storage { message: String },
+}
+
 /// Append-only JSONL ledger for fleet runs.
 #[derive(Debug)]
 pub struct FleetLedger {
@@ -1195,6 +1231,123 @@ impl FleetLedger {
         self.with_read_lock(|| self.rebuild_state_unlocked())
     }
 
+    /// Read one bounded page from the durable Fleet transition history.
+    ///
+    /// A cursor is the opaque digest of a ledger transition, not a global
+    /// worker sequence. That distinction matters because worker `seq` values
+    /// restart for each `(worker, task)` lifecycle. Recent cursors survive a
+    /// process restart and normal appends. Ledger compaction may intentionally
+    /// discard old history; in that case callers receive `CursorUnavailable`
+    /// and must reload the current run projection instead of silently skipping
+    /// an unknown gap.
+    pub fn replay_events(
+        &self,
+        run_id: &FleetRunId,
+        after: Option<&str>,
+        limit: usize,
+    ) -> std::result::Result<FleetEventReplay, FleetEventReplayError> {
+        let (run_exists, all_events, history_compacted) = self
+            .with_read_lock(|| self.scan_runtime_events_unlocked(run_id))
+            .map_err(|error| FleetEventReplayError::Storage {
+                message: error.to_string(),
+            })?;
+        if !run_exists {
+            return Err(FleetEventReplayError::UnknownRun {
+                run_id: run_id.0.clone(),
+            });
+        }
+
+        let limit = limit.clamp(1, 1_000);
+        let (events, has_more, history_truncated) = if let Some(after) = after {
+            let Some(position) = all_events.iter().position(|event| event.cursor == after) else {
+                return Err(FleetEventReplayError::CursorUnavailable {
+                    run_id: run_id.0.clone(),
+                });
+            };
+            let remaining = &all_events[position.saturating_add(1)..];
+            (
+                remaining.iter().take(limit).cloned().collect::<Vec<_>>(),
+                remaining.len() > limit,
+                false,
+            )
+        } else {
+            let start = all_events.len().saturating_sub(limit);
+            (
+                all_events[start..].to_vec(),
+                false,
+                start > 0 || history_compacted,
+            )
+        };
+        let next_cursor = events.last().map(|event| event.cursor.clone());
+        Ok(FleetEventReplay {
+            run_id: run_id.clone(),
+            events,
+            has_more,
+            history_truncated,
+            next_cursor,
+        })
+    }
+
+    fn scan_runtime_events_unlocked(
+        &self,
+        run_filter: &FleetRunId,
+    ) -> Result<(bool, Vec<FleetRuntimeEvent>, bool)> {
+        let mut state = FleetLedgerState::default();
+        let mut events = Vec::new();
+        let mut replay_epoch = "legacy".to_string();
+        let mut history_compacted = false;
+        if !self.ledger_path.exists() {
+            return Ok((false, events, history_compacted));
+        }
+        let file = std::fs::File::open(&self.ledger_path)
+            .with_context(|| format!("opening ledger {}", self.ledger_path.display()))?;
+        let reader = std::io::BufReader::new(file);
+        for (line_no, line) in reader.lines().enumerate() {
+            let line = match line {
+                Ok(line) => line,
+                Err(error) => {
+                    tracing::warn!(
+                        "fleet ledger line {} unreadable during event replay: {}",
+                        line_no + 1,
+                        error
+                    );
+                    continue;
+                }
+            };
+            if line.trim().is_empty() {
+                continue;
+            }
+            let record = match serde_json::from_str::<FleetLedgerRecord>(&line) {
+                Ok(record) => record,
+                Err(error) => {
+                    tracing::warn!(
+                        "fleet ledger line {} parse error during event replay (skipping): {}",
+                        line_no + 1,
+                        error
+                    );
+                    continue;
+                }
+            };
+            if let FleetLedgerRecord::ReplayEpoch { epoch } = &record {
+                replay_epoch.clone_from(epoch);
+                history_compacted = true;
+                apply_record(&mut state, record);
+                continue;
+            }
+            for event in runtime_events_from_record(&record, &state, line_no, &replay_epoch) {
+                if event.run_id == *run_filter {
+                    events.push(event);
+                }
+            }
+            apply_record(&mut state, record);
+        }
+        Ok((
+            state.runs.contains_key(&run_filter.0),
+            events,
+            history_compacted,
+        ))
+    }
+
     fn rebuild_state_unlocked(&self) -> Result<FleetLedgerState> {
         let mut state = FleetLedgerState::default();
         if !self.ledger_path.exists() {
@@ -1281,7 +1434,9 @@ impl FleetLedger {
             let state = self.rebuild_state_unlocked()?;
             after_snapshot();
             let tmp_path = self.ledger_path.with_extension(PARTIAL_SUFFIX);
-            let mut lines = Vec::new();
+            let mut lines = vec![serde_json::to_string(&FleetLedgerRecord::ReplayEpoch {
+                epoch: uuid::Uuid::new_v4().simple().to_string(),
+            })?];
             let mut terminal_lines = Vec::new();
             let mut lifecycle_lines = Vec::new();
             for run in state.runs.values() {
@@ -1456,6 +1611,348 @@ impl FleetLedger {
     }
 }
 
+fn runtime_events_from_record(
+    record: &FleetLedgerRecord,
+    state: &FleetLedgerState,
+    ledger_ordinal: usize,
+    replay_epoch: &str,
+) -> Vec<FleetRuntimeEvent> {
+    match record {
+        FleetLedgerRecord::ReplayEpoch { .. } => Vec::new(),
+        FleetLedgerRecord::RunCreated { run } => vec![FleetRuntimeEvent {
+            cursor: fleet_runtime_cursor(&run.id, "run_created", ledger_ordinal, replay_epoch),
+            event: "fleet.run.created".to_string(),
+            run_id: run.id.clone(),
+            worker_id: None,
+            task_id: None,
+            timestamp: Some(run.created_at.clone()),
+            worker_seq: None,
+            payload: json!({
+                "target": run.target,
+                "workflow": run.workflow,
+                "roles": run.roles,
+                "task_count": run.task_specs.len(),
+                "worker_count": run.worker_specs.len(),
+            }),
+        }],
+        FleetLedgerRecord::RunStatusChanged {
+            run_id,
+            status,
+            timestamp,
+        } => vec![FleetRuntimeEvent {
+            cursor: fleet_runtime_cursor(run_id, "run_status", ledger_ordinal, replay_epoch),
+            event: "fleet.run.status_changed".to_string(),
+            run_id: run_id.clone(),
+            worker_id: None,
+            task_id: None,
+            timestamp: Some(timestamp.clone()),
+            worker_seq: None,
+            payload: json!({ "status": status }),
+        }],
+        FleetLedgerRecord::TaskEnqueued { entry } => vec![FleetRuntimeEvent {
+            cursor: fleet_runtime_cursor(
+                &entry.run_id,
+                "task_enqueued",
+                ledger_ordinal,
+                replay_epoch,
+            ),
+            event: "fleet.task.enqueued".to_string(),
+            run_id: entry.run_id.clone(),
+            worker_id: None,
+            task_id: Some(entry.task_id.clone()),
+            timestamp: Some(entry.enqueued_at.clone()),
+            worker_seq: None,
+            payload: json!({
+                "priority": entry.priority,
+                "attempts": entry.attempts,
+            }),
+        }],
+        FleetLedgerRecord::TaskLeased {
+            run_id,
+            task_id,
+            worker_id,
+            leased_at,
+            lease_expires_at,
+        } => vec![FleetRuntimeEvent {
+            cursor: fleet_runtime_cursor(run_id, "task_leased", ledger_ordinal, replay_epoch),
+            event: "fleet.task.leased".to_string(),
+            run_id: run_id.clone(),
+            worker_id: Some(worker_id.clone()),
+            task_id: Some(task_id.clone()),
+            timestamp: Some(leased_at.clone()),
+            worker_seq: None,
+            payload: json!({ "lease_expires_at": lease_expires_at }),
+        }],
+        FleetLedgerRecord::TaskCompletedOrFailed {
+            run_id,
+            task_id,
+            worker_id,
+            timestamp,
+            status,
+        } => vec![FleetRuntimeEvent {
+            cursor: fleet_runtime_cursor(run_id, "task_terminal", ledger_ordinal, replay_epoch),
+            event: "fleet.task.terminal".to_string(),
+            run_id: run_id.clone(),
+            worker_id: (!worker_id.is_empty()).then(|| worker_id.clone()),
+            task_id: Some(task_id.clone()),
+            timestamp: Some(timestamp.clone()),
+            worker_seq: None,
+            payload: json!({ "status": status }),
+        }],
+        FleetLedgerRecord::TaskAttemptFinalized { event, receipt, .. } => vec![
+            runtime_worker_event("worker", ledger_ordinal, replay_epoch, event),
+            runtime_receipt_event("receipt", ledger_ordinal, replay_epoch, receipt),
+        ],
+        FleetLedgerRecord::EventAppended { event } => {
+            vec![runtime_worker_event(
+                "worker",
+                ledger_ordinal,
+                replay_epoch,
+                event,
+            )]
+        }
+        FleetLedgerRecord::Heartbeat {
+            worker_id,
+            timestamp,
+            cpu_percent,
+            memory_mb,
+        } => active_task_for_replay_worker(state, worker_id)
+            .map(|task| FleetRuntimeEvent {
+                cursor: fleet_runtime_cursor(
+                    &task.entry.run_id,
+                    "heartbeat",
+                    ledger_ordinal,
+                    replay_epoch,
+                ),
+                event: "fleet.worker.heartbeat".to_string(),
+                run_id: task.entry.run_id.clone(),
+                worker_id: Some(worker_id.clone()),
+                task_id: Some(task.entry.task_id.clone()),
+                timestamp: Some(timestamp.clone()),
+                worker_seq: None,
+                payload: json!({
+                    "cpu_percent": cpu_percent,
+                    "memory_mb": memory_mb,
+                }),
+            })
+            .into_iter()
+            .collect(),
+        FleetLedgerRecord::ReceiptRecorded { receipt } => {
+            vec![runtime_receipt_event(
+                "receipt",
+                ledger_ordinal,
+                replay_epoch,
+                receipt,
+            )]
+        }
+        FleetLedgerRecord::AlertSent {
+            run_id,
+            task_id,
+            timestamp,
+            worker_id,
+            attempt,
+            ..
+        } => vec![FleetRuntimeEvent {
+            cursor: fleet_runtime_cursor(run_id, "alert", ledger_ordinal, replay_epoch),
+            event: "fleet.alert.sent".to_string(),
+            run_id: run_id.clone(),
+            worker_id: worker_id.clone(),
+            task_id: Some(task_id.clone()),
+            timestamp: Some(timestamp.clone()),
+            worker_seq: None,
+            payload: json!({ "attempt": attempt }),
+        }],
+        FleetLedgerRecord::TaskLifecycleCheckpoint { .. }
+        | FleetLedgerRecord::EventSequenceCheckpoint { .. } => Vec::new(),
+    }
+}
+
+fn active_task_for_replay_worker<'a>(
+    state: &'a FleetLedgerState,
+    worker_id: &str,
+) -> Option<&'a FleetTaskState> {
+    state.tasks.values().find(|task| {
+        task.status == FleetTaskLedgerStatus::Leased && task.leased_to.as_deref() == Some(worker_id)
+    })
+}
+
+fn runtime_worker_event(
+    cursor_kind: &str,
+    ledger_ordinal: usize,
+    replay_epoch: &str,
+    event: &FleetWorkerEvent,
+) -> FleetRuntimeEvent {
+    let (event_name, payload) = privacy_bounded_worker_payload(&event.payload);
+    FleetRuntimeEvent {
+        cursor: fleet_runtime_cursor(&event.run_id, cursor_kind, ledger_ordinal, replay_epoch),
+        event: event_name,
+        run_id: event.run_id.clone(),
+        worker_id: Some(event.worker_id.clone()),
+        task_id: Some(event.task_id.clone()),
+        timestamp: Some(event.timestamp.clone()),
+        worker_seq: Some(event.seq),
+        payload,
+    }
+}
+
+fn runtime_receipt_event(
+    cursor_kind: &str,
+    ledger_ordinal: usize,
+    replay_epoch: &str,
+    receipt: &FleetReceipt,
+) -> FleetRuntimeEvent {
+    let score = receipt.score.as_ref().map(|score| {
+        json!({
+            "value": score.value,
+            "max": score.max,
+        })
+    });
+    FleetRuntimeEvent {
+        cursor: fleet_runtime_cursor(&receipt.run_id, cursor_kind, ledger_ordinal, replay_epoch),
+        event: "fleet.task.receipt_recorded".to_string(),
+        run_id: receipt.run_id.clone(),
+        worker_id: Some(receipt.worker_id.clone()),
+        task_id: Some(receipt.task_id.clone()),
+        timestamp: Some(receipt.completed_at.clone()),
+        worker_seq: receipt.terminal_seq,
+        payload: json!({
+            "attempt": receipt.attempt,
+            "result": receipt.result,
+            "failure_kind": receipt.failure_kind,
+            "score": score,
+            "artifact_kinds": receipt.artifacts.iter().map(|artifact| &artifact.kind).collect::<Vec<_>>(),
+        }),
+    }
+}
+
+fn privacy_bounded_worker_payload(payload: &FleetWorkerEventPayload) -> (String, Value) {
+    let state = match payload {
+        FleetWorkerEventPayload::Queued => "queued",
+        FleetWorkerEventPayload::Leased { .. } => "leased",
+        FleetWorkerEventPayload::Starting => "starting",
+        FleetWorkerEventPayload::Running => "running",
+        FleetWorkerEventPayload::ModelWait { .. } => "model_wait",
+        FleetWorkerEventPayload::RunningTool { .. } => "running_tool",
+        FleetWorkerEventPayload::WorkflowEvent { .. } => "workflow_event",
+        FleetWorkerEventPayload::Heartbeat { .. } => "heartbeat",
+        FleetWorkerEventPayload::Artifact(_) => "artifact",
+        FleetWorkerEventPayload::Completed { .. } => "completed",
+        FleetWorkerEventPayload::Failed { .. } => "failed",
+        FleetWorkerEventPayload::Cancelled { .. } => "cancelled",
+        FleetWorkerEventPayload::Interrupted { .. } => "interrupted",
+        FleetWorkerEventPayload::Stale { .. } => "stale",
+        FleetWorkerEventPayload::Restarted { .. } => "restarted",
+        FleetWorkerEventPayload::Escalated { .. } => "escalated",
+    };
+    let value = match payload {
+        FleetWorkerEventPayload::Leased { lease_expires_at } => {
+            json!({ "state": state, "lease_expires_at": lease_expires_at })
+        }
+        FleetWorkerEventPayload::ModelWait { model } => {
+            json!({ "state": state, "model": model })
+        }
+        FleetWorkerEventPayload::RunningTool { tool, .. } => {
+            json!({ "state": state, "tool": tool })
+        }
+        FleetWorkerEventPayload::WorkflowEvent {
+            workflow_run_id,
+            event,
+        } => json!({
+            "state": state,
+            "workflow_run_id": workflow_run_id,
+            "event_type": event.get("type").and_then(Value::as_str),
+        }),
+        FleetWorkerEventPayload::Heartbeat {
+            cpu_percent,
+            memory_mb,
+        } => json!({
+            "state": state,
+            "cpu_percent": cpu_percent,
+            "memory_mb": memory_mb,
+        }),
+        FleetWorkerEventPayload::Artifact(artifact) => json!({
+            "state": state,
+            "kind": artifact.kind,
+            "mime_type": artifact.mime_type,
+            "size_bytes": artifact.size_bytes,
+        }),
+        FleetWorkerEventPayload::Completed { exit_code, .. } => {
+            json!({ "state": state, "exit_code": exit_code })
+        }
+        FleetWorkerEventPayload::Failed {
+            reason,
+            recoverable,
+        } => json!({
+            "state": state,
+            "reason": redact_fleet_event_text(reason),
+            "recoverable": recoverable,
+        }),
+        FleetWorkerEventPayload::Stale { last_heartbeat_at } => {
+            json!({ "state": state, "last_heartbeat_at": last_heartbeat_at })
+        }
+        FleetWorkerEventPayload::Restarted { restart_count } => {
+            json!({ "state": state, "restart_count": restart_count })
+        }
+        FleetWorkerEventPayload::Escalated { channel, .. } => {
+            json!({ "state": state, "channel": channel })
+        }
+        FleetWorkerEventPayload::Queued
+        | FleetWorkerEventPayload::Starting
+        | FleetWorkerEventPayload::Running
+        | FleetWorkerEventPayload::Cancelled { .. }
+        | FleetWorkerEventPayload::Interrupted { .. } => json!({ "state": state }),
+    };
+    (format!("fleet.worker.{state}"), value)
+}
+
+fn redact_fleet_event_text(value: &str) -> String {
+    // The shared redactor deliberately treats one whole line as an assignment.
+    // Failure diagnostics often prefix an inline assignment (`provider failed:
+    // api_key=...`), so make a second token-level pass before exposing the
+    // bounded preview. Whitespace is normalized because this is a status
+    // summary, not the forensic worker log.
+    let redacted = bearer_secret_pattern()
+        .replace_all(value, "Bearer [redacted]")
+        .into_owned();
+    let redacted = inline_secret_assignment_pattern()
+        .replace_all(&redacted, "$1$2[redacted]")
+        .into_owned();
+    let redacted = codewhale_config::persistence::redact_secrets(&redacted);
+    let redacted = redacted
+        .split_whitespace()
+        .map(codewhale_config::persistence::redact_secrets)
+        .collect::<Vec<_>>()
+        .join(" ");
+    let mut chars = redacted.chars();
+    let preview = chars.by_ref().take(1_000).collect::<String>();
+    if chars.next().is_some() {
+        format!("{preview}...")
+    } else {
+        preview
+    }
+}
+
+fn fleet_runtime_cursor(
+    run_id: &FleetRunId,
+    kind: &str,
+    ledger_ordinal: usize,
+    replay_epoch: &str,
+) -> String {
+    // Cursor material is deliberately metadata-only: hashing a raw ledger
+    // record would let a caller correlate or guess secret-bearing diagnostic
+    // text. The generation rotates on compaction, while the physical JSONL
+    // ordinal remains stable across ordinary appends and process restarts.
+    let material = format!(
+        "fleet-event-v1\0{replay_epoch}\0{}\0{ledger_ordinal}\0{kind}",
+        run_id.0
+    );
+    format!(
+        "fev1_{}_{}",
+        crate::hashing::sha256_hex(material.as_bytes()),
+        kind
+    )
+}
+
 fn task_key(run_id: &str, task_id: &str) -> String {
     format!("{run_id}:{task_id}")
 }
@@ -1545,6 +2042,7 @@ fn artifact_event_key(event: &FleetWorkerEvent, artifact: &FleetArtifactRef) ->
 
 fn apply_record(state: &mut FleetLedgerState, record: FleetLedgerRecord) {
     match record {
+        FleetLedgerRecord::ReplayEpoch { .. } => {}
         FleetLedgerRecord::RunCreated { run } => {
             state.runs.insert(run.id.0.clone(), *run);
         }
@@ -1910,6 +2408,9 @@ mod tests {
             id: FleetRunId::from(id),
             name: "smoke".to_string(),
             status: FleetRunStatus::Running,
+            target: None,
+            workflow: None,
+            roles: Vec::new(),
             max_workers: None,
             task_specs: vec![],
             worker_specs: vec![],
@@ -1950,6 +2451,153 @@ mod tests {
         );
     }
 
+    #[test]
+    fn fleet_event_replay_is_durable_cursor_bounded_and_privacy_safe() {
+        let tmp = TempDir::new().unwrap();
+        let ledger = FleetLedger::open(tmp.path()).unwrap();
+        let mut run = sample_run("managed-run");
+        run.target = Some(FleetRuntimeTarget::ThisComputer);
+        run.workflow = Some(FleetWorkflowDescriptor {
+            id: "release-check".to_string(),
+            kind: FleetWorkflowKind::Parallel,
+        });
+        run.roles = vec!["reviewer".to_string()];
+        ledger.create_run(&run).unwrap();
+        ledger
+            .update_run_status(&run.id, FleetRunStatus::Paused, "2026-06-12T17:00:10Z")
+            .unwrap();
+        ledger
+            .update_run_status(&run.id, FleetRunStatus::Running, "2026-06-12T17:00:20Z")
+            .unwrap();
+        ledger
+            .enqueue(sample_entry("managed-run", "task-a"))
+            .unwrap();
+        ledger
+            .append_event(FleetWorkerEvent {
+                seq: 1,
+                run_id: run.id.clone(),
+                worker_id: "worker-1".to_string(),
+                task_id: "task-a".to_string(),
+                timestamp: "2026-06-12T17:01:00Z".to_string(),
+                payload: FleetWorkerEventPayload::Artifact(FleetArtifactRef {
+                    kind: FleetArtifactKind::Report,
+                    path: PathBuf::from(".codewhale/private/full-report.md"),
+                    checksum: Some("sha256:private-checksum".to_string()),
+                    mime_type: Some("text/markdown".to_string()),
+                    size_bytes: Some(42),
+                }),
+                extra: BTreeMap::new(),
+            })
+            .unwrap();
+        ledger
+            .record_receipt(FleetReceipt {
+                run_id: run.id.clone(),
+                task_id: "task-a".to_string(),
+                worker_id: "worker-1".to_string(),
+                attempt: Some(1),
+                terminal_seq: Some(2),
+                completed_at: "2026-06-12T17:02:00Z".to_string(),
+                result: FleetTaskResult::Fail,
+                failure_kind: Some(FleetTaskFailureKind::Task),
+                artifacts: vec![FleetArtifactRef {
+                    kind: FleetArtifactKind::Report,
+                    path: PathBuf::from(".codewhale/private/full-report.md"),
+                    checksum: Some("sha256:private-checksum".to_string()),
+                    mime_type: Some("text/markdown".to_string()),
+                    size_bytes: Some(42),
+                }],
+                score: Some(FleetScore {
+                    value: 0.0,
+                    max: Some(1.0),
+                    notes: Some("verifier note contained super-secret".to_string()),
+                }),
+                resolved_route: None,
+                effective_permissions: None,
+            })
+            .unwrap();
+        ledger
+            .append_event(FleetWorkerEvent {
+                seq: 2,
+                run_id: run.id.clone(),
+                worker_id: "worker-1".to_string(),
+                task_id: "task-a".to_string(),
+                timestamp: "2026-06-12T17:02:00Z".to_string(),
+                payload: FleetWorkerEventPayload::Failed {
+                    reason: "provider failed: api_key = super-secret; bearer sk-abcdef1234567890; authorization: Bearer abcdefghijklmno"
+                        .to_string(),
+                    recoverable: true,
+                },
+                extra: BTreeMap::new(),
+            })
+            .unwrap();
+
+        let page = ledger.replay_events(&run.id, None, 100).unwrap();
+        assert_eq!(page.run_id, run.id);
+        assert!(!page.history_truncated);
+        assert!(
+            page.events
+                .iter()
+                .any(|event| event.event == "fleet.run.created")
+        );
+        assert!(
+            page.events
+                .iter()
+                .any(|event| event.event == "fleet.worker.artifact")
+        );
+        assert!(
+            page.events
+                .iter()
+                .any(|event| event.event == "fleet.worker.failed")
+        );
+        let encoded = serde_json::to_string(&page).unwrap();
+        assert!(!encoded.contains("super-secret"));
+        assert!(!encoded.contains("sk-abcdef1234567890"));
+        assert!(!encoded.contains("abcdefghijklmno"));
+        assert!(!encoded.contains("private/full-report.md"));
+        assert!(!encoded.contains("private-checksum"));
+
+        let first_cursor = page.events[0].cursor.clone();
+        drop(ledger);
+        let reopened = FleetLedger::open(tmp.path()).unwrap();
+        let after_restart = reopened
+            .replay_events(&run.id, Some(&first_cursor), 100)
+            .unwrap();
+        assert_eq!(after_restart.events, page.events[1..]);
+        assert!(after_restart.next_cursor.is_some());
+
+        let tail = reopened.replay_events(&run.id, None, 2).unwrap();
+        assert_eq!(tail.events.len(), 2);
+        assert!(tail.history_truncated);
+        assert!(!tail.has_more);
+
+        assert!(matches!(
+            reopened.replay_events(&run.id, Some("fev1_missing_worker"), 100),
+            Err(FleetEventReplayError::CursorUnavailable { .. })
+        ));
+
+        reopened.compact().unwrap();
+        assert!(
+            matches!(
+                reopened.replay_events(&run.id, Some(&first_cursor), 100),
+                Err(FleetEventReplayError::CursorUnavailable { .. })
+            ),
+            "even a retained RunCreated transition must rotate its cursor when compaction deletes intervening history"
+        );
+        let compacted = reopened.replay_events(&run.id, None, 100).unwrap();
+        assert!(compacted.history_truncated);
+        assert!(
+            !compacted.events.is_empty(),
+            "current projection remains replayable after compaction"
+        );
+        assert!(
+            compacted
+                .events
+                .iter()
+                .all(|event| !page.events.iter().any(|old| old.cursor == event.cursor)),
+            "a compaction epoch must rotate every retained transition cursor"
+        );
+    }
+
     #[test]
     fn fleet_ledger_enqueue_and_claim() {
         let tmp = TempDir::new().unwrap();

crates/tui/src/fleet/manager.rs

@@ -23,7 +23,9 @@ use super::executor::{
     authority_envelope_for_worker, build_worker_exec_command_with_launch_spec,
 };
 use super::host::FleetHostErrorKind;
-use super::ledger::{FleetLedger, FleetLedgerState, FleetTaskLedgerStatus, FleetTaskState};
+use super::ledger::{
+    FleetEventReplayError, FleetLedger, FleetLedgerState, FleetTaskLedgerStatus, FleetTaskState,
+};
 use super::scheduler::{FleetScheduler, FleetSchedulerPolicy};
 use super::task_spec::{
     FleetTaskSpecDocument, FleetTaskVerificationInput, load_task_spec_document,
@@ -88,6 +90,18 @@ pub struct FleetRunReport {
     pub warnings: Vec<String>,
 }
 
+/// Product identity captured with a managed Fleet run.
+///
+/// CLI task-spec runs predate these fields and use the default descriptor.
+/// Runtime API callers provide all three values so target and Workflow
+/// identity remain durable and inspectable after the creating client exits.
+#[derive(Debug, Clone, Default)]
+pub struct ManagedFleetRunDescriptor {
+    pub target: Option<FleetRuntimeTarget>,
+    pub workflow: Option<FleetWorkflowDescriptor>,
+    pub roles: Vec<String>,
+}
+
 /// Durable restart transition plus the execution context a caller must drive.
 ///
 /// Restarting is intentionally split from execution so a live foreground
@@ -295,6 +309,15 @@ impl FleetManager {
         self.ledger.rebuild_state()
     }
 
+    pub fn replay_events(
+        &self,
+        run_id: &FleetRunId,
+        after: Option<&str>,
+        limit: usize,
+    ) -> std::result::Result<FleetEventReplay, FleetEventReplayError> {
+        self.ledger.replay_events(run_id, after, limit)
+    }
+
     pub fn load_task_spec(path: &Path) -> Result<FleetTaskSpecDocument> {
         load_task_spec_document(path)
     }
@@ -309,9 +332,39 @@ impl FleetManager {
     }
 
     pub fn create_run(
+        &self,
+        doc: FleetTaskSpecDocument,
+        max_workers: usize,
+    ) -> Result<FleetRunReport> {
+        let mut prepared = self.create_queued_run(doc, max_workers)?;
+        let started = self.start_run(&prepared.run_id)?;
+        prepared.leased = started.leased;
+        prepared.queued = started.queued;
+        Ok(prepared)
+    }
+
+    /// Validate and durably create a queued run without launching work.
+    ///
+    /// Managed clients use this as the first half of an explicit two-step
+    /// launch gate. The CLI's historical `fleet run` path calls `create_run`,
+    /// which immediately invokes [`Self::start_run`] to preserve compatibility.
+    pub fn create_queued_run(
+        &self,
+        doc: FleetTaskSpecDocument,
+        max_workers: usize,
+    ) -> Result<FleetRunReport> {
+        self.create_queued_run_with_descriptor(
+            doc,
+            max_workers,
+            ManagedFleetRunDescriptor::default(),
+        )
+    }
+
+    pub fn create_queued_run_with_descriptor(
         &self,
         mut doc: FleetTaskSpecDocument,
         max_workers: usize,
+        descriptor: ManagedFleetRunDescriptor,
     ) -> Result<FleetRunReport> {
         validate_task_spec_document(&doc)?;
         worker_runtime::canonicalize_fleet_task_roles(&mut doc.tasks);
@@ -347,6 +400,9 @@ impl FleetManager {
             id: run_id.clone(),
             name: doc.name.unwrap_or_else(|| run_id.0.clone()),
             status: FleetRunStatus::Queued,
+            target: descriptor.target,
+            workflow: descriptor.workflow,
+            roles: descriptor.roles,
             max_workers: Some(max_workers),
             task_specs: doc.tasks.clone(),
             worker_specs: doc.workers.clone(),
@@ -367,27 +423,89 @@ impl FleetManager {
                 attempts: 0,
             })?;
         }
-        let initial_status = if run.task_specs.is_empty() {
-            FleetRunStatus::Completed
-        } else {
-            FleetRunStatus::Running
-        };
-        self.ledger
-            .update_run_status(&run.id, initial_status, &timestamp())?;
-        let tick = self.schedule_run(&run.id, max_workers)?;
-        self.refresh_run_status(&run.id)?;
         let state = self.ledger.rebuild_state()?;
         let snapshot = self.status_from_state(Some(&run.id), &state);
         Ok(FleetRunReport {
             run_id: run.id,
             task_count: run.task_specs.len(),
-            leased: tick.leased,
+            leased: 0,
             queued: snapshot.queued,
             worker_ids: run.worker_specs.iter().map(|w| w.id.clone()).collect(),
             warnings,
         })
     }
 
+    /// Activate one durable queued run without leasing work.
+    ///
+    /// Managed clients use this transition before spawning the executor
+    /// driver, so no partial scheduling failure can strand an unowned lease.
+    /// Terminal runs are never reactivated through this method; the existing
+    /// explicit worker restart control remains separate.
+    pub fn activate_run(&self, run_id: &FleetRunId) -> Result<FleetRunReport> {
+        let state = self.ledger.rebuild_state()?;
+        let run =
+            state
+                .runs
+                .get(&run_id.0)
+                .cloned()
+                .ok_or_else(|| FleetControlError::UnknownRun {
+                    run_id: run_id.0.clone(),
+                })?;
+        let lifecycle = state
+            .run_status_overrides
+            .get(&run_id.0)
+            .unwrap_or(&run.status);
+        if matches!(
+            lifecycle,
+            FleetRunStatus::Completed | FleetRunStatus::Failed | FleetRunStatus::Cancelled
+        ) {
+            bail!("fleet run {} is already terminal ({lifecycle:?})", run_id.0);
+        }
+        if !matches!(lifecycle, FleetRunStatus::Running) {
+            self.ledger
+                .update_run_status(run_id, FleetRunStatus::Running, &timestamp())?;
+        }
+        let state = self.ledger.rebuild_state()?;
+        let snapshot = self.status_from_state(Some(run_id), &state);
+        Ok(FleetRunReport {
+            run_id: run.id,
+            task_count: run.task_specs.len(),
+            leased: 0,
+            queued: snapshot.queued,
+            worker_ids: run
+                .worker_specs
+                .iter()
+                .map(|worker| worker.id.clone())
+                .collect(),
+            warnings: Vec::new(),
+        })
+    }
+
+    /// Activate a run and lease its first worker batch for the historical CLI
+    /// path. Managed Runtime callers use [`Self::activate_run`] and start the
+    /// executor driver before it performs scheduling.
+    pub fn start_run(&self, run_id: &FleetRunId) -> Result<FleetRunReport> {
+        let mut report = self.activate_run(run_id)?;
+        let state = self.ledger.rebuild_state()?;
+        let run = state
+            .runs
+            .get(&run_id.0)
+            .ok_or_else(|| FleetControlError::UnknownRun {
+                run_id: run_id.0.clone(),
+            })?;
+        let max_workers = run
+            .max_workers
+            .unwrap_or_else(|| run.worker_specs.len().max(1))
+            .clamp(1, 128);
+        let tick = self.schedule_run(run_id, max_workers)?;
+        self.refresh_run_status(run_id)?;
+        let state = self.ledger.rebuild_state()?;
+        let snapshot = self.status_from_state(Some(run_id), &state);
+        report.leased = tick.leased;
+        report.queued = snapshot.queued;
+        Ok(report)
+    }
+
     pub fn schedule_run(&self, run_id: &FleetRunId, max_workers: usize) -> Result<FleetTickReport> {
         self.schedule_run_excluding(run_id, max_workers, &BTreeSet::new())
     }
@@ -436,9 +554,14 @@ impl FleetManager {
             let Some((entry, task_spec)) = next_enqueued_task_for_run(&state, run_id) else {
                 break;
             };
-            if self.start_worker_task(&worker_id, &entry, &task_spec, Some(max_workers))? {
-                report.leased += 1;
+            if !self.start_worker_task(&worker_id, &entry, &task_spec, Some(max_workers))? {
+                // Busy coordination or a write-claim contention leaves the
+                // task durably queued. Returning to the driver tick avoids a
+                // tight loop over unchanged state and lets live workers make
+                // progress before scheduling retries.
+                break;
             }
+            report.leased += 1;
         }
 
         self.refresh_run_status(run_id)?;
@@ -646,9 +769,26 @@ impl FleetManager {
             // Do not lease new work onto a logical worker until its executor
             // handle has been observed and forgotten below.
             let unavailable_workers = executor.worker_ids().into_iter().collect();
-            self.schedule_run_excluding(run_id, max_workers, &unavailable_workers)?;
+            let scheduling_error = self
+                .schedule_run_excluding(run_id, max_workers, &unavailable_workers)
+                .err();
             self.drive_executor_tick(run_id, executor, codewhale_binary, model)?;
             self.refresh_run_status(run_id)?;
+            if let Some(error) = scheduling_error {
+                if executor.worker_ids().is_empty() {
+                    return Err(error).with_context(|| {
+                        format!(
+                            "scheduling Fleet run {} after draining owned workers",
+                            run_id.0
+                        )
+                    });
+                }
+                tracing::warn!(
+                    run_id = %run_id.0,
+                    error = %error,
+                    "Fleet scheduling paused while already-leased workers continue"
+                );
+            }
             // A separate `fleet interrupt` process can make the ledger
             // terminal while this manager still owns a live host child. Keep
             // driving until the executor has observed that cancellation and
@@ -1157,9 +1297,16 @@ impl FleetManager {
                 let Ok(mut guard) = manager.try_write() else {
                     return Ok(false);
                 };
-                guard
-                    .preflight_worker_coordination(&sub_agent_worker)
-                    .map_err(anyhow::Error::msg)?;
+                if let Err(error) = guard.preflight_worker_coordination(&sub_agent_worker) {
+                    tracing::debug!(
+                        worker_id,
+                        run_id = %entry.run_id.0,
+                        task_id = %entry.task_id,
+                        error = %error,
+                        "Fleet worker coordination is not currently admissible"
+                    );
+                    return Ok(false);
+                }
                 Some(guard)
             }
             None => None,
@@ -2497,6 +2644,184 @@ mod tests {
         assert!(state.tasks.is_empty());
     }
 
+    #[test]
+    fn queued_creation_waits_for_explicit_idempotent_start() {
+        let tmp = TempDir::new().unwrap();
+        let coordination =
+            crate::tools::subagent::new_shared_subagent_manager(tmp.path().to_path_buf(), 4);
+        let manager = FleetManager::open(tmp.path())
+            .unwrap()
+            .with_sub_agent_manager(coordination.clone());
+        let report = manager
+            .create_queued_run_with_descriptor(
+                FleetTaskSpecDocument {
+                    name: Some("managed launch gate".to_string()),
+                    labels: BTreeMap::new(),
+                    security_policy: None,
+                    workers: Vec::new(),
+                    tasks: vec![task("task-a")],
+                },
+                1,
+                ManagedFleetRunDescriptor {
+                    target: Some(FleetRuntimeTarget::ThisComputer),
+                    workflow: Some(FleetWorkflowDescriptor {
+                        id: "managed-launch".to_string(),
+                        kind: FleetWorkflowKind::Parallel,
+                    }),
+                    roles: vec!["reviewer".to_string()],
+                },
+            )
+            .unwrap();
+
+        let queued = manager.rebuild_state().unwrap();
+        assert_eq!(queued.runs[&report.run_id.0].status, FleetRunStatus::Queued);
+        assert_eq!(
+            queued.tasks[&task_key(&report.run_id.0, "task-a")].status,
+            FleetTaskLedgerStatus::Enqueued
+        );
+        assert!(
+            coordination
+                .try_read()
+                .unwrap()
+                .list_worker_records()
+                .is_empty()
+        );
+
+        let activated = manager.activate_run(&report.run_id).unwrap();
+        assert_eq!(activated.leased, 0);
+        let activated_state = manager.rebuild_state().unwrap();
+        assert_eq!(
+            activated_state.tasks[&task_key(&report.run_id.0, "task-a")].status,
+            FleetTaskLedgerStatus::Enqueued
+        );
+        assert!(
+            coordination
+                .try_read()
+                .unwrap()
+                .list_worker_records()
+                .is_empty()
+        );
+
+        let started = manager.start_run(&report.run_id).unwrap();
+        assert_eq!(started.leased, 1);
+        let running = manager.rebuild_state().unwrap();
+        let task = &running.tasks[&task_key(&report.run_id.0, "task-a")];
+        assert_eq!(task.status, FleetTaskLedgerStatus::Leased);
+        assert_eq!(task.entry.attempts, 1);
+        assert_eq!(
+            running.run_status_overrides[&report.run_id.0],
+            FleetRunStatus::Running
+        );
+        assert_eq!(
+            coordination.try_read().unwrap().list_worker_records().len(),
+            1
+        );
+
+        let repeated = manager.start_run(&report.run_id).unwrap();
+        assert_eq!(repeated.leased, 0);
+        let repeated_state = manager.rebuild_state().unwrap();
+        assert_eq!(
+            repeated_state.tasks[&task_key(&report.run_id.0, "task-a")]
+                .entry
+                .attempts,
+            1
+        );
+        assert_eq!(
+            coordination.try_read().unwrap().list_worker_records().len(),
+            1
+        );
+    }
+
+    #[test]
+    fn busy_coordination_yields_without_spinning_or_leasing() {
+        let tmp = TempDir::new().unwrap();
+        let coordination =
+            crate::tools::subagent::new_shared_subagent_manager(tmp.path().to_path_buf(), 4);
+        let manager = FleetManager::open(tmp.path())
+            .unwrap()
+            .with_sub_agent_manager(coordination.clone());
+        let prepared = manager
+            .create_queued_run(
+                FleetTaskSpecDocument {
+                    name: Some("busy coordination".to_string()),
+                    labels: BTreeMap::new(),
+                    security_policy: None,
+                    workers: Vec::new(),
+                    tasks: vec![task("task-a")],
+                },
+                1,
+            )
+            .unwrap();
+
+        let guard = coordination.try_write().unwrap();
+        let blocked = manager.start_run(&prepared.run_id).unwrap();
+        assert_eq!(blocked.leased, 0);
+        let blocked_state = manager.rebuild_state().unwrap();
+        assert_eq!(
+            blocked_state.tasks[&task_key(&prepared.run_id.0, "task-a")].status,
+            FleetTaskLedgerStatus::Enqueued
+        );
+
+        drop(guard);
+        let retried = manager.start_run(&prepared.run_id).unwrap();
+        assert_eq!(retried.leased, 1);
+    }
+
+    #[test]
+    fn cross_run_write_contention_leaves_later_work_queued_until_claim_releases() {
+        let tmp = TempDir::new().unwrap();
+        let coordination =
+            crate::tools::subagent::new_shared_subagent_manager(tmp.path().to_path_buf(), 4);
+        let manager = FleetManager::open(tmp.path())
+            .unwrap()
+            .with_sub_agent_manager(coordination);
+        let write_task = |id: &str| {
+            let mut task = task(id);
+            task.worker = Some(FleetTaskWorkerProfile {
+                agent_profile: None,
+                role: Some("builder".to_string()),
+                loadout: None,
+                model_class: None,
+                model: None,
+                tool_profile: Some("explicit".to_string()),
+                tools: vec!["apply_patch".to_string()],
+                capabilities: Vec::new(),
+            });
+            task.workspace = Some(FleetWorkspaceRequirements {
+                writable_paths: vec![PathBuf::from("src")],
+                ..FleetWorkspaceRequirements::default()
+            });
+            task
+        };
+        let prepare = |name: &str, task_id: &str| {
+            manager
+                .create_queued_run(
+                    FleetTaskSpecDocument {
+                        name: Some(name.to_string()),
+                        labels: BTreeMap::new(),
+                        security_policy: None,
+                        workers: Vec::new(),
+                        tasks: vec![write_task(task_id)],
+                    },
+                    1,
+                )
+                .unwrap()
+        };
+        let first = prepare("first writer", "write-a");
+        let second = prepare("second writer", "write-b");
+
+        assert_eq!(manager.start_run(&first.run_id).unwrap().leased, 1);
+        let blocked = manager.start_run(&second.run_id).unwrap();
+        assert_eq!(blocked.leased, 0);
+        assert_eq!(
+            manager.rebuild_state().unwrap().tasks[&task_key(&second.run_id.0, "write-b")].status,
+            FleetTaskLedgerStatus::Enqueued
+        );
+
+        assert_eq!(manager.stop_run(&first.run_id).unwrap(), 1);
+        assert_eq!(manager.start_run(&second.run_id).unwrap().leased, 1);
+    }
+
     #[test]
     fn restored_lease_without_launch_record_fails_durably_instead_of_poisoning_ticks() {
         let tmp = TempDir::new().unwrap();
@@ -2751,6 +3076,9 @@ mod tests {
                 id: run_id.clone(),
                 name: "resume smoke".to_string(),
                 status: FleetRunStatus::Running,
+                target: None,
+                workflow: None,
+                roles: Vec::new(),
                 max_workers: Some(workers.len().max(1)),
                 task_specs: tasks.to_vec(),
                 worker_specs: workers.to_vec(),

crates/tui/src/fleet/scheduler.rs

@@ -756,6 +756,9 @@ mod tests {
                 id: run_id.clone(),
                 name: "scheduler smoke".to_string(),
                 status: FleetRunStatus::Queued,
+                target: None,
+                workflow: None,
+                roles: Vec::new(),
                 max_workers: Some(workers),
                 task_specs: tasks.clone(),
                 worker_specs: (1..=workers)

crates/tui/src/fleet/worker_runtime.rs

@@ -270,7 +270,7 @@ pub fn fleet_task_to_worker_spec_with_profiles(
     })
 }
 
-fn fleet_write_roots(task_spec: &FleetTaskSpec) -> Result<Vec<String>> {
+pub(crate) fn fleet_write_roots(task_spec: &FleetTaskSpec) -> Result<Vec<String>> {
     let task_root = normalize_fleet_relative_path(
         task_spec
             .workspace

crates/tui/src/runtime_api.rs

@@ -1,5 +1,6 @@
 //! Runtime HTTP/SSE API for local Codewhale automation.
 
+use std::collections::{BTreeMap, BTreeSet};
 use std::convert::Infallible;
 use std::fs;
 use std::net::{IpAddr, SocketAddr, UdpSocket};
@@ -46,10 +47,14 @@ use crate::config::{
     ApiProvider, Config, DEFAULT_TEXT_MODEL, normalize_model_name_for_provider, validate_route,
 };
 use crate::fleet::executor::{FleetExecutor, configured_codewhale_binary};
-use crate::fleet::ledger::{FleetLedgerState, FleetTaskLedgerStatus};
+use crate::fleet::ledger::{FleetEventReplayError, FleetLedgerState, FleetTaskLedgerStatus};
 use crate::fleet::manager::{
     FleetManager, FleetStatusSnapshot, FleetWorkerInspection, FleetWorkerRuntimeProjection,
+    ManagedFleetRunDescriptor,
 };
+use crate::fleet::profile::canonical_public_role_name;
+use crate::fleet::task_spec::FleetTaskSpecDocument;
+use crate::fleet::worker_runtime::fleet_write_roots;
 use crate::mcp::McpPool;
 #[cfg(test)]
 pub(super) use crate::models::{ContentBlock, Message};
@@ -73,7 +78,9 @@ use crate::tools::subagent::{
     new_shared_subagent_manager_with_timeout,
 };
 use codewhale_protocol::fleet::{
-    FleetArtifactKind, FleetRun, FleetRunId, FleetWorkerEventPayload, FleetWorkerStatus,
+    FleetArtifactKind, FleetEventReplay, FleetRun, FleetRunId, FleetRuntimeEvent,
+    FleetRuntimeTarget, FleetSecurityPolicy, FleetTaskSpec, FleetWorkerEventPayload,
+    FleetWorkerSpec, FleetWorkerStatus, FleetWorkflowDescriptor, FleetWorkflowKind,
 };
 
 mod auth;
@@ -388,6 +395,11 @@ fn default_runtime_capabilities() -> RuntimeCapabilities {
         external_tools: true,
         environments: false,
         worker_runtime: true,
+        fleet_run_create: true,
+        fleet_run_start: true,
+        fleet_event_replay: true,
+        fleet_event_stream: true,
+        fleet_local_target: true,
     }
 }
 
@@ -452,6 +464,50 @@ struct ThreadEventsQuery {
     replay_limit: Option<usize>,
 }
 
+const DEFAULT_FLEET_EVENT_REPLAY_LIMIT: usize = 250;
+const MAX_FLEET_EVENT_REPLAY_LIMIT: usize = 1_000;
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct CreateFleetRunRequest {
+    #[serde(default)]
+    name: Option<String>,
+    target: FleetRuntimeTarget,
+    roles: Vec<ManagedFleetRoleRequest>,
+    workflow: ManagedFleetWorkflowRequest,
+    #[serde(default, alias = "workers")]
+    worker_specs: Vec<FleetWorkerSpec>,
+    #[serde(default)]
+    labels: BTreeMap<String, String>,
+    #[serde(default)]
+    security_policy: Option<FleetSecurityPolicy>,
+    #[serde(default)]
+    max_workers: Option<usize>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ManagedFleetRoleRequest {
+    name: String,
+    #[serde(default)]
+    agent_profile: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ManagedFleetWorkflowRequest {
+    id: String,
+    kind: FleetWorkflowKind,
+    #[serde(alias = "task_specs")]
+    tasks: Vec<FleetTaskSpec>,
+}
+
+#[derive(Debug, Deserialize)]
+struct FleetEventsQuery {
+    after: Option<String>,
+    limit: Option<usize>,
+}
+
 #[derive(Debug, Serialize)]
 struct StartTurnResponse {
     thread: ThreadRecord,
@@ -641,18 +697,31 @@ pub fn build_router(state: RuntimeApiState) -> Router {
         .route("/v1/workspace/status", get(workspace_status))
         .route("/v1/agent-runs", get(list_agent_runs))
         .route("/v1/agent-runs/{run_id}", get(get_agent_run))
-        .route("/v1/fleet/runs", get(list_fleet_runs))
+        .route(
+            "/v1/fleet/runs",
+            get(list_fleet_runs).post(create_fleet_run),
+        )
         .route("/v1/fleet/runs/{run_id}", get(get_fleet_run))
         .route(
             "/v1/fleet/runs/{run_id}/workers",
             get(list_fleet_run_workers),
         )
+        .route("/v1/fleet/runs/{run_id}/start", post(start_fleet_run))
+        .route("/v1/fleet/runs/{run_id}/events", get(stream_fleet_events))
+        .route(
+            "/v1/fleet/runs/{run_id}/events/replay",
+            get(replay_fleet_events),
+        )
         .route("/v1/fleet/runs/{run_id}/stop", post(stop_fleet_run))
         .route("/v1/fleet/workers/{worker_id}", get(get_fleet_worker))
         .route(
             "/v1/fleet/workers/{worker_id}/interrupt",
             post(interrupt_fleet_worker),
         )
+        .route(
+            "/v1/fleet/workers/{worker_id}/stop",
+            post(stop_fleet_worker),
+        )
         .route(
             "/v1/fleet/workers/{worker_id}/restart",
             post(restart_fleet_worker),
@@ -1012,6 +1081,454 @@ async fn get_agent_run(
     Ok(Json(run))
 }
 
+async fn create_fleet_run(
+    State(state): State<RuntimeApiState>,
+    Json(request): Json<CreateFleetRunRequest>,
+) -> Result<(StatusCode, Json<Value>), ApiError> {
+    if request.target != FleetRuntimeTarget::ThisComputer {
+        return Err(ApiError::not_implemented(format!(
+            "Fleet target {:?} is not available in this local Runtime; choose this_computer",
+            request.target
+        )));
+    }
+    let (document, descriptor, max_workers) = prepare_managed_fleet_run(request)?;
+    let manager = open_fleet_manager(&state)?;
+    let report = manager
+        .create_queued_run_with_descriptor(document, max_workers, descriptor)
+        .map_err(|error| ApiError::bad_request(format!("Failed to create Fleet run: {error}")))?;
+    let ledger_state = manager
+        .rebuild_state()
+        .map_err(|error| ApiError::internal(format!("Failed to rebuild Fleet state: {error}")))?;
+    let run = ledger_state
+        .runs
+        .get(&report.run_id.0)
+        .ok_or_else(|| ApiError::internal("Created Fleet run was missing from its ledger"))?;
+    Ok((
+        StatusCode::CREATED,
+        Json(json!({
+            "execution": "awaiting_start",
+            "run": fleet_run_detail_json(&manager, run, &ledger_state)?,
+            "warnings": report.warnings,
+        })),
+    ))
+}
+
+fn prepare_managed_fleet_run(
+    request: CreateFleetRunRequest,
+) -> Result<(FleetTaskSpecDocument, ManagedFleetRunDescriptor, usize), ApiError> {
+    if request.security_policy.is_some() {
+        return Err(ApiError::not_implemented(
+            "Managed Fleet security_policy overrides are not executable yet; use named roles and bounded task workspace/tool scopes",
+        ));
+    }
+    if !request.worker_specs.is_empty() {
+        return Err(ApiError::not_implemented(
+            "Managed Fleet custom worker_specs are not available yet; local Runtime worker IDs are generated per run so worker controls cannot collide across Fleets",
+        ));
+    }
+    if request.roles.is_empty() {
+        return Err(ApiError::bad_request(
+            "roles must declare at least one named Fleet role",
+        ));
+    }
+    if request.roles.len() > 128 {
+        return Err(ApiError::bad_request(
+            "roles cannot contain more than 128 entries",
+        ));
+    }
+    let workflow_id = managed_fleet_token("workflow.id", &request.workflow.id)?;
+    let workflow_kind = request.workflow.kind;
+    let name = request
+        .name
+        .as_deref()
+        .map(str::trim)
+        .filter(|name| !name.is_empty())
+        .unwrap_or(workflow_id.as_str())
+        .to_string();
+    if name.len() > 256 || name.chars().any(char::is_control) {
+        return Err(ApiError::bad_request(
+            "name must be one printable line no longer than 256 bytes",
+        ));
+    }
+
+    let mut roles = BTreeMap::new();
+    for role in request.roles {
+        let normalized = canonical_public_role_name(&managed_fleet_token("role.name", &role.name)?);
+        let agent_profile = role
+            .agent_profile
+            .as_deref()
+            .map(|profile| managed_fleet_token("role.agent_profile", profile))
+            .transpose()?;
+        if roles.insert(normalized.clone(), agent_profile).is_some() {
+            return Err(ApiError::bad_request(format!(
+                "duplicate Fleet role '{normalized}'"
+            )));
+        }
+    }
+
+    let mut tasks = request.workflow.tasks;
+    let mut used_roles = BTreeSet::new();
+    for task in &mut tasks {
+        let worker = task.worker.as_mut().ok_or_else(|| {
+            ApiError::bad_request(format!(
+                "Fleet task '{}' must select one named role through worker.role",
+                task.id
+            ))
+        })?;
+        let role = worker.role.as_deref().ok_or_else(|| {
+            ApiError::bad_request(format!(
+                "Fleet task '{}' must select one named role through worker.role",
+                task.id
+            ))
+        })?;
+        let role = canonical_public_role_name(&managed_fleet_token("task.worker.role", role)?);
+        let declared_profile = roles.get(&role).ok_or_else(|| {
+            ApiError::bad_request(format!(
+                "Fleet task '{}' references undeclared role '{role}'",
+                task.id
+            ))
+        })?;
+        if let Some(profile) = declared_profile {
+            match worker.agent_profile.as_deref() {
+                Some(task_profile) if task_profile != profile => {
+                    return Err(ApiError::bad_request(format!(
+                        "Fleet task '{}' overrides role '{role}' agent_profile '{profile}' with '{task_profile}'",
+                        task.id
+                    )));
+                }
+                None => worker.agent_profile = Some(profile.clone()),
+                Some(_) => {}
+            }
+        }
+        worker.role = Some(role.clone());
+        used_roles.insert(role);
+    }
+    let unused_roles = roles
+        .keys()
+        .filter(|role| !used_roles.contains(*role))
+        .cloned()
+        .collect::<Vec<_>>();
+    if !unused_roles.is_empty() {
+        return Err(ApiError::bad_request(format!(
+            "Every declared Fleet role must own a Workflow task; unused roles: {}",
+            unused_roles.join(", ")
+        )));
+    }
+    reject_parallel_write_collisions(&tasks)?;
+
+    let default_workers = roles.len().min(tasks.len()).max(1);
+    let max_workers = request.max_workers.unwrap_or(default_workers);
+    if !(1..=128).contains(&max_workers) {
+        return Err(ApiError::bad_request(
+            "max_workers must be between 1 and 128",
+        ));
+    }
+    let role_names = roles.into_keys().collect::<Vec<_>>();
+    Ok((
+        FleetTaskSpecDocument {
+            name: Some(name),
+            labels: request.labels,
+            security_policy: None,
+            workers: Vec::new(),
+            tasks,
+        },
+        ManagedFleetRunDescriptor {
+            target: Some(request.target),
+            workflow: Some(FleetWorkflowDescriptor {
+                id: workflow_id,
+                kind: workflow_kind,
+            }),
+            roles: role_names,
+        },
+        max_workers,
+    ))
+}
+
+fn managed_fleet_token(field: &str, value: &str) -> Result<String, ApiError> {
+    let value = value.trim();
+    if value.is_empty()
+        || value.len() > 128
+        || !value
+            .chars()
+            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.'))
+    {
+        return Err(ApiError::bad_request(format!(
+            "{field} must be a simple ASCII token no longer than 128 bytes"
+        )));
+    }
+    Ok(value.to_string())
+}
+
+fn reject_parallel_write_collisions(tasks: &[FleetTaskSpec]) -> Result<(), ApiError> {
+    let mut claims: Vec<(String, String)> = Vec::new();
+    for task in tasks {
+        let write_roots = fleet_write_roots(task).map_err(|error| {
+            ApiError::bad_request(format!(
+                "Fleet task '{}' has an invalid write scope: {error}",
+                task.id
+            ))
+        })?;
+        for normalized in write_roots {
+            for (owner, existing) in &claims {
+                if owner != &task.id && managed_paths_overlap(existing.as_str(), &normalized) {
+                    return Err(ApiError::bad_request(format!(
+                        "Parallel Workflow write scope collision: tasks '{owner}' and '{}' both claim overlapping paths",
+                        task.id
+                    )));
+                }
+            }
+            claims.push((task.id.clone(), normalized));
+        }
+    }
+    Ok(())
+}
+
+fn managed_paths_overlap(left: &str, right: &str) -> bool {
+    left == right
+        || left
+            .strip_prefix(right)
+            .is_some_and(|suffix| suffix.starts_with('/'))
+        || right
+            .strip_prefix(left)
+            .is_some_and(|suffix| suffix.starts_with('/'))
+}
+
+async fn start_fleet_run(
+    State(state): State<RuntimeApiState>,
+    Path(run_id): Path<String>,
+) -> Result<(StatusCode, Json<Value>), ApiError> {
+    let manager = open_fleet_manager(&state)?;
+    let durable = manager
+        .rebuild_state()
+        .map_err(|error| ApiError::internal(format!("Failed to rebuild Fleet state: {error}")))?;
+    let run = durable
+        .runs
+        .get(&run_id)
+        .ok_or_else(|| ApiError::not_found(format!("fleet run '{run_id}' not found")))?;
+    match run.target {
+        Some(FleetRuntimeTarget::ThisComputer) => {}
+        Some(target) => {
+            return Err(ApiError::not_implemented(format!(
+                "Fleet target {target:?} is not available in this local Runtime"
+            )));
+        }
+        None => {
+            return Err(ApiError::bad_request(
+                "Fleet run has no explicit Runtime target and cannot be started through the managed API",
+            ));
+        }
+    }
+    if run.workflow.is_none() || run.roles.is_empty() {
+        return Err(ApiError::bad_request(
+            "Fleet run has no managed Workflow/role descriptor and cannot be started through the managed API",
+        ));
+    }
+    let run_id = FleetRunId::from(run_id);
+    let report = manager.activate_run(&run_id).map_err(|error| {
+        let message = format!("Failed to start Fleet run '{}': {error}", run_id.0);
+        if message.contains("already terminal") {
+            ApiError::conflict(message)
+        } else {
+            ApiError::bad_request(message)
+        }
+    })?;
+    let max_workers = durable
+        .runs
+        .get(&run_id.0)
+        .and_then(|run| run.max_workers)
+        .unwrap_or_else(|| report.worker_ids.len().max(1));
+    let workspace = state.workspace.clone();
+    let codewhale_binary = state.fleet_codewhale_binary.clone();
+    let execution_run_id = run_id.clone();
+    tokio::spawn(async move {
+        let mut executor = FleetExecutor::new(&workspace);
+        if let Err(error) = manager
+            .run_to_completion(
+                &execution_run_id,
+                max_workers,
+                &mut executor,
+                &codewhale_binary,
+                None,
+                Duration::from_millis(250),
+            )
+            .await
+        {
+            tracing::error!(
+                run_id = %execution_run_id.0,
+                error = %error,
+                "Runtime API Fleet manager exited with an error"
+            );
+        }
+    });
+    Ok((
+        StatusCode::ACCEPTED,
+        Json(json!({
+            "action": "start",
+            "execution": "scheduled",
+            "run_id": run_id.0,
+            "target": "this_computer",
+            "leased": report.leased,
+            "queued": report.queued,
+            "worker_ids": report.worker_ids,
+        })),
+    ))
+}
+
+async fn replay_fleet_events(
+    State(state): State<RuntimeApiState>,
+    Path(run_id): Path<String>,
+    Query(query): Query<FleetEventsQuery>,
+) -> Result<Json<FleetEventReplay>, ApiError> {
+    let (after, limit) = validate_fleet_events_query(query)?;
+    let replay = load_fleet_event_replay(state, FleetRunId::from(run_id), after, limit)
+        .await
+        .map_err(map_fleet_replay_error)?;
+    Ok(Json(replay))
+}
+
+async fn stream_fleet_events(
+    State(state): State<RuntimeApiState>,
+    Path(run_id): Path<String>,
+    Query(query): Query<FleetEventsQuery>,
+) -> Result<Sse<impl futures_util::Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
+    let (after, limit) = validate_fleet_events_query(query)?;
+    let run_id = FleetRunId::from(run_id);
+    let initial = load_fleet_event_replay(state.clone(), run_id.clone(), after.clone(), limit)
+        .await
+        .map_err(map_fleet_replay_error)?;
+    let event_stream = replay_live_fleet_events(state, run_id, after, limit, initial);
+    Ok(Sse::new(event_stream).keep_alive(
+        KeepAlive::new()
+            .interval(Duration::from_secs(15))
+            .text("keepalive"),
+    ))
+}
+
+fn replay_live_fleet_events(
+    state: RuntimeApiState,
+    run_id: FleetRunId,
+    mut after: Option<String>,
+    limit: usize,
+    initial: FleetEventReplay,
+) -> impl futures_util::Stream<Item = Result<SseEvent, Infallible>> {
+    stream! {
+        let mut page = initial;
+        loop {
+            if page.history_truncated {
+                yield Ok(sse_json(
+                    "fleet.replay.truncated",
+                    json!({
+                        "run_id": run_id.0.clone(),
+                        "reload_projection": true,
+                    }),
+                ));
+            }
+            for event in page.events {
+                after = Some(event.cursor.clone());
+                yield Ok(fleet_sse_event(&event));
+            }
+            if !page.has_more {
+                tokio::time::sleep(Duration::from_millis(250)).await;
+            }
+            match load_fleet_event_replay(
+                state.clone(),
+                run_id.clone(),
+                after.clone(),
+                limit,
+            )
+            .await
+            {
+                Ok(next) => page = next,
+                Err(FleetEventReplayError::CursorUnavailable { .. }) => {
+                    yield Ok(sse_json(
+                        "fleet.replay.cursor_unavailable",
+                        json!({
+                            "run_id": run_id.0.clone(),
+                            "reload_projection": true,
+                        }),
+                    ));
+                    return;
+                }
+                Err(error) => {
+                    tracing::warn!(
+                        run_id = %run_id.0,
+                        error = %error,
+                        "Fleet event stream stopped while reading durable history"
+                    );
+                    yield Ok(sse_json(
+                        "fleet.stream.error",
+                        json!({ "retryable": true }),
+                    ));
+                    return;
+                }
+            }
+        }
+    }
+}
+
+async fn load_fleet_event_replay(
+    state: RuntimeApiState,
+    run_id: FleetRunId,
+    after: Option<String>,
+    limit: usize,
+) -> std::result::Result<FleetEventReplay, FleetEventReplayError> {
+    tokio::task::spawn_blocking(move || {
+        let manager =
+            open_fleet_manager(&state).map_err(|error| FleetEventReplayError::Storage {
+                message: error.message,
+            })?;
+        manager.replay_events(&run_id, after.as_deref(), limit)
+    })
+    .await
+    .map_err(|error| FleetEventReplayError::Storage {
+        message: format!("Fleet replay worker failed: {error}"),
+    })?
+}
+
+fn validate_fleet_events_query(
+    query: FleetEventsQuery,
+) -> Result<(Option<String>, usize), ApiError> {
+    let after = query
+        .after
+        .map(|cursor| cursor.trim().to_string())
+        .filter(|cursor| !cursor.is_empty());
+    if after.as_deref().is_some_and(|cursor| {
+        cursor.len() > 96
+            || !cursor.starts_with("fev1_")
+            || !cursor
+                .chars()
+                .all(|ch| ch.is_ascii_alphanumeric() || ch == '_')
+    }) {
+        return Err(ApiError::bad_request(
+            "after is not a valid Fleet event cursor",
+        ));
+    }
+    let limit = query.limit.unwrap_or(DEFAULT_FLEET_EVENT_REPLAY_LIMIT);
+    if !(1..=MAX_FLEET_EVENT_REPLAY_LIMIT).contains(&limit) {
+        return Err(ApiError::bad_request(format!(
+            "limit must be between 1 and {MAX_FLEET_EVENT_REPLAY_LIMIT}"
+        )));
+    }
+    Ok((after, limit))
+}
+
+fn map_fleet_replay_error(error: FleetEventReplayError) -> ApiError {
+    let message = error.to_string();
+    match error {
+        FleetEventReplayError::UnknownRun { .. } => ApiError::not_found(message),
+        FleetEventReplayError::CursorUnavailable { .. } => ApiError::conflict(message),
+        FleetEventReplayError::Storage { .. } => ApiError::internal(message),
+    }
+}
+
+fn fleet_sse_event(event: &FleetRuntimeEvent) -> SseEvent {
+    let data = serde_json::to_string(event).unwrap_or_else(|_| "{}".to_string());
+    SseEvent::default()
+        .id(event.cursor.clone())
+        .event(event.event.clone())
+        .data(data)
+}
+
 async fn list_fleet_runs(State(state): State<RuntimeApiState>) -> Result<Json<Value>, ApiError> {
     let manager = open_fleet_manager(&state)?;
     let ledger_state = manager
@@ -1106,6 +1623,20 @@ async fn interrupt_fleet_worker(
     })))
 }
 
+async fn stop_fleet_worker(
+    State(state): State<RuntimeApiState>,
+    Path(worker_id): Path<String>,
+) -> Result<Json<Value>, ApiError> {
+    let manager = open_fleet_manager(&state)?;
+    let inspection = manager.interrupt_worker(&worker_id).map_err(|err| {
+        ApiError::bad_request(format!("Failed to stop fleet worker '{worker_id}': {err}"))
+    })?;
+    Ok(Json(json!({
+        "action": "stop",
+        "worker": fleet_worker_json(&inspection),
+    })))
+}
+
 async fn restart_fleet_worker(
     State(state): State<RuntimeApiState>,
     Path(worker_id): Path<String>,
@@ -1170,7 +1701,7 @@ async fn stop_fleet_run(
 }
 
 fn open_fleet_manager(state: &RuntimeApiState) -> Result<FleetManager, ApiError> {
-    let (exec_config, session_model, route_config) = {
+    let (exec_config, fleet_config, session_model, route_config) = {
         let config = state.config.read();
         let exec_config = config
             .fleet
@@ -1179,12 +1710,18 @@ fn open_fleet_manager(state: &RuntimeApiState) -> Result<FleetManager, ApiError>
             .unwrap_or_default();
         // The active session route is the operator: workers without a
         // task/profile model pin inherit the model the user picked in /model.
-        (exec_config, config.default_model(), config.clone())
+        (
+            exec_config,
+            config.fleet_config(),
+            config.default_model(),
+            config.clone(),
+        )
     };
     FleetManager::open(&state.workspace)
         .map(|manager| {
             manager
                 .with_exec_config(exec_config)
+                .with_fleet_config(fleet_config)
                 .with_sub_agent_manager(state.sub_agent_manager.clone())
                 .with_session_model(session_model)
                 .with_route_config(route_config)
@@ -1216,7 +1753,14 @@ fn fleet_run_summary_json(
     Ok(json!({
         "id": run.id.0.clone(),
         "name": run.name.clone(),
+        "lifecycle_status": ledger_state
+            .run_status_overrides
+            .get(&run.id.0)
+            .unwrap_or(&run.status),
         "status": fleet_status_json(&status),
+        "target": run.target,
+        "workflow": run.workflow.clone(),
+        "roles": run.roles.clone(),
         "task_count": run.task_specs.len(),
         "worker_count": run.worker_specs.len(),
         "tasks": task_statuses,
@@ -3996,6 +4540,20 @@ impl ApiError {
         }
     }
 
+    fn conflict(message: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::CONFLICT,
+            message: message.into(),
+        }
+    }
+
+    fn not_implemented(message: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::NOT_IMPLEMENTED,
+            message: message.into(),
+        }
+    }
+
     fn internal(message: impl Into<String>) -> Self {
         Self {
             status: StatusCode::INTERNAL_SERVER_ERROR,

crates/tui/src/runtime_api/tests.rs

@@ -1836,6 +1836,358 @@ async fn fleet_status_runtime_api_exposes_state_and_actions() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn managed_fleet_create_start_and_replay_are_explicit_and_durable() -> Result<()> {
+    let root = std::env::temp_dir().join(format!("codewhale-managed-fleet-{}", Uuid::new_v4()));
+    let workspace = root.join("workspace");
+    fs::create_dir_all(&workspace)?;
+    let marker = root.join("managed-worker-ran");
+    let fake_codewhale = write_fake_fleet_binary(&root, &marker)?;
+    let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, 2);
+    let sessions_dir = root.join("sessions");
+    let Some((addr, _runtime_threads, handle)) =
+        spawn_test_server_with_root_token_mobile_workspace_and_subagents(
+            root.clone(),
+            sessions_dir,
+            None,
+            false,
+            workspace,
+            Some(sub_agent_manager),
+            Some(fake_codewhale.display().to_string()),
+        )
+        .await?
+    else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+    let request = json!({
+        "name": "managed release check",
+        "target": "this_computer",
+        "roles": [
+            {"name": "reviewer"},
+            {"name": "verifier"}
+        ],
+        "workflow": {
+            "id": "release-check",
+            "kind": "parallel",
+            "tasks": [
+                {
+                    "id": "review",
+                    "name": "Review",
+                    "objective": "Review the release locally",
+                    "instructions": "Inspect the prepared release evidence.",
+                    "worker": {"role": "reviewer", "tool_profile": "read-only"}
+                },
+                {
+                    "id": "verify",
+                    "name": "Verify",
+                    "objective": "Verify the release locally",
+                    "instructions": "Verify the prepared release evidence.",
+                    "worker": {"role": "verifier", "tool_profile": "read-only"}
+                }
+            ]
+        },
+        "max_workers": 2
+    });
+
+    let created_response = client
+        .post(format!("http://{addr}/v1/fleet/runs"))
+        .json(&request)
+        .send()
+        .await?;
+    assert_eq!(created_response.status(), StatusCode::CREATED);
+    let created: serde_json::Value = created_response.json().await?;
+    assert_eq!(created["execution"], "awaiting_start");
+    assert_eq!(created["run"]["lifecycle_status"], "queued");
+    assert_eq!(created["run"]["target"], "this_computer");
+    assert_eq!(created["run"]["workflow"]["id"], "release-check");
+    assert_eq!(created["run"]["roles"], json!(["reviewer", "verifier"]));
+    assert_eq!(created["run"]["status"]["queued"], 2);
+    let first_worker_ids = created["run"]["worker_specs"]
+        .as_array()
+        .context("created Fleet response omitted generated workers")?
+        .iter()
+        .map(|worker| {
+            worker["id"]
+                .as_str()
+                .context("generated Fleet worker omitted id")
+                .map(str::to_string)
+        })
+        .collect::<Result<BTreeSet<_>>>()?;
+    assert_eq!(first_worker_ids.len(), 2);
+    assert!(
+        !marker.exists(),
+        "creating a managed Fleet must not cross the explicit launch gate"
+    );
+    let run_id = created["run"]["id"]
+        .as_str()
+        .context("created Fleet response omitted run id")?
+        .to_string();
+
+    let prepared_replay: serde_json::Value = client
+        .get(format!(
+            "http://{addr}/v1/fleet/runs/{run_id}/events/replay?limit=100"
+        ))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(prepared_replay["events"].as_array().unwrap().len(), 3);
+    assert_eq!(prepared_replay["events"][0]["event"], "fleet.run.created");
+    assert!(
+        prepared_replay["events"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .skip(1)
+            .all(|event| event["event"] == "fleet.task.enqueued")
+    );
+
+    let started_response = client
+        .post(format!("http://{addr}/v1/fleet/runs/{run_id}/start"))
+        .send()
+        .await?;
+    assert_eq!(started_response.status(), StatusCode::ACCEPTED);
+    let started: serde_json::Value = started_response.json().await?;
+    assert_eq!(started["execution"], "scheduled");
+    assert_eq!(started["leased"], 0);
+
+    let terminal = tokio::time::timeout(ci_scaled(Duration::from_secs(15)), async {
+        loop {
+            let run: serde_json::Value = client
+                .get(format!("http://{addr}/v1/fleet/runs/{run_id}"))
+                .send()
+                .await
+                .unwrap()
+                .error_for_status()
+                .unwrap()
+                .json()
+                .await
+                .unwrap();
+            if run["status"]["queued"] == 0 && run["status"]["running"] == 0 {
+                break run;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await
+    .context("managed Fleet did not reach a terminal projection")?;
+    assert_eq!(terminal["lifecycle_status"], "completed");
+    assert_eq!(terminal["status"]["completed"], 2);
+    assert!(
+        marker.is_file(),
+        "explicit start did not launch local workers"
+    );
+
+    let replay: serde_json::Value = client
+        .get(format!(
+            "http://{addr}/v1/fleet/runs/{run_id}/events/replay?limit=100"
+        ))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    let events = replay["events"]
+        .as_array()
+        .context("missing replay events")?;
+    assert!(
+        events
+            .iter()
+            .any(|event| event["event"] == "fleet.run.status_changed")
+    );
+    assert_eq!(
+        events
+            .iter()
+            .filter(|event| event["event"] == "fleet.worker.completed")
+            .count(),
+        2
+    );
+    let cursor = replay["next_cursor"]
+        .as_str()
+        .context("terminal replay omitted reconnect cursor")?;
+    let caught_up: serde_json::Value = client
+        .get(format!(
+            "http://{addr}/v1/fleet/runs/{run_id}/events/replay?after={cursor}&limit=100"
+        ))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert!(caught_up["events"].as_array().unwrap().is_empty());
+
+    let stream = client
+        .get(format!(
+            "http://{addr}/v1/fleet/runs/{run_id}/events?limit=1"
+        ))
+        .send()
+        .await?
+        .error_for_status()?;
+    assert_eq!(
+        stream
+            .headers()
+            .get(reqwest::header::CONTENT_TYPE)
+            .and_then(|value| value.to_str().ok()),
+        Some("text/event-stream")
+    );
+    drop(stream);
+
+    let second_created: serde_json::Value = client
+        .post(format!("http://{addr}/v1/fleet/runs"))
+        .json(&request)
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    let second_worker_ids = second_created["run"]["worker_specs"]
+        .as_array()
+        .context("second Fleet response omitted generated workers")?
+        .iter()
+        .map(|worker| {
+            worker["id"]
+                .as_str()
+                .context("second generated Fleet worker omitted id")
+                .map(str::to_string)
+        })
+        .collect::<Result<BTreeSet<_>>>()?;
+    assert!(
+        first_worker_ids.is_disjoint(&second_worker_ids),
+        "managed Fleets must receive run-scoped worker identities"
+    );
+
+    let unsupported = client
+        .post(format!("http://{addr}/v1/fleet/runs"))
+        .json(&json!({
+            "target": "cloud",
+            "roles": [],
+            "workflow": {"id": "cloud-run", "kind": "parallel", "tasks": []}
+        }))
+        .send()
+        .await?;
+    assert_eq!(unsupported.status(), StatusCode::NOT_IMPLEMENTED);
+
+    handle.abort();
+    Ok(())
+}
+
+#[test]
+fn managed_fleet_rejects_parallel_write_scope_collisions() {
+    let request: CreateFleetRunRequest = serde_json::from_value(json!({
+        "target": "this_computer",
+        "roles": [{"name": "builder"}, {"name": "reviewer"}],
+        "workflow": {
+            "id": "collision-check",
+            "kind": "parallel",
+            "tasks": [
+                {
+                    "id": "build",
+                    "name": "Build",
+                    "instructions": "Build the package.",
+                    "worker": {"role": "builder"},
+                    "workspace": {"root": "packages/runtime", "writable_paths": ["src"]}
+                },
+                {
+                    "id": "review",
+                    "name": "Review",
+                    "instructions": "Review the package.",
+                    "worker": {"role": "reviewer"},
+                    "workspace": {"root": "packages", "writable_paths": ["runtime/src/api"]}
+                }
+            ]
+        }
+    }))
+    .unwrap();
+
+    let error = prepare_managed_fleet_run(request).unwrap_err();
+    assert_eq!(error.status, StatusCode::BAD_REQUEST);
+    assert!(error.message.contains("write scope collision"));
+
+    let disjoint: CreateFleetRunRequest = serde_json::from_value(json!({
+        "target": "this_computer",
+        "roles": [{"name": "builder"}, {"name": "reviewer"}],
+        "workflow": {
+            "id": "disjoint-roots",
+            "kind": "parallel",
+            "tasks": [
+                {
+                    "id": "build",
+                    "name": "Build",
+                    "instructions": "Build the package.",
+                    "worker": {"role": "builder"},
+                    "workspace": {"root": "packages/one", "writable_paths": ["src"]}
+                },
+                {
+                    "id": "review",
+                    "name": "Review",
+                    "instructions": "Review the package.",
+                    "worker": {"role": "reviewer"},
+                    "workspace": {"root": "packages/two", "writable_paths": ["src"]}
+                }
+            ]
+        }
+    }))
+    .unwrap();
+    prepare_managed_fleet_run(disjoint).unwrap();
+}
+
+#[test]
+fn managed_fleet_rejects_unenforced_security_policy_overrides() {
+    let request: CreateFleetRunRequest = serde_json::from_value(json!({
+        "target": "this_computer",
+        "roles": [{"name": "reviewer"}],
+        "workflow": {
+            "id": "security-check",
+            "kind": "parallel",
+            "tasks": [{
+                "id": "review",
+                "name": "Review",
+                "instructions": "Review without widening authority.",
+                "worker": {"role": "reviewer"}
+            }]
+        },
+        "security_policy": {
+            "default_trust_level": "operator",
+            "capability_grants": [{"capability": "release"}]
+        }
+    }))
+    .unwrap();
+
+    let error = prepare_managed_fleet_run(request).unwrap_err();
+    assert_eq!(error.status, StatusCode::NOT_IMPLEMENTED);
+    assert!(error.message.contains("not executable yet"));
+}
+
+#[test]
+fn managed_fleet_rejects_caller_assigned_worker_identities() {
+    let request: CreateFleetRunRequest = serde_json::from_value(json!({
+        "target": "this_computer",
+        "roles": [{"name": "reviewer"}],
+        "workflow": {
+            "id": "worker-identity-check",
+            "kind": "parallel",
+            "tasks": [{
+                "id": "review",
+                "name": "Review",
+                "instructions": "Review without a cross-run worker identity.",
+                "worker": {"role": "reviewer"}
+            }]
+        },
+        "worker_specs": [{
+            "id": "shared-worker",
+            "name": "Shared worker",
+            "host": {"kind": "local"}
+        }]
+    }))
+    .unwrap();
+
+    let error = prepare_managed_fleet_run(request).unwrap_err();
+    assert_eq!(error.status, StatusCode::NOT_IMPLEMENTED);
+    assert!(error.message.contains("generated per run"));
+}
+
 #[test]
 fn fleet_worker_json_includes_runtime_state_projection() {
     let inspection = FleetWorkerInspection {
@@ -4900,6 +5252,11 @@ async fn runtime_info_reports_bind_state() -> Result<()> {
     assert_eq!(info["capabilities"]["account_session"], true);
     assert_eq!(info["capabilities"]["external_tools"], true);
     assert_eq!(info["capabilities"]["worker_runtime"], true);
+    assert_eq!(info["capabilities"]["fleet_run_create"], true);
+    assert_eq!(info["capabilities"]["fleet_run_start"], true);
+    assert_eq!(info["capabilities"]["fleet_event_replay"], true);
+    assert_eq!(info["capabilities"]["fleet_event_stream"], true);
+    assert_eq!(info["capabilities"]["fleet_local_target"], true);
     assert_eq!(info["account"]["schema_version"], 1);
     assert_eq!(info["account"]["state"], "signed_out");
     assert_eq!(info["account"]["api_base"], "https://api.codewhale.net");

docs/RUNTIME_API.md

@@ -834,9 +834,9 @@ model is preserved. Cross-origin preflights advertise only `Authorization`,
 `X-DeepSeek-Runtime-Token` request header; custom request headers are not
 allowed. Added in v0.8.10 (#561), tightened in v0.9.1 (#4454).
 
-## Runtime SDK Fleet Helpers
+## Managed Fleet Runtime and SDK helpers
 
-The v0.8.60 Runtime SDK fixture lives in `npm/runtime-sdk` and is exposed as
+The Runtime SDK lives in `npm/runtime-sdk` and is exposed as
 the `@codewhale/runtime-sdk` workspace package. It is deliberately thin: every
 helper calls the local Rust Runtime API and therefore cannot bypass Codewhale's
 sandbox, approval prompts, provider configuration, or fleet ledger authority.
@@ -849,29 +849,83 @@ const client = createRuntimeClient({
   token: process.env.CODEWHALE_RUNTIME_TOKEN,
 });
 
-const { runs } = await client.listFleetRuns();
-const workers = await client.listFleetWorkers(runs[0].id);
-await client.restartWorker(workers.workers[0].worker_id);
+const created = await client.createFleetRun({
+  target: "this_computer",
+  roles: [{ name: "reviewer" }, { name: "verifier" }],
+  workflow: {
+    id: "release-check",
+    kind: "parallel",
+    tasks: [
+      { id: "review", name: "Review", instructions: "Review locally.", worker: { role: "reviewer" } },
+      { id: "verify", name: "Verify", instructions: "Verify locally.", worker: { role: "verifier" } },
+    ],
+  },
+});
+
+// POST /runs only prepares durable work. This call crosses the launch gate.
+await client.startFleetRun(created.run.id);
+
+let cursor;
+for await (const event of client.fleetEvents(created.run.id, { after: cursor })) {
+  if (event.cursor) cursor = event.cursor;
+  if (event.event === "fleet.replay.cursor_unavailable") {
+    // Reload getFleetRun(created.run.id), then reconnect without the old cursor.
+  }
+}
 ```
 
-Fleet helpers cover the v0.8.60 HTTP surface:
+The managed path is deliberately two-step. `POST /v1/fleet/runs` validates and
+persists the run and queue without starting a worker. A separate authenticated
+`POST /start` activates it and schedules the executor driver; its `202` response
+reports `leased: 0` because the driver performs all leasing after it owns the
+run. Creation requires named roles, one task owner per role, a `parallel`
+Workflow, and an explicit Runtime target. v0.9.4 executes
+only `this_computer`; `another_computer` and `cloud` return `501` rather than
+silently executing locally. Worker IDs are generated per run; caller-assigned
+`worker_specs` return `501` until custom workers can be given collision-free
+managed identities. Parallel tasks with overlapping effective write roots are
+rejected before the run is journaled. Managed `security_policy` overrides also
+fail closed until that document can be enforced end to end; executable
+authority comes from each named role's tool posture and bounded task workspace
+scope.
+
+Fleet helpers cover this HTTP surface:
 
 | Helper | Runtime API route |
 |---|---|
+| `createFleetRun(spec)` | `POST /v1/fleet/runs` |
+| `startFleetRun(runId)` | `POST /v1/fleet/runs/{run_id}/start` |
 | `listFleetRuns()` | `GET /v1/fleet/runs` |
 | `getFleetRun(runId)` | `GET /v1/fleet/runs/{run_id}` |
 | `listFleetWorkers(runId)` | `GET /v1/fleet/runs/{run_id}/workers` |
 | `getFleetWorker(workerId)` | `GET /v1/fleet/workers/{worker_id}` |
 | `interruptWorker(workerId)` | `POST /v1/fleet/workers/{worker_id}/interrupt` |
+| `stopWorker(workerId)` | `POST /v1/fleet/workers/{worker_id}/stop` |
 | `restartWorker(workerId)` | `POST /v1/fleet/workers/{worker_id}/restart` |
 | `stopFleetRun(runId)` | `POST /v1/fleet/runs/{run_id}/stop` |
+| `replayFleetEvents(runId, options)` | `GET /v1/fleet/runs/{run_id}/events/replay` |
+| `fleetEvents(runId, options)` | `GET /v1/fleet/runs/{run_id}/events` (SSE) |
 
-`createFleetRun(spec)` and `fleetEvents(runId)` are typed ahead of the current
-Rust routes so editor/web clients can code against the intended SDK contract.
-Until the Runtime API exposes `POST /v1/fleet/runs` and a fleet event stream,
-the SDK raises `RuntimeCapabilityError` with stable capability strings
-(`fleet_run_create`, `fleet_event_stream`) instead of surfacing those gaps as
-generic fetch failures.
+`stopWorker` durably cancels that worker's active task and leaves the rest of
+the Fleet running. `interruptWorker` is the compatibility name for the same
+attempt-fenced cancellation transition. `stopFleetRun` cancels every queued or
+active task and marks the whole run cancelled.
+
+Replay covers aggregate run/task transitions and privacy-bounded individual
+worker transitions. Event bodies omit prompts, tool call IDs, completion text,
+artifact paths/checksums, and cancellation identities; bounded failure reasons
+pass through secret redaction. `cursor` is opaque and stable across ordinary
+appends and Runtime restarts. Clients reconnect with `after=<cursor>`. A fresh
+request returns a bounded newest tail and marks `history_truncated` when older
+history exists. Ledger compaction can remove an old cursor; the JSON endpoint
+then returns `409`, while the SSE endpoint emits
+`fleet.replay.cursor_unavailable`, so the client reloads the current run
+projection instead of accepting a silent gap.
+
+`GET /v1/runtime/info` advertises `fleet_run_create`, `fleet_run_start`,
+`fleet_event_replay`, `fleet_event_stream`, and `fleet_local_target`. Older
+runtimes without a requested route still produce a typed SDK
+`RuntimeCapabilityError`.
 
 Verification:
 

npm/runtime-sdk/README.md

@@ -12,9 +12,29 @@ const client = createRuntimeClient({
   token: process.env.CODEWHALE_RUNTIME_TOKEN,
 });
 
-const { runs } = await client.listFleetRuns();
-const workers = await client.listFleetWorkers(runs[0].id);
-await client.interruptWorker(workers.workers[0].worker_id);
+const created = await client.createFleetRun({
+  target: "this_computer",
+  roles: [{ name: "reviewer" }, { name: "verifier" }],
+  workflow: {
+    id: "release-check",
+    kind: "parallel",
+    tasks: [
+      { id: "review", name: "Review", instructions: "Review locally.", worker: { role: "reviewer" } },
+      { id: "verify", name: "Verify", instructions: "Verify locally.", worker: { role: "verifier" } },
+    ],
+  },
+});
+
+// Creation is durable but does not launch work. Launch remains explicit.
+await client.startFleetRun(created.run.id);
+
+let cursor;
+for await (const event of client.fleetEvents(created.run.id, { after: cursor })) {
+  if (event.cursor) cursor = event.cursor; // persist durable cursors only
+  if (event.event === "fleet.replay.cursor_unavailable") {
+    // Reload getFleetRun(created.run.id), then reconnect without the old cursor.
+  }
+}
 ```
 
 ## Fleet Helpers
@@ -24,12 +44,23 @@ await client.interruptWorker(workers.workers[0].worker_id);
 - `listFleetWorkers(runId)`
 - `getFleetWorker(workerId)`
 - `interruptWorker(workerId)`
+- `stopWorker(workerId)`
 - `restartWorker(workerId)`
 - `stopFleetRun(runId)`
-- `fleetEvents(runId)`
+- `startFleetRun(runId)`
+- `replayFleetEvents(runId, { after, limit })`
+- `fleetEvents(runId, { after, limit })`
 - `createFleetRun(spec)`
 
-`fleetEvents` and `createFleetRun` are typed ahead of the current v0.8.60 Rust
-Runtime API. If the local runtime does not expose those endpoints, the helpers
-raise `RuntimeCapabilityError` with a stable `capability` string instead of a
-generic fetch failure.
+The v0.9.4 Runtime implements the complete local managed-Fleet path. A creation
+request must name its roles, define a `parallel` Workflow, and select the
+explicit `this_computer` target. `another_computer` and `cloud` are contract
+values but fail closed until those targets are implemented. Event cursors are
+opaque and durable across Runtime restarts; if ledger compaction removes an old
+cursor, replay returns a conflict so the client can reload the run projection.
+Local worker IDs are generated per run; managed creation does not yet accept
+caller-assigned `worker_specs` because worker controls address IDs globally.
+
+Older runtimes that do not expose one of these endpoints produce a
+`RuntimeCapabilityError` with a stable capability string instead of a generic
+fetch failure.

npm/runtime-sdk/index.d.ts

@@ -8,6 +8,14 @@ export type FleetRunStatus =
   | "failed"
   | "cancelled";
 
+export type FleetRuntimeTarget = "this_computer" | "another_computer" | "cloud";
+export type FleetWorkflowKind = "parallel";
+
+export interface FleetWorkflowDescriptor {
+  id: string;
+  kind: FleetWorkflowKind;
+}
+
 export type FleetWorkerStatus =
   | "unknown"
   | "online"
@@ -53,7 +61,11 @@ export interface FleetTaskStatusSummary {
 export interface FleetRunSummary {
   id: string;
   name: string;
+  lifecycle_status: FleetRunStatus;
   status: FleetStatusSummary;
+  target?: FleetRuntimeTarget | null;
+  workflow?: FleetWorkflowDescriptor | null;
+  roles: string[];
   task_count: number;
   worker_count: number;
   tasks: FleetTaskStatusSummary[];
@@ -94,7 +106,11 @@ export interface FleetTaskSpec {
 }
 
 export interface FleetTaskWorkerProfile {
+  agent_profile?: string | null;
   role?: string | null;
+  loadout?: string | null;
+  model_class?: string | null;
+  model?: string | null;
   tool_profile?: string | null;
   tools?: string[];
   capabilities?: string[];
@@ -191,7 +207,7 @@ export interface FleetWorkersResponse {
 }
 
 export interface FleetWorkerActionResponse {
-  action: "interrupt" | "restart";
+  action: "interrupt" | "restart" | "stop";
   worker: FleetWorkerInspection;
 }
 
@@ -202,17 +218,82 @@ export interface StopFleetRunResponse {
   status: FleetStatusSummary;
 }
 
-export interface RuntimeClientOptions {
-  baseUrl?: string;
-  token?: string;
-  fetch?: typeof fetch;
+export interface ManagedFleetRole {
+  name: string;
+  agent_profile?: string | null;
+}
+
+export interface ManagedFleetWorkflow extends FleetWorkflowDescriptor {
+  tasks: FleetTaskSpec[];
 }
 
 export interface FleetRunCreateSpec {
   name?: string;
-  task_specs?: FleetTaskSpec[];
-  worker_specs?: FleetWorkerSpec[];
+  target: FleetRuntimeTarget;
+  roles: ManagedFleetRole[];
+  workflow: ManagedFleetWorkflow;
   labels?: Record<string, string>;
+  max_workers?: number;
+}
+
+export interface CreateFleetRunResponse {
+  execution: "awaiting_start";
+  run: FleetRunDetail;
+  warnings: string[];
+}
+
+export interface StartFleetRunResponse {
+  action: "start";
+  execution: "scheduled";
+  run_id: string;
+  target: "this_computer";
+  /** Leasing begins only after the scheduled driver owns the run. */
+  leased: 0;
+  queued: number;
+  worker_ids: string[];
+}
+
+export interface FleetRuntimeEvent {
+  cursor: string;
+  event: string;
+  run_id: string;
+  worker_id?: string | null;
+  task_id?: string | null;
+  timestamp?: string | null;
+  worker_seq?: number | null;
+  payload: Record<string, unknown>;
+}
+
+export interface FleetStreamControlEvent {
+  event:
+    | "fleet.replay.truncated"
+    | "fleet.replay.cursor_unavailable"
+    | "fleet.stream.error";
+  run_id?: string;
+  cursor?: never;
+  reload_projection?: boolean;
+  retryable?: boolean;
+}
+
+export type FleetStreamEvent = FleetRuntimeEvent | FleetStreamControlEvent;
+
+export interface FleetEventReplay {
+  run_id: string;
+  events: FleetRuntimeEvent[];
+  has_more: boolean;
+  history_truncated: boolean;
+  next_cursor?: string | null;
+}
+
+export interface FleetEventOptions {
+  after?: string;
+  limit?: number;
+}
+
+export interface RuntimeClientOptions {
+  baseUrl?: string;
+  token?: string;
+  fetch?: typeof fetch;
 }
 
 export class RuntimeApiError extends Error {
@@ -228,18 +309,21 @@ export class RuntimeCapabilityError extends RuntimeApiError {
 
 export class CodeWhaleRuntimeClient {
   constructor(options?: RuntimeClientOptions);
-  createFleetRun(spec: FleetRunCreateSpec | Record<string, unknown>): Promise<unknown>;
+  createFleetRun(spec: FleetRunCreateSpec): Promise<CreateFleetRunResponse>;
+  startFleetRun(runId: FleetRunId): Promise<StartFleetRunResponse>;
+  replayFleetEvents(runId: FleetRunId, options?: FleetEventOptions): Promise<FleetEventReplay>;
   listFleetRuns(): Promise<FleetRunsResponse>;
   getFleetRun(runId: FleetRunId): Promise<FleetRunDetail>;
   listFleetWorkers(runId: FleetRunId): Promise<FleetWorkersResponse>;
   getFleetWorker(workerId: string): Promise<FleetWorkerInspection>;
   interruptWorker(workerId: string): Promise<FleetWorkerActionResponse>;
+  stopWorker(workerId: string): Promise<FleetWorkerActionResponse>;
   restartWorker(workerId: string): Promise<FleetWorkerActionResponse>;
   stopFleetRun(runId: FleetRunId): Promise<StopFleetRunResponse>;
   fleetEvents(
     runId: FleetRunId,
-    options?: { path?: string },
-  ): AsyncIterable<FleetWorkerEvent>;
+    options?: FleetEventOptions & { path?: string },
+  ): AsyncIterable<FleetStreamEvent>;
 }
 
 export function createRuntimeClient(options?: RuntimeClientOptions): CodeWhaleRuntimeClient;

npm/runtime-sdk/index.js

@@ -37,6 +37,23 @@ export class CodeWhaleRuntimeClient {
     });
   }
 
+  async startFleetRun(runId) {
+    return this.#jsonRequest(`/v1/fleet/runs/${segment(runId)}/start`, {
+      method: "POST",
+      capability: "fleet_run_start",
+    });
+  }
+
+  async replayFleetEvents(runId, options = {}) {
+    const path = fleetEventPath(
+      `/v1/fleet/runs/${segment(runId)}/events/replay`,
+      options,
+    );
+    return this.#jsonRequest(path, {
+      capability: "fleet_event_replay",
+    });
+  }
+
   async listFleetRuns() {
     return this.#jsonRequest("/v1/fleet/runs");
   }
@@ -59,6 +76,12 @@ export class CodeWhaleRuntimeClient {
     });
   }
 
+  async stopWorker(workerId) {
+    return this.#jsonRequest(`/v1/fleet/workers/${segment(workerId)}/stop`, {
+      method: "POST",
+    });
+  }
+
   async restartWorker(workerId) {
     return this.#jsonRequest(`/v1/fleet/workers/${segment(workerId)}/restart`, {
       method: "POST",
@@ -72,10 +95,14 @@ export class CodeWhaleRuntimeClient {
   }
 
   async *fleetEvents(runId, options = {}) {
-    const path = options.path ?? `/v1/fleet/runs/${segment(runId)}/events`;
+    const path = fleetEventPath(
+      options.path ?? `/v1/fleet/runs/${segment(runId)}/events`,
+      options,
+    );
     const response = await this.#rawRequest(path, {
       method: "GET",
       capability: "fleet_event_stream",
+      accept: "text/event-stream",
     });
     const contentType = response.headers.get("content-type") ?? "";
     if (contentType.includes("application/json")) {
@@ -154,6 +181,21 @@ function segment(value) {
   return encodeURIComponent(String(value));
 }
 
+function fleetEventPath(path, options) {
+  const query = new URLSearchParams();
+  if (options.after !== undefined && options.after !== null && String(options.after) !== "") {
+    query.set("after", String(options.after));
+  }
+  if (options.limit !== undefined && options.limit !== null) {
+    query.set("limit", String(options.limit));
+  }
+  const encoded = query.toString();
+  if (!encoded) {
+    return path;
+  }
+  return `${path}${path.includes("?") ? "&" : "?"}${encoded}`;
+}
+
 async function readErrorBody(response) {
   try {
     const text = await response.text();
@@ -169,9 +211,9 @@ async function* parseEventStream(body) {
   for await (const chunk of body) {
     buffer += decoder.decode(chunk, { stream: true });
     let boundary;
-    while ((boundary = buffer.indexOf("\n\n")) >= 0) {
-      const frame = buffer.slice(0, boundary);
-      buffer = buffer.slice(boundary + 2);
+    while ((boundary = eventStreamBoundary(buffer)) !== null) {
+      const frame = buffer.slice(0, boundary.index);
+      buffer = buffer.slice(boundary.index + boundary.length);
       const event = parseSseFrame(frame);
       if (event !== undefined) {
         yield event;
@@ -185,14 +227,43 @@ async function* parseEventStream(body) {
   }
 }
 
+function eventStreamBoundary(buffer) {
+  const lf = buffer.indexOf("\n\n");
+  const crlf = buffer.indexOf("\r\n\r\n");
+  if (lf < 0 && crlf < 0) {
+    return null;
+  }
+  if (crlf >= 0 && (lf < 0 || crlf < lf)) {
+    return { index: crlf, length: 4 };
+  }
+  return { index: lf, length: 2 };
+}
+
 function parseSseFrame(frame) {
-  const data = frame
-    .split(/\r?\n/)
+  const lines = frame.split(/\r?\n/);
+  const eventName = lines
+    .find((line) => line.startsWith("event:"))
+    ?.slice("event:".length)
+    .trimStart();
+  const eventId = lines
+    .find((line) => line.startsWith("id:"))
+    ?.slice("id:".length)
+    .trimStart();
+  const data = lines
     .filter((line) => line.startsWith("data:"))
     .map((line) => line.slice("data:".length).trimStart())
     .join("\n");
   if (!data || data === "[DONE]") {
     return undefined;
   }
-  return JSON.parse(data);
+  const parsed = JSON.parse(data);
+  if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) {
+    if (eventName && parsed.event === undefined) {
+      parsed.event = eventName;
+    }
+    if (eventId && parsed.cursor === undefined) {
+      parsed.cursor = eventId;
+    }
+  }
+  return parsed;
 }

npm/runtime-sdk/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@codewhale/runtime-sdk",
   "version": "0.8.60",
-  "description": "Typed JavaScript helpers for CodeWhale Runtime API fleet endpoints.",
+  "description": "Typed JavaScript helpers for Codewhale Runtime API Fleet endpoints.",
   "license": "MIT",
   "type": "module",
   "main": "index.js",

npm/runtime-sdk/test/fleet-client.test.mjs

@@ -64,7 +64,11 @@ test("worker and run actions use POST endpoints", async () => {
             status: { runs: 1, workers: {} },
           }
         : {
-            action: url.pathname.endsWith("/restart") ? "restart" : "interrupt",
+            action: url.pathname.endsWith("/restart")
+              ? "restart"
+              : url.pathname.endsWith("/stop")
+                ? "stop"
+                : "interrupt",
             worker: { worker_id: "w1", artifacts: [] },
           },
     ),
@@ -72,19 +76,52 @@ test("worker and run actions use POST endpoints", async () => {
   const client = new CodeWhaleRuntimeClient({ fetch });
 
   await client.interruptWorker("w1");
+  await client.stopWorker("w1");
   await client.restartWorker("w1");
+  await client.startFleetRun("run-1");
   await client.stopFleetRun("run-1");
 
   assert.deepEqual(
     fetch.calls.map((call) => [new URL(call.url).pathname, call.init.method]),
     [
       ["/v1/fleet/workers/w1/interrupt", "POST"],
+      ["/v1/fleet/workers/w1/stop", "POST"],
       ["/v1/fleet/workers/w1/restart", "POST"],
+      ["/v1/fleet/runs/run-1/start", "POST"],
       ["/v1/fleet/runs/run-1/stop", "POST"],
     ],
   );
 });
 
+test("managed Fleet helpers send explicit launch metadata and reconnect cursors", async () => {
+  const fetch = fakeFetch((url) =>
+    jsonResponse(
+      url.pathname.endsWith("/events/replay")
+        ? { run_id: "run-1", events: [], has_more: false, history_truncated: false }
+        : { execution: "awaiting_start", run: { id: "run-1" }, warnings: [] },
+    ),
+  );
+  const client = new CodeWhaleRuntimeClient({ fetch });
+  const spec = {
+    target: "this_computer",
+    roles: [{ name: "reviewer" }],
+    workflow: {
+      id: "review",
+      kind: "parallel",
+      tasks: [{ id: "review", name: "Review", instructions: "Review.", worker: { role: "reviewer" } }],
+    },
+  };
+
+  await client.createFleetRun(spec);
+  await client.replayFleetEvents("run-1", { after: "fev1_cursor_worker", limit: 25 });
+
+  assert.deepEqual(JSON.parse(fetch.calls[0].init.body), spec);
+  const replayUrl = new URL(fetch.calls[1].url);
+  assert.equal(replayUrl.pathname, "/v1/fleet/runs/run-1/events/replay");
+  assert.equal(replayUrl.searchParams.get("after"), "fev1_cursor_worker");
+  assert.equal(replayUrl.searchParams.get("limit"), "25");
+});
+
 test("unsupported fleet capabilities raise typed errors", async () => {
   const fetch = fakeFetch(() => jsonResponse({ error: "not found" }, { status: 404 }));
   const client = new CodeWhaleRuntimeClient({ fetch });
@@ -143,7 +180,7 @@ test("fleetEvents parses text/event-stream frames", async () => {
     start(controller) {
       controller.enqueue(
         encoder.encode(
-          'data: {"seq":2,"run_id":"run-1","worker_id":"w1","task_id":"task-1","timestamp":"2026-06-13T00:00:01Z","label":"heartbeat","payload":{"state":"heartbeat","memory_mb":128}}\n\n',
+          'id: fev1_heartbeat_worker\nevent: fleet.worker.heartbeat\ndata: {"cursor":"fev1_heartbeat_worker","event":"fleet.worker.heartbeat","run_id":"run-1","worker_id":"w1","task_id":"task-1","timestamp":"2026-06-13T00:00:01Z","worker_seq":2,"payload":{"state":"heartbeat","memory_mb":128}}\n\n',
         ),
       );
       controller.close();
@@ -159,13 +196,53 @@ test("fleetEvents parses text/event-stream frames", async () => {
   const client = new CodeWhaleRuntimeClient({ fetch });
 
   const events = [];
-  for await (const event of client.fleetEvents("run-1")) {
+  for await (const event of client.fleetEvents("run-1", { after: "fev1_previous", limit: 10 })) {
     events.push(event);
   }
 
   assert.equal(events.length, 1);
   assert.equal(events[0].payload.state, "heartbeat");
   assert.equal(events[0].payload.memory_mb, 128);
+  const eventUrl = new URL(fetch.calls[0].url);
+  assert.equal(eventUrl.searchParams.get("after"), "fev1_previous");
+  assert.equal(eventUrl.searchParams.get("limit"), "10");
+  assert.equal(fetch.calls[0].init.headers.get("accept"), "text/event-stream");
+});
+
+test("fleetEvents preserves SSE control event names", async () => {
+  const encoder = new TextEncoder();
+  const body = new ReadableStream({
+    start(controller) {
+      controller.enqueue(
+        encoder.encode(
+          'event: fleet.replay.cursor_unavailable\r\ndata: {"run_id":"run-1","reload_projection":true}\r\n\r\n',
+        ),
+      );
+      controller.close();
+    },
+  });
+  const client = new CodeWhaleRuntimeClient({
+    fetch: fakeFetch(
+      () =>
+        new Response(body, {
+          status: 200,
+          headers: { "content-type": "text/event-stream" },
+        }),
+    ),
+  });
+
+  const events = [];
+  for await (const event of client.fleetEvents("run-1")) {
+    events.push(event);
+  }
+
+  assert.deepEqual(events, [
+    {
+      event: "fleet.replay.cursor_unavailable",
+      run_id: "run-1",
+      reload_projection: true,
+    },
+  ]);
 });
 
 test("ordinary HTTP errors remain RuntimeApiError", async () => {