diff --git a/apps/decodex/src/orchestrator/daemon.rs b/apps/decodex/src/orchestrator/daemon.rs index 74144d49..05403890 100644 --- a/apps/decodex/src/orchestrator/daemon.rs +++ b/apps/decodex/src/orchestrator/daemon.rs @@ -204,6 +204,9 @@ where if !warnings.is_empty() { hydrate_history_lanes_from_local_ledger(project, state_store, &mut snapshot)?; } + + apply_terminal_history_ledger_outcomes(&mut snapshot); + if warnings.contains(&TRACKER_RATE_LIMIT_WARNING) { let review_state_inspector = GhPullRequestReviewStateInspector { github_token_env_var: Some(project.github().token_env_var().to_owned()), diff --git a/apps/decodex/src/orchestrator/entrypoints.rs b/apps/decodex/src/orchestrator/entrypoints.rs index ab67e343..2a36f29c 100644 --- a/apps/decodex/src/orchestrator/entrypoints.rs +++ b/apps/decodex/src/orchestrator/entrypoints.rs @@ -655,6 +655,7 @@ fn build_operator_status_snapshot_for_tracker_backoff( snapshot.connector_backoffs.push(status.clone()); add_operator_snapshot_warning(&mut snapshot, "external_observer_status_skipped"); + apply_terminal_history_ledger_outcomes(&mut snapshot); refresh_operator_project_summary(&mut snapshot); Ok(snapshot) @@ -1328,6 +1329,7 @@ fn build_operator_state_snapshot_without_live_observers( )?; hydrate_history_lanes_from_local_ledger(project, state_store, &mut snapshot)?; + apply_terminal_history_ledger_outcomes(&mut snapshot); refresh_operator_project_summary(&mut snapshot); Ok(snapshot) diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index 0bb80b60..dcca4a07 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -530,6 +530,7 @@ where }, &mut snapshot, )?; + apply_terminal_history_ledger_outcomes(&mut snapshot); refresh_worktree_ownership( &mut snapshot, Some(workflow.frontmatter().tracker().resolved_completed_state()), @@ -782,6 +783,92 @@ fn hydrate_history_lanes_from_local_ledger( Ok(()) } +fn apply_terminal_history_ledger_outcomes(snapshot: &mut OperatorStatusSnapshot) { + let mut terminal_history_keys = HashSet::new(); + + for lane in &mut snapshot.history_lanes { + if !history_ledger_outcome_supersedes_local_attempts(&lane.ledger_outcome) { + continue; + } + + terminal_history_keys.insert(history_lane_group_key(lane)); + + apply_terminal_history_ledger_outcome_to_latest_run(lane); + } + + if terminal_history_keys.is_empty() { + return; + } + + let active_run_ids = snapshot + .active_runs + .iter() + .map(|run| run.run_id.clone()) + .collect::>(); + let active_issue_keys = snapshot + .active_runs + .iter() + .map(operator_run_group_key) + .collect::>(); + + snapshot.recent_runs.retain(|run| { + let run_group_key = operator_run_group_key(run); + + active_run_ids.contains(&run.run_id) + || active_issue_keys.contains(&run_group_key) + || !terminal_history_keys.contains(&run_group_key) + }); +} + +fn history_ledger_outcome_supersedes_local_attempts( + outcome: &OperatorHistoryLedgerOutcome, +) -> bool { + outcome.ledger_status == "present" + && matches!( + outcome.final_outcome.as_str(), + "cleanup_complete" | "closeout" | "landed" + ) +} + +fn apply_terminal_history_ledger_outcome_to_latest_run(lane: &mut OperatorHistoryLaneStatus) { + let final_outcome = lane.ledger_outcome.final_outcome.clone(); + let final_event_at = lane.ledger_outcome.final_event_at.clone(); + + lane.latest_run.status = final_outcome.clone(); + lane.latest_run.attempt_status = final_outcome; + lane.latest_run.phase = String::from("completed"); + lane.latest_run.wait_reason = None; + lane.latest_run.current_operation = String::from("ledger_outcome"); + lane.latest_run.continuation_pending = false; + lane.latest_run.active_lease = false; + lane.latest_run.queue_lease_state = String::from("not_held"); + lane.latest_run.execution_liveness = String::from("not_running"); + lane.latest_run.suspected_stall = false; + lane.latest_run.retry_kind = None; + lane.latest_run.next_retry_at = None; + + if let Some(final_event_at) = final_event_at { + lane.latest_run.updated_at = final_event_at.clone(); + lane.latest_run.last_run_activity_at = Some(final_event_at); + } +} + +fn history_lane_group_key(lane: &OperatorHistoryLaneStatus) -> String { + let issue_id = lane.issue_id.trim(); + + if !issue_id.is_empty() && !issue_id.eq_ignore_ascii_case("unknown") { + return issue_id.to_ascii_uppercase(); + } + + let issue_key = lane.issue_key.trim(); + + if !issue_key.is_empty() && !issue_key.eq_ignore_ascii_case("unknown") { + return issue_key.to_ascii_uppercase(); + } + + operator_run_group_key(&lane.latest_run) +} + fn refresh_worktree_ownership( snapshot: &mut OperatorStatusSnapshot, completed_state: Option<&str>, @@ -1114,6 +1201,7 @@ fn project_last_activity_at(snapshot: &OperatorStatusSnapshot) -> Option .active_runs .iter() .chain(snapshot.recent_runs.iter()) + .chain(snapshot.history_lanes.iter().map(|lane| &lane.latest_run)) .flat_map(|run| { [ run.last_progress_at.as_deref(), diff --git a/apps/decodex/src/orchestrator/tests/operator/status/history.rs b/apps/decodex/src/orchestrator/tests/operator/status/history.rs index 1cfa1d02..17468503 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/history.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/history.rs @@ -249,8 +249,8 @@ fn live_operator_history_lanes_prefer_linear_ledger_outcome() { .record_run_attempt("xy-355-attempt-1-1777527013", &issue.id, 1, "failed") .expect("failed attempt should record"); state_store - .record_run_attempt("xy-355-attempt-2-1777527613", &issue.id, 2, "succeeded") - .expect("successful attempt should record"); + .record_run_attempt("xy-355-attempt-2-1777527613", &issue.id, 2, "failed") + .expect("stale failed attempt should record"); state_store .clear_worktree(&issue.id) .expect("completed lane cleanup should clear local worktree"); @@ -268,14 +268,15 @@ fn live_operator_history_lanes_prefer_linear_ledger_outcome() { ) .expect("snapshot should build"); let lane = snapshot.history_lanes.first().expect("history lane should exist"); + let snapshot_json = serde_json::to_value(&snapshot).expect("snapshot should serialize"); let rendered = orchestrator::render_operator_status(&snapshot); let outcome_index = rendered.find("outcome: closeout").expect("ledger outcome should render"); let local_index = rendered.find("latest_run_id:").expect("local attempt debug should render"); - assert_eq!(snapshot.recent_runs.len(), 2); + assert!(snapshot.recent_runs.is_empty()); assert_eq!(snapshot.history_lanes.len(), 1); - assert!(snapshot.recent_runs.iter().all(|run| run.project_id == TEST_SERVICE_ID)); - assert!(snapshot.recent_runs.iter().all(|run| { + assert!(lane.attempts.iter().all(|run| run.project_id == TEST_SERVICE_ID)); + assert!(lane.attempts.iter().all(|run| { run.issue_identifier.as_deref() == Some("XY-355") && run.title.as_deref() == Some("Keep completed run rows self describing") })); @@ -284,6 +285,11 @@ fn live_operator_history_lanes_prefer_linear_ledger_outcome() { assert_eq!(lane.title.as_deref(), Some("Keep completed run rows self describing")); assert_eq!(lane.latest_run.issue_identifier.as_deref(), Some("XY-355")); assert_eq!(lane.latest_run.title.as_deref(), Some("Keep completed run rows self describing")); + assert_eq!(lane.latest_run.status, "closeout"); + assert_eq!(lane.latest_run.attempt_status, "closeout"); + assert_eq!(lane.latest_run.phase, "completed"); + assert_eq!(lane.latest_run.current_operation, "ledger_outcome"); + assert!(lane.attempts.iter().any(|attempt| attempt.status == "failed")); assert_eq!(lane.ledger_outcome.ledger_status, "present"); assert_eq!(lane.ledger_outcome.final_outcome, "closeout"); assert_eq!( @@ -306,7 +312,89 @@ fn live_operator_history_lanes_prefer_linear_ledger_outcome() { assert!(rendered.contains("commit_sha: 2222222222222222222222222222222222222222")); assert!(rendered.contains("closeout_status: Done")); assert!(rendered.contains("lifecycle_elapsed_seconds: 600")); + assert!(rendered.contains("attempt_timeline")); + assert!(rendered.contains("status: failed")); assert!(!rendered.contains("pr_url: none")); + assert_eq!( + snapshot_json["history_lanes"][0]["latest_run"]["status"], + "closeout" + ); + assert_eq!( + snapshot_json["history_lanes"][0]["attempts"][0]["status"], + "failed" + ); + assert_eq!( + snapshot_json["recent_runs"] + .as_array() + .expect("recent runs should be an array") + .len(), + 0 + ); +} + +#[test] +fn local_operator_history_lanes_prefer_terminal_ledger_outcome() { + let (_temp_dir, config, workflow) = temp_project_layout(); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let issue = sample_issue_with_sort_fields( + "issue-1", + "XY-799", + "Done", + &[], + Some(3), + "2026-06-08T04:12:00Z", + ); + let local_comments = successful_linear_execution_history_comments(&issue); + + state_store + .upsert_worktree( + TEST_SERVICE_ID, + &issue.id, + "y/decodex-xy-799", + &config.worktree_root().join(&issue.identifier).display().to_string(), + ) + .expect("worktree should remember project ownership"); + state_store + .record_run_attempt("xy-799-attempt-1-1780888320", &issue.id, 1, "failed") + .expect("stale failed attempt should record"); + state_store + .clear_worktree(&issue.id) + .expect("completed lane cleanup should clear local worktree"); + + seed_local_linear_execution_events(&state_store, &local_comments); + + let snapshot = orchestrator::build_operator_state_snapshot_without_live_observers( + &config, + &workflow, + &state_store, + 10, + ) + .expect("snapshot should build"); + let lane = snapshot.history_lanes.first().expect("history lane should exist"); + let snapshot_json = serde_json::to_value(&snapshot).expect("snapshot should serialize"); + + assert!(snapshot.recent_runs.is_empty()); + assert_eq!(lane.latest_run.status, "closeout"); + assert_eq!(lane.latest_run.attempt_status, "closeout"); + assert_eq!(lane.latest_run.phase, "completed"); + assert_eq!(lane.ledger_outcome.final_outcome, "closeout"); + assert_eq!(lane.attempts.len(), 1); + assert_eq!(lane.attempts[0].status, "failed"); + assert_eq!( + snapshot_json["history_lanes"][0]["latest_run"]["status"], + "closeout" + ); + assert_eq!( + snapshot_json["history_lanes"][0]["attempts"][0]["status"], + "failed" + ); + assert_eq!( + snapshot_json["recent_runs"] + .as_array() + .expect("recent runs should be an array") + .len(), + 0 + ); } #[test]