diff --git a/scripts/pi-bridge.test.mjs b/scripts/pi-bridge.test.mjs new file mode 100644 index 0000000..cac909e --- /dev/null +++ b/scripts/pi-bridge.test.mjs @@ -0,0 +1,122 @@ +/** + * 能力桥用例(任务 A3.4 / A3.11)。 + * + * 不启动 Pi:把扩展装到一个假的 Pi 注册表上,直接调它注册出来的工具, + * 验证「Pi 工具调用 → 平台内部端点 → Core.invoke → 假 Provider → 回执」这条链路。 + * 真引擎那一环另有一条需要 APS_PI_INTEGRATION=1 的用例(见 pi-runtime.test.mjs)。 + */ +import assert from 'node:assert/strict'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { pathToFileURL } from 'node:url'; + +const EXTENSION = pathToFileURL(join(process.cwd(), 'server/agent/pi/extension/aps-bridge.ts')).href; + +async function startPlatform() { + process.env.APS_DEV_PROVIDER = '1'; + const { startServer } = await import('../server/dist/index.js'); + return startServer({ host: '127.0.0.1', port: 0, version: 'test', databasePath: ':memory:', engine: 'fake' }); +} + +/** 用假 Pi 加载扩展,返回注册出来的工具表 */ +async function loadBridge(configPath) { + const previous = process.env.APS_BRIDGE_CONFIG; + process.env.APS_BRIDGE_CONFIG = configPath; + try { + const module = await import(EXTENSION); + const tools = new Map(); + module.default({ + registerTool(tool) { + tools.set(tool.name, tool); + }, + }); + return tools; + } finally { + if (previous === undefined) delete process.env.APS_BRIDGE_CONFIG; + else process.env.APS_BRIDGE_CONFIG = previous; + } +} + +async function writeBridgeConfig(server, sessionId, capabilities) { + const dir = await mkdtemp(join(tmpdir(), 'aps-bridge-')); + const file = join(dir, 'bridge.json'); + await writeFile( + file, + JSON.stringify({ baseUrl: server.url, internalToken: server.internalToken, sessionId, capabilities }), + 'utf8', + ); + return { dir, file }; +} + +test('扩展把平台能力注册成 Pi 工具,调用经 Core.invoke 返回回执', async () => { + const server = await startPlatform(); + const created = await ( + await fetch(`${server.url}/v1/sessions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ scope: 'project', projectId: 'p-default', title: '桥用例' }), + }) + ).json(); + const sessionId = created.session.sessionId; + + const { dir, file } = await writeBridgeConfig(server, sessionId, [ + { + toolName: 'fake_echo', + capabilityId: 'fake.echo', + description: '回显输入', + parameters: { type: 'object', properties: { text: { type: 'string' } }, required: ['text'] }, + }, + ]); + + try { + const tools = await loadBridge(file); + assert.ok(tools.has('fake_echo'), `应注册 fake_echo,实际:${[...tools.keys()].join(',')}`); + + const tool = tools.get('fake_echo'); + const result = await tool.execute('call-abc', { text: '你好' }); + const receipt = JSON.parse(result.content[0].text); + assert.equal(receipt.status, 'succeeded'); + assert.deepEqual(receipt.result, { echo: '你好' }); + + // 平台侧留下执行记录,且能经 HTTP 查到 + const execution = await (await fetch(`${server.url}/v1/executions/${receipt.executionId}`)).json(); + assert.equal(execution.status, 'succeeded'); + const record = server.service.core.executions.get(receipt.executionId); + assert.equal(record.sessionId, sessionId, '执行记录应归属发起会话'); + assert.equal(record.capabilityId, 'fake.echo'); + } finally { + await server.close(); + await rm(dir, { recursive: true, force: true }); + delete process.env.APS_DEV_PROVIDER; + } +}); + +test('能力不存在时扩展抛错,Pi 会把它当工具错误(不会伪造成功)', async () => { + const server = await startPlatform(); + const created = await ( + await fetch(`${server.url}/v1/sessions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ scope: 'project', projectId: 'p-default', title: '失败路径' }), + }) + ).json(); + const { dir, file } = await writeBridgeConfig(server, created.session.sessionId, [ + { toolName: 'fake_nope', capabilityId: 'fake.__nope__', description: '不存在的能力', parameters: { type: 'object' } }, + ]); + + try { + const tools = await loadBridge(file); + await assert.rejects(() => tools.get('fake_nope').execute('call-bad', {}), /执行失败|invalid_argument/); + } finally { + await server.close(); + await rm(dir, { recursive: true, force: true }); + delete process.env.APS_DEV_PROVIDER; + } +}); + +test('没有桥配置时不注册任何工具(宁可说做不到,也不给坏工具)', async () => { + const tools = await loadBridge(join(tmpdir(), 'not-exists-bridge.json')); + assert.equal(tools.size, 0); +}); \ No newline at end of file diff --git a/scripts/turn-runner.test.mjs b/scripts/turn-runner.test.mjs new file mode 100644 index 0000000..e015dba --- /dev/null +++ b/scripts/turn-runner.test.mjs @@ -0,0 +1,106 @@ +/** + * 回合执行器按工具执行模式分流(A3.4)。 + * + * 两种模式必须都覆盖,否则 Pi 那条路只有真模型测试才碰得到: + * platform —— 平台执行并回填(假引擎走这条) + * runtime —— 引擎自己执行(Pi + 扩展桥),平台只观察事件、不得重复执行 + */ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +import { runTurn } from '@aps/server/app/turn-runner'; + +function stubSession(mode, events) { + return { + id: 'stub', + toolExecution: mode, + async *send() { + for (const event of events) yield event; + }, + resolveToolResult() {}, + abort() {}, + async dispose() {}, + }; +} + +const toolCallEvents = [ + { type: 'text', delta: '我来调用。' }, + { type: 'tool_call', callId: 'c-1', capabilityId: 'test.echo', args: { q: 1 } }, + { type: 'tool_result', callId: 'c-1', receipt: { executionId: 'exec-1', status: 'succeeded' } }, + { type: 'turn_end', reason: 'stop' }, +]; + +function collect() { + const events = []; + return { events, publish: (event) => events.push(event) }; +} + +test('runtime 模式:平台不执行工具,只观察 tool.call 与 tool.result', async () => { + const { events, publish } = collect(); + let invocations = 0; + const result = await runTurn({ + runtimeSession: stubSession('runtime', toolCallEvents), + sessionId: 's-1', + runId: 'r-1', + assistantMessageId: 'm-1', + text: '调用试试', + publish, + invoke: async () => { + invocations += 1; + return { executionId: 'platform-should-not-run', status: 'succeeded' }; + }, + signal: new AbortController().signal, + }); + + assert.equal(invocations, 0, '引擎模式下平台不得再执行一次(否则等于重复调用)'); + const types = events.map((event) => event.type); + assert.deepEqual(types, ['run.state', 'message.delta', 'tool.call', 'tool.result', 'message.completed', 'run.state']); + assert.equal(result.toolCalls, 1); + assert.equal(result.finishReason, 'stop'); +}); + +test('platform 模式:平台执行并把回执回填给运行时', async () => { + const { events, publish } = collect(); + const resolved = []; + const session = stubSession('platform', toolCallEvents); + session.resolveToolResult = (callId, receipt) => resolved.push({ callId, receipt }); + + let invocations = 0; + await runTurn({ + runtimeSession: session, + sessionId: 's-1', + runId: 'r-1', + assistantMessageId: 'm-1', + text: '调用试试', + publish, + invoke: async () => { + invocations += 1; + return { executionId: 'exec-platform', status: 'succeeded' }; + }, + signal: new AbortController().signal, + }); + + assert.equal(invocations, 1, '平台模式必须执行一次'); + assert.deepEqual(resolved, [{ callId: 'c-1', receipt: { executionId: 'exec-platform', status: 'succeeded' } }]); + const types = events.map((event) => event.type); + assert.ok(types.includes('tool.result'), types.join(',')); +}); + +test('错误回合的结束原因不会被 turn_end 覆盖回 stop', async () => { + const { events, publish } = collect(); + const result = await runTurn({ + runtimeSession: stubSession('runtime', [ + { type: 'error', kind: 'forbidden', message: '401 无效密钥' }, + { type: 'turn_end', reason: 'stop' }, + ]), + sessionId: 's-1', + runId: 'r-1', + assistantMessageId: 'm-1', + text: '触发错误', + publish, + invoke: async () => ({ executionId: 'x', status: 'succeeded' }), + signal: new AbortController().signal, + }); + assert.equal(result.finishReason, 'error'); + assert.equal(events.find((event) => event.type === 'message.completed').data.finishReason, 'error'); +}); \ No newline at end of file diff --git a/server/agent/fake-runtime.ts b/server/agent/fake-runtime.ts index 9c825f8..6e904eb 100644 --- a/server/agent/fake-runtime.ts +++ b/server/agent/fake-runtime.ts @@ -41,6 +41,7 @@ export function createFakeRuntime(options: FakeRuntimeOptions = {}): FakeRuntime return { name: options.name ?? 'fake', + toolExecution: 'platform', stats, async createSession(input: CreateRuntimeSessionInput): Promise { stats.sessions += 1; @@ -115,6 +116,7 @@ export function createFakeRuntime(options: FakeRuntimeOptions = {}): FakeRuntime return { id, + toolExecution: 'platform', send(text: string, sendOptions?: SendOptions): AsyncIterable { return runTurn(text, sendOptions); }, diff --git a/server/agent/pi/extension/aps-bridge.ts b/server/agent/pi/extension/aps-bridge.ts new file mode 100644 index 0000000..77f58d5 --- /dev/null +++ b/server/agent/pi/extension/aps-bridge.ts @@ -0,0 +1,91 @@ +/** + * 平台能力桥(任务 A3.11):把平台的能力目录注册成 Pi 的工具。 + * + * 这个文件由平台以 `--extension ` 加载,运行在 Pi 子进程里(Pi 用 jiti 直接读 TS)。 + * 它自己不判断权限、不碰凭据,只做一件事:把 Pi 的工具调用转成对平台内部回调端点的 HTTP 请求, + * 平台侧收口到 Core.invoke(架构设计 5.1「每次都过 Core」)。 + * + * 配置来自环境变量 APS_BRIDGE_CONFIG 指向的 JSON 文件,由平台在建立会话时写入: + * { baseUrl, internalToken, sessionId, capabilities: [{ toolName, capabilityId, description, parameters }] } + */ +import { readFileSync } from 'node:fs'; + +interface BridgeCapability { + toolName: string; + capabilityId: string; + description: string; + parameters: Record; +} + +interface BridgeConfig { + baseUrl: string; + internalToken: string; + sessionId: string; + capabilities: BridgeCapability[]; +} + +interface PlatformReceipt { + executionId: string; + status: 'accepted' | 'pending-confirmation' | 'succeeded' | 'failed'; + result?: unknown; + errorKind?: string; +} + +/** Pi 扩展的注册接口(只声明用到的部分,避免扩展依赖 Pi 的类型包) */ +interface PiExtensionApi { + registerTool(tool: { + name: string; + label: string; + description: string; + parameters: Record; + execute: (toolCallId: string, params: unknown, signal?: AbortSignal) => Promise<{ content: Array<{ type: 'text'; text: string }>; details: unknown }>; + }): void; +} + +export default function apsBridge(pi: PiExtensionApi): void { + const configPath = process.env.APS_BRIDGE_CONFIG ?? ''; + if (configPath === '') return; + + let config: BridgeConfig; + try { + config = JSON.parse(readFileSync(configPath, 'utf8')) as BridgeConfig; + } catch { + // 配置读不到就不注册任何工具:宁可让模型说做不到,也不要注册一个坏工具 + return; + } + + for (const capability of config.capabilities) { + pi.registerTool({ + name: capability.toolName, + label: capability.capabilityId, + description: capability.description, + parameters: capability.parameters, + async execute(toolCallId: string, params: unknown, signal?: AbortSignal) { + const response = await fetch(`${config.baseUrl}/internal/tool-call`, { + method: 'POST', + headers: { 'content-type': 'application/json', 'x-aps-internal-token': config.internalToken }, + body: JSON.stringify({ + sessionId: config.sessionId, + capabilityId: capability.capabilityId, + params, + idempotencyKey: `call-${toolCallId}`, + }), + ...(signal === undefined ? {} : { signal }), + }); + + if (!response.ok) { + const detail = await response.text().catch(() => ''); + throw new Error(`平台拒绝调用(HTTP ${response.status}):${detail.slice(0, 200)}`); + } + + const receipt = (await response.json()) as PlatformReceipt; + if (receipt.status === 'failed') { + const detail = (receipt.result as { message?: string } | undefined)?.message ?? '未提供原因'; + throw new Error(`能力 ${capability.capabilityId} 执行失败(${receipt.errorKind ?? 'unknown'}):${detail}`); + } + // 成功:把回执原样交给模型(含 executionId,便于它引用证据) + return { content: [{ type: 'text', text: JSON.stringify(receipt) }], details: receipt }; + }, + }); + } +} \ No newline at end of file diff --git a/server/agent/pi/runtime.ts b/server/agent/pi/runtime.ts index 0fa8a26..7afc4dc 100644 --- a/server/agent/pi/runtime.ts +++ b/server/agent/pi/runtime.ts @@ -7,7 +7,7 @@ * - 事件面:把 Pi 的会话事件映射成平台的 RuntimeEvent,上层只认平台类型 * - 状态:平台保存会话历史;Pi 自己的 session 文件放在平台管理目录里,只作同引擎恢复用(ADR-0003) */ -import { existsSync } from 'node:fs'; +import { existsSync, mkdirSync, writeFileSync } from 'node:fs'; import { homedir } from 'node:os'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; @@ -29,6 +29,16 @@ export interface PiRuntimeOptions { extraArgs?: string[]; /** 单回合最长等待(毫秒),超时按引擎错误处理 */ turnTimeoutMs?: number; + /** + * 能力桥:给定平台内部回调地址与令牌。 + * 传了它就会加载平台的能力桥扩展,Pi 才真正拥有平台的能力目录(A3.11)。 + * 用函数是为了延后取值——端口要等监听之后才知道。 + */ + bridge?: () => { baseUrl: string; internalToken: string } | undefined; + /** 能力桥扩展文件路径(默认取仓库里的 server/agent/pi/extension/aps-bridge.ts) */ + extensionPath?: string; + /** 桥配置文件目录(默认 ~/.aps/pi-bridge,内含内部令牌,不入库) */ + bridgeDir?: string; } /** provider → 环境变量名:密钥走环境,不进命令行(命令行会被进程列表看到) */ @@ -124,14 +134,22 @@ export function mapPiEvent(event: JsonAgentSessionEvent): RuntimeEvent[] { } } +/** 默认扩展路径:从编译产物反推仓库里的源文件(Pi 用 jiti 直接读 TS) */ +function defaultExtensionPath(): string { + return fileURLToPath(new URL('../../../agent/pi/extension/aps-bridge.ts', import.meta.url)); +} + export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime { const provider = options.provider ?? 'deepseek'; const model = options.model ?? 'deepseek-chat'; const sessionDir = options.sessionDir ?? join(homedir(), '.aps', 'pi-sessions'); + const bridgeDir = options.bridgeDir ?? join(homedir(), '.aps', 'pi-bridge'); const turnTimeoutMs = options.turnTimeoutMs ?? DEFAULT_TURN_TIMEOUT_MS; return { name: `pi:${provider}/${model}`, + // 工具在 Pi 子进程里执行(经扩展桥回调平台的内部端点),平台只观察事件 + toolExecution: 'runtime', async createSession(input: CreateRuntimeSessionInput): Promise { const env: Record = {}; for (const [key, value] of Object.entries(process.env)) { @@ -141,6 +159,26 @@ export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime { env[API_KEY_ENV[provider] ?? 'DEEPSEEK_API_KEY'] = options.apiKey; } + // 能力桥:写一份会话专属配置给扩展读(含内部令牌,放用户目录、不入库) + const bridge = options.bridge?.(); + if (bridge) { + mkdirSync(bridgeDir, { recursive: true }); + const bridgeConfigPath = join(bridgeDir, `${input.sessionId}.json`); + const bridgeConfig = { + baseUrl: bridge.baseUrl, + internalToken: bridge.internalToken, + sessionId: input.sessionId, + capabilities: input.tools.map((tool) => ({ + toolName: tool.toolName, + capabilityId: tool.capabilityId, + description: tool.description, + parameters: tool.parameters, + })), + }; + writeFileSync(bridgeConfigPath, JSON.stringify(bridgeConfig, null, 2), { mode: 0o600 }); + env.APS_BRIDGE_CONFIG = bridgeConfigPath; + } + const cliPath = options.cliPath ?? defaultCliPath(); const client = new RpcClient({ ...(cliPath === '' ? {} : { cliPath }), @@ -155,6 +193,9 @@ export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime { '--no-skills', '--no-prompt-templates', '--no-context-files', + // 只加载平台自己的能力桥扩展(--no-extensions 关掉自动发现,-e 显式加载仍然有效) + '-e', + options.extensionPath ?? defaultExtensionPath(), '--session-dir', sessionDir, '--session-id', @@ -172,6 +213,7 @@ export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime { return { id: `pi-${input.sessionId}`, + toolExecution: 'runtime', async *send(text: string, sendOptions?: SendOptions): AsyncIterable { const queue: RuntimeEvent[] = []; let wake: (() => void) | undefined; @@ -225,4 +267,4 @@ export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime { }; }, }; -} \ No newline at end of file +} diff --git a/server/agent/runtime.ts b/server/agent/runtime.ts index 5a7ab56..f94754f 100644 --- a/server/agent/runtime.ts +++ b/server/agent/runtime.ts @@ -77,6 +77,8 @@ export interface SendOptions { export interface RuntimeSession { readonly id: string; + /** 与运行时的 toolExecution 一致:决定工具由平台执行还是引擎自己执行 */ + readonly toolExecution: 'platform' | 'runtime'; /** 发一轮输入,产出事件流;流结束表示这一回合结束 */ send(text: string, options?: SendOptions): AsyncIterable; /** 工具执行完回填回执,运行时据此继续生成后续事件 */ @@ -89,5 +91,11 @@ export interface RuntimeSession { export interface AgentRuntime { readonly name: string; + /** + * 工具由谁执行: + * platform —— 运行时只发 tool_call,平台经 Core.invoke 执行后把回执送回来(假引擎走这条) + * runtime —— 工具在引擎子进程里执行(Pi 走扩展桥,扩展自己回调平台的内部端点) + */ + readonly toolExecution: 'platform' | 'runtime'; createSession(input: CreateRuntimeSessionInput): Promise; } \ No newline at end of file diff --git a/server/app/turn-runner.ts b/server/app/turn-runner.ts index 3365d7e..475573c 100644 --- a/server/app/turn-runner.ts +++ b/server/app/turn-runner.ts @@ -61,27 +61,30 @@ export async function runTurn(input: TurnRunnerInput): Promise { 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 }, - }); + // 平台模式:这里执行并回填;引擎模式(Pi + 扩展桥)由扩展自己回调平台执行,平台只观察 + if (runtimeSession.toolExecution === 'platform') { + // 幂等键用 callId:同一回合重放不会重复执行 + const receipt = await invoke(event.capabilityId, event.args, `call-${event.callId}`); + 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': { - // 运行时自行上报的回执(真引擎可能直接给),不重复发事件 + // tool.result 只有这一个出处:运行时上报的回执(平台模式由假引擎回显,引擎模式由 Pi 上报) + publish({ + sessionId, + runId, + type: 'tool.result', + data: { callId: event.callId, executionId: event.receipt.executionId, status: event.receipt.status }, + }); break; } case 'approval_required': { diff --git a/server/index.ts b/server/index.ts index 1e5495a..93ee556 100644 --- a/server/index.ts +++ b/server/index.ts @@ -43,7 +43,10 @@ export async function startServer(options: StartOptions = {}): Promise bridgeRef.info)); const database: Database = openDatabase(options.databasePath ?? config.database.path); const service = new ApplicationService({ ...(options.version === undefined ? {} : { version: options.version }), @@ -56,7 +59,6 @@ export async function startServer(options: StartOptions = {}): Promise((resolve, reject) => { @@ -66,6 +68,8 @@ export async function startServer(options: StartOptions = {}): Promise 配置里的 engine.kind;选 pi 但没密钥时明确回退到假引擎 */ -async function selectRuntime(config: ReturnType, override?: 'fake' | 'pi'): Promise { +async function selectRuntime( + config: ReturnType, + override?: 'fake' | 'pi', + bridge?: () => { baseUrl: string; internalToken: string } | undefined, +): Promise { const kind = override ?? config.engine.kind; if (kind !== 'pi') { const { createFakeRuntime } = await import('./agent/fake-runtime.js'); @@ -97,5 +105,10 @@ async function selectRuntime(config: ReturnType, override?: ' return createFakeRuntime({ name: 'fake', planner: plannerToFakeTurn(createRulePlanner()) }); } const { createPiRuntime } = await import('./agent/pi/runtime.js'); - return createPiRuntime({ provider: config.llm.provider, model: config.llm.model, apiKey: config.llm.apiKey }); -} \ No newline at end of file + return createPiRuntime({ + provider: config.llm.provider, + model: config.llm.model, + apiKey: config.llm.apiKey, + ...(bridge === undefined ? {} : { bridge }), + }); +} diff --git a/server/package.json b/server/package.json index b156414..036128f 100644 --- a/server/package.json +++ b/server/package.json @@ -10,6 +10,10 @@ "types": "./dist/index.d.ts", "default": "./dist/index.js" }, + "./app/turn-runner": { + "types": "./dist/app/turn-runner.d.ts", + "default": "./dist/app/turn-runner.js" + }, "./config": { "types": "./dist/config.d.ts", "default": "./dist/config.js"