admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-30 23:14:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(agent): orchestrator worker thread depth=1 (#930) (#1221)

Browse Source 
Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
oxoxDev
2026-05-05 11:06:43 -07:00
committed by GitHub
co-authored by google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> Claude Opus 4.7
parent 08d5d1a88e
commit a274adc81e
18 changed files with 633 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/types/thread.ts
+1
View File
@@ -6,6 +6,7 @@ export interface Thread {
messageCount: number;
lastMessageAt: string;
createdAt: string;
parentThreadId?: string;
labels: string[];
}
src/openhuman/agent/agents/orchestrator/agent.toml
+1
View File
@@ -83,6 +83,7 @@ named = [
"read_workspace_state",
"ask_user_clarification",
"spawn_subagent",
"spawn_worker_thread",
"composio_list_connections",
# Time + scheduling — lets the orchestrator answer "what time is it",
# "remind me in 10 minutes", "every morning at 8" directly rather than
src/openhuman/agent/agents/orchestrator/prompt.md
+3 -2
View File
@@ -56,7 +56,8 @@ to a sub-agent wastes a turn. Use these directly:
| `read_workspace_state` | Get git status + file tree before planning a code task. |
| `composio_list_connections` | Check which external integrations (Gmail, Notion, GitHub, …) the user has authorised *right now*. Session-start list may be stale. |
| `ask_user_clarification` | Ask one focused question when the request is ambiguous — don't guess. |
| `spawn_subagent` | Escape hatch for agent ids not listed in the delegation table above; only use when direct handling is not sufficient. |
| `spawn_subagent` | Inline delegation: the sub-agent's work is collapsed into a single result in this thread. Use for quick tasks. |
| `spawn_worker_thread` | Dedicated delegation: creates a fresh 'worker' thread for the sub-agent. Use for long, complex, or multi-step tasks to avoid cluttering the parent thread. |
**Scheduling rule of thumb.** To "remind me in 10 minutes", call `current_time`
first. If `cron_add` is available and enabled for this runtime, then call
@@ -89,7 +90,7 @@ transcript. For everyday delegation keep `dedicated_thread` off (the default)
and surface the result inline.
Worker threads are one level deep by design: a sub-agent never sees
`spawn_subagent`, so a worker cannot itself spawn another worker.
`spawn_subagent` or `spawn_worker_thread`, so a worker cannot itself spawn another worker.
## Connecting external services
src/openhuman/agent/harness/subagent_runner/ops.rs
+84 -6
View File
@@ -34,6 +34,7 @@ use crate::openhuman::agent::progress::AgentProgress;
use crate::openhuman::context::prompt::{
render_subagent_system_prompt, PromptContext, PromptTool, SubagentRenderOptions,
};
use crate::openhuman::memory::conversations::ConversationMessage;
use crate::openhuman::providers::{ChatMessage, ChatRequest, Provider, ToolCall};
use crate::openhuman::tools::{Tool, ToolCategory, ToolSpec};
@@ -254,7 +255,10 @@ async fn run_typed_mode(
// should not list these tools either, but we enforce it here so a
// misconfigured TOML can't bypass the rule.
let before = allowed_indices.len();
allowed_indices.retain(|&i| !is_subagent_spawn_tool(parent.all_tools[i].name()));
allowed_indices.retain(|&i| {
let name = parent.all_tools[i].name();
!is_subagent_spawn_tool(name) && name != "spawn_worker_thread"
});
let stripped = before - allowed_indices.len();
if stripped > 0 {
tracing::debug!(
@@ -734,6 +738,7 @@ async fn run_typed_mode(
definition.max_iterations,
task_id,
&definition.id,
options.worker_thread_id.clone(),
handoff_cache.as_deref(),
parent,
)
@@ -813,7 +818,7 @@ async fn run_fork_mode(
.all_tools
.iter()
.map(|t| t.name().to_string())
.filter(|name| !is_subagent_spawn_tool(name))
.filter(|name| !is_subagent_spawn_tool(name) && name != "spawn_worker_thread")
.collect();
let model = parent.model_name.clone();
@@ -840,6 +845,7 @@ async fn run_fork_mode(
task_id,
&definition.id,
None,
None,
parent,
)
.await?;
@@ -891,6 +897,7 @@ async fn run_inner_loop(
max_iterations: usize,
task_id: &str,
agent_id: &str,
worker_thread_id: Option<String>,
handoff_cache: Option<&ResultHandoffCache>,
parent: &ParentExecutionContext,
) -> Result<(String, usize, AggregatedUsage), SubagentRunError> {
@@ -1040,6 +1047,32 @@ async fn run_inner_loop(
}
};
let append_worker_message =
|content: String, sender: String, extra_metadata: serde_json::Value| {
if let Some(ref thread_id) = worker_thread_id {
let message = ConversationMessage {
id: format!("{}:{}", sender, uuid::Uuid::new_v4()),
content,
message_type: "text".to_string(),
extra_metadata,
sender,
created_at: chrono::Utc::now().to_rfc3339(),
};
if let Err(err) = crate::openhuman::memory::conversations::append_message(
parent.workspace_dir.clone(),
thread_id,
message,
) {
tracing::debug!(
agent_id = %agent_id,
thread_id = %thread_id,
error = %err,
"[subagent_runner] failed to append message to worker thread"
);
}
}
};
// Per-turn progress sink shared with the parent — `None` for runs
// that don't have a subscriber (CLI / triage / tests). Cloned upfront
// so the inner loop body doesn't repeatedly re-resolve `parent.on_progress`.
@@ -1126,6 +1159,17 @@ async fn run_inner_loop(
"[subagent_runner] no tool calls — returning final response"
);
history.push(ChatMessage::assistant(response_text.clone()));
append_worker_message(
response_text.clone(),
"agent".to_string(),
serde_json::json!({
"scope": "worker_thread",
"agent_id": agent_id,
"task_id": task_id,
"iteration": iteration + 1,
"final": true,
}),
);
// Persist the final response before returning so the
// transcript always captures the last provider reply.
persist_transcript(history, &usage);
@@ -1146,6 +1190,18 @@ async fn run_inner_loop(
history.push(ChatMessage::assistant(assistant_history_content));
}
append_worker_message(
response_text.clone(),
"agent".to_string(),
serde_json::json!({
"scope": "worker_thread",
"agent_id": agent_id,
"task_id": task_id,
"iteration": iteration + 1,
"tool_calls": native_calls.len(),
}),
);
// Persist the assistant response + tool-call intents **before**
// executing tools. If the session crashes mid-tool-call we
// still have what the model emitted on disk.
@@ -1309,9 +1365,21 @@ async fn run_inner_loop(
} else {
let tool_msg = serde_json::json!({
"tool_call_id": call.id,
"content": result_text,
"content": result_text.clone(),
});
history.push(ChatMessage::tool(tool_msg.to_string()));
append_worker_message(
result_text.clone(),
"user".to_string(),
serde_json::json!({
"scope": "worker_thread",
"agent_id": agent_id,
"task_id": task_id,
"iteration": iteration + 1,
"tool_call_id": call.id,
"tool_name": call.name,
}),
);
}
if let Some(ref tx) = progress_sink {
@@ -1331,9 +1399,19 @@ async fn run_inner_loop(
}
if force_text_mode && !text_mode_result_block.is_empty() {
history.push(ChatMessage::user(format!(
"[Tool results]\n{text_mode_result_block}"
)));
let content = format!("[Tool results]\n{text_mode_result_block}");
history.push(ChatMessage::user(content.clone()));
append_worker_message(
content,
"user".to_string(),
serde_json::json!({
"scope": "worker_thread",
"agent_id": agent_id,
"task_id": task_id,
"iteration": iteration + 1,
"mode": "text",
}),
);
}
// Persist again after tool results have been appended so the
src/openhuman/agent/harness/subagent_runner/ops_tests.rs
+2
View File
@@ -334,6 +334,7 @@ async fn typed_mode_returns_text_through_runner() {
toolkit_override: None,
context: None,
task_id: Some("t1".into()),
worker_thread_id: None,
},
)
.await
@@ -439,6 +440,7 @@ async fn typed_mode_filters_tools_by_skill_filter() {
toolkit_override: None,
context: None,
task_id: None,
worker_thread_id: None,
},
)
.await
src/openhuman/agent/harness/subagent_runner/types.rs
+5
View File
@@ -31,6 +31,11 @@ pub struct SubagentRunOptions {
/// Stable id for tracing / DomainEvents (defaults to a UUID).
pub task_id: Option<String>,
/// Optional thread ID for persistent worker threads. When set,
/// every assistant message and tool result in the inner loop is
/// appended to this thread in the global ConversationStore.
pub worker_thread_id: Option<String>,
}
/// Outcome of a single sub-agent run, returned to the parent.
src/openhuman/memory/conversations/bus.rs
+1
View File
@@ -173,6 +173,7 @@ fn persist_channel_turn(
id: thread_id.clone(),
title,
created_at: created_at.clone(),
parent_thread_id: None,
labels: Some(vec!["work".to_string()]),
},
)?;
src/openhuman/memory/conversations/store.rs
+21 -10
View File
@@ -57,6 +57,8 @@ enum ThreadLogEntry {
created_at: String,
updated_at: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
parent_thread_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
labels: Option<Vec<String>>,
},
Delete {
@@ -87,6 +89,7 @@ impl ConversationStore {
title: request.title.clone(),
created_at: request.created_at.clone(),
updated_at: now,
parent_thread_id: request.parent_thread_id.clone(),
labels: request.labels.clone(),
},
)?;
@@ -159,6 +162,7 @@ impl ConversationStore {
title: title.to_string(),
created_at: entry.created_at.clone(),
updated_at: updated_at.to_string(),
parent_thread_id: entry.parent_thread_id.clone(),
labels: Some(entry.labels.clone()),
},
)?;
@@ -192,6 +196,7 @@ impl ConversationStore {
title: entry.title.clone(),
created_at: entry.created_at.clone(),
updated_at: updated_at.to_string(),
parent_thread_id: entry.parent_thread_id.clone(),
labels: Some(labels),
},
)?;
@@ -352,6 +357,7 @@ impl ConversationStore {
message_count,
last_message_at,
created_at: entry.created_at.clone(),
parent_thread_id: entry.parent_thread_id.clone(),
labels: entry.labels.clone(),
}))
}
@@ -370,24 +376,28 @@ impl ConversationStore {
thread_id,
title,
created_at,
parent_thread_id,
labels,
..
} => {
let (created_at_value, labels_value) = match index.get(&thread_id) {
Some(existing) => (
existing.created_at.clone(),
labels.unwrap_or_else(|| existing.labels.clone()),
),
None => {
let inferred = labels.unwrap_or_else(|| infer_labels(&thread_id));
(created_at, inferred)
}
};
let (created_at_value, parent_thread_id_value, labels_value) =
match index.get(&thread_id) {
Some(existing) => (
existing.created_at.clone(),
parent_thread_id.or_else(|| existing.parent_thread_id.clone()),
labels.unwrap_or_else(|| existing.labels.clone()),
),
None => {
let inferred = labels.unwrap_or_else(|| infer_labels(&thread_id));
(created_at, parent_thread_id, inferred)
}
};
index.insert(
thread_id,
ThreadIndexEntry {
title,
created_at: created_at_value,
parent_thread_id: parent_thread_id_value,
labels: labels_value,
},
);
@@ -414,6 +424,7 @@ impl ConversationStore {
struct ThreadIndexEntry {
title: String,
created_at: String,
parent_thread_id: Option<String>,
labels: Vec<String>,
}
src/openhuman/memory/conversations/store_tests.rs
+13
View File
@@ -18,6 +18,7 @@ fn store_roundtrips_threads_and_messages() {
let created_at = "2026-04-10T12:00:00Z".to_string();
let thread = store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "default-thread".to_string(),
title: "Conversation".to_string(),
created_at: created_at.clone(),
@@ -55,6 +56,7 @@ fn store_updates_message_metadata() {
let (_temp, store) = make_store();
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "default-thread".to_string(),
title: "Conversation".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -95,6 +97,7 @@ fn purge_removes_threads_and_messages() {
let (_temp, store) = make_store();
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "default-thread".to_string(),
title: "Conversation".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -125,6 +128,7 @@ fn purge_removes_threads_and_messages() {
fn ensure_thread_is_idempotent() {
let (_temp, store) = make_store();
let req = CreateConversationThread {
parent_thread_id: None,
id: "t1".to_string(),
title: "Thread".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -141,6 +145,7 @@ fn delete_thread_removes_thread_and_messages() {
let (_temp, store) = make_store();
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "t1".to_string(),
title: "Thread".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -179,6 +184,7 @@ fn get_messages_empty_thread() {
let (_temp, store) = make_store();
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "t1".to_string(),
title: "Empty".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -202,6 +208,7 @@ fn multiple_threads_and_messages() {
for i in 0..3 {
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: format!("t{i}"),
title: format!("Thread {i}"),
created_at: format!("2026-04-10T12:0{i}:00Z"),
@@ -239,6 +246,7 @@ fn update_message_nonexistent_returns_error() {
let (_temp, store) = make_store();
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "t1".to_string(),
title: "Thread".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -260,6 +268,7 @@ fn update_thread_title_persists_latest_title() {
let (_temp, store) = make_store();
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "t1".to_string(),
title: "Chat Apr 10 12:00 PM".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -284,6 +293,7 @@ fn store_handles_labels_and_inference() {
// 1. Explicit labels on ensure
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "t1".to_string(),
title: "Thread 1".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -294,6 +304,7 @@ fn store_handles_labels_and_inference() {
// 2. Inferred labels for morning briefing
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "proactive:morning_briefing".to_string(),
title: "Morning Briefing".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -304,6 +315,7 @@ fn store_handles_labels_and_inference() {
// 3. Inferred labels for other proactive
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "proactive:system".to_string(),
title: "System Notification".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
@@ -314,6 +326,7 @@ fn store_handles_labels_and_inference() {
// 4. Default inferred labels (work)
store
.ensure_thread(CreateConversationThread {
parent_thread_id: None,
id: "user-thread".to_string(),
title: "User Chat".to_string(),
created_at: "2026-04-10T12:00:00Z".to_string(),
src/openhuman/memory/conversations/types.rs
+4
View File
@@ -16,6 +16,8 @@ pub struct ConversationThread {
pub message_count: usize,
pub last_message_at: String,
pub created_at: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent_thread_id: Option<String>,
#[serde(default)]
pub labels: Vec<String>,
}
@@ -42,6 +44,8 @@ pub struct CreateConversationThread {
pub title: String,
pub created_at: String,
#[serde(default)]
pub parent_thread_id: Option<String>,
#[serde(default)]
pub labels: Option<Vec<String>>,
}
src/openhuman/memory/rpc_models.rs
+4
View File
@@ -110,6 +110,8 @@ pub struct ConversationThreadSummary {
pub message_count: usize,
pub last_message_at: String,
pub created_at: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent_thread_id: Option<String>,
#[serde(default)]
pub labels: Vec<String>,
}
@@ -136,6 +138,8 @@ pub struct UpsertConversationThreadRequest {
pub title: String,
pub created_at: String,
#[serde(default)]
pub parent_thread_id: Option<String>,
#[serde(default)]
pub labels: Option<Vec<String>>,
}
src/openhuman/threads/ops.rs
+3
View File
@@ -78,6 +78,7 @@ fn thread_to_summary(thread: ConversationThread) -> ConversationThreadSummary {
message_count: thread.message_count,
last_message_at: thread.last_message_at,
created_at: thread.created_at,
parent_thread_id: thread.parent_thread_id,
labels: thread.labels,
}
}
@@ -132,6 +133,7 @@ pub async fn thread_upsert(
id: request.id,
title: request.title,
created_at: request.created_at,
parent_thread_id: request.parent_thread_id,
labels: request.labels,
},
)?;
@@ -157,6 +159,7 @@ pub async fn thread_create_new(
id,
title,
created_at,
parent_thread_id: None,
// Pass labels through as-is; the store's infer_labels() applies
// the same default on index rebuild, so this is the single source
// of truth for default labels.
src/openhuman/threads/ops_tests.rs
+1
View File
@@ -288,6 +288,7 @@ fn sample_thread() -> ConversationThread {
message_count: 5,
last_message_at: "2026-01-01T00:00:00Z".into(),
created_at: "2026-01-01T00:00:00Z".into(),
parent_thread_id: None,
labels: vec!["work".to_string()],
}
}
src/openhuman/tools/impl/agent/dispatch.rs
+1
View File
@@ -67,6 +67,7 @@ pub(crate) async fn dispatch_subagent(
toolkit_override: skill_filter.map(str::to_string),
context: None,
task_id: Some(task_id.clone()),
worker_thread_id: None,
};
match run_subagent(definition, prompt, options).await {
src/openhuman/tools/impl/agent/mod.rs
+2
View File
@@ -8,6 +8,7 @@ pub(crate) mod onboarding_status;
mod plan_exit;
mod skill_delegation;
mod spawn_subagent;
pub mod spawn_worker_thread;
mod todo_write;
pub(crate) use dispatch::dispatch_subagent;
@@ -20,4 +21,5 @@ pub use delegate::DelegateTool;
pub use plan_exit::{PlanExitTool, PLAN_EXIT_MARKER};
pub use skill_delegation::SkillDelegationTool;
pub use spawn_subagent::SpawnSubagentTool;
pub use spawn_worker_thread::SpawnWorkerThreadTool;
pub use todo_write::{global_todo_store, TodoItem, TodoStatus, TodoStore, TodoWriteTool};
src/openhuman/tools/impl/agent/spawn_subagent.rs
+2
View File
@@ -404,6 +404,7 @@ impl Tool for SpawnSubagentTool {
toolkit_override,
context,
task_id: Some(task_id.clone()),
worker_thread_id: None,
};
let progress_sink = current_parent().and_then(|p| p.on_progress.clone());
@@ -547,6 +548,7 @@ fn persist_worker_thread(
id: thread_id.clone(),
title,
created_at: now.clone(),
parent_thread_id: None,
labels: Some(vec!["worker".to_string()]),
},
)
src/openhuman/tools/impl/agent/spawn_worker_thread.rs
+473
View File
@@ -0,0 +1,473 @@
//! Tool: `spawn_worker_thread` — spawn a dedicated worker thread for a complex delegated task.
//!
//! Unlike `spawn_subagent`, which collapses sub-agent work into a single
//! tool result in the current thread, `spawn_worker_thread` creates a new
//! persisted thread with label `worker`. The sub-agent's full transcript
//! is recorded into that thread, and the parent receives a compact
//! reference (worker thread id) instead of the full output.
//!
//! Worker threads carry a hard cap on depth: a worker thread cannot spawn
//! another worker thread.
use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry;
use crate::openhuman::agent::harness::fork_context::current_parent;
use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRunOptions};
use crate::openhuman::memory::conversations::{
self, ConversationMessage, CreateConversationThread,
};
use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult};
use async_trait::async_trait;
use serde_json::json;
/// Spawns a sub-agent in a dedicated worker thread.
pub struct SpawnWorkerThreadTool;
impl Default for SpawnWorkerThreadTool {
fn default() -> Self {
Self::new()
}
}
impl SpawnWorkerThreadTool {
pub fn new() -> Self {
Self
}
}
#[async_trait]
impl Tool for SpawnWorkerThreadTool {
fn name(&self) -> &str {
"spawn_worker_thread"
}
fn description(&self) -> &str {
"Spawn a dedicated worker thread for a complex delegated task. \
Use this when the task is long or involves many steps that would \
clutter the current conversation. The sub-agent runs in a fresh \
thread labeled 'worker', and you receive the thread ID and a \
summary. Worker threads cannot spawn other worker threads."
}
fn parameters_schema(&self) -> serde_json::Value {
let agent_ids: Vec<String> = AgentDefinitionRegistry::global()
.map(|reg| reg.list().iter().map(|d| d.id.clone()).collect())
.unwrap_or_default();
let agent_id_schema = if agent_ids.is_empty() {
json!({
"type": "string",
"description": "Sub-agent id (e.g. code_executor, researcher, planner)."
})
} else {
json!({
"type": "string",
"enum": agent_ids,
"description": "Sub-agent id from the registry."
})
};
json!({
"type": "object",
"required": ["agent_id", "prompt", "task_title"],
"properties": {
"agent_id": agent_id_schema,
"prompt": {
"type": "string",
"description": "Clear, specific instruction for the sub-agent. The sub-agent has no memory of the parent's conversation, so include all context the sub-agent needs to act."
},
"task_title": {
"type": "string",
"description": "A short, descriptive title for the worker thread (e.g. 'Researching Rust async patterns')."
},
"context": {
"type": "string",
"description": "Optional context blob from prior task results. Rendered as a `[Context]` block before the prompt."
},
"toolkit": {
"type": "string",
"description": "Composio toolkit slug to scope this spawn to (e.g. `gmail`, `notion`)."
}
}
})
}
fn permission_level(&self) -> PermissionLevel {
PermissionLevel::Execute
}
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
let started = std::time::Instant::now();
let agent_id = args
.get("agent_id")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let prompt = args
.get("prompt")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let task_title = args
.get("task_title")
.and_then(|v| v.as_str())
.unwrap_or("Worker Task")
.to_string();
let context = args
.get("context")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let toolkit_override = args
.get("toolkit")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
if agent_id.is_empty() || prompt.is_empty() {
tracing::warn!(
agent_id = %agent_id,
prompt_empty = prompt.is_empty(),
"[spawn_worker_thread] rejected: agent_id and prompt are required"
);
return Ok(ToolResult::error("agent_id and prompt are required"));
}
let parent = current_parent().ok_or_else(|| anyhow::anyhow!("no parent context"))?;
// ── Depth Guard ────────────────────────────────────────────────
// Check if the current thread is already a worker thread.
let current_thread_id = crate::openhuman::providers::thread_context::current_thread_id()
.unwrap_or_else(|| "unknown".to_string());
tracing::info!(
agent_id = %agent_id,
task_title = %task_title,
current_thread_id = %current_thread_id,
toolkit_override = ?toolkit_override,
has_context = context.is_some(),
"[spawn_worker_thread] invoked"
);
let threads = conversations::list_threads(parent.workspace_dir.clone())
.map_err(|e| anyhow::anyhow!(e))?;
if let Some(current_thread) = threads.iter().find(|t| t.id == current_thread_id) {
if current_thread.labels.contains(&"worker".to_string())
|| current_thread.parent_thread_id.is_some()
{
tracing::warn!(
agent_id = %agent_id,
current_thread_id = %current_thread_id,
is_worker_label = current_thread.labels.contains(&"worker".to_string()),
has_parent_thread_id = current_thread.parent_thread_id.is_some(),
elapsed_ms = started.elapsed().as_millis() as u64,
"[spawn_worker_thread] depth guard blocked spawn from worker thread"
);
return Ok(ToolResult::error("Worker threads cannot spawn other worker threads. Depth is capped at 1. Use spawn_subagent for inline delegation instead."));
}
}
let registry = AgentDefinitionRegistry::global()
.ok_or_else(|| anyhow::anyhow!("AgentDefinitionRegistry not initialised"))?;
let definition = registry
.get(&agent_id)
.ok_or_else(|| anyhow::anyhow!("agent_id '{}' not found", agent_id))?;
// ── Create Worker Thread ───────────────────────────────────────
let worker_thread_id = format!("worker-{}", uuid::Uuid::new_v4());
let now = chrono::Utc::now().to_rfc3339();
conversations::ensure_thread(
parent.workspace_dir.clone(),
CreateConversationThread {
id: worker_thread_id.clone(),
title: task_title.clone(),
created_at: now.clone(),
parent_thread_id: Some(current_thread_id.clone()),
labels: Some(vec!["worker".to_string()]),
},
)
.map_err(|e| anyhow::anyhow!(e))?;
tracing::info!(
agent_id = %agent_id,
worker_thread_id = %worker_thread_id,
parent_thread_id = %current_thread_id,
task_title = %task_title,
created_at = %now,
"[spawn_worker_thread] created worker thread"
);
// Append initial user message to the worker thread
conversations::append_message(
parent.workspace_dir.clone(),
&worker_thread_id,
ConversationMessage {
id: format!("user:{}", uuid::Uuid::new_v4()),
content: prompt.clone(),
message_type: "text".to_string(),
extra_metadata: json!({
"scope": "worker_thread",
"agent_id": agent_id,
}),
sender: "user".to_string(),
created_at: now,
},
)
.map_err(|e| anyhow::anyhow!(e))?;
// We don't have an easy way to append a system message to the parent
// thread here without triggering a re-render of the history the model
// sees. Instead, we return the info in the tool result.
// ── Run Subagent ──────────────────────────────────────────────
let options = SubagentRunOptions {
skill_filter_override: None,
toolkit_override,
context,
task_id: None,
worker_thread_id: Some(worker_thread_id.clone()),
};
tracing::debug!(
agent_id = %agent_id,
worker_thread_id = %worker_thread_id,
"[spawn_worker_thread] dispatching run_subagent"
);
match run_subagent(definition, &prompt, options).await {
Ok(outcome) => {
tracing::info!(
agent_id = %agent_id,
worker_thread_id = %worker_thread_id,
task_id = %outcome.task_id,
elapsed_ms = started.elapsed().as_millis() as u64,
"[spawn_worker_thread] completed successfully"
);
let parent_visible = format!(
"Spawned worker thread `{worker_thread_id}` for the task: {task_title}. \
The sub-agent has completed its work. You can find the full transcript \
in the worker thread.\n\n\
[worker_thread_ref]\n{}\n[/worker_thread_ref]",
json!({
"thread_id": worker_thread_id,
"label": "worker",
"agent_id": agent_id,
"task_id": outcome.task_id,
"status": "completed"
})
);
Ok(ToolResult::success(parent_visible))
}
Err(err) => {
tracing::error!(
agent_id = %agent_id,
worker_thread_id = %worker_thread_id,
error = %err,
elapsed_ms = started.elapsed().as_millis() as u64,
"[spawn_worker_thread] execution failed"
);
Ok(ToolResult::error(format!(
"Worker thread execution failed: {err}"
)))
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::openhuman::agent::harness::definition::{
AgentDefinition, DefinitionSource, ModelSpec, PromptSource, SandboxMode, ToolScope,
};
use crate::openhuman::agent::harness::fork_context::with_parent_context;
use crate::openhuman::agent::harness::ParentExecutionContext;
use std::path::PathBuf;
use std::sync::Arc;
use tempfile::TempDir;
struct MockProvider;
#[async_trait]
impl crate::openhuman::providers::Provider for MockProvider {
async fn chat_with_system(
&self,
_: Option<&str>,
_: &str,
_: &str,
_: f64,
) -> anyhow::Result<String> {
Ok("".into())
}
async fn chat(
&self,
_: crate::openhuman::providers::ChatRequest<'_>,
_: &str,
_: f64,
) -> anyhow::Result<crate::openhuman::providers::ChatResponse> {
Ok(crate::openhuman::providers::ChatResponse {
text: Some("done".into()),
tool_calls: vec![],
usage: None,
})
}
fn supports_native_tools(&self) -> bool {
true
}
}
struct MockMemory;
#[async_trait]
impl crate::openhuman::memory::Memory for MockMemory {
async fn store(
&self,
_: &str,
_: &str,
_: &str,
_: crate::openhuman::memory::MemoryCategory,
_: Option<&str>,
) -> anyhow::Result<()> {
Ok(())
}
async fn recall(
&self,
_: &str,
_: usize,
_: crate::openhuman::memory::RecallOpts<'_>,
) -> anyhow::Result<Vec<crate::openhuman::memory::MemoryEntry>> {
Ok(vec![])
}
async fn get(
&self,
_: &str,
_: &str,
) -> anyhow::Result<Option<crate::openhuman::memory::MemoryEntry>> {
Ok(None)
}
async fn list(
&self,
_: Option<&str>,
_: Option<&crate::openhuman::memory::MemoryCategory>,
_: Option<&str>,
) -> anyhow::Result<Vec<crate::openhuman::memory::MemoryEntry>> {
Ok(vec![])
}
async fn forget(&self, _: &str, _: &str) -> anyhow::Result<bool> {
Ok(true)
}
async fn namespace_summaries(
&self,
) -> anyhow::Result<Vec<crate::openhuman::memory::NamespaceSummary>> {
Ok(vec![])
}
async fn count(&self) -> anyhow::Result<usize> {
Ok(0)
}
async fn health_check(&self) -> bool {
true
}
fn name(&self) -> &str {
"mock"
}
}
fn test_parent_ctx(workspace_dir: PathBuf) -> ParentExecutionContext {
ParentExecutionContext {
session_id: "test".into(),
session_key: "test".into(),
session_parent_prefix: None,
model_name: "test".into(),
temperature: 0.4,
workspace_dir,
provider: Arc::new(MockProvider),
memory: Arc::new(MockMemory),
channel: "test".into(),
all_tools: Arc::new(vec![]),
all_tool_specs: Arc::new(vec![]),
skills: Arc::new(vec![]),
memory_context: None,
connected_integrations: vec![],
composio_client: None,
on_progress: None,
agent_config: crate::openhuman::config::AgentConfig::default(),
tool_call_format: crate::openhuman::context::prompt::ToolCallFormat::Native,
}
}
#[tokio::test]
async fn rejects_if_already_worker_thread() {
let temp = TempDir::new().unwrap();
let thread_id = "worker-123";
conversations::ensure_thread(
temp.path().to_path_buf(),
CreateConversationThread {
id: thread_id.to_string(),
title: "Worker".into(),
created_at: "now".into(),
parent_thread_id: None,
labels: Some(vec!["worker".to_string()]),
},
)
.unwrap();
crate::openhuman::providers::thread_context::with_thread_id(thread_id.to_string(), async {
let parent = test_parent_ctx(temp.path().to_path_buf());
with_parent_context(parent, async {
let tool = SpawnWorkerThreadTool::new();
let result = tool
.execute(json!({
"agent_id": "researcher",
"prompt": "do it",
"task_title": "Task"
}))
.await
.unwrap();
assert!(result.is_error);
assert!(result
.output()
.contains("cannot spawn other worker threads"));
})
.await;
})
.await;
}
#[tokio::test]
async fn rejects_if_has_parent_thread_id() {
let temp = TempDir::new().unwrap();
let thread_id = "sub-123";
conversations::ensure_thread(
temp.path().to_path_buf(),
CreateConversationThread {
id: thread_id.to_string(),
title: "Sub".into(),
created_at: "now".into(),
parent_thread_id: Some("parent".into()),
labels: None,
},
)
.unwrap();
crate::openhuman::providers::thread_context::with_thread_id(thread_id.to_string(), async {
let parent = test_parent_ctx(temp.path().to_path_buf());
with_parent_context(parent, async {
let tool = SpawnWorkerThreadTool::new();
let result = tool
.execute(json!({
"agent_id": "researcher",
"prompt": "do it",
"task_title": "Task"
}))
.await
.unwrap();
assert!(result.is_error);
assert!(result
.output()
.contains("cannot spawn other worker threads"));
})
.await;
})
.await;
}
}
src/openhuman/tools/orchestrator_tools.rs
+12 -3
View File
@@ -29,7 +29,7 @@ use crate::openhuman::agent::harness::definition::{
};
use crate::openhuman::context::prompt::ConnectedIntegration;
use super::{ArchetypeDelegationTool, SkillDelegationTool, Tool};
use super::{ArchetypeDelegationTool, SkillDelegationTool, SpawnWorkerThreadTool, Tool};
/// Synthesise the delegation tool list for an agent based on its
/// declarative `subagents` field.
@@ -65,6 +65,11 @@ pub fn collect_orchestrator_tools(
) -> Vec<Box<dyn Tool>> {
let mut tools: Vec<Box<dyn Tool>> = Vec::new();
// Orchestrator-only tool: spawn_worker_thread.
if definition.id == "orchestrator" {
tools.push(Box::new(SpawnWorkerThreadTool::new()));
}
for entry in &definition.subagents {
match entry {
SubagentEntry::AgentId(agent_id) => {
@@ -278,6 +283,7 @@ mod tests {
assert_eq!(
names,
vec![
"spawn_worker_thread", // orchestrator-only, prepended in collect_orchestrator_tools
"research", // researcher's delegate_name override
"delegate_archivist", // archivist has no delegate_name → default
"delegate_gmail",
@@ -308,7 +314,10 @@ mod tests {
let reg = registry_with_targets();
let tools = collect_orchestrator_tools(&orch, &reg, &[]);
let names: Vec<&str> = tools.iter().map(|t| t.name()).collect();
assert_eq!(names, vec!["research", "delegate_archivist"]);
assert_eq!(
names,
vec!["spawn_worker_thread", "research", "delegate_archivist"]
);
}
/// An AgentId entry that points at an id not present in the registry
@@ -324,7 +333,7 @@ mod tests {
let reg = registry_with_targets();
let tools = collect_orchestrator_tools(&orch, &reg, &[]);
let names: Vec<&str> = tools.iter().map(|t| t.name()).collect();
assert_eq!(names, vec!["research"]);
assert_eq!(names, vec!["spawn_worker_thread", "research"]);
}
/// An empty `subagents` list should produce zero tools — regular