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(threads): surface live subagent work in parent thread (#1122) (#1159)

Browse Source
This commit is contained in:
Steven Enamakel
2026-05-03 19:51:46 -07:00
committed by GitHub
parent 2484ea40f2
commit e6b491bcf1
20 changed files with 1170 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/pages/conversations/components/ToolTimelineBlock.tsx
+88 -4
View File
@@ -1,8 +1,85 @@
import type { ToolTimelineEntry } from '../../../store/chatRuntimeSlice';
import type { SubagentActivity, ToolTimelineEntry } from '../../../store/chatRuntimeSlice';
import { formatTimelineEntry } from '../../../utils/toolTimelineFormatting';
import { parseWorkerThreadRef } from '../utils/workerThreadRef';
import { WorkerThreadRefCard } from './WorkerThreadRefCard';
/**
* Render the live activity of one running (or completed) sub-agent
* inside its parent timeline row — the mode/dedicated-thread badge,
* the child iteration counter, the final-run statistics, and the
* flat list of child tool calls the sub-agent has executed.
*
* Kept as a sibling of the existing worker-thread / detail block so
* the surrounding `<details>` chevron + status pill behaviour is
* unaffected — this component only renders when `subagent` is
* present on the entry, which is true for any row produced by the
* `subagent_*` socket events from a current core.
*/
export function SubagentActivityBlock({ subagent }: { subagent: SubagentActivity }) {
const headerBits: string[] = [];
if (subagent.mode) headerBits.push(subagent.mode);
if (subagent.dedicatedThread) headerBits.push('worker thread');
if (subagent.childIteration != null && subagent.childMaxIterations != null) {
headerBits.push(`turn ${subagent.childIteration}/${subagent.childMaxIterations}`);
} else if (subagent.iterations != null) {
headerBits.push(`${subagent.iterations} turn${subagent.iterations === 1 ? '' : 's'}`);
}
if (subagent.elapsedMs != null) {
headerBits.push(
subagent.elapsedMs >= 1000
? `${(subagent.elapsedMs / 1000).toFixed(1)}s`
: `${subagent.elapsedMs}ms`
);
}
return (
<div className="mt-1 space-y-0.5 text-[10px] text-stone-500" data-testid="subagent-activity">
{headerBits.length > 0 ? (
<div className="flex flex-wrap items-center gap-1.5">
{headerBits.map(bit => (
<span
key={bit}
className="rounded-full bg-stone-100 px-1.5 py-0.5 font-medium text-stone-600">
{bit}
</span>
))}
</div>
) : null}
{subagent.toolCalls.length > 0 ? (
<ul className="ml-1 space-y-0.5">
{subagent.toolCalls.map(call => {
const tone =
call.status === 'running'
? 'text-amber-700'
: call.status === 'success'
? 'text-sage-700'
: 'text-coral-700';
return (
<li
key={call.callId}
className="flex items-center gap-1.5"
data-testid="subagent-tool-call">
<span className={`text-[9px] ${tone}`}>•</span>
<span className="font-mono text-[10px] text-stone-700">{call.toolName}</span>
{call.iteration != null ? (
<span className="text-[9px] text-stone-400">·t{call.iteration}</span>
) : null}
<span className={`text-[9px] ${tone}`}>{call.status}</span>
{call.elapsedMs != null && call.status !== 'running' ? (
<span className="text-[9px] text-stone-400">
{call.elapsedMs >= 1000
? `${(call.elapsedMs / 1000).toFixed(1)}s`
: `${call.elapsedMs}ms`}
</span>
) : null}
</li>
);
})}
</ul>
) : null}
</div>
);
}
export function ToolTimelineBlock({ entries }: { entries: ToolTimelineEntry[] }) {
const latestRunningEntryId = [...entries].reverse().find(entry => entry.status === 'running')?.id;
@@ -21,6 +98,12 @@ export function ToolTimelineBlock({ entries }: { entries: ToolTimelineEntry[] })
const detailContent =
normalizeToolBody(formatted.detail) ?? normalizeToolBody(entry.argsBuffer);
const workerRef = parseWorkerThreadRef(formatted.detail ?? entry.detail);
const subagent = entry.subagent;
// A subagent row should always render the expandable details so
// its live activity is visible — even when there is no prompt
// detail to show. Mirrors the rule that a non-subagent row only
// expands when it has detail content.
const expandable = detailContent != null || subagent != null;
const shouldAutoExpand = latestRunningEntryId != null && latestRunningEntryId === entry.id;
const statusTone =
entry.status === 'running'
@@ -46,7 +129,7 @@ export function ToolTimelineBlock({ entries }: { entries: ToolTimelineEntry[] })
return (
<div key={entry.id} className="flex flex-col gap-1 text-xs text-stone-400">
{detailContent ? (
{expandable ? (
<details open={shouldAutoExpand} className="ml-1 group">
<summary className="flex cursor-pointer list-none items-center gap-2 select-none marker:hidden">
<span
@@ -70,12 +153,13 @@ export function ToolTimelineBlock({ entries }: { entries: ToolTimelineEntry[] })
className={`mt-1 rounded-xl rounded-tl-md px-2.5 py-2 text-[11px] whitespace-pre-wrap break-words ${statusTone.bubble}`}>
{formatted.detail}
</div>
) : (
) : detailContent ? (
<pre
className={`mt-1 max-h-24 overflow-y-auto rounded px-2 py-1 font-mono text-[10px] whitespace-pre-wrap break-all ${statusTone.bubble} ${statusTone.code}`}>
{detailContent}
</pre>
)}
) : null}
{subagent ? <SubagentActivityBlock subagent={subagent} /> : null}
</details>
) : (
<div className="ml-1 flex items-center gap-2">
app/src/pages/conversations/components/__tests__/ToolTimelineBlock.test.tsx
+115
View File
@@ -0,0 +1,115 @@
import { render, screen } from '@testing-library/react';
import { Provider } from 'react-redux';
import { describe, expect, it } from 'vitest';
import { store } from '../../../../store';
import type { ToolTimelineEntry } from '../../../../store/chatRuntimeSlice';
import { SubagentActivityBlock, ToolTimelineBlock } from '../ToolTimelineBlock';
// #1122 — guards the parent-thread live subagent rendering. The block
// always expands subagent rows so the activity stays visible while the
// run is in flight, even before the subagent emits any prompt detail.
function renderInStore(ui: React.ReactNode) {
return render(<Provider store={store}>{ui}</Provider>);
}
describe('SubagentActivityBlock', () => {
it('renders mode + dedicated-thread + child-turn pills', () => {
renderInStore(
<SubagentActivityBlock
subagent={{
taskId: 't',
agentId: 'researcher',
mode: 'typed',
dedicatedThread: true,
childIteration: 2,
childMaxIterations: 5,
toolCalls: [],
}}
/>
);
const block = screen.getByTestId('subagent-activity');
expect(block.textContent).toContain('typed');
expect(block.textContent).toContain('worker thread');
expect(block.textContent).toContain('turn 2/5');
});
it('renders final-run statistics on a completed sub-agent', () => {
renderInStore(
<SubagentActivityBlock
subagent={{
taskId: 't',
agentId: 'researcher',
iterations: 3,
elapsedMs: 4200,
toolCalls: [],
}}
/>
);
const block = screen.getByTestId('subagent-activity');
expect(block.textContent).toContain('3 turns');
expect(block.textContent).toContain('4.2s');
});
it('renders one row per child tool call with status + timing', () => {
renderInStore(
<SubagentActivityBlock
subagent={{
taskId: 't',
agentId: 'researcher',
toolCalls: [
{ callId: 'c1', toolName: 'web_search', status: 'success', elapsedMs: 312 },
{ callId: 'c2', toolName: 'composio_execute', status: 'running', iteration: 2 },
{ callId: 'c3', toolName: 'noisy', status: 'error', elapsedMs: 50 },
],
}}
/>
);
const calls = screen.getAllByTestId('subagent-tool-call');
expect(calls).toHaveLength(3);
expect(calls[0].textContent).toContain('web_search');
expect(calls[0].textContent).toContain('success');
expect(calls[0].textContent).toContain('312ms');
expect(calls[1].textContent).toContain('running');
expect(calls[1].textContent).toContain('·t2');
expect(calls[2].textContent).toContain('error');
});
});
describe('ToolTimelineBlock — subagent rendering', () => {
it('expands a subagent row even without prompt detail and shows child tool calls', () => {
const entry: ToolTimelineEntry = {
id: 'tid:subagent:sub-1:researcher',
name: 'subagent:researcher',
round: 1,
status: 'running',
subagent: {
taskId: 'sub-1',
agentId: 'researcher',
mode: 'typed',
childIteration: 1,
childMaxIterations: 5,
toolCalls: [{ callId: 'cc-1', toolName: 'web_search', status: 'running', iteration: 1 }],
},
};
renderInStore(<ToolTimelineBlock entries={[entry]} />);
const calls = screen.getAllByTestId('subagent-tool-call');
expect(calls).toHaveLength(1);
expect(calls[0].textContent).toContain('web_search');
expect(screen.getByTestId('subagent-activity').textContent).toContain('turn 1/5');
});
it('renders a non-subagent row without crashing when there is no detail', () => {
const entry: ToolTimelineEntry = {
id: 'plain',
name: 'list_threads',
round: 0,
status: 'success',
};
renderInStore(<ToolTimelineBlock entries={[entry]} />);
// Plain rows with no detail collapse to a flat label + status pill.
expect(screen.queryByTestId('subagent-activity')).toBeNull();
});
});
app/src/providers/ChatRuntimeProvider.tsx
+98 -10
View File
@@ -8,7 +8,6 @@ import {
type ChatIterationStartEvent,
type ChatSegmentEvent,
type ChatSubagentDoneEvent,
type ChatSubagentSpawnedEvent,
type ChatToolCallEvent,
type ChatToolResultEvent,
type ProactiveMessageEvent,
@@ -334,7 +333,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
})
);
},
onSubagentSpawned: (event: ChatSubagentSpawnedEvent) => {
onSubagentSpawned: event => {
const prev = store.getState().chatRuntime.inferenceStatusByThread[event.thread_id];
dispatch(
setInferenceStatusForThread({
@@ -361,6 +360,13 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
status: 'running',
detail: pendingContext.prompt,
sourceToolName: pendingContext.sourceToolName,
subagent: {
taskId: event.skill_id,
agentId: event.tool_name,
mode: event.subagent?.mode,
dedicatedThread: event.subagent?.dedicated_thread,
toolCalls: [],
},
}),
],
})
@@ -370,14 +376,21 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
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 =>
entry.id === subagentRowId && entry.status === 'running'
? decorateEntry({
...entry,
status: (event.success ? 'success' : 'error') as ToolTimelineEntryStatus,
})
: entry
);
const entries = existing.map(entry => {
if (entry.id !== subagentRowId || entry.status !== 'running') return entry;
return decorateEntry({
...entry,
status: (event.success ? 'success' : 'error') as ToolTimelineEntryStatus,
subagent: entry.subagent
? {
...entry.subagent,
iterations: event.subagent?.iterations ?? entry.subagent.iterations,
elapsedMs: event.subagent?.elapsed_ms ?? entry.subagent.elapsedMs,
outputChars: event.subagent?.output_chars ?? entry.subagent.outputChars,
}
: entry.subagent,
});
});
dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries }));
}
@@ -390,6 +403,81 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
})
);
},
onSubagentIterationStart: event => {
const taskId = event.subagent?.task_id ?? event.skill_id;
const agentId = event.subagent?.agent_id ?? event.tool_name;
const rowId = `${event.thread_id}:subagent:${taskId}:${agentId}`;
const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? [];
const idx = existing.findIndex(entry => entry.id === rowId);
if (idx < 0) return;
const entry = existing[idx];
if (!entry.subagent) return;
const next = [...existing];
next[idx] = {
...entry,
subagent: {
...entry.subagent,
childIteration: event.subagent?.child_iteration ?? entry.subagent.childIteration,
childMaxIterations:
event.subagent?.child_max_iterations ?? entry.subagent.childMaxIterations,
},
};
dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries: next }));
},
onSubagentToolCall: event => {
const taskId = event.subagent?.task_id ?? event.skill_id;
const agentId = event.subagent?.agent_id;
if (!agentId) return;
const rowId = `${event.thread_id}:subagent:${taskId}:${agentId}`;
const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? [];
const idx = existing.findIndex(entry => entry.id === rowId);
if (idx < 0) return;
const entry = existing[idx];
if (!entry.subagent) return;
// De-dupe on call_id — the same call should not append twice if
// the socket layer redelivers (e.g. on reconnect during a run).
if (entry.subagent.toolCalls.some(c => c.callId === event.tool_call_id)) return;
const next = [...existing];
next[idx] = {
...entry,
subagent: {
...entry.subagent,
toolCalls: [
...entry.subagent.toolCalls,
{
callId: event.tool_call_id,
toolName: event.tool_name,
status: 'running',
iteration: event.subagent?.child_iteration,
},
],
},
};
dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries: next }));
},
onSubagentToolResult: event => {
const taskId = event.subagent?.task_id ?? event.skill_id;
const agentId = event.subagent?.agent_id;
if (!agentId) return;
const rowId = `${event.thread_id}:subagent:${taskId}:${agentId}`;
const existing = store.getState().chatRuntime.toolTimelineByThread[event.thread_id] ?? [];
const idx = existing.findIndex(entry => entry.id === rowId);
if (idx < 0) return;
const entry = existing[idx];
if (!entry.subagent) return;
const callIdx = entry.subagent.toolCalls.findIndex(c => c.callId === event.tool_call_id);
if (callIdx < 0) return;
const updatedCalls = [...entry.subagent.toolCalls];
updatedCalls[callIdx] = {
...updatedCalls[callIdx],
status: event.success ? 'success' : 'error',
elapsedMs: event.subagent?.elapsed_ms ?? updatedCalls[callIdx].elapsedMs,
outputChars: event.subagent?.output_chars ?? updatedCalls[callIdx].outputChars,
};
const next = [...existing];
next[idx] = { ...entry, subagent: { ...entry.subagent, toolCalls: updatedCalls } };
dispatch(setToolTimelineForThread({ threadId: event.thread_id, entries: next }));
},
onSegment: (event: ChatSegmentEvent) => {
const eventKey = `segment:${event.thread_id}:${event.request_id}:${event.segment_index}`;
if (
app/src/providers/__tests__/ChatRuntimeProvider.test.tsx
+141
View File
@@ -358,4 +358,145 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria
expect(store.getState().chatRuntime.inferenceStatusByThread['t-err']).toBeUndefined();
});
});
// Live subagent activity (#1122) — the parent thread surfaces a
// subagent's child iterations and tool calls as they happen, then
// settles to the final-run statistics on completion. The asserts here
// are the contract the ToolTimelineBlock UI relies on; if a refactor
// moves the subagent state somewhere else this test is the canary.
describe('live subagent activity (#1122)', () => {
it('builds a live subagent block from spawned → iteration → tool call → done', () => {
const listeners = renderProvider();
const threadId = 'tsa';
act(() => {
listeners.onSubagentSpawned?.({
thread_id: threadId,
request_id: 'r1',
tool_name: 'researcher',
skill_id: 'sub-1',
message: 'spawned',
round: 1,
subagent: { mode: 'typed', dedicated_thread: false, prompt_chars: 42 },
});
});
let timeline = store.getState().chatRuntime.toolTimelineByThread[threadId] ?? [];
expect(timeline).toHaveLength(1);
expect(timeline[0]?.subagent).toMatchObject({
agentId: 'researcher',
taskId: 'sub-1',
mode: 'typed',
dedicatedThread: false,
toolCalls: [],
});
act(() => {
listeners.onSubagentIterationStart?.({
thread_id: threadId,
request_id: 'r1',
round: 1,
tool_name: 'researcher',
skill_id: 'sub-1',
message: 'iter',
subagent: {
agent_id: 'researcher',
task_id: 'sub-1',
child_iteration: 1,
child_max_iterations: 5,
},
});
listeners.onSubagentToolCall?.({
thread_id: threadId,
request_id: 'r1',
round: 1,
tool_name: 'web_search',
skill_id: 'sub-1',
tool_call_id: 'cc-1',
subagent: { agent_id: 'researcher', task_id: 'sub-1', child_iteration: 1 },
});
// Duplicate child tool_call must not double-append.
listeners.onSubagentToolCall?.({
thread_id: threadId,
request_id: 'r1',
round: 1,
tool_name: 'web_search',
skill_id: 'sub-1',
tool_call_id: 'cc-1',
subagent: { agent_id: 'researcher', task_id: 'sub-1', child_iteration: 1 },
});
});
timeline = store.getState().chatRuntime.toolTimelineByThread[threadId] ?? [];
expect(timeline[0]?.subagent?.childIteration).toBe(1);
expect(timeline[0]?.subagent?.childMaxIterations).toBe(5);
expect(timeline[0]?.subagent?.toolCalls).toEqual([
{ callId: 'cc-1', toolName: 'web_search', status: 'running', iteration: 1 },
]);
act(() => {
listeners.onSubagentToolResult?.({
thread_id: threadId,
request_id: 'r1',
round: 1,
tool_name: 'web_search',
skill_id: 'sub-1',
tool_call_id: 'cc-1',
success: true,
subagent: {
agent_id: 'researcher',
task_id: 'sub-1',
child_iteration: 1,
elapsed_ms: 312,
output_chars: 1280,
},
});
listeners.onSubagentDone?.({
thread_id: threadId,
request_id: 'r1',
tool_name: 'researcher',
skill_id: 'sub-1',
message: 'done',
success: true,
round: 1,
subagent: { iterations: 2, elapsed_ms: 4200, output_chars: 980 },
});
});
timeline = store.getState().chatRuntime.toolTimelineByThread[threadId] ?? [];
expect(timeline[0]?.status).toBe('success');
expect(timeline[0]?.subagent?.toolCalls[0]).toMatchObject({
status: 'success',
elapsedMs: 312,
outputChars: 1280,
});
expect(timeline[0]?.subagent).toMatchObject({
iterations: 2,
elapsedMs: 4200,
outputChars: 980,
});
});
it('ignores subagent_tool_call events that arrive before subagent_spawned', () => {
const listeners = renderProvider();
const threadId = 'tsa-orphan';
act(() => {
listeners.onSubagentToolCall?.({
thread_id: threadId,
request_id: 'r1',
round: 1,
tool_name: 'web_search',
skill_id: 'sub-missing',
tool_call_id: 'cc-1',
subagent: { agent_id: 'researcher', task_id: 'sub-missing', child_iteration: 1 },
});
});
// No row was created — the orphan child tool call is dropped rather
// than synthesising a partial subagent row from incomplete data.
const timeline = store.getState().chatRuntime.toolTimelineByThread[threadId] ?? [];
expect(timeline).toHaveLength(0);
});
});
});
app/src/services/__tests__/chatService.test.ts
+92
View File
@@ -76,6 +76,98 @@ describe('chatService.subscribeChatEvents', () => {
expect(onDone).toHaveBeenCalledTimes(1);
});
// #1122 — the new live subagent events must be wired up under their
// canonical snake_case names and dispatch payloads back through the
// listener interface unchanged. Without this coverage the parent
// thread's live subagent block silently goes blank if a future
// refactor renames a socket event.
it('subscribes and forwards live subagent events under canonical names', () => {
const socket = createMockSocket();
vi.mocked(socketService.getSocket).mockReturnValue(socket as never);
const onSubagentSpawned = vi.fn();
const onSubagentDone = vi.fn();
const onSubagentIterationStart = vi.fn();
const onSubagentToolCall = vi.fn();
const onSubagentToolResult = vi.fn();
subscribeChatEvents({
onSubagentSpawned,
onSubagentDone,
onSubagentIterationStart,
onSubagentToolCall,
onSubagentToolResult,
});
const subscribedEvents = socket.on.mock.calls.map(call => call[0]);
expect(subscribedEvents).toEqual([
'subagent_spawned',
'subagent_completed',
'subagent_failed',
'subagent_iteration_start',
'subagent_tool_call',
'subagent_tool_result',
]);
const spawned = {
thread_id: 't',
request_id: 'r',
tool_name: 'researcher',
skill_id: 'sub-1',
message: 'm',
round: 1,
subagent: { mode: 'typed' },
};
socket.emit('subagent_spawned', spawned);
expect(onSubagentSpawned).toHaveBeenCalledWith(spawned);
const iter = {
thread_id: 't',
request_id: 'r',
round: 1,
tool_name: 'researcher',
skill_id: 'sub-1',
message: 'iter',
subagent: {
agent_id: 'researcher',
task_id: 'sub-1',
child_iteration: 1,
child_max_iterations: 5,
},
};
socket.emit('subagent_iteration_start', iter);
expect(onSubagentIterationStart).toHaveBeenCalledWith(iter);
const call = {
thread_id: 't',
request_id: 'r',
round: 1,
tool_name: 'web_search',
skill_id: 'sub-1',
tool_call_id: 'cc-1',
subagent: { agent_id: 'researcher', task_id: 'sub-1', child_iteration: 1 },
};
socket.emit('subagent_tool_call', call);
expect(onSubagentToolCall).toHaveBeenCalledWith(call);
socket.emit('subagent_tool_result', { ...call, success: true });
expect(onSubagentToolResult).toHaveBeenCalledWith({ ...call, success: true });
// Both completion paths route through the same listener.
const done = {
thread_id: 't',
request_id: 'r',
tool_name: 'researcher',
skill_id: 'sub-1',
message: 'done',
success: true,
round: 1,
};
socket.emit('subagent_completed', done);
socket.emit('subagent_failed', { ...done, success: false });
expect(onSubagentDone).toHaveBeenCalledTimes(2);
});
it('removes all handlers on cleanup', () => {
const socket = createMockSocket();
vi.mocked(socketService.getSocket).mockReturnValue(socket as never);
app/src/services/chatService.ts
+134 -1
View File
@@ -137,6 +137,82 @@ export interface ChatSubagentDoneEvent {
message: string;
success: boolean;
round: number;
/** Per-event subagent detail. Mirrors `SubagentProgressDetail` in core. */
subagent?: SubagentProgressDetail;
}
/**
* Per-event subagent detail attached to live subagent activity events
* (`subagent_spawned`, `subagent_completed`, `subagent_iteration_start`,
* `subagent_tool_call`, `subagent_tool_result`).
*
* Matches the Rust `SubagentProgressDetail` struct in
* `src/core/socketio.rs` — every field is optional so older cores that
* don't emit it stay parseable.
*/
export interface SubagentProgressDetail {
mode?: string;
dedicated_thread?: boolean;
prompt_chars?: number;
child_iteration?: number;
child_max_iterations?: number;
agent_id?: string;
task_id?: string;
elapsed_ms?: number;
iterations?: number;
output_chars?: number;
}
/** Extended payload for `subagent_spawned`. */
export interface ChatSubagentSpawnedEventV2 extends ChatSubagentSpawnedEvent {
subagent?: SubagentProgressDetail;
}
/**
* Emitted at the start of each LLM iteration *inside* a running
* sub-agent. Lets the parent thread surface child progress (which round
* the subagent is on, its iteration cap) without flattening it into the
* parent's own iteration counter.
*/
export interface ChatSubagentIterationStartEvent {
thread_id: string;
request_id: string;
/** Parent's iteration index (inherited from the parent context). */
round: number;
/** Subagent's agent id. Mirrored on the flat `tool_name` field. */
tool_name: string;
/** Subagent's task id (the spawn id). */
skill_id: string;
message: string;
subagent?: SubagentProgressDetail;
}
/** Emitted when a sub-agent starts executing one of its own tools. */
export interface ChatSubagentToolCallEvent {
thread_id: string;
request_id: string;
round: number;
/** Child's tool name (e.g. `composio_execute`, `web_search`). */
tool_name: string;
/** Subagent's task id. */
skill_id: string;
/** Provider-assigned tool call id. */
tool_call_id: string;
subagent?: SubagentProgressDetail;
}
/** Emitted when a sub-agent's tool execution finishes. */
export interface ChatSubagentToolResultEvent {
thread_id: string;
request_id: string;
round: number;
tool_name: string;
skill_id: string;
tool_call_id: string;
success: boolean;
/** Stringified JSON `{ output_chars, elapsed_ms }` matching `tool_result`. */
output?: string;
subagent?: SubagentProgressDetail;
}
/**
@@ -187,8 +263,11 @@ export interface ChatEventListeners {
onIterationStart?: (event: ChatIterationStartEvent) => void;
onToolCall?: (event: ChatToolCallEvent) => void;
onToolResult?: (event: ChatToolResultEvent) => void;
onSubagentSpawned?: (event: ChatSubagentSpawnedEvent) => void;
onSubagentSpawned?: (event: ChatSubagentSpawnedEventV2) => void;
onSubagentDone?: (event: ChatSubagentDoneEvent) => void;
onSubagentIterationStart?: (event: ChatSubagentIterationStartEvent) => void;
onSubagentToolCall?: (event: ChatSubagentToolCallEvent) => void;
onSubagentToolResult?: (event: ChatSubagentToolResultEvent) => void;
onSegment?: (event: ChatSegmentEvent) => void;
onTextDelta?: (event: ChatTextDeltaEvent) => void;
onThinkingDelta?: (event: ChatThinkingDeltaEvent) => void;
@@ -214,6 +293,9 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
subagentSpawned: 'subagent_spawned',
subagentCompleted: 'subagent_completed',
subagentFailed: 'subagent_failed',
subagentIterationStart: 'subagent_iteration_start',
subagentToolCall: 'subagent_tool_call',
subagentToolResult: 'subagent_tool_result',
segment: 'chat_segment',
textDelta: 'text_delta',
thinkingDelta: 'thinking_delta',
@@ -335,6 +417,57 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
handlers.push([EVENTS.subagentFailed, onFailed]);
}
if (listeners.onSubagentIterationStart) {
const cb = (payload: unknown) => {
const e = payload as ChatSubagentIterationStartEvent;
chatLog(
'%s thread_id=%s task=%s child_round=%s/%s',
EVENTS.subagentIterationStart,
e.thread_id,
e.skill_id,
e.subagent?.child_iteration,
e.subagent?.child_max_iterations
);
listeners.onSubagentIterationStart?.(e);
};
socket.on(EVENTS.subagentIterationStart, cb);
handlers.push([EVENTS.subagentIterationStart, cb]);
}
if (listeners.onSubagentToolCall) {
const cb = (payload: unknown) => {
const e = payload as ChatSubagentToolCallEvent;
chatLog(
'%s thread_id=%s task=%s child_tool=%s call_id=%s',
EVENTS.subagentToolCall,
e.thread_id,
e.skill_id,
e.tool_name,
e.tool_call_id
);
listeners.onSubagentToolCall?.(e);
};
socket.on(EVENTS.subagentToolCall, cb);
handlers.push([EVENTS.subagentToolCall, cb]);
}
if (listeners.onSubagentToolResult) {
const cb = (payload: unknown) => {
const e = payload as ChatSubagentToolResultEvent;
chatLog(
'%s thread_id=%s task=%s child_tool=%s success=%s',
EVENTS.subagentToolResult,
e.thread_id,
e.skill_id,
e.tool_name,
e.success
);
listeners.onSubagentToolResult?.(e);
};
socket.on(EVENTS.subagentToolResult, cb);
handlers.push([EVENTS.subagentToolResult, cb]);
}
if (listeners.onSegment) {
const cb = (payload: unknown) => {
const e = payload as ChatSegmentEvent;
app/src/store/chatRuntimeSlice.ts
+55
View File
@@ -12,6 +12,54 @@ export interface InferenceStatus {
activeSubagent?: string;
}
/**
* Per-subagent live activity attached to a `subagent:*` timeline row.
*
* Carries everything the parent thread's UI needs to render a live
* subagent block — child iteration counter, mode, dedicated-thread
* flag, final-run statistics, and a flat list of child tool calls
* the subagent has executed during its run. Populated incrementally
* from the new `subagent_*` socket events; absent on plain (legacy)
* subagent rows so older snapshots stay renderable unchanged.
*/
export interface SubagentActivity {
/** Spawn task id (`sub-…`). Stable for the lifetime of one delegation. */
taskId: string;
/** Sub-agent definition id (e.g. `researcher`). */
agentId: string;
/** Resolved spawn mode — `"typed"` or `"fork"`. */
mode?: string;
/** `true` when the spawn requested a dedicated worker thread. */
dedicatedThread?: boolean;
/** Sub-agent's current 1-based iteration index (live). */
childIteration?: number;
/** Sub-agent's iteration cap. */
childMaxIterations?: number;
/** Total iterations once the sub-agent finishes. */
iterations?: number;
/** Wall-clock ms once the sub-agent finishes. */
elapsedMs?: number;
/** Character length of the final assistant text. */
outputChars?: number;
/** Child tool calls executed inside the sub-agent, in arrival order. */
toolCalls: SubagentToolCallEntry[];
}
/** One child tool call performed by a running sub-agent. */
export interface SubagentToolCallEntry {
/** Provider-assigned tool call id. */
callId: string;
/** Child's tool name. */
toolName: string;
status: ToolTimelineEntryStatus;
/** 1-based child iteration the call belongs to. */
iteration?: number;
/** Wall-clock ms the call took (set on completion). */
elapsedMs?: number;
/** Character length of the tool result (set on completion). */
outputChars?: number;
}
export interface ToolTimelineEntry {
id: string;
name: string;
@@ -21,6 +69,13 @@ export interface ToolTimelineEntry {
displayName?: string;
detail?: string;
sourceToolName?: string;
/**
* Live sub-agent activity for `subagent:*` rows. Built up from the
* `subagent_iteration_start` / `subagent_tool_call` /
* `subagent_tool_result` socket events. Absent for non-subagent
* rows and for legacy snapshots emitted by older cores.
*/
subagent?: SubagentActivity;
}
export interface StreamingAssistantState {
src/core/socketio.rs
+60
View File
@@ -72,6 +72,66 @@ pub struct WebChannelEvent {
/// Optional citations attached to `chat_done` payloads.
#[serde(skip_serializing_if = "Option::is_none")]
pub citations: Option<serde_json::Value>,
/// Sub-agent specific progress detail. Populated on
/// `subagent_spawned`, `subagent_completed`, `subagent_iteration_start`,
/// `subagent_tool_call`, and `subagent_tool_result` events so the UI
/// can attribute child activity to the parent's live subagent row
/// without overloading the flat top-level fields. `None` for any
/// non-subagent event.
#[serde(skip_serializing_if = "Option::is_none")]
pub subagent: Option<SubagentProgressDetail>,
}
/// Per-event subagent progress detail attached to `WebChannelEvent`.
///
/// Carries the fields the parent thread's UI needs to render a live
/// subagent block — child iteration counters, mode, child task/agent
/// ids when distinct from the flat `tool_name` (which already carries
/// the agent id on top-level subagent events but not on nested
/// `subagent_tool_*` events where `tool_name` is the *child's* tool),
/// and final-run statistics on `subagent_completed`.
///
/// Every field is optional and skipped from the JSON payload when
/// absent — this keeps the wire format compact for non-subagent events
/// (where the whole struct is `None`) and lets new fields land
/// non-breakingly behind older clients.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct SubagentProgressDetail {
/// Resolved spawn mode — `"typed"` or `"fork"`.
#[serde(skip_serializing_if = "Option::is_none")]
pub mode: Option<String>,
/// Whether the spawn requested a dedicated worker thread.
#[serde(skip_serializing_if = "Option::is_none")]
pub dedicated_thread: Option<bool>,
/// Character length of the delegation prompt (on `subagent_spawned`).
#[serde(skip_serializing_if = "Option::is_none")]
pub prompt_chars: Option<u64>,
/// Sub-agent's child iteration counter (on `subagent_iteration_start`,
/// `subagent_tool_call`, `subagent_tool_result`). 1-based.
#[serde(skip_serializing_if = "Option::is_none")]
pub child_iteration: Option<u32>,
/// Sub-agent's configured iteration cap.
#[serde(skip_serializing_if = "Option::is_none")]
pub child_max_iterations: Option<u32>,
/// Child agent id (on nested `subagent_tool_*` events where the flat
/// `tool_name` is the child's tool, not the agent).
#[serde(skip_serializing_if = "Option::is_none")]
pub agent_id: Option<String>,
/// Spawn task id (on nested `subagent_tool_*` events).
#[serde(skip_serializing_if = "Option::is_none")]
pub task_id: Option<String>,
/// Elapsed wall-clock for the call/run in milliseconds.
#[serde(skip_serializing_if = "Option::is_none")]
pub elapsed_ms: Option<u64>,
/// Total iterations the sub-agent used (on `subagent_completed`).
#[serde(skip_serializing_if = "Option::is_none")]
pub iterations: Option<u32>,
/// Character length of the sub-agent's final assistant text
/// (on `subagent_completed`) or the tool result
/// (on `subagent_tool_result`).
#[serde(skip_serializing_if = "Option::is_none")]
pub output_chars: Option<u64>,
}
#[derive(Debug, Deserialize)]
src/openhuman/agent/harness/fork_context.rs
+10
View File
@@ -17,6 +17,7 @@
//! Both contexts are stashed in `Arc`s so that cloning into the child
//! costs a refcount bump rather than a full copy.
use crate::openhuman::agent::progress::AgentProgress;
use crate::openhuman::config::AgentConfig;
use crate::openhuman::memory::Memory;
use crate::openhuman::providers::{ChatMessage, Provider};
@@ -118,6 +119,15 @@ pub struct ParentExecutionContext {
/// `Some(parent.session_key)`. A grand-child observes
/// `Some("{grandparent_key}__{parent_key}")`.
pub session_parent_prefix: Option<String>,
/// Parent's progress sink. When set, the sub-agent runner emits
/// `AgentProgress::Subagent*` lifecycle events through this channel
/// so the web-channel bridge can stream live child activity (each
/// iteration boundary, child tool call/result) into the parent
/// thread's UI. `None` for parent contexts that don't subscribe to
/// progress (e.g. CLI direct calls); the runner becomes a no-op for
/// child progress in that case.
pub on_progress: Option<tokio::sync::mpsc::Sender<AgentProgress>>,
}
tokio::task_local! {
src/openhuman/agent/harness/session/turn.rs
+1
View File
@@ -1072,6 +1072,7 @@ impl Agent {
tool_call_format: self.tool_dispatcher.tool_call_format(),
session_key: self.session_key.clone(),
session_parent_prefix: self.session_parent_prefix.clone(),
on_progress: self.on_progress.clone(),
}
}
src/openhuman/agent/harness/subagent_runner/ops.rs
+49 -5
View File
@@ -30,6 +30,7 @@ use super::tool_prep::{
use super::types::{SubagentMode, SubagentRunError, SubagentRunOptions, SubagentRunOutcome};
use crate::openhuman::agent::harness::definition::{AgentDefinition, PromptSource};
use crate::openhuman::agent::harness::with_current_sandbox_mode;
use crate::openhuman::agent::progress::AgentProgress;
use crate::openhuman::context::prompt::{
render_subagent_system_prompt, PromptContext, PromptTool, SubagentRenderOptions,
};
@@ -924,6 +925,11 @@ async fn run_inner_loop(
}
};
// 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`.
let progress_sink = parent.on_progress.clone();
for iteration in 0..max_iterations {
tracing::debug!(
task_id = %task_id,
@@ -933,6 +939,17 @@ async fn run_inner_loop(
"[subagent_runner] iteration start"
);
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentIterationStarted {
agent_id: agent_id.to_string(),
task_id: task_id.to_string(),
iteration: (iteration + 1) as u32,
max_iterations: max_iterations as u32,
})
.await;
}
let resp = provider
.chat(
ChatRequest {
@@ -1027,6 +1044,18 @@ async fn run_inner_loop(
// XmlToolDispatcher's `format_results`.
let mut text_mode_result_block = String::new();
for call in &native_calls {
let call_started = Instant::now();
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentToolCallStarted {
agent_id: agent_id.to_string(),
task_id: task_id.to_string(),
call_id: call.id.clone(),
tool_name: call.name.clone(),
iteration: (iteration + 1) as u32,
})
.await;
}
let result_text = if !allowed_names.contains(&call.name) {
tracing::warn!(
task_id = %task_id,
@@ -1118,12 +1147,12 @@ async fn run_inner_loop(
result_text
};
let call_success = !result_text.starts_with("Error");
let call_output_chars = result_text.chars().count();
let call_elapsed_ms = call_started.elapsed().as_millis() as u64;
if force_text_mode {
let status = if result_text.starts_with("Error") {
"error"
} else {
"ok"
};
let status = if call_success { "ok" } else { "error" };
let _ = std::fmt::Write::write_fmt(
&mut text_mode_result_block,
format_args!(
@@ -1138,6 +1167,21 @@ async fn run_inner_loop(
});
history.push(ChatMessage::tool(tool_msg.to_string()));
}
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentToolCallCompleted {
agent_id: agent_id.to_string(),
task_id: task_id.to_string(),
call_id: call.id.clone(),
tool_name: call.name.clone(),
success: call_success,
output_chars: call_output_chars,
elapsed_ms: call_elapsed_ms,
iteration: (iteration + 1) as u32,
})
.await;
}
}
if force_text_mode && !text_mode_result_block.is_empty() {
src/openhuman/agent/harness/subagent_runner/ops_tests.rs
+95
View File
@@ -227,6 +227,7 @@ fn make_parent(provider: Arc<dyn Provider>, tools: Vec<Box<dyn Tool>>) -> Parent
tool_call_format: crate::openhuman::context::prompt::ToolCallFormat::PFormat,
session_key: "0_test".into(),
session_parent_prefix: None,
on_progress: None,
}
}
@@ -609,3 +610,97 @@ async fn runner_errors_outside_parent_context() {
let result = run_subagent(&def, "x", SubagentRunOptions::default()).await;
assert!(matches!(result, Err(SubagentRunError::NoParentContext)));
}
/// #1122 — when the parent attaches a progress sink, the inner loop
/// emits `SubagentIterationStarted` for each round and a paired
/// `SubagentToolCallStarted` / `SubagentToolCallCompleted` for each
/// child tool call. The web-channel bridge translates these into the
/// `subagent_iteration_start` / `subagent_tool_call` /
/// `subagent_tool_result` socket events the parent thread renders.
#[tokio::test]
async fn typed_mode_emits_child_progress_events_when_sink_attached() {
use crate::openhuman::agent::progress::AgentProgress;
let provider = ScriptedProvider::new(vec![
tool_response("file_read", "{\"path\":\"x\"}"),
text_response("done"),
]);
let mut parent = make_parent(provider, vec![stub("file_read")]);
// Wire the parent's progress sink so the runner re-emits child
// lifecycle events through the same channel a real session would
// expose to the web bridge.
let (tx, mut rx) = tokio::sync::mpsc::channel::<AgentProgress>(64);
parent.on_progress = Some(tx);
let def = make_def_named_tools(&["file_read"]);
let outcome = with_parent_context(parent, async {
run_subagent(&def, "read x", SubagentRunOptions::default()).await
})
.await
.expect("runner should succeed");
assert_eq!(outcome.iterations, 2);
// Drain everything the runner sent. The receiver's sender half is
// dropped when `parent` falls out of scope above, so `recv` returns
// None once the queue empties.
let mut events = Vec::new();
while let Some(ev) = rx.recv().await {
events.push(ev);
}
let iter_starts = events
.iter()
.filter(|e| matches!(e, AgentProgress::SubagentIterationStarted { .. }))
.count();
assert_eq!(iter_starts, 2, "one iteration_start per round");
let tool_starts: Vec<_> = events
.iter()
.filter_map(|e| match e {
AgentProgress::SubagentToolCallStarted {
call_id,
tool_name,
iteration,
..
} => Some((call_id.clone(), tool_name.clone(), *iteration)),
_ => None,
})
.collect();
assert_eq!(tool_starts.len(), 1);
assert_eq!(tool_starts[0].1, "file_read");
assert_eq!(tool_starts[0].2, 1);
let tool_done: Vec<_> = events
.iter()
.filter_map(|e| match e {
AgentProgress::SubagentToolCallCompleted {
call_id,
success,
iteration,
..
} => Some((call_id.clone(), *success, *iteration)),
_ => None,
})
.collect();
assert_eq!(tool_done.len(), 1);
assert_eq!(tool_done[0].0, tool_starts[0].0, "matching call_id pair");
assert!(tool_done[0].1, "stub tool returns ok");
assert_eq!(tool_done[0].2, 1);
}
/// Runs without an attached sink must remain backwards compatible — the
/// runner is a no-op for child progress and the outcome is unchanged.
#[tokio::test]
async fn typed_mode_progress_emission_is_a_noop_without_sink() {
let provider = ScriptedProvider::new(vec![text_response("done")]);
let parent = make_parent(provider, vec![]);
assert!(parent.on_progress.is_none());
let def = make_def_named_tools(&[]);
let outcome = with_parent_context(parent, async {
run_subagent(&def, "x", SubagentRunOptions::default()).await
})
.await
.expect("runner should succeed");
assert_eq!(outcome.iterations, 1);
}
src/openhuman/agent/progress.rs
+64 -1
View File
@@ -53,13 +53,35 @@ pub enum AgentProgress {
},
/// A sub-agent was spawned during tool execution.
SubagentSpawned { agent_id: String, task_id: String },
SubagentSpawned {
agent_id: String,
task_id: String,
/// Resolved spawn mode — `"typed"` or `"fork"`. The UI uses this
/// to distinguish narrow-prompt delegations from prefix-replay
/// forks when labelling the live subagent block.
mode: String,
/// `true` when the spawn was requested with
/// `dedicated_thread: true`. The UI links the inline subagent
/// row to the eventual worker thread once the run completes.
dedicated_thread: bool,
/// Character length of the delegated prompt — useful to decide
/// whether to render the prompt detail inline or behind a
/// "show more" affordance.
prompt_chars: usize,
},
/// A sub-agent completed successfully.
SubagentCompleted {
agent_id: String,
task_id: String,
elapsed_ms: u64,
/// Number of LLM iterations the sub-agent actually used. The
/// UI surfaces this in the parent thread's subagent row so a
/// completed delegation reads as "researcher · 3 turns · 4.2s"
/// instead of just "done".
iterations: u32,
/// Character length of the sub-agent's final assistant text.
output_chars: usize,
},
/// A sub-agent failed.
@@ -69,6 +91,47 @@ pub enum AgentProgress {
error: 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
/// outer parent loop uses [`Self::IterationStarted`] for its own
/// rounds. Carries the child's `task_id` so the UI can attribute
/// the round to a specific live subagent row.
SubagentIterationStarted {
agent_id: String,
task_id: String,
/// 1-based child iteration index.
iteration: u32,
/// Maximum iterations configured for this child run.
max_iterations: u32,
},
/// A sub-agent is about to execute a tool. Distinct from
/// [`Self::ToolCallStarted`] so the parent thread can render
/// child-tool activity nested under the subagent row instead of
/// flattened into the parent's tool timeline.
SubagentToolCallStarted {
agent_id: String,
task_id: String,
call_id: String,
tool_name: String,
/// 1-based child iteration index this call belongs to.
iteration: u32,
},
/// A sub-agent's tool execution finished.
SubagentToolCallCompleted {
agent_id: String,
task_id: String,
call_id: String,
tool_name: String,
success: bool,
output_chars: usize,
elapsed_ms: u64,
/// 1-based child iteration index.
iteration: u32,
},
/// A chunk of visible assistant text arrived from the provider
/// while the current iteration is still in flight.
TextDelta {
src/openhuman/agent/triage/escalation.rs
+4
View File
@@ -172,6 +172,10 @@ async fn dispatch_target_agent(agent_id: &str, prompt: &str) -> anyhow::Result<S
// preserving the `{parent}__{child}.jsonl` hierarchy.
session_key: agent.session_key().to_string(),
session_parent_prefix: agent.session_parent_prefix().map(str::to_string),
// Triage runs sub-agents synchronously without streaming progress
// back to a UI; the runner skips child-progress emission when this
// is `None`.
on_progress: None,
};
tracing::debug!(
src/openhuman/channels/proactive.rs
+1
View File
@@ -139,6 +139,7 @@ impl EventHandler for ProactiveMessageSubscriber {
delta_kind: None,
tool_call_id: None,
citations: None,
subagent: None,
});
// 2. If an active external channel is configured, deliver there too.
src/openhuman/channels/providers/presentation.rs
+3
View File
@@ -64,6 +64,7 @@ pub async fn deliver_response(
delta: None,
delta_kind: None,
tool_call_id: None,
subagent: None,
citations: if citations.is_empty() {
None
} else {
@@ -103,6 +104,7 @@ pub async fn deliver_response(
delta: None,
delta_kind: None,
tool_call_id: None,
subagent: None,
citations: if i == 0 && !citations.is_empty() {
Some(serde_json::json!(citations))
} else {
@@ -132,6 +134,7 @@ pub async fn deliver_response(
delta: None,
delta_kind: None,
tool_call_id: None,
subagent: None,
citations: if citations.is_empty() {
None
} else {
src/openhuman/channels/providers/web.rs
+115 -36
View File
@@ -7,7 +7,7 @@ use tokio::sync::{broadcast, Mutex};
use uuid::Uuid;
use crate::core::all::{ControllerFuture, RegisteredController};
use crate::core::socketio::WebChannelEvent;
use crate::core::socketio::{SubagentProgressDetail, WebChannelEvent};
use crate::core::{ControllerSchema, FieldSchema, TypeSchema};
use crate::openhuman::agent::Agent;
use crate::openhuman::config::rpc as config_rpc;
@@ -168,6 +168,7 @@ pub async fn start_chat(
delta_kind: None,
tool_call_id: None,
citations: None,
subagent: None,
});
}
}
@@ -234,6 +235,7 @@ pub async fn start_chat(
delta_kind: None,
tool_call_id: None,
citations: None,
subagent: None,
});
}
}
@@ -326,6 +328,7 @@ pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result<Option<Stri
delta_kind: None,
tool_call_id: None,
citations: None,
subagent: None,
});
}
@@ -596,6 +599,7 @@ fn spawn_progress_bridge(
delta_kind: None,
tool_call_id: None,
citations: None,
subagent: None,
});
}
AgentProgress::IterationStarted {
@@ -624,6 +628,7 @@ fn spawn_progress_bridge(
delta_kind: None,
tool_call_id: None,
citations: None,
subagent: None,
});
}
AgentProgress::ToolCallStarted {
@@ -670,58 +675,57 @@ fn spawn_progress_bridge(
..Default::default()
});
}
AgentProgress::SubagentSpawned { agent_id, task_id } => {
AgentProgress::SubagentSpawned {
agent_id,
task_id,
mode,
dedicated_thread,
prompt_chars,
} => {
publish_web_channel_event(WebChannelEvent {
event: "subagent_spawned".to_string(),
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: request_id.clone(),
full_response: None,
message: Some(format!("Sub-agent '{agent_id}' spawned")),
error_type: None,
tool_name: Some(agent_id),
skill_id: Some(task_id),
args: None,
output: None,
success: None,
round: Some(round),
reaction_emoji: None,
segment_index: None,
segment_total: None,
delta: None,
delta_kind: None,
tool_call_id: None,
citations: None,
subagent: Some(SubagentProgressDetail {
mode: Some(mode),
dedicated_thread: Some(dedicated_thread),
prompt_chars: Some(prompt_chars as u64),
..Default::default()
}),
..Default::default()
});
}
AgentProgress::SubagentCompleted {
agent_id,
task_id,
elapsed_ms,
iterations,
output_chars,
} => {
publish_web_channel_event(WebChannelEvent {
event: "subagent_completed".to_string(),
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: request_id.clone(),
full_response: None,
message: Some(format!(
"Sub-agent '{agent_id}' completed in {elapsed_ms}ms"
)),
error_type: None,
tool_name: Some(agent_id),
skill_id: Some(task_id),
args: None,
output: None,
success: Some(true),
round: Some(round),
reaction_emoji: None,
segment_index: None,
segment_total: None,
delta: None,
delta_kind: None,
tool_call_id: None,
citations: None,
subagent: Some(SubagentProgressDetail {
elapsed_ms: Some(elapsed_ms),
iterations: Some(iterations),
output_chars: Some(output_chars as u64),
..Default::default()
}),
..Default::default()
});
}
AgentProgress::SubagentFailed {
@@ -734,22 +738,97 @@ fn spawn_progress_bridge(
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: request_id.clone(),
full_response: None,
message: Some(error),
error_type: None,
tool_name: Some(agent_id),
skill_id: Some(task_id),
args: None,
output: None,
success: Some(false),
round: Some(round),
reaction_emoji: None,
segment_index: None,
segment_total: None,
delta: None,
delta_kind: None,
tool_call_id: None,
citations: None,
..Default::default()
});
}
AgentProgress::SubagentIterationStarted {
agent_id,
task_id,
iteration,
max_iterations,
} => {
publish_web_channel_event(WebChannelEvent {
event: "subagent_iteration_start".to_string(),
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: request_id.clone(),
message: Some(format!(
"Sub-agent '{agent_id}' iteration {iteration}/{max_iterations}"
)),
tool_name: Some(agent_id),
skill_id: Some(task_id),
round: Some(round),
subagent: Some(SubagentProgressDetail {
child_iteration: Some(iteration),
child_max_iterations: Some(max_iterations),
..Default::default()
}),
..Default::default()
});
}
AgentProgress::SubagentToolCallStarted {
agent_id,
task_id,
call_id,
tool_name,
iteration,
} => {
publish_web_channel_event(WebChannelEvent {
event: "subagent_tool_call".to_string(),
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: request_id.clone(),
tool_name: Some(tool_name),
skill_id: Some(task_id.clone()),
round: Some(round),
tool_call_id: Some(call_id),
subagent: Some(SubagentProgressDetail {
child_iteration: Some(iteration),
agent_id: Some(agent_id),
task_id: Some(task_id),
..Default::default()
}),
..Default::default()
});
}
AgentProgress::SubagentToolCallCompleted {
agent_id,
task_id,
call_id,
tool_name,
success,
output_chars,
elapsed_ms,
iteration,
} => {
publish_web_channel_event(WebChannelEvent {
event: "subagent_tool_result".to_string(),
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: request_id.clone(),
tool_name: Some(tool_name),
skill_id: Some(task_id.clone()),
success: Some(success),
round: Some(round),
tool_call_id: Some(call_id),
output: Some(
json!({"output_chars": output_chars, "elapsed_ms": elapsed_ms})
.to_string(),
),
subagent: Some(SubagentProgressDetail {
child_iteration: Some(iteration),
agent_id: Some(agent_id),
task_id: Some(task_id),
elapsed_ms: Some(elapsed_ms),
output_chars: Some(output_chars as u64),
..Default::default()
}),
..Default::default()
});
}
AgentProgress::TextDelta { delta, iteration } => {
src/openhuman/tools/impl/agent/spawn_subagent.rs
+43 -1
View File
@@ -24,6 +24,7 @@ use crate::openhuman::agent::harness::fork_context::current_parent;
use crate::openhuman::agent::harness::subagent_runner::{
run_subagent, SubagentRunOptions, SubagentRunOutcome,
};
use crate::openhuman::agent::progress::AgentProgress;
use crate::openhuman::memory::conversations::{
self, ConversationMessage, CreateConversationThread,
};
@@ -326,6 +327,23 @@ impl Tool for SpawnSubagentTool {
prompt_chars: prompt.chars().count(),
});
// Mirror the spawn onto the parent's per-turn progress sink so the
// web-channel bridge can stream a live subagent row into the
// parent thread's UI. Best-effort: a closed/missing sink is
// silently ignored — the global DomainEvent above is the
// authoritative record.
if let Some(progress) = current_parent().and_then(|p| p.on_progress.clone()) {
let _ = progress
.send(AgentProgress::SubagentSpawned {
agent_id: definition.id.clone(),
task_id: task_id.clone(),
mode: mode.to_string(),
dedicated_thread,
prompt_chars: prompt.chars().count(),
})
.await;
}
// ── Run the sub-agent ──────────────────────────────────────────
let options = SubagentRunOptions {
skill_filter_override: None,
@@ -334,6 +352,8 @@ impl Tool for SpawnSubagentTool {
task_id: Some(task_id.clone()),
};
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 {
@@ -345,6 +365,18 @@ impl Tool for SpawnSubagentTool {
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;
}
if dedicated_thread {
let workspace_dir = current_parent()
.map(|p| p.workspace_dir.clone())
@@ -403,10 +435,20 @@ impl Tool for SpawnSubagentTool {
);
publish_global(DomainEvent::SubagentFailed {
parent_session,
task_id,
task_id: task_id.clone(),
agent_id: definition.id.clone(),
error: message.clone(),
});
if let Some(ref tx) = progress_sink {
let _ = tx
.send(AgentProgress::SubagentFailed {
agent_id: definition.id.clone(),
task_id: task_id.clone(),
error: message.clone(),
})
.await;
}
// Surface as a non-fatal tool error so the parent model
// can react and (e.g.) retry with different params.
Ok(ToolResult::error(parent_visible_error))
tests/agent_harness_public.rs
+1
View File
@@ -139,6 +139,7 @@ fn stub_parent_context() -> ParentExecutionContext {
tool_call_format: openhuman_core::openhuman::context::prompt::ToolCallFormat::PFormat,
session_key: "test-session".into(),
session_parent_prefix: None,
on_progress: None,
}
}
tests/calendar_grounding_e2e.rs
+1
View File
@@ -156,6 +156,7 @@ async fn test_integrations_agent_has_current_date_context() -> Result<()> {
tool_call_format: openhuman_core::openhuman::context::prompt::ToolCallFormat::PFormat,
session_key: "0_test".into(),
session_parent_prefix: None,
on_progress: None,
};
let def =