fix(agents): fail clearly instead of silently dropping spawn_async_subagent results in thread-less contexts (B40) (#5123)

src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs

@@ -225,6 +225,40 @@ impl Tool for SpawnAsyncSubagentTool {
         let progress_sink = parent.on_progress.clone();
         let parent_thread_id =
             crate::openhuman::inference::provider::thread_context::current_thread_id();
+
+        // Async delivery is thread-addressed: the finished result is inserted
+        // back into the parent chat thread as a follow-up turn
+        // (`background_delivery`). Outside a chat turn (flow `agent` nodes,
+        // CLI, cron) there is no `current_thread_id()` to deliver into, so
+        // `background_delivery::deliver_batch` logs "dropping headless batch"
+        // and the (possibly real, completed) work is silently discarded — the
+        // caller sees "Accepted" and never learns the result never arrived.
+        // Fail loudly instead: the caller has a synchronous alternative
+        // (`spawn_subagent` with `blocking: true`, or a `delegate_*` tool).
+        // Both of those self-heal to blocking dispatch in this situation
+        // rather than reaching this guard — see the `has_delivery_thread`
+        // checks in `spawn_subagent.rs` and `dispatch.rs::dispatch_subagent`.
+        // Only a *direct* `spawn_async_subagent` call lands here.
+        if parent_thread_id.is_none() {
+            log::warn!(
+                "[spawn_async_subagent] refusing fire-and-forget spawn with no delivery thread \
+                 parent={} requested={} — directing caller to synchronous delegation (flow node / \
+                 CLI / cron context, background result would be discarded)",
+                parent.agent_definition_id,
+                definition.id
+            );
+            return Ok(ToolResult::error(
+                "spawn_async_subagent: no parent chat thread available to deliver the result \
+                 into (this looks like a flow node, CLI, or cron run rather than an interactive \
+                 chat turn). Fire-and-forget delegation has nowhere to land its result here and \
+                 the sub-agent's work would be silently discarded. Use synchronous delegation \
+                 instead: call `spawn_subagent` with `blocking: true`, or use a `delegate_*` \
+                 tool — both run the sub-agent inline and hand you its output in this turn. \
+                 For parallel work, model it as parallel flow nodes rather than background \
+                 sub-agents.",
+            ));
+        }
+
         let store = SubagentSessionStore::new(parent.workspace_dir.clone());
         let workspace_descriptor = tool_context.and_then(|ctx| ctx.workspace.clone());
         let effective_action_root = workspace_descriptor
@@ -1040,6 +1074,19 @@ fn reusable_follow_up_message(prompt: &str, context: Option<&str>) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry;
+    use crate::openhuman::agent::harness::fork_context::{
+        with_parent_context, ParentExecutionContext,
+    };
+    use crate::openhuman::config::AgentConfig;
+    use crate::openhuman::context::prompt::ToolCallFormat;
+    use crate::openhuman::inference::provider::Provider;
+    use crate::openhuman::memory::{
+        Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts,
+    };
+    use std::collections::HashSet;
+    use std::path::Path;
+    use std::sync::Arc;
 
     #[test]
     fn parameters_schema_advertises_fire_and_forget_fields() {
@@ -1280,4 +1327,178 @@ mod tests {
         assert!(result.is_error);
         assert!(result.output().contains("prompt"));
     }
+
+    /// B40 / Gap 4: a delegating agent (orchestrator/subconscious) calling
+    /// `spawn_async_subagent` directly from a thread-less context (flow
+    /// `agent` node, CLI, cron) must get a clear, actionable error instead of
+    /// silently accepting the spawn and later dropping its result in
+    /// `background_delivery`'s "headless batch" path. Sets up a real parent
+    /// turn context (so the call gets past the `current_parent()` /
+    /// allowlist / registry checks) but deliberately does NOT wrap the call
+    /// in `with_thread_id`, so `current_thread_id()` is None — the exact
+    /// condition that used to sail through to `tokio::spawn` and lose the
+    /// result.
+    #[tokio::test]
+    async fn errors_clearly_when_no_parent_thread_for_delivery() {
+        let _ = AgentDefinitionRegistry::init_global_builtins();
+        let workspace = tempfile::TempDir::new().expect("workspace");
+
+        let result = with_parent_context(parent_context(workspace.path()), async {
+            SpawnAsyncSubagentTool::new()
+                .execute(json!({
+                    "agent_id": "researcher",
+                    "prompt": "investigate x",
+                }))
+                .await
+        })
+        .await
+        .unwrap();
+
+        assert!(result.is_error);
+        let out = result.output();
+        assert!(out.contains("no parent chat thread"), "{out}");
+        // The recommended escape hatch must name `blocking: true` — plain
+        // `spawn_subagent` defaults to async and would otherwise be steered
+        // straight back into this same guard.
+        assert!(out.contains("spawn_subagent"), "{out}");
+        assert!(out.contains("blocking: true"), "{out}");
+        assert!(out.contains("delegate_"), "{out}");
+    }
+
+    /// The positive half of the branch above: with a chat thread bound, the
+    /// guard must NOT fire. This asserts only that the call gets *past* the
+    /// `parent_thread_id.is_none()` check — driving the full spawn/session
+    /// machinery to a successful "Accepted" is out of scope for a unit test,
+    /// so a later failure is acceptable; a "no parent chat thread" failure is
+    /// not. Pins that the guard keys on thread presence and nothing else.
+    #[tokio::test]
+    async fn guard_does_not_fire_when_parent_thread_is_bound() {
+        let _ = AgentDefinitionRegistry::init_global_builtins();
+        let workspace = tempfile::TempDir::new().expect("workspace");
+
+        let result = with_parent_context(parent_context(workspace.path()), async {
+            crate::openhuman::inference::provider::thread_context::with_thread_id(
+                "t-parent",
+                async {
+                    SpawnAsyncSubagentTool::new()
+                        .execute(json!({
+                            "agent_id": "researcher",
+                            "prompt": "investigate x",
+                        }))
+                        .await
+                },
+            )
+            .await
+        })
+        .await
+        .unwrap();
+
+        assert!(
+            !result.output().contains("no parent chat thread"),
+            "guard fired despite a bound parent thread: {}",
+            result.output()
+        );
+    }
+
+    fn parent_context(workspace_dir: &Path) -> ParentExecutionContext {
+        ParentExecutionContext {
+            workspace_descriptor: None,
+            agent_definition_id: "orchestrator".into(),
+            allowed_subagent_ids: HashSet::from(["researcher".to_string()]),
+            turn_model_source: crate::openhuman::tinyagents::TurnModelSource::new(Arc::new(
+                NoopProvider,
+            )),
+            all_tools: Arc::new(Vec::new()),
+            all_tool_specs: Arc::new(Vec::new()),
+            visible_tool_names: std::collections::HashSet::new(),
+            model_name: "test-model".into(),
+            temperature: 0.0,
+            workspace_dir: workspace_dir.to_path_buf(),
+            memory: Arc::new(NoopMemory),
+            agent_config: AgentConfig::default(),
+            workflows: Arc::new(Vec::new()),
+            memory_context: Arc::new(None),
+            session_id: "parent-session".into(),
+            channel: "test".into(),
+            connected_integrations: Vec::new(),
+            tool_call_format: ToolCallFormat::Native,
+            session_key: "parent-key".into(),
+            session_parent_prefix: None,
+            on_progress: None,
+            run_queue: None,
+        }
+    }
+
+    struct NoopProvider;
+
+    #[async_trait::async_trait]
+    impl Provider for NoopProvider {
+        async fn chat_with_system(
+            &self,
+            _system_prompt: Option<&str>,
+            _message: &str,
+            _model: &str,
+            _temperature: f64,
+        ) -> anyhow::Result<String> {
+            Ok(String::new())
+        }
+    }
+
+    struct NoopMemory;
+
+    #[async_trait::async_trait]
+    impl Memory for NoopMemory {
+        fn name(&self) -> &str {
+            "noop"
+        }
+
+        async fn store(
+            &self,
+            _namespace: &str,
+            _key: &str,
+            _content: &str,
+            _category: MemoryCategory,
+            _session_id: Option<&str>,
+        ) -> anyhow::Result<()> {
+            Ok(())
+        }
+
+        async fn recall(
+            &self,
+            _query: &str,
+            _limit: usize,
+            _opts: RecallOpts<'_>,
+        ) -> anyhow::Result<Vec<MemoryEntry>> {
+            Ok(Vec::new())
+        }
+
+        async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result<Option<MemoryEntry>> {
+            Ok(None)
+        }
+
+        async fn list(
+            &self,
+            _namespace: Option<&str>,
+            _category: Option<&MemoryCategory>,
+            _session_id: Option<&str>,
+        ) -> anyhow::Result<Vec<MemoryEntry>> {
+            Ok(Vec::new())
+        }
+
+        async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result<bool> {
+            Ok(false)
+        }
+
+        async fn namespace_summaries(&self) -> anyhow::Result<Vec<NamespaceSummary>> {
+            Ok(Vec::new())
+        }
+
+        async fn count(&self) -> anyhow::Result<usize> {
+            Ok(0)
+        }
+
+        async fn health_check(&self) -> bool {
+            true
+        }
+    }
 }

src/openhuman/agent_orchestration/tools/spawn_parallel_agents_tests.rs

@@ -367,6 +367,11 @@ fn definition_with_tool_scope(
 
 #[tokio::test]
 async fn rejects_more_tasks_than_parent_parallel_limit() {
+    // The parallel-limit check now runs inside the execution graph, which is
+    // reached only after the registry lookup — so this test needs the global
+    // builtins initialised (as its siblings already do) rather than relying on
+    // whichever test happened to initialise them first.
+    let _ = AgentDefinitionRegistry::init_global_builtins();
     let tool = SpawnParallelAgentsTool::new();
     let parent = parent_context(2);
     let result = with_parent_context(parent, async {
@@ -382,7 +387,11 @@ async fn rejects_more_tasks_than_parent_parallel_limit() {
     .await
     .expect("tool result");
     assert!(result.is_error);
-    assert!(result.output().contains("max_parallel_tools"));
+    assert!(
+        result.output().contains("max_parallel_tools"),
+        "{}",
+        result.output()
+    );
 }
 
 #[tokio::test]

src/openhuman/agent_orchestration/tools/spawn_subagent.rs

@@ -431,7 +431,25 @@ impl Tool for SpawnSubagentTool {
             }
         }
 
-        if !blocking {
+        // Async-by-default only holds where the finished result has somewhere
+        // to land. `spawn_async_subagent` delivers thread-addressed (see
+        // `background_delivery`), so outside a chat turn (flow `agent` node,
+        // CLI, cron) it now refuses outright (B40). Self-heal to blocking
+        // dispatch here rather than forwarding into that guard: the caller
+        // asked to delegate, and running the sub-agent inline is the one mode
+        // that both executes it and returns its output. Mirrors the
+        // `has_delivery_thread` fallback the `delegate_*` tools already do in
+        // `dispatch.rs::dispatch_subagent`.
+        let has_delivery_thread =
+            crate::openhuman::inference::provider::thread_context::current_thread_id().is_some();
+        if !blocking && !has_delivery_thread {
+            log::info!(
+                "[spawn_subagent] async delegation requested for '{}' but no delivery thread \
+                 (flow node / CLI / cron context) — falling back to blocking dispatch",
+                definition.id
+            );
+        }
+        if !blocking && has_delivery_thread {
             let mut async_args = args;
             if let Some(obj) = async_args.as_object_mut() {
                 obj.insert(
@@ -1147,7 +1165,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn legacy_archetype_alias_is_forwarded_to_async_default_path() {
+    async fn legacy_archetype_alias_is_normalized_to_agent_id() {
         let _ = AgentDefinitionRegistry::init_global_builtins();
         let tool = SpawnSubagentTool;
         let result = tool
@@ -1158,18 +1176,50 @@ mod tests {
             .await
             .unwrap();
         assert!(result.is_error);
-        assert!(
-            result
-                .output()
-                .contains("spawn_async_subagent called outside of an agent turn"),
-            "{}",
-            result.output()
-        );
+        // The alias resolved: the call got past argument validation and only
+        // failed later, on the missing parent turn.
         assert!(
             !result.output().contains("agent_id is required"),
             "{}",
             result.output()
         );
+        assert!(
+            result.output().contains("called outside of an agent turn"),
+            "{}",
+            result.output()
+        );
+    }
+
+    /// B40: with no chat thread bound, async-by-default delegation has nowhere
+    /// to deliver a result, so `spawn_subagent` must self-heal to blocking
+    /// dispatch rather than forwarding into `spawn_async_subagent`'s
+    /// thread-less guard — otherwise the guard's own advice ("use
+    /// `spawn_subagent`") would loop straight back into the guard. Asserted
+    /// via which tool owns the downstream error.
+    #[tokio::test]
+    async fn async_default_self_heals_to_blocking_without_delivery_thread() {
+        let _ = AgentDefinitionRegistry::init_global_builtins();
+        let result = SpawnSubagentTool
+            .execute(json!({
+                "agent_id": "researcher",
+                "prompt": "work with no delivery thread",
+            }))
+            .await
+            .unwrap();
+
+        let out = result.output();
+        assert!(
+            !out.contains("spawn_async_subagent"),
+            "thread-less spawn_subagent must not route into the async tool: {out}"
+        );
+        assert!(
+            !out.contains("no parent chat thread"),
+            "thread-less spawn_subagent must not hit the async delivery guard: {out}"
+        );
+        assert!(
+            out.contains("spawn_subagent called outside of an agent turn"),
+            "expected the blocking path's own error: {out}"
+        );
     }
 
     #[tokio::test]

src/openhuman/subconscious/agent/agent.toml

@@ -32,7 +32,16 @@ allowlist = [
 #                     replaced the old scratchpad.
 #   - goals_*       → evolve the user's long-term goals as their world shifts.
 #   - notify_user   → surface something time-sensitive to the user.
-#   - spawn_async_subagent → delegate deeper research / multi-step work.
+#   - spawn_subagent  → delegate deeper research / multi-step work.
+#
+# Delegation here is deliberately `spawn_subagent`, NOT `spawn_async_subagent`:
+# subconscious turns run headless (no chat thread is bound via
+# `thread_context::current_thread_id()` — see `session.rs::process`), and
+# thread-addressed async delivery has nowhere to land its result there. It used
+# to be silently discarded by `background_delivery` ("dropping headless batch");
+# since B40 `spawn_async_subagent` refuses outright in that condition.
+# `spawn_subagent` self-heals to blocking dispatch with no delivery thread, so
+# the sub-agent actually runs AND its output comes back inside this turn.
 named = [
     "memory_diff",
     "agent_prepare_context",
@@ -41,5 +50,5 @@ named = [
     "goals_list",
     "goals_add",
     "goals_edit",
-    "spawn_async_subagent",
+    "spawn_subagent",
 ]

src/openhuman/subconscious/agent/prompt.md

@@ -32,10 +32,12 @@ is to do nothing. Act only when the change genuinely matters to the user.
   must clear a high bar (a real deadline, a risk, something they'd want to
   know now).
 
-- **`spawn_async_subagent`** — Delegate deeper, multi-step work when you
+- **`spawn_subagent`** — Delegate deeper, multi-step work when you
   spot something genuinely actionable that needs research or execution
   (e.g. `agent_id: "researcher"` for web research, `agent_id:
-  "orchestrator"` for coordinated multi-tool work). Fire-and-forget.
+  "orchestrator"` for coordinated multi-tool work). This runs the
+  sub-agent inline and returns its result to you in this turn, so you can
+  act on what it found.
 
 - **`memory_diff` / `agent_prepare_context`** — Already run for you each
   tick. Only call them again if you need to re-check a narrower slice.

src/openhuman/subconscious/profiles/memory.rs

@@ -8,7 +8,7 @@
 //!      gather grounding context.
 //!   3. **reflect** — hand `diff + context` to the slim decision agent, which
 //!      records to-dos (`update_task`), evolves goals (`goals_*`), notifies the
-//!      user (`notify_user`), or delegates (`spawn_async_subagent`).
+//!      user (`notify_user`), or delegates (`spawn_subagent`).
 //!
 //! The generic [`SubconsciousInstance`] runner owns the scheduler + circuit
 //! breaker; this profile owns only what is memory-specific.
@@ -45,7 +45,7 @@ const SUBCONSCIOUS_TOOL_CATALOG: &str = "\
 - update_task: Add or update an actionable item on the user's global to-do board.
 - goals_add: Record a new long-term goal that the changed world makes relevant.
 - goals_edit: Revise an existing long-term goal.
-- spawn_async_subagent: Delegate deeper research or multi-step work.
+- spawn_subagent: Delegate deeper research or multi-step work (runs inline; its result comes back to you).
 ";
 
 /// Construct the live `memory` instance from config (used by the registry /
@@ -192,7 +192,7 @@ impl MemoryProfile {
 
         let mode_guidance = match self.mode {
             SubconsciousMode::Aggressive | SubconsciousMode::EventDriven => {
-                "\n\nYou may delegate deeper work with `spawn_async_subagent` (e.g. research \
+                "\n\nYou may delegate deeper work with `spawn_subagent` (e.g. research \
                  or multi-step execution) when you spot something genuinely actionable."
             }
             _ => "",

src/openhuman/tinyflows/caps.rs

@@ -989,6 +989,19 @@ impl OpenHumanAgentRunner {
     /// `run_via_registry_fallback` already gave `entry.model` — without this,
     /// a custom flow agent's model pin (e.g. `hint:reasoning`) silently
     /// dropped to the default chat model once routed through the harness.
+    ///
+    /// **Synchronous only (B40 / Gap 4).** A flow `agent` node runs here with
+    /// no chat thread bound via `thread_context::current_thread_id()`. If the
+    /// agent it runs is a delegating agent (orchestrator/subconscious) and
+    /// calls `spawn_async_subagent` directly, the tool now refuses (see the
+    /// `parent_thread_id.is_none()` guard in
+    /// `agent_orchestration::tools::spawn_async_subagent`) rather than
+    /// silently discarding the background result once it finishes —
+    /// `background_delivery` has nowhere to deliver it without a thread id.
+    /// This module does not bridge async subagent delivery into flow runs.
+    /// For work that needs to happen in parallel, model it as parallel flow
+    /// nodes (the tinyflows engine already fans those out) instead of
+    /// reaching for a background sub-agent from inside a single node.
     async fn run_via_harness(
         &self,
         agent_ref: &str,