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

feat(subagent): persist sub-agent runs and let orchestrator relay user messages (#3055)

Browse Source
This commit is contained in:
Steven Enamakel
2026-05-30 23:51:33 -07:00
committed by GitHub
parent 6574872c38
commit fbe89ba7b6
46 changed files with 1057 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/lib/i18n/ar.ts
+1
View File
@@ -2453,6 +2453,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'قيد التشغيل',
'conversations.subagent.statusCompleted': 'اكتمل',
'conversations.subagent.statusFailed': 'فشل',
'conversations.subagent.statusAwaitingUser': 'في انتظار المستخدم',
'daemon.serviceBlockingGate.body': 'المحتوى',
'daemon.serviceBlockingGate.downloadHint': 'تلميح التنزيل',
'daemon.serviceBlockingGate.downloadLatest': 'تنزيل أحدث إصدار',
app/src/lib/i18n/bn.ts
+1
View File
@@ -2501,6 +2501,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'চলছে',
'conversations.subagent.statusCompleted': 'সম্পন্ন',
'conversations.subagent.statusFailed': 'ব্যর্থ',
'conversations.subagent.statusAwaitingUser': 'ব্যবহারকারীর অপেক্ষায়',
'daemon.serviceBlockingGate.body': 'বডি',
'daemon.serviceBlockingGate.downloadHint': 'ডাউনলোড হিন্ট',
'daemon.serviceBlockingGate.downloadLatest': 'সর্বশেষ সংস্করণ ডাউনলোড করুন',
app/src/lib/i18n/de.ts
+1
View File
@@ -2565,6 +2565,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'läuft',
'conversations.subagent.statusCompleted': 'abgeschlossen',
'conversations.subagent.statusFailed': 'fehlgeschlagen',
'conversations.subagent.statusAwaitingUser': 'warte auf Benutzer',
'daemon.serviceBlockingGate.body': 'Körper',
'daemon.serviceBlockingGate.downloadHint': 'Hinweis herunterladen',
'daemon.serviceBlockingGate.downloadLatest': 'Lade die neueste Version herunter',
app/src/lib/i18n/en.ts
+1
View File
@@ -2759,6 +2759,7 @@ const en: TranslationMap = {
'conversations.subagent.statusRunning': 'running',
'conversations.subagent.statusCompleted': 'completed',
'conversations.subagent.statusFailed': 'failed',
'conversations.subagent.statusAwaitingUser': 'awaiting user',
'daemon.serviceBlockingGate.body':
'Retrying in the background. This usually resolves in a few seconds.',
'daemon.serviceBlockingGate.downloadHint':
app/src/lib/i18n/es.ts
+1
View File
@@ -2548,6 +2548,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'en ejecución',
'conversations.subagent.statusCompleted': 'completado',
'conversations.subagent.statusFailed': 'fallido',
'conversations.subagent.statusAwaitingUser': 'esperando al usuario',
'daemon.serviceBlockingGate.body': 'Cuerpo',
'daemon.serviceBlockingGate.downloadHint': 'Sugerencia de descarga',
'daemon.serviceBlockingGate.downloadLatest': 'Descargar la última versión',
app/src/lib/i18n/fr.ts
+1
View File
@@ -2557,6 +2557,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'en cours',
'conversations.subagent.statusCompleted': 'terminé',
'conversations.subagent.statusFailed': 'échoué',
'conversations.subagent.statusAwaitingUser': "en attente de l'utilisateur",
'daemon.serviceBlockingGate.body': 'Corps',
'daemon.serviceBlockingGate.downloadHint': 'Indice de téléchargement',
'daemon.serviceBlockingGate.downloadLatest': 'Télécharger la dernière version',
app/src/lib/i18n/hi.ts
+1
View File
@@ -2505,6 +2505,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'चल रहा है',
'conversations.subagent.statusCompleted': 'पूर्ण',
'conversations.subagent.statusFailed': 'विफल',
'conversations.subagent.statusAwaitingUser': 'उपयोगकर्ता की प्रतीक्षा',
'daemon.serviceBlockingGate.body': 'विवरण',
'daemon.serviceBlockingGate.downloadHint': 'डाउनलोड संकेत',
'daemon.serviceBlockingGate.downloadLatest': 'नवीनतम संस्करण डाउनलोड करें',
app/src/lib/i18n/id.ts
+1
View File
@@ -2507,6 +2507,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'berjalan',
'conversations.subagent.statusCompleted': 'selesai',
'conversations.subagent.statusFailed': 'gagal',
'conversations.subagent.statusAwaitingUser': 'menunggu pengguna',
'daemon.serviceBlockingGate.body': 'Isi',
'daemon.serviceBlockingGate.downloadHint': 'Petunjuk unduhan',
'daemon.serviceBlockingGate.downloadLatest': 'Unduh Versi Terbaru',
app/src/lib/i18n/it.ts
+1
View File
@@ -2539,6 +2539,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'in esecuzione',
'conversations.subagent.statusCompleted': 'completato',
'conversations.subagent.statusFailed': 'non riuscito',
'conversations.subagent.statusAwaitingUser': "in attesa dell'utente",
'daemon.serviceBlockingGate.body': 'Corpo',
'daemon.serviceBlockingGate.downloadHint': 'Suggerimento di download',
'daemon.serviceBlockingGate.downloadLatest': "Scarica l'ultima versione",
app/src/lib/i18n/ko.ts
+1
View File
@@ -2481,6 +2481,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': '실행 중',
'conversations.subagent.statusCompleted': '완료됨',
'conversations.subagent.statusFailed': '실패',
'conversations.subagent.statusAwaitingUser': '사용자 대기 중',
'daemon.serviceBlockingGate.body': '본문',
'daemon.serviceBlockingGate.downloadHint': '다운로드 안내',
'daemon.serviceBlockingGate.downloadLatest': '최신 버전 다운로드',
app/src/lib/i18n/pl.ts
+1
View File
@@ -2534,6 +2534,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'w toku',
'conversations.subagent.statusCompleted': 'zakończono',
'conversations.subagent.statusFailed': 'niepowodzenie',
'conversations.subagent.statusAwaitingUser': 'oczekiwanie na użytkownika',
'daemon.serviceBlockingGate.body':
'Rdzeń OpenHuman nie odpowiada. Spróbuj ponownie lub pobierz najnowszą wersję aplikacji.',
'daemon.serviceBlockingGate.downloadHint':
app/src/lib/i18n/pt.ts
+1
View File
@@ -2545,6 +2545,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'em execução',
'conversations.subagent.statusCompleted': 'concluído',
'conversations.subagent.statusFailed': 'falhou',
'conversations.subagent.statusAwaitingUser': 'aguardando usuário',
'daemon.serviceBlockingGate.body': 'Corpo',
'daemon.serviceBlockingGate.downloadHint': 'Dica de download',
'daemon.serviceBlockingGate.downloadLatest': 'Baixar Versão Mais Recente',
app/src/lib/i18n/ru.ts
+1
View File
@@ -2519,6 +2519,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': 'выполняется',
'conversations.subagent.statusCompleted': 'завершено',
'conversations.subagent.statusFailed': 'ошибка',
'conversations.subagent.statusAwaitingUser': 'ожидание пользователя',
'daemon.serviceBlockingGate.body': 'Текст',
'daemon.serviceBlockingGate.downloadHint': 'Подсказка по загрузке',
'daemon.serviceBlockingGate.downloadLatest': 'Скачать последнюю версию',
app/src/lib/i18n/zh-CN.ts
+1
View File
@@ -2380,6 +2380,7 @@ const messages: TranslationMap = {
'conversations.subagent.statusRunning': '运行中',
'conversations.subagent.statusCompleted': '已完成',
'conversations.subagent.statusFailed': '失败',
'conversations.subagent.statusAwaitingUser': '等待用户',
'daemon.serviceBlockingGate.body': '核心服务不可用，请等待或下载最新版本。',
'daemon.serviceBlockingGate.downloadHint': '下载最新版本',
'daemon.serviceBlockingGate.downloadLatest': '下载最新版本',
app/src/pages/conversations/components/SubagentDrawer.tsx
+8 -1
View File
@@ -50,7 +50,7 @@ function transcriptFromMessages(messages: ThreadMessage[]): {
function statusTone(status: ToolTimelineEntryStatus | undefined): {
dot: string;
pill: string;
label: 'statusRunning' | 'statusCompleted' | 'statusFailed';
label: 'statusRunning' | 'statusCompleted' | 'statusFailed' | 'statusAwaitingUser';
} {
if (status === 'success') {
return {
@@ -66,6 +66,13 @@ function statusTone(status: ToolTimelineEntryStatus | undefined): {
label: 'statusFailed',
};
}
if (status === 'awaiting_user') {
return {
dot: 'bg-amber-400 animate-pulse',
pill: 'bg-amber-100 dark:bg-amber-500/20 text-amber-700 dark:text-amber-300',
label: 'statusAwaitingUser',
};
}
return {
dot: 'bg-amber-500 animate-pulse',
pill: 'bg-amber-100 dark:bg-amber-500/20 text-amber-700 dark:text-amber-300',
app/src/providers/ChatRuntimeProvider.tsx
+17
View File
@@ -498,6 +498,23 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
})
);
},
onSubagentAwaitingUser: (event: ChatSubagentDoneEvent) => {
const subagentRowId = `${event.thread_id}:subagent:${event.skill_id}:${event.tool_name}`;
const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? [];
if (existing.length > 0) {
const entries = existing.map(entry => {
if (entry.id !== subagentRowId || entry.status !== 'running') return entry;
return decorateEntry({
...entry,
status: 'awaiting_user' as ToolTimelineEntryStatus,
subagent: entry.subagent
? { ...entry.subagent, status: 'awaiting_user' }
: entry.subagent,
});
});
dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries }));
}
},
onSubagentDone: (event: ChatSubagentDoneEvent) => {
const subagentRowId = `${event.thread_id}:subagent:${event.skill_id}:${event.tool_name}`;
const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? [];
app/src/services/chatService.ts
+19
View File
@@ -341,6 +341,7 @@ export interface ChatEventListeners {
onToolResult?: (event: ChatToolResultEvent) => void;
onSubagentSpawned?: (event: ChatSubagentSpawnedEventV2) => void;
onSubagentDone?: (event: ChatSubagentDoneEvent) => void;
onSubagentAwaitingUser?: (event: ChatSubagentDoneEvent) => void;
onSubagentIterationStart?: (event: ChatSubagentIterationStartEvent) => void;
onSubagentToolCall?: (event: ChatSubagentToolCallEvent) => void;
onSubagentToolResult?: (event: ChatSubagentToolResultEvent) => void;
@@ -373,6 +374,7 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
subagentSpawned: 'subagent_spawned',
subagentCompleted: 'subagent_completed',
subagentFailed: 'subagent_failed',
subagentAwaitingUser: 'subagent_awaiting_user',
subagentIterationStart: 'subagent_iteration_start',
subagentToolCall: 'subagent_tool_call',
subagentToolResult: 'subagent_tool_result',
@@ -501,6 +503,23 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
handlers.push([EVENTS.subagentFailed, onFailed]);
}
if (listeners.onSubagentAwaitingUser) {
const onAwaitingUser = (payload: unknown) => {
const e = payload as ChatSubagentDoneEvent;
chatLog(
'%s thread_id=%s request_id=%s round=%d agent=%s',
EVENTS.subagentAwaitingUser,
e.thread_id,
e.request_id,
e.round,
e.tool_name
);
listeners.onSubagentAwaitingUser?.(e);
};
socket.on(EVENTS.subagentAwaitingUser, onAwaitingUser);
handlers.push([EVENTS.subagentAwaitingUser, onAwaitingUser]);
}
if (listeners.onSubagentIterationStart) {
const cb = (payload: unknown) => {
const e = payload as ChatSubagentIterationStartEvent;
app/src/store/chatRuntimeSlice.ts
+3 -1
View File
@@ -13,7 +13,7 @@ import { resetUserScopedState } from './resetActions';
const turnStateLog = debug('chatRuntime.turnState');
export type ToolTimelineEntryStatus = 'running' | 'success' | 'error';
export type ToolTimelineEntryStatus = 'running' | 'success' | 'error' | 'awaiting_user';
export interface InferenceStatus {
phase: 'thinking' | 'tool_use' | 'subagent';
@@ -38,6 +38,8 @@ export interface SubagentActivity {
taskId: string;
/** Sub-agent definition id (e.g. `researcher`). */
agentId: string;
/** High-level status: `"running"`, `"awaiting_user"`, `"completed"`, `"failed"`. */
status?: string;
/** Human-readable display name from the agent registry (e.g. "Researcher"). */
displayName?: string;
/**
app/test/e2e/specs/chat-harness-subagent-continue.spec.ts
+254
View File
@@ -0,0 +1,254 @@
/**
* Chat harness — orchestrator → subagent continuation flow.
*
* Tests the full sub-agent persistence and continuation loop:
* 1. Orchestrator delegates to `researcher` sub-agent.
* 2. Researcher calls `ask_user_clarification` ("Which repo?").
* 3. The harness exits early, returns [SUBAGENT_AWAITING_USER] to the
* orchestrator.
* 4. Orchestrator surfaces the question to the user.
* 5. User replies in the composer ("the main repo").
* 6. Orchestrator calls `continue_subagent` with the user's answer.
* 7. Researcher resumes from checkpoint, produces final answer.
* 8. Orchestrator produces final synthesis with canary.
*
* Verifies:
* - The tool timeline shows a subagent entry.
* - The final canary text renders in the DOM.
* - The mock LLM received ≥4 POST requests (orchestrator initial +
* researcher initial + orchestrator relay + researcher resumed +
* orchestrator final).
* - Persisted thread JSONL contains the final canary text.
*/
import { waitForApp } from '../helpers/app-helpers';
import {
clickByTitle,
clickSend,
getSelectedThreadId,
hexEncodeThreadId,
typeIntoComposer,
waitForSocketConnected,
} from '../helpers/chat-harness';
import { callOpenhumanRpc } from '../helpers/core-rpc';
import { textExists } from '../helpers/element-helpers';
import { resetApp } from '../helpers/reset-app';
import { navigateViaHash } from '../helpers/shared-flows';
import { getRequestLog, setMockBehavior, startMockServer, stopMockServer } from '../mock-server';
const USER_ID = 'e2e-chat-harness-subagent-continue';
const PROMPT = 'Research the answer and tell me a marker phrase.';
const CANARY_FINAL = 'subagent-continue-canary-9bc3f';
const CLARIFICATION_QUESTION = 'Which repo should I search?';
const USER_ANSWER = 'the main repo';
const RESEARCHER_FINAL_REPLY = 'After searching the main repo, the answer is 42.';
// Five forced responses, popped in order by the mock LLM streamer:
// 1. Orchestrator: emits `research` tool call.
// 2. Researcher: calls `ask_user_clarification` (triggers early exit).
// 3. Orchestrator: sees [SUBAGENT_AWAITING_USER], asks user the question.
// 4. Orchestrator (after user reply): calls `continue_subagent` with user's answer.
// 5. Researcher (resumed): produces final text answer.
// 6. Orchestrator: final synthesis containing the canary.
const FORCED_RESPONSES = [
// 1. Orchestrator: delegate to researcher.
{
content: '',
toolCalls: [
{
id: 'call_research_1',
name: 'research',
arguments: JSON.stringify({ prompt: 'Tell me a marker phrase' }),
},
],
},
// 2. Researcher: ask clarification (tool call).
{
content: '',
toolCalls: [
{
id: 'call_clarify_1',
name: 'ask_user_clarification',
arguments: JSON.stringify({ question: CLARIFICATION_QUESTION }),
},
],
},
// 3. Orchestrator: relay the question to the user.
{ content: `The researcher needs to know: ${CLARIFICATION_QUESTION}` },
// 4. Orchestrator (after user replies): continue_subagent.
{
content: '',
toolCalls: [
{
id: 'call_continue_1',
name: 'continue_subagent',
arguments: JSON.stringify({
task_id: '{{DYNAMIC_TASK_ID}}',
agent_id: 'researcher',
message: USER_ANSWER,
}),
},
],
},
// 5. Researcher (resumed): final answer.
{ content: RESEARCHER_FINAL_REPLY },
// 6. Orchestrator: final synthesis.
{ content: `Done. The result is: ${CANARY_FINAL}` },
];
interface RuntimeSnapshot {
phase?: string;
activeSubagent?: string;
timelineIds: string[];
timelineNames: string[];
}
async function snapshotRuntime(threadId: string): Promise<RuntimeSnapshot> {
return (await browser.execute((tid: string) => {
const winAny = window as unknown as { __OPENHUMAN_STORE__?: { getState: () => unknown } };
const state = winAny.__OPENHUMAN_STORE__?.getState() as
| {
chatRuntime?: {
inferenceStatusByThread?: Record<string, { phase?: string; activeSubagent?: string }>;
toolTimelineByThread?: Record<string, Array<{ id?: string; name?: string }>>;
};
}
| undefined;
const status = state?.chatRuntime?.inferenceStatusByThread?.[tid];
const timeline = state?.chatRuntime?.toolTimelineByThread?.[tid] ?? [];
return {
phase: status?.phase,
activeSubagent: status?.activeSubagent,
timelineIds: timeline.map(e => e?.id ?? ''),
timelineNames: timeline.map(e => e?.name ?? ''),
};
}, threadId)) as RuntimeSnapshot;
}
describe('Chat harness — orchestrator → subagent continuation flow', () => {
before(async function beforeSuite() {
this.timeout(120_000);
await startMockServer();
await waitForApp();
await resetApp(USER_ID);
setMockBehavior('llmForcedResponses', JSON.stringify(FORCED_RESPONSES));
setMockBehavior('llmStreamChunkDelayMs', '10');
});
after(async () => {
setMockBehavior('llmForcedResponses', '');
setMockBehavior('llmStreamChunkDelayMs', '');
await stopMockServer();
});
it('orchestrator delegates, researcher asks clarification, user answers, researcher continues, canary lands', async function () {
this.timeout(120_000);
await navigateViaHash('/chat');
await browser.waitUntil(async () => await textExists('Threads'), {
timeout: 15_000,
timeoutMsg: 'Conversations did not mount',
});
expect(await clickByTitle('New thread', 8_000)).toBe(true);
const threadId = (await browser.waitUntil(async () => await getSelectedThreadId(), {
timeout: 8_000,
timeoutMsg: 'thread.selectedThreadId never populated',
})) as string;
expect(typeof threadId).toBe('string');
// Send the initial prompt.
await typeIntoComposer(PROMPT);
const socketReady = await waitForSocketConnected(30_000);
if (!socketReady) {
console.warn('[subagent-continue] socket did not connect within 30 s');
}
expect(
await browser.waitUntil(async () => await clickSend(), {
timeout: 5_000,
timeoutMsg: 'Send button never enabled',
})
).toBe(true);
// Wait for the orchestrator to relay the clarification question.
await browser.waitUntil(async () => await textExists(CLARIFICATION_QUESTION), {
timeout: 45_000,
timeoutMsg: 'orchestrator never relayed the clarification question',
});
// User answers the clarification.
await typeIntoComposer(USER_ANSWER);
expect(
await browser.waitUntil(async () => await clickSend(), {
timeout: 5_000,
timeoutMsg: 'Send button never enabled for user answer',
})
).toBe(true);
// Watch for subagent timeline entry.
let sawSubagentTimeline = false;
const deadline = Date.now() + 45_000;
while (Date.now() < deadline) {
const snap = await snapshotRuntime(threadId);
if (
snap.timelineIds.some(id => id.includes(':subagent:')) ||
snap.timelineNames.some(n => n.startsWith('subagent:'))
) {
sawSubagentTimeline = true;
}
if (sawSubagentTimeline) break;
if (await textExists(CANARY_FINAL)) break;
await browser.pause(200);
}
expect(sawSubagentTimeline).toBe(true);
// Final canary must land in the DOM.
await browser.waitUntil(async () => await textExists(CANARY_FINAL), {
timeout: 45_000,
timeoutMsg: 'orchestrator never produced the final canary text',
});
// IN_FLIGHT must drain after chat_done.
await browser.waitUntil(
async () => {
const snap = await callOpenhumanRpc<{ result: { entries: Array<unknown> } }>(
'openhuman.test_support_in_flight_chats',
{}
);
return snap.ok && (snap.result?.result?.entries?.length ?? 0) === 0;
},
{ timeout: 10_000, timeoutMsg: 'IN_FLIGHT never cleared after orchestrator finished' }
);
});
it('the mock LLM saw multiple chat-completions requests (parent + sub-agent + resumed sub-agent)', async () => {
const log = getRequestLog() as Array<{ method: string; url: string; body?: string }>;
const llmHits = log.filter(
r => r.method === 'POST' && r.url.includes('/openai/v1/chat/completions')
);
// Orchestrator turn 1 + researcher turn + orchestrator turn 2 (relay question)
// + orchestrator turn 3 (continue) + researcher turn 2 + orchestrator turn 4 (synthesis)
// = 6, but accept ≥4 for robustness.
expect(llmHits.length).toBeGreaterThanOrEqual(4);
});
it('persisted thread file records the final orchestrator text', async () => {
const threadId = await getSelectedThreadId();
expect(typeof threadId).toBe('string');
const relPath = `memory/conversations/threads/${hexEncodeThreadId(threadId as string)}.jsonl`;
let content = '';
const deadline = Date.now() + 30_000;
while (Date.now() < deadline) {
const read = await callOpenhumanRpc<{ result: { content_utf8: string } }>(
'openhuman.test_support_read_workspace_file',
{ rel_path: relPath, max_bytes: 131_072 }
);
if (read.ok && read.result?.result?.content_utf8) {
content = read.result.result.content_utf8;
if (content.includes(CANARY_FINAL)) break;
}
await browser.pause(500);
}
expect(content).toContain(CANARY_FINAL);
});
});
src/core/event_bus/events.rs
+12
View File
@@ -76,6 +76,15 @@ pub enum DomainEvent {
agent_id: String,
error: String,
},
/// A sub-agent called `ask_user_clarification` and paused, waiting
/// for the orchestrator to relay the user's answer via
/// `continue_subagent`.
SubagentAwaitingUser {
parent_session: String,
task_id: String,
agent_id: String,
question: String,
},
/// High-level orchestration accepted a child agent for execution.
AgentOrchestrationSpawned {
session_id: String,
@@ -723,6 +732,7 @@ impl DomainEvent {
| Self::SubagentSpawned { .. }
| Self::SubagentCompleted { .. }
| Self::SubagentFailed { .. }
| Self::SubagentAwaitingUser { .. }
| Self::AgentOrchestrationSpawned { .. }
| Self::AgentOrchestrationCompleted { .. }
| Self::AgentOrchestrationFailed { .. }
@@ -829,6 +839,7 @@ impl DomainEvent {
Self::SubagentSpawned { .. } => "SubagentSpawned",
Self::SubagentCompleted { .. } => "SubagentCompleted",
Self::SubagentFailed { .. } => "SubagentFailed",
Self::SubagentAwaitingUser { .. } => "SubagentAwaitingUser",
Self::AgentOrchestrationSpawned { .. } => "AgentOrchestrationSpawned",
Self::AgentOrchestrationCompleted { .. } => "AgentOrchestrationCompleted",
Self::AgentOrchestrationFailed { .. } => "AgentOrchestrationFailed",
@@ -919,6 +930,7 @@ impl DomainEvent {
Self::SubagentSpawned { agent_id, .. }
| Self::SubagentCompleted { agent_id, .. }
| Self::SubagentFailed { agent_id, .. }
| Self::SubagentAwaitingUser { agent_id, .. }
| Self::AgentOrchestrationSpawned { agent_id, .. }
| Self::AgentOrchestrationCompleted { agent_id, .. }
| Self::AgentOrchestrationFailed { agent_id, .. } => Some(agent_id.as_str()),
src/openhuman/agent/harness/engine/core.rs
+44
View File
@@ -50,6 +50,11 @@ pub(crate) struct TurnEngineOutcome {
/// or an early circuit-breaker halt. `Agent::turn` keys its checkpoint-only
/// history/transcript handling off this.
pub hit_cap: bool,
/// When set, the turn exited early because a specific tool requested
/// it (e.g. `ask_user_clarification` inside a sub-agent). The tool
/// result is in `text`. Callers use this to propagate pause semantics
/// without modifying the checkpoint strategy.
pub early_exit_tool: Option<String>,
}
/// Truncate a digest entry's body so a huge tool result can't blow up the
@@ -81,6 +86,7 @@ pub(crate) async fn run_turn_engine(
multimodal_config: &crate::openhuman::config::MultimodalConfig,
max_iterations: usize,
on_delta: Option<tokio::sync::mpsc::Sender<String>>,
early_exit_tool_names: &[&str],
) -> Result<TurnEngineOutcome> {
let mut context_guard = context_window_for_model(model)
.map(ContextGuard::with_context_window)
@@ -411,6 +417,7 @@ pub(crate) async fn run_turn_engine(
iterations: (iteration + 1) as u32,
cost: turn_cost,
hit_cap: false,
early_exit_tool: None,
});
}
@@ -425,6 +432,7 @@ pub(crate) async fn run_turn_engine(
// message per call with the correct id.
let mut tool_results = String::new();
let mut individual_results: Vec<String> = Vec::new();
let mut early_exit_tool: Option<String> = None;
for (call_idx, call) in tool_calls.iter().enumerate() {
// Stable id threaded through the start/complete pair. The fallback
// includes `call_idx` to stay unique when the same tool name
@@ -478,6 +486,19 @@ pub(crate) async fn run_turn_engine(
);
halt_reason = Some(reason);
}
// Early-exit when a sub-agent calls ask_user_clarification:
// the tool returned successfully with the question text — stop
// the loop so the runner can checkpoint and surface the pause.
if early_exit_tool_names.contains(&call.name.as_str()) && outcome.success {
tracing::info!(
iteration,
tool = call.name.as_str(),
"[agent_loop] early-exit tool detected — requesting early exit"
);
early_exit_tool = Some(call.name.clone());
break;
}
}
// Add assistant message with tool calls + tool results to history.
@@ -510,6 +531,27 @@ pub(crate) async fn run_turn_engine(
observer.after_iteration(history, iteration);
// Early-exit for ask_user_clarification: history already has the
// tool call + result appended, observer persisted the transcript.
// Return the clarification output so the sub-agent runner can
// checkpoint and propagate the pause to the orchestrator.
if let Some(ref exit_tool) = early_exit_tool {
tracing::info!(
iteration,
tool = exit_tool.as_str(),
"[agent_loop] early exit — returning with tool result as output"
);
let exit_text = individual_results.last().cloned().unwrap_or_default();
progress.turn_completed((iteration + 1) as u32).await;
return Ok(TurnEngineOutcome {
text: exit_text,
iterations: (iteration + 1) as u32,
cost: turn_cost,
hit_cap: false,
early_exit_tool,
});
}
// Circuit breaker tripped this iteration: return the root-cause summary
// instead of looping to `max_iterations`. Tool results are already in
// `history`, so the caller still has full context.
@@ -522,6 +564,7 @@ pub(crate) async fn run_turn_engine(
iterations: (iteration + 1) as u32,
cost: turn_cost,
hit_cap: false,
early_exit_tool: None,
});
}
}
@@ -550,5 +593,6 @@ pub(crate) async fn run_turn_engine(
iterations: max_iterations as u32,
cost: turn_cost,
hit_cap: true,
early_exit_tool: None,
})
}
src/openhuman/agent/harness/session/turn.rs
+1
View File
@@ -557,6 +557,7 @@ impl Agent {
&multimodal,
max_iterations,
None, // the web bridge streams via on_progress deltas, not on_delta
&[],
)
.await?;
src/openhuman/agent/harness/subagent_runner/mod.rs
+4 -1
View File
@@ -40,7 +40,10 @@ mod types;
// Public API — the entry point and the shapes it returns.
pub use autonomous::{autonomous_iter_cap, with_autonomous_iter_cap};
pub use ops::run_subagent;
pub use types::{SubagentMode, SubagentRunError, SubagentRunOptions, SubagentRunOutcome};
pub use types::{
SubagentCheckpointData, SubagentMode, SubagentRunError, SubagentRunOptions, SubagentRunOutcome,
SubagentRunStatus,
};
// Crate-internal re-exports: `agent::debug` calls the text-mode protocol
// renderer. The other `tool_prep` helpers are used only inside this module.
src/openhuman/agent/harness/subagent_runner/ops.rs
+95 -11
View File
@@ -1150,22 +1150,31 @@ async fn run_typed_mode(
if let Some(ref ctx) = options.context {
context_parts.push(ctx);
}
let user_message = if context_parts.is_empty() {
task_prompt.to_string()
let mut history: Vec<ChatMessage> = if let Some(ref initial) = options.initial_history {
tracing::info!(
agent_id = %definition.id,
task_id = %task_id,
history_len = initial.len(),
"[subagent_runner] resuming with initial_history (checkpoint replay)"
);
initial.clone()
} else {
format!("[Context]\n{}\n\n{task_prompt}", context_parts.join("\n\n"))
let user_message = if context_parts.is_empty() {
task_prompt.to_string()
} else {
format!("[Context]\n{}\n\n{task_prompt}", context_parts.join("\n\n"))
};
vec![
ChatMessage::system(system_prompt),
ChatMessage::user(user_message),
]
};
let mut history: Vec<ChatMessage> = vec![
ChatMessage::system(system_prompt),
ChatMessage::user(user_message),
];
// ── Run the inner tool-call loop ───────────────────────────────────
// Transcript persistence lives INSIDE the loop (one write per
// provider response), mirroring the main-agent turn loop in
// `session/turn.rs`. No post-loop write needed here.
let (output, iterations, _agg_usage) = run_inner_loop(
let (output, iterations, _agg_usage, early_exit_tool) = run_inner_loop(
subagent_provider.as_ref(),
&mut history,
&parent.all_tools,
@@ -1185,6 +1194,74 @@ async fn run_typed_mode(
)
.await?;
// Determine status: if the turn engine exited early because of
// ask_user_clarification, checkpoint the history and return
// AwaitingUser so the orchestrator can relay the user's answer.
let status = if early_exit_tool.as_deref() == Some("ask_user_clarification") {
let question = output.clone();
let options_vec: Option<Vec<String>> = None;
// Persist checkpoint so `continue_subagent` can resume later.
let checkpoint_dir = options
.checkpoint_dir
.clone()
.unwrap_or_else(|| parent.workspace_dir.join(".openhuman/subagent_checkpoints"));
if let Err(e) = std::fs::create_dir_all(&checkpoint_dir) {
tracing::warn!(
task_id = %task_id,
error = %e,
"[subagent_runner] failed to create checkpoint directory"
);
} else {
let checkpoint_data = super::types::SubagentCheckpointData {
task_id: task_id.to_string(),
agent_id: definition.id.clone(),
worker_thread_id: options.worker_thread_id.clone(),
history: history.clone(),
question: question.clone(),
options: options_vec.clone(),
toolkit_override: options.toolkit_override.clone(),
skill_filter_override: options.skill_filter_override.clone(),
model_override: options.model_override.clone(),
created_at: chrono::Utc::now().to_rfc3339(),
};
let checkpoint_path = checkpoint_dir.join(format!("{task_id}.json"));
match serde_json::to_string_pretty(&checkpoint_data) {
Ok(json) => {
if let Err(e) = std::fs::write(&checkpoint_path, json) {
tracing::warn!(
task_id = %task_id,
path = %checkpoint_path.display(),
error = %e,
"[subagent_runner] failed to write checkpoint"
);
} else {
tracing::info!(
task_id = %task_id,
path = %checkpoint_path.display(),
history_len = history.len(),
"[subagent_runner] checkpoint written for awaiting_user"
);
}
}
Err(e) => {
tracing::warn!(
task_id = %task_id,
error = %e,
"[subagent_runner] failed to serialize checkpoint"
);
}
}
}
super::types::SubagentRunStatus::AwaitingUser {
question,
options: options_vec,
}
} else {
super::types::SubagentRunStatus::Completed
};
Ok(SubagentRunOutcome {
task_id: task_id.to_string(),
agent_id: definition.id.clone(),
@@ -1192,6 +1269,7 @@ async fn run_typed_mode(
iterations,
elapsed: started.elapsed(),
mode: SubagentMode::Typed,
status,
})
}
@@ -1236,7 +1314,7 @@ async fn run_inner_loop(
handoff_cache: Option<&ResultHandoffCache>,
parent: &ParentExecutionContext,
extended_policy: bool,
) -> Result<(String, usize, AggregatedUsage), SubagentRunError> {
) -> Result<(String, usize, AggregatedUsage, Option<String>), SubagentRunError> {
// An autonomous skill run (set via `with_autonomous_iter_cap`) lifts the
// per-agent cap so sub-agents run until done / the circuit breaker trips.
let max_iterations = super::autonomous::autonomous_iter_cap()
@@ -1357,10 +1435,16 @@ async fn run_inner_loop(
&crate::openhuman::config::MultimodalConfig::default(),
max_iterations,
None, // sub-agents don't stream a draft
&["ask_user_clarification"],
)
.await?;
Ok((outcome.text, outcome.iterations as usize, observer.usage))
Ok((
outcome.text,
outcome.iterations as usize,
observer.usage,
outcome.early_exit_tool,
))
}
/// Apply the progressive-disclosure handoff to a tool result. If a cache is
src/openhuman/agent/harness/subagent_runner/ops_tests.rs
+4
View File
@@ -466,6 +466,8 @@ async fn typed_mode_returns_text_through_runner() {
model_override: None,
task_id: Some("t1".into()),
worker_thread_id: None,
initial_history: None,
checkpoint_dir: None,
},
)
.await
@@ -575,6 +577,8 @@ async fn typed_mode_filters_tools_by_skill_filter() {
model_override: None,
task_id: None,
worker_thread_id: None,
initial_history: None,
checkpoint_dir: None,
},
)
.await
src/openhuman/agent/harness/subagent_runner/types.rs
+51
View File
@@ -3,9 +3,12 @@
//! external callers importing these shapes don't drag in the full
//! orchestration machinery.
use std::path::PathBuf;
use std::time::Duration;
use thiserror::Error;
use crate::openhuman::inference::provider::ChatMessage;
/// Per-spawn options that override or augment what the
/// [`AgentDefinition`] specifies. Built by `SpawnSubagentTool::execute`
/// from the parent model's call arguments.
@@ -41,6 +44,33 @@ pub struct SubagentRunOptions {
/// 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>,
/// Pre-populated conversation history for resuming a paused
/// sub-agent (checkpoint + replay). When `Some`, the runner skips
/// system-prompt + user-message construction and uses this history
/// directly — it already contains the system prompt and all prior
/// turns including the clarification tool call/result.
pub initial_history: Option<Vec<ChatMessage>>,
/// Directory for writing/reading checkpoint files when the
/// sub-agent pauses for user input. Defaults to
/// `{workspace_dir}/.openhuman/subagent_checkpoints/`.
pub checkpoint_dir: Option<PathBuf>,
}
/// Terminal status of a sub-agent run.
#[derive(Debug, Clone)]
pub enum SubagentRunStatus {
/// The sub-agent completed normally with a final response.
Completed,
/// The sub-agent called `ask_user_clarification` and is waiting
/// for the orchestrator to relay the user's answer via
/// `continue_subagent`. The checkpoint file contains the full
/// conversation history for resumption.
AwaitingUser {
question: String,
options: Option<Vec<String>>,
},
}
/// Outcome of a single sub-agent run, returned to the parent.
@@ -58,6 +88,8 @@ pub struct SubagentRunOutcome {
pub elapsed: Duration,
/// Which execution mode was used (Typed vs. Fork).
pub mode: SubagentMode,
/// Whether the run completed or paused for user input.
pub status: SubagentRunStatus,
}
/// Which prompt-construction path the runner took for a sub-agent.
@@ -80,6 +112,25 @@ impl SubagentMode {
}
}
/// Serialisable checkpoint written when a sub-agent pauses for user input.
/// Contains everything needed to resume the run from where it left off.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SubagentCheckpointData {
pub task_id: String,
pub agent_id: String,
pub worker_thread_id: Option<String>,
pub history: Vec<ChatMessage>,
pub question: String,
pub options: Option<Vec<String>>,
/// Composio toolkit override, if the paused run was scoped to one.
pub toolkit_override: Option<String>,
/// Skill filter override, if the paused run was scoped to one.
pub skill_filter_override: Option<String>,
/// Model override, if one was set for this run.
pub model_override: Option<String>,
pub created_at: String,
}
/// Errors the runner can surface to the parent. The parent receives a
/// stringified version inside a tool result block.
#[derive(Debug, Error)]
src/openhuman/agent/harness/tool_loop.rs
+1
View File
@@ -305,6 +305,7 @@ pub(crate) async fn run_tool_call_loop(
multimodal_config,
max_iterations,
on_delta,
&[],
)
.await
.map(|outcome| outcome.text)
src/openhuman/agent/progress.rs
+10
View File
@@ -100,6 +100,16 @@ pub enum AgentProgress {
error: String,
},
/// A sub-agent paused and is waiting for user input relayed via
/// `continue_subagent`. The orchestrator surfaces the question to
/// the user and calls `continue_subagent` with the answer.
SubagentAwaitingUser {
agent_id: String,
task_id: String,
question: String,
worker_thread_id: Option<String>,
},
/// A sub-agent's inner LLM iteration is starting. Emitted **only
/// from inside [`crate::openhuman::agent::harness::subagent_runner`]**
/// when the parent context carries an `on_progress` sink — the
src/openhuman/agent/tools/delegate_to_personality.rs
+2
View File
@@ -215,6 +215,8 @@ impl Tool for DelegateToPersonalityTool {
skill_filter_override: None,
task_id: None,
worker_thread_id: None,
initial_history: None,
checkpoint_dir: None,
};
match run_subagent(&definition, &prompt, options).await {
src/openhuman/agent_orchestration/ops.rs
+2
View File
@@ -437,6 +437,8 @@ impl AgentOrchestrationSession {
model_override: request.model,
task_id: Some(orchestration_id.clone()),
worker_thread_id: None,
initial_history: None,
checkpoint_dir: None,
};
let task_session = self.clone();
src/openhuman/agent_orchestration/tools.rs
+3
View File
@@ -1,5 +1,7 @@
#[path = "tools/archetype_delegation.rs"]
mod archetype_delegation;
#[path = "tools/continue_subagent.rs"]
mod continue_subagent;
#[path = "tools/dispatch.rs"]
mod dispatch;
#[path = "tools/skill_delegation.rs"]
@@ -19,6 +21,7 @@ mod worker_thread;
pub(crate) use dispatch::dispatch_subagent;
pub use archetype_delegation::ArchetypeDelegationTool;
pub use continue_subagent::ContinueSubagentTool;
pub use skill_delegation::{SkillDelegationTool, INTEGRATIONS_DELEGATE_TOOL_NAME};
pub use spawn_parallel_agents::SpawnParallelAgentsTool;
pub use spawn_subagent::SpawnSubagentTool;
src/openhuman/agent_orchestration/tools/continue_subagent.rs
+340
View File
@@ -0,0 +1,340 @@
//! Tool: `continue_subagent` — resume a paused sub-agent with a follow-up
//! message (typically the user's answer to a clarification question).
//!
//! When a sub-agent calls `ask_user_clarification`, its harness loop exits
//! early, checkpoints the full conversation history, and returns an
//! `AwaitingUser` envelope to the parent. The orchestrator surfaces the
//! question to the user, and when the user answers, calls this tool to
//! resume the same sub-agent from its checkpoint with the user's response
//! appended to the conversation history.
use crate::core::event_bus::{publish_global, DomainEvent};
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, SubagentCheckpointData, SubagentRunOptions, SubagentRunStatus,
};
use crate::openhuman::agent::progress::AgentProgress;
use crate::openhuman::inference::provider::ChatMessage;
use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult};
use async_trait::async_trait;
use serde_json::json;
pub struct ContinueSubagentTool;
impl Default for ContinueSubagentTool {
fn default() -> Self {
Self::new()
}
}
impl ContinueSubagentTool {
pub fn new() -> Self {
Self
}
}
#[async_trait]
impl Tool for ContinueSubagentTool {
fn name(&self) -> &str {
"continue_subagent"
}
fn description(&self) -> &str {
"Resume a paused sub-agent that requested user input via \
ask_user_clarification. Pass the task_id from the \
[SUBAGENT_AWAITING_USER] envelope and the user's answer \
as the message. The sub-agent continues from its checkpoint \
with full prior context."
}
fn parameters_schema(&self) -> serde_json::Value {
json!({
"type": "object",
"required": ["task_id", "agent_id", "message"],
"properties": {
"task_id": {
"type": "string",
"description": "The task_id from the [SUBAGENT_AWAITING_USER] envelope."
},
"agent_id": {
"type": "string",
"description": "The agent_id from the [SUBAGENT_AWAITING_USER] envelope."
},
"message": {
"type": "string",
"description": "The user's answer or follow-up message to send to the paused sub-agent."
}
}
})
}
fn permission_level(&self) -> PermissionLevel {
PermissionLevel::Execute
}
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
let task_id = args
.get("task_id")
.and_then(|v| v.as_str())
.unwrap_or("")
.trim()
.to_string();
let agent_id = args
.get("agent_id")
.and_then(|v| v.as_str())
.unwrap_or("")
.trim()
.to_string();
let message = args
.get("message")
.and_then(|v| v.as_str())
.unwrap_or("")
.trim()
.to_string();
if task_id.is_empty() {
return Ok(ToolResult::error(
"continue_subagent: `task_id` is required",
));
}
if agent_id.is_empty() {
return Ok(ToolResult::error(
"continue_subagent: `agent_id` is required",
));
}
if message.is_empty() {
return Ok(ToolResult::error(
"continue_subagent: `message` is required (the user's answer)",
));
}
let parent = match current_parent() {
Some(p) => p,
None => {
return Ok(ToolResult::error(
"continue_subagent: no parent execution context available",
));
}
};
// Load checkpoint
let checkpoint_dir = parent.workspace_dir.join(".openhuman/subagent_checkpoints");
let checkpoint_path = checkpoint_dir.join(format!("{task_id}.json"));
let checkpoint_json = match std::fs::read_to_string(&checkpoint_path) {
Ok(json) => json,
Err(e) => {
tracing::warn!(
task_id = %task_id,
path = %checkpoint_path.display(),
error = %e,
"[continue_subagent] checkpoint not found"
);
return Ok(ToolResult::error(format!(
"continue_subagent: no checkpoint found for task_id '{task_id}'. \
The sub-agent may not have paused, or the checkpoint expired."
)));
}
};
let checkpoint: SubagentCheckpointData = match serde_json::from_str(&checkpoint_json) {
Ok(cp) => cp,
Err(e) => {
tracing::error!(
task_id = %task_id,
error = %e,
"[continue_subagent] failed to deserialize checkpoint"
);
return Ok(ToolResult::error(format!(
"continue_subagent: corrupted checkpoint for task_id '{task_id}': {e}"
)));
}
};
if checkpoint.agent_id != agent_id {
return Ok(ToolResult::error(format!(
"continue_subagent: agent_id mismatch — checkpoint has '{}', \
caller passed '{agent_id}'",
checkpoint.agent_id
)));
}
// Look up the agent definition
let registry = match AgentDefinitionRegistry::global() {
Some(reg) => reg,
None => {
return Ok(ToolResult::error(
"continue_subagent: AgentDefinitionRegistry not initialised",
));
}
};
let definition = match registry.get(&agent_id) {
Some(def) => def,
None => {
return Ok(ToolResult::error(format!(
"continue_subagent: unknown agent_id '{agent_id}'"
)));
}
};
// Reconstruct history and append the user's answer
let mut history = checkpoint.history;
history.push(ChatMessage::user(format!(
"[User's answer to your clarification question]\n{message}"
)));
tracing::info!(
task_id = %task_id,
agent_id = %agent_id,
history_len = history.len(),
message_chars = message.chars().count(),
"[continue_subagent] resuming sub-agent with user's answer"
);
let parent_session = parent.session_id.clone();
let progress_sink = parent.on_progress.clone();
// Publish resumed event (reuse SubagentSpawned with a note)
publish_global(DomainEvent::SubagentSpawned {
parent_session: parent_session.clone(),
agent_id: agent_id.clone(),
mode: "typed".to_string(),
task_id: task_id.clone(),
prompt_chars: message.chars().count(),
});
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentSpawned {
agent_id: agent_id.clone(),
task_id: task_id.clone(),
mode: "typed".to_string(),
dedicated_thread: false,
prompt_chars: message.chars().count(),
worker_thread_id: checkpoint.worker_thread_id.clone(),
display_name: definition.display_name.clone(),
})
.await;
}
// Build options with initial_history for replay
let options = SubagentRunOptions {
skill_filter_override: checkpoint.skill_filter_override,
toolkit_override: checkpoint.toolkit_override,
context: None,
model_override: checkpoint.model_override,
task_id: Some(task_id.clone()),
worker_thread_id: checkpoint.worker_thread_id.clone(),
initial_history: Some(history),
checkpoint_dir: Some(checkpoint_dir.clone()),
};
// Run the sub-agent from its checkpoint
match run_subagent(definition, "", options).await {
Ok(outcome) => {
match &outcome.status {
SubagentRunStatus::AwaitingUser {
question,
options: _,
} => {
// Another round of clarification
publish_global(DomainEvent::SubagentAwaitingUser {
parent_session,
task_id: outcome.task_id.clone(),
agent_id: outcome.agent_id.clone(),
question: question.clone(),
});
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentAwaitingUser {
agent_id: outcome.agent_id.clone(),
task_id: outcome.task_id.clone(),
question: question.clone(),
worker_thread_id: checkpoint.worker_thread_id.clone(),
})
.await;
}
let wt_display = checkpoint.worker_thread_id.as_deref().unwrap_or("(none)");
let envelope = format!(
"[SUBAGENT_AWAITING_USER]\n\
task_id: {}\n\
agent_id: {}\n\
worker_thread_id: {}\n\
question: {}\n\
[/SUBAGENT_AWAITING_USER]\n\n\
The sub-agent needs further clarification. \
Surface the above question to the user. When the user responds, \
call continue_subagent again with the same task_id, agent_id, \
and the user's new answer.",
outcome.task_id, outcome.agent_id, wt_display, question,
);
Ok(ToolResult::success(envelope))
}
SubagentRunStatus::Completed => {
// Clean up checkpoint file on successful completion
if let Err(e) = std::fs::remove_file(&checkpoint_path) {
tracing::debug!(
task_id = %task_id,
error = %e,
"[continue_subagent] failed to remove checkpoint (best-effort)"
);
} else {
tracing::info!(
task_id = %task_id,
"[continue_subagent] checkpoint cleaned up after completion"
);
}
publish_global(DomainEvent::SubagentCompleted {
parent_session,
task_id: outcome.task_id.clone(),
agent_id: outcome.agent_id.clone(),
elapsed_ms: outcome.elapsed.as_millis() as u64,
output_chars: outcome.output.chars().count(),
iterations: outcome.iterations,
});
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentCompleted {
agent_id: outcome.agent_id.clone(),
task_id: outcome.task_id.clone(),
elapsed_ms: outcome.elapsed.as_millis() as u64,
iterations: outcome.iterations as u32,
output_chars: outcome.output.chars().count(),
})
.await;
}
Ok(ToolResult::success(outcome.output))
}
}
}
Err(err) => {
let message = err.to_string();
tracing::error!(
task_id = %task_id,
agent_id = %agent_id,
"[continue_subagent] sub-agent execution failed"
);
publish_global(DomainEvent::SubagentFailed {
parent_session,
task_id: task_id.clone(),
agent_id: agent_id.clone(),
error: message.clone(),
});
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentFailed {
agent_id: agent_id.clone(),
task_id: task_id.clone(),
error: message.clone(),
})
.await;
}
Ok(ToolResult::error(format!(
"continue_subagent failed: {message}"
)))
}
}
}
}
src/openhuman/agent_orchestration/tools/dispatch.rs
+2
View File
@@ -88,6 +88,8 @@ pub(crate) async fn dispatch_subagent(
model_override: model_override.map(str::to_string),
task_id: Some(task_id.clone()),
worker_thread_id: None,
initial_history: None,
checkpoint_dir: None,
};
match run_subagent(definition, prompt, options).await {
src/openhuman/agent_orchestration/tools/spawn_parallel_agents.rs
+2
View File
@@ -435,6 +435,8 @@ async fn run_one_parallel_task(
model_override: None,
task_id: Some(task_id.clone()),
worker_thread_id: None,
initial_history: None,
checkpoint_dir: None,
};
match run_subagent(&definition, &prompt, options).await {
Ok(outcome) => {
src/openhuman/agent_orchestration/tools/spawn_subagent.rs
+106 -73
View File
@@ -16,7 +16,7 @@ use crate::core::event_bus::{publish_global, DomainEvent};
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, SubagentRunOutcome,
run_subagent, SubagentRunOptions, SubagentRunOutcome, SubagentRunStatus,
};
use crate::openhuman::agent::progress::AgentProgress;
use crate::openhuman::memory_conversations::{
@@ -132,7 +132,7 @@ impl Tool for SpawnSubagentTool {
},
"dedicated_thread": {
"type": "boolean",
"description": "Temporarily disabled (see tinyhumansai/openhuman#1624). Passing `true` causes this tool to return an explicit error. Omit the field or pass `false` until the worker-thread UI surface lands."
"description": "Legacy compatibility flag. Delegations now always create a persistent worker thread when parent context is available, so this flag no longer gates thread creation."
}
}
})
@@ -176,25 +176,15 @@ impl Tool for SpawnSubagentTool {
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty());
// Worker-thread spawning is temporarily disabled until a proper UI
// showcase lands (see tinyhumansai/openhuman#1624). Return an
// explicit error when callers request a dedicated thread so the
// behaviour is observable rather than silently downgraded.
let dedicated_thread_requested = args
// Worker threads are now always created for delegations that may
// need follow-up (checkpoint + replay for ask_user_clarification).
// The `dedicated_thread` parameter is accepted but no longer
// gates thread creation — every delegation gets a persistent
// worker thread. (#3049 supersedes the #1624 disable.)
let dedicated_thread = args
.get("dedicated_thread")
.and_then(|v| v.as_bool())
.unwrap_or(false);
if dedicated_thread_requested {
log::debug!(
"[spawn_subagent] dedicated_thread requested but temporarily \
disabled (see tinyhumansai/openhuman#1624); rejecting call"
);
return Ok(ToolResult::error(
"spawn_subagent: `dedicated_thread` is temporarily disabled \
(see tinyhumansai/openhuman#1624); retry without it.",
));
}
let dedicated_thread = false;
// ── Validation ─────────────────────────────────────────────────
if agent_id.is_empty() {
@@ -446,71 +436,112 @@ impl Tool for SpawnSubagentTool {
model_override,
task_id: Some(task_id.clone()),
worker_thread_id: worker_thread_id.clone(),
initial_history: None,
checkpoint_dir: None,
};
let progress_sink = current_parent().and_then(|p| p.on_progress.clone());
match run_subagent(definition, &prompt, options).await {
Ok(outcome) => {
publish_global(DomainEvent::SubagentCompleted {
parent_session,
task_id: outcome.task_id.clone(),
agent_id: outcome.agent_id.clone(),
elapsed_ms: outcome.elapsed.as_millis() as u64,
output_chars: outcome.output.chars().count(),
iterations: outcome.iterations,
});
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentCompleted {
agent_id: outcome.agent_id.clone(),
match &outcome.status {
SubagentRunStatus::AwaitingUser {
question,
options: _,
} => {
// Sub-agent paused for user input — publish
// awaiting event and return structured envelope so
// the orchestrator can relay the question and later
// call continue_subagent.
publish_global(DomainEvent::SubagentAwaitingUser {
parent_session,
task_id: outcome.task_id.clone(),
agent_id: outcome.agent_id.clone(),
question: question.clone(),
});
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentAwaitingUser {
agent_id: outcome.agent_id.clone(),
task_id: outcome.task_id.clone(),
question: question.clone(),
worker_thread_id: worker_thread_id.clone(),
})
.await;
}
let wt_display = worker_thread_id.as_deref().unwrap_or("(none)");
let envelope = format!(
"[SUBAGENT_AWAITING_USER]\n\
task_id: {}\n\
agent_id: {}\n\
worker_thread_id: {}\n\
question: {}\n\
[/SUBAGENT_AWAITING_USER]\n\n\
The sub-agent needs clarification before it can continue. \
Surface the above question to the user. When the user responds, \
call continue_subagent with the task_id, agent_id, and the \
user's answer as the message parameter.",
outcome.task_id, outcome.agent_id, wt_display, question,
);
Ok(ToolResult::success(envelope))
}
SubagentRunStatus::Completed => {
publish_global(DomainEvent::SubagentCompleted {
parent_session,
task_id: outcome.task_id.clone(),
agent_id: outcome.agent_id.clone(),
elapsed_ms: outcome.elapsed.as_millis() as u64,
iterations: outcome.iterations as u32,
output_chars: outcome.output.chars().count(),
})
.await;
}
iterations: outcome.iterations,
});
if dedicated_thread {
let workspace_dir = current_parent()
.map(|p| p.workspace_dir.clone())
.unwrap_or_else(|| PathBuf::from("."));
let parent_visible = match persist_worker_thread(
&workspace_dir,
&definition.id,
&prompt,
&outcome,
) {
Ok(thread_id) => {
render_worker_thread_result(&thread_id, &definition.id, &outcome)
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentCompleted {
agent_id: outcome.agent_id.clone(),
task_id: outcome.task_id.clone(),
elapsed_ms: outcome.elapsed.as_millis() as u64,
iterations: outcome.iterations as u32,
output_chars: outcome.output.chars().count(),
})
.await;
}
Err(error) => {
// Persistence failure must not silently swallow the
// sub-agent's work — return the full output and
// surface the worker-thread error so the parent
// model can mention it. We deliberately fall
// through to a `success` ToolResult so the agent
// loop doesn't prepend "Error:" to text the
// sub-agent produced legitimately.
tracing::error!(
target: "spawn_subagent",
agent_id = %definition.id,
error = %error,
"[spawn_subagent] dedicated_thread persistence failed; \
returning full sub-agent output inline"
);
format!(
"{}\n\n[worker_thread_error] failed to persist worker thread: {}",
outcome.output, error
)
}
};
return Ok(ToolResult::success(parent_visible));
}
Ok(ToolResult::success(outcome.output))
if dedicated_thread {
let workspace_dir = current_parent()
.map(|p| p.workspace_dir.clone())
.unwrap_or_else(|| PathBuf::from("."));
let parent_visible = match persist_worker_thread(
&workspace_dir,
&definition.id,
&prompt,
&outcome,
) {
Ok(thread_id) => render_worker_thread_result(
&thread_id,
&definition.id,
&outcome,
),
Err(error) => {
tracing::error!(
target: "spawn_subagent",
agent_id = %definition.id,
error = %error,
"[spawn_subagent] dedicated_thread persistence failed; \
returning full sub-agent output inline"
);
format!(
"{}\n\n[worker_thread_error] failed to persist worker thread: {}",
outcome.output, error
)
}
};
return Ok(ToolResult::success(parent_visible));
}
Ok(ToolResult::success(outcome.output))
}
}
}
Err(err) => {
let message = err.to_string();
@@ -760,6 +791,7 @@ mod tests {
elapsed: Duration::from_millis(120),
iterations: 3,
mode: SubagentMode::Typed,
status: SubagentRunStatus::Completed,
}
}
@@ -933,7 +965,8 @@ mod tests {
}
#[tokio::test]
async fn dedicated_thread_flag_is_rejected_explicitly() {
async fn dedicated_thread_flag_no_longer_returns_disabled_error() {
let _ = AgentDefinitionRegistry::init_global_builtins();
let tool = SpawnSubagentTool;
let result = tool
.execute(json!({
@@ -944,7 +977,7 @@ mod tests {
.await
.unwrap();
assert!(result.is_error);
assert!(result.output().contains("temporarily disabled"));
assert!(!result.output().contains("temporarily disabled"));
}
#[tokio::test]
src/openhuman/agent_orchestration/tools/spawn_worker_thread.rs
+2
View File
@@ -204,6 +204,8 @@ impl Tool for SpawnWorkerThreadTool {
model_override,
task_id: None,
worker_thread_id: Some(worker_thread_id.clone()),
initial_history: None,
checkpoint_dir: None,
};
tracing::debug!(
src/openhuman/channels/providers/web.rs
+31
View File
@@ -1223,6 +1223,37 @@ fn spawn_progress_bridge(
..Default::default()
});
}
AgentProgress::SubagentAwaitingUser {
agent_id,
task_id,
question,
worker_thread_id,
} => {
log::debug!(
"[web_channel][bridge] subagent_awaiting_user agent_id={} task_id={} client_id={} thread_id={} request_id={}",
agent_id,
task_id,
client_id,
thread_id,
request_id,
);
publish_web_channel_event(WebChannelEvent {
event: "subagent_awaiting_user".to_string(),
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: request_id.clone(),
message: Some(question),
tool_name: Some(agent_id),
skill_id: Some(task_id),
success: Some(true),
round: Some(round),
subagent: Some(SubagentProgressDetail {
worker_thread_id,
..Default::default()
}),
..Default::default()
});
}
AgentProgress::SubagentIterationStarted {
agent_id,
task_id,
src/openhuman/skills/run_log.rs
+6
View File
@@ -159,6 +159,12 @@ pub fn format_event(ev: &AgentProgress) -> Option<String> {
AgentProgress::SubagentFailed {
agent_id, error, ..
} => format!(" ⮑ subagent {agent_id} FAILED: {}", truncate(error, 200)),
AgentProgress::SubagentAwaitingUser {
agent_id, question, ..
} => format!(
" ⮑ subagent {agent_id} awaiting user: {}",
truncate(question, 200)
),
AgentProgress::TurnCompleted { iterations } => {
format!("turn completed ({iterations} iterations)")
}
src/openhuman/threads/turn_state/mirror.rs
+10
View File
@@ -162,6 +162,7 @@ impl TurnStateMirror {
subagent: Some(SubagentActivity {
task_id: task_id.clone(),
agent_id: agent_id.clone(),
status: None,
mode: Some(mode.clone()),
dedicated_thread: Some(*dedicated_thread),
child_iteration: None,
@@ -205,6 +206,15 @@ impl TurnStateMirror {
self.flush();
true
}
AgentProgress::SubagentAwaitingUser { task_id, .. } => {
if let Some(entry) = self.find_subagent_entry_mut(task_id) {
if let Some(activity) = entry.subagent.as_mut() {
activity.status = Some("awaiting_user".to_string());
}
}
self.flush();
true
}
AgentProgress::SubagentIterationStarted {
task_id,
iteration,
src/openhuman/threads/turn_state/mirror_tests.rs
+1 -1
View File
@@ -225,7 +225,7 @@ fn subagent_lifecycle_records_and_clears_active() {
dedicated_thread: false,
prompt_chars: 42,
worker_thread_id: None,
display_name: None,
display_name: Some("Researcher".into()),
});
let s = m.snapshot();
assert_eq!(s.active_subagent.as_deref(), Some("researcher"));
src/openhuman/threads/turn_state/types.rs
+4
View File
@@ -73,6 +73,10 @@ pub struct ToolTimelineEntry {
pub struct SubagentActivity {
pub task_id: String,
pub agent_id: String,
/// High-level status: `"running"`, `"awaiting_user"`, `"completed"`,
/// `"failed"`. `None` for legacy snapshots written before this field.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub status: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub mode: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
src/openhuman/tools/ops.rs
+1
View File
@@ -149,6 +149,7 @@ pub fn all_tools_with_runtime(
// returns a single text result. See
// `agent::harness::subagent_runner` for the dispatch path.
Box::new(SpawnSubagentTool::new()),
Box::new(ContinueSubagentTool::new()),
Box::new(SpawnParallelAgentsTool::new()),
Box::new(DelegateToPersonalityTool::new()),
// Coding-harness control flow (issue #1205): a process-global
tests/inference_agent_raw_coverage_e2e.rs
+4 -1
View File
@@ -39,7 +39,7 @@ use openhuman_core::openhuman::agent::harness::definition::{
};
use openhuman_core::openhuman::agent::harness::subagent_runner::{
autonomous_iter_cap, with_autonomous_iter_cap, SubagentMode, SubagentRunError,
SubagentRunOptions, SubagentRunOutcome,
SubagentRunOptions, SubagentRunOutcome, SubagentRunStatus,
};
use openhuman_core::openhuman::agent::harness::{
check_interrupt, current_sandbox_mode, with_current_sandbox_mode, InterruptFence,
@@ -4612,6 +4612,8 @@ async fn agent_subagent_public_types_cover_task_local_and_error_display_paths()
model_override: Some("specialist-model".to_string()),
task_id: Some("task-1".to_string()),
worker_thread_id: Some("thread-1".to_string()),
initial_history: None,
checkpoint_dir: None,
};
assert_eq!(options.skill_filter_override.as_deref(), Some("docs"));
assert_eq!(options.toolkit_override.as_deref(), Some("github"));
@@ -4624,6 +4626,7 @@ async fn agent_subagent_public_types_cover_task_local_and_error_display_paths()
iterations: 3,
elapsed: Duration::from_millis(12),
mode: SubagentMode::Typed,
status: SubagentRunStatus::Completed,
};
assert_eq!(outcome.mode.as_str(), "typed");
assert_eq!(outcome.elapsed.as_millis(), 12);
tests/memory_core_threads_raw_coverage_e2e.rs
+1 -1
View File
@@ -554,7 +554,7 @@ async fn thread_ops_welcome_migration_and_turn_state_cover_error_and_cleanup_pat
dedicated_thread: true,
prompt_chars: 42,
worker_thread_id: None,
display_name: None,
display_name: Some("Researcher".into()),
}));
assert!(mirror.observe(&AgentProgress::SubagentCompleted {
agent_id: "researcher".into(),
tests/memory_raw_coverage_e2e.rs
+1
View File
@@ -590,6 +590,7 @@ fn threads_turn_state_store_skips_corrupt_entries_and_marks_interrupted() {
subagent: Some(SubagentActivity {
task_id: "task-1".into(),
agent_id: "researcher".into(),
status: Some("running".into()),
mode: Some("read".into()),
dedicated_thread: Some(false),
child_iteration: Some(1),
tests/memory_threads_raw_coverage_e2e.rs
+2 -1
View File
@@ -3140,7 +3140,7 @@ fn turn_state_mirror_persists_progress_edges_from_public_events() {
dedicated_thread: true,
prompt_chars: 99,
worker_thread_id: None,
display_name: None,
display_name: Some("Researcher".into()),
}));
assert!(!mirror.observe(&AgentProgress::SubagentIterationStarted {
agent_id: "researcher".into(),
@@ -3407,6 +3407,7 @@ fn turn_state_store_persists_lists_marks_and_clears_snapshots() {
subagent: Some(SubagentActivity {
task_id: "task-1".into(),
agent_id: "agent-1".into(),
status: Some("running".into()),
mode: Some("focused".into()),
dedicated_thread: Some(true),
child_iteration: Some(1),