diff --git a/Cargo.lock b/Cargo.lock index 6d3627ed..d2650c4c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6618,6 +6618,7 @@ dependencies = [ "wasmtime", "wat", "workdir", + "workspace-api", "yoi-plugin-pdk", ] diff --git a/crates/fs-operation/src/lib.rs b/crates/fs-operation/src/lib.rs index 53999081..fb2cd54f 100644 --- a/crates/fs-operation/src/lib.rs +++ b/crates/fs-operation/src/lib.rs @@ -157,6 +157,22 @@ mod tests { } } + fn grep_request(path: &str, pattern: &str) -> GrepRequest { + GrepRequest { + pattern: pattern.to_string(), + path: FsPath::new(path).unwrap(), + glob: None, + file_type: None, + case_insensitive: false, + before_context: 0, + after_context: 0, + multiline: false, + output_mode: GrepOutputMode::Content, + limit: 10, + offset: 0, + } + } + #[test] fn logical_paths_reject_absolute_parent_and_backslash_forms() { assert!(FsPath::new("src/lib.rs").is_ok()); @@ -280,6 +296,261 @@ mod tests { assert!(!grep.output.contains("c.txt")); } + #[test] + fn grep_accepts_a_direct_file_without_searching_siblings() { + let temp = tempfile::tempdir().unwrap(); + let selected = temp.path().join("selected.txt"); + std::fs::write(&selected, "before\nneedle selected\nafter\n").unwrap(); + std::fs::write(temp.path().join("sibling.txt"), "needle sibling\n").unwrap(); + let root = temp.path().canonicalize().unwrap(); + let readable = RootAccess(root.clone()); + + let mut request = grep_request("selected.txt", "needle"); + request.before_context = 1; + request.after_context = 1; + let direct = run_grep(&root, selected, request, &readable).unwrap(); + + assert_eq!(direct.match_count, 1); + assert_eq!(direct.matched_files, 1); + assert_eq!( + direct.output, + concat!( + "selected.txt\n", + " 1 │ before\n", + " > 2 │ needle selected\n", + " 3 │ after\n", + ) + ); + assert!(!direct.output.contains("sibling")); + + let directory = run_grep( + &root, + root.clone(), + GrepRequest { + pattern: "needle".to_string(), + path: FsPath::root(), + glob: None, + file_type: None, + case_insensitive: false, + before_context: 0, + after_context: 0, + multiline: false, + output_mode: GrepOutputMode::Content, + limit: 10, + offset: 0, + }, + &readable, + ) + .unwrap(); + assert_eq!(directory.match_count, 2); + assert_eq!(directory.matched_files, 2); + } + + #[test] + fn grep_direct_file_applies_glob_and_type_filters_for_every_output_mode() { + let temp = tempfile::tempdir().unwrap(); + let nested = temp.path().join("nested"); + std::fs::create_dir(&nested).unwrap(); + let selected = nested.join("selected.rs"); + std::fs::write(&selected, "needle one\nneedle two\n").unwrap(); + let root = temp.path().canonicalize().unwrap(); + let readable = RootAccess(root.clone()); + + for mode in [ + GrepOutputMode::Content, + GrepOutputMode::FilesWithMatches, + GrepOutputMode::Count, + ] { + for (glob, file_type) in [(Some("other/*.rs"), None), (None, Some("python"))] { + let mut request = grep_request("nested/selected.rs", "needle"); + request.output_mode = mode; + request.glob = glob.map(str::to_string); + request.file_type = file_type.map(str::to_string); + + let excluded = run_grep(&root, selected.clone(), request, &readable).unwrap(); + assert_eq!(excluded.output, "", "mode {mode:?}"); + assert_eq!(excluded.match_count, 0, "mode {mode:?}"); + assert_eq!(excluded.matched_files, 0, "mode {mode:?}"); + assert!(!excluded.truncated, "mode {mode:?}"); + } + + let mut request = grep_request("nested/selected.rs", "needle"); + request.output_mode = mode; + request.glob = Some("nested/*.rs".to_string()); + request.file_type = Some("rust".to_string()); + let matched = run_grep(&root, selected.clone(), request, &readable).unwrap(); + + match mode { + GrepOutputMode::Content => { + assert_eq!(matched.match_count, 2); + assert_eq!(matched.matched_files, 1); + assert!(matched.output.starts_with("nested/selected.rs\n")); + assert!(matched.output.contains("> 1 │ needle one")); + assert!(matched.output.contains("> 2 │ needle two")); + } + GrepOutputMode::FilesWithMatches => { + assert_eq!(matched.match_count, 1); + assert_eq!(matched.matched_files, 1); + assert_eq!(matched.output, "nested/selected.rs\n"); + } + GrepOutputMode::Count => { + assert_eq!(matched.match_count, 2); + assert_eq!(matched.matched_files, 1); + assert_eq!(matched.output, "nested/selected.rs:2\n"); + } + } + assert!(!matched.truncated, "mode {mode:?}"); + } + } + + #[test] + fn grep_direct_file_preserves_explicit_hidden_and_gitignored_behavior() { + let temp = tempfile::tempdir().unwrap(); + let hidden = temp.path().join(".hidden.rs"); + let ignored = temp.path().join("ignored.rs"); + std::fs::write(&hidden, "needle hidden\n").unwrap(); + std::fs::write(&ignored, "needle ignored\n").unwrap(); + std::fs::write(temp.path().join(".gitignore"), "ignored.rs\n").unwrap(); + let root = temp.path().canonicalize().unwrap(); + let readable = RootAccess(root.clone()); + + for (path, expected) in [ + (".hidden.rs", "needle hidden"), + ("ignored.rs", "needle ignored"), + ] { + let result = run_grep( + &root, + root.join(path), + grep_request(path, "needle"), + &readable, + ) + .unwrap(); + assert_eq!(result.match_count, 1, "path {path}"); + assert!(result.output.contains(expected), "path {path}"); + } + } + + #[test] + fn grep_direct_file_preserves_case_multiline_and_bounds() { + let temp = tempfile::tempdir().unwrap(); + let selected = temp.path().join("selected.txt"); + std::fs::write(&selected, "NEEDLE first\nstart\nfinish\nneedle last\n").unwrap(); + let root = temp.path().canonicalize().unwrap(); + let readable = RootAccess(root.clone()); + + let mut case_request = grep_request("selected.txt", "needle"); + case_request.case_insensitive = true; + case_request.offset = 1; + case_request.limit = 1; + let bounded = run_grep(&root, selected.clone(), case_request, &readable).unwrap(); + assert_eq!(bounded.match_count, 1); + assert!(!bounded.output.contains("NEEDLE first")); + assert!(bounded.output.contains("needle last")); + assert!(bounded.truncated); + + let mut multiline_request = grep_request("selected.txt", "start\\nfinish"); + multiline_request.multiline = true; + let multiline = run_grep(&root, selected, multiline_request, &readable).unwrap(); + assert_eq!(multiline.match_count, 1); + assert!(multiline.output.contains("start\nfinish")); + } + + #[test] + fn grep_returns_not_found_for_a_missing_direct_path() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let missing = root.join("missing.txt"); + let readable = RootAccess(root.clone()); + + let error = run_grep( + &root, + missing.clone(), + grep_request("missing.txt", "needle"), + &readable, + ) + .unwrap_err(); + + assert!(matches!(error, FsError::NotFound(path) if path == missing)); + } + + #[cfg(unix)] + #[test] + fn grep_keeps_direct_symlink_directory_and_broken_path_guards() { + use std::os::unix::fs::symlink; + + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let readable = RootAccess(root.clone()); + std::fs::create_dir(root.join("target-dir")).unwrap(); + std::fs::write(root.join("target-file.rs"), "needle file\n").unwrap(); + symlink(root.join("target-file.rs"), root.join("file-link.rs")).unwrap(); + symlink(root.join("target-dir"), root.join("directory-link")).unwrap(); + symlink(root.join("missing-target"), root.join("broken-link")).unwrap(); + + let request = |path: &str| grep_request(path, "needle"); + + let file_result = run_grep( + &root, + root.join("file-link.rs"), + request("file-link.rs"), + &readable, + ) + .unwrap(); + assert_eq!(file_result.match_count, 1); + assert!(file_result.output.starts_with("file-link.rs\n")); + + let directory_error = run_grep( + &root, + root.join("directory-link"), + request("directory-link"), + &readable, + ) + .unwrap_err(); + assert!(matches!( + directory_error, + FsError::SymlinkDirectoryNotTraversed { tool: "Grep", path, .. } + if path == root.join("directory-link") + )); + + let broken_error = run_grep( + &root, + root.join("broken-link"), + request("broken-link"), + &readable, + ) + .unwrap_err(); + assert!(matches!( + broken_error, + FsError::BrokenSymlink { path, .. } if path == root.join("broken-link") + )); + } + + #[cfg(unix)] + #[test] + fn grep_rejects_a_direct_special_file_as_invalid_argument() { + use std::os::unix::net::UnixListener; + + let temp = tempfile::tempdir().unwrap(); + let socket = temp.path().join("grep.sock"); + let _listener = UnixListener::bind(&socket).unwrap(); + let root = temp.path().canonicalize().unwrap(); + let readable = RootAccess(root.clone()); + + let error = run_grep( + &root, + socket, + grep_request("grep.sock", "needle"), + &readable, + ) + .unwrap_err(); + + assert!(matches!( + error, + FsError::InvalidArgument(message) + if message.contains("must be a regular file or directory") + )); + } + #[test] fn grep_content_groups_lines_by_file_and_marks_matches() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/fs-operation/src/search.rs b/crates/fs-operation/src/search.rs index 54420f6f..a6d9d336 100644 --- a/crates/fs-operation/src/search.rs +++ b/crates/fs-operation/src/search.rs @@ -7,8 +7,8 @@ use grep_regex::RegexMatcherBuilder; use grep_searcher::sinks::UTF8 as UTF8Sink; use grep_searcher::{BinaryDetection, Searcher, SearcherBuilder, Sink, SinkContext, SinkMatch}; use ignore::WalkBuilder; -use ignore::overrides::OverrideBuilder; -use ignore::types::TypesBuilder; +use ignore::overrides::{Override, OverrideBuilder}; +use ignore::types::{Types, TypesBuilder}; use crate::{FsError, GrepOutputMode, GrepRequest, GrepResult, direct_symlink}; @@ -126,6 +126,38 @@ fn logical_display(root: &Path, path: &Path) -> String { const DEFAULT_HEAD_LIMIT: usize = 250; +fn build_overrides(base: &Path, glob: Option<&str>) -> Result, FsError> { + let Some(glob) = glob else { + return Ok(None); + }; + let mut builder = OverrideBuilder::new(base); + builder + .add(glob) + .map_err(|error| FsError::InvalidGlob(error.to_string()))?; + builder + .build() + .map(Some) + .map_err(|error| FsError::InvalidGlob(error.to_string())) +} + +fn build_types(file_type: Option<&str>) -> Result, FsError> { + let Some(file_type) = file_type else { + return Ok(None); + }; + let mut builder = TypesBuilder::new(); + builder.add_defaults(); + builder.select(file_type); + builder + .build() + .map(Some) + .map_err(|error| FsError::InvalidArgument(format!("invalid type {file_type}: {error}"))) +} + +fn direct_file_selected(path: &Path, overrides: Option<&Override>, types: Option<&Types>) -> bool { + !overrides.is_some_and(|filter| filter.matched(path, false).is_ignore()) + && !types.is_some_and(|filter| filter.matched(path, false).is_ignore()) +} + struct GrepParams { pattern: String, path: Option, @@ -221,13 +253,15 @@ pub fn run_grep( std::io::ErrorKind::NotFound => FsError::NotFound(base.clone()), _ => FsError::io(&base, e), })?; - if !base_meta.is_dir() { + if !base_meta.is_file() && !base_meta.is_dir() { return Err(FsError::InvalidArgument(format!( - "grep search path is not a directory: {}", + "grep search path must be a regular file or directory: {}", base.display() ))); } - if let Some(info) = symlink.as_ref() { + if base_meta.is_dir() + && let Some(info) = symlink.as_ref() + { return Err(FsError::SymlinkDirectoryNotTraversed { tool: "Grep", path: base.clone(), @@ -235,32 +269,9 @@ pub fn run_grep( }); } - let mut wb = WalkBuilder::new(&base); - wb.hidden(true) - .git_ignore(true) - .git_global(true) - .git_exclude(true) - .ignore(true) - .parents(true) - .follow_links(false); - - if let Some(t) = p.file_type.as_deref() { - let mut tb = TypesBuilder::new(); - tb.add_defaults(); - tb.select(t); - let types = tb - .build() - .map_err(|e| FsError::InvalidArgument(format!("invalid type {t}: {e}")))?; - wb.types(types); - } - if let Some(g) = p.glob.as_deref() { - let mut ob = OverrideBuilder::new(&base); - ob.add(g).map_err(|e| FsError::InvalidGlob(e.to_string()))?; - let ov = ob - .build() - .map_err(|e| FsError::InvalidGlob(e.to_string()))?; - wb.overrides(ov); - } + let filter_base = if base_meta.is_file() { root } else { &base }; + let types = build_types(p.file_type.as_deref())?; + let overrides = build_overrides(filter_base, p.glob.as_deref())?; let mode = p.output_mode.unwrap_or_default(); let head_limit = p.head_limit.unwrap_or(DEFAULT_HEAD_LIMIT); @@ -275,74 +286,133 @@ pub fn run_grep( lines: Vec::new(), truncated: false, }; + let mut matching_files_seen = 0; + let mut matches_seen = 0; - // Per-mode walker state. - let mut matching_files_seen: usize = 0; - let mut matches_seen: usize = 0; + if base_meta.is_file() { + if direct_file_selected(&base, overrides.as_ref(), types.as_ref()) { + scan_path( + &mut searcher, + &matcher, + &base, + mode, + &mut report, + &mut matching_files_seen, + &mut matches_seen, + offset, + head_limit, + )?; + } + return Ok(report.into_result(root)); + } - 'walker: for entry in wb.build().flatten() { - if !entry.file_type().map(|t| t.is_file()).unwrap_or(false) { + let mut walker = WalkBuilder::new(&base); + walker + .hidden(true) + .git_ignore(true) + .git_global(true) + .git_exclude(true) + .ignore(true) + .parents(true) + .follow_links(false); + if let Some(types) = types { + walker.types(types); + } + if let Some(overrides) = overrides { + walker.overrides(overrides); + } + + for entry in walker.build().flatten() { + if !entry + .file_type() + .map(|kind| kind.is_file()) + .unwrap_or(false) + { continue; } let path = entry.path(); if !access.is_readable(path) { continue; } - - match mode { - GrepOutputMode::FilesWithMatches => { - let hit = scan_any_match(&mut searcher, &matcher, path)?; - if !hit { - continue; - } - if matching_files_seen >= offset { - report.files.push(path.to_path_buf()); - if report.files.len() >= head_limit { - report.truncated = true; - break 'walker; - } - } - matching_files_seen += 1; - } - GrepOutputMode::Count => { - let count = scan_count(&mut searcher, &matcher, path)?; - if count == 0 { - continue; - } - if matching_files_seen >= offset { - report.counts.push((path.to_path_buf(), count)); - if report.counts.len() >= head_limit { - report.truncated = true; - break 'walker; - } - } - matching_files_seen += 1; - } - GrepOutputMode::Content => { - let before_count = matches_seen; - let mut sink = ContentSink { - path: path.to_path_buf(), - lines: &mut report.lines, - matches_seen: &mut matches_seen, - offset, - head_limit, - }; - searcher - .search_path(&matcher, path, &mut sink) - .map_err(|e| FsError::io(path, e))?; - // If we hit head_limit during this file, stop walking. - if matches_seen >= offset.saturating_add(head_limit) && matches_seen > before_count - { - report.truncated = true; - break 'walker; - } - } + if scan_path( + &mut searcher, + &matcher, + path, + mode, + &mut report, + &mut matching_files_seen, + &mut matches_seen, + offset, + head_limit, + )? { + break; } } Ok(report.into_result(root)) } +#[allow(clippy::too_many_arguments)] +fn scan_path( + searcher: &mut Searcher, + matcher: &grep_regex::RegexMatcher, + path: &Path, + mode: GrepOutputMode, + report: &mut GrepReport, + matching_files_seen: &mut usize, + matches_seen: &mut usize, + offset: usize, + head_limit: usize, +) -> Result { + match mode { + GrepOutputMode::FilesWithMatches => { + if !scan_any_match(searcher, matcher, path)? { + return Ok(false); + } + if *matching_files_seen >= offset { + report.files.push(path.to_path_buf()); + if report.files.len() >= head_limit { + report.truncated = true; + return Ok(true); + } + } + *matching_files_seen += 1; + } + GrepOutputMode::Count => { + let count = scan_count(searcher, matcher, path)?; + if count == 0 { + return Ok(false); + } + if *matching_files_seen >= offset { + report.counts.push((path.to_path_buf(), count)); + if report.counts.len() >= head_limit { + report.truncated = true; + return Ok(true); + } + } + *matching_files_seen += 1; + } + GrepOutputMode::Content => { + let before_count = *matches_seen; + let mut sink = ContentSink { + path: path.to_path_buf(), + lines: &mut report.lines, + matches_seen, + offset, + head_limit, + }; + searcher + .search_path(matcher, path, &mut sink) + .map_err(|error| FsError::io(path, error))?; + if *matches_seen >= offset.saturating_add(head_limit) && *matches_seen > before_count { + report.truncated = true; + return Ok(true); + } + } + } + Ok(false) +} + fn scan_any_match( searcher: &mut Searcher, matcher: &grep_regex::RegexMatcher, diff --git a/crates/manifest/src/config.rs b/crates/manifest/src/config.rs index 9671675d..1eb3e86c 100644 --- a/crates/manifest/src/config.rs +++ b/crates/manifest/src/config.rs @@ -92,6 +92,8 @@ pub struct FeatureConfigPartial { #[serde(default)] pub worker: Option, #[serde(default)] + pub workspace_worker_discovery: Option, + #[serde(default)] pub objective: Option, #[serde(default)] pub manage_workdir: Option, @@ -119,6 +121,11 @@ impl FeatureConfigPartial { ), flow: merge_option(self.flow, other.flow, FeatureFlagConfigPartial::merge), worker: merge_option(self.worker, other.worker, WorkerFeatureConfigPartial::merge), + workspace_worker_discovery: merge_option( + self.workspace_worker_discovery, + other.workspace_worker_discovery, + FeatureFlagConfigPartial::merge, + ), objective: merge_option( self.objective, other.objective, @@ -265,6 +272,10 @@ impl From for FeatureConfig { .worker .map(WorkerFeatureConfig::from) .unwrap_or_default(), + workspace_worker_discovery: value + .workspace_worker_discovery + .map(FeatureFlagConfig::from) + .unwrap_or_default(), objective: value .objective .map(FeatureFlagConfig::from) @@ -394,6 +405,7 @@ impl From for FeatureConfigPartial { sub_worker: Some(value.sub_worker.into()), flow: Some(value.flow.into()), worker: Some(value.worker.into()), + workspace_worker_discovery: Some(value.workspace_worker_discovery.into()), objective: Some(value.objective.into()), manage_workdir: Some(value.manage_workdir.into()), ticket: Some(value.ticket.into()), diff --git a/crates/manifest/src/lib.rs b/crates/manifest/src/lib.rs index 3d057c05..5cb31db9 100644 --- a/crates/manifest/src/lib.rs +++ b/crates/manifest/src/lib.rs @@ -118,6 +118,10 @@ pub struct FeatureConfig { pub flow: FeatureFlagConfig, #[serde(default)] pub worker: WorkerFeatureConfig, + /// Privileged read-only discovery of visible Workspace Workers. Backend + /// source proof remains required for every listing operation. + #[serde(default)] + pub workspace_worker_discovery: FeatureFlagConfig, #[serde(default)] pub objective: FeatureFlagConfig, #[serde(default)] @@ -142,6 +146,7 @@ impl Default for FeatureConfig { sub_worker: FeatureFlagConfig::disabled(), flow: FeatureFlagConfig::disabled(), worker: WorkerFeatureConfig::disabled(), + workspace_worker_discovery: FeatureFlagConfig::disabled(), objective: FeatureFlagConfig::disabled(), manage_workdir: FeatureFlagConfig::disabled(), ticket: TicketFeatureConfig::default(), diff --git a/crates/manifest/src/profile.rs b/crates/manifest/src/profile.rs index d84c195d..b6cf09d0 100644 --- a/crates/manifest/src/profile.rs +++ b/crates/manifest/src/profile.rs @@ -946,9 +946,11 @@ fn apply_role_profile( value["feature"]["sub_worker"] = serde_json::json!({ "enabled": sub_worker }); value["feature"]["flow"] = serde_json::json!({ "enabled": slug == "coder" }); value["feature"]["worker"] = serde_json::json!({ - "enabled": slug == "orchestrator", - "direct_spawn": slug != "orchestrator" + "enabled": matches!(slug, "companion" | "orchestrator"), + "direct_spawn": !matches!(slug, "companion" | "orchestrator") }); + value["feature"]["workspace_worker_discovery"] = + serde_json::json!({ "enabled": slug == "companion" }); value["feature"]["manage_workdir"] = serde_json::json!({ "enabled": matches!(slug, "companion" | "orchestrator") }); @@ -1423,7 +1425,7 @@ mod tests { } #[test] - fn builtin_companion_uses_sub_worker_control_without_worker_control() { + fn builtin_companion_combines_runtime_and_sub_worker_control_with_discovery() { let tmp = TempDir::new().unwrap(); let resolved = ProfileResolver::new() .with_workspace_base(tmp.path()) @@ -1435,7 +1437,9 @@ mod tests { assert!(resolved.manifest.feature.manage_workdir.enabled); assert!(resolved.manifest.feature.sub_worker.enabled); - assert!(!resolved.manifest.feature.worker.enabled); + assert!(resolved.manifest.feature.worker.enabled); + assert!(!resolved.manifest.feature.worker.direct_spawn); + assert!(resolved.manifest.feature.workspace_worker_discovery.enabled); } #[test] diff --git a/crates/tools/src/grep.rs b/crates/tools/src/grep.rs index 27c26ad0..785470d4 100644 --- a/crates/tools/src/grep.rs +++ b/crates/tools/src/grep.rs @@ -22,7 +22,7 @@ enum OutputMode { #[derive(Debug, Deserialize, JsonSchema)] struct GrepParams { pattern: String, - /// Logical Workdir-relative path to search. Defaults to the Workdir root. + /// Logical Workdir-relative file or directory to search. Defaults to the Workdir root. #[serde(default)] path: Option, #[serde(default)] @@ -129,7 +129,7 @@ pub fn grep_tool(session: WorkdirSessionHandle) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(GrepParams); let meta = ToolMeta::new("Grep") - .description("Search Workdir file contents with a regex. Content results group lines by file; `>` marks matching lines and unmarked lines are context. Glob/Grep traversal executes inside the WorkdirSession provider. Results are bounded and Workdir-relative.") + .description("Search a Workdir file or directory with a regex. Content results group lines by file; `>` marks matching lines and unmarked lines are context. Directory traversal executes inside the WorkdirSession provider. Results are bounded and Workdir-relative.") .input_schema(serde_json::to_value(schema).expect("Grep schema serialization")); let tool: Arc = Arc::new(GrepTool { session: session.clone(), diff --git a/crates/workdir/src/http.rs b/crates/workdir/src/http.rs index f5e10d6b..9e152df0 100644 --- a/crates/workdir/src/http.rs +++ b/crates/workdir/src/http.rs @@ -107,6 +107,31 @@ pub enum WorkdirTransportErrorCode { Internal, } +impl WorkdirTransportErrorCode { + pub const fn as_str(self) -> &'static str { + match self { + Self::NotFound => "not_found", + Self::Conflict => "conflict", + Self::Unsupported => "unsupported", + Self::InvalidRequest => "invalid_request", + Self::UnknownCommand => "unknown_command", + Self::Unavailable => "unavailable", + Self::Internal => "internal", + } + } + + /// Shared public HTTP classification for Runtime and Workspace Workdir operation boundaries. + pub const fn http_status(self) -> u16 { + match self { + Self::NotFound | Self::UnknownCommand => 404, + Self::Conflict => 409, + Self::Unsupported | Self::InvalidRequest => 400, + Self::Unavailable => 503, + Self::Internal => 500, + } + } +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkdirTransportError { pub code: WorkdirTransportErrorCode, @@ -126,6 +151,9 @@ impl WorkdirTransportError { message: format!("Workdir capability {capability:?} is not available"), }; } + WorkdirError::UnsupportedOperation(_) => { + (Code::Unsupported, "Workdir operation is not supported") + } WorkdirError::UnknownCommand(_) => { (Code::UnknownCommand, "Workdir command was not found") } @@ -161,7 +189,7 @@ impl WorkdirTransportError { match self.code { Code::NotFound => WorkdirError::NotFound("".into()), Code::Conflict => WorkdirError::Conflict(self.message), - Code::Unsupported => WorkdirError::Unavailable(self.message), + Code::Unsupported => WorkdirError::UnsupportedOperation(self.message), Code::UnknownCommand => WorkdirError::UnknownCommand("".to_string()), Code::InvalidRequest => WorkdirError::InvalidArgument(self.message), Code::Unavailable => WorkdirError::Unavailable(self.message), @@ -517,13 +545,13 @@ mod client { .json::() .await .map(WorkdirTransportError::into_workdir_error) - .unwrap_or_else(|error| { - WorkdirError::Unavailable(format!("Runtime HTTP error: {error}")) + .unwrap_or_else(|_| { + WorkdirError::Transport("Runtime Workdir error response was invalid".to_string()) }) } - fn http_unavailable(error: reqwest::Error) -> WorkdirError { - WorkdirError::Unavailable(format!("Runtime Workdir HTTP request failed: {error}")) + fn http_unavailable(_error: reqwest::Error) -> WorkdirError { + WorkdirError::Transport("Runtime Workdir HTTP request failed".to_string()) } pub use self::RemoteWorkdirSession as ClientSession; @@ -536,6 +564,57 @@ pub use client::{ClientSession as RemoteWorkdirSession, WorkdirHttpAuthorization mod tests { use super::*; + #[test] + fn transport_error_round_trip_keeps_public_classification() { + for (code, expected_status, expected_error) in [ + ( + WorkdirTransportErrorCode::InvalidRequest, + 400, + "invalid argument", + ), + (WorkdirTransportErrorCode::NotFound, 404, "file not found"), + ( + WorkdirTransportErrorCode::UnknownCommand, + 404, + "unknown Workdir session command", + ), + ( + WorkdirTransportErrorCode::Conflict, + 409, + "modified externally", + ), + (WorkdirTransportErrorCode::Unsupported, 400, "unsupported"), + (WorkdirTransportErrorCode::Unavailable, 503, "unavailable"), + (WorkdirTransportErrorCode::Internal, 500, "transport failed"), + ] { + let transport = WorkdirTransportError { + code, + message: "safe provider message".to_string(), + }; + assert_eq!(code.http_status(), expected_status); + let workdir_error = transport.clone().into_workdir_error(); + assert!(workdir_error.to_string().contains(expected_error)); + assert_eq!( + WorkdirTransportError::from_workdir_error(&workdir_error).code, + code + ); + } + } + + #[test] + fn local_validation_errors_share_invalid_request_classification() { + for error in [ + WorkdirError::InvalidGlob("[".to_string()), + WorkdirError::InvalidRegex("(".to_string()), + WorkdirError::InvalidArgument("limit must be positive".to_string()), + ] { + let transport = WorkdirTransportError::from_workdir_error(&error); + assert_eq!(transport.code, WorkdirTransportErrorCode::InvalidRequest); + assert_eq!(transport.code.http_status(), 400); + assert_eq!(transport.message, "Workdir operation request is invalid"); + } + } + #[test] fn transport_failure_remains_distinct_from_session_unavailable() { let transport = WorkdirTransportError::from_workdir_error(&WorkdirError::Transport( diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index 22fb57de..d673b651 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -225,6 +225,9 @@ pub enum WorkdirError { #[error("Workdir session does not support {0:?}")] Unsupported(WorkdirSessionCapability), + #[error("Workdir operation is unsupported: {0}")] + UnsupportedOperation(String), + #[error("invalid Workdir path: {0}")] InvalidPath(String), diff --git a/crates/workdir/src/local.rs b/crates/workdir/src/local.rs index 16686cb4..a897076f 100644 --- a/crates/workdir/src/local.rs +++ b/crates/workdir/src/local.rs @@ -1943,9 +1943,9 @@ mod tests { &workdir, GrepRequest { pattern: "NEEDLE".into(), - path: WorkdirPath::root(), - glob: None, - file_type: None, + path: WorkdirPath::new("src/main.rs").unwrap(), + glob: Some("src/*.rs".into()), + file_type: Some("rust".into()), case_insensitive: false, before_context: 0, after_context: 0, diff --git a/crates/worker-runtime/src/auth.rs b/crates/worker-runtime/src/auth.rs index bf02b35f..05279ea5 100644 --- a/crates/worker-runtime/src/auth.rs +++ b/crates/worker-runtime/src/auth.rs @@ -17,6 +17,7 @@ const WORKER_MUTATION_SOURCE_SIGNING_INPUT_PREFIX: &str = "yoi-worker-source-v1. pub const WORKER_REMOVE_PERMISSION: &str = "workspace:worker-remove"; pub const RUNTIME_REQUEST_SOURCE_PROOF_HEADER: &str = "x-yoi-runtime-request-proof"; pub const WORKSPACE_REQUEST_PERMISSION: &str = "workspace:request"; +pub const WORKSPACE_WORKER_DISCOVERY_PERMISSION: &str = "workspace:worker-discovery"; pub const BACKEND_RESOURCE_FETCH_PERMISSION: &str = "workspace:resource-fetch"; const RUNTIME_REQUEST_SOURCE_PROOF_PREFIX: &str = "yoi-runtime-request-v1"; const RUNTIME_REQUEST_SOURCE_SIGNING_INPUT_PREFIX: &str = "yoi-runtime-request-v1."; diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 20b9e442..610ea770 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -1721,17 +1721,8 @@ impl RuntimeHttpWorkdirError { impl From for RuntimeHttpWorkdirError { fn from(error: workdir::WorkdirError) -> Self { let payload = WorkdirTransportError::from_workdir_error(&error); - let status = match payload.code { - WorkdirTransportErrorCode::NotFound | WorkdirTransportErrorCode::UnknownCommand => { - StatusCode::NOT_FOUND - } - WorkdirTransportErrorCode::Conflict => StatusCode::CONFLICT, - WorkdirTransportErrorCode::Unsupported | WorkdirTransportErrorCode::InvalidRequest => { - StatusCode::BAD_REQUEST - } - WorkdirTransportErrorCode::Unavailable => StatusCode::SERVICE_UNAVAILABLE, - WorkdirTransportErrorCode::Internal => StatusCode::INTERNAL_SERVER_ERROR, - }; + let status = StatusCode::from_u16(payload.code.http_status()) + .expect("Workdir transport error status is valid"); Self { status, payload } } } @@ -1886,8 +1877,8 @@ mod tests { use manifest::{Scope, SharedScope}; use tower::ServiceExt; use workdir::{ - LocalWorkdirSession, ReadRequest, StatRequest, Workdir, WorkdirPath, - WorkdirSessionCapabilities, + GrepOutputMode, GrepRequest, LocalWorkdirSession, ReadRequest, StatRequest, Workdir, + WorkdirPath, WorkdirSessionCapabilities, }; fn test_bundle(profile: ProfileSelector) -> ConfigBundle { @@ -2348,6 +2339,40 @@ mod tests { .expect("owned operation"); assert!(matches!(result, WorkdirSessionOperationResult::Stat(_))); + let grep = WorkdirSessionOperationRequest { + delegations: Vec::new(), + operation: WorkdirSessionOperation::Grep(GrepRequest { + pattern: "hello".into(), + path: WorkdirPath::new("hello.txt").unwrap(), + glob: Some("*.txt".into()), + file_type: Some("txt".into()), + case_insensitive: false, + before_context: 0, + after_context: 0, + multiline: false, + output_mode: GrepOutputMode::Content, + limit: 10, + offset: 0, + }), + }; + let Json(result) = run_workdir_session_operation( + State(state.clone()), + Path("session-1".to_string()), + Some(Extension(auth.clone())), + Ok(Json(grep)), + ) + .await + .expect("grep direct file through provider operation"); + match result { + WorkdirSessionOperationResult::Grep(result) => { + assert_eq!(result.match_count, 1); + assert_eq!(result.matched_files, 1); + assert!(result.output.starts_with("hello.txt\n")); + assert!(result.output.contains("> 1 │ hello")); + } + other => panic!("unexpected workdir grep result: {other:?}"), + } + #[cfg(unix)] { let delegated_visible = WorkdirSessionOperationRequest { diff --git a/crates/worker-runtime/src/worker_source.rs b/crates/worker-runtime/src/worker_source.rs index 4c43e4a1..ed485ab0 100644 --- a/crates/worker-runtime/src/worker_source.rs +++ b/crates/worker-runtime/src/worker_source.rs @@ -9,8 +9,8 @@ use worker::{ use crate::auth::{ RUNTIME_REQUEST_SOURCE_PROOF_HEADER, RuntimeAuthError, RuntimeIdentityMaterial, RuntimeRequestSourceSigner, RuntimeWorkerMutationSourceSigner, WORKER_REMOVE_PERMISSION, - WORKSPACE_REQUEST_PERMISSION, WorkerMutationActorKind, WorkerMutationOperation, - WorkerMutationSourceClaims, new_token_id, + WORKSPACE_REQUEST_PERMISSION, WORKSPACE_WORKER_DISCOVERY_PERMISSION, WorkerMutationActorKind, + WorkerMutationOperation, WorkerMutationSourceClaims, new_token_id, }; use crate::runtime::RuntimeWorkspaceScope; use crate::worker_backend::WorkspacePromptProjectionCache; @@ -343,6 +343,51 @@ impl RuntimeOwnedWorkspaceClient { self.request_timeout = request_timeout; self } + + fn execute_with_permission( + &self, + request: WorkspaceRequest, + permission: &'static str, + ) -> Result { + let base_url = self.base_url.clone(); + let workspace_id = self.workspace_id.clone(); + let runtime_id = self.runtime_id.clone(); + let worker_id = self.worker_id.clone(); + let request_source_signer = self.request_source_signer.clone(); + let request_source_audience = self.request_source_audience.clone(); + let request_timeout = self.request_timeout; + if tokio::runtime::Handle::try_current().is_ok() { + std::thread::spawn(move || { + execute_runtime_owned_workspace_http( + &base_url, + &workspace_id, + &runtime_id, + &worker_id, + request_source_signer.as_ref(), + request_source_audience.as_deref(), + request_timeout, + permission, + request, + ) + }) + .join() + .map_err(|_| { + WorkspaceClientError::Request("workspace request thread panicked".to_string()) + })? + } else { + execute_runtime_owned_workspace_http( + &self.base_url, + &self.workspace_id, + &self.runtime_id, + &self.worker_id, + self.request_source_signer.as_ref(), + self.request_source_audience.as_deref(), + self.request_timeout, + permission, + request, + ) + } + } } impl std::fmt::Debug for RuntimeOwnedWorkspaceClient { @@ -377,42 +422,40 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { &self, request: WorkspaceRequest, ) -> Result { - let base_url = self.base_url.clone(); - let workspace_id = self.workspace_id.clone(); - let runtime_id = self.runtime_id.clone(); - let worker_id = self.worker_id.clone(); - let request_source_signer = self.request_source_signer.clone(); - let request_source_audience = self.request_source_audience.clone(); - let request_timeout = self.request_timeout; - if tokio::runtime::Handle::try_current().is_ok() { - std::thread::spawn(move || { - execute_runtime_owned_workspace_http( - &base_url, - &workspace_id, - &runtime_id, - &worker_id, - request_source_signer.as_ref(), - request_source_audience.as_deref(), - request_timeout, - request, - ) - }) - .join() - .map_err(|_| { - WorkspaceClientError::Request("workspace request thread panicked".to_string()) - })? - } else { - execute_runtime_owned_workspace_http( - &self.base_url, - &self.workspace_id, - &self.runtime_id, - &self.worker_id, - self.request_source_signer.as_ref(), - self.request_source_audience.as_deref(), - self.request_timeout, - request, - ) + self.execute_with_permission(request, WORKSPACE_REQUEST_PERMISSION) + } + + fn list_workspace_workers( + &self, + request: worker::WorkspaceWorkerDiscoveryRequest, + ) -> Result { + let mut path = format!( + "/api/w/{}/worker-discovery/workers?limit={}", + self.workspace_id, request.limit + ); + if let Some(cursor) = request.cursor.as_deref() { + path.push_str("&cursor="); + path.push_str(&percent_encode_query(cursor)); } + if let Some(query) = request.query.as_deref() { + path.push_str("&query="); + path.push_str(&percent_encode_query(query)); + } + let response = self.execute_with_permission( + WorkspaceRequest::get(path), + WORKSPACE_WORKER_DISCOVERY_PERMISSION, + )?; + if !(200..300).contains(&response.status) { + return Err(WorkspaceClientError::Request(format!( + "Workspace Worker discovery failed with HTTP {}: {}", + response.status, response.body + ))); + } + serde_json::from_str(&response.body).map_err(|error| { + WorkspaceClientError::Request(format!( + "invalid Workspace Worker discovery response: {error}" + )) + }) } fn current_prompt_projection( @@ -506,6 +549,19 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { } } +fn percent_encode_query(value: &str) -> String { + let mut encoded = String::with_capacity(value.len()); + for byte in value.bytes() { + if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') { + encoded.push(char::from(byte)); + } else { + use std::fmt::Write as _; + let _ = write!(encoded, "%{byte:02X}"); + } + } + encoded +} + fn execute_runtime_owned_workspace_http( base_url: &str, workspace_id: &str, @@ -514,6 +570,7 @@ fn execute_runtime_owned_workspace_http( request_source_signer: Option<&RuntimeRequestSourceSigner>, request_source_audience: Option<&str>, request_timeout: Option, + permission: &'static str, request: WorkspaceRequest, ) -> Result { if !request.path.starts_with('/') || request.path.starts_with("//") { @@ -553,7 +610,7 @@ fn execute_runtime_owned_workspace_http( audience, workspace_id, Some(worker_id), - WORKSPACE_REQUEST_PERMISSION, + permission, method.as_str(), &request.path, body.as_bytes(), @@ -903,6 +960,85 @@ mod tests { ); } + #[test] + fn workspace_worker_discovery_signs_dedicated_permission_and_encoded_query() { + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::sync::Mutex; + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let received = Arc::new(Mutex::new(String::new())); + let received_for_server = received.clone(); + let body = serde_json::json!({ + "workers": [{ + "subject": { + "kind": "runtime_worker", + "runtime_id": "runtime-b", + "worker_id": "worker-b" + }, + "resource_key": "W-2", + "display_name": "coder two", + "profile": "builtin:coder", + "status": "idle" + }], + "next_cursor": "v1:1" + }) + .to_string(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut bytes = [0_u8; 4096]; + let count = stream.read(&mut bytes).unwrap(); + *received_for_server.lock().unwrap() = + String::from_utf8_lossy(&bytes[..count]).into_owned(); + write!( + stream, + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}", + body.len(), + body + ) + .unwrap(); + }); + + let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap(); + let client = RuntimeOwnedWorkspaceClient::new( + "workspace-a", + format!("http://{address}"), + "runtime-a", + "worker-a", + ) + .with_runtime_request_source(&identity, "server-a"); + let page = client + .list_workspace_workers(worker::WorkspaceWorkerDiscoveryRequest { + cursor: Some("v1:0".to_string()), + limit: 1, + query: Some("coder two".to_string()), + }) + .unwrap(); + assert_eq!(page.workers[0].resource_key, "W-2"); + server.join().unwrap(); + + let request = received.lock().unwrap().clone(); + assert!(request.contains( + "GET /api/w/workspace-a/worker-discovery/workers?limit=1&cursor=v1%3A0&query=coder%20two " + )); + let token = request + .lines() + .find_map(|line| { + line.split_once(':').and_then(|(name, value)| { + name.eq_ignore_ascii_case(RUNTIME_REQUEST_SOURCE_PROOF_HEADER) + .then(|| value.trim()) + }) + }) + .unwrap(); + let claims = decode_runtime_request_source_claims(token).unwrap(); + assert_eq!(claims.permission, WORKSPACE_WORKER_DISCOVERY_PERMISSION); + assert_eq!( + claims.path, + "/api/w/workspace-a/worker-discovery/workers?limit=1&cursor=v1%3A0&query=coder%20two" + ); + } + #[test] fn remote_authority_stamps_and_signs_worker_remove_without_caller_claim_choices() { let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap(); diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index bf951b6c..87437d17 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -27,6 +27,7 @@ toml = { workspace = true } tracing = { workspace = true } tools = { workspace = true } workdir = { workspace = true } +workspace-api = { workspace = true } minijinja = "2.19.0" chrono = "0.4" config-source = { path = "../config-source" } diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 41b33f44..ad7aa5b5 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -884,8 +884,10 @@ where .register_tools(tools::web_builtin_tools(web_config)); } + let worker_enabled = feature_config.worker.enabled; + let sub_worker_enabled = feature_config.sub_worker.enabled; let mut feature_registry = FeatureRegistryBuilder::new(); - if feature_config.sub_worker.enabled { + if sub_worker_enabled && !worker_enabled { feature_registry.add_module( crate::feature::builtin::manage_worker::sub_worker_control_feature( worker.workspace_client_handle(), @@ -956,6 +958,23 @@ where crate::feature::builtin::manage_workdir::manage_workdir_feature(workspace_client), ); } + if feature_config.workspace_worker_discovery.enabled { + let workspace_client = worker.workspace_client_handle(); + let has_workspace_identity = workspace_client.workspace_id().is_some_and(|workspace_id| { + !workspace_id.is_empty() && !workspace_id.chars().any(char::is_control) + }); + if !workspace_client.is_available() || !has_workspace_identity { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "Workspace Worker discovery requires Backend Workspace API authority", + )); + } + feature_registry.add_module( + crate::feature::builtin::workspace_worker_discovery::workspace_worker_discovery_feature( + workspace_client, + ), + ); + } if feature_config.worker.enabled { let workspace_client = worker.workspace_client_handle(); let has_workspace_identity = workspace_client.workspace_id().is_some_and(|workspace_id| { @@ -970,7 +989,7 @@ where feature_registry.add_module( crate::feature::builtin::manage_worker::manage_worker_feature( workspace_client, - Some(spawned_registry.clone()), + sub_worker_enabled.then(|| spawned_registry.clone()), feature_config.worker.direct_spawn, ), ); diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index 010fcf96..2f1be0a7 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -17,6 +17,7 @@ pub mod session_explore; pub mod task; pub mod ticket; pub mod worker_observation; +pub mod workspace_worker_discovery; pub(crate) use memory_extract::{MemoryExtractFeature, MemoryExtractState, render_extract_input}; pub(crate) use session_explore::{SessionExploreFeature, SessionExploreState}; diff --git a/crates/worker/src/feature/builtin/manage_workdir.rs b/crates/worker/src/feature/builtin/manage_workdir.rs index ffc7cbdc..46397ea0 100644 --- a/crates/worker/src/feature/builtin/manage_workdir.rs +++ b/crates/worker/src/feature/builtin/manage_workdir.rs @@ -195,11 +195,16 @@ impl WorkspaceAttachedWorkdirSession { .execute(request) .map_err(workspace_workdir_error)?; if !response.is_success() { - return Err(WorkdirError::Transport(format!( - "Workspace Workdir API returned HTTP {}: {}", - response.status, - bounded_error_body(&response.body) - ))); + return Err( + serde_json::from_str::(&response.body) + .map(workdir::http::WorkdirTransportError::into_workdir_error) + .unwrap_or_else(|_| { + WorkdirError::Transport(format!( + "Workspace Workdir operation failed with HTTP {}", + response.status + )) + }), + ); } serde_json::from_str(&response.body).map_err(|error| { WorkdirError::Transport(format!( @@ -795,6 +800,21 @@ mod tests { } } + fn error_response( + status: u16, + code: workdir::http::WorkdirTransportErrorCode, + message: &str, + ) -> WorkspaceResponse { + WorkspaceResponse { + status, + body: serde_json::to_string(&workdir::http::WorkdirTransportError { + code, + message: message.to_string(), + }) + .unwrap(), + } + } + fn workdir_json(id: &str) -> serde_json::Value { json!({ "working_directory_id": id, @@ -1175,6 +1195,48 @@ mod tests { assert_eq!(validation["delegations"].as_array().unwrap().len(), 1); } + #[tokio::test] + async fn attached_session_preserves_typed_provider_validation_error() { + let client = Arc::new(RecordingWorkspaceClient::new(vec![error_response( + 400, + workdir::http::WorkdirTransportErrorCode::InvalidRequest, + "Workdir operation request is invalid", + )])); + let session = WorkspaceAttachedWorkdirSession::handle(client); + + let error = session + .glob(workdir::GlobRequest { + pattern: "[".to_string(), + path: workdir::WorkdirPath::root(), + limit: 10, + }) + .await + .unwrap_err(); + + assert!(matches!(error, WorkdirError::InvalidArgument(_))); + } + + #[tokio::test] + async fn attached_session_does_not_expose_untyped_workspace_error_body() { + let client = Arc::new(RecordingWorkspaceClient::new(vec![WorkspaceResponse { + status: 502, + body: "secret token and /host/private/path".to_string(), + }])); + let session = WorkspaceAttachedWorkdirSession::handle(client); + + let error = session + .stat(StatRequest { + path: workdir::WorkdirPath::root(), + }) + .await + .unwrap_err(); + let message = error.to_string(); + assert!(matches!(error, WorkdirError::Transport(_))); + assert!(!message.contains("secret token")); + assert!(!message.contains("/host/private/path")); + assert!(message.contains("HTTP 502")); + } + #[tokio::test] async fn nested_attached_session_preserves_full_delegation_chain() { let client = Arc::new(RecordingWorkspaceClient::new(vec![ diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 12d481b4..1ea1950d 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -28,16 +28,45 @@ pub const WORKER_LIFECYCLE_SERVICE_ID: &str = "worker.lifecycle"; pub const WORKER_CONTROL_SERVICE_ID: &str = "worker.control"; const WORKER_LIFECYCLE_SERVICE_VERSION: &str = "1"; +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WorkerControlSubject { + RuntimeWorker { + runtime_id: String, + worker_id: String, + }, + SubWorker { + name: String, + }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkerControlInputKind { + User, + Notify, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkerControlStopKind { + Cancel, + Stop, +} + #[async_trait] pub trait WorkerControlService: Send + Sync { fn workspace_id(&self) -> &str; - fn known_subworkers(&self) -> Vec; - async fn send_subworker( + async fn list_workers(&self) -> Result; + async fn send_input( &self, - name: &str, + subject: WorkerControlSubject, content: String, + kind: WorkerControlInputKind, + ) -> Result; + async fn stop_worker( + &self, + subject: WorkerControlSubject, + kind: WorkerControlStopKind, + reason: Option, ) -> Result; - async fn stop_subworker(&self, name: &str) -> Result; async fn spawn_worker( &self, request: WorkerLifecycleSpawnRequest, @@ -63,24 +92,10 @@ struct WorkspaceWorkerControlService { client: Arc, workspace_id: String, registry: Option>, + runtime_worker_control: bool, } -impl std::fmt::Debug for WorkspaceWorkerControlService { - fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter - .debug_struct("WorkspaceWorkerControlService") - .field("workspace_id", &self.workspace_id) - .field("has_subworker_registry", &self.registry.is_some()) - .finish_non_exhaustive() - } -} - -#[async_trait] -impl WorkerControlService for WorkspaceWorkerControlService { - fn workspace_id(&self) -> &str { - &self.workspace_id - } - +impl WorkspaceWorkerControlService { fn known_subworkers(&self) -> Vec { self.registry .as_ref() @@ -104,51 +119,203 @@ impl WorkerControlService for WorkspaceWorkerControlService { }) .unwrap_or_default() } +} - async fn send_subworker( - &self, - name: &str, - content: String, - ) -> Result { - let record = self - .registry - .as_ref() - .and_then(|registry| registry.get_internal(name)) +impl std::fmt::Debug for WorkspaceWorkerControlService { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("WorkspaceWorkerControlService") + .field("workspace_id", &self.workspace_id) + .field("has_subworker_registry", &self.registry.is_some()) + .finish_non_exhaustive() + } +} + +#[async_trait] +impl WorkerControlService for WorkspaceWorkerControlService { + fn workspace_id(&self) -> &str { + &self.workspace_id + } + + async fn list_workers(&self) -> Result { + let response = if self.runtime_worker_control { + self.execute_runtime(WorkspaceRequest::get(format!( + "/api/w/{}/worker-control/workers", + self.workspace_id + ))) + .await? + } else { + WorkspaceResponse { + status: 200, + body: serde_json::json!({ "items": [] }).to_string(), + } + }; + if !response.is_success() { + return Ok(response); + } + + let mut payload: serde_json::Value = + serde_json::from_str(&response.body).map_err(|error| { + WorkspaceClientError::Request(format!( + "invalid Workspace control response: {error}" + )) + })?; + let items = payload + .get_mut("items") + .and_then(serde_json::Value::as_array_mut) .ok_or_else(|| { WorkspaceClientError::Request( - "unknown Worker or permission not granted".to_string(), + "invalid Workspace control response: missing items".to_string(), ) })?; - record - .session - .send(content) - .await - .map_err(|error| WorkspaceClientError::Request(error.to_string()))?; + let mut seen = std::collections::HashSet::new(); + items.retain(|item| { + item.get("subject") + .map(serde_json::Value::to_string) + .is_none_or(|subject| seen.insert(subject)) + }); + for worker in self.known_subworkers() { + let is_new = worker + .get("subject") + .map(serde_json::Value::to_string) + .is_none_or(|subject| seen.insert(subject)); + if is_new { + items.push(worker); + } + } Ok(WorkspaceResponse { - status: 200, - body: serde_json::json!({ "subject": { "kind": "sub_worker", "name": name } }) - .to_string(), + status: response.status, + body: serde_json::to_string(&payload) + .map_err(|error| WorkspaceClientError::Request(error.to_string()))?, }) } - async fn stop_subworker(&self, name: &str) -> Result { - let registry = self.registry.as_ref().ok_or_else(|| { - WorkspaceClientError::Request("unknown Worker or permission not granted".to_string()) - })?; - let summary = registry - .remove_internal(name) - .await - .map_err(|error| WorkspaceClientError::Request(error.to_string()))? - .ok_or_else(|| { - WorkspaceClientError::Request( - "unknown Worker or permission not granted".to_string(), + async fn send_input( + &self, + subject: WorkerControlSubject, + content: String, + kind: WorkerControlInputKind, + ) -> Result { + match subject { + WorkerControlSubject::RuntimeWorker { + runtime_id, + worker_id, + } => { + let kind = match kind { + WorkerControlInputKind::User => "user", + WorkerControlInputKind::Notify => "notify", + }; + self.execute_runtime(WorkspaceRequest::json( + WorkspaceRequestMethod::Post, + format!( + "/api/w/{}/worker-control/workers/{runtime_id}/{worker_id}/input", + self.workspace_id + ), + serde_json::json!({ + "kind": kind, + "content": content, + }) + .to_string(), + )) + .await + } + WorkerControlSubject::SubWorker { name } => { + if kind != WorkerControlInputKind::User { + return Err(WorkspaceClientError::Request( + "notify is not available for parent-owned subworkers".to_string(), + )); + } + self.ensure_permission( + &super::worker_observation::WorkerObservationSubjectRef::SubWorker { + name: name.clone(), + }, + "send_input", ) - })?; - Ok(WorkspaceResponse { - status: 200, - body: serde_json::to_string(&summary) - .map_err(|error| WorkspaceClientError::Request(error.to_string()))?, - }) + .await?; + let record = self + .registry + .as_ref() + .and_then(|registry| registry.get_internal(&name)) + .ok_or_else(|| { + WorkspaceClientError::Request( + "unknown Worker or permission not granted".to_string(), + ) + })?; + record + .session + .send(content) + .await + .map_err(|error| WorkspaceClientError::Request(error.to_string()))?; + Ok(WorkspaceResponse { + status: 200, + body: serde_json::json!({ + "subject": { "kind": "sub_worker", "name": name } + }) + .to_string(), + }) + } + } + } + + async fn stop_worker( + &self, + subject: WorkerControlSubject, + kind: WorkerControlStopKind, + reason: Option, + ) -> Result { + match subject { + WorkerControlSubject::RuntimeWorker { + runtime_id, + worker_id, + } => { + let action = match kind { + WorkerControlStopKind::Cancel => "cancel", + WorkerControlStopKind::Stop => "stop", + }; + self.execute_runtime(WorkspaceRequest::json( + WorkspaceRequestMethod::Post, + format!( + "/api/w/{}/worker-control/workers/{runtime_id}/{worker_id}/{action}", + self.workspace_id + ), + serde_json::json!({ "reason": reason }).to_string(), + )) + .await + } + WorkerControlSubject::SubWorker { name } => { + if kind != WorkerControlStopKind::Stop { + return Err(WorkspaceClientError::Request( + "cancel is not available for parent-owned subworkers".to_string(), + )); + } + self.ensure_permission( + &super::worker_observation::WorkerObservationSubjectRef::SubWorker { + name: name.clone(), + }, + "stop", + ) + .await?; + let registry = self.registry.as_ref().ok_or_else(|| { + WorkspaceClientError::Request( + "unknown Worker or permission not granted".to_string(), + ) + })?; + let summary = registry + .remove_internal(&name) + .await + .map_err(|error| WorkspaceClientError::Request(error.to_string()))? + .ok_or_else(|| { + WorkspaceClientError::Request( + "unknown Worker or permission not granted".to_string(), + ) + })?; + Ok(WorkspaceResponse { + status: 200, + body: serde_json::to_string(&summary) + .map_err(|error| WorkspaceClientError::Request(error.to_string()))?, + }) + } + } } async fn spawn_worker( @@ -169,6 +336,11 @@ impl WorkerControlService for WorkspaceWorkerControlService { worker_id: &str, reason: &str, ) -> Result { + if !self.runtime_worker_control { + return Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is not enabled for this Worker".to_string(), + )); + } self.client .execute_worker_remove(runtime_id, worker_id, reason) } @@ -177,6 +349,11 @@ impl WorkerControlService for WorkspaceWorkerControlService { &self, request: WorkspaceRequest, ) -> Result { + if !self.runtime_worker_control { + return Err(WorkspaceClientError::Request( + "Runtime Worker control is not enabled for this Worker".to_string(), + )); + } self.client.execute(request) } @@ -204,6 +381,11 @@ impl WorkerControlService for WorkspaceWorkerControlService { runtime_id, worker_id, } => { + if !self.runtime_worker_control { + return Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is not enabled for this Worker".to_string(), + )); + } let response = self.client.execute(WorkspaceRequest::get(format!( "/api/w/{}/worker-control/workers", self.workspace_id @@ -329,10 +511,6 @@ impl std::fmt::Debug for ManageWorkerFeature { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter .debug_struct("ManageWorkerFeature") - .field( - "has_subworker_registry", - &!self.control.known_subworkers().is_empty(), - ) .field("direct_spawn", &self.direct_spawn) .finish_non_exhaustive() } @@ -348,6 +526,7 @@ pub fn manage_worker_feature( client: client.clone(), workspace_id, registry, + runtime_worker_control: true, }); ManageWorkerFeature { client, @@ -356,6 +535,12 @@ pub fn manage_worker_feature( } } +const SUB_WORKER_CONTROL_OPERATIONS: &[WorkerOperation] = &[ + WorkerOperation::List, + WorkerOperation::SendInput, + WorkerOperation::Stop, +]; + pub struct SubWorkerControlFeature { client: Arc, registry: Arc, @@ -369,13 +554,19 @@ impl SubWorkerControlFeature { impl FeatureModule for SubWorkerControlFeature { fn descriptor(&self) -> FeatureDescriptor { - FeatureDescriptor::builtin("sub-worker-control", "SubWorker Control") - .with_description("Parent-owned SubWorker control service provider") + let mut descriptor = FeatureDescriptor::builtin("sub-worker-control", "SubWorker Control") + .with_description( + "Parent-owned SubWorker lifecycle through the canonical Worker control surface", + ) .with_provided_service(ServiceDeclaration::new( ServiceId::builtin(WORKER_CONTROL_SERVICE_ID), WORKER_LIFECYCLE_SERVICE_VERSION, "Parent-owned SubWorker discovery and control operations", - )) + )); + for operation in SUB_WORKER_CONTROL_OPERATIONS { + descriptor = descriptor.with_tool(sub_worker_tool_declaration(*operation)); + } + descriptor } fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { @@ -383,6 +574,7 @@ impl FeatureModule for SubWorkerControlFeature { workspace_id: self.client.workspace_id().unwrap_or_default().to_string(), client: self.client.clone(), registry: Some(self.registry.clone()), + runtime_worker_control: false, }); context.services().provide( ServiceDeclaration::new( @@ -390,8 +582,16 @@ impl FeatureModule for SubWorkerControlFeature { WORKER_LIFECYCLE_SERVICE_VERSION, "Parent-owned SubWorker discovery and control operations", ), - control, - ) + control.clone(), + )?; + for operation in SUB_WORKER_CONTROL_OPERATIONS { + context.tools().register(worker_tool_contribution( + *operation, + control.clone(), + false, + ))?; + } + Ok(()) } } @@ -418,10 +618,7 @@ impl FeatureModule for ManageWorkerFeature { )); for operation in WorkerOperation::ALL { if operation != WorkerOperation::Spawn || self.direct_spawn { - descriptor = descriptor.with_tool(ToolDeclaration::new( - operation.tool_name(), - operation.description(), - )); + descriptor = descriptor.with_tool(worker_tool_declaration(operation)); } } descriptor @@ -463,29 +660,11 @@ impl FeatureModule for ManageWorkerFeature { if operation == WorkerOperation::Spawn && !self.direct_spawn { continue; } - let definition = match operation { - WorkerOperation::List => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Spawn => { - definition::(operation, self.control.clone()) - } - WorkerOperation::SendInput | WorkerOperation::Notify => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Cancel | WorkerOperation::Stop => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Restore => { - definition::(operation, self.control.clone()) - } - WorkerOperation::Remove => { - definition::(operation, self.control.clone()) - } - }; - context - .tools() - .register(ToolContribution::new(operation.tool_name(), definition))?; + context.tools().register(worker_tool_contribution( + operation, + self.control.clone(), + true, + ))?; } Ok(()) } @@ -558,6 +737,34 @@ struct WorkerTargetInput { subject: WorkerSubjectInput, } +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum DirectSubWorkerSubjectInput { + SubWorker { name: String }, +} + +impl From for WorkerSubjectInput { + fn from(subject: DirectSubWorkerSubjectInput) -> Self { + match subject { + DirectSubWorkerSubjectInput::SubWorker { name } => Self::SubWorker { name }, + } + } +} + +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +struct DirectSubWorkerMessageInput { + subject: DirectSubWorkerSubjectInput, + content: String, +} + +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +struct DirectSubWorkerStopInput { + subject: DirectSubWorkerSubjectInput, + reason: Option, +} + #[derive(Debug, Deserialize, JsonSchema)] #[serde(deny_unknown_fields)] struct WorkerMessageInput { @@ -583,6 +790,7 @@ struct WorkerRemoveInput { struct WorkspaceWorkerTool { operation: WorkerOperation, control: Arc, + runtime_worker_control: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -654,15 +862,10 @@ impl Tool for WorkspaceWorkerTool { let response = match self.operation { WorkerOperation::List => { parse::(input_json, "WorkerList")?; - let response = self - .control - .execute_runtime(WorkspaceRequest::get(format!( - "/api/w/{}/worker-control/workers", - self.control.workspace_id() - ))) + self.control + .list_workers() .await - .map_err(control_tool_error)?; - self.with_subworkers(response)? + .map_err(control_tool_error)? } WorkerOperation::Spawn => { let input = parse::(input_json, "WorkerSpawn")?; @@ -701,74 +904,56 @@ impl Tool for WorkspaceWorkerTool { .map_err(control_tool_error)? } WorkerOperation::SendInput | WorkerOperation::Notify => { - let input = parse::(input_json, self.operation.tool_name())?; + let input = if self.runtime_worker_control { + parse::(input_json, self.operation.tool_name())? + } else { + let input = parse::( + input_json, + self.operation.tool_name(), + )?; + WorkerMessageInput { + subject: input.subject.into(), + content: input.content, + } + }; let content = non_empty(input.content, "content")?; if content.len() > 16 * 1024 { return Err(ToolError::ExecutionFailed( "content must contain at most 16384 bytes".to_string(), )); } - match input.subject { - WorkerSubjectInput::SubWorker { name } => { - if self.operation != WorkerOperation::SendInput { - return Err(unsupported_subject(self.operation, "sub_worker")); - } - let subject = subworker_subject(&name)?; - self.control - .ensure_permission(&subject, "send_input") - .await - .map_err(control_tool_error)?; - self.control - .send_subworker(&name, content) - .await - .map_err(control_tool_error)? - } - subject @ WorkerSubjectInput::RuntimeWorker { .. } => { - let (runtime_id, worker_id) = - runtime_subject_ids(&subject, self.operation)?; - self.control.execute_runtime(WorkspaceRequest::json( - WorkspaceRequestMethod::Post, - format!("/api/w/{}/worker-control/workers/{runtime_id}/{worker_id}/input", self.control.workspace_id()), - serde_json::json!({ - "kind": if self.operation == WorkerOperation::Notify { "notify" } else { "user" }, - "content": content, - }).to_string(), - )).await.map_err(control_tool_error)? - } - } + let subject = worker_control_subject(input.subject)?; + let kind = if self.operation == WorkerOperation::Notify { + WorkerControlInputKind::Notify + } else { + WorkerControlInputKind::User + }; + self.control + .send_input(subject, content, kind) + .await + .map_err(control_tool_error)? } WorkerOperation::Cancel | WorkerOperation::Stop => { - let input = parse::(input_json, self.operation.tool_name())?; - match input.subject { - WorkerSubjectInput::SubWorker { name } => { - if self.operation != WorkerOperation::Stop { - return Err(unsupported_subject(self.operation, "sub_worker")); - } - let subject = subworker_subject(&name)?; - self.control - .ensure_permission(&subject, "stop") - .await - .map_err(control_tool_error)?; - self.control - .stop_subworker(&name) - .await - .map_err(control_tool_error)? + let input = if self.runtime_worker_control { + parse::(input_json, self.operation.tool_name())? + } else { + let input = + parse::(input_json, self.operation.tool_name())?; + WorkerStopInput { + subject: input.subject.into(), + reason: input.reason, } - subject @ WorkerSubjectInput::RuntimeWorker { .. } => { - let (runtime_id, worker_id) = - runtime_subject_ids(&subject, self.operation)?; - let action = if self.operation == WorkerOperation::Cancel { - "cancel" - } else { - "stop" - }; - self.control.execute_runtime(WorkspaceRequest::json( - WorkspaceRequestMethod::Post, - format!("/api/w/{}/worker-control/workers/{runtime_id}/{worker_id}/{action}", self.control.workspace_id()), - serde_json::json!({ "reason": input.reason }).to_string(), - )).await.map_err(control_tool_error)? - } - } + }; + let subject = worker_control_subject(input.subject)?; + let kind = if self.operation == WorkerOperation::Cancel { + WorkerControlStopKind::Cancel + } else { + WorkerControlStopKind::Stop + }; + self.control + .stop_worker(subject, kind, input.reason) + .await + .map_err(control_tool_error)? } WorkerOperation::Restore => { let input = parse::(input_json, "WorkerRestore")?; @@ -803,31 +988,18 @@ impl Tool for WorkspaceWorkerTool { } } -impl WorkspaceWorkerTool { - fn with_subworkers( - &self, - mut response: WorkspaceResponse, - ) -> Result { - if !response.is_success() { - return Ok(response); - } - let mut body: serde_json::Value = - serde_json::from_str(&response.body).map_err(|error| { - ToolError::ExecutionFailed(format!("WorkerList returned invalid JSON: {error}")) - })?; - let items = body - .get_mut("items") - .and_then(serde_json::Value::as_array_mut) - .ok_or_else(|| { - ToolError::ExecutionFailed( - "WorkerList response did not contain an items array".to_string(), - ) - })?; - items.extend(self.control.known_subworkers()); - response.body = serde_json::to_string(&body).map_err(|error| { - ToolError::ExecutionFailed(format!("WorkerList could not encode its response: {error}")) - })?; - Ok(response) +fn worker_control_subject(subject: WorkerSubjectInput) -> Result { + match subject { + WorkerSubjectInput::RuntimeWorker { + runtime_id, + worker_id, + } => Ok(WorkerControlSubject::RuntimeWorker { + runtime_id: authority_id(&runtime_id, "runtime_id")?, + worker_id: authority_id(&worker_id, "worker_id")?, + }), + WorkerSubjectInput::SubWorker { name } => Ok(WorkerControlSubject::SubWorker { + name: authority_id(&name, "name")?, + }), } } @@ -847,16 +1019,6 @@ fn runtime_subject_ids( } } -fn subworker_subject( - name: &str, -) -> Result { - Ok( - super::worker_observation::WorkerObservationSubjectRef::SubWorker { - name: authority_id(name, "name")?, - }, - ) -} - fn unsupported_subject(operation: WorkerOperation, kind: &str) -> ToolError { ToolError::InvalidArgument(format!( "{} does not support subject kind '{kind}'", @@ -911,7 +1073,7 @@ fn render_subworker_stop_summary(summary: &SubWorkerStopSummary) -> String { .as_ref() .map(|stat| format!("+{}/-{} Changes · ", stat.added, stat.deleted)) .unwrap_or_default(); - format!("SubWorkerStop - done\n {tools}\n {changes}{elapsed}",) + format!("WorkerStop - done\n {tools}\n {changes}{elapsed}",) } fn format_elapsed(elapsed_ms: u64) -> String { @@ -925,9 +1087,62 @@ fn format_elapsed(elapsed_ms: u64) -> String { } } +fn sub_worker_tool_declaration(operation: WorkerOperation) -> ToolDeclaration { + let description = match operation { + WorkerOperation::List => "List this Worker's direct SubWorkers.", + WorkerOperation::SendInput => "Send a new user turn to a direct SubWorker.", + WorkerOperation::Stop => { + "Stop a direct SubWorker and release all authority delegated to its child session." + } + _ => unreachable!("unsupported direct SubWorker operation"), + }; + ToolDeclaration::new(operation.tool_name(), description) +} + +fn worker_tool_declaration(operation: WorkerOperation) -> ToolDeclaration { + ToolDeclaration::new(operation.tool_name(), operation.description()) +} + +fn worker_tool_contribution( + operation: WorkerOperation, + control: Arc, + runtime_worker_control: bool, +) -> ToolContribution { + let definition = match operation { + WorkerOperation::List => { + definition::(operation, control, runtime_worker_control) + } + WorkerOperation::Spawn => { + definition::(operation, control, runtime_worker_control) + } + WorkerOperation::SendInput | WorkerOperation::Notify => { + if runtime_worker_control { + definition::(operation, control, true) + } else { + definition::(operation, control, false) + } + } + WorkerOperation::Cancel | WorkerOperation::Stop => { + if runtime_worker_control { + definition::(operation, control, true) + } else { + definition::(operation, control, false) + } + } + WorkerOperation::Restore => { + definition::(operation, control, runtime_worker_control) + } + WorkerOperation::Remove => { + definition::(operation, control, runtime_worker_control) + } + }; + ToolContribution::new(operation.tool_name(), definition) +} + fn definition( operation: WorkerOperation, control: Arc, + runtime_worker_control: bool, ) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(I); @@ -938,6 +1153,7 @@ fn definition( let tool: Arc = Arc::new(WorkspaceWorkerTool { operation, control: control.clone(), + runtime_worker_control, }); (meta, tool) }) @@ -1011,10 +1227,35 @@ mod tests { &self, request: WorkspaceRequest, ) -> Result { + let is_worker_list = request.path.ends_with("/worker-control/workers"); self.requests.lock().unwrap().push(request); Ok(WorkspaceResponse { status: 200, - body: "{}".to_string(), + body: if is_worker_list { + serde_json::json!({ + "items": [ + { + "subject": { + "kind": "runtime_worker", + "runtime_id": "runtime-a", + "worker_id": "worker-a" + }, + "summary": { "status": "idle" } + }, + { + "subject": { + "kind": "runtime_worker", + "runtime_id": "runtime-a", + "worker_id": "worker-a" + }, + "summary": { "status": "idle" } + } + ] + }) + .to_string() + } else { + "{}".to_string() + }, }) } @@ -1036,11 +1277,125 @@ mod tests { } } + #[derive(Debug, Default)] + struct RecordingSubWorkerControl { + sent: Mutex>, + stopped: Mutex>, + } + + #[async_trait] + impl WorkerLifecycleService for RecordingSubWorkerControl { + async fn spawn( + &self, + _request: WorkerLifecycleSpawnRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is disabled in this test".to_string(), + )) + } + } + + #[async_trait] + impl WorkerControlService for RecordingSubWorkerControl { + fn workspace_id(&self) -> &str { + "workspace-test" + } + + async fn list_workers(&self) -> Result { + Ok(WorkspaceResponse { + status: 200, + body: serde_json::json!({ "items": [] }).to_string(), + }) + } + + async fn send_input( + &self, + subject: WorkerControlSubject, + content: String, + kind: WorkerControlInputKind, + ) -> Result { + let WorkerControlSubject::SubWorker { name } = subject else { + return Err(WorkspaceClientError::Request( + "expected subworker subject".to_string(), + )); + }; + if kind != WorkerControlInputKind::User { + return Err(WorkspaceClientError::Request( + "expected user input".to_string(), + )); + } + self.sent.lock().unwrap().push((name, content)); + Ok(WorkspaceResponse { + status: 200, + body: serde_json::json!({ "status": "accepted" }).to_string(), + }) + } + + async fn stop_worker( + &self, + subject: WorkerControlSubject, + kind: WorkerControlStopKind, + _reason: Option, + ) -> Result { + let WorkerControlSubject::SubWorker { name } = subject else { + return Err(WorkspaceClientError::Request( + "expected subworker subject".to_string(), + )); + }; + if kind != WorkerControlStopKind::Stop { + return Err(WorkspaceClientError::Request("expected stop".to_string())); + } + self.stopped.lock().unwrap().push(name); + Ok(WorkspaceResponse { + status: 200, + body: serde_json::json!({ "status": "stopped" }).to_string(), + }) + } + + async fn spawn_worker( + &self, + _request: WorkerLifecycleSpawnRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is disabled in this test".to_string(), + )) + } + + fn remove_runtime_worker( + &self, + _runtime_id: &str, + _worker_id: &str, + _reason: &str, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is disabled in this test".to_string(), + )) + } + + async fn execute_runtime( + &self, + _request: WorkspaceRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Runtime Worker control is disabled in this test".to_string(), + )) + } + + async fn ensure_permission( + &self, + _subject: &crate::feature::builtin::WorkerObservationSubjectRef, + _permission: &str, + ) -> Result<(), WorkspaceClientError> { + Ok(()) + } + } + fn test_control(client: Arc) -> Arc { Arc::new(WorkspaceWorkerControlService { client, workspace_id: "workspace%2Ftest".to_string(), registry: None, + runtime_worker_control: true, }) } @@ -1050,6 +1405,7 @@ mod tests { let tool = WorkspaceWorkerTool { operation: WorkerOperation::Spawn, control: test_control(client.clone()), + runtime_worker_control: true, }; tool.execute( &serde_json::json!({ @@ -1128,6 +1484,190 @@ mod tests { assert!(report.services.providers().is_empty()); } + #[tokio::test] + async fn control_service_deduplicates_worker_list_by_stable_subject() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let control = WorkspaceWorkerControlService { + client: client.clone(), + workspace_id: "workspace%2Ftest".to_string(), + registry: None, + runtime_worker_control: true, + }; + + let response = control.list_workers().await.unwrap(); + + let value: serde_json::Value = serde_json::from_str(&response.body).unwrap(); + assert_eq!(value["items"].as_array().unwrap().len(), 1); + assert_eq!( + value["items"][0]["subject"], + serde_json::json!({ + "kind": "runtime_worker", + "runtime_id": "runtime-a", + "worker_id": "worker-a" + }) + ); + assert_eq!(client.requests.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn control_service_routes_by_subject_without_cross_provider_fallback() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let control = WorkspaceWorkerControlService { + client: client.clone(), + workspace_id: "workspace%2Ftest".to_string(), + registry: None, + runtime_worker_control: true, + }; + + control + .send_input( + WorkerControlSubject::RuntimeWorker { + runtime_id: "runtime-1".to_string(), + worker_id: "worker-1".to_string(), + }, + "continue".to_string(), + WorkerControlInputKind::User, + ) + .await + .unwrap(); + let runtime_request_count = client.requests.lock().unwrap().len(); + assert_eq!(runtime_request_count, 1); + + let error = control + .send_input( + WorkerControlSubject::SubWorker { + name: "missing-reviewer".to_string(), + }, + "continue".to_string(), + WorkerControlInputKind::User, + ) + .await + .unwrap_err(); + assert!(matches!(error, WorkspaceClientError::Request(_))); + assert_eq!( + client.requests.lock().unwrap().len(), + runtime_request_count, + "a subworker failure must not fall through to Workspace control" + ); + } + + #[tokio::test] + async fn sub_worker_control_surface_lists_registry_without_workspace_authority() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let runtime_base = tempfile::tempdir().unwrap(); + let runtime_dir = Arc::new( + crate::runtime::dir::RuntimeDir::create(runtime_base.path(), "sub-worker-control") + .await + .unwrap(), + ); + let registry = SpawnedWorkerRegistry::new(runtime_dir); + let control: Arc = Arc::new(WorkspaceWorkerControlService { + client: client.clone(), + workspace_id: "workspace%2Ftest".to_string(), + registry: Some(registry), + runtime_worker_control: false, + }); + let tool = WorkspaceWorkerTool { + operation: WorkerOperation::List, + control, + runtime_worker_control: false, + }; + + let output = tool + .execute( + "{}", + ToolExecutionContext::new("call-list", "batch-list", 0), + ) + .await + .unwrap(); + + assert!(client.requests.lock().unwrap().is_empty()); + let value: serde_json::Value = + serde_json::from_str(output.content.as_deref().unwrap()).unwrap(); + assert_eq!(value["items"], serde_json::json!([])); + } + + #[tokio::test] + async fn canonical_send_and_stop_tools_route_to_direct_subworker_control() { + let control = Arc::new(RecordingSubWorkerControl::default()); + let send_tool = WorkspaceWorkerTool { + operation: WorkerOperation::SendInput, + control: control.clone(), + runtime_worker_control: false, + }; + let stop_tool = WorkspaceWorkerTool { + operation: WorkerOperation::Stop, + control: control.clone(), + runtime_worker_control: false, + }; + + send_tool + .execute( + r#"{"subject":{"kind":"sub_worker","name":"reviewer"},"content":"continue"}"#, + ToolExecutionContext::new("call-send", "batch-control", 0), + ) + .await + .unwrap(); + stop_tool + .execute( + r#"{"subject":{"kind":"sub_worker","name":"reviewer"},"reason":"review completed"}"#, + ToolExecutionContext::new("call-stop", "batch-control", 1), + ) + .await + .unwrap(); + + assert_eq!( + *control.sent.lock().unwrap(), + vec![("reviewer".to_string(), "continue".to_string())] + ); + assert_eq!( + *control.stopped.lock().unwrap(), + vec!["reviewer".to_string()] + ); + } + + #[tokio::test] + async fn sub_worker_control_surface_rejects_runtime_subject_without_workspace_call() { + let client = Arc::new(RecordingWorkspaceClient::default()); + let control: Arc = Arc::new(WorkspaceWorkerControlService { + client: client.clone(), + workspace_id: "workspace%2Ftest".to_string(), + registry: None, + runtime_worker_control: false, + }); + let permission_error = control + .ensure_permission( + &crate::feature::builtin::WorkerObservationSubjectRef::RuntimeWorker { + runtime_id: "runtime-1".to_string(), + worker_id: "worker-1".to_string(), + }, + "stop", + ) + .await + .unwrap_err(); + assert!(matches!( + permission_error, + WorkspaceClientError::Unavailable(_) + )); + + let tool = WorkspaceWorkerTool { + operation: WorkerOperation::Stop, + control, + runtime_worker_control: false, + }; + + let error = tool + .execute( + r#"{"subject":{"kind":"runtime_worker","runtime_id":"runtime-1","worker_id":"worker-1"},"reason":"not authorized"}"#, + ToolExecutionContext::new("call-stop", "batch-stop", 0), + ) + .await + .unwrap_err(); + + assert!(matches!(error, ToolError::InvalidArgument(_))); + assert!(client.requests.lock().unwrap().is_empty()); + } + #[test] fn worker_service_can_remain_enabled_without_direct_spawn_surface() { let client = Arc::new(RecordingWorkspaceClient::default()); @@ -1146,7 +1686,7 @@ mod tests { } #[test] - fn worker_tool_family_is_distinct_from_sub_worker_tools() { + fn canonical_worker_tool_family_uses_one_subject_based_namespace() { assert_eq!( WorkerOperation::ALL.map(WorkerOperation::tool_name), [ @@ -1254,6 +1794,7 @@ mod tests { WorkspaceWorkerTool { operation, control: test_control(client.clone()), + runtime_worker_control: true, } .execute( &args.to_string(), @@ -1278,6 +1819,7 @@ mod tests { let tool = WorkspaceWorkerTool { operation: WorkerOperation::Remove, control: test_control(client.clone()), + runtime_worker_control: true, }; tool.execute( &serde_json::json!({ @@ -1327,6 +1869,7 @@ mod tests { let tool = WorkspaceWorkerTool { operation: WorkerOperation::Remove, control: test_control(client.clone()), + runtime_worker_control: true, }; for reason in [" ".to_string(), "x".repeat(513)] { let _error = tool @@ -1397,7 +1940,7 @@ mod tests { assert_eq!( output.summary, - "SubWorkerStop - done\n 26 Read, 5 Grep\n +215/-148 Changes · 1m 18s" + "WorkerStop - done\n 26 Read, 5 Grep\n +215/-148 Changes · 1m 18s" ); assert_eq!( serde_json::from_str::(output.content.as_deref().unwrap()) diff --git a/crates/worker/src/feature/builtin/workspace_worker_discovery.rs b/crates/worker/src/feature/builtin/workspace_worker_discovery.rs new file mode 100644 index 00000000..4d8ea268 --- /dev/null +++ b/crates/worker/src/feature/builtin/workspace_worker_discovery.rs @@ -0,0 +1,275 @@ +//! Privileged, read-only discovery of Workspace-visible Workers. +//! +//! This feature deliberately stays separate from the canonical `WorkerList` +//! control-grant surface. Discovery results carry the typed subject needed by a +//! later control operation, but discovery itself grants no control authority. + +use std::sync::Arc; + +use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput}; +use async_trait::async_trait; +use serde::Deserialize; +use serde_json::json; + +use crate::feature::{ + FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureInstructionContribution, + FeatureInstructionDeclaration, FeatureInstructionId, FeatureModule, ToolContribution, + ToolDeclaration, +}; +use crate::worker::{WorkspaceClient, WorkspaceWorkerDiscoveryRequest}; + +const FEATURE_ID: &str = "workspace-worker-discovery"; +const TOOL_NAME: &str = "ListWorkspaceWorkers"; +const DEFAULT_LIMIT: usize = 50; +const MAX_LIMIT: usize = 100; +const INSTRUCTION_ID: &str = "workspace-worker-discovery.policy"; +const PROMPT_REF: &str = "common.workspace_worker_discovery"; +const DESCRIPTION: &str = "List or directly find Workspace-visible Workers through Backend authority. Results include each W-key and the typed runtime_worker subject needed by later Worker control calls, but do not grant control authority."; + +fn instruction() -> FeatureInstructionDeclaration { + FeatureInstructionDeclaration::new( + FeatureInstructionId::builtin(INSTRUCTION_ID), + PROMPT_REF, + "Workspace Worker discovery and control-authority separation", + ) + .expect("static Workspace Worker discovery instruction is valid") +} + +#[derive(Clone)] +pub struct WorkspaceWorkerDiscoveryFeature { + client: Arc, +} + +pub fn workspace_worker_discovery_feature( + client: Arc, +) -> WorkspaceWorkerDiscoveryFeature { + WorkspaceWorkerDiscoveryFeature { client } +} + +impl FeatureModule for WorkspaceWorkerDiscoveryFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin(FEATURE_ID, "Workspace Worker Discovery") + .with_description(DESCRIPTION) + .with_instruction(instruction()) + .with_tool(ToolDeclaration::new(TOOL_NAME, DESCRIPTION)) + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + context + .instructions() + .register(FeatureInstructionContribution::new(instruction()))?; + let client = self.client.clone(); + let definition: ToolDefinition = Arc::new(move || { + ( + ToolMeta::new(TOOL_NAME) + .description(DESCRIPTION) + .input_schema(input_schema()), + Arc::new(ListWorkspaceWorkersTool { + client: client.clone(), + }) as Arc, + ) + }); + context + .tools() + .register(ToolContribution::new(TOOL_NAME, definition))?; + Ok(()) + } +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct ListWorkspaceWorkersInput { + #[serde(default)] + cursor: Option, + #[serde(default)] + limit: Option, + #[serde(default)] + query: Option, +} + +#[derive(Clone)] +struct ListWorkspaceWorkersTool { + client: Arc, +} + +#[async_trait] +impl Tool for ListWorkspaceWorkersTool { + async fn execute( + &self, + input_json: &str, + _ctx: ToolExecutionContext, + ) -> Result { + let input: ListWorkspaceWorkersInput = serde_json::from_str(input_json) + .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; + let limit = input.limit.unwrap_or(DEFAULT_LIMIT); + if !(1..=MAX_LIMIT).contains(&limit) { + return Err(ToolError::InvalidArgument(format!( + "limit must be between 1 and {MAX_LIMIT}" + ))); + } + let query = input.query.map(|query| query.trim().to_string()); + if query.as_deref().is_some_and(str::is_empty) { + return Err(ToolError::InvalidArgument( + "query must not be empty".to_string(), + )); + } + if query.as_ref().is_some_and(|query| query.len() > 128) { + return Err(ToolError::InvalidArgument( + "query must not exceed 128 bytes".to_string(), + )); + } + + let page = self + .client + .list_workspace_workers(WorkspaceWorkerDiscoveryRequest { + cursor: input.cursor, + limit, + query, + }) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + let count = page.workers.len(); + Ok(ToolOutput { + summary: format!("Listed {count} Workspace Worker(s)"), + content: Some(serde_json::to_string_pretty(&page).map_err(|error| { + ToolError::ExecutionFailed(format!( + "encode Workspace Worker discovery result: {error}" + )) + })?), + attachments: Vec::new(), + }) + } +} + +fn input_schema() -> serde_json::Value { + json!({ + "type": "object", + "additionalProperties": false, + "properties": { + "cursor": { + "type": "string", + "description": "Opaque cursor returned by a prior page." + }, + "limit": { + "type": "integer", + "minimum": 1, + "maximum": MAX_LIMIT, + "default": DEFAULT_LIMIT + }, + "query": { + "type": "string", + "maxLength": 128, + "description": "Exact W-key or Worker display name lookup." + } + } + }) +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use workspace_api::{ + WorkspaceWorkerDiscoveryItem, WorkspaceWorkerDiscoveryPage, WorkspaceWorkerSubject, + }; + + use super::*; + use crate::worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse}; + + #[derive(Debug)] + struct RecordingClient { + requests: Mutex>, + result: WorkspaceWorkerDiscoveryPage, + unavailable: bool, + } + + impl WorkspaceClient for RecordingClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-1") + } + + fn kind(&self) -> &str { + "recording" + } + + fn is_available(&self) -> bool { + !self.unavailable + } + + fn execute( + &self, + _request: WorkspaceRequest, + ) -> Result { + panic!("discovery must not use generic Workspace request authority") + } + + fn list_workspace_workers( + &self, + request: WorkspaceWorkerDiscoveryRequest, + ) -> Result { + self.requests.lock().unwrap().push(request); + if self.unavailable { + Err(WorkspaceClientError::Unavailable("denied".to_string())) + } else { + Ok(self.result.clone()) + } + } + } + + fn page() -> WorkspaceWorkerDiscoveryPage { + WorkspaceWorkerDiscoveryPage { + workers: vec![WorkspaceWorkerDiscoveryItem { + subject: WorkspaceWorkerSubject::RuntimeWorker { + runtime_id: "arcadia".to_string(), + worker_id: "worker-1".to_string(), + }, + resource_key: "W-12".to_string(), + display_name: "coder-one".to_string(), + profile: Some("builtin:coder".to_string()), + status: Some("idle".to_string()), + }], + next_cursor: Some("v1:1".to_string()), + } + } + + #[tokio::test] + async fn tool_preserves_typed_subject_and_forwards_lookup() { + let client = Arc::new(RecordingClient { + requests: Mutex::new(Vec::new()), + result: page(), + unavailable: false, + }); + let tool = ListWorkspaceWorkersTool { + client: client.clone(), + }; + let output = tool + .execute( + r#"{"query":" W-12 ","limit":1}"#, + ToolExecutionContext::default(), + ) + .await + .unwrap(); + let value: serde_json::Value = + serde_json::from_str(output.content.as_deref().unwrap()).unwrap(); + assert_eq!(value["workers"][0]["resource_key"], "W-12"); + assert_eq!(value["workers"][0]["subject"]["kind"], "runtime_worker"); + assert_eq!(value["workers"][0]["subject"]["runtime_id"], "arcadia"); + let requests = client.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].query.as_deref(), Some("W-12")); + assert_eq!(requests[0].limit, 1); + } + + #[tokio::test] + async fn missing_backend_authority_fails_closed() { + let client = Arc::new(RecordingClient { + requests: Mutex::new(Vec::new()), + result: page(), + unavailable: true, + }); + let error = ListWorkspaceWorkersTool { client } + .execute("{}", ToolExecutionContext::default()) + .await + .unwrap_err(); + assert!(error.to_string().contains("denied")); + } +} diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 8de12fbd..20042594 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -52,6 +52,6 @@ pub use worker::{ LocalWorkingDirectory, WORKER_INPUT_SUBMISSION_EXTENSION_DOMAIN, Worker, WorkerError, WorkerFilesystemAuthority, WorkerRunResult, WorkerWorkspaceContext, WorkspaceClient, WorkspaceClientError, WorkspaceId, WorkspaceIdError, WorkspacePromptCatalogResolution, - WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse, apply_worker_manifest, - marker_workspace_client, unavailable_workspace_client, + WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse, WorkspaceWorkerDiscoveryRequest, + apply_worker_manifest, marker_workspace_client, unavailable_workspace_client, }; diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index 87986cff..fc3c9dc1 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -901,6 +901,23 @@ mod tests { .unwrap() .contains("BOUNDARY_MARKER") ); - catalog.worker_orchestration_guidance_section().unwrap(); + let orchestration = catalog.worker_orchestration_guidance_section().unwrap(); + for name in [ + "SubWorkerSpawn", + "WorkerList", + "WorkerSendInput", + "WorkerStop", + ] { + assert!( + orchestration.contains(name), + "missing canonical tool {name}" + ); + } + for alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] { + assert!( + !orchestration.contains(alias), + "guidance referenced stale alias {alias}" + ); + } } } diff --git a/crates/worker/src/prompt/system.rs b/crates/worker/src/prompt/system.rs index 8dd78771..ffd9da5a 100644 --- a/crates/worker/src/prompt/system.rs +++ b/crates/worker/src/prompt/system.rs @@ -181,12 +181,19 @@ impl ToolCapabilities { "MemoryReadDocument" => capabilities.memory_read_document = true, "MemoryUpdateDocument" => capabilities.memory_update_document = true, "SubWorkerSpawn" => capabilities.sub_worker_spawn = true, - "SubWorkerSend" => capabilities.sub_worker_send = true, - "SubWorkerStop" => capabilities.sub_worker_stop = true, - "SubWorkerList" => capabilities.sub_worker_list = true, _ => {} } } + if capabilities.sub_worker_spawn { + for name in names { + match name.as_str() { + "WorkerSendInput" => capabilities.sub_worker_send = true, + "WorkerStop" => capabilities.sub_worker_stop = true, + "WorkerList" => capabilities.sub_worker_list = true, + _ => {} + } + } + } capabilities } @@ -316,6 +323,35 @@ fn append_trailing_section( mod tests { use super::*; + #[test] + fn sub_worker_capabilities_follow_the_registered_canonical_control_tools() { + let names = [ + "SubWorkerSpawn", + "WorkerList", + "WorkerSendInput", + "WorkerStop", + ] + .map(str::to_string); + let capabilities = ToolCapabilities::from_tool_names(&names); + assert!(capabilities.sub_worker_management()); + assert!(capabilities.sub_worker_list); + assert!(capabilities.sub_worker_send); + assert!(capabilities.sub_worker_stop); + + let stale_aliases = [ + "SubWorkerSpawn", + "SubWorkerList", + "SubWorkerSend", + "SubWorkerStop", + ] + .map(str::to_string); + let capabilities = ToolCapabilities::from_tool_names(&stale_aliases); + assert!(capabilities.sub_worker_spawn); + assert!(!capabilities.sub_worker_list); + assert!(!capabilities.sub_worker_send); + assert!(!capabilities.sub_worker_stop); + } + #[test] fn rejects_legacy_prefix_relative_and_missing_names() { for reference in ["legacy/custom", "custom.md", "../custom", "missing"] { diff --git a/crates/worker/src/spawn/comm_tools.rs b/crates/worker/src/spawn/comm_tools.rs index 49f9ab78..27e6845f 100644 --- a/crates/worker/src/spawn/comm_tools.rs +++ b/crates/worker/src/spawn/comm_tools.rs @@ -1,226 +1,19 @@ -#![cfg_attr(not(test), allow(dead_code, unused_imports))] - -//! Parent-facing tools for in-process Internal SubWorker sessions. +//! Socket communication retained for the legacy top-level Worker callback protocol. //! -//! Legacy direct-child tool constructors are test-only; production exposes the -//! registry through the unified `worker.control` service and Worker tools. -//! There is no Runtime catalog lookup or child socket transport, so a Worker can operate only on -//! its direct Internal children. The socket helper at the bottom remains solely for the legacy -//! top-level Worker callback protocol and is not part of SubWorker communication. +//! 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::sync::Arc; use std::time::Duration; -use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; -use async_trait::async_trait; use protocol::stream::{JsonLineReader, JsonLineWriter}; use protocol::{Event, Method}; -use serde::{Deserialize, Serialize}; use tokio::net::UnixStream; -use crate::spawn::registry::SpawnedWorkerRegistry; - /// 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); -// --------------------------------------------------------------------------- -// Shared input types -// --------------------------------------------------------------------------- - -#[derive(Debug, Deserialize, schemars::JsonSchema)] -struct NameInput { - /// Name of a previously spawned SubWorker. - name: String, -} - -#[derive(Debug, Deserialize, schemars::JsonSchema)] -#[serde(deny_unknown_fields)] -struct SubWorkerListInput {} - -#[derive(Debug, Serialize)] -struct SubWorkerListItem { - name: String, -} - -struct SubWorkerListTool { - registry: Arc, -} - -#[async_trait] -impl Tool for SubWorkerListTool { - async fn execute( - &self, - input_json: &str, - _ctx: agen::tool::ToolExecutionContext, - ) -> Result { - let _input: SubWorkerListInput = serde_json::from_str(input_json).map_err(|error| { - ToolError::InvalidArgument(format!("invalid SubWorkerList input: {error}")) - })?; - let items = self - .registry - .list_internal() - .into_iter() - .map(|record| SubWorkerListItem { - name: record.worker_name, - }) - .collect::>(); - let count = items.len(); - let content = serde_json::to_string_pretty(&serde_json::json!({ "sub_workers": items })) - .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; - Ok(ToolOutput { - summary: format!("listed {count} child SubWorker(s)"), - content: Some(content), - attachments: Vec::new(), - }) - } -} - -#[cfg(test)] -pub fn sub_worker_list_tool(registry: Arc) -> ToolDefinition { - Arc::new(move || { - let schema = schemars::schema_for!(SubWorkerListInput); - let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); - let meta = ToolMeta::new("SubWorkerList") - .description("List child SubWorkers owned by this Worker. Peer Workers and general Runtime Workers are excluded.") - .input_schema(schema_value); - let tool: Arc = Arc::new(SubWorkerListTool { - registry: registry.clone(), - }); - (meta, tool) - }) -} - -// --------------------------------------------------------------------------- -// SubWorkerSend -// --------------------------------------------------------------------------- - -const SEND_TO_POD_DESCRIPTION: &str = "Send a text message to a previously spawned SubWorker. The SubWorker \ -processes it as a user turn. Fails if the SubWorker is already executing a \ -turn — retry after it finishes. Does not wait for the turn to complete; \ -use worker-observation tools to inspect its committed session."; - -#[derive(Debug, Deserialize, schemars::JsonSchema)] -struct SubWorkerSendInput { - /// Target SubWorker name. - name: String, - /// Text delivered to the SubWorker as the next user message. - message: String, -} - -struct SubWorkerSendTool { - registry: Arc, -} - -#[async_trait] -impl Tool for SubWorkerSendTool { - async fn execute( - &self, - input_json: &str, - _ctx: agen::tool::ToolExecutionContext, - ) -> Result { - let input: SubWorkerSendInput = serde_json::from_str(input_json) - .map_err(|e| ToolError::InvalidArgument(format!("invalid SubWorkerSend input: {e}")))?; - if let Some(record) = self.registry.get_internal(&input.name) { - record.session.send(input.message).await.map_err(|error| { - ToolError::ExecutionFailed(format!("send to `{}`: {error}", input.name)) - })?; - return Ok(ToolOutput { - summary: format!("sent message to `{}`", input.name), - content: None, - attachments: Vec::new(), - }); - } - Err(unknown_worker_err(&input.name)) - } -} - -#[cfg(test)] -pub fn sub_worker_send_tool(registry: Arc) -> ToolDefinition { - Arc::new(move || { - let schema = schemars::schema_for!(SubWorkerSendInput); - let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); - let meta = ToolMeta::new("SubWorkerSend") - .description(SEND_TO_POD_DESCRIPTION) - .input_schema(schema_value); - let tool: Arc = Arc::new(SubWorkerSendTool { - registry: registry.clone(), - }); - (meta, tool) - }) -} - -// --------------------------------------------------------------------------- -// SubWorkerStop -// --------------------------------------------------------------------------- - -const STOP_POD_DESCRIPTION: &str = "Cancel and stop a spawned Internal SubWorker session, remove it from the parent's direct-child registry, and reclaim delegated Write scope."; - -struct SubWorkerStopTool { - registry: Arc, -} - -#[async_trait] -impl Tool for SubWorkerStopTool { - async fn execute( - &self, - input_json: &str, - _ctx: agen::tool::ToolExecutionContext, - ) -> Result { - let input: NameInput = serde_json::from_str(input_json) - .map_err(|e| ToolError::InvalidArgument(format!("invalid SubWorkerStop input: {e}")))?; - if let Some(summary) = self - .registry - .remove_internal(&input.name) - .await - .map_err(|error| ToolError::ExecutionFailed(error.to_string()))? - { - return Ok(ToolOutput { - summary: format!( - "SubWorkerStop - done\n {} tool kind{}\n {}ms", - summary.tool_counts.len(), - if summary.tool_counts.len() == 1 { - "" - } else { - "s" - }, - summary.elapsed_ms, - ), - content: Some( - serde_json::to_string(&summary) - .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?, - ), - attachments: Vec::new(), - }); - } - Err(unknown_worker_err(&input.name)) - } -} - -#[cfg(test)] -pub fn sub_worker_stop_tool(registry: Arc) -> ToolDefinition { - Arc::new(move || { - let schema = schemars::schema_for!(NameInput); - let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); - let meta = ToolMeta::new("SubWorkerStop") - .description(STOP_POD_DESCRIPTION) - .input_schema(schema_value); - let tool: Arc = Arc::new(SubWorkerStopTool { - registry: registry.clone(), - }); - (meta, tool) - }) -} - -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - -fn unknown_worker_err(name: &str) -> ToolError { - ToolError::InvalidArgument(format!("no spawned worker named `{name}`")) -} - /// Connect with a timeout, drain the server's connect-time snapshot, /// write one `Method` line, flush, and close. /// diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index 8e4843d7..9cd314dd 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -1226,14 +1226,11 @@ extract_threshold = 4000 Err(tokio::sync::mpsc::error::TryRecvError::Empty) )); - let context = agen::tool::ToolExecutionContext::direct(); - let list = (crate::spawn::comm_tools::sub_worker_list_tool(registry.clone()))().1; - let listed = list.execute("{}", context.clone()).await.unwrap(); assert!( - listed - .content - .unwrap_or_default() - .contains("reviewer-child") + registry + .list_internal() + .iter() + .any(|record| record.worker_name == "reviewer-child") ); let observation = @@ -1256,13 +1253,11 @@ extract_threshold = 4000 .contains("reviewed") ); - let send = (crate::spawn::comm_tools::sub_worker_send_tool(registry.clone()))().1; - send.execute( - r#"{"name":"reviewer-child","message":"review follow-up"}"#, - context.clone(), - ) - .await - .unwrap(); + record + .session + .send("review follow-up".to_string()) + .await + .unwrap(); assert_eq!( record.session.wait_until_idle().await, crate::internal_worker::InternalWorkerSessionStatus::Idle @@ -1277,12 +1272,11 @@ extract_threshold = 4000 assert!(latest_capture.session.entries.len() > first_capture.session.entries.len()); fail_requests.store(true, Ordering::SeqCst); - send.execute( - r#"{"name":"reviewer-child","message":"trigger terminal failure"}"#, - context.clone(), - ) - .await - .unwrap(); + record + .session + .send("trigger terminal failure".to_string()) + .await + .unwrap(); assert_eq!( record.session.wait_until_idle().await, InternalWorkerSessionStatus::Stopped @@ -1298,10 +1292,13 @@ extract_threshold = 4000 ); assert!(registry.get_internal("reviewer-child").is_some()); - let stop = (crate::spawn::comm_tools::sub_worker_stop_tool(registry.clone()))().1; - stop.execute(r#"{"name":"reviewer-child"}"#, context) - .await - .unwrap(); + assert!( + registry + .remove_internal("reviewer-child") + .await + .unwrap() + .is_some() + ); assert!(registry.get_internal("reviewer-child").is_none()); assert!(spawner_scope.snapshot().is_writable(&workspace_root)); @@ -1315,9 +1312,6 @@ extract_threshold = 4000 .await .unwrap(); assert!(spawner_scope.snapshot().is_writable(&workspace_root)); - drop(list); - drop(send); - drop(stop); drop(observation); drop(tool); drop(registry); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 80dd85b2..bd7e5f22 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -251,6 +251,13 @@ impl WorkspacePromptCatalogResolution { } } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkspaceWorkerDiscoveryRequest { + pub cursor: Option, + pub limit: usize, + pub query: Option, +} + /// Path-free Workspace operation authority injected by Runtime/host code. /// /// Workers receive this trait object rather than a Backend URL. The concrete @@ -263,6 +270,18 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync { fn execute(&self, request: WorkspaceRequest) -> Result; + /// Lists Workspace-visible Workers through dedicated Runtime-owned source + /// proof. Implementations must not fall back to generic Workspace request + /// authority. + fn list_workspace_workers( + &self, + _request: WorkspaceWorkerDiscoveryRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "Workspace Worker discovery authority is unavailable".to_string(), + )) + } + /// Resolve the Workspace's current immutable Prompt projection for future /// operation boundaries. Creation and restore continue to use persisted /// launch/session state; this hook never reconstructs historical prompts. diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 92191f3e..364440b2 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -18,7 +18,8 @@ use workdir::{ use worker::{ Event, Method, Worker, WorkerController, WorkerFilesystemAuthority, WorkerHandle, - WorkerManifest, WorkerStatus, WorkerWorkspaceContext, + WorkerManifest, WorkerStatus, WorkerWorkspaceContext, WorkspaceClient, WorkspaceClientError, + WorkspaceRequest, WorkspaceResponse, }; type TestStore = CombinedStore; @@ -179,9 +180,72 @@ async fn make_worker_with_pwd( make_worker_with_pwd_and_manifest(client, MANIFEST_TOML).await } +#[derive(Debug)] +struct AvailableWorkspaceClient; + +impl WorkspaceClient for AvailableWorkspaceClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-1") + } + + fn kind(&self) -> &str { + "test" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &self, + _request: WorkspaceRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable("test".to_string())) + } +} + +#[derive(Debug)] +struct NoopWorkspaceClient; + +impl WorkspaceClient for NoopWorkspaceClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-test") + } + + fn kind(&self) -> &str { + "test-noop" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &self, + _request: WorkspaceRequest, + ) -> Result { + Err(WorkspaceClientError::Unavailable( + "test client does not execute requests".to_string(), + )) + } +} + async fn make_worker_with_pwd_and_manifest( client: MockClient, manifest_toml: &str, +) -> (Worker, std::path::PathBuf) { + make_worker_with_pwd_manifest_and_workspace_context( + client, + manifest_toml, + WorkerWorkspaceContext::local_filesystem(None), + ) + .await +} + +async fn make_worker_with_pwd_manifest_and_workspace_context( + client: MockClient, + manifest_toml: &str, + workspace_context: WorkerWorkspaceContext, ) -> (Worker, std::path::PathBuf) { let manifest = WorkerManifest::from_toml(manifest_toml).unwrap(); let store_tmp = tempfile::tempdir().unwrap(); @@ -202,16 +266,9 @@ async fn make_worker_with_pwd_and_manifest( let worker = Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(client); let authority = WorkerFilesystemAuthority::local(pwd.clone(), pwd.clone()); - let worker = Worker::new( - manifest, - worker, - store, - WorkerWorkspaceContext::local_filesystem(None), - authority, - scope, - ) - .await - .unwrap(); + let worker = Worker::new(manifest, worker, store, workspace_context, authority, scope) + .await + .unwrap(); (worker, pwd) } @@ -663,9 +720,125 @@ permission = "write" "{} role SubWorker tool exposure mismatch: {names:?}", case.role ); + for control_tool in ["WorkerList", "WorkerSendInput", "WorkerStop"] { + assert_eq!( + names.iter().any(|name| name == control_tool), + case.sub_worker_enabled, + "{} role {control_tool} exposure mismatch: {names:?}", + case.role + ); + } + for stale_alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] { + assert!( + !names.iter().any(|name| name == stale_alias), + "{} role exposed stale alias {stale_alias}: {names:?}", + case.role + ); + } } } +#[tokio::test] +async fn worker_and_sub_worker_features_install_one_canonical_control_surface() { + let manifest = r#" +[worker] +name = "combined-worker-control-feature-test" +pwd = "./" + +[model] +scheme = "anthropic" +model_id = "test-model" + +[engine] +max_tokens = 100 + +[feature.worker] +enabled = true +direct_spawn = false + +[feature.sub_worker] +enabled = true + +[[scope.allow]] +target = "./" +permission = "write" + +[[delegation_scope.allow]] +target = "/tmp" +permission = "write" +"#; + let client = MockClient::new(simple_text_events()); + let client_for_assert = client.clone(); + let worker = make_worker_with_pwd_manifest_and_workspace_context( + client, + manifest, + WorkerWorkspaceContext::with_client(None, Arc::new(NoopWorkspaceClient)), + ) + .await + .0; + let handle = spawn_controller(worker).await; + + handle.send(Method::run_text("Hello")).await.unwrap(); + wait_for_status(&handle, WorkerStatus::Idle).await; + + let request = wait_for_captured_request(&client_for_assert).await; + let names = request_tool_names(&request); + assert!(names.iter().any(|name| name == "SubWorkerSpawn")); + assert!(!names.iter().any(|name| name == "WorkerSpawn")); + for control_tool in ["WorkerList", "WorkerSendInput", "WorkerStop"] { + assert_eq!( + names + .iter() + .filter(|name| name.as_str() == control_tool) + .count(), + 1, + "expected one {control_tool} contribution: {names:?}" + ); + } + for stale_alias in ["SubWorkerList", "SubWorkerSend", "SubWorkerStop"] { + assert!(!names.iter().any(|name| name == stale_alias)); + } +} + +#[tokio::test] +async fn workspace_worker_discovery_requires_workspace_authority_and_stays_separate_from_worker_list() + { + let manifest_toml = r#" +[worker] +name = "workspace-worker-discovery-test" +pwd = "./" + +[model] +scheme = "anthropic" +model_id = "test-model" + +[engine] +max_tokens = 100 + +[feature.workspace_worker_discovery] +enabled = true + +[[scope.allow]] +target = "./" +permission = "write" +"#; + let client = MockClient::new(simple_text_events()); + let client_for_assert = client.clone(); + let (worker, _pwd) = make_worker_with_pwd_manifest_and_workspace_context( + client, + manifest_toml, + WorkerWorkspaceContext::with_client(None, Arc::new(AvailableWorkspaceClient)), + ) + .await; + let handle = spawn_controller(worker).await; + handle.send(Method::run_text("Hello")).await.unwrap(); + wait_for_status(&handle, WorkerStatus::Idle).await; + let request = wait_for_captured_request(&client_for_assert).await; + let names = request_tool_names(&request); + assert!(names.iter().any(|name| name == "ListWorkspaceWorkers")); + assert!(!names.iter().any(|name| name == "WorkerList")); +} + #[tokio::test] async fn sub_worker_feature_exposure_does_not_require_delegation_scope() { let manifest = r#" diff --git a/crates/workspace-api/src/lib.rs b/crates/workspace-api/src/lib.rs index 1b354fdc..898312e3 100644 --- a/crates/workspace-api/src/lib.rs +++ b/crates/workspace-api/src/lib.rs @@ -324,6 +324,35 @@ pub struct WorkerCapabilitySummary { pub can_spawn_followup: bool, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum WorkspaceWorkerSubject { + RuntimeWorker { + runtime_id: String, + worker_id: String, + }, +} + +/// Bounded, model-safe projection used by privileged Workspace Worker discovery. +/// Runtime placement appears only in the typed subject required by Worker +/// control operations; provider and launch internals are intentionally omitted. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkspaceWorkerDiscoveryItem { + pub subject: WorkspaceWorkerSubject, + pub resource_key: String, + pub display_name: String, + pub profile: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub status: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkspaceWorkerDiscoveryPage { + pub workers: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub next_cursor: Option, +} + /// Workspace-authoritative Worker projection. /// /// `resource_key` is required here even though Runtime-internal Worker summaries diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 8fb9cd5d..bc7a73d2 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -4450,7 +4450,9 @@ mod tests { .unwrap(); assert!(companion.feature.manage_workdir.enabled); assert!(companion.feature.sub_worker.enabled); - assert!(!companion.feature.worker.enabled); + assert!(companion.feature.worker.enabled); + assert!(!companion.feature.worker.direct_spawn); + assert!(companion.feature.workspace_worker_discovery.enabled); let coder = archive .resolve_profile("builtin:coder", root.path(), "embedded-test-coder") .unwrap(); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 5f32a7cb..c7874ef0 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -44,7 +44,9 @@ use webauthn_rs::prelude::{ PublicKeyCredential, RegisterPublicKeyCredential, RequestChallengeResponse, Webauthn, WebauthnBuilder, }; -use workdir::http::{WorkdirSessionOperation, WorkdirSessionOperationResult}; +use workdir::http::{ + WorkdirSessionOperation, WorkdirSessionOperationResult, WorkdirTransportError, +}; use workdir::workspace::{ MaterializerKind, WorkingDirectoryCleanupTarget, WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse, @@ -60,11 +62,12 @@ use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeE use workspace_api::{ CreateRemoteRuntimeRequest, CreateRepositorySshCredentialRequest, DeleteRepositorySshCredentialRequest, DeleteRepositorySshHostTrustRequest, - ObjectiveCreateRequest, ObjectiveEditRequest, ObjectiveLinkTicketRequest, - ObjectiveStateRequest, PutRepositorySshHostTrustRequest, RepositoryAccessProjection, - RepositorySshCredential, RepositorySshHostTrust, RotateRepositorySshCredentialRequest, - RuntimeConnectionTestResponse, RuntimeManagementSummary, TICKET_ORCHESTRATION_PLANS_QUERY_PATH, - TICKET_RELATIONS_QUERY_PATH, WorkspaceRuntimeResource, + ObjectiveCreateRequest, ObjectiveEditRequest, ObjectiveLinkTicketRequest, ObjectiveStateRequest, + PutRepositorySshHostTrustRequest, RepositoryAccessProjection, RepositorySshCredential, + RepositorySshHostTrust, RotateRepositorySshCredentialRequest, RuntimeConnectionTestResponse, + RuntimeManagementSummary, TICKET_ORCHESTRATION_PLANS_QUERY_PATH, TICKET_RELATIONS_QUERY_PATH, + WorkspaceRuntimeResource, WorkspaceWorkerDiscoveryItem, WorkspaceWorkerDiscoveryPage, + WorkspaceWorkerSubject, }; use crate::auth::{ @@ -1062,7 +1065,13 @@ async fn authorize_scoped_workspace_request( .map_err(|_| StatusCode::BAD_REQUEST.into_response())?; let digest = worker_runtime::auth::request_body_digest(&body); *request.body_mut() = axum::body::Body::from(body); - let permission = if path.starts_with("/api/runtime/v1/workspaces/") + let permission = if path + .split('?') + .next() + .is_some_and(|path| path.ends_with("/worker-discovery/workers")) + { + worker_runtime::auth::WORKSPACE_WORKER_DISCOVERY_PERMISSION + } else if path.starts_with("/api/runtime/v1/workspaces/") || path.contains("/profile-source-archives/") { worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION @@ -1134,7 +1143,13 @@ async fn authorize_workspace_api_request( }; let digest = worker_runtime::auth::request_body_digest(&body); *request.body_mut() = axum::body::Body::from(body); - let permission = if path.starts_with("/api/runtime/v1/workspaces/") + let permission = if path + .split('?') + .next() + .is_some_and(|path| path.ends_with("/worker-discovery/workers")) + { + worker_runtime::auth::WORKSPACE_WORKER_DISCOVERY_PERMISSION + } else if path.starts_with("/api/runtime/v1/workspaces/") || path.contains("/profile-source-archives/") { worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION @@ -2478,6 +2493,10 @@ fn build_inner_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/worker-observation/session", post(scoped_capture_worker_observation_session), ) + .route( + "/api/w/{workspace_id}/worker-discovery/workers", + get(scoped_discover_workspace_workers), + ) .route( "/api/w/{workspace_id}/workers", get(scoped_list_workers).post(scoped_create_workspace_worker), @@ -6852,12 +6871,56 @@ fn validated_current_worker_attachment( Ok(link) } +#[derive(Debug)] +enum WorkdirOperationApiError { + Api(ApiError), + Provider(WorkdirTransportError), +} + +impl From for WorkdirOperationApiError { + fn from(error: ApiError) -> Self { + Self::Api(error) + } +} + +impl From for WorkdirOperationApiError { + fn from(error: Error) -> Self { + Self::Api(error.into()) + } +} + +impl From for WorkdirOperationApiError { + fn from(error: WorkdirTransportError) -> Self { + Self::Provider(error) + } +} + +impl IntoResponse for WorkdirOperationApiError { + fn into_response(self) -> Response { + match self { + Self::Api(error) => error.into_response(), + Self::Provider(error) => { + let status = StatusCode::from_u16(error.code.http_status()) + .expect("Workdir transport error status is valid"); + let log = ApiErrorLog { + kind: format!("workdir_session_operation_{}", error.code.as_str()), + message: error.message.clone(), + diagnostics: Vec::new(), + }; + let mut response = (status, Json(error)).into_response(); + response.extensions_mut().insert(log); + response + } + } + } +} + async fn scoped_execute_current_worker_workdir_operation( State(api): State, AxumPath(path): AxumPath, headers: HeaderMap, Json(request): Json, -) -> ApiResult> { +) -> std::result::Result, WorkdirOperationApiError> { validate_workspace_scope(&api, &path.workspace_id)?; let worker = current_worker_identity(&api, &path.workspace_id, &headers)?; let expected_session_fence = request.expected_session_fence; @@ -6996,7 +7059,8 @@ async fn current_worker_command_session( external_handle: &CommandHandle, delegations: &[workdir::WorkdirDelegationRequest], expected_session_fence: Option<&str>, -) -> ApiResult<(workdir::AppliedWorkdirDelegation, CommandHandle)> { +) -> std::result::Result<(workdir::AppliedWorkdirDelegation, CommandHandle), WorkdirOperationApiError> +{ let _link = validated_current_worker_attachment(api, worker, expected_session_fence)?; let command = api .workdir_sessions @@ -7004,7 +7068,7 @@ async fn current_worker_command_session( .expect("Workdir session registry lock poisoned") .command(worker, external_handle) .ok_or_else(|| { - ApiError::from(current_worker_workdir_operation_error( + WorkdirOperationApiError::Provider(current_worker_workdir_operation_error( worker, workdir::WorkdirError::UnknownCommand(external_handle.0.clone()), )) @@ -7021,14 +7085,10 @@ async fn current_worker_command_session( } fn current_worker_workdir_operation_error( - worker: &RuntimeWorkerRef, + _worker: &RuntimeWorkerRef, error: workdir::WorkdirError, -) -> Error { - Error::RuntimeOperationFailed { - runtime_id: worker.runtime_id.clone(), - code: "workdir_session_operation_failed".to_string(), - message: error.to_string(), - } +) -> WorkdirTransportError { + WorkdirTransportError::from_workdir_error(&error) } async fn execute_workdir_session_operation( @@ -8099,6 +8159,144 @@ async fn scoped_get_workspace_worker( }) } +#[derive(Debug, Deserialize)] +struct WorkspaceWorkerDiscoveryQuery { + cursor: Option, + limit: Option, + query: Option, +} + +async fn scoped_discover_workspace_workers( + State(api): State, + AxumPath(path): AxumPath, + Query(query): Query, + source: Option>, +) -> Response { + let Some(Extension(_source)) = source else { + return ( + StatusCode::FORBIDDEN, + Json(serde_json::json!({ + "error": "workspace_worker_discovery_source_required" + })), + ) + .into_response(); + }; + + if let Err(error) = validate_workspace_scope(&api, &path.workspace_id) { + return error.into_response(); + } + let limit = query.limit.unwrap_or(50).clamp(1, 100); + let cursor = query.cursor.as_deref(); + let query = match query.query.as_deref().map(str::trim) { + Some("") | None => None, + Some(query) if query.len() <= 128 => Some(query), + Some(_) => { + return ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ + "error": "invalid_workspace_worker_discovery_query" + })), + ) + .into_response(); + } + }; + let filter_fingerprint = workspace_worker_discovery_filter_fingerprint(query); + let offset = match query_cursor_offset(cursor, filter_fingerprint) { + Ok(offset) => offset, + Err(()) => { + return ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ + "error": "invalid_workspace_worker_discovery_cursor" + })), + ) + .into_response(); + } + }; + + let workers = match workers_response(api) { + Ok(workers) => workers, + Err(error) => return error.into_response(), + }; + Json(workspace_worker_discovery_page( + workers.items, + query, + offset, + limit, + filter_fingerprint, + )) + .into_response() +} + +fn workspace_worker_discovery_page( + workers: Vec, + query: Option<&str>, + offset: usize, + limit: usize, + filter_fingerprint: u64, +) -> WorkspaceWorkerDiscoveryPage { + let mut workers = workers + .into_iter() + .filter(|worker| worker.workspace.visibility == "workspace_scoped") + .filter(|worker| { + query.is_none_or(|query| { + worker.resource_key == query + || worker.display_name == query + || worker.label == query + }) + }) + .collect::>(); + workers.sort_by(|left, right| left.resource_key.cmp(&right.resource_key)); + let total = workers.len(); + let page = workers + .into_iter() + .skip(offset) + .take(limit) + .map(|worker| WorkspaceWorkerDiscoveryItem { + subject: WorkspaceWorkerSubject::RuntimeWorker { + runtime_id: worker.runtime_id, + worker_id: worker.worker_id, + }, + resource_key: worker.resource_key, + display_name: worker.display_name, + profile: worker.profile, + status: Some(worker.state), + }) + .collect::>(); + let next_offset = offset.saturating_add(page.len()); + WorkspaceWorkerDiscoveryPage { + workers: page, + next_cursor: (next_offset < total) + .then(|| format!("v1:{next_offset}:{filter_fingerprint:016x}")), + } +} + +fn query_cursor_offset( + cursor: Option<&str>, + filter_fingerprint: u64, +) -> std::result::Result { + let Some(cursor) = cursor else { + return Ok(0); + }; + let mut parts = cursor.split(':'); + match (parts.next(), parts.next(), parts.next(), parts.next()) { + (Some("v1"), Some(offset), Some(fingerprint), None) + if u64::from_str_radix(fingerprint, 16).ok() == Some(filter_fingerprint) => + { + offset.parse().map_err(|_| ()) + } + _ => Err(()), + } +} + +fn workspace_worker_discovery_filter_fingerprint(query: Option<&str>) -> u64 { + use std::hash::{Hash, Hasher}; + + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + query.hash(&mut hasher); + hasher.finish() +} + async fn scoped_list_workers( State(api): State, AxumPath(path): AxumPath, @@ -15588,6 +15786,107 @@ mod tests { assert_eq!(conflict.status(), StatusCode::CONFLICT); } + #[tokio::test] + async fn workdir_operation_errors_preserve_typed_public_status_and_code() { + for (code, expected_status) in [ + ( + workdir::http::WorkdirTransportErrorCode::InvalidRequest, + StatusCode::BAD_REQUEST, + ), + ( + workdir::http::WorkdirTransportErrorCode::NotFound, + StatusCode::NOT_FOUND, + ), + ( + workdir::http::WorkdirTransportErrorCode::UnknownCommand, + StatusCode::NOT_FOUND, + ), + ( + workdir::http::WorkdirTransportErrorCode::Conflict, + StatusCode::CONFLICT, + ), + ( + workdir::http::WorkdirTransportErrorCode::Unsupported, + StatusCode::BAD_REQUEST, + ), + ( + workdir::http::WorkdirTransportErrorCode::Unavailable, + StatusCode::SERVICE_UNAVAILABLE, + ), + ( + workdir::http::WorkdirTransportErrorCode::Internal, + StatusCode::INTERNAL_SERVER_ERROR, + ), + ] { + let response = WorkdirOperationApiError::Provider(WorkdirTransportError { + code, + message: "safe provider message".to_string(), + }) + .into_response(); + assert_eq!(response.status(), expected_status); + let log = response.extensions().get::().unwrap(); + assert_eq!( + log.kind, + format!("workdir_session_operation_{}", code.as_str()) + ); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let decoded: WorkdirTransportError = serde_json::from_slice(&body).unwrap(); + assert_eq!(decoded.code, code); + assert_eq!(decoded.message, "safe provider message"); + } + } + + #[tokio::test] + async fn local_and_remote_validation_errors_share_workspace_classification() { + let worker = RuntimeWorkerRef::new("runtime", "worker"); + let local = current_worker_workdir_operation_error( + &worker, + workdir::WorkdirError::InvalidGlob("[".to_string()), + ); + let remote = current_worker_workdir_operation_error( + &worker, + WorkdirTransportError { + code: workdir::http::WorkdirTransportErrorCode::InvalidRequest, + message: "Workdir operation request is invalid".to_string(), + } + .into_workdir_error(), + ); + assert_eq!(local, remote); + + for public in [local, remote] { + let response = WorkdirOperationApiError::Provider(public).into_response(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let decoded: WorkdirTransportError = serde_json::from_slice(&body).unwrap(); + assert_eq!( + decoded.code, + workdir::http::WorkdirTransportErrorCode::InvalidRequest + ); + } + } + + #[tokio::test] + async fn workdir_operation_error_response_redacts_provider_internal_details() { + let error = workdir::WorkdirError::Io { + path: PathBuf::from("/host/private/worktree/secret.txt"), + source: std::io::Error::new(std::io::ErrorKind::PermissionDenied, "token=secret"), + }; + let public = current_worker_workdir_operation_error( + &RuntimeWorkerRef::new("runtime", "worker"), + error, + ); + let response = WorkdirOperationApiError::Provider(public).into_response(); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!(!text.contains("/host/private")); + assert!(!text.contains("token=secret")); + let decoded: WorkdirTransportError = serde_json::from_str(&text).unwrap(); + assert_eq!( + decoded.code, + workdir::http::WorkdirTransportErrorCode::Internal + ); + } + #[test] fn backend_worker_projection_preserves_missing_rows_links_and_redacts_paths() { let worker = WorkerRegistryRecord { @@ -22023,6 +22322,247 @@ mod tests { assert_eq!(valid.status(), StatusCode::OK); } + #[test] + fn workspace_worker_discovery_filters_internal_workers_and_paginates_typed_subjects() { + fn summary( + key: &str, + worker_id: &str, + name: &str, + visibility: &str, + ) -> workspace_api::WorkerSummary { + workspace_api::WorkerSummary { + runtime_id: "runtime-test".to_string(), + worker_id: worker_id.to_string(), + resource_key: key.to_string(), + host_id: "host-test".to_string(), + display_name: name.to_string(), + label: name.to_string(), + profile: Some("builtin:coder".to_string()), + singleton_key: None, + tags: Vec::new(), + workspace: workspace_api::WorkerWorkspaceSummary { + visibility: visibility.to_string(), + identity: "workspace-test".to_string(), + workspace_id: Some(TEST_WORKSPACE_ID.to_string()), + }, + state: "idle".to_string(), + last_seen_at: None, + pinned: false, + retention_state: "normal".to_string(), + implementation: workspace_api::WorkerImplementationSummary { + kind: "remote".to_string(), + display_hint: "remote".to_string(), + }, + capabilities: workspace_api::WorkerCapabilitySummary { + can_stop: true, + can_spawn_followup: false, + }, + working_directory: None, + diagnostics: Vec::new(), + } + } + + let fingerprint = workspace_worker_discovery_filter_fingerprint(None); + let first = workspace_worker_discovery_page( + vec![ + summary("W-3", "internal", "service", "backend_internal"), + summary("W-2", "worker-b", "beta", "workspace_scoped"), + summary("W-1", "worker-a", "alpha", "workspace_scoped"), + ], + None, + 0, + 1, + fingerprint, + ); + assert_eq!(first.workers.len(), 1); + assert_eq!(first.workers[0].resource_key, "W-1"); + assert_eq!( + first.workers[0].subject, + WorkspaceWorkerSubject::RuntimeWorker { + runtime_id: "runtime-test".to_string(), + worker_id: "worker-a".to_string(), + } + ); + let cursor = first.next_cursor.as_deref().unwrap(); + let offset = query_cursor_offset(Some(cursor), fingerprint).unwrap(); + let second = workspace_worker_discovery_page( + vec![ + summary("W-3", "internal", "service", "backend_internal"), + summary("W-2", "worker-b", "beta", "workspace_scoped"), + summary("W-1", "worker-a", "alpha", "workspace_scoped"), + ], + None, + offset, + 1, + fingerprint, + ); + assert_eq!(second.workers[0].resource_key, "W-2"); + assert!(second.next_cursor.is_none()); + + let lookup_fingerprint = workspace_worker_discovery_filter_fingerprint(Some("beta")); + let lookup = workspace_worker_discovery_page( + vec![summary("W-2", "worker-b", "beta", "workspace_scoped")], + Some("beta"), + 0, + 10, + lookup_fingerprint, + ); + assert_eq!(lookup.workers[0].resource_key, "W-2"); + assert!(query_cursor_offset(Some(cursor), lookup_fingerprint).is_err()); + } + + #[tokio::test] + async fn workspace_worker_discovery_does_not_create_or_expand_control_grants() { + let workspace = tempfile::tempdir().unwrap(); + let api = test_api(workspace.path()).await; + seed_worker_source_member(&api, "runtime-test", "worker-caller"); + seed_worker_source_member(&api, "runtime-test", "worker-target"); + let controller = RuntimeWorkerRef::new("runtime-test", "worker-caller"); + let target = RuntimeWorkerRef::new("runtime-test", "worker-target"); + let before = api + .store + .list_active_worker_control_grants(TEST_WORKSPACE_ID, &controller, 10) + .unwrap(); + assert!(before.is_empty()); + + let discovered = workspace_worker_discovery_page( + vec![workspace_api::WorkerSummary { + runtime_id: target.runtime_id.clone(), + worker_id: target.worker_id.clone(), + resource_key: "W-2".to_string(), + host_id: "host-test".to_string(), + display_name: "ungranted-peer".to_string(), + label: "ungranted-peer".to_string(), + profile: Some("builtin:coder".to_string()), + singleton_key: None, + tags: Vec::new(), + workspace: workspace_api::WorkerWorkspaceSummary { + visibility: "workspace_scoped".to_string(), + identity: "workspace-test".to_string(), + workspace_id: Some(TEST_WORKSPACE_ID.to_string()), + }, + state: "idle".to_string(), + last_seen_at: None, + pinned: false, + retention_state: "normal".to_string(), + implementation: workspace_api::WorkerImplementationSummary { + kind: "remote".to_string(), + display_hint: "remote".to_string(), + }, + capabilities: workspace_api::WorkerCapabilitySummary { + can_stop: true, + can_spawn_followup: false, + }, + working_directory: None, + diagnostics: Vec::new(), + }], + None, + 0, + 10, + workspace_worker_discovery_filter_fingerprint(None), + ); + let WorkspaceWorkerSubject::RuntimeWorker { + runtime_id, + worker_id, + } = &discovered.workers[0].subject; + assert_eq!(discovered.workers[0].resource_key, "W-2"); + let discovered_target = RuntimeWorkerRef::new(runtime_id, worker_id); + let error = authorize_known_worker_permission( + &api, + TEST_WORKSPACE_ID, + &controller, + &discovered_target, + "send_input", + ) + .unwrap_err(); + assert!(matches!( + error, + Error::UnknownWorker { worker } if worker == target + )); + + let after = api + .store + .list_active_worker_control_grants(TEST_WORKSPACE_ID, &controller, 10) + .unwrap(); + assert!(after.is_empty()); + } + + #[tokio::test] + async fn workspace_worker_discovery_requires_dedicated_source_proof() { + let workspace = tempfile::tempdir().unwrap(); + let mut api = test_api(workspace.path()).await; + let identity = + worker_runtime::auth::RuntimeIdentityMaterial::generate("runtime-test").unwrap(); + configure_runtime_request_auth(&mut api, &identity, "runtime-test"); + seed_worker_source_member(&api, "runtime-test", "worker-caller"); + seed_worker_source_member(&api, "runtime-test", "worker-target"); + let target_key = api + .store + .resource_key( + TEST_WORKSPACE_ID, + WorkspaceResourceKind::Worker, + "worker-target", + ) + .unwrap() + .unwrap(); + let target = format!( + "/api/w/{TEST_WORKSPACE_ID}/worker-discovery/workers?limit=1&query={target_key}" + ); + let signer = worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&identity); + let issue = |permission| { + signer + .issue( + "server-test", + TEST_WORKSPACE_ID, + Some("worker-caller"), + permission, + "GET", + &target, + b"", + i64::try_from(worker_runtime::auth::unix_now_seconds()).unwrap_or(i64::MAX), + 30, + ) + .unwrap() + }; + let app = build_router(api); + + let generic = app + .clone() + .oneshot( + Request::builder() + .uri(&target) + .header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + issue(worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION), + ) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(generic.status(), StatusCode::UNAUTHORIZED); + + let response = app + .oneshot( + Request::builder() + .uri(&target) + .header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + issue(worker_runtime::auth::WORKSPACE_WORKER_DISCOVERY_PERMISSION), + ) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body: WorkspaceWorkerDiscoveryPage = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert!(body.workers.is_empty()); + assert!(body.next_cursor.is_none()); + } + #[tokio::test] async fn internal_resource_fetch_rest_returns_typed_missing_resource() { let workspace = tempfile::tempdir().unwrap(); diff --git a/docs/design/session-observation.md b/docs/design/session-observation.md index ae7eb9fa..24ce5dc0 100644 --- a/docs/design/session-observation.md +++ b/docs/design/session-observation.md @@ -44,4 +44,4 @@ Observation is read-only evidence access. It does not authorize Ticket, Memory, ## SubWorker output -SubWorkers no longer expose a separate output cursor tool. `SubWorkerList`, `SubWorkerSend`, and `SubWorkerStop` retain parent-owned lifecycle control, while committed child output is read through `worker-observation`. Turn-completion notifications carry no transcript and only tell the parent to inspect the authoritative committed session at a natural boundary. +SubWorkers no longer expose a separate output cursor or lifecycle tool family. `WorkerList`, `WorkerSendInput`, and `WorkerStop` retain parent-owned lifecycle control through a `{ kind: "sub_worker", name }` subject, while committed child output is read through `worker-observation`. Turn-completion notifications carry no transcript and only tell the parent to inspect the authoritative committed session at a natural boundary. diff --git a/resources/profiles/base.dcdl b/resources/profiles/base.dcdl index 2e960d15..c91afed8 100644 --- a/resources/profiles/base.dcdl +++ b/resources/profiles/base.dcdl @@ -28,6 +28,7 @@ feature = { image = { enabled = true; }; sub_worker = { enabled = false; }; worker = { enabled = false; }; + workspace_worker_discovery = { enabled = false; }; objective = { enabled = true; }; ticket = { enabled = true; authoring = true; thread = true; }; merge_request = { diff --git a/resources/profiles/companion.dcdl b/resources/profiles/companion.dcdl index d3720e3a..85f4fbee 100644 --- a/resources/profiles/companion.dcdl +++ b/resources/profiles/companion.dcdl @@ -8,6 +8,8 @@ import "./base.dcdl" // { memory = { enabled = true; }; web = { enabled = true; }; sub_worker = { enabled = true; }; + worker = { enabled = true; direct_spawn = false; }; + workspace_worker_discovery = { enabled = true; }; manage_workdir = { enabled = true; }; ticket = { enabled = true; authoring = true; thread = true; }; }; diff --git a/resources/prompts/catalog.dcdl b/resources/prompts/catalog.dcdl index caa82038..0f7a2a90 100644 --- a/resources/prompts/catalog.dcdl +++ b/resources/prompts/catalog.dcdl @@ -10,6 +10,7 @@ commonMergeRequest = import "./common/merge-request.md"; commonTickets = import "./common/tickets.md"; commonToolUsage = import "./common/tool-usage.md"; commonWorkerObservation = import "./common/worker-observation.md"; +commonWorkspaceWorkerDiscovery = import "./common/workspace-worker-discovery.md"; commonWorkerOrchestration = import "./common/worker-orchestration.md"; commonWorkspace = import "./common/workspace.md"; commonWriting = import "./common/writing.md"; @@ -40,6 +41,7 @@ in tickets = commonTickets.content; tool_usage = commonToolUsage.content; worker_observation = commonWorkerObservation.content; + workspace_worker_discovery = commonWorkspaceWorkerDiscovery.content; worker_orchestration = commonWorkerOrchestration.content; workspace = commonWorkspace.content; writing = commonWriting.content; diff --git a/resources/prompts/common/worker-orchestration.md b/resources/prompts/common/worker-orchestration.md index 6bf73286..c1f815d1 100644 --- a/resources/prompts/common/worker-orchestration.md +++ b/resources/prompts/common/worker-orchestration.md @@ -2,7 +2,7 @@ --- ## SubWorker orchestration -When SubWorker-management tools are available, SubWorker notifications are background signals for the parent Worker to handle at a natural stopping point. Do not ignore routine follow-up, but do not interrupt the current user request unnecessarily. +When SubWorker-management tools are available, create direct children with `SubWorkerSpawn`, discover them with `WorkerList`, continue them with `WorkerSendInput`, and release their delegated authority with `WorkerStop`. Pass the exact `{ kind: "sub_worker", name }` subject returned by `WorkerList`; do not invent direct-only aliases. SubWorker notifications are background signals for the parent Worker to handle at a natural stopping point. Do not ignore routine follow-up, but do not interrupt the current user request unnecessarily. The parent Worker does not need to keep a turn open or call tools solely to wait for a notification. Do not use `sleep` or polling loops just to wait for SubWorker output; if there is no useful immediate work, return control and handle the SubWorker when notified or when the user next asks. diff --git a/resources/prompts/common/workspace-worker-discovery.md b/resources/prompts/common/workspace-worker-discovery.md new file mode 100644 index 00000000..e4f43743 --- /dev/null +++ b/resources/prompts/common/workspace-worker-discovery.md @@ -0,0 +1,6 @@ +Workspace Worker discovery is separate from Worker control authority. + +- Use `ListWorkspaceWorkers` to list accessible Workspace Workers or directly find one by its exact `W-*` key or display name. +- Reuse the returned typed `subject` unchanged when a later `Worker*` control tool requires a target. Do not guess `runtime_id` or `worker_id`. +- Discovery does not grant control. `WorkerList` remains the authoritative list of Workers this Worker may control, and a discovered Worker can still be rejected by every control operation. +- Results exclude service-private/Internal Workers and omit provider, launch, credential, and capability internals.