feat: add in-process standalone worker host

2026-08-30 11:29:59 +09:00
parent 22867faa9c
commit 4d9b211d69
14 changed files with 782 additions and 6 deletions
@@ -0,0 +1,22 @@
+[package]
+name = "standalone"
+description = "In-process standalone Worker host"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+agen.workspace = true
+manifest.workspace = true
+protocol.workspace = true
+session-store.workspace = true
+thiserror.workspace = true
+tokio = { workspace = true, features = ["rt", "sync", "time"] }
+worker.workspace = true
+
+[dev-dependencies]
+async-trait.workspace = true
+futures.workspace = true
+tempfile.workspace = true
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] }
+uuid = { workspace = true, features = ["v7"] }
@@ -0,0 +1,154 @@
+use std::time::Duration;
+
+use agen::llm_client::client::LlmClient;
+use protocol::{Event, Method};
+use session_store::{CombinedStore, FsStore, FsWorkerStore};
+use thiserror::Error;
+use tokio::sync::broadcast;
+use worker::bootstrap::{WorkerBootstrap, WorkerBootstrapError, WorkerBootstrapLayout};
+use worker::controller::WorkerControllerTransport;
+use worker::{WorkerError, WorkerFilesystemAuthority, WorkerWorkspaceContext};
+
+use crate::launch::ResolvedStandaloneLaunch;
+
+const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// One top-level Worker plus its process-owned local Workdir session.
+///
+/// The host deliberately exposes the existing typed Worker protocol rather
+/// than owning an HTTP/WebSocket server or a second execution model.
+pub struct StandaloneHost {
+    handle: worker::WorkerHandle,
+    shutdown: worker::controller::ShutdownReceiver,
+    shutdown_timeout: Duration,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
+pub enum StandaloneStartupError {
+    #[error("the standalone state store could not be opened")]
+    StateStore,
+    #[error("the resolved Worker configuration is invalid")]
+    WorkerConfiguration,
+    #[error("the configured model provider is unavailable")]
+    ModelProvider,
+    #[error("the fixed standalone feature composition could not be installed")]
+    FeatureComposition,
+    #[error("the in-process Worker controller could not start")]
+    Controller,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
+pub enum StandaloneRequestError {
+    #[error("the standalone Worker is no longer accepting requests")]
+    WorkerUnavailable,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
+pub enum StandaloneShutdownError {
+    #[error("the standalone Worker did not stop before the shutdown deadline")]
+    DeadlineExceeded,
+    #[error("the standalone Worker shutdown confirmation was lost")]
+    ConfirmationLost,
+}
+
+impl StandaloneHost {
+    pub async fn start(launch: ResolvedStandaloneLaunch) -> Result<Self, StandaloneStartupError> {
+        Self::start_with_optional_model_client(launch, None).await
+    }
+
+    pub async fn start_with_model_client<C>(
+        launch: ResolvedStandaloneLaunch,
+        model_client: C,
+    ) -> Result<Self, StandaloneStartupError>
+    where
+        C: LlmClient + 'static,
+    {
+        Self::start_with_optional_model_client(launch, Some(Box::new(model_client))).await
+    }
+
+    async fn start_with_optional_model_client(
+        launch: ResolvedStandaloneLaunch,
+        model_client: Option<Box<dyn LlmClient>>,
+    ) -> Result<Self, StandaloneStartupError> {
+        std::fs::create_dir_all(&launch.state_dir)
+            .map_err(|_| StandaloneStartupError::StateStore)?;
+        let session_store = FsStore::new(launch.state_dir.join("sessions"))
+            .map_err(|_| StandaloneStartupError::StateStore)?;
+        let worker_store = FsWorkerStore::new(launch.state_dir.join("workers"))
+            .map_err(|_| StandaloneStartupError::StateStore)?;
+        let store = CombinedStore::new(session_store, worker_store);
+        let filesystem_authority =
+            WorkerFilesystemAuthority::local(launch.cwd.clone(), launch.cwd.clone());
+        let workspace_context = WorkerWorkspaceContext::local_filesystem(None);
+        let runtime_base = launch.state_dir.join("runtime");
+
+        let mut bootstrap = WorkerBootstrap::new(
+            launch.profile.manifest,
+            store,
+            launch.prompt_catalog,
+            workspace_context,
+            filesystem_authority,
+            WorkerBootstrapLayout::Direct { runtime_base },
+            WorkerControllerTransport::InProcess,
+        );
+        if let Some(model_client) = model_client {
+            bootstrap = bootstrap.with_model_client(model_client);
+        }
+        let started = bootstrap.start().await.map_err(classify_startup_error)?;
+
+        Ok(Self {
+            handle: started.handle,
+            shutdown: started.shutdown,
+            shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
+        })
+    }
+
+    pub async fn send(&self, method: Method) -> Result<(), StandaloneRequestError> {
+        self.handle
+            .send(method)
+            .await
+            .map_err(|_| StandaloneRequestError::WorkerUnavailable)
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
+        self.handle.subscribe()
+    }
+
+    pub fn snapshot(&self) -> Event {
+        self.handle.snapshot_event()
+    }
+
+    pub fn with_shutdown_timeout(mut self, shutdown_timeout: Duration) -> Self {
+        self.shutdown_timeout = shutdown_timeout;
+        self
+    }
+
+    pub async fn shutdown(self) -> Result<(), StandaloneShutdownError> {
+        let StandaloneHost {
+            handle,
+            shutdown,
+            shutdown_timeout,
+        } = self;
+        let _ = handle.send(Method::Shutdown).await;
+        match tokio::time::timeout(shutdown_timeout, shutdown).await {
+            Ok(Ok(())) => Ok(()),
+            Ok(Err(_)) => Err(StandaloneShutdownError::ConfirmationLost),
+            Err(_) => Err(StandaloneShutdownError::DeadlineExceeded),
+        }
+    }
+}
+
+fn classify_startup_error(error: WorkerBootstrapError) -> StandaloneStartupError {
+    match error {
+        WorkerBootstrapError::Worker(WorkerError::Provider(_)) => {
+            StandaloneStartupError::ModelProvider
+        }
+        WorkerBootstrapError::Worker(_) => StandaloneStartupError::WorkerConfiguration,
+        WorkerBootstrapError::Controller { source, .. }
+            if source.kind() == std::io::ErrorKind::Other =>
+        {
+            StandaloneStartupError::FeatureComposition
+        }
+        WorkerBootstrapError::Controller { .. } => StandaloneStartupError::Controller,
+    }
+}
@@ -0,0 +1,86 @@
+use std::path::{Path, PathBuf};
+
+use manifest::{
+    ProfileExecutionTarget, ProfileResolveOptions, ProfileResolver, ProfileSelector,
+    ResolvedProfile,
+};
+use thiserror::Error;
+use worker::PromptCatalogSource;
+
+/// Process launch input resolved before any Worker/session side effect occurs.
+#[derive(Debug, Clone)]
+pub struct StandaloneLaunchConfig {
+    pub cwd: PathBuf,
+    pub state_dir: PathBuf,
+    pub profile: ProfileSelector,
+    pub worker_name: String,
+}
+
+pub struct ResolvedStandaloneLaunch {
+    pub cwd: PathBuf,
+    pub state_dir: PathBuf,
+    pub profile: ResolvedProfile,
+    pub prompt_catalog: PromptCatalogSource,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
+pub enum StandaloneLaunchError {
+    #[error("the standalone working directory is unavailable")]
+    WorkingDirectoryUnavailable,
+    #[error("path-based profiles are not standalone launch authority")]
+    PathProfileUnsupported,
+    #[error("the standalone profile could not be resolved")]
+    ProfileResolutionFailed,
+}
+
+impl StandaloneLaunchConfig {
+    pub fn new(
+        cwd: impl Into<PathBuf>,
+        state_dir: impl Into<PathBuf>,
+        profile: ProfileSelector,
+        worker_name: impl Into<String>,
+    ) -> Self {
+        Self {
+            cwd: cwd.into(),
+            state_dir: state_dir.into(),
+            profile,
+            worker_name: worker_name.into(),
+        }
+    }
+
+    /// Resolve only built-in/XDG profile authority and bind standalone scope
+    /// to the canonical process cwd. Repository-local profile discovery is
+    /// deliberately not part of this path.
+    pub fn resolve(self) -> Result<ResolvedStandaloneLaunch, StandaloneLaunchError> {
+        if matches!(self.profile, ProfileSelector::Path { .. }) {
+            return Err(StandaloneLaunchError::PathProfileUnsupported);
+        }
+        let cwd = canonical_directory(&self.cwd)?;
+        let profile = ProfileResolver::new()
+            .with_workspace_base(&cwd)
+            .resolve_for_target(
+                &self.profile,
+                ProfileResolveOptions {
+                    worker_name: Some(self.worker_name),
+                },
+                ProfileExecutionTarget::Standalone,
+            )
+            .map_err(|_| StandaloneLaunchError::ProfileResolutionFailed)?;
+
+        Ok(ResolvedStandaloneLaunch {
+            cwd,
+            state_dir: self.state_dir,
+            profile,
+            prompt_catalog: PromptCatalogSource::builtins_only(),
+        })
+    }
+}
+
+fn canonical_directory(path: &Path) -> Result<PathBuf, StandaloneLaunchError> {
+    let path = std::fs::canonicalize(path)
+        .map_err(|_| StandaloneLaunchError::WorkingDirectoryUnavailable)?;
+    if !path.is_dir() {
+        return Err(StandaloneLaunchError::WorkingDirectoryUnavailable);
+    }
+    Ok(path)
+}
@@ -0,0 +1,13 @@
+//! In-process standalone host for one top-level Yoi Worker.
+//!
+//! The crate composes existing `worker`, `manifest`, `session-store`, and
+//! `workdir` contracts. It intentionally owns no TUI, Runtime, Workspace
+//! Server, HTTP, WebSocket, subprocess Worker, or alternative execution path.
+
+pub mod host;
+pub mod launch;
+
+pub use host::{
+    StandaloneHost, StandaloneRequestError, StandaloneShutdownError, StandaloneStartupError,
+};
+pub use launch::{ResolvedStandaloneLaunch, StandaloneLaunchConfig, StandaloneLaunchError};
@@ -0,0 +1,212 @@
+use std::collections::VecDeque;
+use std::pin::Pin;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use agen::llm_client::client::LlmClient;
+use agen::llm_client::error::ClientError;
+use agen::llm_client::event::{Event as LlmEvent, StopReason};
+use agen::llm_client::types::Request;
+use async_trait::async_trait;
+use futures::{Stream, stream};
+use protocol::{Event, Method};
+use standalone::{StandaloneHost, StandaloneLaunchConfig};
+use uuid::Uuid;
+
+#[derive(Clone)]
+struct ScriptedClient {
+    responses: Arc<Mutex<VecDeque<Vec<LlmEvent>>>>,
+    requests: Arc<Mutex<Vec<Request>>>,
+}
+
+impl ScriptedClient {
+    fn new(responses: Vec<Vec<LlmEvent>>) -> Self {
+        Self {
+            responses: Arc::new(Mutex::new(responses.into())),
+            requests: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+
+    fn requests(&self) -> Vec<Request> {
+        self.requests.lock().expect("requests lock").clone()
+    }
+}
+
+#[async_trait]
+impl LlmClient for ScriptedClient {
+    async fn stream(
+        &self,
+        request: Request,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
+    {
+        self.requests.lock().expect("requests lock").push(request);
+        let response = self
+            .responses
+            .lock()
+            .expect("responses lock")
+            .pop_front()
+            .expect("scripted response");
+        Ok(Box::pin(stream::iter(response.into_iter().map(Ok))))
+    }
+
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        Box::new(self.clone())
+    }
+}
+
+#[tokio::test]
+async fn in_process_host_runs_text_and_read_tool_then_shuts_down() {
+    let temp = tempfile::tempdir().expect("tempdir");
+    std::fs::write(temp.path().join("probe.txt"), "standalone tool evidence\n")
+        .expect("write probe");
+    let worker_name = format!("standalone-{}", Uuid::now_v7());
+    let launch = StandaloneLaunchConfig::new(
+        temp.path(),
+        temp.path().join("state"),
+        manifest::ProfileSelector::Default,
+        &worker_name,
+    )
+    .resolve()
+    .expect("resolve standalone profile");
+
+    let client = ScriptedClient::new(vec![
+        vec![
+            LlmEvent::tool_use_start(0, "read-1", "Read"),
+            LlmEvent::tool_input_delta(0, r#"{"file_path":"probe.txt"}"#),
+            LlmEvent::tool_use_stop(0),
+        ],
+        vec![
+            LlmEvent::text_block_start(0),
+            LlmEvent::text_delta(0, "standalone response"),
+            LlmEvent::text_block_stop(0, Some(StopReason::EndTurn)),
+        ],
+    ]);
+    let inspection = client.clone();
+    let host = StandaloneHost::start_with_model_client(launch, client)
+        .await
+        .expect("start in-process host");
+    let mut events = host.subscribe();
+
+    host.send(Method::run_text("read the probe"))
+        .await
+        .expect("submit input");
+
+    tokio::time::timeout(Duration::from_secs(30), async {
+        let mut saw_text = false;
+        let mut saw_tool_result = false;
+        loop {
+            match events.recv().await.expect("worker event") {
+                Event::TextDelta { text } if text.contains("standalone response") => {
+                    saw_text = true;
+                }
+                Event::ToolResult { .. } => {
+                    saw_tool_result = true;
+                }
+                Event::RunEnd { .. } => {
+                    assert!(saw_text, "stream must expose the model text delta");
+                    assert!(saw_tool_result, "stream must expose the tool result");
+                    break;
+                }
+                _ => {}
+            }
+        }
+    })
+    .await
+    .expect("run completed");
+
+    let requests = inspection.requests();
+    assert_eq!(requests.len(), 2);
+    let tool_names = requests[0]
+        .tools
+        .iter()
+        .map(|tool| tool.name.as_str())
+        .collect::<Vec<_>>();
+    assert!(tool_names.contains(&"Read"));
+    assert!(tool_names.contains(&"TaskCreate"));
+    assert!(tool_names.contains(&"SubWorkerSpawn"));
+    assert!(format!("{:?}", requests[1].items).contains("standalone tool evidence"));
+    assert!(
+        !temp
+            .path()
+            .join("state/runtime")
+            .join(&worker_name)
+            .join("worker.sock")
+            .exists()
+    );
+
+    host.shutdown().await.expect("graceful shutdown");
+}
+
+#[tokio::test]
+async fn state_store_failure_is_redacted_and_starts_no_controller() {
+    let temp = tempfile::tempdir().expect("tempdir");
+    let state_path = temp.path().join("state-file-with-secret-name");
+    std::fs::write(&state_path, "not a directory").expect("write blocking file");
+    let launch = StandaloneLaunchConfig::new(
+        temp.path(),
+        &state_path,
+        manifest::ProfileSelector::Default,
+        format!("standalone-failure-{}", Uuid::now_v7()),
+    )
+    .resolve()
+    .expect("resolve launch");
+    let client = ScriptedClient::new(Vec::new());
+
+    let error = StandaloneHost::start_with_model_client(launch, client)
+        .await
+        .err()
+        .expect("state store startup rejected");
+    assert_eq!(error, standalone::StandaloneStartupError::StateStore);
+    assert_eq!(
+        error.to_string(),
+        "the standalone state store could not be opened"
+    );
+    assert!(!error.to_string().contains("secret-name"));
+    assert!(
+        !temp
+            .path()
+            .join("state-file-with-secret-name/runtime")
+            .exists()
+    );
+}
+
+#[test]
+fn standalone_crate_has_no_tui_runtime_or_workspace_server_dependency() {
+    let manifest = include_str!("../Cargo.toml");
+    let dependencies = manifest
+        .split("[dependencies]")
+        .nth(1)
+        .expect("dependencies section")
+        .split("[dev-dependencies]")
+        .next()
+        .expect("dependency body");
+    for forbidden in ["tui", "worker-runtime", "yoi-workspace-server"] {
+        assert!(
+            !dependencies.lines().any(|line| {
+                line.split_once('=')
+                    .is_some_and(|(name, _)| name.trim() == forbidden)
+            }),
+            "standalone must not depend on {forbidden}"
+        );
+    }
+}
+
+#[test]
+fn launch_rejects_path_profile_before_worker_startup() {
+    let temp = tempfile::tempdir().expect("tempdir");
+    let error = StandaloneLaunchConfig::new(
+        temp.path(),
+        temp.path().join("state"),
+        manifest::ProfileSelector::Path {
+            path: temp.path().join("profile.dcdl"),
+        },
+        "standalone-path-profile",
+    )
+    .resolve()
+    .err()
+    .expect("path profile rejected");
+    assert_eq!(
+        error,
+        standalone::StandaloneLaunchError::PathProfileUnsupported
+    );
+}