feat(backends): resolve external CLI descriptors in the runtime

docs/README.md

@@ -22,6 +22,7 @@ New to NEEDLE? Start with these guides to get up and running quickly.
 | **[Quickstart Example](examples/quickstart/README.md)** | End-to-end walkthrough from empty workspace to first closed bead |
 | **[Quickstart Expected Output](examples/quickstart/expected-output.md)** | Reference output for the quickstart example |
 | **[Configuration Guide](configuration.md)** | Complete configuration reference (config.yaml, environment variables, CLI args) |
+| **[External Bead Backends](external-bead-backends.md)** | Install and explicitly bind an operator-owned CLI descriptor |
 | **[Agent Adapter Authoring](plan/plan.md#agent-adapters)** | How to write custom agent adapters (invoke_template schema) |
 | **[Plugin: Claude Interactive](templates/AGENTS-needle.md)** | NEEDLE workspace template for Claude Code sessions |
 

docs/external-bead-backends.md

@@ -0,0 +1,82 @@
+# External bead backends
+
+NEEDLE can bind a workspace to an operator-installed bead CLI descriptor. The
+workspace selects the descriptor by name; it cannot choose the descriptor
+directory or substitute command templates.
+
+## Install and select a descriptor
+
+Place one YAML descriptor in:
+
+```text
+~/.config/needle/bead-backends/<name>.yaml
+```
+
+Then select its `name` in the workspace's committed `.needle.yaml`:
+
+```yaml
+bead_cli:
+  backend: example-remote
+```
+
+An operator may set `bead_cli.path` to an explicit executable path. Otherwise
+NEEDLE resolves the descriptor's `binary` on `PATH`, then its `detect_paths`.
+The descriptor and resolved executable become one immutable runtime binding.
+Changing `bead_cli.backend`, `bead_cli.path`, or a descriptor requires a worker
+restart; a running worker does not silently switch stores.
+
+Descriptors are trusted operator configuration. Do not install descriptors
+from a repository under work or construct their operations from work-item
+content. Two operator files with the same descriptor name are rejected as
+ambiguous. A user file may intentionally replace a shipped built-in by name.
+
+## Contract
+
+External descriptors currently implement the same complete operation set as a
+shipped backend. Validation rejects missing operations, unsupported strategy
+names, invalid regular expressions, and unresolvable placeholders before any
+store command runs.
+
+NEEDLE executes `version_command` first, with a five-second timeout and bounded
+stdout/stderr. Its output must match `identity_pattern`. A failed, timed-out, or
+mismatched identity check prevents the store from opening. Native bead-rs
+capability probing remains specific to the `bead-rs` descriptor.
+
+Use `strategy: atomic_command` for a descriptor whose `claim` operation performs
+one atomic claim. The command returns one JSON object:
+
+```json
+{"outcome":"claimed","bead_id":"work-123"}
+{"outcome":"race_lost","claimed_by":"worker-b"}
+{"outcome":"not_claimable","reason":"paused"}
+{"outcome":"error","reason":"store unavailable"}
+```
+
+Malformed JSON, unknown outcomes, command failures, and missing required fields
+are errors. They are never interpreted as an empty queue. Arguments are passed
+as an argv vector, so IDs and actor names containing spaces or punctuation are
+not shell-split.
+
+Prompt fragments and canary lookups render from the same resolved descriptor.
+The worker, supervisor, doctor, validation commands, and cross-workspace lookup
+all use explicit workspace bindings rather than rediscovering a CLI.
+
+## Local example
+
+The hermetic fixture is executable without credentials or network access:
+
+```sh
+cargo test --test external_backend_runtime
+```
+
+It creates a temporary operator descriptor directory and fake CLI, selects the
+external backend through normal configuration, verifies identity before store
+mutation, exercises descriptor-rendered claim/release commands, and races two
+claims to prove exactly one winner.
+
+## Current boundary
+
+This extension drives CLI-shaped stores. It does not yet add opaque remote lease
+context, renewal, fenced terminal mutations, or capability-aware subsets of the
+operation contract. Those require the separate remote-lifecycle change; an
+external descriptor must not emulate them by pretending to be bead-rs.

src/bead_store/backend.rs

@@ -134,6 +134,15 @@ pub struct BeadBackend {
     pub error_markers: BeadBackendErrorMarkers,
 }
 
+/// One validated descriptor together with the operator-controlled source that
+/// supplied it. Runtime consumers retain this provenance instead of
+/// rediscovering a similarly named descriptor later.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct LoadedBeadBackend {
+    pub descriptor: BeadBackend,
+    pub source: PathBuf,
+}
+
 fn default_version_command() -> Vec<String> {
     vec!["--version".to_string()]
 }
