| 1 | use super::{ |
| 2 | SharedWorkflowControllers, SharedWorkflowLifecycles, SharedWorkflowRuns, |
| 3 | WorkflowDispatchFailure, WorkflowRunRecord, WorkflowRunStatus, WorkflowUiEvent, |
| 4 | WorkflowUiEventKind, WorkflowWorkLifecycle, |
| 5 | }; |
| 6 | use serde::{Deserialize, Serialize}; |
| 7 | use std::collections::HashMap; |
| 8 | use std::fs::OpenOptions; |
| 9 | use std::io::{BufRead, Write}; |
| 10 | use std::path::{Path, PathBuf}; |
| 11 | use std::sync::{Arc, Mutex, OnceLock}; |
| 12 | use tracing::warn; |
| 13 | |
| 14 | pub(super) const CODEWHALE_DIR: &str = ".codewhale"; |
| 15 | pub(super) const WORKFLOW_RUNS_FILE: &str = "workflow-runs.jsonl"; |
| 16 | |
| 17 | /// Per-workspace workflow state shared across tool-registry rebuilds. |
| 18 | pub(super) struct WorkflowWorkspaceState { |
| 19 | pub runs: SharedWorkflowRuns, |
| 20 | pub controllers: SharedWorkflowControllers, |
| 21 | lifecycles: SharedWorkflowLifecycles, |
| 22 | journal: WorkflowRunJournal, |
| 23 | } |
| 24 | |
| 25 | impl WorkflowWorkspaceState { |
| 26 | pub fn open(workspace: &Path) -> Arc<Self> { |
| 27 | Self::open_inner(workspace, true) |
| 28 | } |
| 29 | |
| 30 | /// Hydrate the journal without rewriting leftover `running` rows to |
| 31 | /// `failed`. Host cancel uses this after a restart so a controller-less |
| 32 | /// run can still be marked cancelled instead of looking like a crash. |
| 33 | pub fn open_preserving_running(workspace: &Path) -> Arc<Self> { |
| 34 | Self::open_inner(workspace, false) |
| 35 | } |
| 36 | |
| 37 | fn open_inner(workspace: &Path, recover_orphans: bool) -> Arc<Self> { |
| 38 | let journal = WorkflowRunJournal::open(workspace); |
| 39 | let runs = Arc::new(Mutex::new(journal.hydrate_runs(recover_orphans))); |
| 40 | Arc::new(Self { |
| 41 | runs, |
| 42 | controllers: Arc::new(Mutex::new(HashMap::new())), |
| 43 | lifecycles: Arc::new(Mutex::new(HashMap::new())), |
| 44 | journal, |
| 45 | }) |
| 46 | } |
| 47 | |
| 48 | pub fn attach_lifecycle(&self, run_id: &str, lifecycle: WorkflowWorkLifecycle) { |
| 49 | self.lifecycles |
| 50 | .lock() |
| 51 | .unwrap_or_else(|poison| poison.into_inner()) |
| 52 | .entry(run_id.to_string()) |
| 53 | .or_insert(lifecycle); |
| 54 | } |
| 55 | |
| 56 | pub fn reconcile_snapshot(&self, record: &WorkflowRunRecord) { |
| 57 | let lifecycle = self |
| 58 | .lifecycles |
| 59 | .lock() |
| 60 | .unwrap_or_else(|poison| poison.into_inner()) |
| 61 | .get(&record.run_id) |
| 62 | .cloned(); |
| 63 | if let Some(lifecycle) = lifecycle |
| 64 | && let Err(err) = lifecycle.reconcile_record(record) |
| 65 | { |
| 66 | warn!( |
| 67 | run_id = record.run_id, |
| 68 | "workflow Work reconciliation failed: {err}" |
| 69 | ); |
| 70 | } |
| 71 | } |
| 72 | |
| 73 | pub fn reconcile_cancel(&self, run_id: &str, outcome: super::CancelOutcome) { |
| 74 | let lifecycle = self |
| 75 | .lifecycles |
| 76 | .lock() |
| 77 | .unwrap_or_else(|poison| poison.into_inner()) |
| 78 | .get(run_id) |
| 79 | .cloned(); |
| 80 | if let Some(lifecycle) = lifecycle |
| 81 | && let Err(err) = lifecycle.reconcile_cancel(outcome) |
| 82 | { |
| 83 | warn!(run_id, "workflow cancellation reconciliation failed: {err}"); |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | pub fn mark_owner_missing(&self, run_id: &str) { |
| 88 | let lifecycle = self |
| 89 | .lifecycles |
| 90 | .lock() |
| 91 | .unwrap_or_else(|poison| poison.into_inner()) |
| 92 | .get(run_id) |
| 93 | .cloned(); |
| 94 | if let Some(lifecycle) = lifecycle { |
| 95 | lifecycle.reconcile_missing(); |
| 96 | } |
| 97 | } |
| 98 | |
| 99 | pub fn try_record_snapshot(&self, record: &WorkflowRunRecord) -> Result<(), String> { |
| 100 | self.journal |
| 101 | .append_snapshot(record) |
| 102 | .map_err(|err| err.to_string()) |
| 103 | } |
| 104 | |
| 105 | pub fn record_snapshot(&self, record: &WorkflowRunRecord) { |
| 106 | if let Err(err) = self.try_record_snapshot(record) { |
| 107 | warn!("workflow journal snapshot failed: {err}"); |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | pub fn record_progress(&self, run_id: &str, message: &str) { |
| 112 | if let Err(err) = self.journal.append_progress(run_id, message) { |
| 113 | warn!("workflow journal progress failed: {err}"); |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | pub fn record_event(&self, run_id: &str, event: &WorkflowUiEvent) { |
| 118 | if let Err(err) = self.journal.append_event(run_id, event) { |
| 119 | warn!("workflow journal event failed: {err}"); |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | /// Durable journal location for full-fidelity run detail (#2974). |
| 124 | pub fn journal_path(&self) -> &Path { |
| 125 | &self.journal.ledger_path |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | fn workspace_store() -> &'static Mutex<HashMap<PathBuf, Arc<WorkflowWorkspaceState>>> { |
| 130 | static STORE: OnceLock<Mutex<HashMap<PathBuf, Arc<WorkflowWorkspaceState>>>> = OnceLock::new(); |
| 131 | STORE.get_or_init(|| Mutex::new(HashMap::new())) |
| 132 | } |
| 133 | |
| 134 | pub(super) fn shared_workflow_state(workspace: &Path) -> Arc<WorkflowWorkspaceState> { |
| 135 | let key = workspace |
| 136 | .canonicalize() |
| 137 | .unwrap_or_else(|_| workspace.to_path_buf()); |
| 138 | let mut store = workspace_store() |
| 139 | .lock() |
| 140 | .unwrap_or_else(|poison| poison.into_inner()); |
| 141 | store |
| 142 | .entry(key) |
| 143 | .or_insert_with(|| WorkflowWorkspaceState::open(workspace)) |
| 144 | .clone() |
| 145 | } |
| 146 | |
| 147 | /// Read-only lookup that never creates workspace state, a journal |
| 148 | /// directory, or a ledger file. Used by the human-only `/structcopy` |
| 149 | /// command (#2033), which must stay side-effect free. |
| 150 | pub(super) fn peek_shared_workflow_state(workspace: &Path) -> Option<Arc<WorkflowWorkspaceState>> { |
| 151 | let key = workspace |
| 152 | .canonicalize() |
| 153 | .unwrap_or_else(|_| workspace.to_path_buf()); |
| 154 | workspace_store() |
| 155 | .lock() |
| 156 | .unwrap_or_else(|poison| poison.into_inner()) |
| 157 | .get(&key) |
| 158 | .cloned() |
| 159 | } |
| 160 | |
| 161 | #[derive(Debug, Clone, Serialize, Deserialize)] |
| 162 | #[serde(tag = "kind", rename_all = "snake_case")] |
| 163 | enum WorkflowJournalRecord { |
| 164 | // Boxed: a full run record dwarfs the progress variant |
| 165 | // (clippy::large_enum_variant). |
| 166 | Snapshot { |
| 167 | run: Box<WorkflowRunRecord>, |
| 168 | }, |
| 169 | Progress { |
| 170 | run_id: String, |
| 171 | message: String, |
| 172 | }, |
| 173 | Event { |
| 174 | run_id: String, |
| 175 | event: Box<WorkflowUiEvent>, |
| 176 | }, |
| 177 | } |
| 178 | |
| 179 | #[derive(Debug)] |
| 180 | struct WorkflowRunJournal { |
| 181 | ledger_path: PathBuf, |
| 182 | } |
| 183 | |
| 184 | impl WorkflowRunJournal { |
| 185 | fn open(workspace: &Path) -> Self { |
| 186 | let dir = workspace.join(CODEWHALE_DIR); |
| 187 | if let Err(err) = std::fs::create_dir_all(&dir) { |
| 188 | warn!( |
| 189 | "workflow journal dir create failed ({}): {err}", |
| 190 | dir.display() |
| 191 | ); |
| 192 | } |
| 193 | let ledger_path = dir.join(WORKFLOW_RUNS_FILE); |
| 194 | if !ledger_path.exists() |
| 195 | && let Err(err) = std::fs::write(&ledger_path, "") |
| 196 | { |
| 197 | warn!( |
| 198 | "workflow journal create failed ({}): {err}", |
| 199 | ledger_path.display() |
| 200 | ); |
| 201 | } |
| 202 | Self { ledger_path } |
| 203 | } |
| 204 | |
| 205 | fn hydrate_runs(&self, recover_orphans: bool) -> HashMap<String, WorkflowRunRecord> { |
| 206 | let file = match std::fs::File::open(&self.ledger_path) { |
| 207 | Ok(file) => file, |
| 208 | Err(_) => return HashMap::new(), |
| 209 | }; |
| 210 | let mut runs = HashMap::new(); |
| 211 | for line in std::io::BufReader::new(file).lines() { |
| 212 | let Ok(line) = line else { continue }; |
| 213 | let trimmed = line.trim(); |
| 214 | if trimmed.is_empty() { |
| 215 | continue; |
| 216 | } |
| 217 | let record = match serde_json::from_str::<WorkflowJournalRecord>(trimmed) { |
| 218 | Ok(record) => record, |
| 219 | Err(err) => { |
| 220 | warn!("workflow journal skipped malformed line: {err}"); |
| 221 | continue; |
| 222 | } |
| 223 | }; |
| 224 | match record { |
| 225 | WorkflowJournalRecord::Snapshot { run } => { |
| 226 | let mut run = *run; |
| 227 | run.normalize_bounded_ledgers(); |
| 228 | runs.insert(run.run_id.clone(), run); |
| 229 | } |
| 230 | WorkflowJournalRecord::Progress { run_id, message } => { |
| 231 | if let Some(run) = runs.get_mut(&run_id) { |
| 232 | run.push_progress(message); |
| 233 | } |
| 234 | } |
| 235 | WorkflowJournalRecord::Event { run_id, event } => { |
| 236 | if let Some(run) = runs.get_mut(&run_id) { |
| 237 | let event = *event; |
| 238 | if let WorkflowUiEventKind::TaskDispatchFailed { |
| 239 | label, |
| 240 | phase, |
| 241 | message, |
| 242 | } = &event.kind |
| 243 | { |
| 244 | run.push_dispatch_failure(WorkflowDispatchFailure { |
| 245 | at_ms: event.at_ms, |
| 246 | label: label.clone(), |
| 247 | phase: phase.clone(), |
| 248 | message: message.clone(), |
| 249 | }); |
| 250 | } |
| 251 | run.push_event(event); |
| 252 | } |
| 253 | } |
| 254 | } |
| 255 | } |
| 256 | // Journals written before #2974 have no counters; rebuild them |
| 257 | // from the retained tail so summaries stay truthful. |
| 258 | for run in runs.values_mut() { |
| 259 | run.normalize_bounded_ledgers(); |
| 260 | run.events_total = run.events_total.max(run.events.len() as u64); |
| 261 | } |
| 262 | // A run journaled as Running belongs to a process that is gone; |
| 263 | // without this it would show as live forever after a restart. |
| 264 | // Host cancel skips this rewrite so it can still mark the line |
| 265 | // cancelled with an honest "nothing live to stop" receipt. |
| 266 | if recover_orphans { |
| 267 | let mut recovered = Vec::new(); |
| 268 | for run in runs.values_mut() { |
| 269 | if run.status == WorkflowRunStatus::Running { |
| 270 | run.status = WorkflowRunStatus::Failed; |
| 271 | run.lifecycle_seq = run.lifecycle_seq.saturating_add(1); |
| 272 | run.completed_at_ms.get_or_insert_with(super::now_ms); |
| 273 | run.error = Some( |
| 274 | "process exited before the run completed (recovered on startup)" |
| 275 | .to_string(), |
| 276 | ); |
| 277 | recovered.push(run.clone()); |
| 278 | } |
| 279 | } |
| 280 | // The recovery decision is owner truth, not a presentation-only |
| 281 | // repair. Append it so another restart replays the same terminal |
| 282 | // sequence instead of rediscovering and incrementing it again. |
| 283 | for run in recovered { |
| 284 | if let Err(err) = self.append_snapshot(&run) { |
| 285 | warn!( |
| 286 | run_id = run.run_id, |
| 287 | "workflow recovery snapshot append failed: {err}" |
| 288 | ); |
| 289 | } |
| 290 | } |
| 291 | } |
| 292 | runs |
| 293 | } |
| 294 | |
| 295 | fn append_record(&self, record: &WorkflowJournalRecord) -> std::io::Result<()> { |
| 296 | let mut line = |
| 297 | serde_json::to_string(record).map_err(|err| std::io::Error::other(err.to_string()))?; |
| 298 | line.push('\n'); |
| 299 | let mut file = OpenOptions::new() |
| 300 | .create(true) |
| 301 | .append(true) |
| 302 | .open(&self.ledger_path)?; |
| 303 | file.write_all(line.as_bytes())?; |
| 304 | file.flush()?; |
| 305 | Ok(()) |
| 306 | } |
| 307 | |
| 308 | fn append_snapshot(&self, record: &WorkflowRunRecord) -> std::io::Result<()> { |
| 309 | self.append_record(&WorkflowJournalRecord::Snapshot { |
| 310 | run: Box::new(record.clone()), |
| 311 | }) |
| 312 | } |
| 313 | |
| 314 | fn append_progress(&self, run_id: &str, message: &str) -> std::io::Result<()> { |
| 315 | self.append_record(&WorkflowJournalRecord::Progress { |
| 316 | run_id: run_id.to_string(), |
| 317 | message: message.to_string(), |
| 318 | }) |
| 319 | } |
| 320 | |
| 321 | fn append_event(&self, run_id: &str, event: &WorkflowUiEvent) -> std::io::Result<()> { |
| 322 | self.append_record(&WorkflowJournalRecord::Event { |
| 323 | run_id: run_id.to_string(), |
| 324 | event: Box::new(event.clone()), |
| 325 | }) |
| 326 | } |
| 327 | } |
| 328 | |
| 329 | #[cfg(test)] |
| 330 | mod tests { |
| 331 | use super::super::{WORKFLOW_RUN_DISPATCH_FAILURES_MAX_RETAINED, WorkflowUiEventKind}; |
| 332 | use super::*; |
| 333 | |
| 334 | /// #5582: a degraded run must never project as an ordinary success |
| 335 | /// to owner-level consumers. |
| 336 | #[test] |
| 337 | fn owner_snapshot_keeps_degraded_distinct_from_completed() { |
| 338 | use crate::work_graph::OwnerState; |
| 339 | assert_eq!( |
| 340 | crate::tools::workflow::owner_state_for_run_status(WorkflowRunStatus::Degraded), |
| 341 | OwnerState::Degraded |
| 342 | ); |
| 343 | assert_eq!( |
| 344 | crate::tools::workflow::owner_state_for_run_status(WorkflowRunStatus::Completed), |
| 345 | OwnerState::Completed |
| 346 | ); |
| 347 | } |
| 348 | |
| 349 | fn sample_record(run_id: &str, status: WorkflowRunStatus) -> WorkflowRunRecord { |
| 350 | WorkflowRunRecord { |
| 351 | run_id: run_id.to_string(), |
| 352 | owner_session_id: Some("session-journal".to_string()), |
| 353 | status, |
| 354 | lifecycle_seq: 1, |
| 355 | started_at_ms: 1, |
| 356 | completed_at_ms: None, |
| 357 | source_path: None, |
| 358 | workflow_id: Some("fixture".to_string()), |
| 359 | workflow_goal: Some("journal test".to_string()), |
| 360 | token_budget: None, |
| 361 | child_ids: Vec::new(), |
| 362 | progress_count: 0, |
| 363 | progress: Vec::new(), |
| 364 | events: Vec::new(), |
| 365 | schema_errors: Vec::new(), |
| 366 | schema_repairs: Vec::new(), |
| 367 | schema_repair_count: 0, |
| 368 | dispatch_failure_count: 0, |
| 369 | dispatch_failures: Vec::new(), |
| 370 | result: None, |
| 371 | execution: None, |
| 372 | error: None, |
| 373 | verify_on_complete: false, |
| 374 | verification: None, |
| 375 | plan_approval: None, |
| 376 | gate_status: Vec::new(), |
| 377 | usage: None, |
| 378 | events_total: 0, |
| 379 | events_dropped: 0, |
| 380 | } |
| 381 | } |
| 382 | |
| 383 | #[test] |
| 384 | fn workflow_journal_hydrates_snapshots_and_progress() { |
| 385 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 386 | let state = WorkflowWorkspaceState::open(tmp.path()); |
| 387 | let running = sample_record("workflow_abc", WorkflowRunStatus::Running); |
| 388 | state.record_snapshot(&running); |
| 389 | state.record_progress("workflow_abc", "phase: scan"); |
| 390 | state.record_event( |
| 391 | "workflow_abc", |
| 392 | &WorkflowUiEvent::at( |
| 393 | 5, |
| 394 | "session-journal", |
| 395 | WorkflowUiEventKind::PhaseStarted { |
| 396 | title: "scan".to_string(), |
| 397 | }, |
| 398 | ), |
| 399 | ); |
| 400 | |
| 401 | let completed = WorkflowRunRecord { |
| 402 | status: WorkflowRunStatus::Completed, |
| 403 | completed_at_ms: Some(99), |
| 404 | progress: vec!["phase: scan".to_string()], |
| 405 | events: vec![WorkflowUiEvent::at( |
| 406 | 5, |
| 407 | "session-journal", |
| 408 | WorkflowUiEventKind::PhaseStarted { |
| 409 | title: "scan".to_string(), |
| 410 | }, |
| 411 | )], |
| 412 | ..sample_record("workflow_abc", WorkflowRunStatus::Completed) |
| 413 | }; |
| 414 | state.record_snapshot(&completed); |
| 415 | state.record_event( |
| 416 | "workflow_abc", |
| 417 | &WorkflowUiEvent::at( |
| 418 | 6, |
| 419 | "session-journal", |
| 420 | WorkflowUiEventKind::HandoffPromoted { |
| 421 | artifact_id: "workflow_abc:scout-1:scout-gate:findings".to_string(), |
| 422 | gate_id: "scout-gate".to_string(), |
| 423 | kind: "findings".to_string(), |
| 424 | from_role: "scout".to_string(), |
| 425 | to_role: "implementer".to_string(), |
| 426 | producer_task_id: "scout-1".to_string(), |
| 427 | }, |
| 428 | ), |
| 429 | ); |
| 430 | state.record_event( |
| 431 | "workflow_abc", |
| 432 | &WorkflowUiEvent::at( |
| 433 | 7, |
| 434 | "session-journal", |
| 435 | WorkflowUiEventKind::HandoffConsumed { |
| 436 | artifact_id: "workflow_abc:scout-1:scout-gate:findings".to_string(), |
| 437 | kind: "findings".to_string(), |
| 438 | from_role: "scout".to_string(), |
| 439 | to_role: "implementer".to_string(), |
| 440 | consumer_task_id: "implementer-1".to_string(), |
| 441 | }, |
| 442 | ), |
| 443 | ); |
| 444 | |
| 445 | let reloaded = WorkflowWorkspaceState::open(tmp.path()); |
| 446 | let runs = reloaded |
| 447 | .runs |
| 448 | .lock() |
| 449 | .expect("runs lock") |
| 450 | .get("workflow_abc") |
| 451 | .cloned() |
| 452 | .expect("hydrated run"); |
| 453 | assert_eq!(runs.status, WorkflowRunStatus::Completed); |
| 454 | assert_eq!(runs.progress, vec!["phase: scan"]); |
| 455 | assert_eq!(runs.events.len(), 3); |
| 456 | assert_eq!(runs.events[0].event_type(), "phase_started"); |
| 457 | let promoted = serde_json::to_value(&runs.events[1]).expect("promoted receipt"); |
| 458 | assert_eq!(promoted["type"], "handoff_promoted"); |
| 459 | assert_eq!( |
| 460 | promoted["artifact_id"], |
| 461 | "workflow_abc:scout-1:scout-gate:findings" |
| 462 | ); |
| 463 | assert_eq!(promoted["gate_id"], "scout-gate"); |
| 464 | assert_eq!(promoted["producer_task_id"], "scout-1"); |
| 465 | assert!(promoted.get("payload").is_none(), "{promoted}"); |
| 466 | let consumed = serde_json::to_value(&runs.events[2]).expect("consumed receipt"); |
| 467 | assert_eq!(consumed["type"], "handoff_consumed"); |
| 468 | assert_eq!(consumed["artifact_id"], promoted["artifact_id"]); |
| 469 | assert_eq!(consumed["consumer_task_id"], "implementer-1"); |
| 470 | assert!(consumed.get("payload").is_none(), "{consumed}"); |
| 471 | assert_eq!(runs.completed_at_ms, Some(99)); |
| 472 | |
| 473 | // The event-line replay above must also survive compaction into a |
| 474 | // final Snapshot record containing both handoff variants. |
| 475 | reloaded.record_snapshot(&runs); |
| 476 | let reopened = WorkflowWorkspaceState::open(tmp.path()); |
| 477 | let compacted = reopened |
| 478 | .runs |
| 479 | .lock() |
| 480 | .expect("runs lock") |
| 481 | .get("workflow_abc") |
| 482 | .cloned() |
| 483 | .expect("snapshot with handoff receipts"); |
| 484 | assert_eq!( |
| 485 | compacted |
| 486 | .events |
| 487 | .iter() |
| 488 | .map(WorkflowUiEvent::event_type) |
| 489 | .collect::<Vec<_>>(), |
| 490 | vec!["phase_started", "handoff_promoted", "handoff_consumed"] |
| 491 | ); |
| 492 | } |
| 493 | |
| 494 | #[test] |
| 495 | fn workflow_journal_rebuilds_a_bounded_exact_rejection_ledger() { |
| 496 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 497 | let state = WorkflowWorkspaceState::open(tmp.path()); |
| 498 | state.record_snapshot(&sample_record( |
| 499 | "workflow_rejections", |
| 500 | WorkflowRunStatus::Running, |
| 501 | )); |
| 502 | let total = WORKFLOW_RUN_DISPATCH_FAILURES_MAX_RETAINED + 5; |
| 503 | for index in 0..total { |
| 504 | let message = format!("invalid task options {index}"); |
| 505 | state.record_progress( |
| 506 | "workflow_rejections", |
| 507 | &format!("dispatch failed for rejected-{index}: {message}"), |
| 508 | ); |
| 509 | state.record_event( |
| 510 | "workflow_rejections", |
| 511 | &WorkflowUiEvent::at( |
| 512 | index as u64, |
| 513 | "session-journal", |
| 514 | WorkflowUiEventKind::TaskDispatchFailed { |
| 515 | label: Some(format!("rejected-{index}")), |
| 516 | phase: Some("fan-out".to_string()), |
| 517 | message, |
| 518 | }, |
| 519 | ), |
| 520 | ); |
| 521 | } |
| 522 | drop(state); |
| 523 | |
| 524 | let reloaded = WorkflowWorkspaceState::open(tmp.path()); |
| 525 | let run = reloaded |
| 526 | .runs |
| 527 | .lock() |
| 528 | .expect("runs lock") |
| 529 | .get("workflow_rejections") |
| 530 | .cloned() |
| 531 | .expect("hydrated rejection run"); |
| 532 | assert_eq!(run.progress_count, total as u64); |
| 533 | assert_eq!(run.progress.len(), total); |
| 534 | assert_eq!(run.dispatch_failure_count, total as u64); |
| 535 | assert_eq!( |
| 536 | run.dispatch_failures.len(), |
| 537 | WORKFLOW_RUN_DISPATCH_FAILURES_MAX_RETAINED |
| 538 | ); |
| 539 | assert_eq!( |
| 540 | run.dispatch_failures |
| 541 | .first() |
| 542 | .and_then(|failure| failure.label.as_deref()), |
| 543 | Some("rejected-5") |
| 544 | ); |
| 545 | drop(reloaded); |
| 546 | |
| 547 | // Restart recovery appends a compact snapshot. Replaying the |
| 548 | // journal again must not double-count its earlier event lines. |
| 549 | let reopened = WorkflowWorkspaceState::open(tmp.path()); |
| 550 | let run = reopened |
| 551 | .runs |
| 552 | .lock() |
| 553 | .expect("runs lock") |
| 554 | .get("workflow_rejections") |
| 555 | .cloned() |
| 556 | .expect("rehydrated rejection run"); |
| 557 | assert_eq!(run.dispatch_failure_count, total as u64); |
| 558 | assert_eq!( |
| 559 | run.dispatch_failures.len(), |
| 560 | WORKFLOW_RUN_DISPATCH_FAILURES_MAX_RETAINED |
| 561 | ); |
| 562 | } |
| 563 | |
| 564 | #[test] |
| 565 | fn workflow_journal_marks_orphaned_running_runs_failed() { |
| 566 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 567 | let state = WorkflowWorkspaceState::open(tmp.path()); |
| 568 | state.record_snapshot(&sample_record( |
| 569 | "workflow_orphan", |
| 570 | WorkflowRunStatus::Running, |
| 571 | )); |
| 572 | |
| 573 | let reloaded = WorkflowWorkspaceState::open(tmp.path()); |
| 574 | let run = reloaded |
| 575 | .runs |
| 576 | .lock() |
| 577 | .expect("runs lock") |
| 578 | .get("workflow_orphan") |
| 579 | .cloned() |
| 580 | .expect("hydrated run"); |
| 581 | assert_eq!(run.status, WorkflowRunStatus::Failed); |
| 582 | assert_eq!( |
| 583 | run.lifecycle_seq, 2, |
| 584 | "restart recovery is a durable owner lifecycle transition" |
| 585 | ); |
| 586 | assert!( |
| 587 | run.completed_at_ms.is_some(), |
| 588 | "restart recovery must terminalize the durable owner record" |
| 589 | ); |
| 590 | assert!( |
| 591 | run.error |
| 592 | .as_deref() |
| 593 | .is_some_and(|error| error.contains("process exited")), |
| 594 | "expected orphan recovery error, got {:?}", |
| 595 | run.error |
| 596 | ); |
| 597 | |
| 598 | let reopened = WorkflowWorkspaceState::open(tmp.path()); |
| 599 | let replayed = reopened |
| 600 | .runs |
| 601 | .lock() |
| 602 | .expect("runs lock") |
| 603 | .get("workflow_orphan") |
| 604 | .cloned() |
| 605 | .expect("durably recovered run"); |
| 606 | assert_eq!(replayed.status, WorkflowRunStatus::Failed); |
| 607 | assert_eq!( |
| 608 | replayed.lifecycle_seq, 2, |
| 609 | "reopening must replay the recovery snapshot without another transition" |
| 610 | ); |
| 611 | } |
| 612 | |
| 613 | #[test] |
| 614 | fn host_cancel_hydrates_a_journal_without_live_process_state() { |
| 615 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 616 | let state = WorkflowWorkspaceState::open(tmp.path()); |
| 617 | let record = sample_record("workflow_prior", WorkflowRunStatus::Running); |
| 618 | state.record_snapshot(&record); |
| 619 | drop(state); |
| 620 | |
| 621 | assert!( |
| 622 | peek_shared_workflow_state(tmp.path()).is_none(), |
| 623 | "writing the journal must not insert process-wide live state" |
| 624 | ); |
| 625 | |
| 626 | let line = super::super::host_cancel_workflow( |
| 627 | tmp.path(), |
| 628 | "workflow_prior", |
| 629 | Some("session-journal"), |
| 630 | ) |
| 631 | .expect("a journaled run must be visible to host cancel after restart"); |
| 632 | assert_eq!(line.run_id, "workflow_prior"); |
| 633 | assert_eq!(line.status, "cancelled"); |
| 634 | assert!( |
| 635 | line.error |
| 636 | .as_deref() |
| 637 | .is_some_and(|error| error.contains("no live process")), |
| 638 | "controller-less cancel must leave an honest receipt, got {:?}", |
| 639 | line.error |
| 640 | ); |
| 641 | |
| 642 | let reopened = WorkflowWorkspaceState::open(tmp.path()); |
| 643 | let replayed = reopened |
| 644 | .runs |
| 645 | .lock() |
| 646 | .expect("runs lock") |
| 647 | .get("workflow_prior") |
| 648 | .cloned() |
| 649 | .expect("cancelled journal line"); |
| 650 | assert_eq!(replayed.status, WorkflowRunStatus::Cancelled); |
| 651 | } |
| 652 | |
| 653 | #[test] |
| 654 | fn host_stage_is_derived_from_typed_owner_events() { |
| 655 | let mut record = sample_record("workflow_stage", WorkflowRunStatus::Running); |
| 656 | record.push_event(WorkflowUiEvent::at( |
| 657 | 1, |
| 658 | "session-journal", |
| 659 | WorkflowUiEventKind::RunStarted { |
| 660 | workflow_id: Some("fixture".to_string()), |
| 661 | workflow_goal: Some("review release".to_string()), |
| 662 | source_path: None, |
| 663 | token_budget: None, |
| 664 | }, |
| 665 | )); |
| 666 | assert_eq!(super::super::host_workflow_stage(&record), "queued"); |
| 667 | |
| 668 | record.push_event(WorkflowUiEvent::at( |
| 669 | 2, |
| 670 | "session-journal", |
| 671 | WorkflowUiEventKind::PhaseStarted { |
| 672 | title: "review".to_string(), |
| 673 | }, |
| 674 | )); |
| 675 | assert_eq!(super::super::host_workflow_stage(&record), "running"); |
| 676 | |
| 677 | record.push_event(WorkflowUiEvent::at( |
| 678 | 3, |
| 679 | "session-journal", |
| 680 | WorkflowUiEventKind::TaskStarted(Box::new(super::super::WorkflowTaskStartedEvent { |
| 681 | task_id: "reviewer-1".to_string(), |
| 682 | label: Some("reviewer".to_string()), |
| 683 | role: None, |
| 684 | profile: None, |
| 685 | model: None, |
| 686 | strength: None, |
| 687 | thinking: None, |
| 688 | requested_reasoning: None, |
| 689 | effective_reasoning: None, |
| 690 | resolved_role: Some("reviewer".to_string()), |
| 691 | resolved_profile: None, |
| 692 | resolved_provider: "local".to_string(), |
| 693 | resolved_model: "stub".to_string(), |
| 694 | route_source: "session".to_string(), |
| 695 | child_route: None, |
| 696 | worktree: false, |
| 697 | workspace: None, |
| 698 | git_branch: None, |
| 699 | parent_task_id: None, |
| 700 | depth: 0, |
| 701 | workflow_run_id: Some("workflow_stage".to_string()), |
| 702 | workflow_phase_id: Some("review".to_string()), |
| 703 | workflow_task_label: Some("reviewer".to_string()), |
| 704 | workflow_child_index: Some(0), |
| 705 | fleet_receipt: None, |
| 706 | })), |
| 707 | )); |
| 708 | assert_eq!(super::super::host_workflow_stage(&record), "waiting"); |
| 709 | |
| 710 | record.push_event(WorkflowUiEvent::at( |
| 711 | 4, |
| 712 | "session-journal", |
| 713 | WorkflowUiEventKind::TaskCompleted { |
| 714 | task_id: "reviewer-1".to_string(), |
| 715 | status: super::super::IrWorkflowRunStatus::Succeeded, |
| 716 | usage: None, |
| 717 | }, |
| 718 | )); |
| 719 | assert_eq!(super::super::host_workflow_stage(&record), "running"); |
| 720 | |
| 721 | record.status = WorkflowRunStatus::Completed; |
| 722 | assert_eq!(super::super::host_workflow_stage(&record), "completed"); |
| 723 | record.status = WorkflowRunStatus::Failed; |
| 724 | assert_eq!(super::super::host_workflow_stage(&record), "failed"); |
| 725 | record.status = WorkflowRunStatus::Cancelled; |
| 726 | assert_eq!(super::super::host_workflow_stage(&record), "cancelled"); |
| 727 | } |
| 728 | |
| 729 | #[test] |
| 730 | fn host_run_details_derive_phases_and_child_states_from_the_journal() { |
| 731 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 732 | let state = WorkflowWorkspaceState::open(tmp.path()); |
| 733 | let mut record = sample_record("workflow_detail", WorkflowRunStatus::Running); |
| 734 | record.workflow_goal = Some("audit provider errors".to_string()); |
| 735 | for message in ["phase: scan", "child slow-1 done", "child slow-2 failed"] { |
| 736 | record.push_progress(message.to_string()); |
| 737 | } |
| 738 | state.record_snapshot(&record); |
| 739 | drop(state); |
| 740 | |
| 741 | let phase: WorkflowUiEvent = serde_json::from_value(serde_json::json!({ |
| 742 | "at_ms": 1, |
| 743 | "owner_session_id": "session-journal", |
| 744 | "type": "phase_started", |
| 745 | "title": "scan" |
| 746 | })) |
| 747 | .expect("phase_started event"); |
| 748 | let started: WorkflowUiEvent = WorkflowUiEvent::at( |
| 749 | 2, |
| 750 | "session-journal", |
| 751 | WorkflowUiEventKind::TaskStarted(Box::new(super::super::WorkflowTaskStartedEvent { |
| 752 | task_id: "child-1".to_string(), |
| 753 | label: Some("slow-1".to_string()), |
| 754 | role: None, |
| 755 | profile: None, |
| 756 | model: None, |
| 757 | strength: None, |
| 758 | thinking: None, |
| 759 | requested_reasoning: None, |
| 760 | effective_reasoning: None, |
| 761 | resolved_role: Some("explore".to_string()), |
| 762 | resolved_profile: None, |
| 763 | resolved_provider: "deepseek".to_string(), |
| 764 | resolved_model: "deepseek-v4-flash".to_string(), |
| 765 | route_source: "session".to_string(), |
| 766 | child_route: None, |
| 767 | worktree: false, |
| 768 | workspace: None, |
| 769 | git_branch: None, |
| 770 | parent_task_id: None, |
| 771 | depth: 0, |
| 772 | workflow_run_id: Some("workflow_detail".to_string()), |
| 773 | workflow_phase_id: Some("scan".to_string()), |
| 774 | workflow_task_label: None, |
| 775 | workflow_child_index: Some(0), |
| 776 | fleet_receipt: None, |
| 777 | })), |
| 778 | ); |
| 779 | let completed: WorkflowUiEvent = serde_json::from_value(serde_json::json!({ |
| 780 | "at_ms": 3, |
| 781 | "owner_session_id": "session-journal", |
| 782 | "type": "task_completed", |
| 783 | "task_id": "child-1", |
| 784 | "status": "failed" |
| 785 | })) |
| 786 | .expect("task_completed event"); |
| 787 | let replay = WorkflowWorkspaceState::open(tmp.path()); |
| 788 | replay.record_event("workflow_detail", &phase); |
| 789 | replay.record_event("workflow_detail", &started); |
| 790 | replay.record_event("workflow_detail", &completed); |
| 791 | drop(replay); |
| 792 | |
| 793 | let details = super::super::host_workflow_run_details(tmp.path(), Some("session-journal")); |
| 794 | assert_eq!(details.len(), 1, "one journaled run"); |
| 795 | let detail = &details[0]; |
| 796 | assert_eq!(detail.line.run_id, "workflow_detail"); |
| 797 | // Journal-only `running` rows hydrate through restart-orphan |
| 798 | // recovery (the same rewrite `WorkflowWorkspaceState::open` |
| 799 | // applies), so the host projection reports the run as failed — |
| 800 | // live in-process runs keep `running` via the shared state. |
| 801 | assert_eq!(detail.line.status, "failed"); |
| 802 | assert_eq!(detail.line.label, "audit provider errors"); |
| 803 | assert_eq!(detail.phases, vec!["scan".to_string()]); |
| 804 | assert_eq!(detail.children.len(), 1); |
| 805 | let child = &detail.children[0]; |
| 806 | assert_eq!(child.task_id, "child-1"); |
| 807 | assert_eq!(child.label.as_deref(), Some("slow-1")); |
| 808 | assert_eq!(child.role.as_deref(), Some("explore")); |
| 809 | assert_eq!(child.model.as_deref(), Some("deepseek-v4-flash")); |
| 810 | assert_eq!(child.phase.as_deref(), Some("scan")); |
| 811 | assert_eq!( |
| 812 | child.state, "failed", |
| 813 | "terminal event must win over running" |
| 814 | ); |
| 815 | assert_eq!(detail.progress_tail.len(), 3); |
| 816 | assert!(!detail.has_result); |
| 817 | |
| 818 | // Session ownership fences the projection: a foreign session |
| 819 | // sees nothing, exactly like every other host control. |
| 820 | assert!( |
| 821 | super::super::host_workflow_run_details(tmp.path(), Some("session-other")).is_empty() |
| 822 | ); |
| 823 | } |
| 824 | } |
| 825 |