Skip to content
Merged
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 22 additions & 10 deletions packages/agent-core-v2/src/agent/tools/agent/agentTool.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@
* profiles mid-session, and a live read would rewrite the tools payload of
* every later request — breaking the provider's prompt cache for a change a
* live agent must not see (new profiles take effect on `/new` or `/reload`).
* The `subagent.spawned` signal is emitted only after the run's task
* registration succeeds, so it carries the task id (and a failed registration
* leaves no spawned row behind); the mirrored `subagent.started` is deferred
* until after it, keeping the spawned → started order consumers rely on.
* Bound at Agent scope.
*/

Expand Down Expand Up @@ -91,7 +95,8 @@ import type { Runtime } from '#/runtime/runtime';
import { ISessionMetadata } from '#/session/sessionMetadata/sessionMetadata';
import { ISessionWorkspaceContext } from '#/session/workspaceContext/workspaceContext';

import { emitAgentRunSpawned, mirrorAgentRun } from '#/session/subagent/mirrorAgentRun';
import { emitAgentRunSpawned, mirrorAgentRun, SubagentStarted } from '#/session/subagent/mirrorAgentRun';
import { IEventDispatcher } from '#/state/eventDispatcher';
Comment thread
wbxl2000 marked this conversation as resolved.
Resolved
import { ISessionSubagentService } from '#/session/subagent/subagent';
import {
buildSubagentModelDescriptions,
Expand Down Expand Up @@ -352,15 +357,6 @@ export class SubagentTool implements ISubagentTool {
});
}

const runInBackground = args.run_in_background === true;
emitAgentRunSpawned(requester, agentId, {
profileName,
parentToolCallId: toolCallId,
description: args.description,
runInBackground,
model: displayModel,
});

