feat: unify headless execution under the Standalone Host
Expose an optional private same-uid Unix operator adapter and JSONL serve/attach CLI using the production Standalone Profile, store, restore lease and shutdown lifecycle. Remove direct Worker subprocess/discovery/readiness/controller sockets and parent socket callbacks; retain metadata-only peer registration and legacy safety readers. Validated real Host/client input, events, provenance, disconnect, admitted shutdown and shared production restore/duplicate exclusion; normal standalone/worker/manifest/protocol/client/yoi/tui/server tests, root check, rustdoc, fmt and state-contract audit. Runtime lib: 476 passed, damaged_checkout_does_not_block_restore_but_is_rejected_when_used fails at missing_git diagnostic; unchanged restore logic, not relaxed or retried. Runtime bins/docs pass.
This commit is contained in:
@@ -11,7 +11,6 @@ default = []
|
||||
|
||||
[dependencies]
|
||||
async-trait = { workspace = true }
|
||||
clap = { version = "4.6.0", features = ["derive"] }
|
||||
agen = { workspace = true, features = ["codex"] }
|
||||
session-store = { workspace = true }
|
||||
secrets = { workspace = true }
|
||||
|
||||
@@ -6,9 +6,7 @@ use thiserror::Error;
|
||||
use workdir::WorkdirSessionHandle;
|
||||
|
||||
use crate::PromptCatalogSource;
|
||||
use crate::controller::{
|
||||
ShutdownReceiver, WorkerController, WorkerControllerTransport, WorkerHandle,
|
||||
};
|
||||
use crate::controller::{ShutdownReceiver, WorkerController, WorkerHandle};
|
||||
use crate::worker::{Worker, WorkerError, WorkerFilesystemAuthority, WorkerWorkspaceContext};
|
||||
use manifest::WorkerManifest;
|
||||
|
||||
@@ -30,7 +28,6 @@ pub struct WorkerBootstrap<St> {
|
||||
workspace_context: WorkerWorkspaceContext,
|
||||
filesystem_authority: WorkerFilesystemAuthority,
|
||||
layout: WorkerBootstrapLayout,
|
||||
transport: WorkerControllerTransport,
|
||||
model_client: Option<Box<dyn LlmClient>>,
|
||||
workdir_session: Option<WorkdirSessionHandle>,
|
||||
subjektiv_host: Option<std::sync::Arc<dyn crate::subjektiv::SubjektivHost>>,
|
||||
@@ -41,7 +38,6 @@ pub struct WorkerBootstrap<St> {
|
||||
pub struct PreparedWorker<C: LlmClient, St: Store> {
|
||||
worker: Worker<C, St>,
|
||||
layout: WorkerBootstrapLayout,
|
||||
transport: WorkerControllerTransport,
|
||||
}
|
||||
|
||||
/// Live controller returned only after Worker construction and feature
|
||||
@@ -68,7 +64,6 @@ impl<St> WorkerBootstrap<St>
|
||||
where
|
||||
St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static,
|
||||
{
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub fn new(
|
||||
manifest: WorkerManifest,
|
||||
store: St,
|
||||
@@ -76,7 +71,6 @@ where
|
||||
workspace_context: WorkerWorkspaceContext,
|
||||
filesystem_authority: WorkerFilesystemAuthority,
|
||||
layout: WorkerBootstrapLayout,
|
||||
transport: WorkerControllerTransport,
|
||||
) -> Self {
|
||||
Self {
|
||||
manifest,
|
||||
@@ -85,7 +79,6 @@ where
|
||||
workspace_context,
|
||||
filesystem_authority,
|
||||
layout,
|
||||
transport,
|
||||
model_client: None,
|
||||
workdir_session: None,
|
||||
subjektiv_host: None,
|
||||
@@ -165,7 +158,7 @@ where
|
||||
)
|
||||
.map_err(WorkerBootstrapError::Worker)?;
|
||||
}
|
||||
Ok(PreparedWorker::new(worker, self.layout, self.transport))
|
||||
Ok(PreparedWorker::new(worker, self.layout))
|
||||
}
|
||||
|
||||
pub async fn prepare_restored(
|
||||
@@ -206,7 +199,7 @@ where
|
||||
)
|
||||
.map_err(WorkerBootstrapError::Worker)?;
|
||||
}
|
||||
Ok(PreparedWorker::new(worker, self.layout, self.transport))
|
||||
Ok(PreparedWorker::new(worker, self.layout))
|
||||
}
|
||||
|
||||
pub async fn start(self) -> Result<BootstrappedWorker, WorkerBootstrapError> {
|
||||
@@ -221,16 +214,8 @@ where
|
||||
{
|
||||
/// Wrap a restored Worker in the same pre-exposure lifecycle used by fresh
|
||||
/// bootstraps.
|
||||
pub fn new(
|
||||
worker: Worker<C, St>,
|
||||
layout: WorkerBootstrapLayout,
|
||||
transport: WorkerControllerTransport,
|
||||
) -> Self {
|
||||
Self {
|
||||
worker,
|
||||
layout,
|
||||
transport,
|
||||
}
|
||||
pub fn new(worker: Worker<C, St>, layout: WorkerBootstrapLayout) -> Self {
|
||||
Self { worker, layout }
|
||||
}
|
||||
|
||||
pub fn worker(&self) -> &Worker<C, St> {
|
||||
@@ -242,7 +227,7 @@ where
|
||||
}
|
||||
|
||||
pub async fn start(self) -> Result<BootstrappedWorker, WorkerBootstrapError> {
|
||||
start_worker_controller(self.worker, self.layout, self.transport).await
|
||||
start_worker_controller(self.worker, self.layout).await
|
||||
}
|
||||
}
|
||||
|
||||
@@ -252,7 +237,6 @@ where
|
||||
pub async fn start_worker_controller<C, St>(
|
||||
worker: Worker<C, St>,
|
||||
layout: WorkerBootstrapLayout,
|
||||
transport: WorkerControllerTransport,
|
||||
) -> Result<BootstrappedWorker, WorkerBootstrapError>
|
||||
where
|
||||
C: LlmClient + Clone + 'static,
|
||||
@@ -261,13 +245,10 @@ where
|
||||
let cleanup_sessions = worker.workdir_sessions();
|
||||
let controller = match layout {
|
||||
WorkerBootstrapLayout::Direct { runtime_base } => {
|
||||
WorkerController::spawn_with_transport_owned(worker, &runtime_base, transport).await
|
||||
WorkerController::spawn_owned(worker, &runtime_base).await
|
||||
}
|
||||
WorkerBootstrapLayout::RuntimeManagedRun { run_dir } => {
|
||||
WorkerController::spawn_runtime_managed_run_with_transport_owned(
|
||||
worker, &run_dir, transport,
|
||||
)
|
||||
.await
|
||||
WorkerController::spawn_runtime_managed_run_owned(worker, &run_dir).await
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
+59
-679
File diff suppressed because it is too large
Load Diff
@@ -65,7 +65,6 @@ async fn prepared(
|
||||
WorkerBootstrapLayout::Direct {
|
||||
runtime_base: temp.path().join("runtime"),
|
||||
},
|
||||
WorkerControllerTransport::InProcess,
|
||||
)
|
||||
.with_model_client(client)
|
||||
.prepare()
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -1,57 +1,6 @@
|
||||
//! `WorkerEvent` send / receive helpers.
|
||||
//!
|
||||
//! This module owns the parent-facing lifecycle-event primitive
|
||||
//! (`WorkerEvent`) that children fire upward on turn-end / error /
|
||||
//! shutdown / scope-sub-delegation. Three responsibilities live here:
|
||||
//!
|
||||
//! - **Send** a `Method::WorkerEvent` to the parent socket, fire-and-forget,
|
||||
//! logging failures without blocking the child.
|
||||
//! - **Render** agent-visible variants into human-readable strings for the
|
||||
//! parent's notification buffer. Control-plane-only variants may still have
|
||||
//! a renderer for diagnostics, but receive-side classification keeps them
|
||||
//! out of LLM history/context.
|
||||
//! - **Apply side effects** on the parent (registry / worker-allocation
|
||||
//! updates) so that the receive path is idempotent and tolerant of
|
||||
//! out-of-order delivery.
|
||||
//!
|
||||
//! Transport is fire-and-forget — the ticket's decision is that
|
||||
//! callbacks are an optimisation and `ListWorkers` + `reclaim_stale` are
|
||||
//! the real fallback. This module is allowed to drop events on the
|
||||
//! floor (with a warn log) rather than retry.
|
||||
//!
|
||||
//! `apply_event_side_effects` takes its dependencies (registry, scope
|
||||
//! lock path, self identity) by reference so the caller owns lifetime
|
||||
//! and locking concerns.
|
||||
//! Agent-visible rendering of typed Internal SubWorker lifecycle events.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use protocol::{Method, WorkerEvent};
|
||||
|
||||
use crate::spawn::comm_tools::connect_and_send;
|
||||
use crate::spawn::registry::SpawnedWorkerRegistry;
|
||||
|
||||
/// Connect to `socket`, send a single `Method::WorkerEvent(event)`, and
|
||||
/// return. Used by children to report up to their parent.
|
||||
///
|
||||
/// This is a synchronous helper — callers that want fire-and-forget
|
||||
/// semantics should wrap the call in `tokio::spawn` themselves.
|
||||
pub async fn send_worker_event(socket: &Path, event: WorkerEvent) -> std::io::Result<()> {
|
||||
connect_and_send(socket, &Method::WorkerEvent(event)).await
|
||||
}
|
||||
|
||||
/// Spawn a fire-and-forget task that sends `event` to `socket`. If
|
||||
/// `socket` is `None`, no send happens (top-level Workers have no parent).
|
||||
/// Any send failure is logged at warn level but otherwise ignored —
|
||||
/// the parent is treated as best-effort.
|
||||
pub fn fire_and_forget(socket: Option<PathBuf>, event: WorkerEvent) {
|
||||
let Some(socket) = socket else { return };
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = send_worker_event(&socket, event).await {
|
||||
tracing::warn!(error = %e, socket = %socket.display(), "WorkerEvent send failed");
|
||||
}
|
||||
});
|
||||
}
|
||||
use protocol::WorkerEvent;
|
||||
|
||||
/// Render a variant into a one-line human-readable string.
|
||||
///
|
||||
@@ -83,16 +32,3 @@ pub fn render_event(event: &WorkerEvent) -> String {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Legacy process callback events have no SubWorker registry authority.
|
||||
///
|
||||
/// 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>,
|
||||
) {
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
pub mod alerter;
|
||||
pub mod event;
|
||||
pub mod protocol_session;
|
||||
pub mod server;
|
||||
|
||||
pub(crate) mod interceptor;
|
||||
pub(crate) mod notify_buffer;
|
||||
|
||||
@@ -1,224 +0,0 @@
|
||||
use std::io;
|
||||
use std::io::ErrorKind;
|
||||
use std::path::PathBuf;
|
||||
|
||||
use protocol::stream::{JsonLineReader, JsonLineWriter};
|
||||
use tokio::net::UnixListener;
|
||||
use tokio::sync::oneshot;
|
||||
use tokio::task::{JoinHandle, JoinSet};
|
||||
|
||||
use crate::controller::WorkerHandle;
|
||||
use crate::ipc::protocol_session::{
|
||||
dispatch_worker_protocol_method, live_log_entry_event, subscribe_worker_protocol_session,
|
||||
};
|
||||
use protocol::{ErrorCode, Event};
|
||||
|
||||
/// Unix socket server for Worker Protocol.
|
||||
///
|
||||
/// Listens on the Worker's runtime directory socket path.
|
||||
/// Each client connection gets bidirectional JSONL:
|
||||
/// - Client writes Method lines → forwarded to WorkerController
|
||||
/// - Worker events → written as Event lines to all connected clients
|
||||
pub struct SocketServer {
|
||||
accept_task: Option<JoinHandle<()>>,
|
||||
shutdown: Option<oneshot::Sender<()>>,
|
||||
path: PathBuf,
|
||||
}
|
||||
|
||||
impl SocketServer {
|
||||
/// Start listening on the WorkerHandle's socket path.
|
||||
pub async fn start(handle: &WorkerHandle) -> Result<Self, io::Error> {
|
||||
let path = handle.runtime_dir.socket_path();
|
||||
|
||||
// Remove stale socket file if it exists
|
||||
let _ = tokio::fs::remove_file(&path).await;
|
||||
|
||||
let listener = UnixListener::bind(&path)?;
|
||||
let handle = handle.clone();
|
||||
let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
|
||||
|
||||
let accept_task = tokio::spawn(async move {
|
||||
let mut connections = JoinSet::new();
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = &mut shutdown_rx => break,
|
||||
accepted = listener.accept() => match accepted {
|
||||
Ok((stream, _)) => {
|
||||
let handle = handle.clone();
|
||||
connections.spawn(handle_connection(stream, handle));
|
||||
}
|
||||
Err(_) => break,
|
||||
},
|
||||
completed = connections.join_next(), if !connections.is_empty() => {
|
||||
let _ = completed;
|
||||
}
|
||||
}
|
||||
}
|
||||
connections.shutdown().await;
|
||||
});
|
||||
|
||||
Ok(Self {
|
||||
accept_task: Some(accept_task),
|
||||
shutdown: Some(shutdown_tx),
|
||||
path,
|
||||
})
|
||||
}
|
||||
|
||||
/// Stop accepting connections and join the server task. Dropping its
|
||||
/// `JoinSet` cancels every active connection task before this returns.
|
||||
pub async fn shutdown(mut self) {
|
||||
if let Some(shutdown) = self.shutdown.take() {
|
||||
let _ = shutdown.send(());
|
||||
}
|
||||
if let Some(task) = self.accept_task.take() {
|
||||
let _ = task.await;
|
||||
}
|
||||
let _ = tokio::fs::remove_file(&self.path).await;
|
||||
}
|
||||
|
||||
/// The socket file path.
|
||||
pub fn path(&self) -> &std::path::Path {
|
||||
&self.path
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for SocketServer {
|
||||
fn drop(&mut self) {
|
||||
if let Some(shutdown) = self.shutdown.take() {
|
||||
let _ = shutdown.send(());
|
||||
}
|
||||
let _ = self.accept_task.take();
|
||||
let _ = std::fs::remove_file(&self.path);
|
||||
}
|
||||
}
|
||||
|
||||
fn is_peer_disconnect_read_error(error: &io::Error) -> bool {
|
||||
matches!(
|
||||
error.kind(),
|
||||
ErrorKind::ConnectionReset
|
||||
| ErrorKind::ConnectionAborted
|
||||
| ErrorKind::BrokenPipe
|
||||
| ErrorKind::UnexpectedEof
|
||||
)
|
||||
}
|
||||
|
||||
async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle) {
|
||||
let (reader, writer) = stream.into_split();
|
||||
let mut reader = JsonLineReader::new(reader);
|
||||
let mut writer = JsonLineWriter::new(writer);
|
||||
|
||||
// Hold the in-flight stream lock while taking the session-log mirror
|
||||
// snapshot. `LogEntry::AnnotatedAssistantItem` is mirror-only for live clients,
|
||||
// so a finalized assistant block must be observed either as an already
|
||||
// committed entry or as the still-present in-flight block. This lock
|
||||
// order matches `append_entry` (in-flight clear before sink publish) and
|
||||
// keeps the snapshot/live boundary gap-free.
|
||||
let mut streams = subscribe_worker_protocol_session(&handle);
|
||||
for alert in streams.alert_snapshot {
|
||||
if writer.write_event(&Event::Alert(alert)).await.is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Send the typed snapshot up front so late attachers can
|
||||
// reconstruct view state without an extra round trip.
|
||||
if writer.write_event(&streams.snapshot_event).await.is_err() {
|
||||
return;
|
||||
}
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
// Live session-log entries → dispatched as the role-specific
|
||||
// wire events. `SegmentLogSink` only broadcasts committed log
|
||||
// entries with live UI meaning; `UserInput` travels this lane so
|
||||
// the visible user line is ordered with `SegmentStart` rotation.
|
||||
entry = streams.log_entries.recv() => {
|
||||
match entry {
|
||||
Ok(entry) => {
|
||||
if let Some(event) = live_log_entry_event(entry) {
|
||||
if writer.write_event(&event).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
|
||||
// Slow client fell behind the broadcast buffer.
|
||||
// Drop the connection so the next reconnect
|
||||
// re-seeds the prefix via subscribe_with_snapshot.
|
||||
break;
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
|
||||
}
|
||||
}
|
||||
// Broadcast events → this client
|
||||
event = streams.events.recv() => {
|
||||
match event {
|
||||
Ok(event) => {
|
||||
let Ok(event) = handle.alerter.freshen_event(event) else {
|
||||
// The queued Alert was evicted; reconnect must rebuild
|
||||
// the bounded prefix rather than forward a stale anchor.
|
||||
break;
|
||||
};
|
||||
if writer.write_event(&event).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
// Client methods → handle or forward to controller
|
||||
method = reader.next_method() => {
|
||||
match method {
|
||||
Ok(Some(method)) => {
|
||||
if let Some(response) = dispatch_worker_protocol_method(&handle, method).await
|
||||
&& writer.write_event(&response).await.is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(None) => break,
|
||||
Err(e) if is_peer_disconnect_read_error(&e) => break,
|
||||
Err(e) => {
|
||||
if writer
|
||||
.write(&Event::Error {
|
||||
code: ErrorCode::InvalidRequest,
|
||||
message: format!("invalid method: {e}"),
|
||||
})
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn peer_disconnect_read_errors_are_connection_close() {
|
||||
for kind in [
|
||||
ErrorKind::ConnectionReset,
|
||||
ErrorKind::ConnectionAborted,
|
||||
ErrorKind::BrokenPipe,
|
||||
ErrorKind::UnexpectedEof,
|
||||
] {
|
||||
let error = io::Error::new(kind, "peer disconnected");
|
||||
assert!(
|
||||
is_peer_disconnect_read_error(&error),
|
||||
"{kind:?} should be treated as a normal peer disconnect"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_data_is_not_peer_disconnect() {
|
||||
let error = io::Error::new(ErrorKind::InvalidData, "malformed method");
|
||||
assert!(!is_peer_disconnect_read_error(&error));
|
||||
}
|
||||
}
|
||||
@@ -7,8 +7,6 @@ pub mod bootstrap;
|
||||
pub mod checkout;
|
||||
pub mod compact;
|
||||
pub mod controller;
|
||||
pub mod discovery;
|
||||
pub mod entrypoint;
|
||||
pub mod feature;
|
||||
mod file_operation;
|
||||
pub mod fs_view;
|
||||
@@ -19,9 +17,9 @@ pub mod job;
|
||||
mod metadata_lifecycle;
|
||||
pub mod model_client;
|
||||
mod paste_artifact_tool;
|
||||
pub mod profile_manifest;
|
||||
pub mod prompt;
|
||||
pub mod runtime;
|
||||
pub mod runtime_command;
|
||||
pub mod segment_log_sink;
|
||||
mod session_capture;
|
||||
mod session_history;
|
||||
@@ -37,6 +35,7 @@ pub mod wip;
|
||||
|
||||
mod internal_worker;
|
||||
mod interrupt_prep;
|
||||
mod peer;
|
||||
mod permission;
|
||||
mod worker;
|
||||
|
||||
@@ -45,10 +44,9 @@ pub use bootstrap::{
|
||||
WorkerBootstrapLayout, start_worker_controller,
|
||||
};
|
||||
pub use compact::token_counter::{EstimateSource, SplitPoint, TokenEstimate};
|
||||
pub use controller::{ShutdownReceiver, WorkerController, WorkerControllerTransport, WorkerHandle};
|
||||
pub use controller::{ShutdownReceiver, WorkerController, WorkerHandle};
|
||||
pub use hook::{Hook, HookEventKind, HookRegistryBuilder};
|
||||
pub use ipc::alerter::Alerter;
|
||||
pub use ipc::server::SocketServer;
|
||||
pub use manifest::{
|
||||
AuthRef, ModelManifest, SchemeKind, Scope, WorkerManifest, WorkerManifestConfig,
|
||||
WorkerMetaConfig,
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
//! Reciprocal peer metadata registration, independent of discovery and transport.
|
||||
//!
|
||||
//! Registering a peer does not start a Worker or grant filesystem, observation, or
|
||||
//! messaging authority. The two metadata writes are not a transaction: a failed
|
||||
//! reciprocal write triggers best-effort restoration of the source's prior peers.
|
||||
|
||||
use serde::Serialize;
|
||||
use session_store::{WorkerMetadataStore, WorkerStoreError, validate_worker_name};
|
||||
|
||||
#[derive(Debug, Serialize, PartialEq, Eq)]
|
||||
pub(crate) struct PeerRegistrationResult {
|
||||
pub source: String,
|
||||
pub peer: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub(crate) enum PeerRegistrationError {
|
||||
#[error("worker state missing for `{worker_name}`")]
|
||||
StateMissing { worker_name: String },
|
||||
#[error("worker `{worker_name}` cannot be registered as a peer of itself")]
|
||||
SelfPeer { worker_name: String },
|
||||
#[error("worker `{worker_name}` does not exist")]
|
||||
MissingWorker { worker_name: String },
|
||||
#[error("worker store error: {0}")]
|
||||
WorkerStore(#[from] WorkerStoreError),
|
||||
}
|
||||
|
||||
pub(crate) fn register_peer(
|
||||
store: &impl WorkerMetadataStore,
|
||||
source: &str,
|
||||
peer: &str,
|
||||
) -> Result<PeerRegistrationResult, PeerRegistrationError> {
|
||||
ensure_existing_peer(store, source, peer)?.ok_or_else(|| PeerRegistrationError::MissingWorker {
|
||||
worker_name: peer.to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
fn ensure_existing_peer(
|
||||
store: &impl WorkerMetadataStore,
|
||||
source: &str,
|
||||
peer: &str,
|
||||
) -> Result<Option<PeerRegistrationResult>, PeerRegistrationError> {
|
||||
validate_worker_name(peer)?;
|
||||
if peer == source {
|
||||
return Err(PeerRegistrationError::SelfPeer {
|
||||
worker_name: peer.to_string(),
|
||||
});
|
||||
}
|
||||
let prior_source_peers = store
|
||||
.read_by_name(source)?
|
||||
.ok_or_else(|| PeerRegistrationError::StateMissing {
|
||||
worker_name: source.to_string(),
|
||||
})?
|
||||
.peers;
|
||||
if store.read_by_name(peer)?.is_none() {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
store.add_peer(source, peer)?;
|
||||
if let Err(error) = store.add_peer(peer, source) {
|
||||
let _ = store.set_peers(source, prior_source_peers);
|
||||
return Err(PeerRegistrationError::WorkerStore(error));
|
||||
}
|
||||
Ok(Some(PeerRegistrationResult {
|
||||
source: source.to_string(),
|
||||
peer: peer.to_string(),
|
||||
}))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use session_store::{FsWorkerStore, WorkerMetadata};
|
||||
|
||||
#[test]
|
||||
fn register_peer_persists_reciprocal_metadata_without_duplicates() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let store = FsWorkerStore::new(root.path()).unwrap();
|
||||
store.write(&WorkerMetadata::new("source", None)).unwrap();
|
||||
store.write(&WorkerMetadata::new("target", None)).unwrap();
|
||||
for _ in 0..2 {
|
||||
let result = register_peer(&store, "source", "target").unwrap();
|
||||
assert_eq!(
|
||||
serde_json::to_value(result).unwrap(),
|
||||
serde_json::json!({
|
||||
"source": "source", "peer": "target",
|
||||
})
|
||||
);
|
||||
}
|
||||
// Reopen the real store rather than asserting only the write calls.
|
||||
let persisted = FsWorkerStore::new(root.path()).unwrap();
|
||||
let source = persisted.read_by_name("source").unwrap().unwrap();
|
||||
let target = persisted.read_by_name("target").unwrap().unwrap();
|
||||
assert_eq!(source.peers.len(), 1);
|
||||
assert_eq!(source.peers[0].worker_name, "target");
|
||||
assert_eq!(target.peers.len(), 1);
|
||||
assert_eq!(target.peers[0].worker_name, "source");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_self_or_missing_peer_leaves_metadata_unchanged() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let store = FsWorkerStore::new(root.path()).unwrap();
|
||||
store.write(&WorkerMetadata::new("source", None)).unwrap();
|
||||
assert!(matches!(
|
||||
register_peer(&store, "source", "source"),
|
||||
Err(PeerRegistrationError::SelfPeer { .. })
|
||||
));
|
||||
assert!(matches!(
|
||||
register_peer(&store, "source", "missing"),
|
||||
Err(PeerRegistrationError::MissingWorker { .. })
|
||||
));
|
||||
assert!(matches!(
|
||||
register_peer(&store, "source", "../escape"),
|
||||
Err(PeerRegistrationError::WorkerStore(_))
|
||||
));
|
||||
assert!(matches!(
|
||||
register_peer(&store, "missing-source", "source"),
|
||||
Err(PeerRegistrationError::StateMissing { .. })
|
||||
));
|
||||
assert!(
|
||||
store
|
||||
.read_by_name("source")
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.peers
|
||||
.is_empty()
|
||||
);
|
||||
assert_eq!(store.list_names().unwrap(), vec!["source"]);
|
||||
}
|
||||
|
||||
struct FailTargetPeerStore(FsWorkerStore);
|
||||
|
||||
impl WorkerMetadataStore for FailTargetPeerStore {
|
||||
fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> {
|
||||
if metadata.worker_name == "target" && !metadata.peers.is_empty() {
|
||||
return Err(WorkerStoreError::Io(std::io::Error::other(
|
||||
"target write failed",
|
||||
)));
|
||||
}
|
||||
self.0.write(metadata)
|
||||
}
|
||||
|
||||
fn read_by_name(&self, name: &str) -> Result<Option<WorkerMetadata>, WorkerStoreError> {
|
||||
self.0.read_by_name(name)
|
||||
}
|
||||
|
||||
fn list_names(&self) -> Result<Vec<String>, WorkerStoreError> {
|
||||
self.0.list_names()
|
||||
}
|
||||
|
||||
fn delete_by_name(&self, name: &str) -> Result<(), WorkerStoreError> {
|
||||
self.0.delete_by_name(name)
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reciprocal_write_failure_restores_exact_prior_source_peers() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let inner = FsWorkerStore::new(root.path()).unwrap();
|
||||
for name in ["source", "target", "existing"] {
|
||||
inner.write(&WorkerMetadata::new(name, None)).unwrap();
|
||||
}
|
||||
inner.add_peer("source", "existing").unwrap();
|
||||
for had_target in [false, true] {
|
||||
if had_target {
|
||||
inner.add_peer("source", "target").unwrap();
|
||||
}
|
||||
let before = inner.read_by_name("source").unwrap().unwrap().peers;
|
||||
let store = FailTargetPeerStore(inner.clone());
|
||||
assert!(matches!(
|
||||
register_peer(&store, "source", "target"),
|
||||
Err(PeerRegistrationError::WorkerStore(_))
|
||||
));
|
||||
assert_eq!(inner.read_by_name("source").unwrap().unwrap().peers, before);
|
||||
assert!(
|
||||
inner
|
||||
.read_by_name("target")
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.peers
|
||||
.is_empty()
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
//! Runtime manifest preparation from Backend-resolved profiles.
|
||||
//!
|
||||
//! This module has no process entrypoint or local profile discovery authority.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use crate::PromptCatalogSource;
|
||||
use manifest::{Permission, ScopeConfig, ScopeRule, WorkerManifest};
|
||||
|
||||
pub fn resolve_runtime_profile_manifest_from_manifest(
|
||||
mut manifest: WorkerManifest,
|
||||
workspace_root: &Path,
|
||||
worker_name: &str,
|
||||
) -> Result<(WorkerManifest, PromptCatalogSource), String> {
|
||||
if manifest.worker.name.is_empty() {
|
||||
manifest.worker.name = worker_name.to_string();
|
||||
}
|
||||
apply_profile_launch_policy(&mut manifest, workspace_root);
|
||||
// Do not run plugin discovery here: runtime-created Workers receive their
|
||||
// resolved manifest/profile archive from Backend authority, not by scanning
|
||||
// materialized workdir-local plugin stores.
|
||||
Ok((manifest, PromptCatalogSource::builtins_only()))
|
||||
}
|
||||
|
||||
pub fn resolve_runtime_profile_manifest_from_manifest_without_filesystem(
|
||||
mut manifest: WorkerManifest,
|
||||
_workspace_root: &Path,
|
||||
worker_name: &str,
|
||||
) -> Result<(WorkerManifest, PromptCatalogSource), String> {
|
||||
if manifest.worker.name.is_empty() {
|
||||
manifest.worker.name = worker_name.to_string();
|
||||
}
|
||||
manifest.scope = ScopeConfig::default();
|
||||
manifest.delegation_scope = ScopeConfig::default();
|
||||
// Same as the filesystem-capable runtime path: no local discovery.
|
||||
Ok((manifest, PromptCatalogSource::builtins_only()))
|
||||
}
|
||||
|
||||
fn read_rule(target: PathBuf) -> ScopeRule {
|
||||
ScopeRule {
|
||||
target,
|
||||
permission: Permission::Read,
|
||||
recursive: true,
|
||||
symlink_policy: Default::default(),
|
||||
}
|
||||
}
|
||||
|
||||
fn write_rule(target: PathBuf) -> ScopeRule {
|
||||
ScopeRule {
|
||||
target,
|
||||
permission: Permission::Write,
|
||||
recursive: true,
|
||||
symlink_policy: Default::default(),
|
||||
}
|
||||
}
|
||||
|
||||
fn workspace_scope(
|
||||
workspace_root: &Path,
|
||||
permission: Permission,
|
||||
deny_write: &[PathBuf],
|
||||
) -> ScopeConfig {
|
||||
let allow_rule = ScopeRule {
|
||||
target: workspace_root.to_path_buf(),
|
||||
permission,
|
||||
recursive: true,
|
||||
symlink_policy: Default::default(),
|
||||
};
|
||||
let deny = deny_write
|
||||
.iter()
|
||||
.cloned()
|
||||
.map(write_rule)
|
||||
.collect::<Vec<_>>();
|
||||
ScopeConfig {
|
||||
allow: vec![allow_rule],
|
||||
deny,
|
||||
}
|
||||
}
|
||||
|
||||
fn workspace_worktree_delegation(workspace_root: &Path) -> ScopeConfig {
|
||||
ScopeConfig {
|
||||
allow: vec![
|
||||
read_rule(workspace_root.to_path_buf()),
|
||||
write_rule(workspace_root.join(".worktree")),
|
||||
],
|
||||
deny: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn append_missing_rules(target: &mut Vec<ScopeRule>, defaults: Vec<ScopeRule>) {
|
||||
for rule in defaults {
|
||||
if !target.contains(&rule) {
|
||||
target.push(rule);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn apply_scope_launch_defaults(scope: &mut ScopeConfig, defaults: ScopeConfig) {
|
||||
// Profile resolution has already applied explicit profile/workspace override scope rules.
|
||||
// Launch policy contributes runtime defaults on top rather than replacing those grants.
|
||||
append_missing_rules(&mut scope.allow, defaults.allow);
|
||||
append_missing_rules(&mut scope.deny, defaults.deny);
|
||||
}
|
||||
|
||||
fn apply_profile_launch_policy(manifest: &mut WorkerManifest, workspace_root: &Path) {
|
||||
let worktree_root = workspace_root.join(".worktree");
|
||||
let default_scope = workspace_scope(
|
||||
workspace_root,
|
||||
Permission::Write,
|
||||
std::slice::from_ref(&worktree_root),
|
||||
);
|
||||
apply_scope_launch_defaults(&mut manifest.scope, default_scope);
|
||||
manifest.delegation_scope = workspace_worktree_delegation(workspace_root);
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn manifest() -> WorkerManifest {
|
||||
let mut config = manifest::WorkerManifestConfig::resolution_defaults();
|
||||
config.worker.name = Some("resolved".into());
|
||||
WorkerManifest::try_from(config).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runtime_launch_preserves_explicit_scope_without_repository_discovery() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
std::fs::create_dir(root.path().join(".yoi")).unwrap();
|
||||
std::fs::write(root.path().join(".yoi/manifest.toml"), "invalid TOML").unwrap();
|
||||
let mut original = manifest();
|
||||
let explicit = read_rule(root.path().join("reference"));
|
||||
original.scope.allow.push(explicit.clone());
|
||||
let (resolved, _) =
|
||||
resolve_runtime_profile_manifest_from_manifest(original, root.path(), "fallback")
|
||||
.unwrap();
|
||||
assert_eq!(resolved.worker.name, "resolved");
|
||||
assert!(resolved.scope.allow.contains(&explicit));
|
||||
assert!(
|
||||
resolved
|
||||
.scope
|
||||
.allow
|
||||
.contains(&write_rule(root.path().to_path_buf()))
|
||||
);
|
||||
assert!(
|
||||
resolved
|
||||
.scope
|
||||
.deny
|
||||
.contains(&write_rule(root.path().join(".worktree")))
|
||||
);
|
||||
assert_eq!(
|
||||
resolved.delegation_scope.allow,
|
||||
vec![
|
||||
read_rule(root.path().to_path_buf()),
|
||||
write_rule(root.path().join(".worktree")),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runtime_without_filesystem_clears_scope_and_delegation_grants() {
|
||||
let mut original = manifest();
|
||||
original.worker.name.clear();
|
||||
original
|
||||
.scope
|
||||
.allow
|
||||
.push(write_rule(PathBuf::from("/work")));
|
||||
original
|
||||
.delegation_scope
|
||||
.allow
|
||||
.push(write_rule(PathBuf::from("/work")));
|
||||
let (resolved, _) = resolve_runtime_profile_manifest_from_manifest_without_filesystem(
|
||||
original,
|
||||
Path::new("/unused"),
|
||||
"runtime-worker",
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(resolved.worker.name, "runtime-worker");
|
||||
assert!(resolved.scope.allow.is_empty());
|
||||
assert!(resolved.scope.deny.is_empty());
|
||||
assert!(resolved.delegation_scope.allow.is_empty());
|
||||
assert!(resolved.delegation_scope.deny.is_empty());
|
||||
}
|
||||
}
|
||||
@@ -1,17 +1,12 @@
|
||||
use std::io;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use manifest::{ScopeRule, paths};
|
||||
use manifest::ScopeRule;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::fs;
|
||||
|
||||
use crate::shared_state::WorkerSharedState;
|
||||
|
||||
/// One spawned-child record mirrored to `spawned_workers.json`.
|
||||
///
|
||||
/// Written by the spawner after registry changes so runtime-local tools
|
||||
/// have a materialised snapshot. Durable restore uses Worker state metadata;
|
||||
/// this file is not the authoritative source.
|
||||
/// Legacy process-child projection used only by persisted scope reclamation.
|
||||
/// Socket fields retain the stored format; they are never connected to.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct SpawnedWorkerRecord {
|
||||
/// Spawned Worker's identity.
|
||||
@@ -25,56 +20,33 @@ pub struct SpawnedWorkerRecord {
|
||||
pub callback_address: PathBuf,
|
||||
}
|
||||
|
||||
/// Manages the Worker's runtime directory on tmpfs.
|
||||
/// Artifact directory retained by an in-process Controller.
|
||||
///
|
||||
/// ```text
|
||||
/// <runtime_dir>/{worker_name}/
|
||||
/// ├── pid
|
||||
/// ├── status.json
|
||||
/// ├── manifest.toml
|
||||
/// ├── history.json
|
||||
/// └── sock (created by socket listener, not by RuntimeDir)
|
||||
/// ```
|
||||
///
|
||||
/// `<runtime_dir>` is resolved via [`manifest::paths::runtime_dir`].
|
||||
/// Files are written atomically (write tmp → rename).
|
||||
/// The directory is removed on drop.
|
||||
/// Worker liveness and configuration belong to the Host, not filesystem projections.
|
||||
/// Persistent run directories and pre-existing data are never removed on drop.
|
||||
pub struct RuntimeDir {
|
||||
path: PathBuf,
|
||||
write_legacy_snapshots: bool,
|
||||
preserve_on_drop: bool,
|
||||
socket_file_name: &'static str,
|
||||
}
|
||||
|
||||
impl RuntimeDir {
|
||||
/// Create the runtime directory and write the PID file.
|
||||
/// Create an ephemeral directory without PID, status, manifest or socket files.
|
||||
pub async fn create(base: &Path, worker_name: &str) -> Result<Self, io::Error> {
|
||||
let path = base.join(worker_name);
|
||||
fs::create_dir_all(&path).await?;
|
||||
|
||||
let pid = std::process::id().to_string();
|
||||
fs::write(path.join("pid"), pid.as_bytes()).await?;
|
||||
|
||||
Ok(Self {
|
||||
path,
|
||||
write_legacy_snapshots: true,
|
||||
preserve_on_drop: false,
|
||||
socket_file_name: "sock",
|
||||
})
|
||||
Self::create_transient(base, worker_name).await
|
||||
}
|
||||
|
||||
/// Create an ephemeral Runtime-owned artifact directory.
|
||||
///
|
||||
/// Runtime-managed Workers keep their status in the owning Runtime and do
|
||||
/// not materialize legacy pid/status/manifest projections.
|
||||
/// Preserve pre-existing directories rather than claiming their contents for cleanup.
|
||||
pub async fn create_transient(base: &Path, worker_name: &str) -> Result<Self, io::Error> {
|
||||
fs::create_dir_all(base).await?;
|
||||
let path = base.join(worker_name);
|
||||
fs::create_dir_all(&path).await?;
|
||||
let preserve_on_drop = match fs::create_dir(&path).await {
|
||||
Ok(()) => false,
|
||||
Err(error) if error.kind() == io::ErrorKind::AlreadyExists && path.is_dir() => true,
|
||||
Err(error) => return Err(error),
|
||||
};
|
||||
Ok(Self {
|
||||
path,
|
||||
write_legacy_snapshots: false,
|
||||
preserve_on_drop: false,
|
||||
socket_file_name: "sock",
|
||||
preserve_on_drop,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -100,39 +72,11 @@ impl RuntimeDir {
|
||||
std::fs::File::open(parent)?.sync_all()?;
|
||||
Ok(Self {
|
||||
path: path.to_path_buf(),
|
||||
write_legacy_snapshots: false,
|
||||
preserve_on_drop: true,
|
||||
socket_file_name: "worker.sock",
|
||||
})
|
||||
}
|
||||
|
||||
/// Create in the default base directory resolved via
|
||||
/// [`manifest::paths::runtime_dir`].
|
||||
pub async fn create_default(worker_name: &str) -> Result<Self, io::Error> {
|
||||
let base = default_base()?;
|
||||
Self::create(&base, worker_name).await
|
||||
}
|
||||
|
||||
/// Write status.json atomically.
|
||||
pub async fn write_status(&self, state: &WorkerSharedState) -> Result<(), io::Error> {
|
||||
if !self.write_legacy_snapshots {
|
||||
return Ok(());
|
||||
}
|
||||
let content = state.status_json();
|
||||
atomic_write(&self.path.join("status.json"), content.as_bytes()).await
|
||||
}
|
||||
|
||||
/// Write manifest.toml (typically once at startup).
|
||||
pub async fn write_manifest(&self, toml: &str) -> Result<(), io::Error> {
|
||||
if !self.write_legacy_snapshots {
|
||||
return Ok(());
|
||||
}
|
||||
atomic_write(&self.path.join("manifest.toml"), toml.as_bytes()).await
|
||||
}
|
||||
|
||||
/// Write `spawned_workers.json` atomically. The entries are the full
|
||||
/// set of spawned children known to this Worker — callers pass the
|
||||
/// replacement list, no incremental merge.
|
||||
/// Rewrite the legacy projection during persisted process-child reclamation.
|
||||
pub async fn write_spawned_workers(
|
||||
&self,
|
||||
records: &[SpawnedWorkerRecord],
|
||||
@@ -145,26 +89,10 @@ impl RuntimeDir {
|
||||
pub fn path(&self) -> &Path {
|
||||
&self.path
|
||||
}
|
||||
|
||||
/// Path where the Unix socket should be created. External callers
|
||||
/// that only know the worker name (e.g. the TUI's attach flow)
|
||||
/// predict the same path via [`manifest::paths::worker_socket_path`].
|
||||
pub fn socket_path(&self) -> PathBuf {
|
||||
self.path.join(self.socket_file_name)
|
||||
}
|
||||
|
||||
pub async fn close_socket(&self) -> Result<(), io::Error> {
|
||||
match fs::remove_file(self.socket_path()).await {
|
||||
Ok(()) => Ok(()),
|
||||
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
|
||||
Err(error) => Err(error),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for RuntimeDir {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::remove_file(self.socket_path());
|
||||
if !self.preserve_on_drop {
|
||||
let _ = std::fs::remove_dir_all(&self.path);
|
||||
}
|
||||
@@ -179,156 +107,72 @@ async fn atomic_write(target: &Path, content: &[u8]) -> Result<(), io::Error> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Resolve the default base directory for runtime data.
|
||||
///
|
||||
/// Thin wrapper over [`manifest::paths::runtime_dir`] that converts a
|
||||
/// missing-env situation into an `io::Error`.
|
||||
pub fn default_base() -> Result<PathBuf, io::Error> {
|
||||
paths::runtime_dir().ok_or_else(|| {
|
||||
io::Error::new(
|
||||
io::ErrorKind::NotFound,
|
||||
"could not resolve runtime directory (no YOI_HOME / \
|
||||
YOI_RUNTIME_DIR / XDG_RUNTIME_DIR / HOME)",
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::shared_state::WorkerSharedState;
|
||||
|
||||
fn test_state() -> WorkerSharedState {
|
||||
WorkerSharedState::new(
|
||||
"test-worker".into(),
|
||||
session_store::new_segment_id(),
|
||||
"[engine]\nname = \"test-worker\"".into(),
|
||||
protocol::Greeting {
|
||||
worker_name: "test-worker".into(),
|
||||
cwd: "/tmp".into(),
|
||||
provider: "anthropic".into(),
|
||||
model: "claude".into(),
|
||||
scope_summary: String::new(),
|
||||
tools: Vec::new(),
|
||||
context_window: 200_000,
|
||||
context_tokens: 0,
|
||||
reasoning: None,
|
||||
context_usage: None,
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn creates_directory_and_pid() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let rt = RuntimeDir::create(tmp.path(), "my-worker").await.unwrap();
|
||||
|
||||
assert!(rt.path().join("pid").exists());
|
||||
let pid = std::fs::read_to_string(rt.path().join("pid")).unwrap();
|
||||
assert_eq!(pid, std::process::id().to_string());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn write_status_creates_file() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let rt = RuntimeDir::create(tmp.path(), "my-worker").await.unwrap();
|
||||
let state = test_state();
|
||||
|
||||
rt.write_status(&state).await.unwrap();
|
||||
|
||||
let content = std::fs::read_to_string(rt.path().join("status.json")).unwrap();
|
||||
let parsed: serde_json::Value = serde_json::from_str(&content).unwrap();
|
||||
assert_eq!(parsed["state"], "idle");
|
||||
assert_eq!(parsed["worker_name"], "test-worker");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn write_status_reflects_changes() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let rt = RuntimeDir::create(tmp.path(), "my-worker").await.unwrap();
|
||||
let state = test_state();
|
||||
|
||||
state.transition(protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
|
||||
protocol::WorkerRunState::Running,
|
||||
)));
|
||||
rt.write_status(&state).await.unwrap();
|
||||
|
||||
let content = std::fs::read_to_string(rt.path().join("status.json")).unwrap();
|
||||
let parsed: serde_json::Value = serde_json::from_str(&content).unwrap();
|
||||
assert_eq!(parsed["state"], "running");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn write_manifest_creates_file() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let rt = RuntimeDir::create(tmp.path(), "my-worker").await.unwrap();
|
||||
|
||||
rt.write_manifest("[engine]\nname = \"test\"")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let content = std::fs::read_to_string(rt.path().join("manifest.toml")).unwrap();
|
||||
assert_eq!(content, "[engine]\nname = \"test\"");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn transient_directory_does_not_write_liveness_snapshots() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let rt = RuntimeDir::create_transient(tmp.path(), "runtime-worker")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
rt.write_status(&test_state()).await.unwrap();
|
||||
rt.write_manifest("[worker]\nname = \"runtime-worker\"")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(!rt.path().join("pid").exists());
|
||||
assert!(!rt.path().join("status.json").exists());
|
||||
assert!(!rt.path().join("manifest.toml").exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn write_spawned_workers_creates_file() {
|
||||
use manifest::{Permission, ScopeRule};
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let rt = RuntimeDir::create(tmp.path(), "my-worker").await.unwrap();
|
||||
|
||||
let records = vec![SpawnedWorkerRecord {
|
||||
worker_name: "child".into(),
|
||||
socket_path: "/run/yoi/child/sock".into(),
|
||||
scope_delegated: vec![ScopeRule {
|
||||
target: "/tmp/work".into(),
|
||||
permission: Permission::Write,
|
||||
recursive: true,
|
||||
symlink_policy: Default::default(),
|
||||
}],
|
||||
callback_address: "/run/yoi/my-worker/sock".into(),
|
||||
}];
|
||||
rt.write_spawned_workers(&records).await.unwrap();
|
||||
|
||||
let content = std::fs::read_to_string(rt.path().join("spawned_workers.json")).unwrap();
|
||||
let parsed: Vec<SpawnedWorkerRecord> = serde_json::from_str(&content).unwrap();
|
||||
assert_eq!(parsed.len(), 1);
|
||||
assert_eq!(parsed[0].worker_name, "child");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn socket_path() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let rt = RuntimeDir::create(tmp.path(), "my-worker").await.unwrap();
|
||||
assert_eq!(rt.socket_path(), rt.path().join("sock"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn drop_removes_directory() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let dir_path;
|
||||
async fn transient_controller_directory_has_no_liveness_projections_and_is_released() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let path;
|
||||
{
|
||||
let rt = RuntimeDir::create(tmp.path(), "my-worker").await.unwrap();
|
||||
dir_path = rt.path().to_owned();
|
||||
assert!(dir_path.exists());
|
||||
let runtime = RuntimeDir::create(root.path(), "worker").await.unwrap();
|
||||
path = runtime.path().to_owned();
|
||||
assert_eq!(std::fs::read_dir(&path).unwrap().count(), 0);
|
||||
}
|
||||
assert!(!dir_path.exists());
|
||||
assert!(!path.exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn preexisting_data_is_not_overwritten_or_removed_on_drop() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let path = root.path().join("worker");
|
||||
std::fs::create_dir(&path).unwrap();
|
||||
std::fs::write(path.join("pid"), "retained legacy data").unwrap();
|
||||
{
|
||||
let _runtime = RuntimeDir::create(root.path(), "worker").await.unwrap();
|
||||
}
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(path.join("pid")).unwrap(),
|
||||
"retained legacy data"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn persistent_run_preserves_logs_and_rejects_directory_reuse() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let path = root.path().join("runs/1");
|
||||
{
|
||||
let _runtime = RuntimeDir::create_worker_run(&path).await.unwrap();
|
||||
for name in ["artifacts", "spawned", "worker.out.log", "worker.err.log"] {
|
||||
assert!(path.join(name).exists(), "missing {name}");
|
||||
}
|
||||
}
|
||||
assert!(path.exists());
|
||||
let error = RuntimeDir::create_worker_run(&path).await.err().unwrap();
|
||||
assert_eq!(error.kind(), io::ErrorKind::AlreadyExists);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_reclamation_projection_retains_its_stored_format() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let runtime = RuntimeDir::create(root.path(), "worker").await.unwrap();
|
||||
let record: SpawnedWorkerRecord = serde_json::from_value(serde_json::json!({
|
||||
"worker_name": "legacy-child", "socket_path": "/old/child/sock",
|
||||
"scope_delegated": [], "callback_address": "/old/parent/sock",
|
||||
}))
|
||||
.unwrap();
|
||||
runtime.write_spawned_workers(&[record]).await.unwrap();
|
||||
let value: serde_json::Value = serde_json::from_slice(
|
||||
&std::fs::read(runtime.path().join("spawned_workers.json")).unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(value[0]["socket_path"], "/old/child/sock");
|
||||
assert_eq!(value[0]["callback_address"], "/old/parent/sock");
|
||||
runtime.write_spawned_workers(&[]).await.unwrap();
|
||||
assert_eq!(
|
||||
std::fs::read_to_string(runtime.path().join("spawned_workers.json")).unwrap(),
|
||||
"[]"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -72,7 +72,8 @@ pub struct Allocation {
|
||||
pub worker_name: String,
|
||||
/// Owning process. Checked with `kill(pid, 0)` for stale detection.
|
||||
pub pid: u32,
|
||||
/// Worker's Unix socket path.
|
||||
/// Legacy stored endpoint hint. Empty means the Host owns connectivity.
|
||||
/// Allocation admission and stale reclamation use identity/PID, never this path.
|
||||
pub socket: PathBuf,
|
||||
/// Allow rules granted to this Worker (write + read).
|
||||
pub scope_allow: Vec<ScopeRule>,
|
||||
@@ -338,6 +339,51 @@ mod tests {
|
||||
assert_eq!(guard.data().allocations[0].worker_name, "a");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_legacy_socket_roundtrips_without_path_resolution_or_cleanup() {
|
||||
let root = TempDir::new().unwrap();
|
||||
let path = root.path().join("workers.json");
|
||||
let retained = root.path().join("retained-data");
|
||||
fs::write(&retained, "do not touch").unwrap();
|
||||
let segment = sid();
|
||||
{
|
||||
let mut guard = open_empty(&path);
|
||||
register_worker(
|
||||
&mut guard,
|
||||
"in-process".into(),
|
||||
std::process::id(),
|
||||
PathBuf::new(),
|
||||
Vec::new(),
|
||||
segment,
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
let stored: serde_json::Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
|
||||
assert_eq!(stored["allocations"][0]["socket"], "");
|
||||
let mut guard = LockFileGuard::open(&path).unwrap();
|
||||
assert!(guard.data().allocations[0].socket.as_os_str().is_empty());
|
||||
super::super::reclaim_stale_with(&mut guard, |pid| {
|
||||
assert_eq!(pid, std::process::id());
|
||||
true
|
||||
});
|
||||
assert_eq!(
|
||||
guard.data().allocations.len(),
|
||||
1,
|
||||
"missing endpoint is not stale ownership"
|
||||
);
|
||||
assert!(guard.data().find_by_segment(segment).is_some());
|
||||
super::super::reclaim_stale_with(&mut guard, |_| false);
|
||||
drop(guard);
|
||||
assert!(
|
||||
LockFileGuard::open(&path)
|
||||
.unwrap()
|
||||
.data()
|
||||
.allocations
|
||||
.is_empty()
|
||||
);
|
||||
assert_eq!(fs::read_to_string(retained).unwrap(), "do not touch");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn atomic_save_reopens_complete_table_at_both_rename_boundaries() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
|
||||
@@ -1,165 +0,0 @@
|
||||
use std::ffi::OsString;
|
||||
use std::fmt;
|
||||
use std::io;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
const POD_RUNTIME_COMMAND_ENV: &str = "YOI_POD_RUNTIME_COMMAND";
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct WorkerRuntimeCommand {
|
||||
pub program: PathBuf,
|
||||
pub prefix_args: Vec<OsString>,
|
||||
}
|
||||
|
||||
impl WorkerRuntimeCommand {
|
||||
pub fn new(program: impl Into<PathBuf>, prefix_args: Vec<OsString>) -> Self {
|
||||
Self {
|
||||
program: program.into(),
|
||||
prefix_args,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn for_current_exe() -> io::Result<Self> {
|
||||
Ok(Self::for_executable(std::env::current_exe()?))
|
||||
}
|
||||
|
||||
pub fn for_executable(program: impl Into<PathBuf>) -> Self {
|
||||
Self::new(program, vec![OsString::from("worker")])
|
||||
}
|
||||
|
||||
/// Resolve the Worker runtime command used for subprocess launches.
|
||||
///
|
||||
/// The default launch path is always the current `yoi` executable plus
|
||||
/// the unified `worker` prefix argument. During development, a non-empty
|
||||
/// `YOI_POD_RUNTIME_COMMAND` value replaces only the executable path;
|
||||
/// the `worker` prefix is still added here and the env value is not parsed as a
|
||||
/// shell command.
|
||||
pub fn resolve() -> io::Result<Self> {
|
||||
Self::resolve_from_env_value(
|
||||
std::env::var_os(POD_RUNTIME_COMMAND_ENV),
|
||||
std::env::current_exe,
|
||||
)
|
||||
}
|
||||
|
||||
fn resolve_from_env_value<F>(
|
||||
override_program: Option<OsString>,
|
||||
current_exe: F,
|
||||
) -> io::Result<Self>
|
||||
where
|
||||
F: FnOnce() -> io::Result<PathBuf>,
|
||||
{
|
||||
if let Some(program) = override_program.filter(|program| !program.as_os_str().is_empty()) {
|
||||
return Ok(Self::for_executable(program));
|
||||
}
|
||||
|
||||
Ok(Self::for_executable(current_exe()?))
|
||||
}
|
||||
|
||||
pub fn program(&self) -> &Path {
|
||||
&self.program
|
||||
}
|
||||
|
||||
pub fn prefix_args(&self) -> &[OsString] {
|
||||
&self.prefix_args
|
||||
}
|
||||
|
||||
pub fn argv_with<I, S>(&self, args: I) -> Vec<OsString>
|
||||
where
|
||||
I: IntoIterator<Item = S>,
|
||||
S: Into<OsString>,
|
||||
{
|
||||
let mut argv = self.prefix_args.clone();
|
||||
argv.extend(args.into_iter().map(Into::into));
|
||||
argv
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for WorkerRuntimeCommand {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}", self.program.display())?;
|
||||
for arg in &self.prefix_args {
|
||||
write!(f, " {}", arg.to_string_lossy())?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn yoi_binary_defaults_to_worker_prefix() {
|
||||
let command = WorkerRuntimeCommand::for_executable("/opt/yoi/bin/yoi");
|
||||
|
||||
assert_eq!(command.program(), Path::new("/opt/yoi/bin/yoi"));
|
||||
assert_eq!(command.prefix_args(), [OsString::from("worker")]);
|
||||
assert_eq!(
|
||||
command.argv_with(["--worker", "agent"]),
|
||||
vec!["worker", "--worker", "agent"]
|
||||
.into_iter()
|
||||
.map(OsString::from)
|
||||
.collect::<Vec<_>>()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn any_runtime_executable_gets_worker_prefix() {
|
||||
let command = WorkerRuntimeCommand::for_executable("/opt/yoi/bin/custom-runtime");
|
||||
|
||||
assert_eq!(command.program(), Path::new("/opt/yoi/bin/custom-runtime"));
|
||||
assert_eq!(command.prefix_args(), [OsString::from("worker")]);
|
||||
assert_eq!(
|
||||
command.argv_with(["--worker", "agent"]),
|
||||
vec!["worker", "--worker", "agent"]
|
||||
.into_iter()
|
||||
.map(OsString::from)
|
||||
.collect::<Vec<_>>()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_uses_current_exe_when_override_is_unset() {
|
||||
let command = WorkerRuntimeCommand::resolve_from_env_value(None, || {
|
||||
Ok(PathBuf::from("/opt/yoi/bin/yoi"))
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
command,
|
||||
WorkerRuntimeCommand::for_executable("/opt/yoi/bin/yoi")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_uses_current_exe_when_override_is_empty() {
|
||||
let command = WorkerRuntimeCommand::resolve_from_env_value(Some(OsString::new()), || {
|
||||
Ok(PathBuf::from("/opt/yoi/bin/yoi"))
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
command,
|
||||
WorkerRuntimeCommand::for_executable("/opt/yoi/bin/yoi")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_override_replaces_only_program_and_keeps_worker_prefix() {
|
||||
let command = WorkerRuntimeCommand::resolve_from_env_value(
|
||||
Some(OsString::from("/tmp/rebuilt yoi")),
|
||||
|| panic!("override must not inspect current_exe"),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(command.program(), Path::new("/tmp/rebuilt yoi"));
|
||||
assert_eq!(command.prefix_args(), [OsString::from("worker")]);
|
||||
assert_eq!(
|
||||
command.argv_with(["--worker", "agent"]),
|
||||
vec!["worker", "--worker", "agent"]
|
||||
.into_iter()
|
||||
.map(OsString::from)
|
||||
.collect::<Vec<_>>()
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1,181 +0,0 @@
|
||||
//! Socket communication retained for the legacy top-level Worker callback protocol.
|
||||
//!
|
||||
//! Direct Internal SubWorker lifecycle is exposed through `worker.control` and the
|
||||
//! canonical Worker tools, not a second SubWorker-specific tool family.
|
||||
|
||||
use std::path::Path;
|
||||
use std::time::Duration;
|
||||
|
||||
use protocol::stream::{JsonLineReader, JsonLineWriter};
|
||||
use protocol::{Event, Method};
|
||||
use tokio::net::UnixStream;
|
||||
|
||||
/// Timeout applied to each socket-level operation — connect, write,
|
||||
/// read. Kept short so a stuck child doesn't block the spawner's turn.
|
||||
const SOCKET_OP_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
|
||||
/// Connect with a timeout, drain the server's connect-time snapshot,
|
||||
/// write one `Method` line, flush, and close.
|
||||
///
|
||||
/// The Worker socket protocol sends replayed alerts and an initial
|
||||
/// `Event::Snapshot` before it starts reading client methods. Send-only
|
||||
/// callers must consume that prefix; otherwise a large snapshot can block
|
||||
/// the server's writer before it reaches the method-read branch. Any
|
||||
/// socket error maps to an `io::Error`; the caller decides whether to
|
||||
/// surface it to the LLM or treat it as "worker stopped".
|
||||
pub(crate) async fn connect_and_send(socket: &Path, method: &Method) -> std::io::Result<()> {
|
||||
let stream = tokio::time::timeout(SOCKET_OP_TIMEOUT, UnixStream::connect(socket))
|
||||
.await
|
||||
.map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "connect timed out"))??;
|
||||
let (r, w) = stream.into_split();
|
||||
let mut reader = JsonLineReader::new(r);
|
||||
let mut writer = JsonLineWriter::new(w);
|
||||
|
||||
drain_initial_snapshot(&mut reader).await?;
|
||||
|
||||
tokio::time::timeout(SOCKET_OP_TIMEOUT, writer.write(method))
|
||||
.await
|
||||
.map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "write timed out"))??;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn drain_initial_snapshot<R>(reader: &mut JsonLineReader<R>) -> std::io::Result<()>
|
||||
where
|
||||
R: tokio::io::AsyncBufRead + Unpin,
|
||||
{
|
||||
loop {
|
||||
let event = tokio::time::timeout(SOCKET_OP_TIMEOUT, reader.next::<Event>())
|
||||
.await
|
||||
.map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "read timed out"))??;
|
||||
match event {
|
||||
Some(Event::Snapshot { .. }) => return Ok(()),
|
||||
Some(_) => continue,
|
||||
None => {
|
||||
return Err(std::io::Error::new(
|
||||
std::io::ErrorKind::UnexpectedEof,
|
||||
"worker closed connection before Snapshot event",
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
use protocol::{Alert, AlertLevel, AlertSource, Greeting, WorkerEvent, WorkerStatus};
|
||||
use tempfile::TempDir;
|
||||
use tokio::net::UnixListener;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
fn snapshot(entries: Vec<serde_json::Value>) -> Event {
|
||||
Event::Snapshot {
|
||||
session: protocol::SessionSnapshot {
|
||||
pending_submissions: protocol::PendingSubmissionsSnapshot::default(),
|
||||
entries: entries
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.map(|(index, value)| protocol::SessionSnapshotEntry {
|
||||
use_policy: None,
|
||||
entry_id: format!("test-{index}"),
|
||||
timestamp: index as u64,
|
||||
provenance: protocol::SessionEntryProvenance::LegacyUnknown,
|
||||
derived_from: Vec::new(),
|
||||
data: protocol::SessionSnapshotEntryData::RunError {
|
||||
message: value.to_string(),
|
||||
failure: None,
|
||||
},
|
||||
})
|
||||
.collect(),
|
||||
},
|
||||
greeting: Greeting {
|
||||
worker_name: "server".into(),
|
||||
cwd: "/tmp".into(),
|
||||
provider: "test".into(),
|
||||
model: "test".into(),
|
||||
scope_summary: String::new(),
|
||||
tools: Vec::new(),
|
||||
context_window: 200_000,
|
||||
context_tokens: 0,
|
||||
reasoning: None,
|
||||
context_usage: None,
|
||||
},
|
||||
state: WorkerStatus::Idle.into(),
|
||||
in_flight: Default::default(),
|
||||
internal_workers: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn serve_initial_events_then_method(
|
||||
listener: UnixListener,
|
||||
events: Vec<Event>,
|
||||
) -> JoinHandle<Option<Method>> {
|
||||
tokio::spawn(async move {
|
||||
let (stream, _) = listener.accept().await.ok()?;
|
||||
let (r, w) = stream.into_split();
|
||||
let mut reader = JsonLineReader::new(r);
|
||||
let mut writer = JsonLineWriter::new(w);
|
||||
for event in events {
|
||||
writer.write(&event).await.ok()?;
|
||||
}
|
||||
reader.next::<Method>().await.ok().flatten()
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn connect_and_send_drains_initial_alert_and_snapshot_before_method() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let socket = tmp.path().join("worker.sock");
|
||||
let listener = UnixListener::bind(&socket).unwrap();
|
||||
let received = serve_initial_events_then_method(
|
||||
listener,
|
||||
vec![
|
||||
Event::Alert(Alert {
|
||||
id: None,
|
||||
position: None,
|
||||
level: AlertLevel::Warn,
|
||||
source: AlertSource::Worker,
|
||||
message: "replayed alert".into(),
|
||||
timestamp_ms: 0,
|
||||
}),
|
||||
snapshot(Vec::new()),
|
||||
],
|
||||
);
|
||||
|
||||
let method = Method::Shutdown {
|
||||
command: protocol::WorkerCommandEnvelope::new(1),
|
||||
};
|
||||
connect_and_send(&socket, &method).await.unwrap();
|
||||
|
||||
let method = received.await.unwrap().expect("expected method");
|
||||
assert!(matches!(method, Method::Shutdown { .. }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn connect_and_send_delivers_method_after_large_initial_snapshot() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let socket = tmp.path().join("worker.sock");
|
||||
let listener = UnixListener::bind(&socket).unwrap();
|
||||
let large_payload = "x".repeat(2 * 1024 * 1024);
|
||||
let received = serve_initial_events_then_method(
|
||||
listener,
|
||||
vec![snapshot(vec![
|
||||
serde_json::json!({ "payload": large_payload }),
|
||||
])],
|
||||
);
|
||||
let expected = Method::WorkerEvent(WorkerEvent::TurnEnded {
|
||||
worker_name: "child".into(),
|
||||
});
|
||||
|
||||
connect_and_send(&socket, &expected).await.unwrap();
|
||||
|
||||
let method = received.await.unwrap().expect("expected method");
|
||||
match method {
|
||||
Method::WorkerEvent(WorkerEvent::TurnEnded { worker_name }) => {
|
||||
assert_eq!(worker_name, "child")
|
||||
}
|
||||
other => panic!("expected TurnEnded WorkerEvent, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,3 @@
|
||||
pub mod comm_tools;
|
||||
mod published;
|
||||
pub mod registry;
|
||||
pub mod tool;
|
||||
|
||||
@@ -678,8 +678,8 @@ pub(crate) fn test_connection(client: Arc<dyn WorkspaceClient>) -> SubjektivHost
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{
|
||||
Method, WorkerBootstrap, WorkerBootstrapLayout, WorkerControllerTransport,
|
||||
WorkerFilesystemAuthority, WorkerManifest, WorkerWorkspaceContext,
|
||||
Method, WorkerBootstrap, WorkerBootstrapLayout, WorkerFilesystemAuthority, WorkerManifest,
|
||||
WorkerWorkspaceContext,
|
||||
};
|
||||
use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
|
||||
use agen::llm_client::{ClientError, LlmClient, Request};
|
||||
@@ -944,7 +944,6 @@ enabled = true
|
||||
WorkerBootstrapLayout::Direct {
|
||||
runtime_base: temp.path().join("runtime"),
|
||||
},
|
||||
WorkerControllerTransport::InProcess,
|
||||
)
|
||||
.with_model_client(TaskScriptClient::default())
|
||||
.with_subjektiv_host(host)
|
||||
@@ -1137,7 +1136,6 @@ enabled = true
|
||||
WorkerBootstrapLayout::Direct {
|
||||
runtime_base: temp.path().join("runtime"),
|
||||
},
|
||||
WorkerControllerTransport::InProcess,
|
||||
)
|
||||
.with_model_client(ScriptedClient(tx))
|
||||
.with_subjektiv_host(host.clone())
|
||||
@@ -1318,7 +1316,6 @@ enabled = true
|
||||
WorkerBootstrapLayout::Direct {
|
||||
runtime_base: temp.path().join("runtime"),
|
||||
},
|
||||
WorkerControllerTransport::InProcess,
|
||||
)
|
||||
.with_model_client(ScriptedClient(tx))
|
||||
.with_subjektiv_host(host.clone())
|
||||
@@ -1377,7 +1374,6 @@ enabled = true
|
||||
WorkerBootstrapLayout::Direct {
|
||||
runtime_base: temp.path().join("runtime"),
|
||||
},
|
||||
WorkerControllerTransport::InProcess,
|
||||
)
|
||||
.with_model_client(ScriptedClient(tx));
|
||||
let bootstrap = if attach {
|
||||
|
||||
+7
-169
@@ -531,7 +531,6 @@ use crate::prompt::agents_md::read_agents_md;
|
||||
use crate::prompt::catalog::{CatalogError, PromptCatalog, WorkspacePromptProjection};
|
||||
use crate::prompt::source::PromptCatalogSource;
|
||||
use crate::prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTemplate};
|
||||
use crate::runtime::dir;
|
||||
use crate::runtime::worker_allocation::{
|
||||
self, ScopeAllocationGuard, ScopeLockError, SegmentActivationGuard,
|
||||
};
|
||||
@@ -3456,18 +3455,13 @@ pub struct Worker<C: LlmClient, St: Store> {
|
||||
/// authority before the outer allocation guard reacquires the same lock.
|
||||
failed_activation_authority: Mutex<Option<SegmentActivationGuard>>,
|
||||
/// Scope allocation in the machine-wide lock file. `Some` for
|
||||
/// Workers built via `from_manifest` / `from_manifest_spawned` /
|
||||
/// Workers built via `from_manifest` /
|
||||
/// `restore_from_manifest` (production paths); `None` for the
|
||||
/// low-level `Worker::new` constructor used in tests, which bypasses
|
||||
/// the registry. Kept purely for its `Drop` impl, which releases
|
||||
/// the allocation when the Worker is dropped.
|
||||
#[allow(dead_code)]
|
||||
scope_allocation: Option<ScopeAllocationGuard>,
|
||||
/// Socket path of the spawning Worker. `Some` only for Workers built via
|
||||
/// `from_manifest_spawned`. Consumed by the controller to fire
|
||||
/// `Method::WorkerEvent` reports upward (turn end, error, shutdown,
|
||||
/// scope sub-delegation).
|
||||
callback_socket: Option<PathBuf>,
|
||||
/// Transient launch role for Ticket role sessions. This is process-local
|
||||
/// runtime identity used by controller policy; it is not model-visible and
|
||||
/// is not persisted into Ticket claim/session records.
|
||||
@@ -3751,7 +3745,6 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
|
||||
pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())),
|
||||
failed_activation_authority: Mutex::new(None),
|
||||
scope_allocation: None,
|
||||
callback_socket: None,
|
||||
runtime_ticket_role: None,
|
||||
prompts,
|
||||
inject_resident_summary: true,
|
||||
@@ -3916,8 +3909,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
|
||||
self.runtime_ticket_role.as_deref()
|
||||
}
|
||||
|
||||
/// Set the process-local Ticket role marker. Intended for entrypoint
|
||||
/// launch metadata, not for model-visible prompts or durable claims.
|
||||
/// Set the process-local Ticket role marker supplied by the Host at launch,
|
||||
/// not by model-visible prompts or durable claims.
|
||||
pub fn set_runtime_ticket_role(&mut self, role: Option<String>) {
|
||||
self.runtime_ticket_role = role;
|
||||
}
|
||||
@@ -4788,15 +4781,6 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
|
||||
self.pending_notifies.clone()
|
||||
}
|
||||
|
||||
/// Parent callback socket set by `from_manifest_spawned`.
|
||||
///
|
||||
/// Consumed by the Controller to fire `Method::WorkerEvent` upward on
|
||||
/// lifecycle transitions. `None` for top-level Workers, in which case
|
||||
/// the Controller silently skips the send.
|
||||
pub fn callback_socket(&self) -> Option<&PathBuf> {
|
||||
self.callback_socket.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn install_compaction_budget(
|
||||
&mut self,
|
||||
budget: Option<crate::compact::worker::CompactWorkerInterceptor>,
|
||||
@@ -8304,14 +8288,10 @@ where
|
||||
// Register this Worker in the machine-wide worker-allocation
|
||||
// before building anything else, so a spawn that conflicts on
|
||||
// scope fails fast.
|
||||
let socket_path = dir::default_base()
|
||||
.map_err(ScopeLockError::from)?
|
||||
.join(&manifest.worker.name)
|
||||
.join("sock");
|
||||
let scope_allocation = worker_allocation::install_top_level(
|
||||
manifest.worker.name.clone(),
|
||||
std::process::id(),
|
||||
socket_path,
|
||||
PathBuf::new(), // Reserved legacy allocation field; no Worker endpoint exists.
|
||||
common.scope.allow_rules(),
|
||||
segment_id,
|
||||
)?;
|
||||
@@ -8372,7 +8352,6 @@ where
|
||||
pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())),
|
||||
failed_activation_authority: Mutex::new(None),
|
||||
scope_allocation: Some(scope_allocation),
|
||||
callback_socket: None,
|
||||
runtime_ticket_role: None,
|
||||
prompts: common.prompts,
|
||||
inject_resident_summary: true,
|
||||
@@ -8773,7 +8752,6 @@ where
|
||||
pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())),
|
||||
failed_activation_authority: Mutex::new(None),
|
||||
scope_allocation: None,
|
||||
callback_socket: None,
|
||||
runtime_ticket_role: None,
|
||||
prompts: common.prompts,
|
||||
inject_resident_summary: true,
|
||||
@@ -8796,141 +8774,6 @@ where
|
||||
Ok(worker)
|
||||
}
|
||||
|
||||
/// Build a Worker spawned by another Worker (sibling process).
|
||||
///
|
||||
/// Behaves like [`Worker::from_manifest`] but claims the scope
|
||||
/// allocation that the spawner pre-registered via
|
||||
/// [`worker_allocation::delegate_scope`], rather than installing a new
|
||||
/// top-level entry. `callback_socket` carries the spawner's
|
||||
/// Unix-socket path so the spawned Worker can send `Method::Notify`
|
||||
/// back to the spawner.
|
||||
pub async fn from_manifest_spawned(
|
||||
manifest: WorkerManifest,
|
||||
store: St,
|
||||
loader: PromptCatalogSource,
|
||||
callback_socket: PathBuf,
|
||||
) -> Result<Self, WorkerError> {
|
||||
let cwd = current_cwd()?;
|
||||
let authority = WorkerFilesystemAuthority::local(cwd.clone(), cwd.clone());
|
||||
let workspace_context = WorkerWorkspaceContext::local_filesystem(None);
|
||||
Self::from_manifest_spawned_with_context(
|
||||
manifest,
|
||||
store,
|
||||
loader,
|
||||
callback_socket,
|
||||
workspace_context,
|
||||
authority,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn from_manifest_spawned_with_context(
|
||||
manifest: WorkerManifest,
|
||||
store: St,
|
||||
loader: PromptCatalogSource,
|
||||
callback_socket: PathBuf,
|
||||
workspace_context: WorkerWorkspaceContext,
|
||||
filesystem_authority: WorkerFilesystemAuthority,
|
||||
) -> Result<Self, WorkerError> {
|
||||
let common = prepare_worker_common_with_context(
|
||||
&manifest,
|
||||
&loader,
|
||||
/* parse_template */ true,
|
||||
workspace_context,
|
||||
filesystem_authority,
|
||||
manifest.scope.clone(),
|
||||
)?;
|
||||
|
||||
// A spawned child starts its own conversation, so it mints a
|
||||
// fresh Session rather than joining the spawner's.
|
||||
let session_id = session_store::new_session_id();
|
||||
let segment_id = session_store::new_segment_id();
|
||||
let scope_allocation = worker_allocation::adopt_allocation(
|
||||
manifest.worker.name.clone(),
|
||||
std::process::id(),
|
||||
segment_id,
|
||||
)?;
|
||||
|
||||
let mut worker =
|
||||
Engine::<Box<dyn LlmClient>, Mutable, SessionHistoryMetadata>::new_annotated(
|
||||
common.client,
|
||||
);
|
||||
apply_worker_manifest(&mut worker, &manifest.engine);
|
||||
worker.set_cache_key(Some(segment_id.to_string()));
|
||||
let worker_metadata_writer = Some(worker_metadata_writer_for_store(&store));
|
||||
let worker_metadata_segment_cas = Some(worker_metadata_segment_cas_for_store(&store));
|
||||
let scope = SharedScope::new(common.scope);
|
||||
let workdir_sessions =
|
||||
workdir_sessions_from_authority(&common.filesystem_authority, &scope);
|
||||
|
||||
let mut worker = Self {
|
||||
backend_job: std::sync::OnceLock::new(),
|
||||
manifest,
|
||||
engine: Some(worker),
|
||||
session: WorkerSession::new(session_id, Vec::new()),
|
||||
last_run_interrupted: false,
|
||||
store,
|
||||
worker_metadata_writer,
|
||||
worker_metadata_segment_cas,
|
||||
segment_state: SegmentState::new(session_id, segment_id, 0),
|
||||
filesystem_authority: common.filesystem_authority,
|
||||
workdir_sessions,
|
||||
workspace_context: common.workspace_context,
|
||||
flow_runtime_state: Arc::new(Mutex::new(None)),
|
||||
flow_feature_enabled: false,
|
||||
scope,
|
||||
hook_builder: HookRegistryBuilder::new(),
|
||||
hook_registry: None,
|
||||
feature_background_tasks: FeatureBackgroundTaskRegistry::default(),
|
||||
feature_invocations: FeatureInvocationRegistry::default(),
|
||||
manifest_lifecycle_features_enabled: true,
|
||||
interceptor_installed: false,
|
||||
compaction_service_budget: None,
|
||||
compact_state: None,
|
||||
usage_tracker: Arc::new(UsageTracker::new()),
|
||||
metrics_tracker: Arc::new(crate::compact::metrics_tracker::MetricsTracker::new()),
|
||||
usage_history: Arc::new(Mutex::new(Vec::new())),
|
||||
tracker: None,
|
||||
task_feature: TaskFeature::new(),
|
||||
worker_observation_provider: None,
|
||||
system_prompt_template: common.system_prompt_template,
|
||||
feature_instructions: common.feature_instructions,
|
||||
alerter: None,
|
||||
working_event_tx: None,
|
||||
internal_worker_registry: None,
|
||||
pending_compaction_cleanup: Mutex::new(None),
|
||||
in_flight: None,
|
||||
ai_activity_counter: Arc::new(AtomicUsize::new(0)),
|
||||
pending_notifies: NotifyBuffer::new(),
|
||||
notification_coalesce_delay: DEFAULT_NOTIFICATION_COALESCE_DELAY,
|
||||
pending_attachments: Arc::new(Mutex::new(Vec::<SystemItem>::new())),
|
||||
pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())),
|
||||
failed_activation_authority: Mutex::new(None),
|
||||
scope_allocation: Some(scope_allocation),
|
||||
callback_socket: Some(callback_socket),
|
||||
runtime_ticket_role: None,
|
||||
prompts: common.prompts,
|
||||
inject_resident_summary: true,
|
||||
feature_resident_summary_source: None,
|
||||
session_context_source: None,
|
||||
prepared_initial_seed: None,
|
||||
prepared_initial_prompt_completion: None,
|
||||
subjektiv_host: None,
|
||||
feature_installation: crate::feature::FeatureInstallation::Activate,
|
||||
restore_feature_lifecycle_pending: false,
|
||||
feature_system_prompt_override: None,
|
||||
user_segments: Vec::new(),
|
||||
sink: SegmentLogSink::new(),
|
||||
history_persistence_wired: false,
|
||||
log_writer: None,
|
||||
notification_committer: None,
|
||||
};
|
||||
worker.apply_permissions_from_manifest();
|
||||
worker.apply_prune_from_manifest();
|
||||
worker.write_worker_metadata_pending()?;
|
||||
Ok(worker)
|
||||
}
|
||||
|
||||
/// Restore a Worker by resolving its name-keyed metadata to an active
|
||||
/// `(SessionId, SegmentId)` and then using the normal session-log restore
|
||||
/// path. The metadata stores only the active pointer; lineage and origin
|
||||
@@ -9215,14 +9058,10 @@ where
|
||||
// the lookup + install inside a single `LockFileGuard` is what
|
||||
// makes "no two live Workers write to the same session log"
|
||||
// actually structural rather than a hopeful pre-check.
|
||||
let socket_path = dir::default_base()
|
||||
.map_err(ScopeLockError::from)?
|
||||
.join(&manifest.worker.name)
|
||||
.join("sock");
|
||||
let scope_allocation = worker_allocation::install_top_level_with_deny(
|
||||
manifest.worker.name.clone(),
|
||||
std::process::id(),
|
||||
socket_path,
|
||||
PathBuf::new(), // Reserved legacy allocation field; no Worker endpoint exists.
|
||||
common.scope.allow_rules(),
|
||||
common.scope.deny_rules(),
|
||||
segment_id,
|
||||
@@ -9320,7 +9159,6 @@ where
|
||||
pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())),
|
||||
failed_activation_authority: Mutex::new(None),
|
||||
scope_allocation: Some(scope_allocation),
|
||||
callback_socket: None,
|
||||
runtime_ticket_role: None,
|
||||
prompts: common.prompts,
|
||||
inject_resident_summary: true,
|
||||
@@ -10416,6 +10254,8 @@ struct WorkerCommon {
|
||||
feature_instructions: Vec<FeatureInstructionDeclaration>,
|
||||
}
|
||||
|
||||
/// Migration-only safety probe for retained process-child metadata.
|
||||
/// New Internal SubWorkers never publish socket metadata or use this path.
|
||||
async fn restored_child_reachable(child: &WorkerSpawnedChild) -> bool {
|
||||
tokio::time::timeout(
|
||||
RESTORE_RECONCILIATION_REACHABILITY_TIMEOUT,
|
||||
@@ -17962,7 +17802,6 @@ permission = "write"
|
||||
WorkerWorkspaceContext::no_workspace(),
|
||||
WorkerFilesystemAuthority::None,
|
||||
layout,
|
||||
crate::WorkerControllerTransport::InProcess,
|
||||
)
|
||||
.with_model_client(NoopClient)
|
||||
.with_subjektiv_host(host)
|
||||
@@ -18282,7 +18121,6 @@ permission = "write"
|
||||
.join("runtime/runs")
|
||||
.join(uuid::Uuid::now_v7().to_string()),
|
||||
},
|
||||
crate::WorkerControllerTransport::InProcess,
|
||||
)
|
||||
.with_model_client(NoopClient)
|
||||
}
|
||||
|
||||
@@ -1738,42 +1738,24 @@ async fn snapshot_includes_user_input_for_in_flight_turn() {
|
||||
// The Running event is the in-flight visibility fence: the committed
|
||||
// annotated input must already be available to an immediately attaching
|
||||
// subscriber rather than racing behind this status transition.
|
||||
let stream = tokio::net::UnixStream::connect(handle.runtime_dir.socket_path())
|
||||
.await
|
||||
.unwrap();
|
||||
let (reader, _writer) = stream.into_split();
|
||||
let mut reader = protocol::stream::JsonLineReader::new(reader);
|
||||
|
||||
loop {
|
||||
let event = reader.next::<Event>().await.unwrap().unwrap();
|
||||
match event {
|
||||
Event::Snapshot { session, .. } => {
|
||||
let found = session.entries.iter().any(|entry| match &entry.data {
|
||||
protocol::SessionSnapshotEntryData::UserInput { segments } => {
|
||||
protocol::Segment::flatten_to_text(segments) == "hello in-flight"
|
||||
}
|
||||
protocol::SessionSnapshotEntryData::Message {
|
||||
role: protocol::SessionMessageRole::User,
|
||||
content,
|
||||
} => content.iter().any(|part| {
|
||||
matches!(
|
||||
part,
|
||||
protocol::SessionContentPart::Text { text }
|
||||
if text == "hello in-flight"
|
||||
)
|
||||
}),
|
||||
_ => false,
|
||||
});
|
||||
assert!(
|
||||
found,
|
||||
"snapshot must carry the in-flight UserInput entry: {session:?}"
|
||||
);
|
||||
return;
|
||||
let Event::Snapshot { session, .. } = handle.snapshot_event() else {
|
||||
panic!("expected snapshot");
|
||||
};
|
||||
assert!(
|
||||
session.entries.iter().any(|entry| match &entry.data {
|
||||
protocol::SessionSnapshotEntryData::UserInput { segments } => {
|
||||
protocol::Segment::flatten_to_text(segments) == "hello in-flight"
|
||||
}
|
||||
Event::Alert(_) => continue,
|
||||
other => panic!("expected Snapshot first, got {other:?}"),
|
||||
}
|
||||
}
|
||||
protocol::SessionSnapshotEntryData::Message {
|
||||
role: protocol::SessionMessageRole::User,
|
||||
content,
|
||||
} => content.iter().any(|part| matches!(part,
|
||||
protocol::SessionContentPart::Text { text } if text == "hello in-flight"
|
||||
)),
|
||||
_ => false,
|
||||
}),
|
||||
"snapshot must carry the in-flight UserInput entry: {session:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -1855,24 +1837,10 @@ async fn attach_snapshot_includes_current_status() {
|
||||
.unwrap();
|
||||
wait_for_status(&handle, WorkerStatus::Running).await;
|
||||
|
||||
let stream = tokio::net::UnixStream::connect(handle.runtime_dir.socket_path())
|
||||
.await
|
||||
.unwrap();
|
||||
let (reader, _writer) = stream.into_split();
|
||||
let mut reader = protocol::stream::JsonLineReader::new(reader);
|
||||
|
||||
// First event after connect is the snapshot — it carries the current status.
|
||||
loop {
|
||||
let event = reader.next::<Event>().await.unwrap().unwrap();
|
||||
match event {
|
||||
Event::Snapshot { state, .. } => {
|
||||
assert_eq!(state.catalog_status(), WorkerStatus::Running);
|
||||
return;
|
||||
}
|
||||
Event::Alert(_) => continue,
|
||||
other => panic!("expected Snapshot, got {other:?}"),
|
||||
}
|
||||
}
|
||||
let Event::Snapshot { state, .. } = handle.snapshot_event() else {
|
||||
panic!("expected snapshot");
|
||||
};
|
||||
assert_eq!(state.catalog_status(), WorkerStatus::Running);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -2928,7 +2896,7 @@ async fn status_json_reflects_worker_name() {
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Socket transport tests
|
||||
// Controller lifecycle tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[tokio::test]
|
||||
@@ -2966,283 +2934,148 @@ async fn shutdown_closes_method_admission_before_terminal_confirmation() {
|
||||
shutdown_rx.await.unwrap();
|
||||
}
|
||||
|
||||
/// Uses the real Controller and metadata store; only the model client is replaced.
|
||||
#[tokio::test]
|
||||
async fn shutdown_joins_socket_server_with_active_connection() {
|
||||
use tokio::net::UnixStream;
|
||||
|
||||
async fn register_peer_returns_existing_response_and_persists_reciprocal_metadata() {
|
||||
use session_store::{WorkerMetadata, WorkerMetadataStore};
|
||||
let worker = make_worker(MockClient::new(simple_text_events())).await;
|
||||
let runtime_base = tempfile::tempdir().unwrap();
|
||||
let (handle, shutdown_rx) = WorkerController::spawn(worker, runtime_base.path())
|
||||
.await
|
||||
let store = worker.store().clone();
|
||||
store
|
||||
.write(&WorkerMetadata::new("test-worker", None))
|
||||
.unwrap();
|
||||
let socket_path = handle.runtime_dir.socket_path();
|
||||
let _connection = UnixStream::connect(&socket_path).await.unwrap();
|
||||
|
||||
store
|
||||
.write(&WorkerMetadata::new("peer-worker", None))
|
||||
.unwrap();
|
||||
let handle = spawn_controller(worker).await;
|
||||
let mut events = handle.subscribe();
|
||||
handle
|
||||
.send(Method::Shutdown {
|
||||
command: worker_command(&handle),
|
||||
.send(Method::RegisterPeer {
|
||||
name: "peer-worker".into(),
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), shutdown_rx)
|
||||
assert!(
|
||||
drain_until(
|
||||
&mut events,
|
||||
std::time::Duration::from_secs(2),
|
||||
|event| matches!(
|
||||
event,
|
||||
Event::PeerRegistered { result } if result == &serde_json::json!({
|
||||
"source": "test-worker", "peer": "peer-worker",
|
||||
})
|
||||
)
|
||||
)
|
||||
.await
|
||||
.expect("controller should join its socket tasks")
|
||||
.expect("controller shutdown signal should remain open");
|
||||
assert!(!socket_path.exists());
|
||||
);
|
||||
assert_eq!(
|
||||
store.read_by_name("test-worker").unwrap().unwrap().peers[0].worker_name,
|
||||
"peer-worker"
|
||||
);
|
||||
assert_eq!(
|
||||
store.read_by_name("peer-worker").unwrap().unwrap().peers[0].worker_name,
|
||||
"test-worker"
|
||||
);
|
||||
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Idle);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn socket_run_receives_events() {
|
||||
use protocol::stream::{JsonLineReader, JsonLineWriter};
|
||||
use tokio::net::UnixStream;
|
||||
|
||||
let client = MockClient::new(simple_text_events());
|
||||
let worker = make_worker(client).await;
|
||||
async fn register_peer_missing_target_is_rejected_without_starting_a_worker() {
|
||||
use session_store::{WorkerMetadata, WorkerMetadataStore};
|
||||
let worker = make_worker(MockClient::new(simple_text_events())).await;
|
||||
let store = worker.store().clone();
|
||||
store
|
||||
.write(&WorkerMetadata::new("test-worker", None))
|
||||
.unwrap();
|
||||
let handle = spawn_controller(worker).await;
|
||||
|
||||
// Give the socket server a moment to bind
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
|
||||
let sock_path = handle.runtime_dir.socket_path();
|
||||
let stream = UnixStream::connect(&sock_path).await.unwrap();
|
||||
let (reader, writer) = stream.into_split();
|
||||
let mut reader = JsonLineReader::new(reader);
|
||||
let mut writer = JsonLineWriter::new(writer);
|
||||
|
||||
// Send run method via socket
|
||||
writer
|
||||
.write(&Method::submit_text(
|
||||
protocol::new_submission_request_id(),
|
||||
"Hello",
|
||||
))
|
||||
let mut events = handle.subscribe();
|
||||
handle
|
||||
.send(Method::RegisterPeer {
|
||||
name: "missing-peer".into(),
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Collect events
|
||||
let mut saw_turn_start = false;
|
||||
let mut saw_text_delta = false;
|
||||
let mut saw_turn_end = false;
|
||||
|
||||
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = reader.next::<Event>() => {
|
||||
match event {
|
||||
Ok(Some(Event::TurnStart { .. })) => saw_turn_start = true,
|
||||
Ok(Some(Event::TextDelta { .. })) => saw_text_delta = true,
|
||||
Ok(Some(Event::TurnEnd { .. })) => {
|
||||
saw_turn_end = true;
|
||||
break;
|
||||
}
|
||||
Ok(None) | Err(_) => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep_until(deadline) => break,
|
||||
}
|
||||
}
|
||||
|
||||
assert!(saw_turn_start, "should see turn_start via socket");
|
||||
assert!(saw_text_delta, "should see text_delta via socket");
|
||||
assert!(saw_turn_end, "should see turn_end via socket");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn socket_worker_event_turn_ended_while_idle_auto_starts_turn() {
|
||||
use protocol::stream::{JsonLineReader, JsonLineWriter};
|
||||
use tokio::net::UnixStream;
|
||||
|
||||
let client = MockClient::new(simple_text_events());
|
||||
let worker = make_worker(client).await;
|
||||
let handle = spawn_controller(worker).await;
|
||||
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
|
||||
let sock_path = handle.runtime_dir.socket_path();
|
||||
let stream = UnixStream::connect(&sock_path).await.unwrap();
|
||||
let (reader, writer) = stream.into_split();
|
||||
let mut reader = JsonLineReader::new(reader);
|
||||
let mut writer = JsonLineWriter::new(writer);
|
||||
|
||||
writer
|
||||
.write(&Method::WorkerEvent(protocol::WorkerEvent::TurnEnded {
|
||||
worker_name: "child".into(),
|
||||
}))
|
||||
assert!(
|
||||
drain_until(
|
||||
&mut events,
|
||||
std::time::Duration::from_secs(2),
|
||||
|event| matches!(
|
||||
event,
|
||||
Event::Error { code: worker::ErrorCode::InvalidRequest, message }
|
||||
if message.contains("missing-peer") && message.contains("does not exist")
|
||||
)
|
||||
)
|
||||
.await
|
||||
);
|
||||
assert!(store.read_by_name("missing-peer").unwrap().is_none());
|
||||
assert!(
|
||||
store
|
||||
.read_by_name("test-worker")
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.peers
|
||||
.is_empty()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_process_discovery_methods_require_host_or_runtime_authority() {
|
||||
let worker = make_worker(MockClient::new(simple_text_events())).await;
|
||||
let handle = spawn_controller(worker).await;
|
||||
let mut events = handle.subscribe();
|
||||
for method in [
|
||||
Method::ListWorkers,
|
||||
Method::RestoreWorker {
|
||||
name: "old-worker".into(),
|
||||
},
|
||||
] {
|
||||
handle.send(method).await.unwrap();
|
||||
assert!(
|
||||
drain_until(
|
||||
&mut events,
|
||||
std::time::Duration::from_secs(2),
|
||||
|event| matches!(
|
||||
event,
|
||||
Event::Error { code: worker::ErrorCode::InvalidRequest, message }
|
||||
if message.contains("Host or Runtime")
|
||||
)
|
||||
)
|
||||
.await
|
||||
);
|
||||
}
|
||||
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Idle);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn all_controller_launch_modes_leave_external_transport_to_the_host() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
for mode in ["direct", "runtime", "run"] {
|
||||
let worker = make_worker(MockClient::new(simple_text_events())).await;
|
||||
let base = root.path().join(mode);
|
||||
let (handle, shutdown) = match mode {
|
||||
"direct" => WorkerController::spawn(worker, &base).await,
|
||||
"runtime" => WorkerController::spawn_runtime_managed(worker, &base).await,
|
||||
"run" => {
|
||||
WorkerController::spawn_runtime_managed_run(worker, &base.join("runs/1")).await
|
||||
}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
.unwrap();
|
||||
|
||||
let mut saw_worker_event_echo = false;
|
||||
let mut saw_turn_start = false;
|
||||
let mut saw_turn_end = false;
|
||||
|
||||
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
|
||||
// The SystemItem and TurnEnd events arrive through independent
|
||||
// broadcast lanes (sink fan-out vs `working_event_tx`), so their relative
|
||||
// order on the wire is non-deterministic. Keep reading until both
|
||||
// are observed (or the deadline trips), rather than breaking on
|
||||
// the first TurnEnd.
|
||||
loop {
|
||||
if saw_worker_event_echo && saw_turn_end {
|
||||
break;
|
||||
}
|
||||
tokio::select! {
|
||||
event = reader.next::<Event>() => {
|
||||
match event {
|
||||
Ok(Some(Event::SystemItem { ref item, .. }))
|
||||
if item.get("kind").and_then(|k| k.as_str()) == Some("worker_event")
|
||||
&& item
|
||||
.pointer("/presentation/event/worker_name")
|
||||
.and_then(|v| v.as_str()) == Some("child") =>
|
||||
{
|
||||
saw_worker_event_echo = true;
|
||||
}
|
||||
Ok(Some(Event::TurnStart { .. })) => saw_turn_start = true,
|
||||
Ok(Some(Event::TurnEnd { .. })) => {
|
||||
saw_turn_end = true;
|
||||
}
|
||||
Ok(None) | Err(_) => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep_until(deadline) => break,
|
||||
}
|
||||
}
|
||||
|
||||
assert!(
|
||||
saw_worker_event_echo,
|
||||
"WorkerEvent::TurnEnded via socket should be echoed as Event::SystemItem(WorkerEvent)"
|
||||
);
|
||||
assert!(
|
||||
saw_turn_start,
|
||||
"WorkerEvent::TurnEnded via socket should auto-start a turn"
|
||||
);
|
||||
assert!(
|
||||
saw_turn_end,
|
||||
"auto-triggered turn should reach turn_end via socket"
|
||||
);
|
||||
}
|
||||
|
||||
async fn socket_error_after_method_line(
|
||||
handle: &WorkerHandle,
|
||||
line: &[u8],
|
||||
) -> (worker::ErrorCode, String) {
|
||||
use protocol::stream::JsonLineReader;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use tokio::net::UnixStream;
|
||||
|
||||
let sock_path = handle.runtime_dir.socket_path();
|
||||
let stream = UnixStream::connect(&sock_path).await.unwrap();
|
||||
let (reader, mut writer) = stream.into_split();
|
||||
let mut reader = JsonLineReader::new(reader);
|
||||
|
||||
writer.write_all(line).await.unwrap();
|
||||
|
||||
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = reader.next::<Event>() => {
|
||||
match event {
|
||||
Ok(Some(Event::Error { code, message })) => return (code, message),
|
||||
Ok(Some(_)) => {}
|
||||
Ok(None) => panic!("socket closed before invalid-method error"),
|
||||
Err(e) => panic!("socket read failed before invalid-method error: {e}"),
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep_until(deadline) => {
|
||||
panic!("timed out waiting for invalid-method error")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn socket_schema_invalid_method_returns_error() {
|
||||
let client = MockClient::new(simple_text_events());
|
||||
let worker = make_worker(client).await;
|
||||
let handle = spawn_controller(worker).await;
|
||||
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
|
||||
let (code, message) = socket_error_after_method_line(&handle, b"{\"bad\":\"json\"}\n").await;
|
||||
|
||||
assert_eq!(code, worker::ErrorCode::InvalidRequest);
|
||||
assert!(
|
||||
message.contains("invalid method"),
|
||||
"expected invalid-method diagnostic, got: {message}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn socket_malformed_method_returns_error() {
|
||||
let client = MockClient::new(simple_text_events());
|
||||
let worker = make_worker(client).await;
|
||||
let handle = spawn_controller(worker).await;
|
||||
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
|
||||
let (code, message) = socket_error_after_method_line(&handle, b"{not-json}\n").await;
|
||||
|
||||
assert_eq!(code, worker::ErrorCode::InvalidRequest);
|
||||
assert!(
|
||||
message.contains("invalid method"),
|
||||
"expected invalid-method diagnostic, got: {message}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn socket_peer_close_without_method_does_not_broadcast_error() {
|
||||
use protocol::stream::JsonLineReader;
|
||||
use tokio::net::UnixStream;
|
||||
|
||||
let client = MockClient::new(simple_text_events());
|
||||
let worker = make_worker(client).await;
|
||||
let handle = spawn_controller(worker).await;
|
||||
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
|
||||
let mut broadcast_rx = handle.subscribe();
|
||||
let sock_path = handle.runtime_dir.socket_path();
|
||||
let stream = UnixStream::connect(&sock_path).await.unwrap();
|
||||
let (reader, writer) = stream.into_split();
|
||||
let mut reader = JsonLineReader::new(reader);
|
||||
|
||||
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = reader.next::<Event>() => {
|
||||
match event {
|
||||
Ok(Some(Event::Snapshot { .. })) => break,
|
||||
Ok(Some(_)) => {}
|
||||
Ok(None) => panic!("socket closed before connect-time snapshot"),
|
||||
Err(e) => panic!("socket read failed before connect-time snapshot: {e}"),
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep_until(deadline) => {
|
||||
panic!("timed out waiting for connect-time snapshot")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
drop(writer);
|
||||
drop(reader);
|
||||
|
||||
let deadline = tokio::time::Instant::now() + std::time::Duration::from_millis(200);
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = broadcast_rx.recv() => {
|
||||
match event {
|
||||
Ok(Event::Error { code, message }) => {
|
||||
panic!("peer close without Method broadcast error {code:?}: {message}")
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
|
||||
panic!("broadcast receiver lagged while checking peer close: {n}")
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep_until(deadline) => break,
|
||||
for legacy in ["pid", "status.json", "manifest.toml", "sock", "worker.sock"] {
|
||||
assert!(
|
||||
!handle.runtime_dir.path().join(legacy).exists(),
|
||||
"{mode} created {legacy}"
|
||||
);
|
||||
}
|
||||
handle
|
||||
.send(Method::Shutdown {
|
||||
command: worker_command(&handle),
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), shutdown)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4006,27 +3839,15 @@ fn item_text_contains(item: &Item, needle: &str) -> bool {
|
||||
}
|
||||
|
||||
async fn snapshot_contains_user_input(handle: &WorkerHandle, needle: &str) -> bool {
|
||||
let stream = tokio::net::UnixStream::connect(handle.runtime_dir.socket_path())
|
||||
.await
|
||||
.unwrap();
|
||||
let (reader, _writer) = stream.into_split();
|
||||
let mut reader = protocol::stream::JsonLineReader::new(reader);
|
||||
|
||||
loop {
|
||||
let event = reader.next::<Event>().await.unwrap().unwrap();
|
||||
match event {
|
||||
Event::Snapshot { session, .. } => {
|
||||
return session.entries.into_iter().any(|entry| match entry.data {
|
||||
protocol::SessionSnapshotEntryData::UserInput { segments } => {
|
||||
protocol::Segment::flatten_to_text(&segments).contains(needle)
|
||||
}
|
||||
_ => false,
|
||||
});
|
||||
}
|
||||
Event::Alert(_) => continue,
|
||||
other => panic!("expected Snapshot first, got {other:?}"),
|
||||
let Event::Snapshot { session, .. } = handle.snapshot_event() else {
|
||||
panic!("expected snapshot");
|
||||
};
|
||||
session.entries.into_iter().any(|entry| match entry.data {
|
||||
protocol::SessionSnapshotEntryData::UserInput { segments } => {
|
||||
protocol::Segment::flatten_to_text(&segments).contains(needle)
|
||||
}
|
||||
}
|
||||
_ => false,
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -1,15 +1,10 @@
|
||||
//! Legacy process callback events are diagnostics only after Internal SubWorker migration.
|
||||
//! Agent-visible Internal SubWorker event summaries.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use protocol::{Permission, ScopeRule, WorkerEvent};
|
||||
use tempfile::TempDir;
|
||||
use worker::ipc::event::{apply_event_side_effects, render_event};
|
||||
use worker::runtime::dir::RuntimeDir;
|
||||
use worker::spawn::registry::SpawnedWorkerRegistry;
|
||||
use protocol::WorkerEvent;
|
||||
use worker::ipc::event::render_event;
|
||||
|
||||
#[test]
|
||||
fn render_event_keeps_bounded_legacy_diagnostics() {
|
||||
fn render_event_identifies_child_and_failure() {
|
||||
let rendered = render_event(&WorkerEvent::Errored {
|
||||
worker_name: "legacy-child".into(),
|
||||
message: "boom".into(),
|
||||
@@ -17,30 +12,3 @@ fn render_event_keeps_bounded_legacy_diagnostics() {
|
||||
assert!(rendered.contains("legacy-child"));
|
||||
assert!(rendered.contains("boom"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn legacy_callback_cannot_register_process_subworker_authority() {
|
||||
let runtime_base = TempDir::new().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: "legacy-parent".into(),
|
||||
sub_worker: "legacy-child".into(),
|
||||
sub_socket: "/tmp/legacy-child.sock".into(),
|
||||
scope: vec![ScopeRule {
|
||||
target: scope_root.path().to_path_buf(),
|
||||
permission: Permission::Write,
|
||||
recursive: true,
|
||||
symlink_policy: Default::default(),
|
||||
}],
|
||||
};
|
||||
|
||||
apply_event_side_effects(&event, ®istry, "parent", &None).await;
|
||||
|
||||
assert!(!runtime_dir.path().join("spawned_workers.json").exists());
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user