aps-agent/server/app/turn-runner.ts

133 lines
4.6 KiB
TypeScript
Raw Normal View History

/**
* 回合执行器:把运行时的 RuntimeEvent 翻译成平台事件,并让工具调用经 Core.invoke 落地。
*
* 这是引擎与平台之间唯一的翻译层(架构设计 5.2「事件」一行):
* text → message.delta
* tool_call → tool.call(参数只发摘要)+ 调 Core.invoke + 把回执送回运行时
* tool_result → tool.result
* approval_required → approval.requested
* error → error
* turn_end → message.completed + run.state(idle)
*
* A4 之后,写能力的判定与确认卡插在「拿到 tool_call」与「调 Core.invoke」之间。
*/
import type { ExecutionReceipt } from '@aps/contracts';
import { digestParams } from '../core/execution/store.js';
import type { EventInput } from './events.js';
import type { RuntimeSession } from '../agent/runtime.js';
export interface TurnRunnerInput {
runtimeSession: RuntimeSession;
sessionId: string;
runId: string;
assistantMessageId: string;
text: string;
/** 事件发布(EventBus.publish) */
publish: (event: EventInput) => void;
/** 工具出口:只能是 Core.invoke */
invoke: (capabilityId: string, args: unknown, idempotencyKey: string) => Promise<ExecutionReceipt>;
signal: AbortSignal;
}
export interface TurnResult {
text: string;
finishReason: 'stop' | 'cancelled' | 'error';
toolCalls: number;
}
export async function runTurn(input: TurnRunnerInput): Promise<TurnResult> {
const { runtimeSession, sessionId, runId, assistantMessageId, publish, invoke, signal } = input;
let emitted = '';
let toolCalls = 0;
let finishReason: TurnResult['finishReason'] = 'stop';
publish({ sessionId, runId, type: 'run.state', data: { state: 'running' } });
try {
for await (const event of runtimeSession.send(input.text, { signal })) {
switch (event.type) {
case 'text': {
emitted += event.delta;
publish({ sessionId, runId, type: 'message.delta', data: { messageId: assistantMessageId, text: event.delta } });
break;
}
case 'tool_call': {
toolCalls += 1;
publish({
sessionId,
runId,
type: 'tool.call',
data: { callId: event.callId, capabilityId: event.capabilityId, argsDigest: digestParams(event.args) },
});
// 幂等键用 callId:同一回合重放不会重复执行
const receipt = await invoke(event.capabilityId, event.args, `call-${event.callId}`);
publish({
sessionId,
runId,
type: 'tool.result',
data: { callId: event.callId, executionId: receipt.executionId, status: receipt.status },
});
runtimeSession.resolveToolResult(event.callId, receipt);
if (receipt.status === 'failed') {
publish({
sessionId,
runId,
type: 'error',
data: { code: `tool.${receipt.errorKind ?? 'failed'}`, message: toolFailureText(receipt), retryable: true },
});
}
break;
}
case 'tool_result': {
// 运行时自行上报的回执(真引擎可能直接给),不重复发事件
break;
}
case 'approval_required': {
publish({ sessionId, runId, type: 'approval.requested', data: event.request });
break;
}
case 'error': {
publish({
sessionId,
runId,
type: 'error',
data: { code: `engine.${event.kind}`, message: event.message, retryable: event.kind !== 'forbidden', errorKind: event.kind },
});
finishReason = 'error';
break;
}
case 'turn_end': {
finishReason = event.reason;
break;
}
default:
break;
}
}
} catch (error) {
finishReason = 'error';
publish({
sessionId,
runId,
type: 'error',
data: { code: 'engine.failed', message: error instanceof Error ? error.message : '引擎执行失败', retryable: true },
});
}
if (signal.aborted && finishReason === 'stop') finishReason = 'cancelled';
publish({
sessionId,
runId,
type: 'message.completed',
data: { messageId: assistantMessageId, finishReason },
});
publish({ sessionId, runId, type: 'run.state', data: { state: 'idle' } });
return { text: emitted, finishReason, toolCalls };
}
function toolFailureText(receipt: ExecutionReceipt): string {
const detail = receipt.result as { message?: string } | undefined;
return detail?.message ?? `能力调用失败(${receipt.errorKind ?? 'unknown'})`;
}