diff --git a/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs b/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs index b4b5cf7d2..0ec3bb495 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs +++ b/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 { + 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> { + Ok(Vec::new()) + } + + async fn get(&self, _namespace: &str, _key: &str) -> anyhow::Result> { + Ok(None) + } + + async fn list( + &self, + _namespace: Option<&str>, + _category: Option<&MemoryCategory>, + _session_id: Option<&str>, + ) -> anyhow::Result> { + Ok(Vec::new()) + } + + async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result { + Ok(false) + } + + async fn namespace_summaries(&self) -> anyhow::Result> { + Ok(Vec::new()) + } + + async fn count(&self) -> anyhow::Result { + Ok(0) + } + + async fn health_check(&self) -> bool { + true + } + } } diff --git a/src/openhuman/agent_orchestration/tools/spawn_parallel_agents_tests.rs b/src/openhuman/agent_orchestration/tools/spawn_parallel_agents_tests.rs index 5a094213b..b3c11047e 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_parallel_agents_tests.rs +++ b/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] diff --git a/src/openhuman/agent_orchestration/tools/spawn_subagent.rs b/src/openhuman/agent_orchestration/tools/spawn_subagent.rs index 5b1dcbe5f..262bde83e 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_subagent.rs +++ b/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] diff --git a/src/openhuman/subconscious/agent/agent.toml b/src/openhuman/subconscious/agent/agent.toml index 19bbe91d0..8723c59e6 100644 --- a/src/openhuman/subconscious/agent/agent.toml +++ b/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", ] diff --git a/src/openhuman/subconscious/agent/prompt.md b/src/openhuman/subconscious/agent/prompt.md index 45e5bd124..b3907460a 100644 --- a/src/openhuman/subconscious/agent/prompt.md +++ b/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. diff --git a/src/openhuman/subconscious/profiles/memory.rs b/src/openhuman/subconscious/profiles/memory.rs index 49f818066..6de59593a 100644 --- a/src/openhuman/subconscious/profiles/memory.rs +++ b/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." } _ => "", diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs index 54c2e1ef6..cfb530ffe 100644 --- a/src/openhuman/tinyflows/caps.rs +++ b/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,