From fe373b5656a316694090f9cbf231bb5a1707bd28 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 13 Aug 2026 03:25:57 +0900 Subject: [PATCH] runtime: remove embedded worker socket transport --- Cargo.lock | 1 + crates/worker-runtime/Cargo.toml | 1 + crates/worker-runtime/src/worker_backend.rs | 248 ++++++++++++++++-- crates/worker/src/controller.rs | 80 +++++- crates/worker/src/lib.rs | 2 +- crates/workspace-server/src/server.rs | 1 + ...f-termination-and-runtime-restore-panic.md | 45 ++++ 7 files changed, 345 insertions(+), 33 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5b925ccb..672ddbc4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6056,6 +6056,7 @@ dependencies = [ "ring", "serde", "serde_json", + "serial_test", "session-store", "sha2 0.11.0", "tar", diff --git a/crates/worker-runtime/Cargo.toml b/crates/worker-runtime/Cargo.toml index d4ca9d64..e073df2f 100644 --- a/crates/worker-runtime/Cargo.toml +++ b/crates/worker-runtime/Cargo.toml @@ -49,6 +49,7 @@ workdir.workspace = true [dev-dependencies] futures.workspace = true llm-engine.workspace = true +serial_test = "3.4.0" tempfile.workspace = true tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } tokio-tungstenite.workspace = true diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 1c4b047a..45e359bd 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -54,8 +54,8 @@ use worker::feature::builtin::{ use worker::ipc::protocol_session::{live_log_entry_event, subscribe_worker_protocol_session}; use worker::{ PromptLoader, SegmentLogSink, WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN, Worker, - WorkerController, WorkerError, WorkerFilesystemAuthority, WorkerHandle, WorkerSharedState, - WorkerWorkspaceContext, WorkspaceClient, WorkspaceId, + WorkerController, WorkerControllerTransport, WorkerError, WorkerFilesystemAuthority, + WorkerHandle, WorkerSharedState, WorkerWorkspaceContext, WorkspaceClient, WorkspaceId, }; const DEFAULT_BACKEND_ID: &str = "worker-crate"; @@ -221,6 +221,7 @@ pub struct ProfileRuntimeWorkerFactory { runtime_id: Option, worker_mutation_identity: Option, embedded_worker_mutation_dispatcher: Option>, + controller_transport: WorkerControllerTransport, } impl ProfileRuntimeWorkerFactory { @@ -235,6 +236,7 @@ impl ProfileRuntimeWorkerFactory { runtime_id: None, worker_mutation_identity: None, embedded_worker_mutation_dispatcher: None, + controller_transport: WorkerControllerTransport::UnixSocket, } } @@ -264,6 +266,14 @@ impl ProfileRuntimeWorkerFactory { self } + pub fn with_controller_transport( + mut self, + controller_transport: WorkerControllerTransport, + ) -> Self { + self.controller_transport = controller_transport; + self + } + pub fn with_runtime_store_dir(mut self, runtime_store_dir: impl Into) -> Self { self.worker_aggregate_root = Some(runtime_store_dir.into().join("workers")); self @@ -649,14 +659,18 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { let run_dir = worker_aggregate_dir .join("runs") .join(request.run_generation.to_string()); - let (handle, shutdown_rx) = WorkerController::spawn_runtime_managed_run(worker, &run_dir) - .await - .map_err(|err| { - format!( - "failed to spawn Worker controller in {}: {err}", - run_dir.display() - ) - })?; + let (handle, shutdown_rx) = WorkerController::spawn_runtime_managed_run_with_transport( + worker, + &run_dir, + self.controller_transport, + ) + .await + .map_err(|err| { + format!( + "failed to spawn Worker controller in {}: {err}", + run_dir.display() + ) + })?; if flow_transition_enabled { handle.shared_state.enable_flow_transition(); } @@ -799,14 +813,18 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { let run_dir = worker_aggregate_dir .join("runs") .join(request.run_generation.to_string()); - let (handle, shutdown_rx) = WorkerController::spawn_runtime_managed_run(worker, &run_dir) - .await - .map_err(|err| { - format!( - "failed to spawn restored Worker controller in {}: {err}", - run_dir.display() - ) - })?; + let (handle, shutdown_rx) = WorkerController::spawn_runtime_managed_run_with_transport( + worker, + &run_dir, + self.controller_transport, + ) + .await + .map_err(|err| { + format!( + "failed to spawn restored Worker controller in {}: {err}", + run_dir.display() + ) + })?; if flow_transition_enabled { handle.shared_state.enable_flow_transition(); } @@ -2079,14 +2097,29 @@ mod tests { } fn sample_profile_archive() -> crate::profile_archive::ProfileSourceArchive { - let entrypoints = - BTreeMap::from([("default".to_string(), "profiles/default.dcdl".to_string())]); + let entrypoints = BTreeMap::from([ + ("default".to_string(), "profiles/default.dcdl".to_string()), + ( + "builtin:default".to_string(), + "profiles/default.dcdl".to_string(), + ), + ( + "builtin:companion".to_string(), + "profiles/default.dcdl".to_string(), + ), + ]); let sources = BTreeMap::from([( "profiles/default.dcdl".to_string(), r#"{ slug = "default"; description = "Default"; scope = "workspace_read"; + model = { + scheme = "anthropic"; + model_id = "test-model"; + auth = { kind = "none"; }; + }; + engine = { max_tokens = 100; }; }"# .to_string(), )]); @@ -2343,6 +2376,7 @@ mod tests { } #[tokio::test] + #[serial_test::serial(worker_allocation)] async fn restore_pending_worker_uses_saved_manifest_snapshot() { let root = tempfile::tempdir().unwrap(); let runtime_store_dir = root.path().join("runtime"); @@ -2431,6 +2465,180 @@ mod tests { assert!(!run_dir.join("worker.sock").exists()); } + #[tokio::test] + #[serial_test::serial(worker_allocation)] + async fn in_process_restore_does_not_bind_unix_socket_under_overlong_store_path() { + let root = tempfile::tempdir().unwrap(); + let long_component = "embedded-workspace-store-segment".repeat(4); + let runtime_store_dir = root.path().join(long_component); + let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(1)); + let worker_aggregate_dir = runtime_store_dir.join("workers/1"); + let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref); + let session_id = session_store::new_session_id(); + let manifest = manifest::WorkerManifest::from_toml(&format!( + r#" + [worker] + name = "{}" + pwd = "{}" + + [model] + scheme = "anthropic" + model_id = "test-model" + auth = {{ kind = "none" }} + + [engine] + max_tokens = 100 + + [[scope.allow]] + target = "{}" + permission = "write" + "#, + worker_name, + root.path().display(), + root.path().display(), + )) + .unwrap(); + WorkerAggregateStore::new(&worker_aggregate_dir, &worker_name) + .unwrap() + .set_active( + &worker_name, + Some(session_store::WorkerActiveSegmentRef::pending_segment( + session_id, + )), + Some(serde_json::to_value(&manifest).unwrap()), + ) + .unwrap(); + + let run_dir = runtime_store_dir.join("workers/1/runs/2"); + let socket_path = run_dir.join("worker.sock"); + assert!( + socket_path.as_os_str().as_encoded_bytes().len() > 107, + "test path must exceed Linux sockaddr_un.sun_path capacity: {}", + socket_path.display() + ); + + let controller = ProfileRuntimeWorkerFactory::new(root.path()) + .with_runtime_store_dir(&runtime_store_dir) + .with_controller_transport(WorkerControllerTransport::InProcess) + .restore_controller(WorkerExecutionRestoreRequest { + worker_ref: worker_ref.clone(), + run_generation: 2, + request: create_request("embedded restore"), + workspace_scope: None, + context: test_execution_context(worker_ref), + previous_working_directory: None, + working_directory: None, + config_bundle: None, + }) + .await + .expect("in-process restore must not bind the overlong Unix socket path"); + + assert_eq!( + controller.handle.shared_state.get_status(), + WorkerStatus::Idle + ); + assert!(!socket_path.exists()); + assert!(run_dir.join("worker.out.log").is_file()); + assert!(run_dir.join("worker.err.log").is_file()); + controller.handle.send(Method::Shutdown).await.unwrap(); + if let Some(receiver) = controller.shutdown.lock().await.take() { + receiver.await.unwrap(); + } + assert!(!socket_path.exists()); + } + + #[test] + #[serial_test::serial(worker_allocation)] + fn in_process_runtime_reopens_persisted_worker_without_overlong_unix_socket() { + let root = tempfile::tempdir().unwrap(); + let long_component = "embedded-workspace-store-segment".repeat(4); + let runtime_store_dir = root.path().join(long_component); + let runtime_options = crate::fs_store::FsRuntimeStoreOptions { + root: runtime_store_dir.clone(), + display_name: Some("embedded".to_string()), + }; + + let backend = Arc::new( + WorkerRuntimeExecutionBackend::new( + ProfileRuntimeWorkerFactory::new(root.path()) + .with_runtime_store_dir(&runtime_store_dir) + .with_controller_transport(WorkerControllerTransport::InProcess), + ) + .unwrap(), + ); + let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend( + runtime_options.clone(), + backend.clone(), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let mut request = create_request("embedded singleton"); + request.profile = ProfileSelector::Builtin("default".to_string()); + let worker = runtime.create_worker(request).unwrap(); + let first_run_socket = runtime_store_dir.join("workers/1/runs/1/worker.sock"); + assert!( + first_run_socket.as_os_str().as_encoded_bytes().len() > 107, + "test path must exceed Linux sockaddr_un.sun_path capacity: {}", + first_run_socket.display() + ); + assert!(!first_run_socket.exists()); + + let (handle, shutdown) = { + let workers = backend.workers.lock().unwrap(); + let execution = workers.get(&worker.worker_ref).unwrap(); + (execution.handle.clone(), execution.shutdown.clone()) + }; + backend + .run_on_adapter_runtime(async move { + handle + .send(Method::Shutdown) + .await + .map_err(|error| error.to_string())?; + if let Some(receiver) = shutdown.lock().await.take() { + receiver.await.map_err(|error| error.to_string())?; + } + Ok(()) + }) + .unwrap(); + drop(runtime); + drop(backend); + + let restored_backend = Arc::new( + WorkerRuntimeExecutionBackend::new( + ProfileRuntimeWorkerFactory::new(root.path()) + .with_runtime_store_dir(&runtime_store_dir) + .with_controller_transport(WorkerControllerTransport::InProcess), + ) + .unwrap(), + ); + let restored = EmbeddedRuntime::with_fs_store_and_execution_backend( + runtime_options, + restored_backend.clone(), + ) + .expect("persisted in-process Worker must restore without binding its run path"); + let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap(); + let diagnostics = restored.diagnostics().unwrap(); + assert_eq!( + restored_worker.status, + crate::catalog::WorkerStatus::Idle, + "restore diagnostics: {diagnostics:#?}" + ); + assert!(!diagnostics.iter().any(|diagnostic| { + diagnostic.code == "worker_execution_restore_failed" + && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref) + })); + let restored_run = runtime_store_dir.join("workers/1/runs/2"); + assert!(!restored_run.join("worker.sock").exists()); + assert!(restored_run.join("worker.out.log").is_file()); + assert!(restored_run.join("worker.err.log").is_file()); + + restored + .stop_worker(&worker.worker_ref, Some("test cleanup".to_string())) + .unwrap(); + drop(restored); + drop(restored_backend); + } + #[test] fn builtin_profile_selector_is_not_double_prefixed() { assert_eq!( diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index f72568ac..6aa641f6 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -200,6 +200,18 @@ fn should_auto_run_notification(status: WorkerStatus, auto_run: bool) -> bool { pub type ShutdownReceiver = oneshot::Receiver<()>; +/// Client transport exposed by a Worker controller. +/// +/// Process-hosted Workers use a Unix socket for external attach clients. Runtimes +/// that retain the returned [`WorkerHandle`] in the same process can disable that +/// redundant listener and drive the controller directly through its channels. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum WorkerControllerTransport { + #[default] + UnixSocket, + InProcess, +} + pub struct WorkerController; impl WorkerController { @@ -211,7 +223,14 @@ impl WorkerController { C: LlmClient + Clone + 'static, St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, { - Self::spawn_inner(worker, runtime_base, false, None).await + Self::spawn_inner( + worker, + runtime_base, + false, + None, + WorkerControllerTransport::UnixSocket, + ) + .await } /// Spawn a Worker owned by `worker-runtime`. @@ -227,7 +246,14 @@ impl WorkerController { C: LlmClient + Clone + 'static, St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, { - Self::spawn_inner(worker, runtime_base, true, None).await + Self::spawn_inner( + worker, + runtime_base, + true, + None, + WorkerControllerTransport::UnixSocket, + ) + .await } /// Spawn into an exact persistent `runs/` directory. @@ -235,6 +261,25 @@ impl WorkerController { worker: Worker, run_dir: &Path, ) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error> + where + C: LlmClient + Clone + 'static, + St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, + { + Self::spawn_runtime_managed_run_with_transport( + worker, + run_dir, + WorkerControllerTransport::UnixSocket, + ) + .await + } + + /// Spawn into an exact persistent `runs/` directory using the + /// requested client transport. + pub async fn spawn_runtime_managed_run_with_transport( + worker: Worker, + run_dir: &Path, + transport: WorkerControllerTransport, + ) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error> where C: LlmClient + Clone + 'static, St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, @@ -242,7 +287,7 @@ impl WorkerController { let parent = run_dir .parent() .ok_or_else(|| std::io::Error::other("run path has no parent"))?; - Self::spawn_inner(worker, parent, true, Some(run_dir)).await + Self::spawn_inner(worker, parent, true, Some(run_dir), transport).await } async fn spawn_inner( @@ -250,14 +295,21 @@ impl WorkerController { runtime_base: &Path, runtime_managed: bool, runtime_run: Option<&Path>, + transport: WorkerControllerTransport, ) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error> where C: LlmClient + Clone + 'static, St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, { let session = worker.workdir_session().cloned(); - let result = - Self::spawn_initialized(worker, runtime_base, runtime_managed, runtime_run).await; + let result = Self::spawn_initialized( + worker, + runtime_base, + runtime_managed, + runtime_run, + transport, + ) + .await; if result.is_err() && let Some(session) = session && let Err(error) = session.close().await @@ -272,6 +324,7 @@ impl WorkerController { runtime_base: &Path, runtime_managed: bool, runtime_run: Option<&Path>, + transport: WorkerControllerTransport, ) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error> where C: LlmClient + Clone + 'static, @@ -287,10 +340,9 @@ impl WorkerController { let in_flight = InFlightEvents::new(event_tx.clone()); worker.attach_in_flight_events(in_flight.clone()); - // Runtime directory is created before tool registration because - // the spawn-tool factories need its socket path, and before the - // initial status/history writes consume the greeting we build - // after registration is complete. + // Runtime directory is created before tool registration because it owns + // bounded tool artifacts, and before initial status/history writes consume + // the greeting we build after registration is complete. let runtime_dir = Arc::new(if let Some(run_dir) = runtime_run { RuntimeDir::create_worker_run(run_dir).await? } else if runtime_managed { @@ -411,7 +463,10 @@ impl WorkerController { sink: worker.sink(), }; - let socket_server = SocketServer::start(&handle).await?; + let socket_server = match transport { + WorkerControllerTransport::UnixSocket => Some(SocketServer::start(&handle).await?), + WorkerControllerTransport::InProcess => None, + }; // === 5. controller_loop === // Clone cancel sender and notification buffer before moving worker @@ -908,13 +963,14 @@ async fn controller_loop( spawner_name: String, spawned_registry: Arc, shutdown_tx: oneshot::Sender<()>, - socket_server: SocketServer, + socket_server: Option, shutdown_after_idle: ShutdownAfterIdleRequest, ) where C: LlmClient + Clone + 'static, St: Store + WorkerMetadataStore + Clone + 'static, { - // Hold socket server alive for the lifetime of the controller task. + // Hold an optional external attach server alive for the controller lifetime. + // In-process runtimes retain and drive the WorkerHandle directly. let _socket_server = socket_server; let discovery_runtime_base = runtime_dir diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 7df10a5a..469ad674 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -23,7 +23,7 @@ mod permission; mod worker; pub use compact::token_counter::{EstimateSource, SplitPoint, TokenEstimate}; -pub use controller::{ShutdownReceiver, WorkerController, WorkerHandle}; +pub use controller::{ShutdownReceiver, WorkerController, WorkerControllerTransport, WorkerHandle}; pub use hook::{Hook, HookEventKind, HookRegistryBuilder}; pub use ipc::alerter::Alerter; pub use ipc::server::SocketServer; diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 910a6bee..60d6c48f 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -655,6 +655,7 @@ impl WorkspaceApi { worker_remove_dispatcher.clone(), ) .with_runtime_store_dir(config.embedded_runtime_store_root.clone()) + .with_controller_transport(worker::WorkerControllerTransport::InProcess) .with_resource_client(Arc::new(resource_broker.clone())), ) .map_err(|err| { diff --git a/docs/report/2026-08-12-dogfood-restart-self-termination-and-runtime-restore-panic.md b/docs/report/2026-08-12-dogfood-restart-self-termination-and-runtime-restore-panic.md index 494d1afb..8ee032a3 100644 --- a/docs/report/2026-08-12-dogfood-restart-self-termination-and-runtime-restore-panic.md +++ b/docs/report/2026-08-12-dogfood-restart-self-termination-and-runtime-restore-panic.md @@ -119,6 +119,51 @@ Server and Runtime externally. Post-restart checks confirmed: The reusable regression gate is `scripts/isolated-startup-smoke.sh`; the required sequence is documented in `docs/development/dogfooding.md`. +### Follow-up: embedded singleton restore failure + +The external restart recovered the remote Runtime Workers, but it did not recover +the two persisted embedded singleton executions. Workspace Memory Consolidation +(`embedded-worker-runtime/3`) and Workspace Orchestrator +(`embedded-worker-runtime/6`) are projected as `stopped` without a user stop. +The embedded Runtime diagnostics record `worker_execution_restore_failed` for +both with `path must be shorter than SUN_LEN`. Their persisted records still +contain execution bindings and session files, so this is a failed automatic +restore after Server restart, not a graceful terminal stop. Follow-up Ticket: +`00001KZVFQPSK`. + +The isolated pre-dogfood gate originally covered remote Runtime persistence reopen +but not persisted embedded singleton restoration; that missing scenario allowed +this failure through even after the remote smoke passed. + +### Embedded IPC resolution + +The embedded Runtime retained every controller's `WorkerHandle` in the Server +process, but `WorkerController` still unconditionally bound `worker.sock` below +its persistent `runs/` directory. The workspace-owned embedded store +path plus Worker/run components exceeded Linux `sockaddr_un.sun_path`, so restore +failed before the in-process handle could be registered. + +The controller now has an explicit transport policy. Standalone and remote +Runtime factories keep the default Unix-socket transport for external attach +clients; the Server-owned embedded factory selects `InProcess` and drives the +same controller exclusively through `WorkerHandle` channels. Persistent run +logs and artifact directories remain unchanged, but embedded spawn and restore +no longer create a socket file. + +Regression coverage now includes both direct restore and complete fs-store +reopen under a path longer than `SUN_LEN`, with assertions that the Worker +returns to `idle`, no `worker_execution_restore_failed` diagnostic exists, and +neither run generation contains `worker.sock`. The existing isolated +production-binary smoke remains the remote Runtime restart gate; embedding the +Server-owned Runtime into that script requires a separate clean embedded spawn +fixture because the normal product singleton lifecycle is not a generic smoke +fixture. + +Memory Consolidation also has an earlier independent session failure: +`memory tools require Backend Workspace API authority and an authenticated +workspace id`. That authority problem is not the cause of the `SUN_LEN` restore +failure and should remain separate. + ## Current state observed during diagnosis - `yoi-server` is not running. Legacy `target/debug/worker-runtime` PID `1820761` remains listening on `127.0.0.1:38800`; it uses the separate standalone Runtime catalog containing Workers 43, 57, 58, 59, 60, 61, 62, and 63.