diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index fbbedf50..5af8355b 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -45,12 +45,12 @@ use workdir::{ #[derive(Clone)] pub struct WorkerHandle { method_tx: mpsc::Sender, - event_tx: broadcast::Sender, + working_event_tx: broadcast::Sender, pub shared_state: Arc, pub runtime_dir: Arc, pub alerter: Alerter, pub in_flight: InFlightEvents, - /// Segment-log mirror + broadcast handle. The IPC server snapshots + /// Segment-log mirror + session-entry channel. The IPC server snapshots /// it on every new connection (Event::Snapshot) and forwards /// subsequent commits (Event::Entry) on the receiver. pub sink: SegmentLogSink, @@ -63,7 +63,7 @@ impl WorkerHandle { } pub fn subscribe(&self) -> broadcast::Receiver { - self.event_tx.subscribe() + self.working_event_tx.subscribe() } pub fn committed_entries(&self) -> Vec { @@ -117,7 +117,7 @@ impl WorkerHandle { /// Broadcast an event to all listeners (including socket clients). pub fn send_event(&self, event: Event) -> Result> { - self.event_tx.send(event) + self.working_event_tx.send(event) } /// Emit a user-facing alert. Thin wrapper over `Alerter::alert`. @@ -129,19 +129,19 @@ impl WorkerHandle { async fn set_controller_status( shared_state: &Arc, runtime_dir: &RuntimeDir, - event_tx: &broadcast::Sender, + working_event_tx: &broadcast::Sender, status: WorkerStatus, ) { shared_state.set_status(status); let _ = runtime_dir.write_status(shared_state).await; - let _ = event_tx.send(Event::Status { status }); + let _ = working_event_tx.send(Event::Status { status }); } async fn finish_controller_run( worker: &mut Worker, shared_state: &Arc, runtime_dir: &RuntimeDir, - event_tx: &broadcast::Sender, + working_event_tx: &broadcast::Sender, new_status: WorkerStatus, ) where C: LlmClient + Clone + 'static, @@ -157,7 +157,7 @@ async fn finish_controller_run( // the terminal run boundary so reconnect snapshots cannot append stale // partial text/tool arguments after newer entries. worker.clear_in_flight_events(); - set_controller_status(shared_state, runtime_dir, event_tx, new_status).await; + set_controller_status(shared_state, runtime_dir, working_event_tx, new_status).await; worker.spawn_post_run_memory_jobs(); } @@ -353,9 +353,9 @@ impl WorkerController { // bash-output scope) === let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); let (method_tx, method_rx) = mpsc::channel::(32); - let (event_tx, _) = broadcast::channel::(256); - let alerter = Alerter::new(event_tx.clone()); - let in_flight = InFlightEvents::new(event_tx.clone()); + let (working_event_tx, _) = broadcast::channel::(256); + let alerter = Alerter::new(working_event_tx.clone()); + let in_flight = InFlightEvents::new(working_event_tx.clone()); worker.attach_in_flight_events(in_flight.clone()); // Runtime directory is created before tool registration because it owns @@ -395,7 +395,7 @@ impl WorkerController { // Also hand the raw broadcast sender so Worker-internal operations // can emit typed lifecycle `Event`s (currently: compact progress). worker.attach_internal_worker_registry(spawned_registry.clone()); - worker.attach_event_tx(event_tx.clone()); + worker.attach_working_event_tx(working_event_tx.clone()); // Bash spills long outputs to a per-worker subdir under the runtime // dir. Push a recursive `allow(Read)` for that path into the @@ -430,7 +430,7 @@ impl WorkerController { worker.wire_history_persistence(); // === 2. Engine event bridge wiring === - wire_event_bridges_on_engine(&mut worker, &event_tx, &alerter, &in_flight); + wire_event_bridges_on_engine(&mut worker, &working_event_tx, &alerter, &in_flight); // === 3. Tool registration (builtin / memory / spawn-orchestration) === let fs_for_view = register_worker_tools( @@ -477,7 +477,7 @@ impl WorkerController { let handle = WorkerHandle { method_tx, - event_tx: event_tx.clone(), + working_event_tx: working_event_tx.clone(), shared_state: shared_state.clone(), runtime_dir: runtime_dir.clone(), alerter: alerter.clone(), @@ -502,7 +502,7 @@ impl WorkerController { tokio::spawn(controller_loop( worker, method_rx, - event_tx, + working_event_tx, shared_state, runtime_dir, cancel_tx, @@ -640,7 +640,7 @@ fn protocol_command_status(status: WorkdirCommandStatus) -> ProtocolCommandStatu } /// Wire the per-event broadcast bridges on the Worker's Engine. Each callback -/// re-publishes a worker-level signal as a `protocol::Event` on `event_tx` +/// re-publishes a worker-level signal as a `protocol::Event` on `working_event_tx` /// so subscribers (TUI, socket clients) get a single typed stream. /// /// `Worker::wire_history_persistence` is called separately to wire the @@ -649,7 +649,7 @@ fn protocol_command_status(status: WorkdirCommandStatus) -> ProtocolCommandStatu /// / `AnnotatedToolResult` commit through the sync writer. pub(crate) fn wire_event_bridges_on_engine( worker: &mut Worker, - event_tx: &broadcast::Sender, + working_event_tx: &broadcast::Sender, alerter: &Alerter, in_flight: &InFlightEvents, ) where @@ -659,12 +659,12 @@ pub(crate) fn wire_event_bridges_on_engine( let ai_activity = worker.ai_activity_counter(); let worker = worker.engine_mut(); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); worker.on_turn_start(move |turn| { let _ = tx.send(Event::TurnStart { turn }); }); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); worker.on_turn_end(move |turn| { let _ = tx.send(Event::TurnEnd { turn, @@ -672,17 +672,17 @@ pub(crate) fn wire_event_bridges_on_engine( }); }); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); worker.on_llm_call_start(move |llm_call| { let _ = tx.send(Event::LlmCallStart { llm_call }); }); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); worker.on_llm_call_end(move |llm_call| { let _ = tx.send(Event::LlmCallEnd { llm_call }); }); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); worker.on_llm_retry(move |llm_call, notice| { let _ = tx.send(Event::LlmRetry { llm_call, @@ -695,7 +695,7 @@ pub(crate) fn wire_event_bridges_on_engine( }); }); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); worker.on_llm_continuation(move |llm_call, attempt, max_attempts, reason| { let _ = tx.send(Event::LlmContinuation { llm_call, @@ -768,7 +768,7 @@ pub(crate) fn wire_event_bridges_on_engine( }); }); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); let activity = ai_activity.clone(); worker.on_tool_result(move |result| { activity.fetch_add(1, Ordering::SeqCst); @@ -793,7 +793,7 @@ pub(crate) fn wire_event_bridges_on_engine( }); }); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); worker.on_usage(move |event| { let _ = tx.send(Event::Usage { input_tokens: event.input_tokens, @@ -802,7 +802,7 @@ pub(crate) fn wire_event_bridges_on_engine( }); }); - let tx = event_tx.clone(); + let tx = working_event_tx.clone(); worker.on_error(move |event| { let _ = tx.send(Event::Error { code: ErrorCode::ProviderError, @@ -1156,7 +1156,7 @@ where async fn controller_loop( mut worker: Worker, mut method_rx: mpsc::Receiver, - event_tx: broadcast::Sender, + working_event_tx: broadcast::Sender, shared_state: Arc, runtime_dir: Arc, cancel_tx: mpsc::Sender<()>, @@ -1213,7 +1213,7 @@ async fn controller_loop( set_controller_status( &shared_state, &runtime_dir, - &event_tx, + &working_event_tx, WorkerStatus::Running, ) .await; @@ -1230,7 +1230,7 @@ async fn controller_loop( }, ), &mut method_rx, - &event_tx, + &working_event_tx, &cancel_tx, &pause_tx, &shared_state, @@ -1255,7 +1255,7 @@ async fn controller_loop( }, ), &mut method_rx, - &event_tx, + &working_event_tx, &cancel_tx, &pause_tx, &shared_state, @@ -1273,7 +1273,7 @@ async fn controller_loop( drive_turn( worker.run_for_notification(kind), &mut method_rx, - &event_tx, + &working_event_tx, &cancel_tx, &pause_tx, &shared_state, @@ -1291,7 +1291,7 @@ async fn controller_loop( drive_turn( worker.resume(), &mut method_rx, - &event_tx, + &working_event_tx, &cancel_tx, &pause_tx, &shared_state, @@ -1315,16 +1315,16 @@ async fn controller_loop( &mut worker, &shared_state, &runtime_dir, - &event_tx, + &working_event_tx, new_status, ) .await; if shutdown { - let _ = event_tx.send(Event::Shutdown); + let _ = working_event_tx.send(Event::Shutdown); break; } if take_shutdown_request_after_status(&shutdown_after_idle, new_status) { - let _ = event_tx.send(Event::Shutdown); + let _ = working_event_tx.send(Event::Shutdown); break; } continue; @@ -1342,7 +1342,7 @@ async fn controller_loop( // already rejects `Run` while a turn is live, so // this branch is only reachable across a race window // around status flips. - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker is already executing a turn".into(), }); @@ -1393,7 +1393,7 @@ async fn controller_loop( Method::Resume => { if shared_state.get_status() != WorkerStatus::Paused { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::NotPaused, message: "Worker is not paused".into(), }); @@ -1409,20 +1409,20 @@ async fn controller_loop( set_controller_status( &shared_state, &runtime_dir, - &event_tx, + &working_event_tx, WorkerStatus::Idle, ) .await; } Err(error) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: worker_error_code(&error), message: error.to_string(), }); } }, WorkerStatus::Idle | WorkerStatus::Stopped => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::NotRunning, message: "Worker is not running".into(), }); @@ -1439,7 +1439,7 @@ async fn controller_loop( // Worker is Idle (Running turns go through `drive_turn`, // not this outer match), so there is nothing to pause. if shared_state.get_status() != WorkerStatus::Paused { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::NotRunning, message: "Worker is not running".into(), }); @@ -1449,21 +1449,21 @@ async fn controller_loop( Method::Compact => match shared_state.get_status() { WorkerStatus::Idle => { if let Err(error) = worker.manual_compact().await { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: worker_error_code(&error), message: error.to_string(), }); } } WorkerStatus::Paused => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::InvalidRequest, message: "Cannot compact while the Worker is paused; resume or start a fresh turn first" .into(), }); } WorkerStatus::Running | WorkerStatus::Stopped => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker is already executing a turn; compact can only run while idle" @@ -1474,10 +1474,10 @@ async fn controller_loop( Method::ListRewindTargets => match shared_state.get_status() { WorkerStatus::Idle | WorkerStatus::Paused => { - emit_rewind_targets(&worker, &event_tx) + emit_rewind_targets(&worker, &working_event_tx) } WorkerStatus::Running | WorkerStatus::Stopped => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker is already executing a turn; rewind can only run while idle or paused" .into(), @@ -1490,23 +1490,28 @@ async fn controller_loop( expected_head_entries, } => match shared_state.get_status() { WorkerStatus::Idle => { - if apply_rewind(&mut worker, &event_tx, target, expected_head_entries) { + if apply_rewind( + &mut worker, + &working_event_tx, + target, + expected_head_entries, + ) { worker.clear_in_flight_events(); shared_state.set_status(WorkerStatus::Idle); - let _ = event_tx.send(Event::Status { + let _ = working_event_tx.send(Event::Status { status: WorkerStatus::Idle, }); } } WorkerStatus::Paused => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::InvalidRequest, message: "Cannot apply rewind while the Worker is paused; resume or wait for idle first" .into(), }); } WorkerStatus::Running | WorkerStatus::Stopped => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker is already executing a turn; rewind can only run while idle or paused" .into(), @@ -1515,24 +1520,24 @@ async fn controller_loop( }, Method::Shutdown => { - let _ = event_tx.send(Event::Shutdown); + let _ = working_event_tx.send(Event::Shutdown); break; } Method::ListWorkers => match discovery.list_visible().await { Ok(workers) => match serde_json::to_value(workers) { Ok(workers) => { - let _ = event_tx.send(Event::WorkersListed { workers }); + let _ = working_event_tx.send(Event::WorkersListed { workers }); } Err(error) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::Internal, message: format!("serialize visible workers: {error}"), }); } }, Err(error) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::InvalidRequest, message: error.to_string(), }); @@ -1542,17 +1547,17 @@ async fn controller_loop( Method::RestoreWorker { name } => match discovery.restore(&name).await { Ok(result) => match serde_json::to_value(result) { Ok(result) => { - let _ = event_tx.send(Event::WorkerRestored { result }); + let _ = working_event_tx.send(Event::WorkerRestored { result }); } Err(error) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::Internal, message: format!("serialize worker restore result: {error}"), }); } }, Err(error) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::InvalidRequest, message: error.to_string(), }); @@ -1562,17 +1567,17 @@ async fn controller_loop( Method::RegisterPeer { name } => match discovery.register_peer(&name) { Ok(result) => match serde_json::to_value(result) { Ok(result) => { - let _ = event_tx.send(Event::PeerRegistered { result }); + let _ = working_event_tx.send(Event::PeerRegistered { result }); } Err(error) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::Internal, message: format!("serialize peer registration result: {error}"), }); } }, Err(error) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::InvalidRequest, message: error.to_string(), }); @@ -1691,7 +1696,7 @@ async fn handle_inbound_worker_event( async fn drive_turn( worker_future: F, method_rx: &mut mpsc::Receiver, - event_tx: &broadcast::Sender, + working_event_tx: &broadcast::Sender, cancel_tx: &mpsc::Sender<()>, pause_tx: &mpsc::Sender<()>, shared_state: &Arc, @@ -1727,7 +1732,7 @@ where set_controller_status( shared_state, runtime_dir, - event_tx, + working_event_tx, WorkerStatus::Running, ) .await; @@ -1745,11 +1750,11 @@ where WorkerRunResult::LimitReached => (WorkerStatus::Idle, RunResult::LimitReached), WorkerRunResult::RolledBack => (WorkerStatus::Idle, RunResult::RolledBack), WorkerRunResult::Interrupted { .. } if pause_requested => { - let _ = event_tx.send(Event::RunEnd { result: RunResult::Paused }); + let _ = working_event_tx.send(Event::RunEnd { result: RunResult::Paused }); return (WorkerStatus::Paused, shutdown_requested); } WorkerRunResult::Interrupted { code, message } => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code, message: message.clone(), }); @@ -1765,7 +1770,7 @@ where return (WorkerStatus::Idle, shutdown_requested); } }; - let _ = event_tx.send(Event::RunEnd { result: run_result }); + let _ = working_event_tx.send(Event::RunEnd { result: run_result }); if parent_originated && matches!(run_result, RunResult::Finished) { crate::ipc::event::fire_and_forget( parent_socket.cloned(), @@ -1782,13 +1787,13 @@ where // intentionally skip `WorkerEvent::Errored` upward: // that channel is reserved for worker runtime // failures, not deliberate interruptions. - let _ = event_tx.send(Event::RunEnd { result: RunResult::Paused }); + let _ = working_event_tx.send(Event::RunEnd { result: RunResult::Paused }); (WorkerStatus::Paused, shutdown_requested) } Err(e) => { let code = worker_error_code(&e); let message = e.to_string(); - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code, message: message.clone(), }); @@ -1819,13 +1824,13 @@ where let _ = cancel_tx.try_send(()); } Some(Method::Run { .. } | Method::RunTracked { .. } | Method::Resume) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker is already executing a turn".into(), }); } Some(Method::Compact | Method::ListRewindTargets | Method::RewindTo { .. }) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker is already executing a turn; rewind/compact can only run while idle or paused" .into(), @@ -1839,7 +1844,7 @@ where } Some(Method::ListCompletions { .. }) => {} Some(Method::ListWorkers | Method::RestoreWorker { .. } | Method::RegisterPeer { .. }) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::AlreadyRunning, message: "Worker discovery/control requests are only handled while the Worker is idle or paused" .into(), @@ -1872,20 +1877,20 @@ where } } -fn emit_rewind_targets(worker: &Worker, event_tx: &broadcast::Sender) +fn emit_rewind_targets(worker: &Worker, working_event_tx: &broadcast::Sender) where C: LlmClient + 'static, St: Store, { match worker.list_rewind_targets() { Ok((head_entries, targets)) => { - let _ = event_tx.send(Event::RewindTargets { + let _ = working_event_tx.send(Event::RewindTargets { head_entries, targets, }); } Err(err) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::Internal, message: err.to_string(), }); @@ -1895,7 +1900,7 @@ where fn apply_rewind( worker: &mut Worker, - event_tx: &broadcast::Sender, + working_event_tx: &broadcast::Sender, target: RewindTargetId, expected_head_entries: usize, ) -> bool @@ -1907,7 +1912,7 @@ where Ok(applied) => { let session = session_store::public_snapshot::project_current_session_snapshot(&applied.entries); - let _ = event_tx.send(Event::RewindApplied { + let _ = working_event_tx.send(Event::RewindApplied { session, input: applied.input, summary: applied.summary, @@ -1915,7 +1920,7 @@ where true } Err(err) => { - let _ = event_tx.send(Event::Error { + let _ = working_event_tx.send(Event::Error { code: ErrorCode::InvalidRequest, message: err.to_string(), }); @@ -2049,7 +2054,7 @@ mod tests { // would observe channel-closed and confuse the select! arm. _method_tx: mpsc::Sender, method_rx: mpsc::Receiver, - event_tx: broadcast::Sender, + working_event_tx: broadcast::Sender, cancel_tx: mpsc::Sender<()>, _cancel_rx: mpsc::Receiver<()>, pause_tx: mpsc::Sender<()>, @@ -2070,7 +2075,7 @@ mod tests { .expect("runtime dir create"), ); let (method_tx, method_rx) = mpsc::channel::(16); - let (event_tx, _) = broadcast::channel::(16); + let (working_event_tx, _) = broadcast::channel::(16); let (cancel_tx, cancel_rx) = mpsc::channel::<()>(1); let (pause_tx, pause_rx) = mpsc::channel::<()>(1); let shared_state = Arc::new(WorkerSharedState::new( @@ -2095,7 +2100,7 @@ mod tests { DriveTurnEnv { _method_tx: method_tx, method_rx, - event_tx, + working_event_tx, cancel_tx, _cancel_rx: cancel_rx, pause_tx, @@ -2157,7 +2162,7 @@ mod tests { let (status, shutdown) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, @@ -2200,7 +2205,7 @@ mod tests { let (status, shutdown) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, @@ -2230,7 +2235,7 @@ mod tests { let (status, _) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, @@ -2268,7 +2273,7 @@ mod tests { let (status, _) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, @@ -2312,7 +2317,7 @@ mod tests { let (status, _) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, @@ -2354,7 +2359,7 @@ mod tests { let (status, shutdown) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, @@ -2393,7 +2398,7 @@ mod tests { let (status, shutdown) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, @@ -2430,7 +2435,7 @@ mod tests { let (status, shutdown) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, @@ -2453,7 +2458,7 @@ mod tests { #[tokio::test] async fn compact_method_is_rejected_while_running() { let mut env = make_env().await; - let mut events = env.event_tx.subscribe(); + let mut events = env.working_event_tx.subscribe(); env._method_tx .send(Method::Compact) .await @@ -2466,7 +2471,7 @@ mod tests { let (status, shutdown) = drive_turn( worker_future, &mut env.method_rx, - &env.event_tx, + &env.working_event_tx, &env.cancel_tx, &env.pause_tx, &env.shared_state, diff --git a/crates/worker/src/in_flight.rs b/crates/worker/src/in_flight.rs index 095fffb8..39e1d844 100644 --- a/crates/worker/src/in_flight.rs +++ b/crates/worker/src/in_flight.rs @@ -15,7 +15,7 @@ pub struct InFlightBlockId(u64); #[derive(Debug, Clone)] pub struct InFlightEvents { inner: Arc>, - event_tx: broadcast::Sender, + working_event_tx: broadcast::Sender, } #[derive(Debug)] @@ -47,14 +47,14 @@ enum TrackedBlock { } impl InFlightEvents { - pub(crate) fn new(event_tx: broadcast::Sender) -> Self { + pub(crate) fn new(working_event_tx: broadcast::Sender) -> Self { Self { inner: Arc::new(Mutex::new(InFlightInner { next_block_id: 1, blocks: Vec::new(), commands: Vec::new(), })), - event_tx, + working_event_tx, } } @@ -84,7 +84,7 @@ impl InFlightEvents { current.push_str(&text); *finished = false; } - let _ = self.event_tx.send(Event::TextDelta { text }); + let _ = self.working_event_tx.send(Event::TextDelta { text }); } pub(crate) fn text_done(&self, block_id: InFlightBlockId, text: String) { @@ -100,7 +100,7 @@ impl InFlightEvents { } *finished = true; } - let _ = self.event_tx.send(Event::TextDone { text }); + let _ = self.working_event_tx.send(Event::TextDone { text }); } pub(crate) fn thinking_start(&self) -> InFlightBlockId { @@ -111,7 +111,7 @@ impl InFlightEvents { text: String::new(), finished: false, }); - let _ = self.event_tx.send(Event::ThinkingStart); + let _ = self.working_event_tx.send(Event::ThinkingStart); block_id } @@ -126,7 +126,7 @@ impl InFlightEvents { current.push_str(&text); *finished = false; } - let _ = self.event_tx.send(Event::ThinkingDelta { text }); + let _ = self.working_event_tx.send(Event::ThinkingDelta { text }); } pub(crate) fn thinking_done(&self, block_id: InFlightBlockId, text: String) { @@ -142,7 +142,7 @@ impl InFlightEvents { } *finished = true; } - let _ = self.event_tx.send(Event::ThinkingDone { text }); + let _ = self.working_event_tx.send(Event::ThinkingDone { text }); } pub(crate) fn tool_call_start(&self, id: String, name: String) -> InFlightBlockId { @@ -155,7 +155,9 @@ impl InFlightEvents { args: String::new(), state: InFlightToolCallState::Pending, }); - let _ = self.event_tx.send(Event::ToolCallStart { id, name }); + let _ = self + .working_event_tx + .send(Event::ToolCallStart { id, name }); block_id } @@ -171,7 +173,7 @@ impl InFlightEvents { *state = InFlightToolCallState::StreamingArgs; } let _ = self - .event_tx + .working_event_tx .send(Event::ToolCallArgsDelta { id, json: delta }); } @@ -191,7 +193,7 @@ impl InFlightEvents { } *state = InFlightToolCallState::Done; } - let _ = self.event_tx.send(Event::ToolCallDone { + let _ = self.working_event_tx.send(Event::ToolCallDone { id, name, arguments: args, @@ -210,7 +212,7 @@ impl InFlightEvents { pub(crate) fn publish_command_event(&self, event: CommandEvent) { self.lock().apply_command_event(&event); - let _ = self.event_tx.send(Event::Command { event }); + let _ = self.working_event_tx.send(Event::Command { event }); } pub(crate) fn replace_command_snapshot(&self, commands: Vec) { @@ -492,13 +494,13 @@ mod tests { #[test] fn snapshot_boundary_does_not_duplicate_or_gap_delta_sent_after_subscribe() { - let (event_tx, _) = broadcast::channel(16); - let in_flight = InFlightEvents::new(event_tx.clone()); + let (working_event_tx, _) = broadcast::channel(16); + let in_flight = InFlightEvents::new(working_event_tx.clone()); let block_id = in_flight.start_text_block(); in_flight.text_delta(block_id, "hel".into()); let guard = in_flight.snapshot_guard(); - let mut rx = event_tx.subscribe(); + let mut rx = working_event_tx.subscribe(); let snapshot = snapshot_from_guard(&guard); drop(guard); @@ -526,9 +528,9 @@ mod tests { use crate::segment_log_sink::SegmentLogSink; use session_store::{LogEntry, LoggedRole}; - let (event_tx, _) = broadcast::channel(16); + let (working_event_tx, _) = broadcast::channel(16); let sink = SegmentLogSink::new(); - let in_flight = InFlightEvents::new(event_tx); + let in_flight = InFlightEvents::new(working_event_tx); let block_id = in_flight.start_text_block(); in_flight.text_delta(block_id, "done".into()); in_flight.text_done(block_id, "done".into()); @@ -580,9 +582,9 @@ mod tests { use crate::segment_log_sink::SegmentLogSink; use session_store::{LogEntry, LoggedRole}; - let (event_tx, _) = broadcast::channel(16); + let (working_event_tx, _) = broadcast::channel(16); let sink = SegmentLogSink::new(); - let in_flight = InFlightEvents::new(event_tx); + let in_flight = InFlightEvents::new(working_event_tx); let block_id = in_flight.start_text_block(); in_flight.text_delta(block_id, "done".into()); in_flight.text_done(block_id, "done".into()); @@ -615,8 +617,8 @@ mod tests { #[test] fn committed_item_clears_matching_in_flight_block() { - let (event_tx, _) = broadcast::channel(16); - let in_flight = InFlightEvents::new(event_tx); + let (working_event_tx, _) = broadcast::channel(16); + let in_flight = InFlightEvents::new(working_event_tx); let block_id = in_flight.start_text_block(); in_flight.text_delta(block_id, "done".into()); in_flight.clear_for_committed_item_then( @@ -635,8 +637,8 @@ mod tests { #[test] fn committed_reasoning_summary_clears_matching_in_flight_thinking_blocks() { - let (event_tx, _) = broadcast::channel(16); - let in_flight = InFlightEvents::new(event_tx); + let (working_event_tx, _) = broadcast::channel(16); + let in_flight = InFlightEvents::new(working_event_tx); let first = in_flight.thinking_start(); in_flight.thinking_delta(first, "summary A".into()); in_flight.thinking_done(first, "".into()); @@ -660,8 +662,8 @@ mod tests { #[test] fn committed_encrypted_only_reasoning_clears_empty_finished_thinking_block() { - let (event_tx, _) = broadcast::channel(16); - let in_flight = InFlightEvents::new(event_tx); + let (working_event_tx, _) = broadcast::channel(16); + let in_flight = InFlightEvents::new(working_event_tx); let first = in_flight.thinking_start(); in_flight.thinking_done(first, "".into()); let second = in_flight.thinking_start(); @@ -689,9 +691,9 @@ mod tests { #[test] fn command_events_are_bounded_and_recoverable_from_snapshot() { - let (event_tx, _) = broadcast::channel(16); - let mut rx = event_tx.subscribe(); - let in_flight = InFlightEvents::new(event_tx); + let (working_event_tx, _) = broadcast::channel(16); + let mut rx = working_event_tx.subscribe(); + let in_flight = InFlightEvents::new(working_event_tx); in_flight.publish_command_event(CommandEvent::Started { command_id: "command-1".into(), tool_call_id: Some("tool-1".into()), @@ -740,9 +742,9 @@ mod tests { #[test] fn clear_discards_uncommitted_blocks_without_protocol_event() { - let (event_tx, _) = broadcast::channel(16); - let mut rx = event_tx.subscribe(); - let in_flight = InFlightEvents::new(event_tx); + let (working_event_tx, _) = broadcast::channel(16); + let mut rx = working_event_tx.subscribe(); + let in_flight = InFlightEvents::new(working_event_tx); let text = in_flight.start_text_block(); in_flight.text_delta(text, "stale".into()); let tool = in_flight.tool_call_start("call-1".into(), "Bash".into()); @@ -770,8 +772,8 @@ mod tests { #[test] fn snapshot_omits_empty_finished_thinking_blocks() { - let (event_tx, _) = broadcast::channel(16); - let in_flight = InFlightEvents::new(event_tx); + let (working_event_tx, _) = broadcast::channel(16); + let in_flight = InFlightEvents::new(working_event_tx); let empty_finished = in_flight.thinking_start(); in_flight.thinking_done(empty_finished, "".into()); let empty_running = in_flight.thinking_start(); diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index e033369e..6dc23c51 100644 --- a/crates/worker/src/internal_worker.rs +++ b/crates/worker/src/internal_worker.rs @@ -752,7 +752,7 @@ pub(crate) async fn prepare_internal_worker_session( } let actor_in_flight = in_flight.clone(); worker.attach_alerter(alerter.clone()); - worker.attach_event_tx(event_tx.clone()); + worker.attach_working_event_tx(event_tx.clone()); worker.attach_in_flight_events(in_flight.clone()); wire_event_bridges_on_engine(&mut worker, &event_tx, &alerter, &in_flight); diff --git a/crates/worker/src/ipc/alerter.rs b/crates/worker/src/ipc/alerter.rs index 4ae3d0e0..1ab5f55b 100644 --- a/crates/worker/src/ipc/alerter.rs +++ b/crates/worker/src/ipc/alerter.rs @@ -28,15 +28,15 @@ pub struct Alerter { } struct Inner { - event_tx: broadcast::Sender, + working_event_tx: broadcast::Sender, buffer: Mutex>, } impl Alerter { - pub fn new(event_tx: broadcast::Sender) -> Self { + pub fn new(working_event_tx: broadcast::Sender) -> Self { Self { inner: Arc::new(Inner { - event_tx, + working_event_tx, buffer: Mutex::new(VecDeque::with_capacity(MAX_BUFFERED_ALERTS)), }), } @@ -66,7 +66,7 @@ impl Alerter { buf.pop_front(); } buf.push_back(alert.clone()); - let _ = self.inner.event_tx.send(Event::Alert(alert)); + let _ = self.inner.working_event_tx.send(Event::Alert(alert)); } } @@ -81,7 +81,7 @@ impl Alerter { .buffer .lock() .expect("alerter buffer mutex poisoned"); - let rx = self.inner.event_tx.subscribe(); + let rx = self.inner.working_event_tx.subscribe(); let snapshot: Vec = buf.iter().cloned().collect(); (snapshot, rx) } diff --git a/crates/worker/src/segment_log_sink.rs b/crates/worker/src/segment_log_sink.rs index 38192af7..91c2792b 100644 --- a/crates/worker/src/segment_log_sink.rs +++ b/crates/worker/src/segment_log_sink.rs @@ -51,7 +51,7 @@ struct SinkInner { /// survives session swaps so existing subscribers keep their /// receiver — they observe the swap as a freshly broadcast /// `LogEntry::AnnotatedSegmentStart` and reset their view accordingly. - broadcast_tx: broadcast::Sender, + session_entry_tx: broadcast::Sender, } impl SegmentLogSink { @@ -59,11 +59,11 @@ impl SegmentLogSink { /// has been written (deferred SegmentStart) or as a placeholder in /// tests. pub fn new() -> Self { - let (broadcast_tx, _) = broadcast::channel(BROADCAST_CAPACITY); + let (session_entry_tx, _) = broadcast::channel(BROADCAST_CAPACITY); Self { inner: Arc::new(SinkInner { mirror: StdMutex::new(Vec::new()), - broadcast_tx, + session_entry_tx, }), } } @@ -72,11 +72,11 @@ impl SegmentLogSink { /// Used by restore / fork-at-restore code paths that materialise /// the existing log before the sink starts taking new commits. pub fn with_initial(entries: Vec) -> Self { - let (broadcast_tx, _) = broadcast::channel(BROADCAST_CAPACITY); + let (session_entry_tx, _) = broadcast::channel(BROADCAST_CAPACITY); Self { inner: Arc::new(SinkInner { mirror: StdMutex::new(entries), - broadcast_tx, + session_entry_tx, }), } } @@ -111,7 +111,7 @@ impl SegmentLogSink { // SendError means there are zero subscribers; harmless. The // mirror lock is held across `send` so subscribers cannot // observe an inconsistent (snapshot, receiver) pair. - let _ = self.inner.broadcast_tx.send(entry); + let _ = self.inner.session_entry_tx.send(entry); } } @@ -144,7 +144,7 @@ impl SegmentLogSink { .expect("session log mirror mutex poisoned"); mirror.clear(); mirror.push(initial.clone()); - let _ = self.inner.broadcast_tx.send(initial); + let _ = self.inner.session_entry_tx.send(initial); } /// Atomically swap the mirror to the supplied replacement-session prefix @@ -161,7 +161,7 @@ impl SegmentLogSink { .expect("session log mirror mutex poisoned"); *mirror = entries; if let Some(initial) = first { - let _ = self.inner.broadcast_tx.send(initial); + let _ = self.inner.session_entry_tx.send(initial); } } @@ -199,7 +199,7 @@ impl SegmentLogSink { .lock() .expect("session log mirror mutex poisoned"); let snapshot = mirror.clone(); - let rx = self.inner.broadcast_tx.subscribe(); + let rx = self.inner.session_entry_tx.subscribe(); (snapshot, rx) } diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index 2c769f6e..62fec8d4 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -414,10 +414,10 @@ impl SpawnedWorkerRegistry { pub(crate) fn attach_parent_protocol( &self, - event_tx: broadcast::Sender, + working_event_tx: broadcast::Sender, parent_session_id: String, ) { - *self.parent_protocol.lock().unwrap() = Some((event_tx, parent_session_id)); + *self.parent_protocol.lock().unwrap() = Some((working_event_tx, parent_session_id)); for record in self.internal_records.lock().unwrap().clone() { self.start_protocol_forwarding(record); } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 01788a13..cf2e579e 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -1149,7 +1149,7 @@ pub struct Worker { /// etc.). Attached by the Controller alongside `alerter`. Unlike /// notifications, events sent here are NOT replayed to clients that /// connect after the fact — they are fire-and-forget broadcasts. - event_tx: Option>, + working_event_tx: Option>, /// Parent-owned projection/control boundary for observable Internal service Workers. /// Service Workers are never exposed through the model-facing SubWorker control surface. internal_worker_registry: Option>, @@ -1304,7 +1304,7 @@ impl Worker system_prompt_template: None, feature_instructions: self.feature_instructions.clone(), alerter: self.alerter.clone(), - event_tx: self.event_tx.clone(), + working_event_tx: self.working_event_tx.clone(), internal_worker_registry: self.internal_worker_registry.clone(), in_flight: self.in_flight.clone(), ai_activity_counter: self.ai_activity_counter.clone(), @@ -1497,7 +1497,7 @@ impl Worker { system_prompt_template: None, feature_instructions: Vec::new(), alerter: None, - event_tx: None, + working_event_tx: None, internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), @@ -2190,13 +2190,13 @@ impl Worker { /// The Controller wires this alongside [`attach_alerter`] so that /// Worker-internal operations (currently: compaction) can surface /// progress to connected clients. - pub fn attach_event_tx(&mut self, event_tx: broadcast::Sender) { + pub fn attach_working_event_tx(&mut self, working_event_tx: broadcast::Sender) { let session_id = self.session_id().to_string(); let registry = self.internal_worker_registry.get_or_insert_with( crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services, ); - registry.attach_parent_protocol(event_tx.clone(), session_id); - self.event_tx = Some(event_tx); + registry.attach_parent_protocol(working_event_tx.clone(), session_id); + self.working_event_tx = Some(working_event_tx); } pub(crate) fn attach_internal_worker_registry( @@ -2240,10 +2240,10 @@ impl Worker { } /// Broadcast a typed `Event` to connected clients. No-op when no - /// `event_tx` is attached (tests / direct `Worker::new` usage) or when + /// `working_event_tx` is attached (tests / direct `Worker::new` usage) or when /// no clients are currently subscribed. fn send_event(&self, event: Event) { - if let Some(tx) = self.event_tx.as_ref() { + if let Some(tx) = self.working_event_tx.as_ref() { let _ = tx.send(event); } } @@ -4407,7 +4407,7 @@ impl Worker { .with_memory_settings(&memory_cfg) .emit( self.workspace_client(), - self.event_tx.as_ref(), + self.working_event_tx.as_ref(), memory::audit::WorkerLifecycleStatus::Skipped, "extract_threshold_disabled", None, @@ -4438,7 +4438,7 @@ impl Worker { .with_memory_settings(&memory_cfg) .emit( self.workspace_client(), - self.event_tx.as_ref(), + self.working_event_tx.as_ref(), memory::audit::WorkerLifecycleStatus::Skipped, "extract_already_in_flight", None, @@ -4503,7 +4503,7 @@ impl Worker { Some(model_audit_from_manifest(model)), ) .with_memory_settings(memory_cfg); - let event_tx = self.event_tx.as_ref(); + let working_event_tx = self.working_event_tx.as_ref(); let pointer_snapshot = self .extract_pointer @@ -4519,7 +4519,7 @@ impl Worker { if tokens_since < threshold { audit.emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Skipped, format!( "token_threshold_not_reached tokens_since={tokens_since} threshold={threshold}" @@ -4536,7 +4536,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Skipped, "no_new_history_items", None, @@ -4564,7 +4564,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Skipped, "empty_segment_log", None, @@ -4583,7 +4583,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Skipped, "no_new_segment_entries", None, @@ -4613,7 +4613,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Started, format!( "token_threshold_reached tokens_since={tokens_since} threshold={threshold}" @@ -4637,7 +4637,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Failed, format!("client_build_failed: {err}"), None, @@ -4659,7 +4659,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Failed, format!("prompt_render_failed: {err}"), None, @@ -4737,7 +4737,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Cancelled, "worker_cancelled: internal Worker run rolled back before AI output", usage, @@ -4760,7 +4760,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, lifecycle_status_for_worker_error(&err.source), format!("worker_failed: {}", err.source), usage, @@ -4812,7 +4812,7 @@ impl Worker { audit .emit( self.workspace_client(), - event_tx, + working_event_tx, memory::audit::WorkerLifecycleStatus::Completed, reason, usage, @@ -4847,7 +4847,7 @@ impl Worker { .with_memory_settings(&memory_cfg) .emit( self.workspace_client(), - self.event_tx.as_ref(), + self.working_event_tx.as_ref(), memory::audit::WorkerLifecycleStatus::Skipped, "consolidation_threshold_disabled", None, @@ -4889,7 +4889,7 @@ impl Worker { .with_memory_settings(&memory_cfg) .emit( self.workspace_client(), - self.event_tx.as_ref(), + self.working_event_tx.as_ref(), memory::audit::WorkerLifecycleStatus::Skipped, "consolidation_backend_operation_failed", None, @@ -4942,18 +4942,18 @@ fn model_audit_from_manifest(model: &manifest::ModelManifest) -> memory::audit:: } fn emit_memory_worker_event( - event_tx: Option<&broadcast::Sender>, + working_event_tx: Option<&broadcast::Sender>, run_id: uuid::Uuid, worker: memory::audit::AuditWorker, status: memory::audit::WorkerLifecycleStatus, trigger: memory::audit::AuditTrigger, reason: &str, ) { - let Some(event_tx) = event_tx else { + let Some(working_event_tx) = working_event_tx else { return; }; let message = format!("memory {} {}: {reason}", worker.label(), status.label()); - let _ = event_tx.send(Event::MemoryWorker(protocol::MemoryWorkerEvent { + let _ = working_event_tx.send(Event::MemoryWorker(protocol::MemoryWorkerEvent { worker: worker.label().to_string(), status: status.label().to_string(), run_id: run_id.to_string(), @@ -5003,7 +5003,7 @@ impl WorkerAuditBase { async fn emit( &self, workspace_client: &dyn WorkspaceClient, - event_tx: Option<&broadcast::Sender>, + working_event_tx: Option<&broadcast::Sender>, status: memory::audit::WorkerLifecycleStatus, reason: impl Into, usage: Option, @@ -5034,7 +5034,7 @@ impl WorkerAuditBase { .await; if should_emit_memory_worker_event(self.worker, status, &reason) { emit_memory_worker_event( - event_tx, + working_event_tx, self.run_id, self.worker, status, @@ -5218,7 +5218,7 @@ where system_prompt_template: common.system_prompt_template, feature_instructions: common.feature_instructions, alerter: None, - event_tx: None, + working_event_tx: None, internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), @@ -5302,7 +5302,7 @@ where system_prompt_template: common.system_prompt_template, feature_instructions: common.feature_instructions, alerter: None, - event_tx: None, + working_event_tx: None, internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), @@ -5421,7 +5421,7 @@ where system_prompt_template: common.system_prompt_template, feature_instructions: common.feature_instructions, alerter: None, - event_tx: None, + working_event_tx: None, internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), @@ -5797,7 +5797,7 @@ where system_prompt_template: None, feature_instructions: common.feature_instructions, alerter: None, - event_tx: None, + working_event_tx: None, internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index e67519fe..708fd60a 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -382,7 +382,7 @@ async fn compact_emits_session_start_carrying_summary_and_task_snapshot() { let mut worker = make_worker(client).await; let (tx, _rx_keep) = broadcast::channel::(64); - worker.attach_event_tx(tx); + worker.attach_working_event_tx(tx); worker.run_text("first").await.unwrap(); let session_id = worker.session_id(); @@ -429,7 +429,7 @@ async fn pre_run_compact_success_broadcasts_start_and_done() { let mut worker = make_worker(client).await; let (tx, mut rx) = broadcast::channel::(64); - worker.attach_event_tx(tx); + worker.attach_working_event_tx(tx); worker.run_text("first").await.unwrap(); // Drain run events so only compact events remain in `rx`. @@ -539,7 +539,7 @@ async fn mid_turn_compact_success_broadcasts_start_and_done() { let mut worker = make_worker_with_manifest(MID_TURN_MANIFEST_TOML, client).await; let (tx, mut rx) = broadcast::channel::(64); - worker.attach_event_tx(tx); + worker.attach_working_event_tx(tx); // First run populates usage_history above the request threshold. worker.run_text("first").await.unwrap(); @@ -718,7 +718,7 @@ async fn pre_run_compact_failure_broadcasts_start_and_failed() { let mut worker = make_worker(client).await; let (tx, mut rx) = broadcast::channel::(64); - worker.attach_event_tx(tx); + worker.attach_working_event_tx(tx); worker.run_text("first").await.unwrap(); let _ = drain(&mut rx); diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 364440b2..968f8292 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -1515,7 +1515,7 @@ async fn notify_while_idle_auto_starts_turn_and_injects_system_message() { // after that. wait_for_status(&handle, WorkerStatus::Idle).await; // The live echo arrives via the sink's `Event::SystemItem` lane, - // not on the `event_tx` broadcast that `handle.subscribe()` taps. + // not on the `working_event_tx` broadcast that `handle.subscribe()` taps. // Verify the notification landed on the sink mirror instead. let (entries, _) = handle.sink.subscribe_with_snapshot(); let saw_notify_in_mirror = entries.iter().any(|e| { @@ -1899,7 +1899,7 @@ async fn socket_worker_event_turn_ended_while_idle_auto_starts_turn() { let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2); // The SystemItem and TurnEnd events arrive through independent - // broadcast lanes (sink fan-out vs `event_tx`), so their relative + // broadcast lanes (sink fan-out vs `working_event_tx`), so their relative // order on the wire is non-deterministic. Keep reading until both // are observed (or the deadline trips), rather than breaking on // the first TurnEnd.