Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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
53 changes: 50 additions & 3 deletions packages/agent-core/src/agent/turn/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -604,7 +604,17 @@ export class TurnFlow {
this.agent.emitEvent(errorEvent);
}
if (ended.reason !== 'completed') {
this.trackTurnInterrupted(turnId, this.currentStepByTurn.get(turnId) ?? this.currentStep);
// Fallback for turns that end abnormally without a `turn.interrupted`
// loop event reaching `trackLoopTelemetry` (e.g. a user-prompt hook block
// or an abort that bypasses the step loop). `ended.reason` maps onto the
// same interrupt-reason taxonomy the loop-event path uses; for a
// `cancelled` end the signal's reason decides user_cancelled vs aborted.
const interruptReason = telemetryInterruptReason(ended.reason, isUserCancellation(signal.reason));
this.trackTurnInterrupted(
turnId,
this.currentStepByTurn.get(turnId) ?? this.currentStep,
interruptReason,
);
}
this.telemetryModeByTurn.delete(turnId);
this.currentStepByTurn.delete(turnId);
Expand Down Expand Up @@ -938,7 +948,11 @@ export class TurnFlow {
if (event.reason === 'error' && event.activeStep !== undefined) {
this.stepFailureByTurn.set(turnId, event);
}
this.trackTurnInterrupted(turnId, interruptedStep(event));
this.trackTurnInterrupted(
turnId,
interruptedStep(event),
event.interruptReason ?? telemetryInterruptReason(event.reason, false),
);
return;
}
this.trackToolLifecycle(event, turnId);
Expand Down Expand Up @@ -1024,12 +1038,17 @@ export class TurnFlow {
return false;
}

