| 1 | //! Shared Fleet control-plane surface (#1888, #4022). |
| 2 | //! |
| 3 | //! `codewhale fleet …` and the `/fleet …` slash command (and therefore its |
| 4 | //! hotbar action) run the *same* verbs against the *same* durable ledger and |
| 5 | //! render the *same* [`ControlReceipt`]. Nothing here formats twice: the CLI's |
| 6 | //! `print_status` / `print_inspection` delegate to the renderers below. |
| 7 | //! |
| 8 | //! Vocabulary: Fleet = who, Workflow = order, Lane = one running Workflow, |
| 9 | //! Runtime = where/how. Auto-Review is a permission posture and never appears |
| 10 | //! here as a role. |
| 11 | |
| 12 | use std::path::{Path, PathBuf}; |
| 13 | |
| 14 | use codewhale_lane::control::{ |
| 15 | Availability, ControlContext, ControlDomain, ControlFailure, ControlFailureKind, |
| 16 | ControlOperation, ControlReceipt, ControlSurface, DEFAULT_RUN_LIST_LIMIT, Known, RunListPage, |
| 17 | RunRouteDto, RunSummaryDto, RunUsageDto, UnknownReason, parse_target, redact_path, |
| 18 | sanitize_line, |
| 19 | }; |
| 20 | use codewhale_protocol::fleet::{ |
| 21 | FleetArtifactKind, FleetReceipt, FleetRun, FleetRunId, FleetRunStatus, FleetWorkerEventPayload, |
| 22 | FleetWorkerStatus, |
| 23 | }; |
| 24 | |
| 25 | use super::ledger::FleetLedgerState; |
| 26 | use super::manager::{FleetControlError, FleetManager, FleetStatusSnapshot, FleetWorkerInspection}; |
| 27 | |
| 28 | /// Maximum worker rows rendered in one durable status payload. |
| 29 | pub const MAX_STATUS_WORKER_ROWS: usize = 24; |
| 30 | /// Maximum artifact rows rendered in one inspection payload. |
| 31 | pub const MAX_INSPECTION_ARTIFACT_ROWS: usize = 24; |
| 32 | |
| 33 | /// The durable Fleet ledger for `workspace`, without creating it. |
| 34 | #[must_use] |
| 35 | pub fn fleet_ledger_path(workspace: &Path) -> PathBuf { |
| 36 | workspace.join(".codewhale").join("fleet.jsonl") |
| 37 | } |
| 38 | |
| 39 | /// Read-only availability probe for the Fleet domain. |
| 40 | /// |
| 41 | /// [`FleetManager::open`] creates the ledger as a side effect, so a status |
| 42 | /// surface must probe first; otherwise "this workspace has no Fleet ledger" |
| 43 | /// silently becomes "here is an empty Fleet ledger I just made". |
| 44 | #[must_use] |
| 45 | pub fn fleet_control_context(workspace: &Path) -> ControlContext { |
| 46 | ControlContext::probe(None, Some(&fleet_ledger_path(workspace))) |
| 47 | } |
| 48 | |
| 49 | // --------------------------------------------------------------------------- |
| 50 | // Labels and renderers (single source for CLI and TUI) |
| 51 | // --------------------------------------------------------------------------- |
| 52 | |
| 53 | #[must_use] |
| 54 | pub fn worker_status_label(status: &FleetWorkerStatus) -> &'static str { |
| 55 | match status { |
| 56 | FleetWorkerStatus::Unknown => "unknown", |
| 57 | FleetWorkerStatus::Online => "online", |
| 58 | FleetWorkerStatus::Busy => "busy", |
| 59 | FleetWorkerStatus::Offline => "offline", |
| 60 | FleetWorkerStatus::Unhealthy => "unhealthy", |
| 61 | FleetWorkerStatus::Draining => "draining", |
| 62 | FleetWorkerStatus::Retired => "retired", |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | #[must_use] |
| 67 | pub fn run_status_label(status: &FleetRunStatus) -> &'static str { |
| 68 | match status { |
| 69 | FleetRunStatus::Pending => "pending", |
| 70 | FleetRunStatus::Queued => "queued", |
| 71 | FleetRunStatus::Running => "running", |
| 72 | FleetRunStatus::Paused => "paused", |
| 73 | FleetRunStatus::Completed => "completed", |
| 74 | FleetRunStatus::Failed => "failed", |
| 75 | FleetRunStatus::Cancelled => "cancelled", |
| 76 | } |
| 77 | } |
| 78 | |
| 79 | #[must_use] |
| 80 | pub fn artifact_kind_label(kind: &FleetArtifactKind) -> String { |
| 81 | match kind { |
| 82 | FleetArtifactKind::Log => "log".to_string(), |
| 83 | FleetArtifactKind::Patch => "patch".to_string(), |
| 84 | FleetArtifactKind::TestResult => "test_result".to_string(), |
| 85 | FleetArtifactKind::Report => "report".to_string(), |
| 86 | FleetArtifactKind::Checkpoint => "checkpoint".to_string(), |
| 87 | FleetArtifactKind::Receipt => "receipt".to_string(), |
| 88 | FleetArtifactKind::Other(value) => value.clone(), |
| 89 | } |
| 90 | } |
| 91 | |
| 92 | #[must_use] |
| 93 | pub fn event_label(payload: &FleetWorkerEventPayload) -> String { |
| 94 | match payload { |
| 95 | FleetWorkerEventPayload::Queued => "queued".to_string(), |
| 96 | FleetWorkerEventPayload::Leased { .. } => "leased".to_string(), |
| 97 | FleetWorkerEventPayload::Starting => "starting".to_string(), |
| 98 | FleetWorkerEventPayload::Running => "running".to_string(), |
| 99 | FleetWorkerEventPayload::ModelWait { model } => model |
| 100 | .as_ref() |
| 101 | .map(|model| format!("model_wait model={model}")) |
| 102 | .unwrap_or_else(|| "model_wait".to_string()), |
| 103 | FleetWorkerEventPayload::RunningTool { tool, call_id } => call_id |
| 104 | .as_ref() |
| 105 | .map(|call_id| format!("running_tool tool={tool} call_id={call_id}")) |
| 106 | .unwrap_or_else(|| format!("running_tool tool={tool}")), |
| 107 | FleetWorkerEventPayload::WorkflowEvent { |
| 108 | workflow_run_id, |
| 109 | event, |
| 110 | } => event |
| 111 | .get("type") |
| 112 | .and_then(serde_json::Value::as_str) |
| 113 | .map(|kind| format!("workflow_event run_id={workflow_run_id} type={kind}")) |
| 114 | .unwrap_or_else(|| format!("workflow_event run_id={workflow_run_id}")), |
| 115 | FleetWorkerEventPayload::Heartbeat { .. } => "heartbeat".to_string(), |
| 116 | FleetWorkerEventPayload::Artifact(artifact) => { |
| 117 | format!("artifact kind={}", artifact_kind_label(&artifact.kind)) |
| 118 | } |
| 119 | FleetWorkerEventPayload::Completed { exit_code, summary } => match (exit_code, summary) { |
| 120 | (Some(code), Some(summary)) => format!("completed exit_code={code} {summary}"), |
| 121 | (Some(code), None) => format!("completed exit_code={code}"), |
| 122 | (None, Some(summary)) => format!("completed {summary}"), |
| 123 | (None, None) => "completed".to_string(), |
| 124 | }, |
| 125 | FleetWorkerEventPayload::Failed { |
| 126 | reason, |
| 127 | recoverable, |
| 128 | } => format!("failed recoverable={recoverable} reason={reason}"), |
| 129 | FleetWorkerEventPayload::Cancelled { cancelled_by } => cancelled_by |
| 130 | .as_ref() |
| 131 | .map(|by| format!("cancelled by={by}")) |
| 132 | .unwrap_or_else(|| "cancelled".to_string()), |
| 133 | FleetWorkerEventPayload::Interrupted { signal } => signal |
| 134 | .as_ref() |
| 135 | .map(|signal| format!("interrupted signal={signal}")) |
| 136 | .unwrap_or_else(|| "interrupted".to_string()), |
| 137 | FleetWorkerEventPayload::Stale { last_heartbeat_at } => last_heartbeat_at |
| 138 | .as_ref() |
| 139 | .map(|ts| format!("stale last_heartbeat_at={ts}")) |
| 140 | .unwrap_or_else(|| "stale".to_string()), |
| 141 | FleetWorkerEventPayload::Restarted { restart_count } => { |
| 142 | format!("restarted count={restart_count}") |
| 143 | } |
| 144 | FleetWorkerEventPayload::Escalated { channel, alert_id } => alert_id |
| 145 | .as_ref() |
| 146 | .map(|alert_id| format!("escalated channel={channel} alert_id={alert_id}")) |
| 147 | .unwrap_or_else(|| format!("escalated channel={channel}")), |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | /// Durable status snapshot as bounded lines. |
| 152 | #[must_use] |
| 153 | pub fn status_lines(status: &FleetStatusSnapshot) -> Vec<String> { |
| 154 | let mut lines = vec![format!( |
| 155 | "fleet: runs={} queued={} running={} completed={} partial={} failed={} restarted={} \ |
| 156 | escalated={} transport_failed={} task_failed={} verifier_failed={} cancelled={} stale={}", |
| 157 | status.runs, |
| 158 | status.queued, |
| 159 | status.running, |
| 160 | status.completed, |
| 161 | status.partial, |
| 162 | status.failed, |
| 163 | status.restarted, |
| 164 | status.escalated, |
| 165 | status.transport_failed, |
| 166 | status.task_failed, |
| 167 | status.verifier_failed, |
| 168 | status.cancelled, |
| 169 | status.stale |
| 170 | )]; |
| 171 | if !status.workers.is_empty() { |
| 172 | lines.push("workers:".to_string()); |
| 173 | for (worker_id, worker_status) in status.workers.iter().take(MAX_STATUS_WORKER_ROWS) { |
| 174 | lines.push(format!( |
| 175 | " {worker_id} {}", |
| 176 | worker_status_label(worker_status) |
| 177 | )); |
| 178 | } |
| 179 | let omitted = status.workers.len().saturating_sub(MAX_STATUS_WORKER_ROWS); |
| 180 | if omitted > 0 { |
| 181 | lines.push(format!( |
| 182 | " [{omitted} more worker(s) omitted by the {MAX_STATUS_WORKER_ROWS}-row bound]" |
| 183 | )); |
| 184 | } |
| 185 | } |
| 186 | lines |
| 187 | } |
| 188 | |
| 189 | /// Exactly the text `codewhale fleet status` has always printed. |
| 190 | #[must_use] |
| 191 | pub fn render_fleet_status_snapshot(status: &FleetStatusSnapshot) -> String { |
| 192 | status_lines(status).join("\n") |
| 193 | } |
| 194 | |
| 195 | /// Durable worker inspection as bounded lines. |
| 196 | #[must_use] |
| 197 | pub fn inspection_lines(inspection: &FleetWorkerInspection) -> Vec<String> { |
| 198 | let mut lines = vec![ |
| 199 | format!("worker: {}", inspection.worker_id), |
| 200 | format!("status: {}", worker_status_label(&inspection.status)), |
| 201 | ]; |
| 202 | if let Some(run_id) = &inspection.current_run_id { |
| 203 | lines.push(format!("run: {}", run_id.0)); |
| 204 | } |
| 205 | if let Some(task_id) = &inspection.current_task_id { |
| 206 | lines.push(format!("task: {task_id}")); |
| 207 | } |
| 208 | if let Some(objective) = &inspection.objective { |
| 209 | lines.push(format!("objective: {objective}")); |
| 210 | } |
| 211 | if let Some(role) = &inspection.role { |
| 212 | lines.push(format!("role: {role}")); |
| 213 | } |
| 214 | if let Some(host) = &inspection.host { |
| 215 | lines.push(format!("host: {host}")); |
| 216 | } |
| 217 | if let Some(heartbeat) = &inspection.latest_heartbeat_at { |
| 218 | lines.push(format!("heartbeat: {heartbeat}")); |
| 219 | } |
| 220 | if let Some(event) = &inspection.latest_event { |
| 221 | lines.push(format!( |
| 222 | "latest_event: seq={} {}", |
| 223 | event.seq, |
| 224 | event_label(&event.payload) |
| 225 | )); |
| 226 | } |
| 227 | if !inspection.artifacts.is_empty() { |
| 228 | lines.push("artifacts:".to_string()); |
| 229 | for artifact in inspection |
| 230 | .artifacts |
| 231 | .iter() |
| 232 | .take(MAX_INSPECTION_ARTIFACT_ROWS) |
| 233 | { |
| 234 | lines.push(format!( |
| 235 | " {} {}", |
| 236 | artifact_kind_label(&artifact.kind), |
| 237 | artifact.path.display() |
| 238 | )); |
| 239 | } |
| 240 | let omitted = inspection |
| 241 | .artifacts |
| 242 | .len() |
| 243 | .saturating_sub(MAX_INSPECTION_ARTIFACT_ROWS); |
| 244 | if omitted > 0 { |
| 245 | lines.push(format!(" [{omitted} more artifact(s) omitted]")); |
| 246 | } |
| 247 | } |
| 248 | if let Some(receipt) = &inspection.receipt_summary { |
| 249 | lines.push(format!("receipt: {receipt}")); |
| 250 | } |
| 251 | if let Some(error) = &inspection.last_error { |
| 252 | lines.push(format!("last_error: {error}")); |
| 253 | } |
| 254 | if let Some(alert) = &inspection.alert_state { |
| 255 | lines.push(format!("alert: {alert}")); |
| 256 | } |
| 257 | lines |
| 258 | } |
| 259 | |
| 260 | #[must_use] |
| 261 | pub fn render_inspection(inspection: &FleetWorkerInspection) -> String { |
| 262 | inspection_lines(inspection).join("\n") |
| 263 | } |
| 264 | |
| 265 | /// Artifact listing lines for `codewhale fleet artifacts`. |
| 266 | #[must_use] |
| 267 | pub fn artifact_lines(inspection: &FleetWorkerInspection) -> Vec<String> { |
| 268 | if inspection.artifacts.is_empty() { |
| 269 | return vec!["artifacts: none".to_string()]; |
| 270 | } |
| 271 | let mut lines = vec!["artifacts:".to_string()]; |
| 272 | for artifact in inspection |
| 273 | .artifacts |
| 274 | .iter() |
| 275 | .take(MAX_INSPECTION_ARTIFACT_ROWS) |
| 276 | { |
| 277 | let size = artifact |
| 278 | .size_bytes |
| 279 | .map(|size| format!(" size={size}")) |
| 280 | .unwrap_or_default(); |
| 281 | let mime = artifact |
| 282 | .mime_type |
| 283 | .as_ref() |
| 284 | .map(|mime| format!(" mime={mime}")) |
| 285 | .unwrap_or_default(); |
| 286 | lines.push(format!( |
| 287 | " {} {}{}{}", |
| 288 | artifact_kind_label(&artifact.kind), |
| 289 | artifact.path.display(), |
| 290 | size, |
| 291 | mime |
| 292 | )); |
| 293 | } |
| 294 | let omitted = inspection |
| 295 | .artifacts |
| 296 | .len() |
| 297 | .saturating_sub(MAX_INSPECTION_ARTIFACT_ROWS); |
| 298 | if omitted > 0 { |
| 299 | lines.push(format!(" [{omitted} more artifact(s) omitted]")); |
| 300 | } |
| 301 | lines |
| 302 | } |
| 303 | |
| 304 | #[must_use] |
| 305 | pub fn render_artifacts(inspection: &FleetWorkerInspection) -> String { |
| 306 | artifact_lines(inspection).join("\n") |
| 307 | } |
| 308 | |
| 309 | // --------------------------------------------------------------------------- |
| 310 | // Fleet run DTOs |
| 311 | // --------------------------------------------------------------------------- |
| 312 | |
| 313 | fn label(run: &FleetRun, key: &str) -> Option<String> { |
| 314 | run.labels.get(key).cloned() |
| 315 | } |
| 316 | |
| 317 | /// Project the most recent receipt's resolved route onto the shared DTO. |
| 318 | /// |
| 319 | /// Everything the receipt records exactly becomes `Known`; everything it does |
| 320 | /// not record stays typed-unknown. In particular the Fleet receipt persists |
| 321 | /// the *effective* reasoning tier only, so `requested_reasoning` is |
| 322 | /// `not_recorded` rather than being back-filled from the effective value. |
| 323 | fn route_dto(receipt: Option<&FleetReceipt>) -> RunRouteDto { |
| 324 | let Some(route) = receipt.and_then(|receipt| receipt.resolved_route.as_ref()) else { |
| 325 | return RunRouteDto::all_unknown(UnknownReason::NotRecorded); |
| 326 | }; |
| 327 | RunRouteDto { |
| 328 | provider_id: Known::Known(route.provider_id.clone()), |
| 329 | provider_exact_id: Known::from_option(route.provider_exact_id.clone()), |
| 330 | model: Known::Known( |
| 331 | route |
| 332 | .canonical_model |
| 333 | .clone() |
| 334 | .unwrap_or_else(|| route.wire_model_id.clone()), |
| 335 | ), |
| 336 | requested_reasoning: Known::unknown(), |
| 337 | effective_reasoning: Known::from_option(route.reasoning_effort.clone()), |
| 338 | route_source: Known::Known(route.source.clone()), |
| 339 | } |
| 340 | } |
| 341 | |
| 342 | /// Project one durable Fleet run into the shared run DTO. |
| 343 | #[must_use] |
| 344 | pub fn fleet_run_summary(run: &FleetRun, receipt: Option<&FleetReceipt>) -> RunSummaryDto { |
| 345 | RunSummaryDto { |
| 346 | domain: ControlDomain::Fleet, |
| 347 | run_id: run.id.0.clone(), |
| 348 | status: run_status_label(&run.status).to_string(), |
| 349 | // The Fleet ledger fences lifecycle per task, not per run. |
| 350 | lifecycle_seq: Known::not_applicable(), |
| 351 | runtime: Known::from_option(label(run, "runtime")), |
| 352 | workflow: Known::from_option(label(run, "workflow")), |
| 353 | fleet: Known::from_option( |
| 354 | label(run, "fleet").or_else(|| (!run.name.trim().is_empty()).then(|| run.name.clone())), |
| 355 | ), |
| 356 | issue: Known::from_option(label(run, "issue")), |
| 357 | goal: Known::from_option(label(run, "goal").or_else(|| label(run, "objective"))), |
| 358 | started_at: Known::Known(run.created_at.clone()), |
| 359 | stopped_at: Known::from_option(run.completed_at.clone()), |
| 360 | // Fleet runs are workspace-scoped; there is no per-run worktree, and |
| 361 | // the Lane-shaped Runtime handles do not apply to a Fleet run. |
| 362 | location: Known::not_applicable(), |
| 363 | branch: Known::not_applicable(), |
| 364 | runtime_session: Known::not_applicable(), |
| 365 | runtime_socket: Known::not_applicable(), |
| 366 | attach: Known::not_applicable(), |
| 367 | log: Known::not_applicable(), |
| 368 | route: route_dto(receipt), |
| 369 | // The ledger does not persist token counts or wall-clock duration. |
| 370 | usage: RunUsageDto::all_unknown(UnknownReason::NotRecorded), |
| 371 | } |
| 372 | } |
| 373 | |
| 374 | /// Bounded page of durable Fleet runs, newest first. |
| 375 | #[must_use] |
| 376 | pub fn fleet_run_page(state: &FleetLedgerState, limit: usize) -> RunListPage { |
| 377 | let mut summaries: Vec<RunSummaryDto> = state |
| 378 | .runs |
| 379 | .values() |
| 380 | .map(|run| { |
| 381 | let status = state |
| 382 | .run_status_overrides |
| 383 | .get(&run.id.0) |
| 384 | .cloned() |
| 385 | .unwrap_or_else(|| run.status.clone()); |
| 386 | let receipt = state |
| 387 | .receipts |
| 388 | .values() |
| 389 | .filter(|receipt| receipt.run_id.0 == run.id.0) |
| 390 | .max_by(|a, b| a.completed_at.cmp(&b.completed_at)); |
| 391 | let mut summary = fleet_run_summary(run, receipt); |
| 392 | summary.status = run_status_label(&status).to_string(); |
| 393 | summary |
| 394 | }) |
| 395 | .collect(); |
| 396 | // Newest first, matching `lane list`. Sort on the parsed UTC instant, not |
| 397 | // the rendered text: two timestamps written at different offsets order |
| 398 | // wrongly under a string compare. Unparseable timestamps sort last rather |
| 399 | // than being silently interleaved, and the exact id breaks every tie so |
| 400 | // the order is total and stable. |
| 401 | summaries.sort_by(|a, b| { |
| 402 | instant_of(&a.started_at) |
| 403 | .cmp(&instant_of(&b.started_at)) |
| 404 | .reverse() |
| 405 | .then_with(|| a.run_id.cmp(&b.run_id)) |
| 406 | }); |
| 407 | RunListPage::bounded(summaries, limit) |
| 408 | } |
| 409 | |
| 410 | /// Parse a recorded timestamp into a comparable UTC instant. |
| 411 | /// |
| 412 | /// `None` for unknown or unparseable values, which `Option`'s ordering places |
| 413 | /// before every real instant — and therefore last under the reversed |
| 414 | /// newest-first sort. |
| 415 | fn instant_of(value: &Known<String>) -> Option<chrono::DateTime<chrono::Utc>> { |
| 416 | let raw = value.as_known()?; |
| 417 | chrono::DateTime::parse_from_rfc3339(raw) |
| 418 | .ok() |
| 419 | .map(|parsed| parsed.with_timezone(&chrono::Utc)) |
| 420 | } |
| 421 | |
| 422 | // --------------------------------------------------------------------------- |
| 423 | // Executor — the one code path behind `codewhale fleet …` and `/fleet …` |
| 424 | // --------------------------------------------------------------------------- |
| 425 | |
| 426 | /// Run a Fleet control verb against the durable workspace ledger, using a |
| 427 | /// default manager. |
| 428 | /// |
| 429 | /// The slash command and hotbar use this. The CLI uses |
| 430 | /// [`execute_fleet_control_with`] so its configured manager (exec config, |
| 431 | /// stale-after window, session model, route config) still applies — same code |
| 432 | /// path, same receipt, caller-owned policy. |
| 433 | #[must_use] |
| 434 | pub fn execute_fleet_control( |
| 435 | surface: ControlSurface, |
| 436 | workspace: &Path, |
| 437 | operation: ControlOperation, |
| 438 | raw_target: Option<&str>, |
| 439 | ) -> ControlReceipt { |
| 440 | let descriptor = operation.descriptor(); |
| 441 | let availability = descriptor.availability(surface, fleet_control_context(workspace)); |
| 442 | if !availability.is_available() { |
| 443 | return ControlReceipt::unavailable(descriptor, surface, availability); |
| 444 | } |
| 445 | match FleetManager::open(workspace) { |
| 446 | Ok(manager) => execute_fleet_control_with( |
| 447 | surface, |
| 448 | workspace, |
| 449 | fleet_control_context(workspace), |
| 450 | &manager, |
| 451 | operation, |
| 452 | raw_target, |
| 453 | ), |
| 454 | Err(err) => ControlReceipt::failed( |
| 455 | descriptor, |
| 456 | surface, |
| 457 | None, |
| 458 | ControlFailure::backend(format!("{err:#}")), |
| 459 | ), |
| 460 | } |
| 461 | } |
| 462 | |
| 463 | /// Run a Fleet control verb against a caller-configured [`FleetManager`]. |
| 464 | /// |
| 465 | /// The CLI and the slash command both land here, so availability, target |
| 466 | /// selection, lifecycle outcome, retryability, and the sanitized failure are |
| 467 | /// decided once. `fleet.restart` is declared `SurfaceLimited` to the CLI |
| 468 | /// because it drives the manager loop to completion; this function reports |
| 469 | /// that as a typed unavailability on other surfaces rather than quietly doing |
| 470 | /// a different, smaller thing. |
| 471 | #[must_use] |
| 472 | pub fn execute_fleet_control_with( |
| 473 | surface: ControlSurface, |
| 474 | workspace: &Path, |
| 475 | ctx: ControlContext, |
| 476 | manager: &FleetManager, |
| 477 | operation: ControlOperation, |
| 478 | raw_target: Option<&str>, |
| 479 | ) -> ControlReceipt { |
| 480 | let descriptor = operation.descriptor(); |
| 481 | if descriptor.domain != ControlDomain::Fleet { |
| 482 | return ControlReceipt::rejected( |
| 483 | descriptor, |
| 484 | surface, |
| 485 | None, |
| 486 | ControlFailure::new( |
| 487 | ControlFailureKind::InvalidTarget, |
| 488 | format!("{} is not a Fleet verb", descriptor.id), |
| 489 | ), |
| 490 | ); |
| 491 | } |
| 492 | |
| 493 | let availability = descriptor.availability(surface, ctx); |
| 494 | if !availability.is_available() { |
| 495 | return ControlReceipt::unavailable(descriptor, surface, availability); |
| 496 | } |
| 497 | |
| 498 | let target = match parse_target(descriptor, raw_target) { |
| 499 | Ok(target) => target, |
| 500 | Err(failure) => return ControlReceipt::rejected(descriptor, surface, None, failure), |
| 501 | }; |
| 502 | |
| 503 | match operation { |
| 504 | ControlOperation::FleetList => match manager.rebuild_state() { |
| 505 | Ok(state) => ControlReceipt::inspected(descriptor, surface, None) |
| 506 | .with_runs(fleet_run_page(&state, DEFAULT_RUN_LIST_LIMIT)) |
| 507 | .with_detail([format!( |
| 508 | "ledger: {}", |
| 509 | redact_path(&fleet_ledger_path(workspace)) |
| 510 | )]), |
| 511 | Err(err) => ControlReceipt::failed( |
| 512 | descriptor, |
| 513 | surface, |
| 514 | None, |
| 515 | ControlFailure::backend(format!("{err:#}")), |
| 516 | ), |
| 517 | }, |
| 518 | ControlOperation::FleetStatus => match manager.status() { |
| 519 | Ok(status) => ControlReceipt::inspected(descriptor, surface, None).with_detail( |
| 520 | status_lines(&status).into_iter().chain([format!( |
| 521 | "ledger: {}", |
| 522 | redact_path(&fleet_ledger_path(workspace)) |
| 523 | )]), |
| 524 | ), |
| 525 | Err(err) => ControlReceipt::failed( |
| 526 | descriptor, |
| 527 | surface, |
| 528 | None, |
| 529 | ControlFailure::backend(format!("{err:#}")), |
| 530 | ), |
| 531 | }, |
| 532 | ControlOperation::FleetInterrupt => { |
| 533 | let Some(target) = target else { |
| 534 | return ControlReceipt::rejected( |
| 535 | descriptor, |
| 536 | surface, |
| 537 | None, |
| 538 | ControlFailure::invalid_target(format!( |
| 539 | "{} needs an exact worker id", |
| 540 | descriptor.id |
| 541 | )), |
| 542 | ); |
| 543 | }; |
| 544 | // Exact identity against the real ledger before any mutation: a |
| 545 | // worker id that was never seen in this workspace is `not_found`, |
| 546 | // which is a different fact from "known worker, nothing leased" |
| 547 | // (a conflict). Checking here keeps a typo from reaching the |
| 548 | // mutation path at all. |
| 549 | match manager.rebuild_state() { |
| 550 | Ok(state) => { |
| 551 | if !state.workers.contains_key(&target.id) { |
| 552 | return ControlReceipt::rejected( |
| 553 | descriptor, |
| 554 | surface, |
| 555 | Some(target.clone()), |
| 556 | ControlFailure::not_found(format!( |
| 557 | "no fleet worker with id {} in this workspace's ledger", |
| 558 | target.id |
| 559 | )), |
| 560 | ); |
| 561 | } |
| 562 | } |
| 563 | Err(err) => { |
| 564 | return ControlReceipt::failed( |
| 565 | descriptor, |
| 566 | surface, |
| 567 | Some(target), |
| 568 | ControlFailure::backend(format!("{err:#}")), |
| 569 | ); |
| 570 | } |
| 571 | } |
| 572 | // Bind before matching so the borrow of `target.id` is over |
| 573 | // before the arms move `target` into the receipt. |
| 574 | let interrupted = manager.interrupt_worker(&target.id); |
| 575 | match interrupted { |
| 576 | Ok(inspection) => ControlReceipt::transitioned(descriptor, surface, Some(target)) |
| 577 | .with_detail(inspection_lines(&inspection)), |
| 578 | Err(err) => { |
| 579 | // The manager refuses when the exact worker has no active |
| 580 | // task. That is a state conflict, not a transient backend |
| 581 | // fault: retrying it will keep failing until work is |
| 582 | // leased again. Classified by type, not by message text. |
| 583 | let message = format!("{err:#}"); |
| 584 | let failure = match err.downcast_ref::<FleetControlError>() { |
| 585 | Some(FleetControlError::NoActiveTask { .. }) => { |
| 586 | ControlFailure::conflict(message) |
| 587 | } |
| 588 | Some(FleetControlError::UnknownRun { .. }) => { |
| 589 | ControlFailure::not_found(message) |
| 590 | } |
| 591 | None => ControlFailure::backend(message), |
| 592 | }; |
| 593 | ControlReceipt::failed(descriptor, surface, Some(target), failure) |
| 594 | } |
| 595 | } |
| 596 | } |
| 597 | ControlOperation::FleetResume => { |
| 598 | let Some(target) = target else { |
| 599 | return ControlReceipt::rejected( |
| 600 | descriptor, |
| 601 | surface, |
| 602 | None, |
| 603 | ControlFailure::invalid_target(format!( |
| 604 | "{} needs an exact run id", |
| 605 | descriptor.id |
| 606 | )), |
| 607 | ); |
| 608 | }; |
| 609 | // Exact identity first. `resume_run` reconciles by run id and, for |
| 610 | // an id that is not in the ledger, would still write a run-status |
| 611 | // record keyed by whatever string the caller typed — durable |
| 612 | // pollution from a typo, reported as a benign no-op. Refuse before |
| 613 | // any write happens (#4022). |
| 614 | match manager.rebuild_state() { |
| 615 | Ok(state) => { |
| 616 | if !state.runs.contains_key(&target.id) { |
| 617 | return ControlReceipt::rejected( |
| 618 | descriptor, |
| 619 | surface, |
| 620 | Some(target.clone()), |
| 621 | ControlFailure::not_found( |
| 622 | FleetControlError::UnknownRun { |
| 623 | run_id: target.id.clone(), |
| 624 | } |
| 625 | .to_string(), |
| 626 | ), |
| 627 | ); |
| 628 | } |
| 629 | } |
| 630 | Err(err) => { |
| 631 | return ControlReceipt::failed( |
| 632 | descriptor, |
| 633 | surface, |
| 634 | Some(target), |
| 635 | ControlFailure::backend(format!("{err:#}")), |
| 636 | ); |
| 637 | } |
| 638 | } |
| 639 | let resumed = manager.resume_run(&FleetRunId::from(target.id.clone())); |
| 640 | match resumed { |
| 641 | Ok(report) => { |
| 642 | let reconciled = report.reclaimed_stale |
| 643 | + report.restarted |
| 644 | + report.failed |
| 645 | + report.escalated; |
| 646 | let detail = [format!( |
| 647 | "fleet resume: {} reclaimed_stale={} restarted={} failed={} escalated={}", |
| 648 | report.run_id.0, |
| 649 | report.reclaimed_stale, |
| 650 | report.restarted, |
| 651 | report.failed, |
| 652 | report.escalated |
| 653 | )] |
| 654 | .into_iter() |
| 655 | .chain(status_lines(&report.status)); |
| 656 | if reconciled == 0 { |
| 657 | ControlReceipt::no_change(descriptor, surface, Some(target)) |
| 658 | .with_detail(detail) |
| 659 | } else { |
| 660 | ControlReceipt::transitioned(descriptor, surface, Some(target)) |
| 661 | .with_detail(detail) |
| 662 | } |
| 663 | } |
| 664 | Err(err) => ControlReceipt::failed( |
| 665 | descriptor, |
| 666 | surface, |
| 667 | Some(target), |
| 668 | ControlFailure::backend(format!("{err:#}")), |
| 669 | ), |
| 670 | } |
| 671 | } |
| 672 | // `fleet.restart` is CLI-only (it drives the manager loop). The |
| 673 | // availability gate above already rejected it elsewhere; this arm |
| 674 | // keeps the refusal explicit if the table ever changes. |
| 675 | _ => ControlReceipt::unavailable( |
| 676 | descriptor, |
| 677 | surface, |
| 678 | Availability::Unavailable { |
| 679 | reason: codewhale_lane::UnavailableReason::SurfaceNotSupported, |
| 680 | hint: sanitize_line(descriptor.cli_invocation), |
| 681 | }, |
| 682 | ), |
| 683 | } |
| 684 | } |
| 685 | |
| 686 | #[cfg(test)] |
| 687 | mod tests { |
| 688 | use super::*; |
| 689 | use codewhale_lane::{ControlAuthority, LifecycleOutcome, PersistenceScope, UnavailableReason}; |
| 690 | use codewhale_protocol::fleet::{FleetResolvedRoute, FleetTaskResult}; |
| 691 | use std::collections::BTreeMap; |
| 692 | |
| 693 | fn run(id: &str) -> FleetRun { |
| 694 | FleetRun { |
| 695 | id: FleetRunId::from(id.to_string()), |
| 696 | name: "stopship".to_string(), |
| 697 | status: FleetRunStatus::Running, |
| 698 | target: None, |
| 699 | workflow: None, |
| 700 | roles: Vec::new(), |
| 701 | max_workers: Some(2), |
| 702 | task_specs: Vec::new(), |
| 703 | worker_specs: Vec::new(), |
| 704 | labels: BTreeMap::new(), |
| 705 | security_policy: None, |
| 706 | created_at: "2026-07-26T00:00:00Z".to_string(), |
| 707 | updated_at: None, |
| 708 | completed_at: None, |
| 709 | } |
| 710 | } |
| 711 | |
| 712 | fn receipt_with_route(run_id: &str) -> FleetReceipt { |
| 713 | FleetReceipt { |
| 714 | run_id: FleetRunId::from(run_id.to_string()), |
| 715 | task_id: "task-1".to_string(), |
| 716 | worker_id: "worker-1".to_string(), |
| 717 | attempt: Some(1), |
| 718 | terminal_seq: Some(9), |
| 719 | completed_at: "2026-07-26T00:01:00Z".to_string(), |
| 720 | result: FleetTaskResult::Pass, |
| 721 | failure_kind: None, |
| 722 | artifacts: Vec::new(), |
| 723 | score: None, |
| 724 | resolved_route: Some(FleetResolvedRoute { |
| 725 | provider_id: "deepseek".to_string(), |
| 726 | provider_exact_id: Some("custom".to_string()), |
| 727 | provider_kind: "deepseek".to_string(), |
| 728 | canonical_model: Some("deepseek-v3".to_string()), |
| 729 | wire_model_id: "deepseek-chat".to_string(), |
| 730 | protocol: "chat_completions".to_string(), |
| 731 | role: Some("implementer".to_string()), |
| 732 | loadout: None, |
| 733 | model_class: None, |
| 734 | model_route: None, |
| 735 | reasoning_effort: Some("high".to_string()), |
| 736 | role_source: None, |
| 737 | loadout_source: None, |
| 738 | model_class_source: None, |
| 739 | model_source: None, |
| 740 | source: "resolver".to_string(), |
| 741 | }), |
| 742 | effective_permissions: None, |
| 743 | } |
| 744 | } |
| 745 | |
| 746 | #[test] |
| 747 | fn fleet_run_dto_uses_exact_route_and_types_what_the_ledger_omits() { |
| 748 | let run = run("run-1"); |
| 749 | let receipt = receipt_with_route("run-1"); |
| 750 | let summary = fleet_run_summary(&run, Some(&receipt)); |
| 751 | |
| 752 | assert_eq!(summary.domain, ControlDomain::Fleet); |
| 753 | assert_eq!(summary.run_id, "run-1"); |
| 754 | assert_eq!(summary.status, "running"); |
| 755 | assert_eq!( |
| 756 | summary.route.provider_id, |
| 757 | Known::Known("deepseek".to_string()) |
| 758 | ); |
| 759 | assert_eq!( |
| 760 | summary.route.provider_exact_id, |
| 761 | Known::Known("custom".to_string()), |
| 762 | "the exact provider-table id must not collapse into the generic id" |
| 763 | ); |
| 764 | assert_eq!(summary.route.model, Known::Known("deepseek-v3".to_string())); |
| 765 | assert_eq!( |
| 766 | summary.route.effective_reasoning, |
| 767 | Known::Known("high".to_string()) |
| 768 | ); |
| 769 | assert_eq!( |
| 770 | summary.route.requested_reasoning.unknown_reason(), |
| 771 | Some(UnknownReason::NotRecorded), |
| 772 | "the ledger records the effective tier only; do not invent the request" |
| 773 | ); |
| 774 | assert_eq!(summary.route.reasoning_downgraded(), None); |
| 775 | assert_eq!( |
| 776 | summary.route.route_source, |
| 777 | Known::Known("resolver".to_string()) |
| 778 | ); |
| 779 | assert_eq!( |
| 780 | summary.usage.total_tokens.unknown_reason(), |
| 781 | Some(UnknownReason::NotRecorded) |
| 782 | ); |
| 783 | assert_eq!( |
| 784 | summary.lifecycle_seq.unknown_reason(), |
| 785 | Some(UnknownReason::NotApplicable) |
| 786 | ); |
| 787 | } |
| 788 | |
| 789 | #[test] |
| 790 | fn a_run_without_a_receipt_reports_every_route_field_unknown() { |
| 791 | let summary = fleet_run_summary(&run("run-2"), None); |
| 792 | for reason in [ |
| 793 | summary.route.provider_id.unknown_reason(), |
| 794 | summary.route.model.unknown_reason(), |
| 795 | summary.route.effective_reasoning.unknown_reason(), |
| 796 | summary.route.route_source.unknown_reason(), |
| 797 | ] { |
| 798 | assert_eq!(reason, Some(UnknownReason::NotRecorded)); |
| 799 | } |
| 800 | } |
| 801 | |
| 802 | #[test] |
| 803 | fn fleet_run_pages_are_bounded() { |
| 804 | let mut state = FleetLedgerState::default(); |
| 805 | for index in 0..10 { |
| 806 | let id = format!("run-{index:03}"); |
| 807 | state.runs.insert(id.clone(), run(&id)); |
| 808 | } |
| 809 | let page = fleet_run_page(&state, 3); |
| 810 | assert_eq!(page.total, 10); |
| 811 | assert_eq!(page.runs.len(), 3); |
| 812 | assert_eq!(page.truncated, 7); |
| 813 | } |
| 814 | |
| 815 | #[test] |
| 816 | fn an_absent_ledger_is_reported_and_not_created() { |
| 817 | let dir = tempfile::tempdir().unwrap(); |
| 818 | for surface in ControlSurface::ALL { |
| 819 | let receipt = |
| 820 | execute_fleet_control(*surface, dir.path(), ControlOperation::FleetStatus, None); |
| 821 | assert_eq!( |
| 822 | receipt.availability.reason(), |
| 823 | Some(UnavailableReason::NoFleetLedger), |
| 824 | "{surface}" |
| 825 | ); |
| 826 | assert_eq!(receipt.outcome, LifecycleOutcome::Rejected); |
| 827 | } |
| 828 | assert!( |
| 829 | !fleet_ledger_path(dir.path()).exists(), |
| 830 | "a read verb must not create the durable ledger" |
| 831 | ); |
| 832 | } |
| 833 | |
| 834 | #[test] |
| 835 | fn durable_fleet_status_is_identical_on_every_surface() { |
| 836 | let dir = tempfile::tempdir().unwrap(); |
| 837 | // Creating the manager is what makes the ledger exist. |
| 838 | FleetManager::open(dir.path()).unwrap(); |
| 839 | let mut rendered = std::collections::BTreeSet::new(); |
| 840 | for surface in ControlSurface::ALL { |
| 841 | let receipt = |
| 842 | execute_fleet_control(*surface, dir.path(), ControlOperation::FleetStatus, None); |
| 843 | assert_eq!(receipt.operation_id, "fleet.status"); |
| 844 | assert_eq!(receipt.authority, ControlAuthority::Read); |
| 845 | assert_eq!(receipt.persistence, PersistenceScope::FleetLedger); |
| 846 | assert_eq!(receipt.outcome, LifecycleOutcome::Inspected); |
| 847 | assert!( |
| 848 | receipt |
| 849 | .detail |
| 850 | .iter() |
| 851 | .any(|line| line.starts_with("fleet: runs=")), |
| 852 | "the durable ledger snapshot must be the payload" |
| 853 | ); |
| 854 | let mut normalized = receipt.clone(); |
| 855 | normalized.surface = ControlSurface::Cli; |
| 856 | rendered.insert(normalized.render()); |
| 857 | } |
| 858 | assert_eq!(rendered.len(), 1, "surfaces rendered different results"); |
| 859 | } |
| 860 | |
| 861 | #[test] |
| 862 | fn fleet_restart_is_cli_only_and_says_so_elsewhere() { |
| 863 | let dir = tempfile::tempdir().unwrap(); |
| 864 | FleetManager::open(dir.path()).unwrap(); |
| 865 | { |
| 866 | let surface = ControlSurface::Slash; |
| 867 | let receipt = execute_fleet_control( |
| 868 | surface, |
| 869 | dir.path(), |
| 870 | ControlOperation::FleetRestart, |
| 871 | Some("worker-1"), |
| 872 | ); |
| 873 | assert_eq!(receipt.outcome, LifecycleOutcome::Rejected); |
| 874 | assert_eq!( |
| 875 | receipt.availability.reason(), |
| 876 | Some(UnavailableReason::SurfaceNotSupported) |
| 877 | ); |
| 878 | assert!( |
| 879 | receipt |
| 880 | .availability |
| 881 | .hint() |
| 882 | .is_some_and(|hint| hint.contains("codewhale fleet restart")) |
| 883 | ); |
| 884 | } |
| 885 | } |
| 886 | |
| 887 | #[test] |
| 888 | fn interrupt_requires_an_exact_worker_id() { |
| 889 | let dir = tempfile::tempdir().unwrap(); |
| 890 | FleetManager::open(dir.path()).unwrap(); |
| 891 | for bad in [None, Some(""), Some("worker one"), Some("../escape")] { |
| 892 | let receipt = execute_fleet_control( |
| 893 | ControlSurface::Slash, |
| 894 | dir.path(), |
| 895 | ControlOperation::FleetInterrupt, |
| 896 | bad, |
| 897 | ); |
| 898 | assert_eq!( |
| 899 | receipt.failure.as_ref().map(|failure| failure.kind), |
| 900 | Some(ControlFailureKind::InvalidTarget), |
| 901 | "{bad:?}" |
| 902 | ); |
| 903 | } |
| 904 | } |
| 905 | |
| 906 | /// #4022: an id that is not in the ledger must be refused as `not_found` |
| 907 | /// *before* any durable write. Resuming a typo used to reconcile nothing, |
| 908 | /// write a run-status record under the typed id, and report `no_change`. |
| 909 | #[test] |
| 910 | fn resuming_an_unknown_run_is_not_found_and_writes_nothing() { |
| 911 | let dir = tempfile::tempdir().unwrap(); |
| 912 | FleetManager::open(dir.path()).unwrap(); |
| 913 | let ledger = fleet_ledger_path(dir.path()); |
| 914 | let before = std::fs::read(&ledger).unwrap(); |
| 915 | |
| 916 | for surface in ControlSurface::ALL { |
| 917 | let receipt = execute_fleet_control( |
| 918 | *surface, |
| 919 | dir.path(), |
| 920 | ControlOperation::FleetResume, |
| 921 | Some("run-does-not-exist"), |
| 922 | ); |
| 923 | assert_eq!(receipt.outcome, LifecycleOutcome::Rejected, "{surface}"); |
| 924 | assert_eq!( |
| 925 | receipt.failure.as_ref().map(|failure| failure.kind), |
| 926 | Some(ControlFailureKind::NotFound), |
| 927 | "{surface}" |
| 928 | ); |
| 929 | } |
| 930 | |
| 931 | assert_eq!( |
| 932 | std::fs::read(&ledger).unwrap(), |
| 933 | before, |
| 934 | "a refused resume must not append to the durable ledger" |
| 935 | ); |
| 936 | let state = FleetManager::open(dir.path()) |
| 937 | .unwrap() |
| 938 | .rebuild_state() |
| 939 | .unwrap(); |
| 940 | assert!( |
| 941 | !state |
| 942 | .run_status_overrides |
| 943 | .contains_key("run-does-not-exist"), |
| 944 | "a caller-supplied id must never become a durable ledger key" |
| 945 | ); |
| 946 | } |
| 947 | |
| 948 | /// #4022: newest-first ordering is computed on parsed instants, so a run |
| 949 | /// recorded at a non-UTC offset still sorts by when it actually happened. |
| 950 | #[test] |
| 951 | fn runs_sort_by_utc_instant_not_by_rendered_text() { |
| 952 | let mut state = FleetLedgerState::default(); |
| 953 | // 2026-07-26T00:30:00+02:00 == 2026-07-25T22:30:00Z, i.e. *earlier* |
| 954 | // than the UTC-stamped run even though its text sorts later. |
| 955 | let mut earlier = run("run-offset"); |
| 956 | earlier.created_at = "2026-07-26T00:30:00+02:00".to_string(); |
| 957 | let mut later = run("run-utc"); |
| 958 | later.created_at = "2026-07-25T23:00:00Z".to_string(); |
| 959 | state.runs.insert("run-offset".to_string(), earlier); |
| 960 | state.runs.insert("run-utc".to_string(), later); |
| 961 | |
| 962 | let page = fleet_run_page(&state, DEFAULT_RUN_LIST_LIMIT); |
| 963 | assert_eq!( |
| 964 | page.runs |
| 965 | .iter() |
| 966 | .map(|run| run.run_id.as_str()) |
| 967 | .collect::<Vec<_>>(), |
| 968 | vec!["run-utc", "run-offset"], |
| 969 | "a string compare would have put run-offset first" |
| 970 | ); |
| 971 | } |
| 972 | |
| 973 | #[test] |
| 974 | fn status_and_inspection_rendering_stay_bounded() { |
| 975 | let mut snapshot = FleetStatusSnapshot::default(); |
| 976 | for index in 0..(MAX_STATUS_WORKER_ROWS + 5) { |
| 977 | snapshot |
| 978 | .workers |
| 979 | .insert(format!("worker-{index:03}"), FleetWorkerStatus::Online); |
| 980 | } |
| 981 | let lines = status_lines(&snapshot); |
| 982 | // summary + "workers:" + capped rows + the omission notice |
| 983 | assert_eq!(lines.len(), 1 + 1 + MAX_STATUS_WORKER_ROWS + 1); |
| 984 | assert!(lines.last().unwrap().contains("5 more worker(s) omitted")); |
| 985 | } |
| 986 | } |
| 987 |