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

merge: orchestration

Browse Source
This commit is contained in:
Hare
2026-07-11 13:44:15 +09:00
parent 6823f0e72d 7d3b364728
commit b6bb22f956
32 changed files with 1796 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/session-store/src/worker_metadata.rs
+49
View File
@@ -100,8 +100,13 @@ pub struct WorkerMetadata {
pub worker_name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub active: Option<WorkerActiveSegmentRef>,
/// Legacy local path hint retained for host/runtime compatibility. It is not
/// Worker workspace identity or authority.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workspace_root: Option<PathBuf>,
/// Path-free workspace identity supplied by the host/runtime boundary.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workspace_id: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub spawned_children: Vec<WorkerSpawnedChild>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
@@ -119,6 +124,7 @@ impl WorkerMetadata {
worker_name: worker_name.into(),
active,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: Vec::new(),
@@ -130,6 +136,11 @@ impl WorkerMetadata {
self.workspace_root = Some(workspace_root);
self
}
pub fn with_workspace_id(mut self, workspace_id: impl Into<String>) -> Self {
self.workspace_id = Some(workspace_id.into());
self
}
}
/// Sync persistence backend for Worker metadata.
@@ -180,6 +191,44 @@ pub trait WorkerMetadataStore: Send + Sync {
}
/// Set the active pointer and workspace ownership while preserving unrelated fields.
fn set_active_with_workspace_context(
&self,
worker_name: &str,
active: Option<WorkerActiveSegmentRef>,
resolved_manifest_snapshot: Option<serde_json::Value>,
workspace_id: Option<String>,
workspace_root: Option<PathBuf>,
) -> Result<WorkerMetadata, WorkerStoreError> {
self.update_by_name(worker_name, |metadata| {
metadata.active = active;
metadata.resolved_manifest_snapshot = resolved_manifest_snapshot;
if let Some(workspace_id) = workspace_id {
metadata.workspace_id = Some(workspace_id);
}
if let Some(workspace_root) = workspace_root {
metadata.workspace_root = Some(workspace_root);
}
})
}
/// Set the active pointer and path-free workspace identity while preserving unrelated fields.
fn set_active_with_workspace_id(
&self,
worker_name: &str,
active: Option<WorkerActiveSegmentRef>,
resolved_manifest_snapshot: Option<serde_json::Value>,
workspace_id: Option<String>,
) -> Result<WorkerMetadata, WorkerStoreError> {
self.update_by_name(worker_name, |metadata| {
metadata.active = active;
metadata.resolved_manifest_snapshot = resolved_manifest_snapshot;
if let Some(workspace_id) = workspace_id {
metadata.workspace_id = Some(workspace_id);
}
})
}
/// Set the active pointer and legacy local workspace-root hint while preserving unrelated fields.
fn set_active_with_workspace_root(
&self,
worker_name: &str,
crates/worker-runtime/Cargo.toml
+1
View File
@@ -32,6 +32,7 @@ reqwest = { version = "0.13", optional = true, default-features = false, feature
tar.workspace = true
thiserror = { workspace = true }
tokio = { workspace = true, features = ["net", "rt", "sync", "time"] }
toml.workspace = true
tower = { workspace = true, features = ["util"], optional = true }
worker.workspace = true
crates/worker-runtime/src/worker_backend.rs
+121 -16
View File
@@ -9,7 +9,7 @@
use std::collections::HashMap;
use std::future::Future;
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::time::Duration;
@@ -35,7 +35,10 @@ use session_store::{CombinedStore, FsWorkerStore};
use tokio::runtime::Runtime;
use tokio::sync::broadcast;
use worker::{Worker, WorkerController, WorkerHandle};
use worker::{
Worker, WorkerController, WorkerFilesystemAuthority, WorkerHandle, WorkerWorkspaceContext,
WorkspaceId,
};
const DEFAULT_BACKEND_ID: &str = "worker-crate";
const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10);
@@ -230,6 +233,41 @@ fn sanitize_worker_name_component(value: &str) -> String {
.collect()
}
#[derive(Debug, Clone)]
enum RuntimeWorkspaceBackendRef {
None,
LocalFilesystem { root: PathBuf },
}
impl RuntimeWorkspaceBackendRef {
fn from_working_directory(binding: Option<&WorkingDirectoryBinding>) -> Self {
match binding {
Some(binding) => Self::LocalFilesystem {
root: binding.root().to_path_buf(),
},
None => Self::None,
}
}
fn worker_context(&self) -> WorkerWorkspaceContext {
match self {
Self::None => WorkerWorkspaceContext::no_workspace(),
Self::LocalFilesystem { root } => local_workspace_context(root),
}
}
}
fn local_workspace_context(root: &Path) -> WorkerWorkspaceContext {
WorkerWorkspaceContext::local_filesystem(read_workspace_id_hint(root))
}
fn read_workspace_id_hint(root: &Path) -> Option<WorkspaceId> {
let contents = std::fs::read_to_string(root.join(".yoi/workspace.toml")).ok()?;
let value = toml::from_str::<toml::Value>(&contents).ok()?;
let id = value.get("id")?.as_str()?.to_string();
WorkspaceId::new(id).ok()
}
#[cfg(feature = "http-server")]
async fn fetch_profile_source_archive_http(
location: &ProfileSourceArchiveHttpRef,
@@ -298,11 +336,19 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
.as_ref()
.map(|binding| binding.root().to_path_buf())
.unwrap_or_else(|| self.profile_base_dir.clone());
let cwd = request
let filesystem_authority = request
.working_directory
.as_ref()
.map(|binding| binding.cwd().to_path_buf())
.unwrap_or_else(|| self.cwd.clone());
.map(|binding| {
WorkerFilesystemAuthority::local(
binding.root().to_path_buf(),
binding.cwd().to_path_buf(),
)
})
.unwrap_or(WorkerFilesystemAuthority::None);
let workspace_backend_ref =
RuntimeWorkspaceBackendRef::from_working_directory(request.working_directory.as_ref());
let workspace_context = workspace_backend_ref.worker_context();
let selector = profile.as_deref().unwrap_or("builtin:default");
let archive = self
.resolve_profile_source_archive(&request.request.profile_source)
@@ -335,9 +381,15 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
})?;
let store = CombinedStore::new(session_store, worker_metadata_store);
let worker = Worker::from_manifest_with_context(manifest, store, loader, worker_root, cwd)
.await
.map_err(|err| format!("failed to create Worker from profile: {err}"))?;
let worker = Worker::from_manifest_with_context(
manifest,
store,
loader,
workspace_context,
filesystem_authority,
)
.await
.map_err(|err| format!("failed to create Worker from profile: {err}"))?;
let runtime_base = self.runtime_base_dir()?;
let (handle, _shutdown_rx) = WorkerController::spawn(worker, &runtime_base)
@@ -806,7 +858,7 @@ where
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::pin::Pin;
use std::process::Command;
@@ -902,18 +954,31 @@ mod tests {
FsStore::new(&self.store_dir).map_err(|err| err.to_string())?,
FsWorkerStore::new(&self.worker_metadata_dir).map_err(|err| err.to_string())?,
);
let cwd = request
let filesystem_authority = request
.working_directory
.as_ref()
.map(|binding| binding.cwd().to_path_buf())
.map(|binding| {
let cwd = binding.cwd().to_path_buf();
self.observed_cwds.lock().unwrap().push(cwd.clone());
WorkerFilesystemAuthority::local(binding.root().to_path_buf(), cwd)
})
.unwrap_or(WorkerFilesystemAuthority::None);
let scope_root = request
.working_directory
.as_ref()
.map(|binding| binding.root().to_path_buf())
.unwrap_or_else(|| self.cwd.clone());
self.observed_cwds.lock().unwrap().push(cwd.clone());
let scope = Scope::writable(&cwd).map_err(|err| err.to_string())?;
let workspace_backend_ref = RuntimeWorkspaceBackendRef::from_working_directory(
request.working_directory.as_ref(),
);
let workspace_context = workspace_backend_ref.worker_context();
let scope = Scope::writable(&scope_root).map_err(|err| err.to_string())?;
let worker = Worker::new(
manifest,
Engine::new(self.client.clone()),
store,
cwd,
workspace_context,
filesystem_authority,
scope,
)
.await
@@ -925,6 +990,20 @@ mod tests {
}
}
fn core_filesystem_tool_names() -> BTreeSet<&'static str> {
["Read", "Write", "Edit", "Glob", "Grep", "Bash"]
.into_iter()
.collect()
}
fn captured_tool_names(client: &MockClient, index: usize) -> BTreeSet<String> {
client.captured.lock().unwrap()[index]
.tools
.iter()
.map(|tool| tool.name.clone())
.collect()
}
fn simple_text_events() -> Vec<LlmEvent> {
vec![
LlmEvent::text_block_start(0),
@@ -1109,7 +1188,7 @@ mod tests {
cwd: cwd.path().to_path_buf(),
store_dir: store.path().join("sessions"),
worker_metadata_dir: store.path().join("workers"),
observed_cwds,
observed_cwds: observed_cwds.clone(),
};
let backend = WorkerRuntimeExecutionBackend::new(factory).unwrap();
let runtime = EmbeddedRuntime::with_execution_backend(
@@ -1148,6 +1227,14 @@ mod tests {
}
assert_eq!(client.captured.lock().unwrap().len(), 1);
assert!(observed_cwds.lock().unwrap().is_empty());
let names = captured_tool_names(&client, 0);
for forbidden in core_filesystem_tool_names() {
assert!(
!names.contains(forbidden),
"no-workdir Worker unexpectedly exposed {forbidden}; tools={names:?}"
);
}
let observations = runtime
.read_worker_observation_events(&detail.worker_ref, WorkerObservationCursor::zero())
.unwrap();
@@ -1166,7 +1253,7 @@ mod tests {
let store = tempfile::tempdir().unwrap();
let observed_cwds = Arc::new(Mutex::new(Vec::new()));
let factory = MockFactory {
client,
client: client.clone(),
runtime_base: runtime_base.path().to_path_buf(),
cwd: repo.path().to_path_buf(),
store_dir: store.path().join("sessions"),
@@ -1191,6 +1278,24 @@ mod tests {
request.working_directory_request = Some(working_directory_request(repo.path()));
let detail = runtime.create_worker(request).unwrap();
runtime
.send_input(&detail.worker_ref, WorkerInput::user("inspect tools"))
.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(5);
while client.captured.lock().unwrap().is_empty() {
assert!(
std::time::Instant::now() < deadline,
"timed out waiting for materialized-worker request"
);
std::thread::sleep(Duration::from_millis(20));
}
let names = captured_tool_names(&client, 0);
for expected in core_filesystem_tool_names() {
assert!(
names.contains(expected),
"local Worker did not expose {expected}; tools={names:?}"
);
}
assert!(detail.execution.working_directory.is_some());
let cwds = observed_cwds.lock().unwrap();
crates/worker/src/controller.rs
+84 -40
View File
@@ -274,7 +274,9 @@ impl WorkerController {
manifest_toml.clone(),
greeting,
));
shared_state.set_fs_view(crate::fs_view::WorkerFsView::new(fs_for_view));
if let Some(fs_for_view) = fs_for_view {
shared_state.set_fs_view(crate::fs_view::WorkerFsView::new(fs_for_view));
}
shared_state.set_workflows(
worker
.workflow_completions()
@@ -520,15 +522,17 @@ fn install_ticket_event_companion_notify_hook<C, St>(
return;
}
let Some(companion_worker_name) = companion_worker_name_for_workspace(worker.workspace_root())
else {
let Some(local) = worker.local_working_directory() else {
return;
};
let Some(companion_worker_name) = companion_worker_name_for_workspace(&local.root) else {
return;
};
if companion_worker_name == worker.manifest().worker.name {
return;
}
let Ok(ticket_config) = TicketConfig::load_workspace(worker.cwd()) else {
let Ok(ticket_config) = TicketConfig::load_workspace(&local.cwd) else {
return;
};
let backend_root = ticket_config.backend_root().to_path_buf();
@@ -540,7 +544,7 @@ fn install_ticket_event_companion_notify_hook<C, St>(
worker.worker_metadata_store(),
worker.manifest().worker.name.clone(),
runtime_base,
worker.cwd().to_path_buf(),
Some(local.cwd.clone()),
spawned_registry,
);
match discovery.ensure_existing_peer(&companion_worker_name) {
@@ -589,7 +593,7 @@ async fn register_worker_tools<C, St>(
spawner_socket: PathBuf,
runtime_base: PathBuf,
spawned_registry: Arc<SpawnedWorkerRegistry>,
) -> std::io::Result<tools::ScopedFs>
) -> std::io::Result<Option<tools::ScopedFs>>
where
C: LlmClient + Clone + 'static,
St: Store + WorkerMetadataStore + Clone + 'static,
@@ -597,8 +601,8 @@ where
// Worker-immutable snapshots taken before the mutable worker borrow
// below so the worker borrow doesn't conflict with reads on `worker`.
let scope_handle = worker.scope().clone();
let cwd = worker.cwd().to_path_buf();
let workspace_root = worker.workspace_root().to_path_buf();
let local_filesystem = worker.local_working_directory().cloned();
let local_workspace_root = local_filesystem.as_ref().map(|local| local.root.clone());
let task_feature = worker.task_feature();
let session_id_for_usage = worker.segment_id().to_string();
let memory_config = worker.manifest().memory.clone();
@@ -611,24 +615,24 @@ where
let worker_metadata_store = worker.store().clone();
let self_parent_socket = worker.callback_socket().cloned();
// The Worker's SharedScope (already augmented with the bash-output
// Read rule by the caller) is the single source of truth — every
// ScopedFs (builtin tools, fs_view, compact worker) reads from it,
// and any future scope mutation (SpawnWorker-style revoke, future
// GrantScope) propagates through it.
let fs = tools::ScopedFs::with_shared_scope(scope_handle.clone(), cwd.clone());
let tracker = tools::Tracker::new();
// Same ScopedFs also powers the IPC `ListCompletions` query — keep
// a clone for the FS view we attach below, since the tools consume
// `fs` itself.
let fs_for_view = fs.clone();
worker
.engine_mut()
.register_tools(tools::core_builtin_tools(
fs,
tracker.clone(),
bash_output_dir,
));
// The Worker's SharedScope is the single source of truth for every
// ScopedFs when local filesystem authority exists. No-workdir Workers
// deliberately skip constructing/registering filesystem and Bash tools.
let (fs_for_view, tracker) = if let Some(local) = local_filesystem.as_ref() {
let fs = tools::ScopedFs::with_shared_scope(scope_handle.clone(), local.cwd.clone());
let tracker = tools::Tracker::new();
let fs_for_view = fs.clone();
worker
.engine_mut()
.register_tools(tools::core_builtin_tools(
fs,
tracker.clone(),
bash_output_dir,
));
(Some(fs_for_view), Some(tracker))
} else {
(None, None)
};
if feature_config.web.enabled {
worker
.engine_mut()
@@ -649,11 +653,20 @@ where
}
};
// Ticket tools are typed operations over the currently checked-out work
// tree. Use the Worker cwd rather than the runtime workspace root so a
// dedicated Orchestrator worktree gets its own `.yoi/tickets` backend.
// tree. They require explicit local filesystem authority and must not
// use workspace identity as a cwd fallback.
let ticket_cwd = local_filesystem
.as_ref()
.map(|local| &local.cwd)
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"ticket tools require local Worker filesystem authority",
)
})?;
feature_registry.add_module(
crate::feature::builtin::ticket::ticket_tools_feature_with_options(
&cwd,
ticket_cwd,
feature_config.ticket.enabled.then_some(ticket_access),
feature_config.ticket_orchestration.enabled,
),
@@ -665,10 +678,12 @@ where
) {
feature_registry = feature_registry.with_module(module);
}
if let Some(module) =
crate::feature::mcp::discover_stdio_tool_feature(&mcp_config, &workspace_root).await
{
feature_registry = feature_registry.with_module(module);
if let Some(workspace_root) = local_workspace_root.as_ref() {
if let Some(module) =
crate::feature::mcp::discover_stdio_tool_feature(&mcp_config, workspace_root).await
{
feature_registry = feature_registry.with_module(module);
}
}
{
@@ -684,7 +699,13 @@ where
"[feature.memory].enabled = true requires a [memory] configuration section",
)
})?;
let layout = memory::WorkspaceLayout::resolve(mem, &workspace_root);
let workspace_root = local_workspace_root.as_ref().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"memory tools require local Worker filesystem authority",
)
})?;
let layout = memory::WorkspaceLayout::resolve(mem, workspace_root);
let query_cfg = memory::tool::QueryConfig::from(mem);
worker.register_tool(memory::tool::read_tool_with_usage(
layout.clone(),
@@ -709,12 +730,27 @@ where
"[feature.workers].enabled = true requires non-empty [[delegation_scope.allow]]",
));
}
let spawner_cwd = local_filesystem
.as_ref()
.map(|local| local.cwd.clone())
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"worker spawn tools require local Worker filesystem authority",
)
})?;
let spawner_workspace_root = local_workspace_root.clone().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"worker spawn tools require local Worker filesystem authority",
)
})?;
worker.register_tool(spawn_worker_tool(
spawner_name.clone(),
spawner_socket,
runtime_base.clone(),
workspace_root.clone(),
cwd.clone(),
spawner_workspace_root,
spawner_cwd.clone(),
spawned_registry.clone(),
self_parent_socket,
spawner_manifest,
@@ -728,7 +764,7 @@ where
worker_metadata_store,
spawner_name,
runtime_base,
cwd,
Some(spawner_cwd),
spawned_registry,
);
worker.register_tool(list_workers_tool(discovery.clone()));
@@ -737,7 +773,9 @@ where
}
}
let _feature_install_report = worker.install_features(feature_registry);
worker.attach_tracker(tracker);
if let Some(tracker) = tracker {
worker.attach_tracker(tracker);
}
Ok(fs_for_view)
}
@@ -774,11 +812,14 @@ async fn controller_loop<C, St>(
.parent()
.map(PathBuf::from)
.unwrap_or_else(|| runtime_dir.path().to_path_buf());
let discovery_cwd = worker
.local_working_directory()
.map(|local| local.cwd.clone());
let discovery = WorkerDiscovery::new(
worker.store().clone(),
spawner_name.clone(),
discovery_runtime_base,
worker.cwd().to_path_buf(),
discovery_cwd,
spawned_registry.clone(),
);
let mut pending: Option<PendingRun> = None;
@@ -1445,7 +1486,10 @@ where
.collect();
protocol::Greeting {
worker_name: manifest.worker.name.clone(),
cwd: worker.cwd().display().to_string(),
cwd: worker
.local_working_directory()
.map(|local| local.cwd.display().to_string())
.unwrap_or_default(),
provider: provider_name,
model: model_id,
scope_summary: worker.scope_snapshot().summary(),
crates/worker/src/discovery.rs
+28 -10
View File
@@ -42,7 +42,7 @@ pub struct WorkerDiscovery<St> {
store: St,
self_worker_name: String,
runtime_base: PathBuf,
cwd: PathBuf,
cwd: Option<PathBuf>,
store_dir: Option<PathBuf>,
spawned_registry: Arc<SpawnedWorkerRegistry>,
}
@@ -55,7 +55,7 @@ where
store: St,
self_worker_name: String,
runtime_base: PathBuf,
cwd: PathBuf,
cwd: Option<PathBuf>,
spawned_registry: Arc<SpawnedWorkerRegistry>,
) -> Self {
let store_dir = store.root_dir();
@@ -432,13 +432,19 @@ where
) -> Result<(), WorkerDiscoveryError> {
let runtime_command =
WorkerRuntimeCommand::resolve().map_err(WorkerDiscoveryError::RestoreSpawn)?;
let Some(cwd) = &self.cwd else {
return Err(WorkerDiscoveryError::NotRestorable {
worker_name: worker_name.to_string(),
reason: "restore requires local Worker filesystem authority".into(),
});
};
let mut command = Command::new(runtime_command.program());
command
.args(runtime_command.prefix_args())
.arg("--worker")
.arg(worker_name)
.arg("--require-worker-state")
.current_dir(&self.cwd)
.current_dir(cwd)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
@@ -1139,6 +1145,7 @@ mod tests {
worker_name: "parent".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: vec![
child("child-live", &live_socket),
child("child-stale", &stale_socket),
@@ -1159,6 +1166,7 @@ mod tests {
active_child_segment,
)),
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: Vec::new(),
@@ -1173,6 +1181,7 @@ mod tests {
active_child_segment,
)),
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: Vec::new(),
@@ -1184,6 +1193,7 @@ mod tests {
worker_name: "child-pending".into(),
active: Some(WorkerActiveSegmentRef::pending_segment(pending_session_id)),
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: Vec::new(),
@@ -1198,6 +1208,7 @@ mod tests {
new_segment_id(),
)),
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: Vec::new(),
@@ -1209,6 +1220,7 @@ mod tests {
worker_name: "peer".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: vec![session_store::WorkerPeer {
@@ -1228,7 +1240,7 @@ mod tests {
store.clone(),
"parent".into(),
runtime_base.clone(),
root.path().to_path_buf(),
Some(root.path().to_path_buf()),
registry,
);
@@ -1355,7 +1367,7 @@ mod tests {
store.clone(),
"source".into(),
runtime_base.clone(),
root.path().to_path_buf(),
Some(root.path().to_path_buf()),
SpawnedWorkerRegistry::new(runtime_dir),
);
let result = discovery.register_peer("target").unwrap();
@@ -1390,7 +1402,7 @@ mod tests {
store,
"source".into(),
runtime_base,
root.path().to_path_buf(),
Some(root.path().to_path_buf()),
SpawnedWorkerRegistry::new(runtime_dir),
);
@@ -1415,6 +1427,7 @@ mod tests {
worker_name: "source".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: vec![session_store::WorkerPeer {
@@ -1430,7 +1443,7 @@ mod tests {
store.clone(),
"source".into(),
runtime_base,
root.path().to_path_buf(),
Some(root.path().to_path_buf()),
SpawnedWorkerRegistry::new(runtime_dir),
);
@@ -1455,6 +1468,7 @@ mod tests {
worker_name: "source".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: vec![session_store::WorkerPeer {
@@ -1468,6 +1482,7 @@ mod tests {
worker_name: "target".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: vec![session_store::WorkerPeer {
@@ -1481,7 +1496,7 @@ mod tests {
store,
"source".into(),
runtime_base.clone(),
root.path().to_path_buf(),
Some(root.path().to_path_buf()),
SpawnedWorkerRegistry::new(runtime_dir),
);
@@ -1573,6 +1588,7 @@ mod tests {
worker_name: "source".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: vec![session_store::WorkerPeer {
@@ -1586,6 +1602,7 @@ mod tests {
worker_name: "target".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: vec![session_store::WorkerPeer {
@@ -1599,7 +1616,7 @@ mod tests {
store,
"source".into(),
runtime_base.clone(),
root.path().to_path_buf(),
Some(root.path().to_path_buf()),
SpawnedWorkerRegistry::new(runtime_dir),
);
@@ -1689,6 +1706,7 @@ mod tests {
worker_name: "source".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: vec![child("target", &socket)],
reclaimed_children: Vec::new(),
peers: Vec::new(),
@@ -1701,7 +1719,7 @@ mod tests {
store,
"source".into(),
runtime_base,
root.path().to_path_buf(),
Some(root.path().to_path_buf()),
SpawnedWorkerRegistry::new(runtime_dir),
);
crates/worker/src/entrypoint.rs
+35 -11
View File
@@ -2,7 +2,10 @@ use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use crate::{PromptLoader, Worker, WorkerController};
use crate::{
PromptLoader, Worker, WorkerController, WorkerFilesystemAuthority, WorkerWorkspaceContext,
WorkspaceId,
};
use clap::{CommandFactory, FromArgMatches, Parser};
use manifest::{
Permission, ProfileResolveOptions, ProfileResolver, ProfileSelector, ScopeConfig, ScopeRule,
@@ -103,6 +106,24 @@ fn runtime_workspace_root(cli: &Cli) -> Result<PathBuf, String> {
}
}
fn runtime_workspace_context(workspace_root: &Path) -> WorkerWorkspaceContext {
WorkerWorkspaceContext::local_filesystem(read_workspace_id_hint(workspace_root))
}
fn read_workspace_id_hint(workspace_root: &Path) -> Option<WorkspaceId> {
let path = workspace_root.join(".yoi/workspace.toml");
let contents = std::fs::read_to_string(path).ok()?;
let value = toml::from_str::<toml::Value>(&contents).ok()?;
let id = value.get("id")?.as_str()?.to_string();
match WorkspaceId::new(id) {
Ok(id) => Some(id),
Err(err) => {
tracing::warn!("ignoring invalid workspace id in .yoi/workspace.toml: {err}");
None
}
}
}
fn runtime_worker_name(cli: &Cli, workspace_root: &Path) -> String {
cli.worker
.as_deref()
@@ -511,6 +532,9 @@ async fn run_cli_inner(cli: Cli) -> ExitCode {
}
};
let store = CombinedStore::new(session_store, worker_metadata_store);
let filesystem_authority =
WorkerFilesystemAuthority::local(workspace_root.clone(), cwd.clone());
let workspace_context = runtime_workspace_context(&workspace_root);
let mut worker = if cli.adopt {
let callback = match cli.callback.clone() {
@@ -525,8 +549,8 @@ async fn run_cli_inner(cli: Cli) -> ExitCode {
store,
loader,
callback,
workspace_root.clone(),
cwd.clone(),
workspace_context.clone(),
filesystem_authority.clone(),
)
.await
{
@@ -556,8 +580,8 @@ async fn run_cli_inner(cli: Cli) -> ExitCode {
manifest,
store,
loader,
workspace_root.clone(),
cwd.clone(),
workspace_context.clone(),
filesystem_authority.clone(),
)
.await
{
@@ -576,8 +600,8 @@ async fn run_cli_inner(cli: Cli) -> ExitCode {
manifest,
store,
loader,
workspace_root.clone(),
cwd.clone(),
workspace_context.clone(),
filesystem_authority.clone(),
)
.await
{
@@ -597,8 +621,8 @@ async fn run_cli_inner(cli: Cli) -> ExitCode {
manifest,
store,
loader,
workspace_root.clone(),
cwd.clone(),
workspace_context.clone(),
filesystem_authority.clone(),
)
.await
{
@@ -619,8 +643,8 @@ async fn run_cli_inner(cli: Cli) -> ExitCode {
manifest,
store,
loader,
workspace_root.clone(),
cwd.clone(),
workspace_context.clone(),
filesystem_authority.clone(),
)
.await
{
crates/worker/src/lib.rs
+4 -1
View File
@@ -38,4 +38,7 @@ pub use provider::{ProviderError, build_client};
pub use runtime::dir::RuntimeDir;
pub use segment_log_sink::SegmentLogSink;
pub use shared_state::WorkerSharedState;
pub use worker::{Worker, WorkerError, WorkerRunResult, apply_worker_manifest};
pub use worker::{
LocalWorkingDirectory, Worker, WorkerError, WorkerFilesystemAuthority, WorkerRunResult,
WorkerWorkspaceContext, WorkspaceClient, WorkspaceId, WorkspaceIdError, apply_worker_manifest,
};
crates/worker/src/prompt/system.rs
+8 -6
View File
@@ -13,7 +13,9 @@
//! `set_system_prompt`. Subsequent turns and compactions reuse that
//! materialised string verbatim.
use std::borrow::Cow;
use std::collections::BTreeMap;
#[cfg(test)]
use std::path::Path;
use std::sync::Arc;
@@ -146,7 +148,7 @@ impl std::fmt::Debug for SystemPromptTemplate {
/// templates cannot drop them on the floor.
pub struct SystemPromptContext<'a> {
pub now: DateTime<Utc>,
pub cwd: &'a Path,
pub cwd: Cow<'a, str>,
/// Language policy exposed to instruction templates as `{{ language }}`.
pub language: &'a str,
pub scope: &'a Scope,
@@ -189,7 +191,7 @@ impl<'a> SystemPromptContext<'a> {
"datetime".into(),
Value::from(self.now.to_rfc3339_opts(SecondsFormat::Secs, true)),
);
root.insert("cwd".into(), Value::from(self.cwd.display().to_string()));
root.insert("cwd".into(), Value::from(self.cwd.as_ref()));
root.insert("language".into(), Value::from(self.language));
root.insert(
"tools".into(),
@@ -442,7 +444,7 @@ mod tests {
) -> SystemPromptContext<'a> {
SystemPromptContext {
now: fixed_now(),
cwd,
cwd: cwd.display().to_string().into(),
language: manifest::defaults::WORKER_LANGUAGE,
scope,
tool_names: tools,
@@ -461,7 +463,7 @@ mod tests {
) -> SystemPromptContext<'a> {
SystemPromptContext {
now: fixed_now(),
cwd,
cwd: cwd.display().to_string().into(),
language: manifest::defaults::WORKER_LANGUAGE,
scope,
tool_names: Vec::new(),
@@ -480,7 +482,7 @@ mod tests {
) -> SystemPromptContext<'a> {
SystemPromptContext {
now: fixed_now(),
cwd,
cwd: cwd.display().to_string().into(),
language: manifest::defaults::WORKER_LANGUAGE,
scope,
tool_names: Vec::new(),
@@ -499,7 +501,7 @@ mod tests {
) -> SystemPromptContext<'a> {
SystemPromptContext {
now: fixed_now(),
cwd,
cwd: cwd.display().to_string().into(),
language: manifest::defaults::WORKER_LANGUAGE,
scope,
tool_names: Vec::new(),
crates/worker/src/ticket_event_notify.rs
+3 -1
View File
@@ -384,6 +384,7 @@ mod tests {
worker_name: "orchestrator".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: Vec::new(),
@@ -395,6 +396,7 @@ mod tests {
worker_name: "companion".into(),
active: None,
workspace_root: None,
workspace_id: None,
spawned_children: Vec::new(),
reclaimed_children: Vec::new(),
peers: Vec::new(),
@@ -414,7 +416,7 @@ mod tests {
store,
"orchestrator".into(),
runtime_base.clone(),
root.path().to_path_buf(),
Some(root.path().to_path_buf()),
SpawnedWorkerRegistry::new(runtime_dir),
),
"companion",
crates/worker/src/worker.rs
+462 -150
View File
File diff suppressed because it is too large Load Diff
crates/worker/tests/compact_events_test.rs
+10 -3
View File
@@ -164,9 +164,16 @@ async fn make_worker_with_manifest(
std::mem::forget(pwd_tmp);
let worker = Engine::new(client);
let mut worker = Worker::new(manifest, worker, store, pwd, scope)
.await
.unwrap();
let mut worker = Worker::new(
manifest,
worker,
store,
worker::WorkerWorkspaceContext::local_filesystem(None),
worker::WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()),
scope,
)
.await
.unwrap();
worker.enable_worker_metadata_write_through().unwrap();
worker
}
crates/worker/tests/consolidation_test.rs
+10 -3
View File
@@ -170,9 +170,16 @@ async fn make_worker_with(
let scope = worker::Scope::writable(&pwd).unwrap();
let worker = Engine::new(client);
Worker::new(manifest, worker, store, pwd, scope)
.await
.unwrap()
Worker::new(
manifest,
worker,
store,
worker::WorkerWorkspaceContext::local_filesystem(None),
worker::WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()),
scope,
)
.await
.unwrap()
}
fn write_n_staging(layout: &WorkspaceLayout, n: usize) -> Vec<uuid::Uuid> {
crates/worker/tests/controller_test.rs
+15 -4
View File
@@ -12,7 +12,10 @@ use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
use session_store::{CombinedStore, FsWorkerStore};
use session_store::{FsStore, LogEntry};
use worker::{Event, Method, Worker, WorkerController, WorkerHandle, WorkerManifest, WorkerStatus};
use worker::{
Event, Method, Worker, WorkerController, WorkerFilesystemAuthority, WorkerHandle,
WorkerManifest, WorkerStatus, WorkerWorkspaceContext,
};
type TestStore = CombinedStore<FsStore, FsWorkerStore>;
@@ -186,9 +189,17 @@ async fn make_worker_with_pwd_and_manifest(
std::mem::forget(pwd_tmp);
let worker = Engine::new(client);
let worker = Worker::new(manifest, worker, store, pwd.clone(), scope)
.await
.unwrap();
let authority = WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone());
let worker = Worker::new(
manifest,
worker,
store,
WorkerWorkspaceContext::local_filesystem(None),
authority,
scope,
)
.await
.unwrap();
(worker, pwd)
}
crates/worker/tests/session_metrics_test.rs
+30 -9
View File
@@ -190,9 +190,16 @@ async fn make_worker(
let mut worker = Engine::new(client);
worker.register_tool(big_content_tool_definition(tool_name));
let worker = Worker::new(manifest, worker, store, pwd, scope)
.await
.unwrap();
let worker = Worker::new(
manifest,
worker,
store,
worker::WorkerWorkspaceContext::local_filesystem(None),
worker::WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()),
scope,
)
.await
.unwrap();
(worker, store_tmp, pwd_tmp)
}
@@ -453,9 +460,16 @@ async fn metric_write_failure_emits_warn_alert_and_does_not_abort_run() {
// the failure path: at least one metric attempts to write.
let client = MockClient::new(vec![text_response_with_cache("hi", 0, 0)]);
let worker = Engine::new(client);
let mut worker = Worker::new(manifest, worker, store.clone(), pwd, scope)
.await
.unwrap();
let mut worker = Worker::new(
manifest,
worker,
store.clone(),
worker::WorkerWorkspaceContext::local_filesystem(None),
worker::WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()),
scope,
)
.await
.unwrap();
let (tx, mut rx) = broadcast::channel::<Event>(64);
let alerter = worker::Alerter::new(tx);
@@ -522,9 +536,16 @@ permission = "write"
let pwd = pwd_tmp.path().to_path_buf();
let scope = worker::Scope::writable(&pwd).unwrap();
let worker = Engine::new(client);
let mut worker = Worker::new(manifest, worker, store.clone(), pwd, scope)
.await
.unwrap();
let mut worker = Worker::new(
manifest,
worker,
store.clone(),
worker::WorkerWorkspaceContext::local_filesystem(None),
worker::WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()),
scope,
)
.await
.unwrap();
let session_id = worker.session_id();
let segment_id = worker.segment_id();
worker.run_text("hello").await.unwrap();
crates/worker/tests/system_prompt_template_test.rs
+9 -1
View File
@@ -123,7 +123,15 @@ async fn make_worker_with_body(
std::mem::forget(user_prompts_tmp);
let worker = Engine::new(client);
let mut worker = Worker::new(manifest, worker, store, pwd.clone(), scope).await?;
let mut worker = Worker::new(
manifest,
worker,
store,
worker::WorkerWorkspaceContext::local_filesystem(None),
worker::WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()),
scope,
)
.await?;
let template = SystemPromptTemplate::parse("$user/test", loader)
.map_err(|source| WorkerError::InvalidSystemPromptTemplate { source })?;
crates/workspace-server/src/server.rs
+74 -19
View File
@@ -752,6 +752,7 @@ pub struct WorkerLaunchRuntimeOption {
pub display_name: String,
pub built_in: bool,
pub can_spawn_worker: bool,
pub working_directory_required: bool,
pub status: String,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
@@ -2640,6 +2641,9 @@ async fn create_workspace_worker(
relative_cwd: selection.relative_cwd,
});
validate_working_directory_claim_for_browser(resolved_working_directory.as_ref())?;
if resolved_working_directory.is_none() {
reject_no_workdir_for_non_embedded_runtime(&request.runtime_id)?;
}
let result = api
.runtime
.spawn_worker(
@@ -2809,11 +2813,36 @@ struct RuntimeConfigBundleAvailabilityQuery {
digest: String,
}
fn reject_no_workdir_for_non_embedded_runtime(
runtime_id: &str,
) -> std::result::Result<(), ApiError> {
if runtime_id == EMBEDDED_WORKER_RUNTIME_ID {
return Ok(());
}
Err(ApiError::with_diagnostics(
Error::RuntimeOperationFailed {
runtime_id: runtime_id.to_string(),
code: "workspace_worker_workdir_required".to_string(),
message: "Only the embedded Runtime can launch a Worker without a working directory"
.to_string(),
},
vec![RuntimeDiagnostic {
code: "workspace_worker_workdir_required".to_string(),
severity: DiagnosticSeverity::Error,
message: "Select a working directory for this Runtime, or choose the embedded Runtime for a conversation-only Worker."
.to_string(),
}],
))
}
async fn create_runtime_worker(
State(api): State<WorkspaceApi>,
AxumPath(runtime_id): AxumPath<String>,
Json(mut request): Json<WorkerSpawnRequest>,
) -> ApiResult<Json<WorkerSpawnResult>> {
if request.working_directory_request.is_none() && request.resolved_working_directory.is_none() {
reject_no_workdir_for_non_embedded_runtime(&runtime_id)?;
}
request.resolved_working_directory_request = request
.working_directory_request
.as_ref()
@@ -3765,6 +3794,7 @@ fn worker_launch_options_response(api: &WorkspaceApi) -> WorkerLaunchOptionsResp
display_name: runtime.label,
built_in,
can_spawn_worker: runtime.capabilities.can_spawn_worker,
working_directory_required: !built_in,
status: runtime.status,
diagnostics: runtime.diagnostics,
}
@@ -4694,6 +4724,7 @@ impl IntoResponse for ApiError {
|| code.starts_with("unsupported_worker_profile")
|| code.starts_with("working_directory_")
|| code.starts_with("workspace_cleanup_")
|| code == "workspace_worker_workdir_required"
|| code.ends_with("_already_exists")
|| code.ends_with("_not_config_managed")
|| code.ends_with("_unsupported") =>
@@ -6034,13 +6065,17 @@ mod tests {
);
let launch_options = get_json(app.clone(), "/api/workers/launch-options").await;
assert!(
launch_options["runtimes"]
.as_array()
.unwrap()
.iter()
.any(|runtime| runtime["runtime_id"] == "team-runtime")
);
let runtimes = launch_options["runtimes"].as_array().unwrap();
let embedded_runtime = runtimes
.iter()
.find(|runtime| runtime["runtime_id"] == EMBEDDED_WORKER_RUNTIME_ID)
.expect("embedded runtime launch option");
assert_eq!(embedded_runtime["working_directory_required"], false);
let team_runtime = runtimes
.iter()
.find(|runtime| runtime["runtime_id"] == "team-runtime")
.expect("team runtime launch option");
assert_eq!(team_runtime["working_directory_required"], true);
let deleted = request_json(
app.clone(),
@@ -6091,18 +6126,6 @@ mod tests {
)
.await;
assert_eq!(added["restart_required"], false);
let created = post_json(
app.clone(),
"/api/workers",
serde_json::json!({
"runtime_id": "busy-runtime",
"display_name": "Remote Test Worker",
"profile": "runtime_default",
"initial_text": ""
}),
)
.await;
assert_eq!(created["runtime_id"], "busy-runtime");
let workers = get_json(app.clone(), "/api/workers").await;
assert!(
workers["items"]
@@ -6306,6 +6329,38 @@ mod tests {
assert!(!projected.contains("http://"));
}
#[tokio::test]
async fn browser_worker_create_rejects_non_embedded_no_workdir() {
let dir = tempfile::tempdir().unwrap();
let app = test_app(dir.path()).await;
let response = request_json(
app,
"POST",
"/api/workers",
Some(serde_json::json!({
"runtime_id": "remote-runtime",
"display_name": "Remote Worker",
"profile": "runtime_default",
"initial_text": ""
})),
StatusCode::BAD_REQUEST,
)
.await;
assert!(
response["message"]
.as_str()
.unwrap_or_default()
.contains("workspace_worker_workdir_required")
);
assert!(
response["diagnostics"]
.as_array()
.unwrap()
.iter()
.any(|diagnostic| { diagnostic["code"] == "workspace_worker_workdir_required" })
);
}
#[tokio::test]
async fn runtime_worker_spawn_rejects_raw_working_directory_fields() {
let dir = tempfile::tempdir().unwrap();