chore: merge develop into companion integration

# Conflicts:
#	crates/workspace-server/src/server.rs

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();

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<Option<Override>, 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<Option<Types>, 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<PathBuf>,
@@ -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<bool, FsError> {
+    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,

crates/manifest/src/config.rs

@@ -92,6 +92,8 @@ pub struct FeatureConfigPartial {
     #[serde(default)]
     pub worker: Option<WorkerFeatureConfigPartial>,
     #[serde(default)]
+    pub workspace_worker_discovery: Option<FeatureFlagConfigPartial>,
+    #[serde(default)]
     pub objective: Option<FeatureFlagConfigPartial>,
     #[serde(default)]
     pub manage_workdir: Option<FeatureFlagConfigPartial>,
@@ -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<FeatureConfigPartial> 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<FeatureConfig> 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()),

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(),

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]

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<String>,
     #[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<dyn Tool> = Arc::new(GrepTool {
             session: session.clone(),

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("<remote>".into()),
             Code::Conflict => WorkdirError::Conflict(self.message),
-            Code::Unsupported => WorkdirError::Unavailable(self.message),
+            Code::Unsupported => WorkdirError::UnsupportedOperation(self.message),
             Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".to_string()),
             Code::InvalidRequest => WorkdirError::InvalidArgument(self.message),
             Code::Unavailable => WorkdirError::Unavailable(self.message),
@@ -517,13 +545,13 @@ mod client {
             .json::<WorkdirTransportError>()
             .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(

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),
 

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,

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.";

crates/worker-runtime/src/http_server.rs

@@ -1721,17 +1721,8 @@ impl RuntimeHttpWorkdirError {
 impl From<workdir::WorkdirError> 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 {

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<WorkspaceResponse, WorkspaceClientError> {
+        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<WorkspaceResponse, WorkspaceClientError> {
-        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<workspace_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
+        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<Duration>,
+    permission: &'static str,
     request: WorkspaceRequest,
 ) -> Result<WorkspaceResponse, WorkspaceClientError> {
     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();

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" }

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,
             ),
         );

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};

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::<workdir::http::WorkdirTransportError>(&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![

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<dyn WorkspaceClient>,
+}
+
+pub fn workspace_worker_discovery_feature(
+    client: Arc<dyn WorkspaceClient>,
+) -> 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<dyn Tool>,
+            )
+        });
+        context
+            .tools()
+            .register(ToolContribution::new(TOOL_NAME, definition))?;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ListWorkspaceWorkersInput {
+    #[serde(default)]
+    cursor: Option<String>,
+    #[serde(default)]
+    limit: Option<usize>,
+    #[serde(default)]
+    query: Option<String>,
+}
+
+#[derive(Clone)]
+struct ListWorkspaceWorkersTool {
+    client: Arc<dyn WorkspaceClient>,
+}
+
+#[async_trait]
+impl Tool for ListWorkspaceWorkersTool {
+    async fn execute(
+        &self,
+        input_json: &str,
+        _ctx: ToolExecutionContext,
+    ) -> Result<ToolOutput, ToolError> {
+        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<Vec<WorkspaceWorkerDiscoveryRequest>>,
+        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<WorkspaceResponse, WorkspaceClientError> {
+            panic!("discovery must not use generic Workspace request authority")
+        }
+
+        fn list_workspace_workers(
+            &self,
+            request: WorkspaceWorkerDiscoveryRequest,
+        ) -> Result<WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
+            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"));
+    }
+}

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,
 };

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}"
+            );
+        }
     }
 }

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"] {

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<SpawnedWorkerRegistry>,
-}
-
-#[async_trait]
-impl Tool for SubWorkerListTool {
-    async fn execute(
-        &self,
-        input_json: &str,
-        _ctx: agen::tool::ToolExecutionContext,
-    ) -> Result<ToolOutput, ToolError> {
-        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::<Vec<_>>();
-        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<SpawnedWorkerRegistry>) -> 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<dyn Tool> = 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<SpawnedWorkerRegistry>,
-}
-
-#[async_trait]
-impl Tool for SubWorkerSendTool {
-    async fn execute(
-        &self,
-        input_json: &str,
-        _ctx: agen::tool::ToolExecutionContext,
-    ) -> Result<ToolOutput, ToolError> {
-        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<SpawnedWorkerRegistry>) -> 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<dyn Tool> = 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<SpawnedWorkerRegistry>,
-}
-
-#[async_trait]
-impl Tool for SubWorkerStopTool {
-    async fn execute(
-        &self,
-        input_json: &str,
-        _ctx: agen::tool::ToolExecutionContext,
-    ) -> Result<ToolOutput, ToolError> {
-        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<SpawnedWorkerRegistry>) -> 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<dyn Tool> = 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.
 ///

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);

crates/worker/src/worker.rs

@@ -251,6 +251,13 @@ impl WorkspacePromptCatalogResolution {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WorkspaceWorkerDiscoveryRequest {
+    pub cursor: Option<String>,
+    pub limit: usize,
+    pub query: Option<String>,
+}
+
 /// 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<WorkspaceResponse, WorkspaceClientError>;
 
+    /// 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<workspace_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
+        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.

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<FsStore, FsWorkerStore>;
@@ -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<WorkspaceResponse, WorkspaceClientError> {
+        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<WorkspaceResponse, WorkspaceClientError> {
+        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<MockClient, TestStore>, 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<MockClient, TestStore>, 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#"

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<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub status: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct WorkspaceWorkerDiscoveryPage {
+    pub workers: Vec<WorkspaceWorkerDiscoveryItem>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub next_cursor: Option<String>,
+}
+
 /// Workspace-authoritative Worker projection.
 ///
 /// `resource_key` is required here even though Runtime-internal Worker summaries

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();

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<ApiError> for WorkdirOperationApiError {
+    fn from(error: ApiError) -> Self {
+        Self::Api(error)
+    }
+}
+
+impl From<Error> for WorkdirOperationApiError {
+    fn from(error: Error) -> Self {
+        Self::Api(error.into())
+    }
+}
+
+impl From<WorkdirTransportError> 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<WorkspaceApi>,
     AxumPath(path): AxumPath<ScopedWorkspacePath>,
     headers: HeaderMap,
     Json(request): Json<WorkspaceWorkdirSessionOperationRequest>,
-) -> ApiResult<Json<WorkdirSessionOperationResult>> {
+) -> std::result::Result<Json<WorkdirSessionOperationResult>, 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<String>,
+    limit: Option<usize>,
+    query: Option<String>,
+}
+
+async fn scoped_discover_workspace_workers(
+    State(api): State<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+    Query(query): Query<WorkspaceWorkerDiscoveryQuery>,
+    source: Option<Extension<crate::worker_source::VerifiedRuntimeRequestSource>>,
+) -> 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<workspace_api::WorkerSummary>,
+    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::<Vec<_>>();
+    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::<Vec<_>>();
+    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<usize, ()> {
+    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<WorkspaceApi>,
     AxumPath(path): AxumPath<ScopedWorkspacePath>,
@@ -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::<ApiErrorLog>().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();