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

worker: retire process sub-worker registry authority

Browse Source
This commit is contained in:
Hare
2026-08-07 02:03:09 +09:00
parent 8f0d7fa3c0
commit cf394403a6
4 changed files with 152 additions and 850 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+2 -16
View File
@@ -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"
);
}
crates/worker/src/ipc/event.rs
+9 -81
View File
@@ -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<SpawnedWorkerRegistry>,
self_name: &str,
self_parent_socket: &Option<PathBuf>,
_event: &WorkerEvent,
_registry: &Arc<SpawnedWorkerRegistry>,
_self_name: &str,
_self_parent_socket: &Option<PathBuf>,
) {
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<PathBuf>,
self_name: &str,
sub_worker: String,
sub_socket: PathBuf,
scope: Vec<ScopeRule>,
) {
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);
}
crates/worker/src/spawn/registry.rs
+116 -350
View File
@@ -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<dyn Fn(&[SpawnedWorkerRecord]) -> io::Result<()> + Send + Sync>;
type RegistryReclaimWriter = Arc<dyn Fn(&SpawnedWorkerRecord) -> 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<Vec<SpawnedWorkerRecord>>,
internal_records: std::sync::Mutex<Vec<InternalSpawnedWorkerRecord>>,
cursors: Mutex<HashMap<String, usize>>,
mutations: Mutex<()>,
runtime_dir: Option<Arc<RuntimeDir>>,
state_writer: Option<RegistryStateWriter>,
reclaim_writer: Option<RegistryReclaimWriter>,
parent_name: Option<String>,
parent_scope: Option<SharedScope>,
}
pub struct SpawnedWorkerRegistryLoad {
pub registry: Arc<SpawnedWorkerRegistry>,
/// True when obsolete process-child metadata was consumed and cleared.
pub reclaimed_unreachable: bool,
}
impl SpawnedWorkerRegistry {
pub fn new(runtime_dir: Arc<RuntimeDir>) -> Arc<Self> {
/// Empty registry used by tests and non-spawning projections.
pub fn new(_runtime_dir: Arc<RuntimeDir>) -> Arc<Self> {
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<Self> {
pub(crate) fn new_internal(_parent_name: String, parent_scope: SharedScope) -> Arc<Self> {
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<St>(
runtime_dir: Arc<RuntimeDir>,
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<St>(
runtime_dir: Arc<RuntimeDir>,
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) => {
warn!(
error = %err,
worker = %child.worker_name,
"dropping corrupt persisted spawned-worker record"
);
continue;
}
};
match record_from_worker_state(child) {
Ok(record) => {
warn!(
worker = %record.worker_name,
"reclaiming legacy persisted process Sub-worker during Internal session restore"
"reclaiming legacy persisted process SubWorker during Internal session restore"
);
pruned_records.push(record);
valid_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 {
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::<Vec<_>>();
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<SpawnedWorkerRecord> {
self.records
.lock()
.await
.iter()
.find(|r| r.worker_name == worker_name)
.cloned()
}
pub async fn list(&self) -> Vec<SpawnedWorkerRecord> {
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<Option<SpawnedWorkerRecord>> {
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<St>(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<St>(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<String>,
parent_scope: Option<SharedScope>,
reclaim_writer: Option<RegistryReclaimWriter>,
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::<Vec<_>>();
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<St>(
store: &St,
worker_name: &str,
records: &[SpawnedWorkerRecord],
) -> Result<(), WorkerStoreError>
where
St: WorkerMetadataStore,
{
let children = records
.iter()
.map(record_to_worker_state)
.collect::<Result<Vec<_>, _>>()?;
store.set_spawned_children(worker_name, children)?;
Ok(())
}
fn record_to_worker_state(
record: &SpawnedWorkerRecord,
) -> Result<WorkerSpawnedChild, serde_json::Error> {
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<SpawnedWorkerRecord, serde_json::Error> {
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::<Result<Vec<_>, _>>()?,
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::<Vec<_>>();
let _ = parent_scope.update(|current| current.with_removed_deny_rules(write_rules));
}
}
fn delegated_write_rules(record: &InternalSpawnedWorkerRecord) -> Vec<ScopeRule> {
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::<Vec<_>>();
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<SpawnedWorkerRecord> {
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::<io::Result<Vec<_>>>()?;
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)
}
crates/worker/tests/worker_events_test.rs
+23 -401
View File
@@ -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<Mutex<()>> = 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<String>,
prev_xdg: Option<String>,
_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
/// `<dir>/workers.json` and Worker runtime sub-dirs at `<dir>/{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<Option<Method>> {
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::<Method>().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<SpawnedWorkerRegistry> {
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(),
})
let runtime_dir = Arc::new(
RuntimeDir::create(runtime_base.path(), "parent")
.await
.unwrap();
let event = WorkerEvent::ShutDown {
worker_name: "child".into(),
};
apply_event_side_effects(&event, &registry, "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, &registry, "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();
.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, &registry, "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, &registry, "parent", &None).await;
// Duplicate delivery must not error and must not overwrite.
apply_event_side_effects(&event, &registry, "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, &registry, "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(),
},
&registry,
"parent",
&None,
)
.await;
apply_event_side_effects(
&WorkerEvent::Errored {
worker_name: "child".into(),
message: "x".into(),
},
&registry,
"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(),
},
&registry,
"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());
}