use super::{TuiEvent, send_critical, state};
use crossbeam_channel::Sender;
use std::{
sync::{
Arc, Mutex,
atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
mpsc,
},
thread::JoinHandle,
};
#[derive(Debug, Clone)]
pub(crate) struct CompactionActivityFinal {
pub(crate) id: crate::output::ActivityId,
pub(crate) status: crate::output::ActivityStatus,
pub(crate) metadata: crate::output::ActivityMetadata,
pub(crate) preview: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct CompactionRuntimeRefresh {
pub(crate) config: crate::config::EffectiveConfig,
pub(crate) session_id: String,
pub(crate) provider: String,
pub(crate) model: String,
}
#[derive(Debug, Clone)]
#[allow(clippy::large_enum_variant)]
pub(crate) enum WorkerFinalEvent {
Done,
RunCanceled {
prompt: String,
},
Error(String),
CompactionFinished {
result: Result<Option<crate::compaction::CompactionResult>, String>,
activity: CompactionActivityFinal,
runtime_config: Option<Box<CompactionRuntimeRefresh>>,
},
OAuthFinished {
provider_id: String,
result: Result<String, String>,
},
CustomProviderFinished {
provider_id: String,
result: Result<String, String>,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum WorkerShutdownPolicy {
Cancel,
WaitForCompletion,
}
#[derive(Debug, PartialEq, Eq)]
pub(super) enum WorkerFinalEventEffect {
None,
CompactionRuntimeRefresh(Box<CompactionRuntimeRefresh>),
CustomProviderFinished {
provider_id: String,
result: Result<String, String>,
},
}
impl WorkerFinalEvent {
pub(super) fn from_tui_event(event: TuiEvent) -> Option<Self> {
Some(match event {
TuiEvent::Done => Self::Done,
TuiEvent::RunCanceled { prompt } => Self::RunCanceled { prompt },
TuiEvent::Error(error) => Self::Error(error),
TuiEvent::CompactionFinishedAuthoritative {
result,
activity,
runtime_config,
} => Self::CompactionFinished {
result,
activity,
runtime_config,
},
TuiEvent::OAuthFinished {
provider_id,
result,
..
} => Self::OAuthFinished {
provider_id,
result,
},
TuiEvent::CustomProviderFinished {
provider_id,
result,
..
} => Self::CustomProviderFinished {
provider_id,
result,
},
_ => return None,
})
}
fn into_tui_event(self, worker_id: u64) -> TuiEvent {
match self {
Self::Done => TuiEvent::Done,
Self::RunCanceled { prompt } => TuiEvent::RunCanceled { prompt },
Self::Error(error) => TuiEvent::Error(error),
Self::CompactionFinished {
result,
activity,
runtime_config,
} => TuiEvent::CompactionFinishedAuthoritative {
result,
activity,
runtime_config,
},
Self::OAuthFinished {
provider_id,
result,
} => TuiEvent::OAuthFinished {
worker_id,
provider_id,
result,
},
Self::CustomProviderFinished {
provider_id,
result,
} => TuiEvent::CustomProviderFinished {
worker_id,
provider_id,
result,
},
}
}
}
#[derive(Debug)]
pub(super) struct WorkerOutcomeState {
worker_id: u64,
final_event: Mutex<Option<WorkerFinalEvent>>,
completion_delivery: AtomicU8,
pub(crate) usage: Arc<Mutex<super::session_usage::SessionUsageLedger>>,
}
const COMPLETION_DELIVERY_PENDING: u8 = 0;
const COMPLETION_DELIVERY_SENT: u8 = 1;
const COMPLETION_DELIVERY_FAILED: u8 = 2;
const UNEXPECTED_WORKER_FAILURE: &str = "worker exited unexpectedly before completion";
static NEXT_WORKER_ID: AtomicU64 = AtomicU64::new(1);
impl Default for WorkerOutcomeState {
fn default() -> Self {
Self {
worker_id: NEXT_WORKER_ID.fetch_add(1, Ordering::Relaxed),
final_event: Mutex::new(None),
completion_delivery: AtomicU8::new(COMPLETION_DELIVERY_PENDING),
usage: Default::default(),
}
}
}
impl WorkerOutcomeState {
pub(super) fn worker_id(&self) -> u64 {
self.worker_id
}
fn store_final_event(&self, event: Option<WorkerFinalEvent>) {
let mut final_event = self
.final_event
.lock()
.unwrap_or_else(|error| error.into_inner());
if self.completion_delivery.load(Ordering::Acquire) == COMPLETION_DELIVERY_PENDING {
*final_event = event;
}
}
pub(super) fn final_event(&self) -> Option<WorkerFinalEvent> {
self.final_event
.lock()
.unwrap_or_else(|error| error.into_inner())
.clone()
}
fn mark_completion_delivered(&self) -> bool {
self.completion_delivery
.compare_exchange(
COMPLETION_DELIVERY_PENDING,
COMPLETION_DELIVERY_SENT,
Ordering::SeqCst,
Ordering::SeqCst,
)
.is_ok()
}
pub(super) fn mark_completion_delivery_failed(&self) -> bool {
self.completion_delivery
.compare_exchange(
COMPLETION_DELIVERY_PENDING,
COMPLETION_DELIVERY_FAILED,
Ordering::SeqCst,
Ordering::SeqCst,
)
.is_ok()
}
pub(super) fn mark_unexpected_worker_failure_if_pending(&self) -> bool {
let mut final_event = self
.final_event
.lock()
.unwrap_or_else(|error| error.into_inner());
let marked = self
.completion_delivery
.compare_exchange(
COMPLETION_DELIVERY_PENDING,
COMPLETION_DELIVERY_FAILED,
Ordering::SeqCst,
Ordering::SeqCst,
)
.is_ok();
if marked {
*final_event = Some(WorkerFinalEvent::Error(
UNEXPECTED_WORKER_FAILURE.to_string(),
));
}
marked
}
pub(super) fn completion_delivery_failed(&self) -> bool {
self.completion_delivery.load(Ordering::SeqCst) == COMPLETION_DELIVERY_FAILED
}
}
pub(super) struct WorkerState {
pub(super) handle: JoinHandle<()>,
pub(super) cancel: Arc<AtomicBool>,
pub(super) login_manual: Option<mpsc::Sender<String>>,
pub(super) outcome: Arc<WorkerOutcomeState>,
pub(super) outcome_reconciled: bool,
pub(super) shutdown_policy: WorkerShutdownPolicy,
pub(super) steering: crate::agent::steering::AgentSteering,
pub(super) accepts_steering: bool,
}
pub(super) fn send_completion(
sender: &Sender<TuiEvent>,
outcome: &WorkerOutcomeState,
final_event: Option<WorkerFinalEvent>,
) {
outcome.store_final_event(final_event.clone());
let worker_id = outcome.worker_id();
let final_event = final_event.map(|event| Box::new(event.into_tui_event(worker_id)));
match send_critical(
sender,
TuiEvent::RunFinished {
worker_id,
final_event,
},
) {
Ok(()) => {
let _ = outcome.mark_completion_delivered();
}
Err(error) => {
if outcome.mark_completion_delivery_failed() {
let _ = sender.try_send(TuiEvent::WorkerCompletionDeliveryFailed {
worker_id,
error: error.to_string(),
});
}
}
}
}
pub(super) fn apply_worker_final_event(
state: &mut state::MissionControlState,
event: WorkerFinalEvent,
) -> WorkerFinalEventEffect {
match event {
WorkerFinalEvent::Done => {
state.highlight_live_completion(std::time::Instant::now());
state.finish_assistant_streaming();
state.status = "run complete".to_string();
WorkerFinalEventEffect::None
}
WorkerFinalEvent::RunCanceled { prompt } => {
state.activity_motion = Default::default();
state.finish_assistant_streaming();
let preview = crate::tui::transcript::sanitize_preview(&prompt);
state.mark_running_prompt_canceled();
state.record_canceled_transcript(&preview);
state.status = format!("canceled: {preview}");
WorkerFinalEventEffect::None
}
WorkerFinalEvent::Error(error) => {
state.activity_motion = Default::default();
state.finish_assistant_streaming();
if error.starts_with("provider not configured for '")
|| error.starts_with("openai-codex credential refresh failed:")
{
state.provider_ready = false;
}
state.record_error_transcript(&error);
state.status = error;
WorkerFinalEventEffect::None
}
WorkerFinalEvent::CompactionFinished {
result,
activity,
runtime_config,
} => {
state.apply_activity_event(crate::output::ActivityEvent::Started {
id: activity.id.clone(),
parent_id: None,
kind: crate::output::ActivityKind::Compaction,
status: crate::output::ActivityStatus::Running,
metadata: activity.metadata.clone(),
});
if let Some(preview) = activity.preview {
state.apply_activity_event(crate::output::ActivityEvent::FinalPreview {
id: activity.id.clone(),
preview,
metadata: Some(activity.metadata.clone()),
status: Some(activity.status),
});
}
state.apply_activity_event(crate::output::ActivityEvent::Finished {
id: activity.id,
status: activity.status,
metadata: Some(activity.metadata),
});
match result {
Ok(Some(result)) => {
let fast_observation =
result
.requested_service_tier
.as_deref()
.map(|requested_service_tier| crate::fast::FastObservation {
requested_service_tier: requested_service_tier.to_string(),
outcome: result.fast_outcome.clone(),
request_sequence: 1,
run_order: None,
});
if let Some(observation) = fast_observation.as_ref() {
state.apply_output_event(&crate::output::OutputEvent::FastObservation {
provider_id: result.provider.clone(),
model: result.model.clone(),
requested_service_tier: observation.requested_service_tier.clone(),
outcome: observation.outcome.clone(),
request_sequence: observation.request_sequence,
run_order: Some(super::NEXT_TUI_RUN_ID.fetch_add(1, Ordering::Relaxed)),
});
}
state.record_compaction_complete_transcript(Some(&result.summary));
state.status = result
.rotation_warning
.clone()
.map(|warning| crate::output::sanitize_display_text(&warning))
.unwrap_or_else(|| {
fast_observation
.as_ref()
.and_then(|observation| {
crate::fast::fast_observation_warning(
&result.provider,
&result.model,
observation,
)
})
.map(|warning| crate::output::sanitize_display_text(&warning))
.unwrap_or_else(|| {
crate::compaction::compaction_success_message(&result)
})
});
}
Ok(None) => {
let message =
"compaction produced empty summary; no checkpoint was written".to_string();
state.finish_compaction_failure_transcript(&message, false);
state.status = message;
}
Err(error) => {
let canceled = error.to_ascii_lowercase().contains("cancel");
state.finish_compaction_failure_transcript(&error, canceled);
state.status = error;
}
}
runtime_config
.map(WorkerFinalEventEffect::CompactionRuntimeRefresh)
.unwrap_or(WorkerFinalEventEffect::None)
}
WorkerFinalEvent::OAuthFinished {
provider_id,
result,
} => {
match result {
Ok(message) => {
if state.set_connect_provider_success(&provider_id, message.clone(), false) {
if state.provider == provider_id {
state.provider_ready = true;
}
state.status = message;
}
}
Err(error) => {
let error = crate::output::redact_sensitive_text(&error);
if state.set_connect_provider_error(error.clone()) {
state.status = error;
}
}
}
WorkerFinalEventEffect::None
}
WorkerFinalEvent::CustomProviderFinished {
provider_id,
result,
} => {
let result = result
.map(|message| crate::output::redact_sensitive_text(&message))
.map_err(|error| crate::output::redact_sensitive_text(&error));
match &result {
Ok(message) => {
let _ = state.set_connect_provider_success(&provider_id, message.clone(), true);
state.status = message.clone();
}
Err(error) => {
let _ = state.set_connect_provider_error(error.clone());
state.status = error.clone();
}
}
WorkerFinalEventEffect::CustomProviderFinished {
provider_id,
result,
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crossbeam_channel::bounded;
#[test]
fn send_completion_emits_single_composite_event() {
let (sender, receiver) = bounded::<TuiEvent>(1);
let outcome = WorkerOutcomeState::default();
send_completion(&sender, &outcome, Some(WorkerFinalEvent::Done));
assert!(!outcome.completion_delivery_failed());
assert!(!outcome.mark_unexpected_worker_failure_if_pending());
assert!(matches!(
outcome.final_event(),
Some(WorkerFinalEvent::Done)
));
assert!(matches!(
receiver.try_recv().unwrap(),
TuiEvent::RunFinished {
worker_id: _,
final_event: Some(event),
} if matches!(*event, TuiEvent::Done)
));
assert!(receiver.try_recv().is_err());
}
#[test]
fn final_event_lock_recovers_after_poison() {
let outcome = WorkerOutcomeState::default();
let _ = std::panic::catch_unwind(|| {
let _guard = outcome.final_event.lock().unwrap();
panic!("poison final event lock");
});
outcome.store_final_event(Some(WorkerFinalEvent::Error("kept".to_string())));
assert!(matches!(
outcome.final_event(),
Some(WorkerFinalEvent::Error(message)) if message == "kept"
));
}
#[test]
fn apply_worker_final_event_records_compaction_summary_transcript() {
let mut state = state::MissionControlState::default();
apply_worker_final_event(
&mut state,
WorkerFinalEvent::CompactionFinished {
result: Ok(Some(crate::compaction::CompactionResult {
session_id: "session-1".to_string(),
summary: "worker summary".to_string(),
provider: "local".to_string(),
model: "compact-model".to_string(),
requested_service_tier: None,
fast_outcome: crate::fast::FastOutcome::NotRequested,
rotation_warning: None,
})),
activity: CompactionActivityFinal {
id: crate::output::ActivityId::new("compaction:test"),
status: crate::output::ActivityStatus::Success,
metadata: crate::output::ActivityMetadata::new("compaction"),
preview: Some("worker summary".to_string()),
},
runtime_config: None,
},
);
assert_eq!(state.transcript, vec!["compact: complete • worker summary"]);
assert!(
state
.status
.contains("compaction complete: session session-1")
);
}
#[test]
fn fast_warning_transcript_survives_done_completion() {
let mut state = state::MissionControlState::default();
state.apply_output_event(&crate::output::OutputEvent::FastObservation {
provider_id: "provider".to_string(),
model: "model".to_string(),
requested_service_tier: "priority".to_string(),
outcome: crate::fast::FastOutcome::Different("default".to_string()),
request_sequence: 1,
run_order: Some(1),
});
let warning = state.status.clone();
assert!(
state
.transcript
.iter()
.any(|line| line.starts_with("warning: "))
);
apply_worker_final_event(&mut state, WorkerFinalEvent::Done);
assert_eq!(state.status, "run complete");
assert!(
state
.transcript
.iter()
.any(|line| line == &format!("warning: {warning}"))
);
assert_eq!(
state
.transcript
.iter()
.filter(|line| line.starts_with("warning: "))
.count(),
1
);
}
#[test]
fn compaction_finished_success_reconciles_one_authoritative_activity() {
let (sender, receiver) = bounded::<TuiEvent>(1);
let outcome = WorkerOutcomeState::default();
send_completion(
&sender,
&outcome,
Some(WorkerFinalEvent::CompactionFinished {
result: Ok(Some(crate::compaction::CompactionResult {
session_id: "session-1".to_string(),
summary: "authoritative summary".to_string(),
provider: "local".to_string(),
model: "compact-model".to_string(),
requested_service_tier: None,
fast_outcome: crate::fast::FastOutcome::NotRequested,
rotation_warning: None,
})),
activity: CompactionActivityFinal {
id: crate::output::ActivityId::new("compaction:success"),
status: crate::output::ActivityStatus::Success,
metadata: crate::output::ActivityMetadata::new("compaction"),
preview: Some("authoritative summary".to_string()),
},
runtime_config: None,
}),
);
let TuiEvent::RunFinished {
worker_id: _,
final_event: Some(event),
} = receiver.try_recv().unwrap()
else {
panic!("missing authoritative completion event");
};
let mut state = state::MissionControlState::default();
apply_worker_final_event(
&mut state,
match *event {
TuiEvent::CompactionFinishedAuthoritative {
result,
activity,
runtime_config,
} => WorkerFinalEvent::CompactionFinished {
result,
activity,
runtime_config,
},
_ => panic!("unexpected completion event"),
},
);
let id = crate::output::ActivityId::new("compaction:success");
assert_eq!(state.roots, vec![id.clone()]);
assert_eq!(state.nodes.len(), 1);
assert_eq!(
state.nodes[&id].status,
crate::output::ActivityStatus::Success
);
assert_eq!(state.nodes[&id].preview, "authoritative summary");
assert_eq!(
state.transcript,
vec!["compact: complete • authoritative summary"]
);
}
#[test]
fn compaction_finished_fallback_reconciles_failed_and_canceled_activity() {
let cases: [(crate::output::ActivityStatus, &str, Option<&str>); 2] = [
(
crate::output::ActivityStatus::Failed,
"compaction failed",
Some("compaction failed"),
),
(
crate::output::ActivityStatus::Canceled,
"compaction canceled",
None,
),
];
for (status, error, preview) in cases {
let (sender, _receiver) = bounded::<TuiEvent>(1);
sender.send(TuiEvent::Done).unwrap();
let outcome = WorkerOutcomeState::default();
let id = crate::output::ActivityId::new(format!("compaction:{status:?}"));
send_completion(
&sender,
&outcome,
Some(WorkerFinalEvent::CompactionFinished {
result: Err(error.to_string()),
activity: CompactionActivityFinal {
id: id.clone(),
status,
metadata: crate::output::ActivityMetadata::new("compaction"),
preview: preview.map(str::to_string),
},
runtime_config: None,
}),
);
let mut state = state::MissionControlState::default();
apply_worker_final_event(&mut state, outcome.final_event().unwrap());
assert_eq!(state.roots, vec![id.clone()]);
assert_eq!(state.nodes.len(), 1);
assert_eq!(state.nodes[&id].status, status);
assert_eq!(state.nodes[&id].metadata.label, "compaction");
assert_eq!(state.nodes[&id].preview, preview.unwrap_or_default());
assert!(state.transcript.iter().any(|line| line.contains(error)));
}
}
#[test]
fn compaction_finished_fallback_reconciles_empty_summary_and_redacts_failure() {
let cases = [
(
Ok(None),
crate::output::ActivityStatus::Failed,
None,
"empty summary; no checkpoint was written",
None,
"error: compaction produced empty summary; no checkpoint was written",
),
(
Err("provider failed api_key=secret-value".to_string()),
crate::output::ActivityStatus::Failed,
Some("provider failed api_key=secret-value"),
"failed; no checkpoint was written",
Some("provider failed api_key=<redacted>"),
"error: provider failed api_key=<redacted>",
),
];
for (result, status, preview_input, note, expected_preview, transcript) in cases {
let (sender, _receiver) = bounded::<TuiEvent>(1);
sender.send(TuiEvent::Done).unwrap();
let outcome = WorkerOutcomeState::default();
let id = crate::output::ActivityId::new(format!("compaction:fallback:{status:?}"));
let metadata = crate::output::ActivityMetadata {
label: "compaction".to_string(),
detail: Some(note.to_string()),
fields: vec![("note".to_string(), note.to_string())],
};
let preview = preview_input.map(crate::output::redact_sensitive_text);
send_completion(
&sender,
&outcome,
Some(WorkerFinalEvent::CompactionFinished {
result,
activity: CompactionActivityFinal {
id: id.clone(),
status,
metadata,
preview,
},
runtime_config: None,
}),
);
let mut state = state::MissionControlState::default();
apply_worker_final_event(&mut state, outcome.final_event().unwrap());
assert_eq!(state.nodes.len(), 1);
assert_eq!(state.roots, vec![id.clone()]);
assert_eq!(
state.nodes[&id].kind,
crate::output::ActivityKind::Compaction
);
assert_eq!(state.nodes[&id].status, status);
assert_eq!(state.nodes[&id].metadata.detail.as_deref(), Some(note));
assert_eq!(
state.nodes[&id].preview,
expected_preview.unwrap_or_default()
);
assert!(state.transcript.iter().any(|line| line == transcript));
assert!(
state
.transcript
.iter()
.all(|line| !line.contains("secret-value"))
);
}
}
}