Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

refactor: name worker event channels by role

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