private trackTurnInterrupted(turnId: number, atStep: number): void {
private trackTurnInterrupted(
turnId: number,
atStep: number,
interruptReason: TelemetryInterruptReason,
): void {
if (this.interruptedTelemetryTurnIds.has(turnId)) return;
this.interruptedTelemetryTurnIds.add(turnId);
this.agent.telemetry.track('turn_interrupted', {
mode: this.telemetryModeByTurn.get(turnId) ?? this.telemetryMode(),
at_step: atStep,
interrupt_reason: interruptReason,
...this.requestProtocolProps(),
});
}
Expand Down Expand Up @@ -1232,6 +1251,34 @@ function interruptedStep(event: LoopTurnInterruptedEvent): number {
return event.activeStep ?? event.attemptedSteps;
}

/**
* Telemetry-facing interrupt reason. The loop reports `LoopInterruptReason`
* (`aborted` | `max_steps` | `error`); we split `aborted` into a deliberate
* user cancel vs. any other programmatic abort so telemetry can tell them
* apart. `filtered` is folded in for the fallback path (turn ends flagged
* `filtered` never emit a `turn.interrupted` loop event).
*/
type TelemetryInterruptReason =
| 'user_cancelled'
| 'aborted'
| 'max_steps'
| 'error'
| 'filtered';

function telemetryInterruptReason(
reason: LoopTurnInterruptedEvent['reason'] | Exclude<TurnEndedEvent['reason'], 'completed'>,
userCancelled: boolean,
): TelemetryInterruptReason {
if ((reason === 'aborted' || reason === 'cancelled') && userCancelled) {
return 'user_cancelled';
}
if (reason === 'aborted' || reason === 'cancelled') return 'aborted';
if (reason === 'failed') return 'error';
// Remaining values are `max_steps` | `error` | `filtered`, which match the
// telemetry enum.
return reason;
}

interface ApiErrorClassification {
readonly errorType: string;
readonly statusCode?: number;
Expand Down
6 changes: 6 additions & 0 deletions packages/agent-core/src/loop/events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import type { ToolInputDisplay } from '../tools/display';
import type { ExecutableToolResult, LoopStepStopReason, ToolUpdate } from './types';

export type LoopInterruptReason = 'aborted' | 'max_steps' | 'error';
export type LoopInterruptCause = LoopInterruptReason | 'user_cancelled';

export interface LoopStepBeginEvent {
readonly type: 'step.begin';
Expand Down Expand Up @@ -94,6 +95,11 @@ export interface LoopTurnInterruptedEvent {
readonly attemptedSteps: number;
readonly activeStep?: number | undefined;
readonly message?: string | undefined;
/**
* Telemetry-facing interrupt cause. `aborted` is split into a deliberate user
* cancel vs. any other abort; `max_steps`/`error` mirror `reason`.
*/
readonly interruptReason?: LoopInterruptCause | undefined;
}

export interface LoopTextDeltaEvent {
Expand Down
10 changes: 9 additions & 1 deletion packages/agent-core/src/loop/run-turn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import { addUsage, emptyUsage, type TokenUsage } from '@moonshot-ai/kosong';

import type { Logger } from '#/logging/types';

import { isUserCancellation } from '../utils/abort';
import {
createMaxStepsExceededError,
errorMessage,
Expand Down Expand Up @@ -148,7 +149,12 @@ export async function runTurn(input: RunTurnInput): Promise<TurnResult> {
}
} catch (error) {
if (isAbortError(error) || signal.aborted) {
dispatchEvent(makeInterruptedEvent('aborted', steps, activeStep));
// A deliberate user cancel travels as the signal's reason (and may be the
// thrown error itself). Report it distinctly from a timeout or other
// programmatic abort so telemetry can tell the two apart.
const interruptReason =
isUserCancellation(signal.reason) || isUserCancellation(error) ? 'user_cancelled' : 'aborted';
dispatchEvent(makeInterruptedEvent('aborted', steps, activeStep, undefined, interruptReason));
return { stopReason: 'aborted', steps, usage };
}
const reason: LoopInterruptReason = isMaxStepsExceededError(error) ? 'max_steps' : 'error';
Expand All @@ -164,12 +170,14 @@ function makeInterruptedEvent(
attemptedSteps: number,
activeStep: number | undefined,
message?: string | undefined,
interruptReason: LoopTurnInterruptedEvent['interruptReason'] = reason,
): LoopTurnInterruptedEvent {
return {
type: 'turn.interrupted',
reason,
attemptedSteps,
...(activeStep !== undefined ? { activeStep } : {}),
...(message !== undefined ? { message } : {}),
interruptReason,
};
}
54 changes: 53 additions & 1 deletion packages/agent-core/test/agent/turn.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,59 @@ describe('Agent turn flow', () => {
});
expect(records).toContainEqual({
event: 'turn_interrupted',
properties: { mode: 'agent', at_step: 0 },
properties: { mode: 'agent', at_step: 0, interrupt_reason: 'error' },
});
});

it('reports turn_interrupted telemetry as user_cancelled on manual abort', async () => {
const records: TelemetryRecord[] = [];
const ctx = testAgent({
kaos: createCommandKaos('should-not-run'),
telemetry: recordingTelemetry(records),
});
ctx.configure({ tools: ['Bash'] });

ctx.mockNextResponse({ type: 'text', text: 'I will run Bash.' }, bashCall());
await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Run a command' }] });
await ctx.untilApprovalRequest();

// User presses stop: the RPC cancel carries no explicit reason, which the
// turn treats as a deliberate user cancellation.
await ctx.rpc.cancel({ turnId: 0 });
await ctx.untilTurnEnd();

const interrupted = records.find((candidate) => candidate.event === 'turn_interrupted');
expect(interrupted).toEqual({
event: 'turn_interrupted',
properties: expect.objectContaining({
mode: 'agent',
interrupt_reason: 'user_cancelled',
}),
});
});

it('reports turn_interrupted telemetry as aborted on programmatic abort', async () => {
const records: TelemetryRecord[] = [];
const ctx = testAgent({
kaos: createCommandKaos('should-not-run'),
telemetry: recordingTelemetry(records),
});
ctx.configure({ tools: ['Bash'] });

ctx.mockNextResponse({ type: 'text', text: 'I will run Bash.' }, bashCall());
await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Run a command' }] });
await ctx.untilApprovalRequest();

// A programmatic abort (e.g. a subagent deadline timeout) carries a plain
// AbortError as its reason, not a UserCancellationError, so telemetry must
// not report it as a user cancellation.
ctx.agent.turn.cancel(0, abortError());
await ctx.untilTurnEnd();

const interrupted = records.find((candidate) => candidate.event === 'turn_interrupted');
expect(interrupted).toEqual({
event: 'turn_interrupted',
properties: expect.objectContaining({ mode: 'agent', interrupt_reason: 'aborted' }),
});
});

Expand Down
Loading