@@ -409,16 +418,35 @@ fn allowed_placeholders(operation: &str) -> &'static [&'static str] {
     }
 }
 
-/// Load built-ins plus user YAML descriptors, overriding built-ins by name.
+/// Load built-ins plus user YAML descriptors. A user descriptor may override a
+/// built-in by name, but two operator files defining the same name are
+/// rejected as ambiguous.
 pub fn load_bead_backends(
     dir: &Path,
     built_ins: &[BeadBackend],
 ) -> Result<HashMap<String, BeadBackend>> {
+    Ok(load_bead_backends_with_sources(dir, built_ins)?
+        .into_iter()
+        .map(|(name, loaded)| (name, loaded.descriptor))
+        .collect())
+}
+
+/// Load and validate backend descriptors while preserving their source.
+pub fn load_bead_backends_with_sources(
+    dir: &Path,
+    built_ins: &[BeadBackend],
+) -> Result<HashMap<String, LoadedBeadBackend>> {
     let mut backends = HashMap::new();
     for backend in built_ins {
         let source = PathBuf::from(format!("<builtin:{}>", backend.name));
         backend.validate(&source)?;
-        backends.insert(backend.name.clone(), backend.clone());
+        backends.insert(
+            backend.name.clone(),
+            LoadedBeadBackend {
+                descriptor: backend.clone(),
+                source,
+            },
+        );
     }
 
     if !dir.exists() {
@@ -449,11 +477,33 @@ pub fn load_bead_backends(
         let backend: BeadBackend = serde_yaml::from_str(&text)
             .with_context(|| format!("invalid YAML in bead backend file: {}", path.display()))?;
         backend.validate(&path)?;
-        backends.insert(backend.name.clone(), backend);
+        if let Some(previous) = backends.get(&backend.name) {
+            if !is_builtin_source(&previous.source) {
+                bail!(
+                    "ambiguous bead backend descriptor '{}': defined by both {} and {}",
+                    backend.name,
+                    previous.source.display(),
+                    path.display()
+                );
+            }
+        }
+        backends.insert(
+            backend.name.clone(),
+            LoadedBeadBackend {
+                descriptor: backend,
+                source: path,
+            },
+        );
     }
     Ok(backends)
 }
 
+fn is_builtin_source(source: &Path) -> bool {
+    source
+        .to_str()
+        .is_some_and(|value| value.starts_with("<builtin:") && value.ends_with('>'))
+}
+
 /// Shipped descriptors. User files can replace this descriptor by name.
 pub fn builtin_bead_backends() -> Vec<BeadBackend> {
     vec![builtin_bead_rs()]

src/bead_store/cli_store.rs

@@ -16,8 +16,8 @@ use crate::types::{Bead, BeadId, BeadStatus, ClaimResult};
 
 use super::{
     execute_create_id_strategy, execute_labels_strategy, spawn_with_etxtbsy_retry_child,
-    validate_strategy_name, BeadBackend, BeadOperationSpec, BeadStore, ClaimStrategy, Filters,
-    NewChild, ParseShape, ParsedStrategy, RepairReport,
+    validate_strategy_name, BeadBackend, BeadOperationSpec, BeadPromptCommands, BeadStore,
+    ClaimStrategy, Filters, NewChild, ParseShape, ParsedStrategy, RepairReport,
 };
 
 const DEFAULT_TIMEOUT_SECS: u64 = 30;
@@ -323,19 +323,77 @@ impl CliBeadStore {
 
     async fn claim_auto_inner(&self, actor: &str) -> Result<ClaimResult> {
         let values = HashMap::from([("actor", actor.to_string())]);
-        let stdout = self.run_operation("claim_auto", &values).await?;
+        self.claim_from_operation("claim_auto", &values).await
+    }
+
+    async fn claim_from_operation(
+        &self,
+        operation: &str,
+        values: &HashMap<&str, String>,
+    ) -> Result<ClaimResult> {
+        let stdout = self.run_operation(operation, values).await?;
         let json: serde_json::Value = serde_json::from_str(&stdout)
-            .with_context(|| format!("{} claim returned invalid JSON", self.backend.name))?;
+            .with_context(|| format!("{} {operation} returned invalid JSON", self.backend.name))?;
         let bead_id = json
             .get("bead_id")
             .or_else(|| json.pointer("/data/bead_id"))
             .and_then(serde_json::Value::as_str)
             .filter(|id| !id.is_empty());
-        match bead_id {
-            Some(id) => Ok(ClaimResult::Claimed(self.show(&BeadId::from(id)).await?)),
-            None => Ok(ClaimResult::NotClaimable {
-                reason: "no beads available".to_string(),
+        match json.get("outcome").and_then(serde_json::Value::as_str) {
+            Some("claimed") => {
+                let id = bead_id.ok_or_else(|| {
+                    anyhow::anyhow!(
+                        "{} {operation} claimed response omitted bead_id",
+                        self.backend.name
+                    )
+                })?;
+                Ok(ClaimResult::Claimed(self.show(&BeadId::from(id)).await?))
+            }
+            Some("race_lost") => {
+                let claimed_by = json
+                    .get("claimed_by")
+                    .and_then(serde_json::Value::as_str)
+                    .filter(|value| !value.is_empty())
+                    .ok_or_else(|| {
+                        anyhow::anyhow!(
+                            "{} {operation} race_lost response omitted claimed_by",
+                            self.backend.name
+                        )
+                    })?;
+                Ok(ClaimResult::RaceLost {
+                    claimed_by: claimed_by.to_string(),
+                })
+            }
+            Some("not_claimable") => Ok(ClaimResult::NotClaimable {
+                reason: json
+                    .get("reason")
+                    .and_then(serde_json::Value::as_str)
+                    .unwrap_or("no work is currently claimable")
+                    .to_string(),
+            }),
+            Some("error") => Ok(ClaimResult::ClaimError {
+                reason: json
+                    .get("reason")
+                    .and_then(serde_json::Value::as_str)
+                    .unwrap_or("backend rejected claim")
+                    .to_string(),
             }),
+            Some(other) => bail!(
+                "{} {operation} returned unknown claim outcome '{other}'",
+                self.backend.name
+            ),
+            None if bead_id.is_some() => Ok(ClaimResult::Claimed(
+                self.show(&BeadId::from(bead_id.unwrap())).await?,
+            )),
+            None if json.get("bead_id").is_some() || json.pointer("/data/bead_id").is_some() => {
+                Ok(ClaimResult::NotClaimable {
+                    reason: "no beads available".to_string(),
+                })
+            }
+            None => bail!(
+                "{} {operation} response omitted normalized outcome",
+                self.backend.name
+            ),
         }
     }
 
@@ -399,6 +457,22 @@ impl CliBeadStore {
 
 #[async_trait]
 impl BeadStore for CliBeadStore {
+    fn prompt_commands(&self) -> Option<BeadPromptCommands> {
+        let values = HashMap::from([
+            ("blocked", "<blocked-id>".to_string()),
+            ("blocker", "<blocker-id>".to_string()),
+        ]);
+        let dep_add = self.render_operation("dep_add", &values).ok()?;
+        let cli = shell_quote(&self.binary.display().to_string());
+        Some(BeadPromptCommands {
+            dep_add: std::iter::once(cli.clone())
+                .chain(dep_add)
+                .collect::<Vec<_>>()
+                .join(" "),
+            cli,
+        })
+    }
+
     fn is_corruption_error(&self, message: &str) -> bool {
         self.backend
             .error_contains_any(message, &self.backend.error_markers.corruption)
@@ -612,11 +686,16 @@ impl BeadStore for CliBeadStore {
     }
 
     async fn claim(&self, id: &BeadId, actor: &str) -> Result<ClaimResult> {
-        if matches!(
-            self.strategy("claim")?,
-            ParsedStrategy::Claim(ClaimStrategy::BatchOp)
-        ) {
-            return self.claim_via_batch(id, actor).await;
+        match self.strategy("claim")? {
+            ParsedStrategy::Claim(ClaimStrategy::AtomicCommand) => {
+                let values = HashMap::from([("id", id.to_string()), ("actor", actor.to_string())]);
+                return self.claim_from_operation("claim", &values).await;
+            }
+            ParsedStrategy::Claim(ClaimStrategy::BatchOp) => {
+                return self.claim_via_batch(id, actor).await;
+            }
+            ParsedStrategy::Claim(ClaimStrategy::CompareAndSet) => {}
+            _ => bail!("backend '{}' has invalid claim strategy", self.backend.name),
         }
         let shown = self.show(id).await?;
         if shown.status != BeadStatus::Open {
@@ -988,6 +1067,18 @@ impl BeadStore for CliBeadStore {
     }
 }
 
+fn shell_quote(value: &str) -> String {
+    if !value.is_empty()
+        && value
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || b"_@%+=:,./-".contains(&byte))
+    {
+        value.to_string()
+    } else {
+        format!("'{}'", value.replace('\'', "'\\''"))
+    }
+}
+
 fn bf_update_batch_args(
     id: &BeadId,
     status: Option<&str>,