const run = await this.subagents.run(
agentId,
{ kind: 'prompt', prompt: promptText },
Expand All @@ -370,6 +366,7 @@ export class SubagentTool implements ISubagentTool {
profileName,
prompt: promptText,
signal: controller.signal,
deferStarted: true,
cancel: (reason) => {
controller.abort(reason);
},
Expand Down Expand Up @@ -493,6 +490,21 @@ export class SubagentTool implements ISubagentTool {
};
}

const requester = this.lifecycle.get(this.callerAgentId);
if (requester !== undefined) {
emitAgentRunSpawned(requester, handle.agentId, {
Comment thread
wbxl2000 marked this conversation as resolved.
Resolved
profileName: handle.profileName,
parentToolCallId: toolCallId,
description: args.description,
runInBackground,
model: handle.model,
taskId,
});
void requester.accessor
.get(IEventDispatcher)
?.dispatch(new SubagentStarted({ subagentId: handle.agentId }));
}

if (runInBackground) {
return {
output: formatBackgroundAgentResult(taskId, handle, args.description, allowBackground),
Expand Down
15 changes: 13 additions & 2 deletions packages/agent-core-v2/src/session/subagent/mirrorAgentRun.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,12 @@
* session recordings and dashboards stay valid. The spawned signal also
* reports the child's bound model alias and its effective thinking effort, so
* clients can render both at spawn instead of waiting for the first
* `agent.status.updated` frame.
* `agent.status.updated` frame. A caller that registers the run as a task
* (the `Agent` tool) emits spawned only after registration, so the signal
* also carries the task id clients bind cancel/status actions to; such a
* caller passes `deferStarted` and dispatches `subagent.started` itself after
* spawned, because consumers drop started until spawned has established the
* row (and a failed registration must leave no started row behind).
*/

/* oxlint-disable typescript-eslint/no-unsafe-declaration-merging, eslint-plugin-import/namespace -- Event2 class+payload-interface declaration merging is the sanctioned event-declaration idiom. */
Expand Down Expand Up @@ -49,6 +54,7 @@ export interface SubagentSpawnedPayload {
readonly runInBackground: boolean;
readonly model?: string;
readonly thinkingEffort?: string;
readonly taskId?: string;
}

export class SubagentSpawned extends Event2<SubagentSpawnedPayload> {
Expand Down Expand Up @@ -99,6 +105,7 @@ export interface AgentRunSpawnedMeta {
readonly swarmIndex?: number;
readonly runInBackground?: boolean;
readonly model?: string;
readonly taskId?: string;
}

export interface MirrorAgentRunOptions {
Expand All @@ -107,6 +114,7 @@ export interface MirrorAgentRunOptions {
readonly suppressRateLimitFailureEvent?: boolean;
readonly signal: AbortSignal;
readonly cancel?: (reason?: unknown) => void;
readonly deferStarted?: boolean;
}

export function emitAgentRunSpawned(
Expand All @@ -131,6 +139,7 @@ export function emitAgentRunSpawned(
runInBackground: meta.runInBackground ?? false,
model: meta.model,
thinkingEffort: childProfile?.getEffectiveThinkingLevel(),
taskId: meta.taskId,
}),
);
childProfile?.republishStatus();
Expand All @@ -151,7 +160,9 @@ export async function mirrorAgentRun(
const dispatcher = requester.accessor.get(IEventDispatcher);
const subagents = requester.accessor.get(ISessionSubagentService);
const agentLifecycle = requester.accessor.get(IAgentLifecycleService);
void dispatcher?.dispatch(new SubagentStarted({ subagentId: run.agentId }));
if (options.deferStarted !== true) {
void dispatcher?.dispatch(new SubagentStarted({ subagentId: run.agentId }));
}
if (options.prompt !== undefined) {
const cancelAndRethrow = (reason: unknown): never => {
options.cancel?.(reason);
Expand Down
35 changes: 35 additions & 0 deletions packages/agent-core-v2/test/tool/tool.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1858,6 +1858,36 @@ describe('Agent tool execution contract', () => {
completion.resolve({ summary: 'finished later' });
});

it('emits spawned with the registered task id ahead of started', async () => {
const completion = deferred<{ readonly summary: string }>();
const lifecycle = createAgentLifecycleStub({
createAgentIds: ['agent-child'],
runCompletion: () => completion.promise,
});
const context = createAgentToolContext(lifecycle);

const result = await executeAgentTool(context, {
prompt: 'Investigate',
description: 'Find cause',
run_in_background: true,
});

if (typeof result.output !== 'string') throw new TypeError('expected string output');
const taskId = result.output.match(/task_id: (agent-[0-9a-z]{8})/)?.[1];
expect(taskId).toBeDefined();
expect(lifecycle.publishedEvents).toContainEqual(
expect.objectContaining({
type: 'subagent.spawned',
subagentId: 'agent-child',
taskId,
}),
);
const eventOrder = lifecycle.publishedEvents.map((event) => event.type);
expect(eventOrder.indexOf('subagent.spawned')).toBeGreaterThanOrEqual(0);
expect(eventOrder.indexOf('subagent.started')).toBeGreaterThan(eventOrder.indexOf('subagent.spawned'));
Comment thread
wbxl2000 marked this conversation as resolved.
Resolved
completion.resolve({ summary: 'finished later' });
});

it('rejects background subagents when background execution is disabled', async () => {
const lifecycle = createAgentLifecycleStub();
const context = createAgentToolContext(lifecycle);
Expand Down Expand Up @@ -1961,6 +1991,11 @@ describe('Agent tool execution contract', () => {
output: 'Too many background tasks are already running.',
});
expect(lifecycle.create).toHaveBeenCalledTimes(2);
expect(
lifecycle.publishedEvents.filter(
(event) => (event as { subagentId?: string }).subagentId === 'agent-second',
),
).toEqual([]);
completions[0]?.resolve({ summary: 'finished later' });
});

Expand Down
1 change: 1 addition & 0 deletions packages/kap-server/src/protocol/events-zod.ts
Original file line number Diff line number Diff line change
Expand Up @@ -851,6 +851,7 @@ export const subagentSpawnedEventSchema = z.object({
runInBackground: z.boolean(),
model: z.string().optional(),
thinkingEffort: z.string().optional(),
taskId: z.string().optional(),
Comment thread
wbxl2000 marked this conversation as resolved.
Resolved
}) satisfies z.ZodType<SubagentSpawnedPayload>;

export const subagentStartedEventSchema = z.object({
Expand Down
38 changes: 34 additions & 4 deletions packages/kap-server/src/services/transcript/coreEventMap.ts
Original file line number Diff line number Diff line change
Expand Up @@ -289,6 +289,10 @@ export class AgentTranscriptProjector {
private readonly tasks = new Map<string, TranscriptTask>();
/** shell `commandId` → transcript `taskId` (`shell.output` is keyed by command id only). */
private readonly shellTasks = new Map<string, string>();
/** subagent agent id → registered task id, for Agent-tool runs whose spawned
carried the registration (`taskId`): the task row keys by the task id so
`/tasks/{id}` actions resolve, and lifecycle events fold back to it. */
private readonly subagentTaskIds = new Map<string, string>();
/** interaction id → the pending entity as last emitted (resolve spreads it). */
private readonly interactions = new Map<string, TranscriptInteraction>();
/** promptId → the prompt queue entity as last emitted (`prompt.upsert` replaces). */
Expand Down Expand Up @@ -982,6 +986,13 @@ export class AgentTranscriptProjector {
outputTail: prev?.outputTail ?? '',
startedAt: prev?.startedAt ?? epochMsToIso(info.startedAt),
endedAt: info.endedAt === null ? prev?.endedAt : epochMsToIso(info.endedAt),
// task.upsert replaces the row wholesale: a subagent.completed/failed
// folded into this row earlier (background Agent run, see
// onSubagentSpawned) would otherwise lose its result details here.
resultSummary: prev?.resultSummary,
usage: prev?.usage,
error: prev?.error,
stateReason: prev?.stateReason,
}));
const ops: TranscriptOperation[] = [{ op: 'task.upsert', task }];
if (event.type === 'task.started') {
Expand Down Expand Up @@ -1127,9 +1138,24 @@ export class AgentTranscriptProjector {
description?: string;
swarmIndex?: number;
runInBackground: boolean;
taskId?: string;
}): TranscriptOperation[] {
const task = this.upsertTask(event.subagentId, (prev) => ({
taskId: event.subagentId,
// An Agent-tool spawned carries the run's task registration: key the row
// by that task id (not the agent id) so `/tasks/{id}` cancel/status
// actions resolve, and fold the later `task.started` upsert into it
// instead of surfacing a second row.
const taskKey = event.taskId ?? event.subagentId;
if (event.taskId !== undefined) {
this.subagentTaskIds.set(event.subagentId, event.taskId);
} else {
// A taskless (re)spawn — e.g. the same child resumed through AgentSwarm —
// must not route its lifecycle into a stale Agent-task row from a
// previous registration: drop the old mapping so this run keys by agent
// id end to end.
this.subagentTaskIds.delete(event.subagentId);
}
Comment thread
wbxl2000 marked this conversation as resolved.
Resolved
const task = this.upsertTask(taskKey, (prev) => ({
Comment thread
wbxl2000 marked this conversation as resolved.
Resolved
taskId: taskKey,
kind: 'subagent',
state: 'running',
// `runInBackground` subagents are detached from birth; foreground runs
Expand Down Expand Up @@ -1183,8 +1209,12 @@ export class AgentTranscriptProjector {
: event.type === 'subagent.failed'
? 'failed'
: 'running';
const task = this.upsertTask(event.subagentId, (prev) => ({
taskId: event.subagentId,
// Fold into the task-id-keyed row when the spawn registered one (see
// onSubagentSpawned) — the agent-id-keyed fallback is for spawns without
// a task registration (swarm/session-init/tower).
const taskKey = this.subagentTaskIds.get(event.subagentId) ?? event.subagentId;
Comment thread
wbxl2000 marked this conversation as resolved.
Resolved
const task = this.upsertTask(taskKey, (prev) => ({
taskId: taskKey,
kind: 'subagent',
state,
detached: prev?.detached ?? true,
Expand Down
92 changes: 92 additions & 0 deletions packages/kap-server/test/services/transcript.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1014,6 +1014,98 @@ describe('AgentTranscriptProjector', () => {
});
});

it('keys an Agent-tool subagent row by its registered task id and folds the lifecycle', () => {
const projector = new AgentTranscriptProjector('main');
const tx = new AgentTranscript('main');
const feed = (event: ProjectorBusEvent): void => void tx.apply(projector.map(event));

feed(
ev({
type: 'subagent.spawned',
subagentId: 'agent-1',
subagentName: 'explore',
parentToolCallId: 'call-1',
description: 'Inspect files',
runInBackground: true,
taskId: 'task-9',
}),
);
feed(
ev({
type: 'task.started',
info: {
taskId: 'task-9',
kind: 'agent',
description: 'Inspect files',
status: 'running',
detached: true,
agentId: 'agent-1',
startedAt: 1_700_000_000_000,
endedAt: null,
},
}),
);
feed(ev({ type: 'subagent.completed', subagentId: 'agent-1', resultSummary: 'done' }));
feed(
ev({
type: 'task.terminated',
info: {
taskId: 'task-9',
kind: 'agent',
description: 'Inspect files',
status: 'completed',
detached: true,
agentId: 'agent-1',
startedAt: 1_700_000_000_000,
endedAt: 1_700_000_001_000,
},
}),
);

expect(tx.getTask('task-9')).toMatchObject({
kind: 'subagent',
state: 'completed',
agentId: 'agent-1',
description: 'Inspect files',
detached: true,
resultSummary: 'done',
});
expect(tx.getTask('agent-1')).toBeUndefined();
});

it('drops the stale task mapping when a child respawns without a task id', () => {
const projector = new AgentTranscriptProjector('main');
const tx = new AgentTranscript('main');
const feed = (event: ProjectorBusEvent): void => void tx.apply(projector.map(event));

feed(
ev({
type: 'subagent.spawned',
subagentId: 'agent-1',
subagentName: 'explore',
parentToolCallId: 'call-1',
description: 'Inspect files',
runInBackground: true,
taskId: 'task-9',
}),
);
feed(ev({ type: 'subagent.completed', subagentId: 'agent-1', resultSummary: 'done' }));
feed(
ev({
type: 'subagent.spawned',
subagentId: 'agent-1',
subagentName: 'worker',
parentToolCallId: 'call-2',
description: 'scan again',
runInBackground: false,
}),
);
feed(ev({ type: 'subagent.started', subagentId: 'agent-1' }));

expect(tx.getTask('task-9')).toMatchObject({ state: 'completed', resultSummary: 'done' });
expect(tx.getTask('agent-1')).toMatchObject({ kind: 'subagent', state: 'running' });
});

it('projects goal updates into meta.goal plus an inline marker', () => {
const projector = new AgentTranscriptProjector('main');
const tx = new AgentTranscript('main');
Expand Down
6 changes: 6 additions & 0 deletions packages/protocol/src/events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -848,6 +848,11 @@ export interface SubagentSpawnedEvent {
/** The child's effective thinking effort at spawn (same vocabulary as
* `agent.status.updated`). Optional for cross-version tolerance. */
readonly thinkingEffort?: string;
/** Background-task id the run registered under in the caller's task store.
* Emitted after task registration, so cancel/status actions can bind to
* the task store without waiting for `task.started`. Optional for
* cross-version tolerance (older producers never send it). */
readonly taskId?: string;
}

export interface SubagentStartedEvent {
Expand Down Expand Up @@ -1779,6 +1784,7 @@ export const subagentSpawnedEventSchema = z.object({
runInBackground: z.boolean(),
model: z.string().optional(),
thinkingEffort: z.string().optional(),
taskId: z.string().optional(),
}) satisfies z.ZodType<SubagentSpawnedEvent>;

export const subagentStartedEventSchema = z.object({
Expand Down
Loading