diff --git a/packages/codingcode/src/approval/tool-policy.ts b/packages/codingcode/src/approval/tool-policy.ts index a19daacc..8f008182 100644 --- a/packages/codingcode/src/approval/tool-policy.ts +++ b/packages/codingcode/src/approval/tool-policy.ts @@ -6,6 +6,7 @@ export const PLAN_TOOL_NAMES: readonly string[] = [ 'search_code', 'todo_write', 'todo_read', + 'ask_user', 'fetch_url', 'submit_plan', ]; @@ -21,6 +22,7 @@ export const BUILD_TOOL_NAMES: readonly string[] = [ 'web_search', 'todo_write', 'todo_read', + 'ask_user', 'spawn_agent', 'wait_agent', ]; diff --git a/packages/codingcode/src/askuser/port.ts b/packages/codingcode/src/askuser/port.ts new file mode 100644 index 00000000..f620d106 --- /dev/null +++ b/packages/codingcode/src/askuser/port.ts @@ -0,0 +1,19 @@ +import { Context } from 'effect'; +import type { Effect } from 'effect'; + +export type AskAnswer = + | { readonly status: 'answered'; readonly options: readonly string[]; readonly custom: string | null } + | { readonly status: 'no_answer' }; + +export interface AskWaitShape { + /** 挂起等待该调用 id 的用户回答 */ + waitForAsk(id: string, sessionId: string): Effect.Effect; + /** 用户回答回流；id + sessionId 双键校验，未命中返回 false */ + resolveAsk(id: string, sessionId: string, answer: AskAnswer): Effect.Effect; + /** 会话收尾时按 sessionId 关掉所有待答问题（fail-closed → no_answer），返回清理条数 */ + cancelPendingFor(sessionId: string): Effect.Effect; + /** 挂起方被打断时清掉等待条目（正常完成时条目已被 resolveAsk 删除，此处为幂等 no-op） */ + forget(id: string): Effect.Effect; +} + +export class AskWaitService extends Context.Tag('AskWait')() {} diff --git a/packages/codingcode/src/askuser/wait.ts b/packages/codingcode/src/askuser/wait.ts new file mode 100644 index 00000000..f4769aaa --- /dev/null +++ b/packages/codingcode/src/askuser/wait.ts @@ -0,0 +1,51 @@ +import { Layer, Effect, Deferred } from 'effect'; +import type { AskAnswer } from './port.js'; +import { AskWaitService } from './port.js'; + +interface PendingEntry { + deferred: Deferred.Deferred; + sessionId: string; +} + +export const AskWaitLayer = Layer.effect( + AskWaitService, + Effect.sync(() => { + const pendingAsks = new Map(); + + return { + waitForAsk: (id: string, sessionId: string): Effect.Effect => + Effect.gen(function* () { + const d = yield* Deferred.make(); + pendingAsks.set(id, { deferred: d, sessionId }); + return yield* Deferred.await(d); + }), + + resolveAsk: (id: string, sessionId: string, answer: AskAnswer): Effect.Effect => + Effect.sync(() => { + const entry = pendingAsks.get(id); + if (!entry || entry.sessionId !== sessionId) return false; + pendingAsks.delete(id); + Deferred.unsafeDone(entry.deferred, Effect.succeed(answer)); + return true; + }), + + cancelPendingFor: (sessionId: string): Effect.Effect => + Effect.sync(() => { + let cleared = 0; + for (const [id, entry] of pendingAsks) { + if (entry.sessionId !== sessionId) continue; + pendingAsks.delete(id); + Deferred.unsafeDone( + entry.deferred, + Effect.succeed({ status: 'no_answer' } as AskAnswer) + ); + cleared++; + } + return cleared; + }), + + forget: (id: string): Effect.Effect => + Effect.sync(() => pendingAsks.delete(id)), + }; + }) +); diff --git a/packages/codingcode/src/layer.ts b/packages/codingcode/src/layer.ts index 08657a62..23f8b356 100644 --- a/packages/codingcode/src/layer.ts +++ b/packages/codingcode/src/layer.ts @@ -9,6 +9,7 @@ import { ApprovalLayer } from './approval/approval.js'; import { ApprovalWaitLayer } from './approval/wait.js'; import { EventSinkLayer } from './sink/sink.js'; import { TodoLayer } from './todo/todo.js'; +import { AskWaitLayer } from './askuser/wait.js'; import { SessionLayer } from './session/session.js'; import { ToolExecutorLayer } from './tools/tools.js'; import { ContextLayer } from './context/context.js'; @@ -29,6 +30,7 @@ const InfraLayer = Layer.mergeAll( McpLayer, EventSinkLayer, ApprovalWaitLayer, + AskWaitLayer, TodoLayer ); diff --git a/packages/codingcode/src/server/app.ts b/packages/codingcode/src/server/app.ts index 8a423e50..340eae38 100644 --- a/packages/codingcode/src/server/app.ts +++ b/packages/codingcode/src/server/app.ts @@ -8,6 +8,7 @@ import { addMessagesRoutes } from './routes/messages.js'; import { addAssetsRoutes } from './routes/assets.js'; import { addModelsRoutes } from './routes/models.js'; import { addApprovalRoutes } from './routes/approval.js'; +import { addAskUserRoutes } from './routes/askuser.js'; import { addSettingsRoutes } from './routes/settings.js'; import { addAutomationsRoutes } from './routes/automations.js'; import { addSubagentsRoutes } from './routes/subagents.js'; @@ -28,6 +29,7 @@ export const buildRouter = (): HttpRouter.HttpRouter => addAssetsRoutes, addModelsRoutes, addApprovalRoutes, + addAskUserRoutes, addSettingsRoutes, addAutomationsRoutes, addSubagentsRoutes diff --git a/packages/codingcode/src/server/handler.ts b/packages/codingcode/src/server/handler.ts index 744a247f..e34a7942 100644 --- a/packages/codingcode/src/server/handler.ts +++ b/packages/codingcode/src/server/handler.ts @@ -4,6 +4,7 @@ import * as HttpServerResponse from '@effect/platform/HttpServerResponse'; import { Effect, Stream } from 'effect'; import { AgentError } from '../util/error.js'; import { ApprovalWaitService } from '../approval/wait-port.js'; +import { AskWaitService } from '../askuser/port.js'; import type { FrameBody } from '../sink/types.js'; import { createFrameAssembler, encodeFrame } from './frame-io.js'; import { errorCodeOf, isAppError, type AppError } from './http-error.js'; @@ -80,9 +81,10 @@ export const readJsonFrom = (web: Request): Effect.Effect => export const sseResponse = ( frames: Stream.Stream, opts: { sessionId: string; onDone?: () => void } -): Effect.Effect => +): Effect.Effect => Effect.gen(function* () { const wait = yield* ApprovalWaitService; + const askWait = yield* AskWaitService; const assembler = createFrameAssembler({ sessionId: opts.sessionId }); const encoder = new TextEncoder(); const body = frames.pipe( @@ -90,6 +92,7 @@ export const sseResponse = ( Stream.ensuring( Effect.sync(() => { Effect.runSync(wait.cancelPendingFor(opts.sessionId)); + Effect.runSync(askWait.cancelPendingFor(opts.sessionId)); opts.onDone?.(); }) ) diff --git a/packages/codingcode/src/server/routes/askuser.ts b/packages/codingcode/src/server/routes/askuser.ts new file mode 100644 index 00000000..e6eb7b85 --- /dev/null +++ b/packages/codingcode/src/server/routes/askuser.ts @@ -0,0 +1,25 @@ +import * as HttpRouter from '@effect/platform/HttpRouter'; +import { Effect } from 'effect'; +import { AskWaitService, type AskAnswer } from '../../askuser/port.js'; +import { json, pathParams, readJsonOrEmpty, type Handler, type Router } from '../handler.js'; + +/** 严格解析回答 body；无法构成有效回答时 fail-closed 为 no_answer */ +function parseAskBody(body: { options?: unknown; custom?: unknown }): AskAnswer { + const custom = typeof body.custom === 'string' ? body.custom : ''; + const raw = Array.isArray(body.options) ? body.options : []; + const options = raw.filter((item): item is string => typeof item === 'string' && /^[A-D]$/.test(item)); + if (options.length === 0 && custom.trim() === '') return { status: 'no_answer' }; + return { status: 'answered', options, custom: custom.trim() || null }; +} + +const resolve: Handler = Effect.gen(function* () { + const { id, sessionId } = yield* pathParams; + const body = yield* readJsonOrEmpty<{ options?: unknown; custom?: unknown }>(); + + const svc = yield* AskWaitService; + const resolved = yield* svc.resolveAsk(id ?? '', sessionId ?? '', parseAskBody(body)); + return json({ ok: resolved }); +}); + +export const addAskUserRoutes = (router: Router): Router => + router.pipe(HttpRouter.post('/api/sessions/:sessionId/askuser/:id', resolve)); diff --git a/packages/codingcode/src/sink/types.ts b/packages/codingcode/src/sink/types.ts index 7f452639..3db09a15 100644 --- a/packages/codingcode/src/sink/types.ts +++ b/packages/codingcode/src/sink/types.ts @@ -48,6 +48,13 @@ export type RuntimeEvent = readonly tool: string; readonly args: Readonly>; } + | { + readonly type: 'ask_user'; + readonly id: string; + readonly question: string; + readonly questiontype: 'single' | 'multiple' | 'text'; + readonly options?: readonly string[]; + } | { readonly type: 'subagent_event'; readonly sessionId: string; diff --git a/packages/codingcode/src/tools/catalog.ts b/packages/codingcode/src/tools/catalog.ts index d29ccada..0c5ff3ca 100644 --- a/packages/codingcode/src/tools/catalog.ts +++ b/packages/codingcode/src/tools/catalog.ts @@ -14,6 +14,7 @@ import { webFetchTool } from './domains/web/fetch.js'; import { webSearchTool } from './domains/web/search.js'; import { todoWriteTool } from './domains/self/todo-write.js'; import { todoReadTool } from './domains/self/todo-read.js'; +import { askUserTool } from './domains/self/ask-user.js'; import { spawnAgentTool } from './domains/subagent/spawn.js'; import { waitAgentTool } from './domains/subagent/wait.js'; import { submitPlanTool } from './domains/subagent/submit-plan.js'; @@ -29,6 +30,7 @@ const ALL_TOOLS: ToolDefinition[] = [ webSearchTool, todoWriteTool, todoReadTool, + askUserTool, spawnAgentTool, waitAgentTool, submitPlanTool, diff --git a/packages/codingcode/src/tools/domains/self/ask-user.ts b/packages/codingcode/src/tools/domains/self/ask-user.ts new file mode 100644 index 00000000..ab901ade --- /dev/null +++ b/packages/codingcode/src/tools/domains/self/ask-user.ts @@ -0,0 +1,111 @@ +import { z } from 'zod'; +import { Effect } from 'effect'; +import type { ToolDefinition } from '../../types.js'; +import { AgentError } from '../../../util/error.js'; +import { EventSinkService } from '../../../sink/port.js'; +import { AskWaitService, type AskAnswer } from '../../../askuser/port.js'; + +type AskQuestionType = 'single' | 'multiple' | 'text'; + +const askUserSchema = z.object({ + questiontype: z.enum(['single', 'multiple', 'text']), + question: z.string().min(1).max(500), + options: z.array(z.string().min(1).max(200)).max(4).optional(), +}); + +const NO_ANSWER_OUTPUT = '用户未回答（用户取消或连接中断）'; + +function trimOrNull(v: string | null | undefined): string | null { + const t = v?.trim(); + return t ? t : null; +} + +function formatOutput( + questiontype: AskQuestionType, + options: readonly string[], + answer: AskAnswer +): string | undefined { + if (answer.status === 'no_answer') return undefined; + + if (questiontype === 'text') { + const text = trimOrNull(answer.custom); + return text ? `用户回答：${text}` : undefined; + } + + const picked = answer.options + .filter((letter) => /^[A-D]$/.test(letter)) + .map((letter) => { + const index = letter.charCodeAt(0) - 65; + return options[index] !== undefined ? `${letter}:${options[index]}` : null; + }) + .filter((item): item is string => item !== null); + const custom = trimOrNull(answer.custom); + if (picked.length === 0 && custom === null) return undefined; + + const items = picked.length > 0 ? picked : [`自定义：${custom}`]; + let out = `用户选择了：[${items.map((item) => JSON.stringify(item)).join(',')}]`; + if (picked.length > 0 && custom !== null) out += `；用户补充输入：${custom}`; + return out; +} + +export const askUserTool: ToolDefinition = { + name: 'ask_user', + concurrencySafe: true, + description: + 'Ask the end user a structured question and block until they answer. Use when missing information ' + + 'blocks the correct next action. questiontype: single (radio), multiple (checkbox), text (free input). ' + + 'options: 2-4 choice texts for single/multiple, ignored for text; letters A-D are assigned by order ' + + 'automatically. The user may also type a custom answer besides the options. Returns e.g. ' + + '用户选择了：["A:红色"] or 用户回答：.... Multiple ask_user calls may be batched, each blocks on its ' + + 'own answer. If it returns 用户未回答, proceed with reasonable assumptions, state them explicitly, ' + + 'and re-ask at most once.', + parameters: askUserSchema, + execute: (args, ctx) => + Effect.gen(function* () { + const { questiontype, question, options } = args as { + questiontype: AskQuestionType; + question: string; + options?: string[]; + }; + + if (questiontype !== 'text' && (!options || options.length < 2)) { + return yield* Effect.fail( + new AgentError('TOOL_EXECUTION_FAILED', 'ask_user choice questions require 2-4 options') + ); + } + + const sessionId = ctx?.sessionId; + if (!sessionId) { + return yield* Effect.fail( + new AgentError('TOOL_EXECUTION_FAILED', 'ask_user requires sessionId') + ); + } + const callId = ctx?.callId; + if (!callId) { + return yield* Effect.fail( + new AgentError('TOOL_EXECUTION_FAILED', 'ask_user requires callId') + ); + } + + const waitSvc = yield* AskWaitService; + const sink = yield* EventSinkService; + + yield* sink.emit(sessionId, { + family: 'event', + event: { + type: 'ask_user', + id: callId, + question, + questiontype, + options: questiontype === 'text' ? undefined : options, + }, + }); + + const answer = yield* waitSvc.waitForAsk(callId, sessionId).pipe( + Effect.ensuring(waitSvc.forget(callId)) + ); + + const choices = questiontype === 'text' ? [] : (options ?? []); + return formatOutput(questiontype, choices, answer) ?? NO_ANSWER_OUTPUT; + }), +}; diff --git a/packages/codingcode/src/tools/tools.ts b/packages/codingcode/src/tools/tools.ts index 4c96a52e..b62d0387 100644 --- a/packages/codingcode/src/tools/tools.ts +++ b/packages/codingcode/src/tools/tools.ts @@ -18,7 +18,7 @@ import { createToolCatalog } from './catalog.js'; type Lookup = (name: string) => ToolRunner | undefined; function toToolCtx(opts: ToolExecCall): ToolExecCtx { - const { turnId: _turnId, callId: _callId, ...ctx } = opts; + const { turnId: _turnId, ...ctx } = opts; return ctx; } diff --git a/packages/codingcode/src/tools/types.ts b/packages/codingcode/src/tools/types.ts index a932db62..869427a0 100644 --- a/packages/codingcode/src/tools/types.ts +++ b/packages/codingcode/src/tools/types.ts @@ -20,6 +20,8 @@ export interface ToolDefinition { export interface ToolExecCtx { signal?: AbortSignal; sessionId?: string; + /** 本次调用的 tool_call id（由 execSingle 以 tc.id 注入）；ask_user 用作等待键 */ + callId?: string; projectPath?: string; activeProfile?: ProfileName; model: string; @@ -37,7 +39,7 @@ export interface ToolRunner { export type ToolLookup = (name: string) => ToolRunner | undefined; -export type ToolExecOpts = Omit & { +export type ToolExecOpts = Omit & { turnId?: number; onBeforeMutation?: Effect.Effect; }; diff --git a/packages/codingcode/test/server/askuser-route.test.ts b/packages/codingcode/test/server/askuser-route.test.ts new file mode 100644 index 00000000..53812960 --- /dev/null +++ b/packages/codingcode/test/server/askuser-route.test.ts @@ -0,0 +1,117 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import { Effect, Fiber, ManagedRuntime } from 'effect'; +import { createServer } from '../../src/server/index.js'; +import { AskWaitService, type AskAnswer } from '../../src/askuser/port.js'; +import { AskWaitLayer } from '../../src/askuser/wait.js'; + +// askuser 路由只在请求期依赖 AskWaitService（见 sse.test.ts：createServer 不在装配期索取全部服务） +const rt = ManagedRuntime.make(AskWaitLayer); + +beforeEach(async () => { + // 清掉上一例可能残留的待答条目，保证测试隔离 + const svc = await rt.runPromise(AskWaitService); + await rt.runPromise(svc.cancelPendingFor('sess-1')); +}); + +async function postAsk(sessionId: string, id: string, body: unknown): Promise<{ ok: boolean }> { + const app = await createServer(rt); + const res = await app.request(`/api/sessions/${sessionId}/askuser/${id}`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + }); + expect(res.status).toBe(200); + return (await res.json()) as { ok: boolean }; +} + +/** 注册一个等待者并返回读取其收到答案的 Effect；runFork 同步执行到 Deferred.await，返回即已注册 */ +function waitRegistered(svc: Awaited>, id: string, sessionId: string) { + const fiber = rt.runFork(svc.waitForAsk(id, sessionId)); + return Effect.map(Fiber.join(fiber), (answer) => answer); +} + +const getService = () => rt.runPromise(AskWaitService); + +describe('POST /api/sessions/:sessionId/askuser/:id', () => { + it('合法 body {options:[A], custom} → answered 且 custom 去空白', async () => { + const svc = await getService(); + const waiter = waitRegistered(svc, 'tc-1', 'sess-1'); + const res = await postAsk('sess-1', 'tc-1', { options: ['A'], custom: ' 备注 ' }); + expect(res).toEqual({ ok: true }); + + const answer = await rt.runPromise(waiter); + expect(answer).toEqual({ status: 'answered', options: ['A'], custom: '备注' }); + }); + + it('仅 custom → answered(options 为空数组)', async () => { + const svc = await getService(); + const waiter = waitRegistered(svc, 'tc-2', 'sess-1'); + await postAsk('sess-1', 'tc-2', { custom: '其他颜色' }); + + const answer = await rt.runPromise(waiter); + expect(answer).toEqual({ status: 'answered', options: [], custom: '其他颜色' }); + }); + + it('多字母选项保留顺序 {options:[B,A]}', async () => { + const svc = await getService(); + const waiter = waitRegistered(svc, 'tc-3', 'sess-1'); + await postAsk('sess-1', 'tc-3', { options: ['B', 'A'] }); + + const answer = await rt.runPromise(waiter); + expect(answer).toEqual({ status: 'answered', options: ['B', 'A'], custom: null }); + }); + + it('空 body {} → fail-closed 为 no_answer', async () => { + const svc = await getService(); + const waiter = waitRegistered(svc, 'tc-4', 'sess-1'); + await postAsk('sess-1', 'tc-4', {}); + + const answer = await rt.runPromise(waiter); + expect(answer).toEqual({ status: 'no_answer' }); + }); + + it('非法选项被过滤（非字符串 / 越界字母）+ 空白 custom → no_answer', async () => { + const svc = await getService(); + const waiter = waitRegistered(svc, 'tc-5', 'sess-1'); + await postAsk('sess-1', 'tc-5', { options: ['a', 42, 'Z', null], custom: ' ' }); + + const answer = await rt.runPromise(waiter); + expect(answer).toEqual({ status: 'no_answer' }); + }); + + it('未知 id → ok:false', async () => { + const res = await postAsk('sess-1', 'no-such-id', { options: ['A'] }); + expect(res).toEqual({ ok: false }); + }); + + it('跨会话 resolve 被拒绝 → ok:false 且原等待者不受污染', async () => { + const svc = await getService(); + const waiter = waitRegistered(svc, 'tc-6', 'sess-1'); + + const res = await postAsk('sess-other', 'tc-6', { options: ['A'] }); + expect(res).toEqual({ ok: false }); + + // 原等待者仍未被解决——用正确会话补一发 resolve 验证条目完好 + const ok = await rt.runPromise( + svc.resolveAsk('tc-6', 'sess-1', { status: 'answered', options: ['C'], custom: null }) + ); + expect(ok).toBe(true); + const answer: AskAnswer = await rt.runPromise(waiter); + expect(answer).toEqual({ status: 'answered', options: ['C'], custom: null }); + }); + + it('malformed JSON body → fail-closed no_answer（回合不被毒死）', async () => { + const svc = await getService(); + const waiter = waitRegistered(svc, 'tc-7', 'sess-1'); + const app = await createServer(rt); + const res = await app.request('/api/sessions/sess-1/askuser/tc-7', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: 'not-json{', + }); + expect(res.status).toBe(200); + + const answer = await rt.runPromise(waiter); + expect(answer).toEqual({ status: 'no_answer' }); + }); +}); diff --git a/packages/codingcode/test/server/sse.test.ts b/packages/codingcode/test/server/sse.test.ts index c788c232..6913141b 100644 --- a/packages/codingcode/test/server/sse.test.ts +++ b/packages/codingcode/test/server/sse.test.ts @@ -12,6 +12,7 @@ import { join } from 'path'; import { createServer, type ServerApp } from '../../src/server/index.js'; import { AgentService } from '../../src/agent/port.js'; import { ApprovalWaitService } from '../../src/approval/wait-port.js'; +import { AskWaitService } from '../../src/askuser/port.js'; import { AgentError } from '../../src/util/error.js'; import type { FrameBody } from '../../src/sink/types.js'; import type { IncomingPart } from '../../src/llm/types.js'; @@ -20,6 +21,7 @@ const TEXT_FRAME: FrameBody = { family: 'event', event: { type: 'text_delta', te let cwd: string; let cancelled: string[] = []; +let askCancelled: string[] = []; function makeServer(frames: () => AsyncGenerator): Promise { const layer = Layer.mergeAll( @@ -38,6 +40,16 @@ function makeServer(frames: () => AsyncGenerator): Pro }), resolveConfirm: () => Effect.succeed(false), waitForConfirm: () => Effect.dieMessage('not implemented'), + } as any), + Layer.succeed(AskWaitService, { + cancelPendingFor: (sessionId: string) => + Effect.sync(() => { + askCancelled.push(sessionId); + return 0; + }), + resolveAsk: () => Effect.succeed(false), + waitForAsk: () => Effect.dieMessage('not implemented'), + forget: () => Effect.succeed(false), } as any) ); return createServer(ManagedRuntime.make(layer)); @@ -60,6 +72,7 @@ const decode = (body: string) => beforeEach(() => { cwd = mkdtempSync(join(tmpdir(), 'codingcode-sse-')); cancelled = []; + askCancelled = []; }); afterEach(() => { @@ -103,6 +116,7 @@ describe('SSE /api/sessions/:id/messages', () => { }); it('领域错误（AgentError）带出自己的 code —— 不是笼统的 INTERNAL_ERROR', async () => { + // eslint-disable-next-line require-yield -- 抛错流正是本用例要覆盖的形状 const app = await makeServer(async function* () { throw new AgentError('SESSION_CORRUPTED', 'nope'); }); @@ -111,10 +125,10 @@ describe('SSE /api/sessions/:id/messages', () => { expect(frames[0].fatal.code).toBe('SESSION_CORRUPTED'); }); - it('流结束时取消该会话的挂起审批', async () => { - // eslint-disable-next-line require-yield -- 空流正是本用例要覆盖的形状 + it('流结束时取消该会话的挂起审批与待答问题', async () => { const app = await makeServer(async function* () {}); await (await send(app, 'sess-9')).text(); expect(cancelled).toEqual(['sess-9']); + expect(askCancelled).toEqual(['sess-9']); }); }); diff --git a/packages/codingcode/test/tools/ask-user.test.ts b/packages/codingcode/test/tools/ask-user.test.ts new file mode 100644 index 00000000..58f438b7 --- /dev/null +++ b/packages/codingcode/test/tools/ask-user.test.ts @@ -0,0 +1,429 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import { Effect, Exit, Fiber, Layer, ManagedRuntime } from 'effect'; +import { askUserTool } from '../../src/tools/domains/self/ask-user.js'; +import { AskWaitService, type AskAnswer } from '../../src/askuser/port.js'; +import { AskWaitLayer } from '../../src/askuser/wait.js'; +import { EventSinkService } from '../../src/sink/port.js'; +import type { FrameBody } from '../../src/sink/types.js'; + +// ---------- 测试环境：真实 AskWaitLayer + 捕获型 EventSink mock ---------- + +const frames: FrameBody[] = []; + +const SinkMockLayer = Layer.succeed(EventSinkService, { + detach: () => Effect.void, + emit: (_sessionId: string, body: FrameBody) => + Effect.sync(() => { + frames.push(body); + }), +} as any); + +const rt = ManagedRuntime.make(Layer.mergeAll(AskWaitLayer, SinkMockLayer)); + +beforeEach(() => { + frames.length = 0; +}); + +const CTX = { sessionId: 'sess-1', callId: 'tc-1', model: 'test-model' } as const; + +/** 起一个 execute fiber，并以重试方式 resolve（emit 之后 waitForAsk 注册之间存在竞态） */ +function askAndResolve( + args: unknown, + ctx: { sessionId: string; callId: string; model: string }, + answer: AskAnswer +) { + return Effect.gen(function* () { + const svc = yield* AskWaitService; + const fiber = yield* Effect.fork(askUserTool.execute(args, ctx)); + let resolved = false; + for (let i = 0; i < 500 && !resolved; i++) { + resolved = yield* svc.resolveAsk(ctx.callId, ctx.sessionId, answer); + if (!resolved) yield* Effect.sleep(2); + } + if (!resolved) throw new Error('ask entry never registered'); + return yield* Fiber.join(fiber); + }); +} + +// ---------- 1. 参数校验 ---------- + +describe('ask_user 参数校验', () => { + it('schema 拒绝非法 questiontype', async () => { + await expect( + askUserTool.parameters.parseAsync({ questiontype: 'checkbox', question: 'q' }) + ).rejects.toThrow(); + }); + + it('schema 拒绝空 question', async () => { + await expect( + askUserTool.parameters.parseAsync({ questiontype: 'text', question: '' }) + ).rejects.toThrow(); + }); + + it('schema 拒绝超过 4 个选项', async () => { + await expect( + askUserTool.parameters.parseAsync({ + questiontype: 'single', + question: 'q', + options: ['1', '2', '3', '4', '5'], + }) + ).rejects.toThrow(); + }); + + it('schema 拒绝空字符串选项', async () => { + await expect( + askUserTool.parameters.parseAsync({ questiontype: 'single', question: 'q', options: ['a', ''] }) + ).rejects.toThrow(); + }); + + it('single 缺 options → AgentError', async () => { + const err = await rt.runPromise( + askUserTool.execute({ questiontype: 'single', question: 'q' }, CTX).pipe(Effect.flip) + ); + expect(err.code).toBe('TOOL_EXECUTION_FAILED'); + }); + + it('multiple 仅 1 个选项 → AgentError', async () => { + const exit = await rt.runPromiseExit( + askUserTool.execute({ questiontype: 'multiple', question: 'q', options: ['唯一'] }, CTX) + ); + expect(Exit.isFailure(exit)).toBe(true); + }); + + it('text 携带 options 被忽略且正常工作', async () => { + const svc = await rt.runPromise(AskWaitService); + const fiber = rt.runFork( + askUserTool.execute( + { questiontype: 'text', question: '什么颜色', options: ['红', '蓝'] }, + CTX + ) + ); + await rt.runPromise(svc.resolveAsk('tc-1', 'sess-1', { status: 'answered', options: [], custom: '绿色' })); + const output = await rt.runPromise(Fiber.join(fiber)); + expect(output).toBe('用户回答：绿色'); + // 帧不带 options + const emitted = (frames[0] as { event: { options?: unknown } }).event; + expect(emitted.options).toBeUndefined(); + }); + + it('缺 sessionId → AgentError', async () => { + const exit = await rt.runPromiseExit( + askUserTool.execute( + { questiontype: 'single', question: 'q', options: ['a', 'b'] }, + { callId: 'tc-1', model: 'm' } + ) + ); + expect(Exit.isFailure(exit)).toBe(true); + }); + + it('缺 callId → AgentError', async () => { + const exit = await rt.runPromiseExit( + askUserTool.execute( + { questiontype: 'single', question: 'q', options: ['a', 'b'] }, + { sessionId: 's', model: 'm' } + ) + ); + expect(Exit.isFailure(exit)).toBe(true); + }); +}); + +// ---------- 2 & 3. 等待-回流（真实 AskWaitLayer）与输出格式化 ---------- + +describe('ask_user 等待-回流与 output 格式', () => { + it('单选：用户选择了：["A:红色"]', async () => { + const output = await rt.runPromise( + askAndResolve( + { questiontype: 'single', question: '什么颜色', options: ['红色', '蓝色'] }, + CTX, + { status: 'answered', options: ['A'], custom: null } + ) + ); + expect(output).toBe('用户选择了：["A:红色"]'); + }); + + it('多选：用户选择了：["A:红色","C:绿色"]', async () => { + const output = await rt.runPromise( + askAndResolve( + { questiontype: 'multiple', question: '要哪些', options: ['红色', '蓝色', '绿色'] }, + CTX, + { status: 'answered', options: ['A', 'C'], custom: null } + ) + ); + expect(output).toBe('用户选择了：["A:红色","C:绿色"]'); + }); + + it('仅自定义：用户选择了：["自定义：其他颜色"]', async () => { + const output = await rt.runPromise( + askAndResolve( + { questiontype: 'single', question: '什么颜色', options: ['红色', '蓝色'] }, + CTX, + { status: 'answered', options: [], custom: '其他颜色' } + ) + ); + expect(output).toBe('用户选择了：["自定义：其他颜色"]'); + }); + + it('选项+补充（多选；单选在前端与自定义输入互斥，不产生此形态）：["A:红色"]；补充', async () => { + const output = await rt.runPromise( + askAndResolve( + { questiontype: 'multiple', question: '什么颜色', options: ['红色', '蓝色'] }, + CTX, + { status: 'answered', options: ['A'], custom: '备注' } + ) + ); + expect(output).toBe('用户选择了：["A:红色"]；用户补充输入：备注'); + }); + + it('填空：用户回答：其他颜色', async () => { + const output = await rt.runPromise( + askAndResolve( + { questiontype: 'text', question: '什么颜色' }, + CTX, + { status: 'answered', options: [], custom: '其他颜色' } + ) + ); + expect(output).toBe('用户回答：其他颜色'); + }); + + it('非 A-D 的选项字母被过滤；越界字母不产出条目', async () => { + const output = await rt.runPromise( + askAndResolve( + { questiontype: 'single', question: '什么颜色', options: ['红色', '蓝色'] }, + CTX, + { status: 'answered', options: ['a', 'E', 'B'], custom: null } + ) + ); + // 'a' 小写不匹配 /E/ 越界均被滤掉，仅保留 'B' + expect(output).toBe('用户选择了：["B:蓝色"]'); + }); + + it('emit 的 ask_user 帧自包含且 id 即 callId', async () => { + await rt.runPromise( + askAndResolve( + { questiontype: 'single', question: '什么颜色', options: ['红色', '蓝色'] }, + CTX, + { status: 'answered', options: ['A'], custom: null } + ) + ); + expect(frames).toEqual([ + { + family: 'event', + event: { + type: 'ask_user', + id: 'tc-1', + question: '什么颜色', + questiontype: 'single', + options: ['红色', '蓝色'], + }, + }, + ]); + }); + + it('未 resolve 前一直挂起（回合不提前收场）', async () => { + const output = await rt.runPromise( + Effect.gen(function* () { + const svc = yield* AskWaitService; + const fiber = yield* Effect.fork( + askUserTool.execute( + { questiontype: 'single', question: 'q', options: ['a', 'b'] }, + CTX + ) + ); + // 帧已发出，但 50ms 内不 resolve → join 竞速应输给 sleep（即仍挂起） + const raced = yield* Effect.race( + Fiber.join(fiber), + Effect.sleep(50).pipe(Effect.as('still-hanging')) + ); + expect(raced).toBe('still-hanging'); + // 收尾：resolve 让 fiber 干净退出 + yield* svc.resolveAsk('tc-1', 'sess-1', { status: 'answered', options: ['A'], custom: null }); + return yield* Fiber.join(fiber); + }) + ); + expect(output).toBe('用户选择了：["A:a"]'); + }); + + it('resolveAsk 未知 id 返回 false', async () => { + const svc = await rt.runPromise(AskWaitService); + const ok = await rt.runPromise( + svc.resolveAsk('no-such-id', 'sess-1', { status: 'answered', options: ['A'], custom: null }) + ); + expect(ok).toBe(false); + }); + + it('跨 sessionId resolve 被拒绝', async () => { + const svc = await rt.runPromise(AskWaitService); + const fiber = rt.runFork( + askUserTool.execute( + { questiontype: 'single', question: 'q', options: ['a', 'b'] }, + CTX + ) + ); + await Effect.runPromise(Effect.sleep(20)); + const ok = await rt.runPromise( + svc.resolveAsk('tc-1', 'other-session', { status: 'answered', options: ['A'], custom: null }) + ); + expect(ok).toBe(false); + // 正确会话仍可 resolve（条目未被污染） + const ok2 = await rt.runPromise( + svc.resolveAsk('tc-1', 'sess-1', { status: 'answered', options: ['A'], custom: null }) + ); + expect(ok2).toBe(true); + expect(await rt.runPromise(Fiber.join(fiber))).toBe('用户选择了：["A:a"]'); + }); + + it('同批两个 ask_user 并发挂起、分别 resolve 互不干扰', async () => { + const output = await rt.runPromise( + Effect.gen(function* () { + const svc = yield* AskWaitService; + const fiber1 = yield* Effect.fork( + askUserTool.execute( + { questiontype: 'single', question: '颜色', options: ['红色', '蓝色'] }, + { ...CTX, callId: 'tc-1' } + ) + ); + const fiber2 = yield* Effect.fork( + askUserTool.execute( + { questiontype: 'text', question: '名字' }, + { ...CTX, callId: 'tc-2' } + ) + ); + // 两个条目都注册后，先解 tc-2 再解 tc-1（乱序） + for (let i = 0; i < 500; i++) { + const framesOk = frames.length === 2; + if (!framesOk) { + yield* Effect.sleep(2); + continue; + } + const r2 = yield* svc.resolveAsk('tc-2', 'sess-1', { + status: 'answered', + options: [], + custom: '张三', + }); + if (!r2) { + yield* Effect.sleep(2); + continue; + } + let r1 = false; + for (let j = 0; j < 500 && !r1; j++) { + r1 = yield* svc.resolveAsk('tc-1', 'sess-1', { + status: 'answered', + options: ['B'], + custom: null, + }); + if (!r1) yield* Effect.sleep(2); + } + if (!r1) throw new Error('tc-1 never registered'); + return yield* Effect.all([Fiber.join(fiber1), Fiber.join(fiber2)]); + } + throw new Error('frames never emitted'); + }) + ); + expect(output).toEqual([ + '用户选择了：["B:蓝色"]', + '用户回答：张三', + ]); + }); +}); + +// ---------- 4. fail-closed：取消 / 打断 ---------- + +describe('ask_user fail-closed 生命周期', () => { + it('cancelPendingFor 后挂起的 execute 以 ok + 未回答文案收场，回合不挂起', async () => { + const output = await rt.runPromise( + Effect.gen(function* () { + const svc = yield* AskWaitService; + const fiber = yield* Effect.fork( + askUserTool.execute( + { questiontype: 'single', question: 'q', options: ['a', 'b'] }, + CTX + ) + ); + // 等条目注册 + for (let i = 0; i < 500; i++) { + const cleared = yield* svc.cancelPendingFor('sess-1'); + if (cleared > 0) break; + yield* Effect.sleep(2); + } + return yield* Fiber.join(fiber); + }) + ); + expect(output).toBe('用户未回答（用户取消或连接中断）'); + }); + + it('cancelPendingFor 只清理本会话，返回清理条数', async () => { + const count = await rt.runPromise( + Effect.gen(function* () { + const svc = yield* AskWaitService; + yield* Effect.fork( + askUserTool.execute( + { questiontype: 'single', question: 'q', options: ['a', 'b'] }, + { ...CTX, callId: 'tc-a1', sessionId: 'sess-A' } + ) + ); + yield* Effect.fork( + askUserTool.execute( + { questiontype: 'single', question: 'q', options: ['a', 'b'] }, + { ...CTX, callId: 'tc-b1', sessionId: 'sess-B' } + ) + ); + yield* Effect.sleep(50); // 给两个条目留足注册时间 + return yield* svc.cancelPendingFor('sess-A'); + }) + ); + expect(count).toBe(1); + // 收尾：清掉 sess-B 的残留条目，不污染后续测试 + const svc = await rt.runPromise(AskWaitService); + await rt.runPromise(svc.cancelPendingFor('sess-B')); + }); + + it('fiber 被打断时 forget 清理等待条目，后续 resolve 落空', async () => { + const noDangling = await rt.runPromise( + Effect.gen(function* () { + const svc = yield* AskWaitService; + const fiber = yield* Effect.fork( + askUserTool.execute( + { questiontype: 'single', question: 'q', options: ['a', 'b'] }, + CTX + ) + ); + // 等条目注册（注册后帧已发出） + for (let i = 0; i < 500; i++) { + if (frames.length > 0) break; + yield* Effect.sleep(2); + } + yield* Effect.sleep(5); // waitForAsk 注册紧随 emit + yield* Fiber.interrupt(fiber); + return yield* svc.resolveAsk('tc-1', 'sess-1', { + status: 'answered', + options: ['A'], + custom: null, + }); + }) + ); + expect(noDangling).toBe(false); + }); + + it('正常完成后 forget 幂等（条目已被 resolveAsk 删除）', async () => { + const result = await rt.runPromise( + Effect.gen(function* () { + const svc = yield* AskWaitService; + const fiber = yield* Effect.fork(svc.waitForAsk('manual-1', 'sess-1')); + // 等注册完成再 resolve（复用重试模式） + let resolved = false; + for (let i = 0; i < 500 && !resolved; i++) { + resolved = yield* svc.resolveAsk('manual-1', 'sess-1', { status: 'no_answer' }); + if (!resolved) yield* Effect.sleep(2); + } + const forgot = yield* svc.forget('manual-1'); + const answer = yield* Fiber.join(fiber); + return { resolved, forgot, answer }; + }) + ); + expect(result).toEqual({ + resolved: true, + forgot: false, + answer: { status: 'no_answer' }, + }); + }); +}); diff --git a/packages/codingcode/test/tools/executor-concurrency.test.ts b/packages/codingcode/test/tools/executor-concurrency.test.ts index 39a2f3f5..8a59850a 100644 --- a/packages/codingcode/test/tools/executor-concurrency.test.ts +++ b/packages/codingcode/test/tools/executor-concurrency.test.ts @@ -417,7 +417,7 @@ describe('executeBatch 保序波次调度', () => { expect(h.hookPoints.filter((p) => p === 'tool.execute.before')).toHaveLength(1); }); - it('13 个内置工具的 concurrencySafe 与分类表逐项一致，且无未覆盖工具', () => { + it('14 个内置工具的 concurrencySafe 与分类表逐项一致，且无未覆盖工具', () => { const expected: Record = { read_file: true, search_code: true, @@ -426,6 +426,7 @@ describe('executeBatch 保序波次调度', () => { web_search: true, todo_write: true, todo_read: true, + ask_user: true, spawn_agent: true, wait_agent: true, write_file: false, diff --git a/packages/codingcode/test/tools/executor-ctx-projection.test.ts b/packages/codingcode/test/tools/executor-ctx-projection.test.ts index abd9e2d9..2a1d7ddb 100644 --- a/packages/codingcode/test/tools/executor-ctx-projection.test.ts +++ b/packages/codingcode/test/tools/executor-ctx-projection.test.ts @@ -70,24 +70,26 @@ describe('工具执行上下文投影', () => { // 精确断言键集合：既防漏抄，也防多塞 expect(Object.keys(captured!).sort()).toEqual([ 'activeProfile', + 'callId', 'model', 'projectPath', 'sessionId', 'signal', ]); expect(captured!.sessionId).toBe('sess-1'); + expect(captured!.callId).toBe('c1'); expect(captured!.projectPath).toBe('/proj'); expect(captured!.activeProfile).toBe('plan'); expect(captured!.model).toBe('m-1'); expect(captured!.signal).toBe(signal); }); - it('编排私有字段一个不漏：turnId / callId 到不了工具', async () => { + it('turnId 到不了工具，callId 以 tc.id 透传', async () => { await runProbe({ turnId: 7, projectPath: '/proj', model: 'm-1' }, 'sess-1'); expect(captured).toBeDefined(); expect('turnId' in captured!).toBe(false); - expect('callId' in captured!).toBe(false); + expect(captured!.callId).toBe('c1'); }); it('sessionId 缺省时不伪造该字段', async () => { diff --git a/packages/codingcode/test/types/tool-exec-shapes.test.ts b/packages/codingcode/test/types/tool-exec-shapes.test.ts index ca73f66e..c8bac14c 100644 --- a/packages/codingcode/test/types/tool-exec-shapes.test.ts +++ b/packages/codingcode/test/types/tool-exec-shapes.test.ts @@ -35,13 +35,21 @@ describe('工具执行参数的形状收口', () => { void bad; }); + it('callId 不进 ToolExecOpts —— 它由 execSingle 按 tc.id 注入', () => { + // @ts-expect-error callId 由 execSingle 注入，不放进批次选项 + const bad: ToolExecOpts = { model: 'm', callId: 'c-1' }; + void bad; + }); + it('编排私有字段不外泄到工具可见的 ctx', () => { // @ts-expect-error turnId 只进 hook payload，不属于 ToolExecCtx const badTurn: ToolExecCtx = { model: 'm', turnId: 1 }; void badTurn; - // @ts-expect-error callId 只进 hook payload，不属于 ToolExecCtx - const badCall: ToolExecCtx = { model: 'm', callId: 'c-1' }; - void badCall; + }); + + it('callId 随 ctx 透传给工具（ask_user 等待键）', () => { + const ctx: ToolExecCtx = { model: 'm', callId: 'c-1' }; + void ctx; }); it('ToolExecCall = ToolExecOpts + 批次级 sessionId / callId', () => { diff --git a/packages/desktop/shared/types.ts b/packages/desktop/shared/types.ts index 52d81942..dfaa50b0 100644 --- a/packages/desktop/shared/types.ts +++ b/packages/desktop/shared/types.ts @@ -8,7 +8,8 @@ export type ItemType = | 'tool_call' | 'tool_result' | 'error' - | 'subagent'; + | 'subagent' + | 'ask_user'; export type Item = | { id: string; type: 'message'; role: 'user' | 'assistant'; parts: ContentPart[]; partial?: boolean } @@ -40,6 +41,16 @@ export type Item = sessionId: string; agentName: string; status: 'running' | 'completed' | 'failed'; + } + /** ask_user 提问条目：id = `ask:`，等待用户作答 */ + | { + id: string; + type: 'ask_user'; + callId: string; + question: string; + questiontype: 'single' | 'multiple' | 'text'; + options: string[]; + status: 'waiting' | 'answered'; }; export type TodoStatus = 'pending' | 'in_progress' | 'completed'; diff --git a/packages/desktop/src/agent/AgentWorkspace.tsx b/packages/desktop/src/agent/AgentWorkspace.tsx index faec6e40..2176723d 100644 --- a/packages/desktop/src/agent/AgentWorkspace.tsx +++ b/packages/desktop/src/agent/AgentWorkspace.tsx @@ -25,6 +25,7 @@ import { import MessageStream from './MessageStream'; import TodoPanel from './TodoPanel'; import ApprovalPanel from './ApprovalPanel'; +import AskUserPanel from './AskUserPanel'; import ProfileIndicator from './ProfileIndicator'; import PlanPanel from '../shared/PlanPanel'; import MediaView from '../shared/MediaView'; @@ -878,6 +879,7 @@ export default function AgentWorkspace({ sendMessage, sendQueuedInput, abort }:
+ {isCompressing && (
diff --git a/packages/desktop/src/agent/AskUserPanel.tsx b/packages/desktop/src/agent/AskUserPanel.tsx new file mode 100644 index 00000000..13705518 --- /dev/null +++ b/packages/desktop/src/agent/AskUserPanel.tsx @@ -0,0 +1,211 @@ +import { useState, useMemo, useCallback, useEffect } from 'react'; +import type { Item } from '@shared/types'; +import { useAgentStore } from '../stores/agent.store'; +import { sendAskUserAnswer } from '../lib/core-api'; + +interface AskUserPanelProps { + threadId: string; +} + +const letterOf = (index: number): string => String.fromCharCode(65 + index); + +export default function AskUserPanel({ threadId }: AskUserPanelProps) { + const [index, setIndex] = useState(0); + const [choice, setChoice] = useState(null); + const [choices, setChoices] = useState([]); + const [custom, setCustom] = useState(''); + const [submitting, setSubmitting] = useState(false); + + const waitingKey = useAgentStore((s) => { + const thread = s.threads[threadId]; + if (!thread) return ''; + return thread.turns + .flatMap((t) => t.items) + .filter((i) => i.type === 'ask_user' && i.status === 'waiting') + .map((i) => i.id) + .join(','); + }); + + const waitingItems = useMemo(() => { + if (!waitingKey) return []; + const thread = useAgentStore.getState().threads[threadId]; + if (!thread) return []; + return thread.turns.flatMap((turn) => + turn.items.filter( + (i): i is Item & { type: 'ask_user' } => i.type === 'ask_user' && i.status === 'waiting' + ) + ); + }, [waitingKey, threadId]); + + // 切题或题目集合变化时重置表单 + useEffect(() => { + setChoice(null); + setChoices([]); + setCustom(''); + }, [waitingKey, index]); + + const clampedIndex = Math.min(index, Math.max(waitingItems.length - 1, 0)); + const current = waitingItems[clampedIndex] ?? null; + + const customText = custom.trim(); + const selectedCount = + current?.questiontype === 'single' + ? choice + ? 1 + : 0 + : current?.questiontype === 'multiple' + ? choices.length + : 0; + const canSubmit = + !!current && + !submitting && + (current.questiontype === 'text' + ? customText !== '' + : selectedCount > 0 || customText !== ''); + + const handleSubmit = useCallback(async () => { + if (!current || !canSubmit) return; + const options = + current.questiontype === 'single' + ? choice + ? [choice] + : [] + : current.questiontype === 'multiple' + ? choices + : []; + setSubmitting(true); + try { + await sendAskUserAnswer(threadId, current.callId, { + options, + custom: customText || undefined, + }); + useAgentStore.getState().markAskAnswered(threadId, current.callId); + } catch (e) { + console.error('Failed to submit ask_user answer:', e); + } finally { + setSubmitting(false); + } + }, [current, canSubmit, choice, choices, customText, threadId]); + + if (!current) return null; + + return ( +
+
+
+
+ ❓ + + 模型提问{waitingItems.length > 1 ? `（${clampedIndex + 1}/${waitingItems.length}）` : ''} + +
+ {waitingItems.length > 1 && ( +
+ + +
+ )} +
+
+
{current.question}
+ + {current.questiontype === 'single' && ( +
+ {current.options.slice(0, 4).map((text, i) => ( + + ))} +
+ )} + + {current.questiontype === 'multiple' && ( +
+ {current.options.slice(0, 4).map((text, i) => ( + + ))} +
+ )} + + {current.questiontype === 'text' ? ( +