diff --git a/crates/session-store/src/worker_metadata.rs b/crates/session-store/src/worker_metadata.rs index 52bca7ae..7fd8cdf5 100644 --- a/crates/session-store/src/worker_metadata.rs +++ b/crates/session-store/src/worker_metadata.rs @@ -795,6 +795,26 @@ where fn read_by_name(&self, worker_name: &str) -> Result, WorkerStoreError> { self.worker_metadata_store.read_by_name(worker_name) } + fn update_by_name( + &self, + worker_name: &str, + update: F, + ) -> Result + where + F: FnOnce(&mut WorkerMetadata), + { + self.worker_metadata_store + .update_by_name(worker_name, update) + } + fn compare_and_swap_active( + &self, + worker_name: &str, + expected: &WorkerActiveSegmentRef, + replacement: WorkerActiveSegmentRef, + ) -> Result { + self.worker_metadata_store + .compare_and_swap_active(worker_name, expected, replacement) + } fn list_names(&self) -> Result, WorkerStoreError> { self.worker_metadata_store.list_names() } @@ -1030,6 +1050,46 @@ mod tests { assert_eq!(restored.reclaimed_children[0].scope_delegated, vec![scope]); } + #[test] + fn combined_store_delegates_atomic_active_segment_cas() { + let temp = tempfile::tempdir().unwrap(); + let metadata = FsWorkerStore::new(temp.path().join("workers")).unwrap(); + let store = CombinedStore::new( + crate::FsStore::new(temp.path().join("sessions")).unwrap(), + metadata, + ); + let session_id = crate::new_session_id(); + let old = WorkerActiveSegmentRef::active_segment(session_id, crate::new_segment_id()); + store + .write(&WorkerMetadata::new("agent", Some(old.clone()))) + .unwrap(); + let barrier = Arc::new(std::sync::Barrier::new(3)); + let handles = [crate::new_segment_id(), crate::new_segment_id()].map(|segment_id| { + let store = store.clone(); + let old = old.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + barrier.wait(); + store + .compare_and_swap_active( + "agent", + &old, + WorkerActiveSegmentRef::active_segment(session_id, segment_id), + ) + .unwrap() + }) + }); + barrier.wait(); + assert_eq!( + handles + .into_iter() + .map(|handle| handle.join().unwrap()) + .filter(|won| *won) + .count(), + 1 + ); + } + #[test] fn active_segment_cas_allows_exactly_one_concurrent_winner() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 86722b63..be192281 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -1403,7 +1403,21 @@ impl App { } } Event::CompactionProgress { compaction } => { - self.compaction_progress = compaction; + self.compaction_progress = compaction.filter(|progress| { + matches!( + (&self.worker_state.state, progress.trigger), + ( + protocol::WorkerState::Busy(protocol::WorkerBusyState::Maintenance( + protocol::WorkerMaintenanceState::Compacting + )), + protocol::CompactionTrigger::Manual + ) | ( + protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(_)), + protocol::CompactionTrigger::PreRun + | protocol::CompactionTrigger::RequestThreshold + ) + ) + }); } Event::CompactStart { lifecycle } => { let should_apply = match &self.active_compaction { @@ -1522,9 +1536,19 @@ impl App { Event::WorkerState { snapshot } => { self.rewind_refresh_fence = false; self.apply_worker_state_snapshot(&snapshot); + if let Some(progress) = self.compaction_progress.take() { + let _ = self.handle_worker_event(Event::CompactionProgress { + compaction: Some(progress), + }); + } } Event::CommandAcknowledged { acknowledgement } => { self.apply_worker_state_snapshot(&acknowledgement.state); + if let Some(progress) = self.compaction_progress.take() { + let _ = self.handle_worker_event(Event::CompactionProgress { + compaction: Some(progress), + }); + } } // Command telemetry is an operational Web Console surface. The // TUI continues to render the final Bash ToolResult from history. @@ -1694,7 +1718,7 @@ impl App { } } self.active_compaction = None; - self.compaction_progress = compaction; + let _ = self.handle_worker_event(Event::CompactionProgress { compaction }); } fn append_assistant_text(&mut self, text: &str) { @@ -4299,9 +4323,25 @@ mod completion_flow_tests { )); } + #[test] + fn compaction_progress_is_hidden_when_worker_state_is_inconsistent() { + let mut app = App::new("test".into()); + app.handle_worker_event(Event::CompactionProgress { + compaction: Some(protocol::InFlightCompaction { + phase: protocol::CompactionPhase::Preparing, + started_at_ms: 100, + trigger: protocol::CompactionTrigger::Manual, + }), + }); + assert!(app.compaction_progress.is_none()); + } + #[test] fn snapshot_restores_and_runtime_clear_removes_compaction_progress() { let mut app = App::new("test".into()); + app.worker_state.state = protocol::WorkerState::Busy( + protocol::WorkerBusyState::Maintenance(protocol::WorkerMaintenanceState::Compacting), + ); app.apply_in_flight_snapshot(InFlightSnapshot { compaction: Some(protocol::InFlightCompaction { phase: protocol::CompactionPhase::Summarizing, diff --git a/crates/tui/src/ui.rs b/crates/tui/src/ui.rs index b23209df..b7496a3e 100644 --- a/crates/tui/src/ui.rs +++ b/crates/tui/src/ui.rs @@ -139,10 +139,17 @@ fn draw_run_status(frame: &mut Frame, app: &App, area: Rect) { } fn run_status_line(app: &App, now: Instant) -> Line<'static> { - let elapsed = app - .run_started_at - .and_then(|started_at| now.checked_duration_since(started_at)) - .unwrap_or_default(); + let elapsed = if let Some(progress) = &app.compaction_progress { + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64; + std::time::Duration::from_millis(now_ms.saturating_sub(progress.started_at_ms)) + } else { + app.run_started_at + .and_then(|started_at| now.checked_duration_since(started_at)) + .unwrap_or_default() + }; let spinner_index = ((elapsed.as_millis() / RUN_SPINNER_FRAME_MS) as usize) % RUN_SPINNER_FRAMES.len(); let request_label = if app.run_requests == 1 { @@ -161,8 +168,13 @@ fn run_status_line(app: &App, now: Instant) -> Line<'static> { Span::raw(" "), ]; if let Some(progress) = &app.compaction_progress { + let phase = match progress.phase { + protocol::CompactionPhase::Preparing => "preparing", + protocol::CompactionPhase::Summarizing => "summarizing", + protocol::CompactionPhase::Committing => "committing", + }; spans.push(Span::styled( - format!("Compacting · {:?}", progress.phase).to_lowercase(), + format!("Compacting · {phase}"), Style::default().fg(Color::Cyan), )); spans.push(Span::styled(" | ", Style::default().fg(Color::DarkGray))); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index dc713560..eaffebd5 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -5610,18 +5610,12 @@ impl Worker { session_id: old_loc.session_id, segment_id: new_segment_id, }; - // Move the live writer lease under the same exclusive compaction owner - // before publishing the durable pointer. If the CAS loses, restore the - // derived lease and leave the staged Segment unreachable. - if self.scope_allocation.is_some() { - worker_allocation::update_segment(&self.manifest.worker.name, new_segment_id)?; - } - if let Err(error) = self.compare_and_swap_worker_metadata_segment(old_loc, new_location) { - if self.scope_allocation.is_some() { - worker_allocation::update_segment(&self.manifest.worker.name, old_loc.segment_id)?; - } - return Err(error); - } + // The writer lease is stable for the Worker lifetime and is keyed by + // worker name. Compaction must not transfer or rewrite that lease: the + // active Segment is derived exclusively from the CAS-protected Worker + // metadata pointer below. A lost CAS therefore leaves every live and + // durable authority on the previous Segment. + self.compare_and_swap_worker_metadata_segment(old_loc, new_location)?; // All live mutations after the durable commit are infallible and happen // before the replacement SegmentStart is broadcast. This keeps the diff --git a/web/workspace/src/lib/workspace/console/WorkerRunStatus.svelte b/web/workspace/src/lib/workspace/console/WorkerRunStatus.svelte index b91b08c1..b057d9f9 100644 --- a/web/workspace/src/lib/workspace/console/WorkerRunStatus.svelte +++ b/web/workspace/src/lib/workspace/console/WorkerRunStatus.svelte @@ -1,4 +1,5 @@ diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte index 2910d25c..5b5ff42d 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/+page.svelte @@ -248,6 +248,18 @@ liveWorkerState ?? (worker?.state === "stopped" ? "stopped" : "loading"), ); const workerRunning = $derived(workerState === "running"); + const compactionProgress = $derived.by(() => { + const state = consoleProjection.workerState?.state; + if (!state || typeof state !== "object" || !("busy" in state)) return null; + const progress = consoleProjection.compaction; + if (!progress) return null; + const busy = state.busy; + if (!busy || typeof busy !== "object") return null; + const valid = progress.trigger === "manual" + ? "maintenance" in busy && busy.maintenance === "compacting" + : "run" in busy; + return valid ? progress : null; + }); const workerPaused = $derived(workerState === "paused"); const composerEditable = $derived(protocolState === "open" && !sending); const draftHasText = $derived(draft.content.trim().length > 0); @@ -1907,7 +1919,7 @@ requests={consoleProjection.runActivity.requests} uploadTokens={consoleProjection.runActivity.uploadTokens} outputTokens={consoleProjection.runActivity.outputTokens} - compaction={consoleProjection.compaction} + compaction={compactionProgress} /> {/if}