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)]
#[expect(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,
}
}
}
}