Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 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
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ describe('Claude provider task lifecycle', () => {
})).toEqual({ type: 'started', sessionId: 's1', taskId: 't1' });
expect(readClaudeProviderTaskActivity({
type: 'system', subtype: 'task_progress', session_id: 's1', task_id: 't1',
})).toEqual({ type: 'progress', sessionId: 's1', taskId: 't1' });
})).toEqual({ type: 'progress', sessionId: 's1', taskId: 't1', source: 'system-task-progress' });
for (const status of ['completed', 'failed', 'stopped']) {
expect(readClaudeProviderTaskActivity({
type: 'system', subtype: 'task_notification', session_id: 's1', task_id: 't1', status,
Expand Down Expand Up @@ -486,6 +486,14 @@ describe('readClaudeSessionHookProviderTaskActivity', () => {
});
});

/** Applies typed Claude SDK system rows from session `s1` to the ledger. */
function sdkRowApplier(ledger: ReturnType<typeof createClaudeProviderActivityLedger>) {
return (row: Record<string, unknown>) => {
const activity = readClaudeProviderTaskActivity({ type: 'system', session_id: 's1', ...row });
if (activity) ledger.apply(activity);
};
}

describe('createClaudeProviderActivityLedger', () => {
it('abandons only active tasks and keeps early terminal evidence', () => {
const ledger = createClaudeProviderActivityLedger();
Expand Down Expand Up @@ -578,6 +586,167 @@ describe('createClaudeProviderActivityLedger', () => {
}
});

it('keeps an agent that stopped with live background children resumable by its own progress', () => {
// Row shapes observed from Claude Code 2.1.302 stream-json: an Agent that ends its turn
// while its own backgrounded Bash runs reports `completed`, then wakes when the child settles.
const ledger = createClaudeProviderActivityLedger();
const apply = sdkRowApplier(ledger);
apply({ subtype: 'task_started', task_id: 'agent-1', task_type: 'local_agent', is_backgrounded: true });
apply({
subtype: 'task_started',
task_id: 'bash-1',
task_type: 'local_bash',
is_backgrounded: true,
owned_by_subagent: true,
parent_task_id: 'agent-1',
});
expect(ledger.getActiveProviderTaskCount()).toBe(2);

apply({ subtype: 'task_updated', task_id: 'agent-1', patch: { status: 'completed' } });
apply({ subtype: 'task_notification', task_id: 'agent-1', status: 'completed' });
expect(ledger.getActiveProviderTaskCount()).toBe(1);

const subagentStop = readClaudeSessionHookProviderTaskActivity({
hook_event_name: 'SubagentStop',
session_id: 's1',
agent_id: 'agent-1',
});
expect(subagentStop).not.toBeNull();
if (subagentStop) ledger.apply(subagentStop);
expect(ledger.getActiveProviderTaskCount()).toBe(1);

apply({ subtype: 'task_notification', task_id: 'bash-1', status: 'completed' });
// An Agent that never wakes holds nothing open once its children settle.
expect(ledger.getActiveProviderTaskCount()).toBe(0);
if (subagentStop) ledger.apply(subagentStop);
expect(ledger.getActiveProviderTaskCount()).toBe(0);

apply({ subtype: 'task_progress', task_id: 'agent-1' });
expect(ledger.hasActiveProviderTask({ sessionId: 's1', taskId: 'agent-1' })).toBe(true);
Comment thread
danljungstrom marked this conversation as resolved.

apply({ subtype: 'task_notification', task_id: 'agent-1', status: 'completed' });
expect(ledger.getActiveProviderTaskCount()).toBe(0);
apply({ subtype: 'task_progress', task_id: 'agent-1' });
expect(ledger.getActiveProviderTaskCount()).toBe(0);
});

it.each([
['failed', { subtype: 'task_notification', task_id: 'agent-1', status: 'failed' }],
['killed', { subtype: 'task_updated', task_id: 'agent-1', patch: { status: 'killed' } }],
])('does not let a %s Agent with live children wake on later progress', (_status, terminalRow) => {
const ledger = createClaudeProviderActivityLedger();
const apply = sdkRowApplier(ledger);
apply({ subtype: 'task_started', task_id: 'agent-1', task_type: 'local_agent', is_backgrounded: true });
apply({
subtype: 'task_started',
task_id: 'bash-1',
task_type: 'local_bash',
is_backgrounded: true,
owned_by_subagent: true,
parent_task_id: 'agent-1',
});
apply(terminalRow);
apply({ subtype: 'task_progress', task_id: 'agent-1' });
expect(ledger.hasActiveProviderTask({ sessionId: 's1', taskId: 'agent-1' })).toBe(false);
});

it.each([
['failed', { type: 'system', subtype: 'task_notification', task_id: 'agent-1', status: 'failed' }],
['stopped', { type: 'system', subtype: 'task_updated', task_id: 'agent-1', patch: { status: 'killed' } }],
['TaskStop', {
hook_event_name: 'PostToolUse', tool_name: 'TaskStop',
tool_input: { task_id: 'agent-1' }, tool_response: { task_id: 'agent-1', status: 'stopped' },
}],
['StopFailure', { hook_event_name: 'StopFailure', agent_id: 'agent-1' }],
])('retires an awaiting Agent after %s evidence', (_label, terminalRow) => {
const ledger = createClaudeProviderActivityLedger();
const apply = sdkRowApplier(ledger);
apply({ subtype: 'task_started', task_id: 'agent-1', task_type: 'local_agent', is_backgrounded: true });
apply({
subtype: 'task_started', task_id: 'bash-1', task_type: 'local_bash',
is_backgrounded: true, parent_task_id: 'agent-1',
});
apply({ subtype: 'task_notification', task_id: 'agent-1', status: 'completed' });
const row = { session_id: 's1', ...terminalRow };
const terminal = readClaudeProviderTaskActivity(row) ?? readClaudeSessionHookProviderTaskActivity(row);
if (!terminal) throw new Error('expected terminal evidence');
ledger.apply(terminal);
apply({ subtype: 'task_notification', task_id: 'bash-1', status: 'completed' });
apply({ subtype: 'task_progress', task_id: 'agent-1' });
expect(ledger.getActiveProviderTaskCount()).toBe(0);
});

it('keeps an agent resumable while a nested agent it owns still waits on or resumes its own work', () => {
const createNested = () => {
const ledger = createClaudeProviderActivityLedger();
const apply = sdkRowApplier(ledger);
apply({ subtype: 'task_started', task_id: 'agent-a', task_type: 'local_agent', is_backgrounded: true });
apply({
subtype: 'task_started',
task_id: 'agent-b',
task_type: 'local_agent',
is_backgrounded: true,
owned_by_subagent: true,
parent_task_id: 'agent-a',
});
apply({
subtype: 'task_started',
task_id: 'bash-b',
task_type: 'local_bash',
is_backgrounded: true,
owned_by_subagent: true,
parent_task_id: 'agent-b',
});
return { ledger, apply };
};
const stop = (apply: (row: Record<string, unknown>) => void, taskId: string) => {
apply({ subtype: 'task_updated', task_id: taskId, patch: { status: 'completed' } });
apply({ subtype: 'task_notification', task_id: taskId, status: 'completed' });
};
const isActive = (ledger: ReturnType<typeof createClaudeProviderActivityLedger>, taskId: string) =>
ledger.hasActiveProviderTask({ sessionId: 's1', taskId });

// A stops while B is itself stopped, waiting on its own shell.
const waiting = createNested();
stop(waiting.apply, 'agent-b');
stop(waiting.apply, 'agent-a');
waiting.apply({ subtype: 'task_notification', task_id: 'bash-b', status: 'completed' });
waiting.apply({ subtype: 'task_progress', task_id: 'agent-b' });
stop(waiting.apply, 'agent-b');
waiting.apply({ subtype: 'task_progress', task_id: 'agent-a' });
expect(isActive(waiting.ledger, 'agent-a')).toBe(true);

// A stops after B has woken from its own wait and is working again.
const resumed = createNested();
stop(resumed.apply, 'agent-b');
resumed.apply({ subtype: 'task_notification', task_id: 'bash-b', status: 'completed' });
resumed.apply({ subtype: 'task_progress', task_id: 'agent-b' });
stop(resumed.apply, 'agent-a');
stop(resumed.apply, 'agent-b');
resumed.apply({ subtype: 'task_progress', task_id: 'agent-a' });
expect(isActive(resumed.ledger, 'agent-a')).toBe(true);
});

it('keeps the parent link when the typed child launch follows an earlier hook launch', () => {
const ledger = createClaudeProviderActivityLedger();
const apply = sdkRowApplier(ledger);
apply({ subtype: 'task_started', task_id: 'agent-a', task_type: 'local_agent', is_backgrounded: true });
ledger.apply({ type: 'started', sessionId: 's1', taskId: 'agent-b', source: 'hook-agent-launch' });
apply({
subtype: 'task_started',
task_id: 'agent-b',
task_type: 'local_agent',
is_backgrounded: true,
owned_by_subagent: true,
parent_task_id: 'agent-a',
});

apply({ subtype: 'task_notification', task_id: 'agent-a', status: 'completed' });
apply({ subtype: 'task_notification', task_id: 'agent-b', status: 'completed' });
apply({ subtype: 'task_progress', task_id: 'agent-a' });
expect(ledger.hasActiveProviderTask({ sessionId: 's1', taskId: 'agent-a' })).toBe(true);
});

it('does not let progress admit an unknown task', () => {
const ledger = createClaudeProviderActivityLedger();
expect(ledger.apply({ type: 'progress', sessionId: 's1', taskId: 'unknown' })).toBe(false);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,14 @@ export type ClaudeProviderTaskActivity =
admission?: 'known-only';
source?: ClaudeProviderTaskActivitySource;
confirmsActivity?: false;
/** The task (an Agent) that owns this one, from the typed SDK row. */
parentTaskId?: string;
}>)
| (ClaudeProviderTaskIdentity & Readonly<{
type: 'progress';
confirmsActivity?: false;
source?: 'system-task-progress';
}>)
| (ClaudeProviderTaskIdentity & Readonly<{ type: 'progress'; confirmsActivity?: false }>)
| (ClaudeProviderTaskIdentity & Readonly<{
type: 'terminal';
terminalStatus?: 'completed' | 'failed' | 'stopped';
Expand Down Expand Up @@ -144,6 +150,7 @@ export function normalizeClaudeProviderTaskEvent(value: unknown, contextualSessi
normalizedString(row.task_type ?? row.taskType),
row.is_backgrounded === true,
);
const parentTaskId = normalizeClaudeAgentSdkProviderTaskId(row.parent_task_id);
return {
activity: {
type: 'started',
Expand All @@ -152,14 +159,15 @@ export function normalizeClaudeProviderTaskEvent(value: unknown, contextualSessi
...(admission === 'known-only'
? { admission: 'known-only' as const }
: {}),
...(parentTaskId ? { parentTaskId } : {}),
},
interruptTarget,
backgroundTask,
};
}
if (row.subtype === 'task_progress') {
return {
activity: { type: 'progress', sessionId, taskId },
activity: { type: 'progress', sessionId, taskId, source: 'system-task-progress' },
interruptTarget,
backgroundTask,
};
Expand Down Expand Up @@ -271,6 +279,10 @@ export function readClaudeSessionHookProviderTaskActivity(
return { type: 'terminal', sessionId, taskId: requestedTaskId };
}

function withParent(parentTaskId: string | undefined): { parentTaskId?: string } {
return parentTaskId ? { parentTaskId } : {};
}

function keyOf(identity: ClaudeProviderTaskIdentity): string {
return JSON.stringify([identity.sessionId, identity.taskId]);
}
Expand All @@ -282,11 +294,20 @@ export function createClaudeProviderActivityLedger() {
identity: ClaudeProviderTaskIdentity;
sources: Set<ClaudeProviderTaskActivitySource>;
observed: boolean;
parentTaskId?: string;
}>
| Readonly<{
phase: 'terminal-before-confirmation';
identity: ClaudeProviderTaskIdentity;
backgroundConfirmed: boolean;
/**
* Claude reports an Agent `completed` each time it stops, including when it stops to
* wait on background children it still owns; it wakes when they settle. Only the
* Agent's own typed progress may then reactivate it, so an Agent that never wakes
* cannot hold the count open.
*/
awaitingChildren?: true;
parentTaskId?: string;
}>;
const tasks = new Map<string, ProviderTaskRecord>();
/**
Expand All @@ -307,10 +328,23 @@ export function createClaudeProviderActivityLedger() {
ownedSessionIds.size > 0 && !ownedSessionIds.has(sessionId)
);

/** A child still owns live work while active or while it waits on its own children. */
const hasActiveChild = (identity: ClaudeProviderTaskIdentity): boolean => {
for (const entry of tasks.values()) {
if (
entry.parentTaskId === identity.taskId
&& entry.identity.sessionId === identity.sessionId
&& (entry.phase === 'active' || entry.awaitingChildren === true)
) return true;
}
return false;
};

const noteActive = (
identity: ClaudeProviderTaskIdentity,
source: ClaudeProviderTaskActivitySource,
confirmsActivity = true,
parentTaskId?: string,
): boolean => {
const key = keyOf(identity);
const current = tasks.get(key);
Expand All @@ -320,16 +354,33 @@ export function createClaudeProviderActivityLedger() {
if (!current.backgroundConfirmed) tasks.set(key, { ...current, backgroundConfirmed: true });
return false;
}
tasks.set(key, { phase: 'active', identity, sources: new Set([source]), observed: confirmsActivity });
tasks.set(key, {
phase: 'active',
identity,
sources: new Set([source]),
observed: confirmsActivity,
...withParent(current.parentTaskId),
});
return true;
}
if (current?.phase === 'active') {
current.sources.add(source);
if (current.observed || !confirmsActivity) return false;
tasks.set(key, { ...current, observed: true });
// Hook and typed launches race; only the typed row carries the owning Agent.
const parent = withParent(current.parentTaskId ?? parentTaskId);
if (current.observed || !confirmsActivity) {
if (parent.parentTaskId !== current.parentTaskId) tasks.set(key, { ...current, ...parent });
return false;
}
tasks.set(key, { ...current, ...parent, observed: true });
return true;
}
tasks.set(key, { phase: 'active', identity, sources: new Set([source]), observed: confirmsActivity });
tasks.set(key, {
phase: 'active',
identity,
sources: new Set([source]),
observed: confirmsActivity,
...withParent(parentTaskId),
});
return true;
};

Expand Down Expand Up @@ -357,13 +408,25 @@ export function createClaudeProviderActivityLedger() {
const current = tasks.get(key);
if (activity.type === 'terminal') {
if (current?.phase === 'active') {
const identity = { sessionId: activity.sessionId, taskId: activity.taskId };
// Only a completed stop can be interim; a failed or stopped Agent never wakes.
const awaitsChildren = activity.terminalStatus === 'completed' && hasActiveChild(identity);
tasks.set(key, {
phase: 'terminal-before-confirmation',
identity: { sessionId: activity.sessionId, taskId: activity.taskId },
identity,
backgroundConfirmed: true,
...(awaitsChildren ? { awaitingChildren: true as const } : {}),
...withParent(current.parentTaskId),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
});
return true;
}
if (current?.awaitingChildren === true && activity.terminalStatus !== 'completed') {
// A later stop/failure is final even when the Agent was already waiting.
// Duplicate completed rows still describe the same interim stop.
const { awaitingChildren: _settled, ...terminal } = current;
tasks.set(key, terminal);
return true;
Comment thread
happier-bot marked this conversation as resolved.
}
if (!current && (activity.rememberIfUnknown === true || activity.terminalStatus !== undefined)) {
tasks.set(key, {
phase: 'terminal-before-confirmation',
Expand All @@ -375,6 +438,21 @@ export function createClaudeProviderActivityLedger() {
}
if (activity.type === 'progress' || activity.admission === 'known-only') {
if (activity.type === 'progress' && activity.confirmsActivity === false) return false;
if (
activity.type === 'progress'
&& activity.source === 'system-task-progress'
&& current?.phase === 'terminal-before-confirmation'
&& current.awaitingChildren === true
) {
tasks.set(key, {
phase: 'active',
identity: current.identity,
sources: new Set(['system-task-progress']),
observed: true,
...withParent(current.parentTaskId),
});
return true;
}
if (activity.type === 'started' && current && activity.source === 'hook-agent-resume') {
if (current.phase === 'terminal-before-confirmation' && !current.backgroundConfirmed) return false;
return noteActive({ sessionId: activity.sessionId, taskId: activity.taskId }, activity.source, activity.confirmsActivity !== false);
Expand All @@ -386,6 +464,7 @@ export function createClaudeProviderActivityLedger() {
{ sessionId: activity.sessionId, taskId: activity.taskId },
activity.source ?? 'system-task-started',
activity.confirmsActivity !== false,
activity.parentTaskId,
);
},
/**
Expand Down
Loading
Loading