mirror of
https://github.com/MoonshotAI/kimi-code.git
synced 2026-08-20 06:05:36 +00:00
fix: dedupe turn
This commit is contained in:
_Kerman
parent
886faeb4b9
commit
3cd3edb47c
7 changed files with 360 additions and 536 deletions
|
|
@ -1,9 +1,6 @@
|
|||
/**
|
||||
* `turn` domain barrel — re-exports the turn contract (`turn`) and its scoped
|
||||
* services (`turnService`, `loopRunner`). Importing this barrel registers the
|
||||
* `ITurnService` and `ILoopRunner` bindings into the scope registry.
|
||||
* `turnRunner` domain barrel - re-exports the turnRunner service contract and implementation.
|
||||
*/
|
||||
|
||||
export * from './turn';
|
||||
export * from './turnService';
|
||||
export * from './loopRunner';
|
||||
|
|
|
|||
|
|
@ -1,19 +0,0 @@
|
|||
/**
|
||||
* `turn` domain (L4) — `ILoopRunner` implementation.
|
||||
*
|
||||
* Runs the per-turn loop. Bound at Turn scope.
|
||||
*/
|
||||
|
||||
import { InstantiationType } from '#/_base/di/extensions';
|
||||
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
|
||||
|
||||
import { ILoopRunner } from './turn';
|
||||
|
||||
export class LoopRunner implements ILoopRunner {
|
||||
declare readonly _serviceBrand: undefined;
|
||||
run(): Promise<void> {
|
||||
return Promise.resolve();
|
||||
}
|
||||
}
|
||||
|
||||
registerScopedService(LifecycleScope.Turn, ILoopRunner, LoopRunner, InstantiationType.Delayed, 'turn');
|
||||
|
|
@ -1,61 +1,51 @@
|
|||
/**
|
||||
* `turn` domain (L4) — drives the turn lifecycle.
|
||||
*
|
||||
* Defines the public contract of a turn: the `ITurnService` used by upper layers
|
||||
* to start, steer, retry, and cancel a turn and to observe its events, the
|
||||
* per-turn `ITurnContext`, and the `ILoopRunner` that runs the turn loop.
|
||||
* `ITurnService` is Agent-scoped; `ILoopRunner` is Turn-scoped.
|
||||
*/
|
||||
import { createDecorator } from "#/_base/di";
|
||||
import type { ContextMessage, PromptOrigin } from '#/context';
|
||||
|
||||
import type { Event } from '#/_base/event';
|
||||
import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation';
|
||||
import type { Hooks } from '#/hooks';
|
||||
|
||||
export interface TurnStartEvent {
|
||||
readonly turnId: string;
|
||||
}
|
||||
export interface TurnToolEvent {
|
||||
readonly turnId: string;
|
||||
readonly toolCallId: string;
|
||||
readonly toolName: string;
|
||||
}
|
||||
export interface TurnStepEvent {
|
||||
readonly turnId: string;
|
||||
readonly step: number;
|
||||
}
|
||||
export interface TurnEndEvent {
|
||||
readonly turnId: string;
|
||||
readonly reason: string;
|
||||
|
||||
export interface TurnResult {
|
||||
readonly reason: 'completed' | 'cancelled' | 'failed' | 'filtered';
|
||||
readonly error?: unknown;
|
||||
}
|
||||
|
||||
export interface ITurnService {
|
||||
readonly _serviceBrand: undefined;
|
||||
readonly onWillStartTurn: Event<TurnStartEvent>;
|
||||
readonly onWillExecuteTool: Event<TurnToolEvent>;
|
||||
readonly onDidFinalizeTool: Event<TurnToolEvent>;
|
||||
readonly onDidEndStep: Event<TurnStepEvent>;
|
||||
readonly onDidEndTurn: Event<TurnEndEvent>;
|
||||
readonly hasActiveTurn: boolean;
|
||||
readonly currentId: string | undefined;
|
||||
prompt(input: string): Promise<void>;
|
||||
steer(content: string, origin?: string): void;
|
||||
retry(): Promise<void>;
|
||||
cancel(reason?: string): void;
|
||||
export interface Turn {
|
||||
readonly id: number;
|
||||
readonly abortController: AbortController;
|
||||
readonly ready: Promise<void>;
|
||||
readonly result: Promise<TurnResult>;
|
||||
}
|
||||
|
||||
export const ITurnService: ServiceIdentifier<ITurnService> =
|
||||
createDecorator<ITurnService>('turnService');
|
||||
|
||||
export interface ITurnContext {
|
||||
readonly turnId: string;
|
||||
export interface TurnStepContext {
|
||||
readonly turn: Turn;
|
||||
continueTurn: boolean;
|
||||
}
|
||||
|
||||
export const ITurnContext: ServiceIdentifier<ITurnContext> =
|
||||
createDecorator<ITurnContext>('turnContext');
|
||||
|
||||
export interface ILoopRunner {
|
||||
readonly _serviceBrand: undefined;
|
||||
run(): Promise<void>;
|
||||
export interface TurnRunContext {
|
||||
readonly turn: Turn;
|
||||
readonly origin: PromptOrigin;
|
||||
readonly promptMessage?: ContextMessage;
|
||||
result?: TurnResult;
|
||||
}
|
||||
|
||||
export const ILoopRunner: ServiceIdentifier<ILoopRunner> =
|
||||
createDecorator<ILoopRunner>('loopRunner');
|
||||
export interface TurnEndedContext {
|
||||
readonly turn: Turn;
|
||||
readonly result: TurnResult;
|
||||
}
|
||||
|
||||
|
||||
export interface ITurnRunner {
|
||||
launch(origin: PromptOrigin): Turn;
|
||||
getActiveTurn(): Turn | undefined;
|
||||
cancel(turnId?: number, reason?: unknown): void;
|
||||
|
||||
readonly hooks: Hooks<{
|
||||
onLaunched: { turn: Turn };
|
||||
onEnded: TurnEndedContext;
|
||||
beforeStep: TurnStepContext;
|
||||
afterStep: TurnStepContext;
|
||||
}>;
|
||||
}
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-redeclare
|
||||
export const ITurnRunner = createDecorator<ITurnRunner>('agentTurnRunnerService');
|
||||
|
|
|
|||
|
|
@ -1,117 +1,324 @@
|
|||
/**
|
||||
* `turn` domain (L4) — `ITurnService` implementation.
|
||||
*
|
||||
* Drives the turn lifecycle and emits its events; runs the turn loop through
|
||||
* `loopRunner`, drives agent lifecycle through `agent-lifecycle`, reads
|
||||
* history through `context`, enqueues follow-up through `injection`, drives
|
||||
* LLM generation through `kosong`, logs through `log`, checks permissions
|
||||
* through `permission`, reports telemetry through `telemetry`, executes tools
|
||||
* through `tool`, and checks usage through `usage`. Bound at Agent scope.
|
||||
*/
|
||||
|
||||
import { Disposable } from '#/_base/di/lifecycle';
|
||||
import { Emitter, type Event } from '#/_base/event';
|
||||
import { InstantiationType } from '#/_base/di/extensions';
|
||||
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
|
||||
import { IAgentLifecycleService } from '#/agent-lifecycle/agentLifecycle';
|
||||
import { IContextService } from '#/context/context';
|
||||
import { IInjectionService } from '#/injection/injection';
|
||||
import { ILLMService } from '#/kosong/kosong';
|
||||
import { ILogService } from '#/log/log';
|
||||
import { IPermissionService } from '#/permission/permission';
|
||||
import { ITelemetryService } from '#/telemetry/telemetry';
|
||||
import { IToolService } from '#/tool/tool';
|
||||
import { IUsageService } from '#/usage/usage';
|
||||
|
||||
import {
|
||||
type TurnEndEvent,
|
||||
type TurnStartEvent,
|
||||
type TurnStepEvent,
|
||||
type TurnToolEvent,
|
||||
ILoopRunner,
|
||||
ITurnService,
|
||||
IInstantiationService,
|
||||
registerSingleton,
|
||||
SyncDescriptor,
|
||||
} from "#/_base/di";
|
||||
import { toKimiErrorPayload, type KimiErrorPayload } from "#/_base/errors";
|
||||
import { isUserCancellation, userCancellationReason } from "#/_base/utils/abort";
|
||||
import type { ContextMessage, PromptOrigin } from '#/context';
|
||||
import { USER_PROMPT_ORIGIN } from '#/context';
|
||||
import { IContextMemory } from '#/contextMemory/contextMemory';
|
||||
import { IEventBus } from '#/eventBus/eventBus';
|
||||
import { IExternalHooksService } from '#/externalHooks/externalHooks';
|
||||
import { OrderedHookSlot } from '#/hooks';
|
||||
import { ILoopService } from '#/loop/loop';
|
||||
import { IPlanService } from '#/plan/plan';
|
||||
import { ITelemetryService } from '#/telemetry/telemetry';
|
||||
import { IUsageService } from '#/usage/usage';
|
||||
import { IWireRecord } from '#/wireRecord/wireRecord';
|
||||
import type {
|
||||
Turn,
|
||||
TurnEndedContext,
|
||||
TurnResult,
|
||||
TurnStepContext,
|
||||
} from './turn';
|
||||
import { ITurnRunner } from './turn';
|
||||
|
||||
let nextTurnId = 0;
|
||||
|
||||
export class TurnService extends Disposable implements ITurnService {
|
||||
declare readonly _serviceBrand: undefined;
|
||||
|
||||
private readonly _onWillStartTurn = this._register(new Emitter<TurnStartEvent>());
|
||||
readonly onWillStartTurn: Event<TurnStartEvent> = this._onWillStartTurn.event;
|
||||
private readonly _onWillExecuteTool = this._register(new Emitter<TurnToolEvent>());
|
||||
readonly onWillExecuteTool: Event<TurnToolEvent> = this._onWillExecuteTool.event;
|
||||
private readonly _onDidFinalizeTool = this._register(new Emitter<TurnToolEvent>());
|
||||
readonly onDidFinalizeTool: Event<TurnToolEvent> = this._onDidFinalizeTool.event;
|
||||
private readonly _onDidEndStep = this._register(new Emitter<TurnStepEvent>());
|
||||
readonly onDidEndStep: Event<TurnStepEvent> = this._onDidEndStep.event;
|
||||
private readonly _onDidEndTurn = this._register(new Emitter<TurnEndEvent>());
|
||||
readonly onDidEndTurn: Event<TurnEndEvent> = this._onDidEndTurn.event;
|
||||
|
||||
private active: { readonly turnId: string; cancelled: boolean } | undefined;
|
||||
private readonly steerBuffer: { content: string; origin?: string }[] = [];
|
||||
|
||||
constructor(
|
||||
@IContextService _context: IContextService,
|
||||
@IToolService _tool: IToolService,
|
||||
@IPermissionService _permission: IPermissionService,
|
||||
@ILLMService _llm: ILLMService,
|
||||
@IInjectionService _injection: IInjectionService,
|
||||
@IUsageService _usage: IUsageService,
|
||||
@ITelemetryService _telemetry: ITelemetryService,
|
||||
@ILogService _log: ILogService,
|
||||
@IAgentLifecycleService _agentLifecycle: IAgentLifecycleService,
|
||||
@ILoopRunner private readonly loopRunner: ILoopRunner,
|
||||
) {
|
||||
super();
|
||||
}
|
||||
|
||||
get hasActiveTurn(): boolean {
|
||||
return this.active !== undefined;
|
||||
}
|
||||
get currentId(): string | undefined {
|
||||
return this.active?.turnId;
|
||||
}
|
||||
|
||||
async prompt(input: string): Promise<void> {
|
||||
if (this.active !== undefined) {
|
||||
this.steer(input);
|
||||
return;
|
||||
}
|
||||
await this.launch(input);
|
||||
}
|
||||
|
||||
steer(content: string, origin?: string): void {
|
||||
this.steerBuffer.push({ content, origin });
|
||||
}
|
||||
|
||||
retry(): Promise<void> {
|
||||
throw new Error('TODO: TurnService.retry');
|
||||
}
|
||||
|
||||
cancel(reason?: string): void {
|
||||
if (this.active === undefined) return;
|
||||
this.active.cancelled = true;
|
||||
const turnId = this.active.turnId;
|
||||
this.active = undefined;
|
||||
this._onDidEndTurn.fire({ turnId, reason: reason ?? 'cancelled' });
|
||||
}
|
||||
|
||||
private async launch(input: string): Promise<void> {
|
||||
const turnId = `turn-${nextTurnId++}`;
|
||||
this.active = { turnId, cancelled: false };
|
||||
this._onWillStartTurn.fire({ turnId });
|
||||
try {
|
||||
await this.loopRunner.run();
|
||||
this._onDidEndStep.fire({ turnId, step: 0 });
|
||||
} finally {
|
||||
if (this.active?.turnId === turnId) {
|
||||
this.active = undefined;
|
||||
this._onDidEndTurn.fire({ turnId, reason: 'completed' });
|
||||
}
|
||||
}
|
||||
void input;
|
||||
declare module '../types' {
|
||||
interface WireRecordMap {
|
||||
'turn.launch': {
|
||||
turnId: number;
|
||||
origin: PromptOrigin;
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
registerScopedService(LifecycleScope.Agent, ITurnService, TurnService, InstantiationType.Delayed, 'turn');
|
||||
export class TurnRunnerService implements ITurnRunner {
|
||||
private nextTurnId = 0;
|
||||
private activeTurn: Turn | undefined;
|
||||
private readonly readyControllers = new WeakMap<Turn, ControlledPromise<void>>();
|
||||
private readonly readySettled = new WeakSet<Turn>();
|
||||
private readonly currentStepByTurn = new Map<number, number>();
|
||||
private readonly interruptedTelemetryTurnIds = new Set<number>();
|
||||
private readonly telemetryModeByTurn = new Map<number, 'agent' | 'plan'>();
|
||||
|
||||
readonly hooks = {
|
||||
onLaunched: new OrderedHookSlot<{ turn: Turn }>(),
|
||||
onEnded: new OrderedHookSlot<TurnEndedContext>(),
|
||||
beforeStep: new OrderedHookSlot<TurnStepContext>(),
|
||||
afterStep: new OrderedHookSlot<TurnStepContext>(),
|
||||
};
|
||||
|
||||
constructor(
|
||||
@ILoopService private readonly loop: ILoopService,
|
||||
@IUsageService private readonly usage: IUsageService,
|
||||
@IEventBus private readonly events: IEventBus,
|
||||
@IWireRecord private readonly wireRecord: IWireRecord,
|
||||
@IContextMemory private readonly context: IContextMemory,
|
||||
@IExternalHooksService private readonly externalHooks: IExternalHooksService,
|
||||
@IInstantiationService private readonly instantiation: IInstantiationService,
|
||||
@ITelemetryService private readonly telemetry: ITelemetryService,
|
||||
) {
|
||||
wireRecord.register('turn.launch', (record) => {
|
||||
this.restoreLaunch(record.turnId);
|
||||
});
|
||||
this.hooks.beforeStep.register('turn-before-step-event', async (ctx, next) => {
|
||||
await next();
|
||||
this.resolveReady(ctx.turn);
|
||||
});
|
||||
this.events.on((event) => {
|
||||
if (event.type === 'turn.step.started') {
|
||||
this.currentStepByTurn.set(event.turnId, event.step);
|
||||
return;
|
||||
}
|
||||
if (event.type === 'turn.step.interrupted') {
|
||||
this.trackTurnInterrupted(event.turnId, event.step);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
launch(origin: PromptOrigin): Turn {
|
||||
if (this.activeTurn !== undefined) {
|
||||
throw new Error(`Cannot launch a new turn while turn ${this.activeTurn.id} is active`);
|
||||
}
|
||||
|
||||
const turnId = this.nextTurnId;
|
||||
this.wireRecord.append({ type: 'turn.launch', turnId, origin });
|
||||
this.restoreLaunch(turnId);
|
||||
const abortController = new AbortController();
|
||||
const ready = createControlledPromise<void>();
|
||||
const turn: MutableTurn = {
|
||||
id: turnId,
|
||||
abortController,
|
||||
ready: ready.promise,
|
||||
result: Promise.resolve({ reason: 'failed' }),
|
||||
};
|
||||
this.readyControllers.set(turn, ready);
|
||||
void ready.promise.catch(() => undefined);
|
||||
this.activeTurn = turn;
|
||||
turn.result = this.runTurn(turn, origin);
|
||||
void this.hooks.onLaunched.run({ turn });
|
||||
return turn;
|
||||
}
|
||||
|
||||
getActiveTurn(): Turn | undefined {
|
||||
return this.activeTurn;
|
||||
}
|
||||
|
||||
cancel(turnId?: number, reason?: unknown): void {
|
||||
const turn = this.activeTurn;
|
||||
if (turn === undefined) return;
|
||||
if (turnId !== undefined && turn.id !== turnId) return;
|
||||
turn.abortController.abort(reason ?? userCancellationReason());
|
||||
}
|
||||
|
||||
private async runTurn(turn: Turn, origin: PromptOrigin): Promise<TurnResult> {
|
||||
const startedAt = Date.now();
|
||||
const telemetryMode = this.telemetryMode();
|
||||
this.telemetryModeByTurn.set(turn.id, telemetryMode);
|
||||
let result: TurnResult | undefined;
|
||||
try {
|
||||
this.usage.beginTurn();
|
||||
this.telemetry.track('turn_started', { mode: telemetryMode });
|
||||
this.events.emit({ type: 'turn.started', turnId: turn.id, origin });
|
||||
const promptHookResult = await this.applyUserPromptHook(turn, origin);
|
||||
if (promptHookResult !== undefined) {
|
||||
result = promptHookResult;
|
||||
return result;
|
||||
}
|
||||
result = await this.loop.runTurn(turn, {
|
||||
beforeStep: this.hooks.beforeStep,
|
||||
afterStep: this.hooks.afterStep,
|
||||
});
|
||||
return result;
|
||||
} catch (error) {
|
||||
if (turn.abortController.signal.aborted) {
|
||||
result = { reason: 'cancelled', error: turn.abortController.signal.reason };
|
||||
this.rejectReady(turn, turn.abortController.signal.reason);
|
||||
return result;
|
||||
}
|
||||
this.externalHooks.triggerStopFailure(error, turn.abortController.signal);
|
||||
this.rejectReady(turn, error);
|
||||
result = { reason: 'failed', error };
|
||||
return result;
|
||||
} finally {
|
||||
if (result !== undefined) {
|
||||
this.rejectReady(turn, result);
|
||||
}
|
||||
this.usage.endTurn();
|
||||
if (this.activeTurn === turn) {
|
||||
this.activeTurn = undefined;
|
||||
}
|
||||
if (result !== undefined) {
|
||||
const ended = toTurnEndedEvent(turn, result, Date.now() - startedAt);
|
||||
if (
|
||||
ended.reason === 'cancelled' &&
|
||||
isUserCancellation(turn.abortController.signal.reason)
|
||||
) {
|
||||
this.externalHooks.triggerInterrupt({ turnId: turn.id, reason: 'cancelled' });
|
||||
}
|
||||
this.events.emit(ended);
|
||||
if (ended.error !== undefined) {
|
||||
this.events.emit({ type: 'error', ...ended.error });
|
||||
}
|
||||
if (ended.reason !== 'completed') {
|
||||
this.trackTurnInterrupted(turn.id, this.currentStepByTurn.get(turn.id) ?? 0);
|
||||
}
|
||||
}
|
||||
if (result !== undefined) {
|
||||
await this.hooks.onEnded.run({ turn, result });
|
||||
}
|
||||
this.currentStepByTurn.delete(turn.id);
|
||||
this.interruptedTelemetryTurnIds.delete(turn.id);
|
||||
this.telemetryModeByTurn.delete(turn.id);
|
||||
}
|
||||
}
|
||||
|
||||
private resolveReady(turn: Turn): void {
|
||||
if (this.readySettled.has(turn)) return;
|
||||
this.readySettled.add(turn);
|
||||
this.readyControllers.get(turn)?.resolve();
|
||||
}
|
||||
|
||||
private restoreLaunch(turnId: number): void {
|
||||
if (Number.isInteger(turnId) && turnId >= this.nextTurnId) {
|
||||
this.nextTurnId = turnId + 1;
|
||||
}
|
||||
}
|
||||
|
||||
private async applyUserPromptHook(
|
||||
turn: Turn,
|
||||
origin: PromptOrigin,
|
||||
): Promise<TurnResult | undefined> {
|
||||
if (origin.kind !== 'user') return undefined;
|
||||
const promptMessage = this.context.getHistory().at(-1);
|
||||
if (!shouldRunUserPromptHook(promptMessage)) return undefined;
|
||||
|
||||
const hookResult = await this.externalHooks.triggerUserPromptSubmit(
|
||||
promptMessage.content,
|
||||
turn.abortController.signal,
|
||||
);
|
||||
if (hookResult?.action === 'block') {
|
||||
this.append({
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: hookResult.text }],
|
||||
toolCalls: [],
|
||||
origin: { kind: 'hook_result', event: hookResult.event, blocked: true },
|
||||
});
|
||||
this.events.emit({
|
||||
type: 'hook.result',
|
||||
turnId: turn.id,
|
||||
hookEvent: hookResult.event,
|
||||
content: hookResult.message,
|
||||
blocked: true,
|
||||
});
|
||||
return { reason: 'completed' };
|
||||
}
|
||||
|
||||
if (hookResult?.action === 'append') {
|
||||
this.append({
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: hookResult.text }],
|
||||
toolCalls: [],
|
||||
origin: { kind: 'hook_result', event: hookResult.event },
|
||||
});
|
||||
this.events.emit({
|
||||
type: 'hook.result',
|
||||
turnId: turn.id,
|
||||
hookEvent: hookResult.event,
|
||||
content: hookResult.message,
|
||||
});
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
private append(...messages: ContextMessage[]): void {
|
||||
if (messages.length === 0) return;
|
||||
this.context.spliceHistory(this.context.getHistory().length, 0, messages);
|
||||
}
|
||||
|
||||
private rejectReady(turn: Turn, reason: unknown): void {
|
||||
if (this.readySettled.has(turn)) return;
|
||||
this.readySettled.add(turn);
|
||||
this.readyControllers.get(turn)?.reject(reason);
|
||||
}
|
||||
|
||||
private trackTurnInterrupted(turnId: number, atStep: number): void {
|
||||
if (this.interruptedTelemetryTurnIds.has(turnId)) return;
|
||||
this.interruptedTelemetryTurnIds.add(turnId);
|
||||
this.telemetry.track('turn_interrupted', {
|
||||
mode: this.telemetryModeByTurn.get(turnId) ?? this.telemetryMode(),
|
||||
at_step: atStep,
|
||||
});
|
||||
}
|
||||
|
||||
private telemetryMode(): 'agent' | 'plan' {
|
||||
const planMode = this.instantiation.invokeFunction((accessor) =>
|
||||
accessor.get(IPlanService),
|
||||
);
|
||||
return planMode.isActive ? 'plan' : 'agent';
|
||||
}
|
||||
}
|
||||
|
||||
function shouldRunUserPromptHook(message: ContextMessage | undefined): message is ContextMessage {
|
||||
if (message === undefined || message.role !== 'user') return false;
|
||||
return (message.origin ?? USER_PROMPT_ORIGIN).kind === 'user';
|
||||
}
|
||||
|
||||
function toTurnEndedEvent(
|
||||
turn: Turn,
|
||||
result: TurnResult,
|
||||
durationMs: number,
|
||||
): {
|
||||
type: 'turn.ended';
|
||||
turnId: number;
|
||||
reason: TurnResult['reason'];
|
||||
error?: KimiErrorPayload;
|
||||
durationMs: number;
|
||||
} {
|
||||
if (result.reason !== 'failed' || result.error === undefined) {
|
||||
return { type: 'turn.ended', turnId: turn.id, reason: result.reason, durationMs };
|
||||
}
|
||||
return {
|
||||
type: 'turn.ended',
|
||||
turnId: turn.id,
|
||||
reason: result.reason,
|
||||
error: summarizeTurnError(result.error, turn.id),
|
||||
durationMs,
|
||||
};
|
||||
}
|
||||
|
||||
const LLM_NOT_SET_MESSAGE = 'LLM not set, send "/login" to login';
|
||||
|
||||
function summarizeTurnError(error: unknown, turnId: number): KimiErrorPayload {
|
||||
const payload = toKimiErrorPayload(error);
|
||||
const details = { ...payload.details, turnId };
|
||||
// Substitute a friendlier, login-aware message for model-not-configured. The
|
||||
// raw "Model not set" / "Provider not set" text is not actionable.
|
||||
if (payload.code === 'model.not_configured') {
|
||||
return { ...payload, message: LLM_NOT_SET_MESSAGE, details };
|
||||
}
|
||||
return { ...payload, details };
|
||||
}
|
||||
|
||||
interface ControlledPromise<T> {
|
||||
readonly promise: Promise<T>;
|
||||
resolve(value: T | PromiseLike<T>): void;
|
||||
reject(reason?: unknown): void;
|
||||
}
|
||||
|
||||
type MutableTurn = {
|
||||
-readonly [K in keyof Turn]: Turn[K];
|
||||
};
|
||||
|
||||
function createControlledPromise<T>(): ControlledPromise<T> {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((res, rej) => {
|
||||
resolve = res;
|
||||
reject = rej;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
registerSingleton(ITurnRunner, new SyncDescriptor(TurnRunnerService, [], true));
|
||||
|
|
|
|||
|
|
@ -1,6 +0,0 @@
|
|||
/**
|
||||
* `turnRunner` domain barrel - re-exports the turnRunner service contract and implementation.
|
||||
*/
|
||||
|
||||
export * from './turnRunner';
|
||||
export * from './turnRunnerService';
|
||||
|
|
@ -1,21 +0,0 @@
|
|||
import { createDecorator } from "#/_base/di";
|
||||
import type { PromptOrigin } from '../../../agent/context';
|
||||
|
||||
import type { Hooks } from '../hooks';
|
||||
import type { Turn, TurnEndedContext, TurnStepContext } from '../types';
|
||||
|
||||
export interface ITurnRunner {
|
||||
launch(origin: PromptOrigin): Turn;
|
||||
getActiveTurn(): Turn | undefined;
|
||||
cancel(turnId?: number, reason?: unknown): void;
|
||||
|
||||
readonly hooks: Hooks<{
|
||||
onLaunched: { turn: Turn };
|
||||
onEnded: TurnEndedContext;
|
||||
beforeStep: TurnStepContext;
|
||||
afterStep: TurnStepContext;
|
||||
}>;
|
||||
}
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-redeclare
|
||||
export const ITurnRunner = createDecorator<ITurnRunner>('agentTurnRunnerService');
|
||||
|
|
@ -1,324 +0,0 @@
|
|||
import {
|
||||
IInstantiationService,
|
||||
registerSingleton,
|
||||
SyncDescriptor,
|
||||
} from "#/_base/di";
|
||||
import type { ContextMessage, PromptOrigin } from '../../../agent/context';
|
||||
import { USER_PROMPT_ORIGIN } from '../../../agent/context';
|
||||
import { toKimiErrorPayload, type KimiErrorPayload } from "#/_base/errors";
|
||||
import { isUserCancellation, userCancellationReason } from "#/_base/utils/abort";
|
||||
import { IContextMemory } from '../contextMemory/contextMemory';
|
||||
import { IEventBus } from '../eventBus/eventBus';
|
||||
import { IExternalHooksService } from '../externalHooks/externalHooks';
|
||||
import { OrderedHookSlot } from '../hooks';
|
||||
import { ILoopService } from '../loop/loop';
|
||||
import { IPlanService } from '../plan/planMode';
|
||||
import { ITelemetryService } from '../telemetry/telemetry';
|
||||
import type {
|
||||
Turn,
|
||||
TurnEndedContext,
|
||||
TurnResult,
|
||||
TurnStepContext,
|
||||
} from '../types';
|
||||
import { IUsageService } from '../usage/usage';
|
||||
import { IWireRecord } from '../wireRecord/wireRecord';
|
||||
import { ITurnRunner } from './turnRunner';
|
||||
|
||||
declare module '../types' {
|
||||
interface WireRecordMap {
|
||||
'turn.launch': {
|
||||
turnId: number;
|
||||
origin: PromptOrigin;
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export class TurnRunnerService implements ITurnRunner {
|
||||
private nextTurnId = 0;
|
||||
private activeTurn: Turn | undefined;
|
||||
private readonly readyControllers = new WeakMap<Turn, ControlledPromise<void>>();
|
||||
private readonly readySettled = new WeakSet<Turn>();
|
||||
private readonly currentStepByTurn = new Map<number, number>();
|
||||
private readonly interruptedTelemetryTurnIds = new Set<number>();
|
||||
private readonly telemetryModeByTurn = new Map<number, 'agent' | 'plan'>();
|
||||
|
||||
readonly hooks = {
|
||||
onLaunched: new OrderedHookSlot<{ turn: Turn }>(),
|
||||
onEnded: new OrderedHookSlot<TurnEndedContext>(),
|
||||
beforeStep: new OrderedHookSlot<TurnStepContext>(),
|
||||
afterStep: new OrderedHookSlot<TurnStepContext>(),
|
||||
};
|
||||
|
||||
constructor(
|
||||
@ILoopService private readonly loop: ILoopService,
|
||||
@IUsageService private readonly usage: IUsageService,
|
||||
@IEventBus private readonly events: IEventBus,
|
||||
@IWireRecord private readonly wireRecord: IWireRecord,
|
||||
@IContextMemory private readonly context: IContextMemory,
|
||||
@IExternalHooksService private readonly externalHooks: IExternalHooksService,
|
||||
@IInstantiationService private readonly instantiation: IInstantiationService,
|
||||
@ITelemetryService private readonly telemetry: ITelemetryService,
|
||||
) {
|
||||
wireRecord.register('turn.launch', (record) => {
|
||||
this.restoreLaunch(record.turnId);
|
||||
});
|
||||
this.hooks.beforeStep.register('turn-before-step-event', async (ctx, next) => {
|
||||
await next();
|
||||
this.resolveReady(ctx.turn);
|
||||
});
|
||||
this.events.on((event) => {
|
||||
if (event.type === 'turn.step.started') {
|
||||
this.currentStepByTurn.set(event.turnId, event.step);
|
||||
return;
|
||||
}
|
||||
if (event.type === 'turn.step.interrupted') {
|
||||
this.trackTurnInterrupted(event.turnId, event.step);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
launch(origin: PromptOrigin): Turn {
|
||||
if (this.activeTurn !== undefined) {
|
||||
throw new Error(`Cannot launch a new turn while turn ${this.activeTurn.id} is active`);
|
||||
}
|
||||
|
||||
const turnId = this.nextTurnId;
|
||||
this.wireRecord.append({ type: 'turn.launch', turnId, origin });
|
||||
this.restoreLaunch(turnId);
|
||||
const abortController = new AbortController();
|
||||
const ready = createControlledPromise<void>();
|
||||
const turn: MutableTurn = {
|
||||
id: turnId,
|
||||
abortController,
|
||||
ready: ready.promise,
|
||||
result: Promise.resolve({ reason: 'failed' }),
|
||||
};
|
||||
this.readyControllers.set(turn, ready);
|
||||
void ready.promise.catch(() => undefined);
|
||||
this.activeTurn = turn;
|
||||
turn.result = this.runTurn(turn, origin);
|
||||
void this.hooks.onLaunched.run({ turn });
|
||||
return turn;
|
||||
}
|
||||
|
||||
getActiveTurn(): Turn | undefined {
|
||||
return this.activeTurn;
|
||||
}
|
||||
|
||||
cancel(turnId?: number, reason?: unknown): void {
|
||||
const turn = this.activeTurn;
|
||||
if (turn === undefined) return;
|
||||
if (turnId !== undefined && turn.id !== turnId) return;
|
||||
turn.abortController.abort(reason ?? userCancellationReason());
|
||||
}
|
||||
|
||||
private async runTurn(turn: Turn, origin: PromptOrigin): Promise<TurnResult> {
|
||||
const startedAt = Date.now();
|
||||
const telemetryMode = this.telemetryMode();
|
||||
this.telemetryModeByTurn.set(turn.id, telemetryMode);
|
||||
let result: TurnResult | undefined;
|
||||
try {
|
||||
this.usage.beginTurn();
|
||||
this.telemetry.track('turn_started', { mode: telemetryMode });
|
||||
this.events.emit({ type: 'turn.started', turnId: turn.id, origin });
|
||||
const promptHookResult = await this.applyUserPromptHook(turn, origin);
|
||||
if (promptHookResult !== undefined) {
|
||||
result = promptHookResult;
|
||||
return result;
|
||||
}
|
||||
result = await this.loop.runTurn(turn, {
|
||||
beforeStep: this.hooks.beforeStep,
|
||||
afterStep: this.hooks.afterStep,
|
||||
});
|
||||
return result;
|
||||
} catch (error) {
|
||||
if (turn.abortController.signal.aborted) {
|
||||
result = { reason: 'cancelled', error: turn.abortController.signal.reason };
|
||||
this.rejectReady(turn, turn.abortController.signal.reason);
|
||||
return result;
|
||||
}
|
||||
this.externalHooks.triggerStopFailure(error, turn.abortController.signal);
|
||||
this.rejectReady(turn, error);
|
||||
result = { reason: 'failed', error };
|
||||
return result;
|
||||
} finally {
|
||||
if (result !== undefined) {
|
||||
this.rejectReady(turn, result);
|
||||
}
|
||||
this.usage.endTurn();
|
||||
if (this.activeTurn === turn) {
|
||||
this.activeTurn = undefined;
|
||||
}
|
||||
if (result !== undefined) {
|
||||
const ended = toTurnEndedEvent(turn, result, Date.now() - startedAt);
|
||||
if (
|
||||
ended.reason === 'cancelled' &&
|
||||
isUserCancellation(turn.abortController.signal.reason)
|
||||
) {
|
||||
this.externalHooks.triggerInterrupt({ turnId: turn.id, reason: 'cancelled' });
|
||||
}
|
||||
this.events.emit(ended);
|
||||
if (ended.error !== undefined) {
|
||||
this.events.emit({ type: 'error', ...ended.error });
|
||||
}
|
||||
if (ended.reason !== 'completed') {
|
||||
this.trackTurnInterrupted(turn.id, this.currentStepByTurn.get(turn.id) ?? 0);
|
||||
}
|
||||
}
|
||||
if (result !== undefined) {
|
||||
await this.hooks.onEnded.run({ turn, result });
|
||||
}
|
||||
this.currentStepByTurn.delete(turn.id);
|
||||
this.interruptedTelemetryTurnIds.delete(turn.id);
|
||||
this.telemetryModeByTurn.delete(turn.id);
|
||||
}
|
||||
}
|
||||
|
||||
private resolveReady(turn: Turn): void {
|
||||
if (this.readySettled.has(turn)) return;
|
||||
this.readySettled.add(turn);
|
||||
this.readyControllers.get(turn)?.resolve();
|
||||
}
|
||||
|
||||
private restoreLaunch(turnId: number): void {
|
||||
if (Number.isInteger(turnId) && turnId >= this.nextTurnId) {
|
||||
this.nextTurnId = turnId + 1;
|
||||
}
|
||||
}
|
||||
|
||||
private async applyUserPromptHook(
|
||||
turn: Turn,
|
||||
origin: PromptOrigin,
|
||||
): Promise<TurnResult | undefined> {
|
||||
if (origin.kind !== 'user') return undefined;
|
||||
const promptMessage = this.context.getHistory().at(-1);
|
||||
if (!shouldRunUserPromptHook(promptMessage)) return undefined;
|
||||
|
||||
const hookResult = await this.externalHooks.triggerUserPromptSubmit(
|
||||
promptMessage.content,
|
||||
turn.abortController.signal,
|
||||
);
|
||||
if (hookResult?.action === 'block') {
|
||||
this.append({
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: hookResult.text }],
|
||||
toolCalls: [],
|
||||
origin: { kind: 'hook_result', event: hookResult.event, blocked: true },
|
||||
});
|
||||
this.events.emit({
|
||||
type: 'hook.result',
|
||||
turnId: turn.id,
|
||||
hookEvent: hookResult.event,
|
||||
content: hookResult.message,
|
||||
blocked: true,
|
||||
});
|
||||
return { reason: 'completed' };
|
||||
}
|
||||
|
||||
if (hookResult?.action === 'append') {
|
||||
this.append({
|
||||
role: 'user',
|
||||
content: [{ type: 'text', text: hookResult.text }],
|
||||
toolCalls: [],
|
||||
origin: { kind: 'hook_result', event: hookResult.event },
|
||||
});
|
||||
this.events.emit({
|
||||
type: 'hook.result',
|
||||
turnId: turn.id,
|
||||
hookEvent: hookResult.event,
|
||||
content: hookResult.message,
|
||||
});
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
private append(...messages: ContextMessage[]): void {
|
||||
if (messages.length === 0) return;
|
||||
this.context.spliceHistory(this.context.getHistory().length, 0, messages);
|
||||
}
|
||||
|
||||
private rejectReady(turn: Turn, reason: unknown): void {
|
||||
if (this.readySettled.has(turn)) return;
|
||||
this.readySettled.add(turn);
|
||||
this.readyControllers.get(turn)?.reject(reason);
|
||||
}
|
||||
|
||||
private trackTurnInterrupted(turnId: number, atStep: number): void {
|
||||
if (this.interruptedTelemetryTurnIds.has(turnId)) return;
|
||||
this.interruptedTelemetryTurnIds.add(turnId);
|
||||
this.telemetry.track('turn_interrupted', {
|
||||
mode: this.telemetryModeByTurn.get(turnId) ?? this.telemetryMode(),
|
||||
at_step: atStep,
|
||||
});
|
||||
}
|
||||
|
||||
private telemetryMode(): 'agent' | 'plan' {
|
||||
const planMode = this.instantiation.invokeFunction((accessor) =>
|
||||
accessor.get(IPlanService),
|
||||
);
|
||||
return planMode.isActive ? 'plan' : 'agent';
|
||||
}
|
||||
}
|
||||
|
||||
function shouldRunUserPromptHook(message: ContextMessage | undefined): message is ContextMessage {
|
||||
if (message === undefined || message.role !== 'user') return false;
|
||||
return (message.origin ?? USER_PROMPT_ORIGIN).kind === 'user';
|
||||
}
|
||||
|
||||
function toTurnEndedEvent(
|
||||
turn: Turn,
|
||||
result: TurnResult,
|
||||
durationMs: number,
|
||||
): {
|
||||
type: 'turn.ended';
|
||||
turnId: number;
|
||||
reason: TurnResult['reason'];
|
||||
error?: KimiErrorPayload;
|
||||
durationMs: number;
|
||||
} {
|
||||
if (result.reason !== 'failed' || result.error === undefined) {
|
||||
return { type: 'turn.ended', turnId: turn.id, reason: result.reason, durationMs };
|
||||
}
|
||||
return {
|
||||
type: 'turn.ended',
|
||||
turnId: turn.id,
|
||||
reason: result.reason,
|
||||
error: summarizeTurnError(result.error, turn.id),
|
||||
durationMs,
|
||||
};
|
||||
}
|
||||
|
||||
const LLM_NOT_SET_MESSAGE = 'LLM not set, send "/login" to login';
|
||||
|
||||
function summarizeTurnError(error: unknown, turnId: number): KimiErrorPayload {
|
||||
const payload = toKimiErrorPayload(error);
|
||||
const details = { ...payload.details, turnId };
|
||||
// Substitute a friendlier, login-aware message for model-not-configured. The
|
||||
// raw "Model not set" / "Provider not set" text is not actionable.
|
||||
if (payload.code === 'model.not_configured') {
|
||||
return { ...payload, message: LLM_NOT_SET_MESSAGE, details };
|
||||
}
|
||||
return { ...payload, details };
|
||||
}
|
||||
|
||||
interface ControlledPromise<T> {
|
||||
readonly promise: Promise<T>;
|
||||
resolve(value: T | PromiseLike<T>): void;
|
||||
reject(reason?: unknown): void;
|
||||
}
|
||||
|
||||
type MutableTurn = {
|
||||
-readonly [K in keyof Turn]: Turn[K];
|
||||
};
|
||||
|
||||
function createControlledPromise<T>(): ControlledPromise<T> {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((res, rej) => {
|
||||
resolve = res;
|
||||
reject = rej;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
registerSingleton(ITurnRunner, new SyncDescriptor(TurnRunnerService, [], true));
|
||||
Loading…
Add table
Add a link
Reference in a new issue