From cf394403a69b4763c2684448fd572277db6ddd31 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 7 Aug 2026 02:03:09 +0900 Subject: [PATCH] worker: retire process sub-worker registry authority --- crates/worker/src/controller.rs | 18 +- crates/worker/src/ipc/event.rs | 90 +---- crates/worker/src/spawn/registry.rs | 468 ++++++---------------- crates/worker/tests/worker_events_test.rs | 426 ++------------------ 4 files changed, 152 insertions(+), 850 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index c6be98e8..f8af7119 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1577,7 +1577,6 @@ fn worker_error_code(e: &WorkerError) -> ErrorCode { #[cfg(test)] mod tests { use super::*; - use crate::runtime::dir::SpawnedWorkerRecord; use protocol::WorkerEvent; use protocol::stream::{JsonLineReader, JsonLineWriter}; use std::time::Duration; @@ -1834,17 +1833,8 @@ mod tests { } #[tokio::test] - async fn running_scope_sub_delegated_applies_side_effects_without_notify_buffer() { + async fn running_legacy_scope_callback_has_no_registry_authority_or_notify() { let mut env = make_env().await; - env.spawned_registry - .add(SpawnedWorkerRecord { - worker_name: "child".into(), - socket_path: "/tmp/child.sock".into(), - scope_delegated: vec![], - callback_address: "/tmp/parent.sock".into(), - }) - .await - .expect("seed child record"); env._method_tx .send(Method::WorkerEvent(WorkerEvent::ScopeSubDelegated { parent_worker: "child".into(), @@ -1875,13 +1865,9 @@ mod tests { assert_eq!(status, WorkerStatus::Idle); assert!(!shutdown); - assert!( - env.spawned_registry.get("grandchild").await.is_some(), - "ScopeSubDelegated side effects must still register the grandchild" - ); assert!( env.notify_buffer.is_empty(), - "control-plane-only ScopeSubDelegated must not enter the agent-visible notify buffer" + "legacy ScopeSubDelegated must not enter the agent-visible notify buffer" ); } diff --git a/crates/worker/src/ipc/event.rs b/crates/worker/src/ipc/event.rs index d4155719..2e4bba72 100644 --- a/crates/worker/src/ipc/event.rs +++ b/crates/worker/src/ipc/event.rs @@ -26,9 +26,8 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; -use protocol::{Method, ScopeRule, WorkerEvent}; +use protocol::{Method, WorkerEvent}; -use crate::runtime::dir::SpawnedWorkerRecord; use crate::spawn::comm_tools::connect_and_send; use crate::spawn::registry::SpawnedWorkerRegistry; @@ -85,86 +84,15 @@ pub fn render_event(event: &WorkerEvent) -> String { } } -/// Apply the variant-specific side effect on the parent side. +/// Legacy process callback events have no SubWorker registry authority. /// -/// All operations are idempotent so that out-of-order delivery (e.g. -/// `TurnEnded` arriving after `ShutDown`) does not produce errors: -/// -/// - `TurnEnded` / `Errored`: no system work; the LLM handles the -/// semantic response. -/// - `ShutDown`: remove the child from `spawned_workers.json`, Worker state, -/// and reclaim its delegated scope/allocation. Missing entries are swallowed. -/// - `ScopeSubDelegated`: register the grandchild locally and re-emit -/// upward to our own parent if we have one. Duplicate grandchild -/// entries (re-delivery) are swallowed. +/// Internal SubWorker lifecycle is applied directly through typed session handles. A callback from +/// an externally adopted Worker may still be rendered for diagnostics, but it cannot add/remove +/// Internal children or transfer filesystem authority. pub async fn apply_event_side_effects( - event: &WorkerEvent, - registry: &Arc, - self_name: &str, - self_parent_socket: &Option, + _event: &WorkerEvent, + _registry: &Arc, + _self_name: &str, + _self_parent_socket: &Option, ) { - match event { - WorkerEvent::TurnEnded { .. } | WorkerEvent::Errored { .. } => {} - - WorkerEvent::ShutDown { worker_name } => { - if let Err(e) = registry.remove(worker_name).await { - tracing::warn!(error = %e, worker = %worker_name, "registry remove on ShutDown failed"); - } - } - - WorkerEvent::ScopeSubDelegated { - parent_worker, - sub_worker, - sub_socket, - scope, - } => { - if registry.get(sub_worker).await.is_some() { - return; - } - let callback_address = registry - .get(parent_worker) - .await - .map(|r| r.socket_path) - .unwrap_or_else(PathBuf::new); - let record = SpawnedWorkerRecord { - worker_name: sub_worker.clone(), - socket_path: sub_socket.clone(), - scope_delegated: scope.clone(), - callback_address, - }; - if let Err(e) = registry.add(record).await { - tracing::warn!( - error = %e, - sub_worker = %sub_worker, - "registry add on ScopeSubDelegated failed" - ); - } - reemit_scope_sub_delegated( - self_parent_socket, - self_name, - sub_worker.clone(), - sub_socket.clone(), - scope.clone(), - ); - } - } -} - -fn reemit_scope_sub_delegated( - self_parent_socket: &Option, - self_name: &str, - sub_worker: String, - sub_socket: PathBuf, - scope: Vec, -) { - let Some(parent_socket) = self_parent_socket.clone() else { - return; - }; - let event = WorkerEvent::ScopeSubDelegated { - parent_worker: self_name.to_string(), - sub_worker, - sub_socket, - scope, - }; - fire_and_forget(Some(parent_socket), event); } diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index 23071156..77eb1147 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -12,12 +12,10 @@ use std::collections::HashMap; use std::io; use std::sync::Arc; -use std::time::Duration; use manifest::{Permission, ScopeRule, SharedScope}; use session_store::{ - WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerSpawnedScopeRule, - WorkerStoreError, + WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerStoreError, }; use tokio::sync::Mutex; use tracing::warn; @@ -26,11 +24,6 @@ use crate::internal_worker::InternalWorkerSessionHandle; use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord}; use crate::runtime::worker_allocation; -type RegistryStateWriter = Arc io::Result<()> + Send + Sync>; -type RegistryReclaimWriter = Arc io::Result<()> + Send + Sync>; - -const REGISTRY_CLEANUP_TIMEOUT: Duration = Duration::from_secs(15); - #[derive(Clone)] pub(crate) struct InternalSpawnedWorkerRecord { pub worker_name: String, @@ -39,55 +32,35 @@ pub(crate) struct InternalSpawnedWorkerRecord { } pub struct SpawnedWorkerRegistry { - records: Mutex>, internal_records: std::sync::Mutex>, cursors: Mutex>, - mutations: Mutex<()>, - runtime_dir: Option>, - state_writer: Option, - reclaim_writer: Option, - parent_name: Option, parent_scope: Option, } pub struct SpawnedWorkerRegistryLoad { pub registry: Arc, + /// True when obsolete process-child metadata was consumed and cleared. pub reclaimed_unreachable: bool, } impl SpawnedWorkerRegistry { - pub fn new(runtime_dir: Arc) -> Arc { + /// Empty registry used by tests and non-spawning projections. + pub fn new(_runtime_dir: Arc) -> Arc { Arc::new(Self { - records: Mutex::new(Vec::new()), internal_records: std::sync::Mutex::new(Vec::new()), cursors: Mutex::new(HashMap::new()), - mutations: Mutex::new(()), - runtime_dir: Some(runtime_dir), - state_writer: None, - reclaim_writer: None, - parent_name: None, parent_scope: None, }) } - pub(crate) fn new_internal(parent_name: String, parent_scope: SharedScope) -> Arc { + pub(crate) fn new_internal(_parent_name: String, parent_scope: SharedScope) -> Arc { Arc::new(Self { - records: Mutex::new(Vec::new()), internal_records: std::sync::Mutex::new(Vec::new()), cursors: Mutex::new(HashMap::new()), - mutations: Mutex::new(()), - runtime_dir: None, - state_writer: None, - reclaim_writer: None, - parent_name: Some(parent_name), parent_scope: Some(parent_scope), }) } - /// Build a registry from the spawner's durable Worker state, pruning child - /// records whose socket path is already gone. The surviving list is - /// written through to both `spawned_workers.json` and Worker state so runtime - /// and durable views start aligned. pub async fn load_from_worker_state( runtime_dir: Arc, store: St, @@ -96,12 +69,14 @@ impl SpawnedWorkerRegistry { where St: WorkerMetadataStore + Clone + Send + Sync + 'static, { - let loaded = + Ok( Self::load_from_worker_state_with_reclaim(runtime_dir, store, worker_name, None) - .await?; - Ok(loaded.registry) + .await? + .registry, + ) } + /// Clear obsolete process-child state instead of attempting socket reconnection. pub async fn load_from_worker_state_with_reclaim( runtime_dir: Arc, store: St, @@ -116,81 +91,49 @@ impl SpawnedWorkerRegistry { .map_err(store_error_to_io)?; let persisted_children = metadata .as_ref() - .map(|m| m.spawned_children.clone()) + .map(|metadata| metadata.spawned_children.clone()) .unwrap_or_default(); - - let records = Vec::with_capacity(persisted_children.len()); - let mut pruned_records = Vec::new(); + let mut valid_records = Vec::new(); for child in &persisted_children { - let record = match record_from_worker_state(child) { - Ok(record) => record, - Err(err) => { + match record_from_worker_state(child) { + Ok(record) => { warn!( - error = %err, - worker = %child.worker_name, - "dropping corrupt persisted spawned-worker record" + worker = %record.worker_name, + "reclaiming legacy persisted process SubWorker during Internal session restore" ); - continue; + valid_records.push(record); } - }; - warn!( - worker = %record.worker_name, - "reclaiming legacy persisted process Sub-worker during Internal session restore" - ); - pruned_records.push(record); + Err(error) => warn!( + error = %error, + worker = %child.worker_name, + "clearing corrupt legacy persisted process SubWorker record" + ), + } } - runtime_dir.write_spawned_workers(&records).await?; - let state_writer = worker_state_writer(store.clone(), worker_name.clone()); - let reclaim_writer = worker_state_reclaim_writer(store.clone(), worker_name.clone()); - if metadata.is_none() { - state_writer(&records)?; - } - - let mut reclaimed_unreachable = false; - if !pruned_records.is_empty() { - let reclaimed = pruned_records + // Runtime projection is migration input only; the normal Internal registry is never + // materialized into spawned_workers.json. + runtime_dir.write_spawned_workers(&[]).await?; + if !persisted_children.is_empty() { + let reclaimed = persisted_children .iter() - .map(|record| WorkerReclaimedChild { - worker_name: record.worker_name.clone(), - scope_delegated: record - .scope_delegated - .iter() - .map(|rule| WorkerSpawnedScopeRule { - target: rule.target.clone(), - permission: match rule.permission { - Permission::Read => "read".to_string(), - Permission::Write => "write".to_string(), - }, - recursive: rule.recursive, - }) - .collect(), - }) + .map(reclaimed_child_from_metadata) .collect(); store .reclaim_spawned_children(&worker_name, reclaimed) .map_err(store_error_to_io)?; - reclaimed_unreachable = true; } - if parent_scope.is_some() { - for record in &pruned_records { - reclaim_record(&worker_name, parent_scope.as_ref(), record)?; - } + for record in &valid_records { + reclaim_record(&worker_name, parent_scope.as_ref(), record)?; } Ok(SpawnedWorkerRegistryLoad { registry: Arc::new(Self { - records: Mutex::new(records), internal_records: std::sync::Mutex::new(Vec::new()), cursors: Mutex::new(HashMap::new()), - mutations: Mutex::new(()), - runtime_dir: Some(runtime_dir), - state_writer: Some(state_writer), - reclaim_writer: Some(reclaim_writer), - parent_name: Some(worker_name), parent_scope, }), - reclaimed_unreachable, + reclaimed_unreachable: !persisted_children.is_empty(), }) } @@ -247,276 +190,23 @@ impl SpawnedWorkerRegistry { }; self.cursors.lock().await.remove(worker_name); if let (Some(record), Some(parent_scope)) = (&removed, &self.parent_scope) { - let write_rules = record - .scope_delegated - .iter() - .filter(|rule| rule.permission == Permission::Write) - .cloned() - .collect::>(); parent_scope - .update(|current| current.with_removed_deny_rules(write_rules)) + .update(|current| current.with_removed_deny_rules(delegated_write_rules(record))) .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?; } Ok(removed) } - /// Append a new legacy process record and persist the full list. - /// error if either persisted write fails; the in-memory state is still - /// updated in that case — the next successful write will reconcile. - pub async fn add(&self, record: SpawnedWorkerRecord) -> io::Result<()> { - let _mutation = self.mutations.lock().await; - let snapshot = { - let mut records = self.records.lock().await; - records.push(record); - records.clone() - }; - self.persist_records(&snapshot).await - } - - /// Look up a record by worker name. Cloned so callers can drop the lock. - pub async fn get(&self, worker_name: &str) -> Option { - self.records - .lock() - .await - .iter() - .find(|r| r.worker_name == worker_name) - .cloned() - } - - pub async fn list(&self) -> Vec { - self.records.lock().await.clone() - } - - /// Remove the record for `worker_name`, persist, clear its cursor, and - /// reclaim any delegated Write scope owned by that child. Returns the - /// removed record (if any). - pub async fn remove(&self, worker_name: &str) -> io::Result> { - let _mutation = self.mutations.lock().await; - let (removed, snapshot) = { - let mut records = self.records.lock().await; - let idx = records.iter().position(|r| r.worker_name == worker_name); - let removed = idx.map(|i| records.remove(i)); - let snapshot = records.clone(); - (removed, snapshot) - }; - self.persist_records(&snapshot).await?; - self.cursors.lock().await.remove(worker_name); - if let Some(record) = &removed { - self.reclaim_removed_record(record.clone()).await?; - } - Ok(removed) - } - - async fn reclaim_removed_record(&self, record: SpawnedWorkerRecord) -> io::Result<()> { - let parent_name = self.parent_name.clone(); - let parent_scope = self.parent_scope.clone(); - let reclaim_writer = self.reclaim_writer.clone(); - let worker_name = record.worker_name.clone(); - let reclaim = tokio::task::spawn_blocking(move || { - reclaim_removed_record_blocking(parent_name, parent_scope, reclaim_writer, record) - }); - tokio::time::timeout(REGISTRY_CLEANUP_TIMEOUT, reclaim) - .await - .map_err(|_| { - io::Error::new( - io::ErrorKind::TimedOut, - format!("timed out reclaiming spawned worker `{worker_name}`"), - ) - })? - .map_err(|err| io::Error::other(format!("spawned-worker reclaim task failed: {err}")))? - } - - /// Read-only cursor lookup. Returns 0 when no cursor has been set. pub async fn cursor(&self, worker_name: &str) -> usize { + *self.cursors.lock().await.get(worker_name).unwrap_or(&0) + } + + pub async fn set_cursor(&self, worker_name: &str, value: usize) { self.cursors .lock() .await - .get(worker_name) - .copied() - .unwrap_or(0) + .insert(worker_name.to_owned(), value); } - - pub async fn set_cursor(&self, worker_name: &str, cursor: usize) { - self.cursors - .lock() - .await - .insert(worker_name.to_string(), cursor); - } - - async fn persist_records(&self, records: &[SpawnedWorkerRecord]) -> io::Result<()> { - if let Some(runtime_dir) = &self.runtime_dir { - runtime_dir.write_spawned_workers(records).await?; - } - if let Some(write_state) = &self.state_writer { - write_state(records)?; - } - Ok(()) - } -} - -fn worker_state_writer(store: St, worker_name: String) -> RegistryStateWriter -where - St: WorkerMetadataStore + Clone + Send + Sync + 'static, -{ - Arc::new(move |records| { - write_records_to_worker_state(&store, &worker_name, records).map_err(store_error_to_io) - }) -} - -fn worker_state_reclaim_writer(store: St, worker_name: String) -> RegistryReclaimWriter -where - St: WorkerMetadataStore + Clone + Send + Sync + 'static, -{ - Arc::new(move |record| { - let reclaimed = WorkerReclaimedChild { - worker_name: record.worker_name.clone(), - scope_delegated: record - .scope_delegated - .iter() - .map(|rule| WorkerSpawnedScopeRule { - target: rule.target.clone(), - permission: match rule.permission { - Permission::Read => "read".to_string(), - Permission::Write => "write".to_string(), - }, - recursive: rule.recursive, - }) - .collect(), - }; - store - .reclaim_spawned_children(&worker_name, vec![reclaimed]) - .map(|_| ()) - .map_err(store_error_to_io) - }) -} - -fn reclaim_removed_record_blocking( - parent_name: Option, - parent_scope: Option, - reclaim_writer: Option, - record: SpawnedWorkerRecord, -) -> io::Result<()> { - if let Some(parent_name) = parent_name { - reclaim_record(&parent_name, parent_scope.as_ref(), &record)?; - } else { - release_child_allocation(&record.worker_name)?; - } - if let Some(write_reclaim) = reclaim_writer { - write_reclaim(&record)?; - } - Ok(()) -} - -fn reclaim_record( - parent_name: &str, - parent_scope: Option<&SharedScope>, - record: &SpawnedWorkerRecord, -) -> io::Result<()> { - let write_rules = record - .scope_delegated - .iter() - .filter(|rule| rule.permission == Permission::Write) - .cloned() - .collect::>(); - - let lock_path = worker_allocation::default_allocation_path() - .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?; - let mut guard = worker_allocation::LockFileGuard::open(&lock_path) - .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?; - worker_allocation::reclaim_delegated_scope( - &mut guard, - parent_name, - &record.worker_name, - &record.scope_delegated, - ) - .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?; - - if let Some(scope) = parent_scope { - scope - .update(|current| current.with_removed_deny_rules(write_rules)) - .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?; - } - - Ok(()) -} - -fn release_child_allocation(worker_name: &str) -> io::Result<()> { - let lock_path = worker_allocation::default_allocation_path() - .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?; - let mut guard = worker_allocation::LockFileGuard::open(&lock_path) - .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?; - match worker_allocation::release_worker(&mut guard, worker_name) { - Ok(()) | Err(worker_allocation::ScopeLockError::UnknownWorker(_)) => Ok(()), - Err(err) => Err(io::Error::new(io::ErrorKind::Other, err)), - } -} - -fn write_records_to_worker_state( - store: &St, - worker_name: &str, - records: &[SpawnedWorkerRecord], -) -> Result<(), WorkerStoreError> -where - St: WorkerMetadataStore, -{ - let children = records - .iter() - .map(record_to_worker_state) - .collect::, _>>()?; - store.set_spawned_children(worker_name, children)?; - Ok(()) -} - -fn record_to_worker_state( - record: &SpawnedWorkerRecord, -) -> Result { - Ok(WorkerSpawnedChild { - worker_name: record.worker_name.clone(), - socket_path: record.socket_path.clone(), - scope_delegated: record - .scope_delegated - .iter() - .map(|rule| WorkerSpawnedScopeRule { - target: rule.target.clone(), - permission: match rule.permission { - Permission::Read => "read".to_string(), - Permission::Write => "write".to_string(), - }, - recursive: rule.recursive, - }) - .collect(), - callback_address: record.callback_address.clone(), - }) -} - -fn record_from_worker_state( - child: &WorkerSpawnedChild, -) -> Result { - Ok(SpawnedWorkerRecord { - worker_name: child.worker_name.clone(), - socket_path: child.socket_path.clone(), - scope_delegated: child - .scope_delegated - .iter() - .map(|rule| { - Ok(ScopeRule { - target: rule.target.clone(), - permission: match rule.permission.as_str() { - "read" => Permission::Read, - "write" => Permission::Write, - other => { - return Err(serde_json::Error::io(io::Error::new( - io::ErrorKind::InvalidData, - format!("invalid permission `{other}`"), - ))); - } - }, - recursive: rule.recursive, - }) - }) - .collect::, _>>()?, - callback_address: child.callback_address.clone(), - }) } impl Drop for SpawnedWorkerRegistry { @@ -529,14 +219,90 @@ impl Drop for SpawnedWorkerRegistry { }; let write_rules = records .iter() - .flat_map(|record| record.scope_delegated.iter()) - .filter(|rule| rule.permission == Permission::Write) - .cloned() + .flat_map(delegated_write_rules) .collect::>(); let _ = parent_scope.update(|current| current.with_removed_deny_rules(write_rules)); } } +fn delegated_write_rules(record: &InternalSpawnedWorkerRecord) -> Vec { + record + .scope_delegated + .iter() + .filter(|rule| rule.permission == Permission::Write) + .cloned() + .collect() +} + +fn reclaimed_child_from_metadata(child: &WorkerSpawnedChild) -> WorkerReclaimedChild { + WorkerReclaimedChild { + worker_name: child.worker_name.clone(), + scope_delegated: child.scope_delegated.clone(), + } +} + +fn reclaim_record( + parent_name: &str, + parent_scope: Option<&SharedScope>, + record: &SpawnedWorkerRecord, +) -> io::Result<()> { + if let Ok(path) = worker_allocation::default_allocation_path() { + if let Ok(mut guard) = worker_allocation::LockFileGuard::open(&path) { + match worker_allocation::reclaim_delegated_scope( + &mut guard, + parent_name, + &record.worker_name, + &record.scope_delegated, + ) { + Ok(()) | Err(worker_allocation::ScopeLockError::UnknownWorker(_)) => {} + Err(error) => return Err(io::Error::other(error)), + } + } + } + if let Some(parent_scope) = parent_scope { + let write_rules = record + .scope_delegated + .iter() + .filter(|rule| rule.permission == Permission::Write) + .cloned() + .collect::>(); + parent_scope + .update(|current| current.with_removed_deny_rules(write_rules)) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?; + } + Ok(()) +} + +fn record_from_worker_state(child: &WorkerSpawnedChild) -> io::Result { + let scope_delegated = child + .scope_delegated + .iter() + .map(|rule| { + let permission = match rule.permission.as_str() { + "read" => Permission::Read, + "write" => Permission::Write, + other => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unsupported spawned-worker permission `{other}`"), + )); + } + }; + Ok(ScopeRule { + target: rule.target.clone(), + permission, + recursive: rule.recursive, + }) + }) + .collect::>>()?; + Ok(SpawnedWorkerRecord { + worker_name: child.worker_name.clone(), + socket_path: child.socket_path.clone(), + scope_delegated, + callback_address: child.callback_address.clone(), + }) +} + fn store_error_to_io(error: WorkerStoreError) -> io::Error { io::Error::other(error) } diff --git a/crates/worker/tests/worker_events_test.rs b/crates/worker/tests/worker_events_test.rs index 5860dc45..ee4c8c15 100644 --- a/crates/worker/tests/worker_events_test.rs +++ b/crates/worker/tests/worker_events_test.rs @@ -1,423 +1,45 @@ -//! Integration tests for the `WorkerEvent` send / receive primitive. -//! -//! These tests drive `worker_events::fire_and_forget` and -//! `worker_events::apply_event_side_effects` directly — the full -//! Controller wiring is exercised by the existing controller / -//! spawn-worker tests, which rely on the same primitives. +//! Legacy process callback events are diagnostics only after Internal SubWorker migration. -use std::path::PathBuf; -use std::sync::{Arc, LazyLock, Mutex}; -use std::time::Duration; +use std::sync::Arc; -use protocol::stream::{JsonLineReader, JsonLineWriter}; -use protocol::{Event, Greeting, Method, Permission, ScopeRule, WorkerEvent, WorkerStatus}; +use protocol::{Permission, ScopeRule, WorkerEvent}; use tempfile::TempDir; -use tokio::net::UnixListener; -use worker::ipc::event::{apply_event_side_effects, fire_and_forget, render_event}; -use worker::runtime::dir::{RuntimeDir, SpawnedWorkerRecord}; -use worker::runtime::worker_allocation::{self, LockFileGuard}; +use worker::ipc::event::{apply_event_side_effects, render_event}; +use worker::runtime::dir::RuntimeDir; use worker::spawn::registry::SpawnedWorkerRegistry; -/// Serialises tests that mutate `YOI_RUNTIME_DIR`. -static ENV_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); - -/// Take `ENV_LOCK` and clear any env vars that would outrank -/// `YOI_RUNTIME_DIR`; restore previous values on drop. -struct EnvGuard { - prev_home: Option, - prev_xdg: Option, - _lock: std::sync::MutexGuard<'static, ()>, -} - -impl EnvGuard { - fn acquire() -> Self { - let lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); - let prev_home = std::env::var("YOI_HOME").ok(); - let prev_xdg = std::env::var("XDG_RUNTIME_DIR").ok(); - unsafe { - std::env::remove_var("YOI_HOME"); - std::env::remove_var("XDG_RUNTIME_DIR"); - } - Self { - prev_home, - prev_xdg, - _lock: lock, - } - } -} - -impl Drop for EnvGuard { - fn drop(&mut self) { - unsafe { - match &self.prev_home { - Some(v) => std::env::set_var("YOI_HOME", v), - None => std::env::remove_var("YOI_HOME"), - } - match &self.prev_xdg { - Some(v) => std::env::set_var("XDG_RUNTIME_DIR", v), - None => std::env::remove_var("XDG_RUNTIME_DIR"), - } - std::env::remove_var("YOI_RUNTIME_DIR"); - } - } -} - -/// Point `YOI_RUNTIME_DIR` at `dir`. The worker-allocation then lives at -/// `/workers.json` and Worker runtime sub-dirs at `/{worker_name}/`. -fn set_runtime_dir(dir: &std::path::Path) { - unsafe { - std::env::set_var("YOI_RUNTIME_DIR", dir); - } -} - -fn clear_runtime_dir() { - unsafe { - std::env::remove_var("YOI_RUNTIME_DIR"); - } -} - -/// Minimal connect-time snapshot used by mock parent sockets. -fn empty_snapshot() -> Event { - Event::Snapshot { - entries: Vec::new(), - greeting: Greeting { - worker_name: "parent".into(), - cwd: "/tmp".into(), - provider: "test".into(), - model: "test".into(), - scope_summary: String::new(), - tools: Vec::new(), - context_window: 200_000, - context_tokens: 0, - }, - status: WorkerStatus::Idle, - in_flight: Default::default(), - } -} - -/// Accept a single connection, send the protocol's connect-time snapshot, -/// read one `Method`, and return it. -fn accept_one_method(listener: UnixListener) -> tokio::task::JoinHandle> { - tokio::spawn(async move { - let (stream, _) = listener.accept().await.ok()?; - let (reader, writer) = stream.into_split(); - let mut w = JsonLineWriter::new(writer); - w.write(&empty_snapshot()).await.ok()?; - let mut r = JsonLineReader::new(reader); - r.next::().await.ok().flatten() - }) -} - #[test] -fn render_event_all_variants_mention_worker_name() { - let t1 = render_event(&WorkerEvent::TurnEnded { - worker_name: "alpha".into(), - }); - assert!(t1.contains("alpha"), "{t1}"); - - let t2 = render_event(&WorkerEvent::Errored { - worker_name: "bravo".into(), +fn render_event_keeps_bounded_legacy_diagnostics() { + let rendered = render_event(&WorkerEvent::Errored { + worker_name: "legacy-child".into(), message: "boom".into(), }); - assert!(t2.contains("bravo") && t2.contains("boom"), "{t2}"); - - let t3 = render_event(&WorkerEvent::ShutDown { - worker_name: "charlie".into(), - }); - assert!(t3.contains("charlie"), "{t3}"); - - let t4 = render_event(&WorkerEvent::ScopeSubDelegated { - parent_worker: "delta".into(), - sub_worker: "echo".into(), - sub_socket: "/tmp/sock".into(), - scope: vec![], - }); - assert!(t4.contains("delta") && t4.contains("echo"), "{t4}"); + assert!(rendered.contains("legacy-child")); + assert!(rendered.contains("boom")); } #[tokio::test] -async fn fire_and_forget_delivers_worker_event_to_listener() { - let dir = TempDir::new().unwrap(); - let socket_path = dir.path().join("parent.sock"); - let listener = UnixListener::bind(&socket_path).unwrap(); - let received = accept_one_method(listener); - - fire_and_forget( - Some(socket_path.clone()), - WorkerEvent::TurnEnded { - worker_name: "child".into(), - }, - ); - - let method = tokio::time::timeout(Duration::from_secs(2), received) - .await - .expect("send timed out") - .unwrap() - .expect("no method received"); - match method { - Method::WorkerEvent(WorkerEvent::TurnEnded { worker_name }) => { - assert_eq!(worker_name, "child") - } - other => panic!("expected TurnEnded, got {other:?}"), - } -} - -#[tokio::test] -async fn fire_and_forget_with_none_socket_is_noop() { - // Nothing binds and nothing listens; the call must not panic and - // must not leak a task that never completes. - fire_and_forget( - None, - WorkerEvent::ShutDown { - worker_name: "x".into(), - }, - ); - // Yield once so any accidentally-spawned task would surface. - tokio::time::sleep(Duration::from_millis(50)).await; -} - -/// Build a registry backed by a fresh runtime dir. -async fn fresh_registry( - runtime_base: &std::path::Path, - worker_name: &str, -) -> Arc { - let rd = RuntimeDir::create(runtime_base, worker_name).await.unwrap(); - SpawnedWorkerRegistry::new(Arc::new(rd)) -} - -#[tokio::test] -async fn apply_shutdown_removes_from_registry_and_tolerates_missing() { - let _env = EnvGuard::acquire(); - let scope_dir = TempDir::new().unwrap(); - set_runtime_dir(scope_dir.path()); - +async fn legacy_callback_cannot_register_process_subworker_authority() { let runtime_base = TempDir::new().unwrap(); - let registry = fresh_registry(runtime_base.path(), "parent").await; - - // Seed a child record; then ShutDown for it should remove it. - registry - .add(SpawnedWorkerRecord { - worker_name: "child".into(), - socket_path: "/tmp/child.sock".into(), - scope_delegated: vec![], - callback_address: "/tmp/parent.sock".into(), - }) - .await - .unwrap(); - - let event = WorkerEvent::ShutDown { - worker_name: "child".into(), - }; - apply_event_side_effects(&event, ®istry, "parent", &None).await; - assert!(registry.get("child").await.is_none()); - - // Second ShutDown for the same (now-missing) child must be a no-op, - // not an error — this is the idempotency guarantee for out-of-order - // delivery. - apply_event_side_effects(&event, ®istry, "parent", &None).await; - assert!(registry.get("child").await.is_none()); - - clear_runtime_dir(); -} - -#[tokio::test] -async fn apply_scope_sub_delegated_adds_grandchild_then_duplicate_is_noop() { - let _env = EnvGuard::acquire(); - let scope_dir = TempDir::new().unwrap(); - set_runtime_dir(scope_dir.path()); - - let runtime_base = TempDir::new().unwrap(); - let registry = fresh_registry(runtime_base.path(), "grandparent").await; - - // Seed the intermediate child so callback_address lookup succeeds. - registry - .add(SpawnedWorkerRecord { - worker_name: "child".into(), - socket_path: "/tmp/child.sock".into(), - scope_delegated: vec![], - callback_address: "/tmp/grandparent.sock".into(), - }) - .await - .unwrap(); - + let runtime_dir = Arc::new( + RuntimeDir::create(runtime_base.path(), "parent") + .await + .unwrap(), + ); + let registry = SpawnedWorkerRegistry::new(runtime_dir.clone()); + let scope_root = TempDir::new().unwrap(); let event = WorkerEvent::ScopeSubDelegated { - parent_worker: "child".into(), - sub_worker: "grandchild".into(), - sub_socket: "/tmp/grandchild.sock".into(), + parent_worker: "legacy-parent".into(), + sub_worker: "legacy-child".into(), + sub_socket: "/tmp/legacy-child.sock".into(), scope: vec![ScopeRule { - target: scope_dir.path().to_path_buf(), + target: scope_root.path().to_path_buf(), permission: Permission::Write, recursive: true, }], }; - apply_event_side_effects(&event, ®istry, "grandparent", &None).await; - let gc = registry - .get("grandchild") - .await - .expect("grandchild missing after ScopeSubDelegated"); - assert_eq!(gc.socket_path, PathBuf::from("/tmp/grandchild.sock")); - assert_eq!(gc.callback_address, PathBuf::from("/tmp/child.sock")); + apply_event_side_effects(&event, ®istry, "parent", &None).await; - // Duplicate delivery must not error and must not overwrite. - apply_event_side_effects(&event, ®istry, "grandparent", &None).await; - let gc2 = registry.get("grandchild").await.unwrap(); - assert_eq!(gc2.socket_path, PathBuf::from("/tmp/grandchild.sock")); - - clear_runtime_dir(); -} - -#[tokio::test] -async fn apply_scope_sub_delegated_reemits_to_own_parent() { - let _env = EnvGuard::acquire(); - let scope_dir = TempDir::new().unwrap(); - set_runtime_dir(scope_dir.path()); - - let runtime_base = TempDir::new().unwrap(); - let registry = fresh_registry(runtime_base.path(), "B").await; - - // Bind a listener at "A's" socket so we can watch the re-emission - // climb one level up the tree. - let sock_dir = TempDir::new().unwrap(); - let a_socket = sock_dir.path().join("A.sock"); - let listener = UnixListener::bind(&a_socket).unwrap(); - let received = accept_one_method(listener); - - // Seed the child record that the event claims spawned the grandchild. - registry - .add(SpawnedWorkerRecord { - worker_name: "C".into(), - socket_path: "/tmp/C.sock".into(), - scope_delegated: vec![], - callback_address: "/tmp/B.sock".into(), - }) - .await - .unwrap(); - - let event = WorkerEvent::ScopeSubDelegated { - parent_worker: "C".into(), - sub_worker: "D".into(), - sub_socket: "/tmp/D.sock".into(), - scope: vec![], - }; - - // Self is B, and B's parent socket is A's listener. - apply_event_side_effects(&event, ®istry, "B", &Some(a_socket.clone())).await; - - // A must see the re-emission with parent_worker set to "B" (the - // sender from A's perspective), not "C" (the original sender's - // local view). - let method = tokio::time::timeout(Duration::from_secs(2), received) - .await - .expect("re-emission timed out") - .unwrap() - .expect("no method received on A's socket"); - match method { - Method::WorkerEvent(WorkerEvent::ScopeSubDelegated { - parent_worker, - sub_worker, - .. - }) => { - assert_eq!(parent_worker, "B"); - assert_eq!(sub_worker, "D"); - } - other => panic!("expected re-emitted ScopeSubDelegated, got {other:?}"), - } - - clear_runtime_dir(); -} - -#[tokio::test] -async fn apply_turn_ended_and_errored_are_system_noops() { - let _env = EnvGuard::acquire(); - let scope_dir = TempDir::new().unwrap(); - set_runtime_dir(scope_dir.path()); - - let runtime_base = TempDir::new().unwrap(); - let registry = fresh_registry(runtime_base.path(), "parent").await; - - // Seed a child to verify it survives the no-op path. - registry - .add(SpawnedWorkerRecord { - worker_name: "child".into(), - socket_path: "/tmp/child.sock".into(), - scope_delegated: vec![], - callback_address: "/tmp/parent.sock".into(), - }) - .await - .unwrap(); - - apply_event_side_effects( - &WorkerEvent::TurnEnded { - worker_name: "child".into(), - }, - ®istry, - "parent", - &None, - ) - .await; - apply_event_side_effects( - &WorkerEvent::Errored { - worker_name: "child".into(), - message: "x".into(), - }, - ®istry, - "parent", - &None, - ) - .await; - - assert!(registry.get("child").await.is_some()); - clear_runtime_dir(); -} - -#[tokio::test] -async fn shutdown_releases_scope_allocation_when_present() { - let _env = EnvGuard::acquire(); - let scope_dir = TempDir::new().unwrap(); - let lock_path = scope_dir.path().join("workers.json"); - set_runtime_dir(scope_dir.path()); - - // Install a top-level allocation for "kid" so ShutDown has - // something to release. - let guard = worker_allocation::install_top_level( - "kid".into(), - std::process::id(), - "/tmp/kid.sock".into(), - vec![], - session_store::new_segment_id(), - ) - .unwrap(); - std::mem::forget(guard); - - let runtime_base = TempDir::new().unwrap(); - let registry = fresh_registry(runtime_base.path(), "parent").await; - registry - .add(SpawnedWorkerRecord { - worker_name: "kid".into(), - socket_path: "/tmp/kid.sock".into(), - scope_delegated: vec![], - callback_address: "/tmp/parent.sock".into(), - }) - .await - .unwrap(); - - apply_event_side_effects( - &WorkerEvent::ShutDown { - worker_name: "kid".into(), - }, - ®istry, - "parent", - &None, - ) - .await; - - // Allocation is gone from the worker-allocation. - let g = LockFileGuard::open(&lock_path).unwrap(); - assert!( - g.data().find("kid").is_none(), - "ShutDown should have released the scope allocation" - ); - - clear_runtime_dir(); + assert!(!runtime_dir.path().join("spawned_workers.json").exists()); }