feat(agent-core): stream live subagent events into background task output

.changeset/stream-agent-task-events.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Stream live subagent events into the background task output buffer. While a subagent launched through the `Agent` tool is still running, `TaskOutput` and the `/tasks` panel now show its turns, tool calls, and thinking/assistant text as they happen, instead of only the final summary after completion.

docs/en/reference/tools.md

@@ -109,7 +109,7 @@ Background task tools manage tasks started via `Bash`, `Agent`, or `AskUserQuest
 
 **`TaskList`** returns the list of background tasks. Optional parameters: `active_only` (defaults to true; lists only running tasks) and `limit` (defaults to 20; range 1–100).
 
-**`TaskOutput`** returns the status and output of a task given its `task_id`. The inline preview includes at most the most recent 32 KB of content; the full log is saved to disk, and the tool also returns an `output_path` with a suggestion to use `Read` for paginated access. Optional `block` (defaults to false) and `timeout` (seconds to wait; defaults to 30; range 0–3600) parameters allow waiting for the task to complete before returning.
+**`TaskOutput`** returns the status and output of a task given its `task_id`. The inline preview includes at most the most recent 32 KB of content; the full log is saved to disk, and the tool also returns an `output_path` with a suggestion to use `Read` for paginated access. For `Agent` subagent tasks, the output streams live: turns, tool calls, and thinking/assistant text appear as they happen, so this is how you watch a subagent work in real time. Optional `block` (defaults to false) and `timeout` (seconds to wait; defaults to 30; range 0–3600) parameters allow waiting for the task to complete before returning.
 
 **`TaskStop`** accepts a `task_id` and optional `reason` (defaults to `Stopped by TaskStop`). Safe to call on tasks that are already in a terminal state.
 

docs/zh/reference/tools.md

@@ -109,7 +109,7 @@ Plan 模式是一种受约束的工作状态：进入后 `Write` 与 `Edit` 只
 
 **`TaskList`** 返回后台任务列表。可选参数 `active_only`（默认 true，仅列出运行中的任务）和 `limit`（默认 20，取值范围 1–100）。
 
-**`TaskOutput`** 根据 `task_id` 返回任务状态与输出。内联预览最多包含最近 32 KB 的内容；完整日志保存在磁盘上，工具会一并返回 `output_path` 并提示通过 `Read` 分页读取。可选 `block`（默认 false）和 `timeout`（等待秒数，默认 30，取值范围 0–3600）参数可用于等待任务完成后再返回。
+**`TaskOutput`** 根据 `task_id` 返回任务状态与输出。内联预览最多包含最近 32 KB 的内容；完整日志保存在磁盘上，工具会一并返回 `output_path` 并提示通过 `Read` 分页读取。对于 `Agent` 子代理任务，输出是实时流式的：轮次、工具调用和思考/回复内容都会随执行即时写入，可以用它实时观察子代理的工作过程。可选 `block`（默认 false）和 `timeout`（等待秒数，默认 30，取值范围 0–3600）参数可用于等待任务完成后再返回。
 
 **`TaskStop`** 接受 `task_id` 和可选的 `reason`（默认 `Stopped by TaskStop`）。对已处于终止状态的任务也能安全调用。
 

packages/agent-core/src/agent/background/agent-task.ts

@@ -1,4 +1,6 @@
+import type { AgentEventListener } from '../../agent';
 import { errorMessage, isAbortError } from '../../loop/errors';
+import type { AgentEvent, AssistantDeltaEvent, ThinkingDeltaEvent } from '../../rpc';
 import {
   type BackgroundTask,
   type BackgroundTaskInfoBase,
@@ -40,6 +42,8 @@ export class AgentBackgroundTask implements BackgroundTask {
       sink.signal.addEventListener('abort', requestAbort, { once: true });
     }
 
+    const unsubscribe = this.handle.subscribeToEvents?.(createEventStreamer(sink));
+
     try {
       const outcome = await this.handle.completion;
       sink.appendOutput(outcome.result);
@@ -51,6 +55,7 @@ export class AgentBackgroundTask implements BackgroundTask {
       }
       await sink.settle({ status: 'failed', stopReason: errorMessage(error) });
     } finally {
+      unsubscribe?.();
       sink.signal.removeEventListener('abort', requestAbort);
     }
   }
@@ -68,3 +73,73 @@ export class AgentBackgroundTask implements BackgroundTask {
     };
   }
 }
+
+const EVENT_PREVIEW_LENGTH = 200;
+
+type TextStreamKind = 'thinking' | 'assistant';
+
+/**
+ * Build the event listener that mirrors a subagent's live activity into the
+ * task output buffer. Thinking/assistant text streams verbatim so the buffer
+ * reads as a transcript, with a `[thinking]` / `[assistant]` marker each time
+ * the stream kind changes; structured events become one-line markers.
+ */
+function createEventStreamer(sink: BackgroundTaskSink): AgentEventListener {
+  let lastStream: TextStreamKind | undefined;
+  return (event) => {
+    if (isTextDeltaEvent(event)) {
+      const streamKind: TextStreamKind = event.type === 'thinking.delta' ? 'thinking' : 'assistant';
+      if (lastStream !== streamKind) {
+        sink.appendOutput(`\n[${streamKind}]\n`);
+        lastStream = streamKind;
+      }
+      sink.appendOutput(event.delta);
+      return;
+    }
+    const line = formatAgentEvent(event);
+    if (line !== undefined) {
+      lastStream = undefined;
+      sink.appendOutput(`\n${line}\n`);
+    }
+  };
+}
+
+function isTextDeltaEvent(
+  event: AgentEvent,
+): event is ThinkingDeltaEvent | AssistantDeltaEvent {
+  return event.type === 'thinking.delta' || event.type === 'assistant.delta';
+}
+
+function formatAgentEvent(event: AgentEvent): string | undefined {
+  switch (event.type) {
+    case 'turn.started':
+      return `[turn ${event.turnId} started]`;
+    case 'turn.ended':
+      return `[turn ${event.turnId} ended: ${event.reason}]${formatDetail(event.error?.message)}`;
+    case 'turn.step.retrying':
+      return `[retrying] step ${event.step} (attempt ${event.nextAttempt}/${event.maxAttempts}): ${event.errorMessage}`;
+    case 'turn.step.interrupted':
+      return `[interrupted] step ${event.step}: ${event.reason}${formatDetail(event.message)}`;
+    case 'tool.call.started':
+      return `[tool] ${event.name}${formatDetail(event.description ?? event.args)}`;
+    case 'tool.result':
+      return `[result${event.isError === true ? ' error' : ''}]${formatDetail(event.output)}`;
+    case 'error':
+      return `[error] ${event.message}`;
+    case 'warning':
+      return `[warning] ${event.message}`;
+    default:
+      return undefined;
+  }
+}
+
+function formatDetail(value: unknown): string {
+  if (value === undefined || value === null || value === '') return '';
+  const text = (typeof value === 'string' ? value : (JSON.stringify(value) ?? ''))
+    .replaceAll(/\s+/g, ' ')
+    .trim();
+  if (text.length === 0) return '';
+  const preview =
+    text.length > EVENT_PREVIEW_LENGTH ? `${text.slice(0, EVENT_PREVIEW_LENGTH)}…` : text;
+  return `: ${preview}`;
+}

packages/agent-core/src/agent/background/index.ts

@@ -145,8 +145,10 @@ const NOTIFICATION_FALLBACK_PREVIEW_BYTES = 3_000;
  * command (e.g. `b3sum --length <huge>`) whose output would otherwise grow
  * without bound — filling the disk, or retaining each pending-write chunk until
  * Node aborts with an out-of-memory crash. Scoped to process tasks (foreground
- * and background); subagent and user-question results are appended once and must
- * always be persisted, so they are intentionally not capped here.
+ * and background). Agent tasks stream the subagent's live event transcript and
+ * question tasks append their answer; both are unbounded but text-sized (and
+ * killing a subagent over log volume would be worse than the leak), so they
+ * are intentionally not capped here.
  */
 const MAX_TASK_OUTPUT_BYTES = 16 * 1024 * 1024; // 16 MiB
 
@@ -723,9 +725,10 @@ export class BackgroundManager {
     // live-forward buffer or the on-disk write chain until the process runs out
     // of memory or fills the disk. Trip once, then request graceful termination
     // through the shared stop path (SIGTERM → grace → SIGKILL). Scoped to
-    // process tasks (foreground and background): subagent and user-question tasks
-    // append their bounded result in one shot and must always persist it, so they
-    // are intentionally not capped here.
+    // process tasks (foreground and background): agent tasks stream a live
+    // event transcript and question tasks append an answer — unbounded but
+    // text-sized, and never worth killing the task over — so they are
+    // intentionally not capped here.
     if (
       !entry.outputLimitTripped &&
       entry.task.kind === 'process' &&

packages/agent-core/src/agent/index.ts

@@ -68,6 +68,9 @@ export * from './goal';
 
 export type AgentType = 'main' | 'sub' | 'independent';
 
+/** In-process listener for {@link Agent.onEvent}. */
+export type AgentEventListener = (event: AgentEvent) => void;
+
 export interface AgentOptions {
   readonly kaos: Kaos;
   readonly config?: KimiConfig;
@@ -663,8 +666,29 @@ export class Agent {
     };
   }
 
+  private readonly eventListeners = new Set<AgentEventListener>();
+
+  /**
+   * Subscribe to this agent's events in-process; returns an unsubscribe
+   * function. Listener failures are swallowed so event fan-out can never
+   * break the agent loop.
+   */
+  onEvent(listener: AgentEventListener): () => void {
+    this.eventListeners.add(listener);
+    return () => {
+      this.eventListeners.delete(listener);
+    };
+  }
+
   emitEvent(event: AgentEvent): void {
     if (this.records.restoring) return;
+    for (const listener of this.eventListeners) {
+      try {
+        listener(event);
+      } catch {
+        // Listener failures must not break the agent loop.
+      }
+    }
     void this.rpc?.emitEvent?.(event);
   }
 

packages/agent-core/src/session/subagent-host.ts

@@ -4,7 +4,7 @@ import {
   type TokenUsage,
 } from '@moonshot-ai/kosong';
 
-import type { Agent } from '../agent';
+import type { Agent, AgentEventListener } from '../agent';
 import type { PromptOrigin } from '../agent/context';
 import { ErrorCodes } from '../errors';
 import { DenyAllPermissionPolicy } from '../agent/permission/policies/deny-all';
@@ -128,11 +128,17 @@ type SubagentCompletion = {
   readonly usage?: TokenUsage;
 };
 
+export type SubagentEventListener = AgentEventListener;
+
+export type SubscribeToSubagentEvents = (callback: SubagentEventListener) => () => void;
+
 export type SubagentHandle = {
   readonly agentId: string;
   readonly profileName: string;
   readonly resumed: boolean;
   readonly completion: Promise<SubagentCompletion>;
+  /** Subscribe to the child agent's live events; the returned function unsubscribes. */
+  readonly subscribeToEvents?: SubscribeToSubagentEvents;
 };
 
 export class SessionSubagentHost {
@@ -173,6 +179,7 @@ export class SessionSubagentHost {
       profileName: profile.name,
       resumed: false,
       completion,
+      subscribeToEvents: agent.onEvent.bind(agent),
     };
   }
 
@@ -189,7 +196,13 @@ export class SessionSubagentHost {
         throw error;
       }
     });
-    return { agentId, profileName, resumed: true, completion };
+    return {
+      agentId,
+      profileName,
+      resumed: true,
+      completion,
+      subscribeToEvents: child.onEvent.bind(child),
+    };
   }
 
   async retry(agentId: string, options: RunSubagentOptions): Promise<SubagentHandle> {
@@ -211,7 +224,13 @@ export class SessionSubagentHost {
         throw error;
       }
     });
-    return { agentId, profileName, resumed: true, completion };
+    return {
+      agentId,
+      profileName,
+      resumed: true,
+      completion,
+      subscribeToEvents: child.onEvent.bind(child),
+    };
   }
 
   private async ensureIdleSubagent(

packages/agent-core/src/tools/builtin/collaboration/agent.ts

@@ -288,12 +288,10 @@ export class AgentTool implements BuiltinTool<AgentToolInput> {
   ): Promise<ExecutableToolResult> {
     const info = this.backgroundManager.getTask(taskId);
     if (info?.status === 'completed') {
-      return {
-        output: formatForegroundAgentSuccess(
-          handle,
-          await this.backgroundManager.readOutput(taskId),
-        ),
-      };
+      // The task output buffer carries the live transcript plus the final
+      // result; the parent agent only wants the result itself.
+      const { result } = await handle.completion;
+      return { output: formatForegroundAgentSuccess(handle, result) };
     }
     const timedOut = info?.status === 'timed_out';
     const message =