From 79334ece2a761879f9b7fef854db75d878eb0a0e Mon Sep 17 00:00:00 2001 From: axb Date: Tue, 11 Aug 2026 22:42:59 +0800 Subject: [PATCH] fix(executor): align interactive form defer handling --- .../en/wegent/developer-guide/architecture.md | 2 + .../zh/wegent/developer-guide/architecture.md | 2 + executor/src/process/mod.rs | 13 +- .../agent_runtime_capabilities_contract.rs | 79 +++- .../agent-conversation-regression.spec.ts | 391 +++++++++++++++++- frontend/e2e/utils/mock-model-server.ts | 144 ++++++- 6 files changed, 610 insertions(+), 21 deletions(-) diff --git a/docs/en/wegent/developer-guide/architecture.md b/docs/en/wegent/developer-guide/architecture.md index ab4f94ca25..3daf7c6e81 100644 --- a/docs/en/wegent/developer-guide/architecture.md +++ b/docs/en/wegent/developer-guide/architecture.md @@ -366,6 +366,8 @@ EXECUTOR_IMAGE: wegent-executor:latest # Executor image Rust executor is the only executor runtime implementation. Backend Chat shell work may still use an in-process path, while other tasks run through standalone/local executor. In Wework packaged App local-first mode, the app does not start a local Backend; it calls the executor sidecar directly over Tauri app IPC. Codex runtime control uses `codex app-server --stdio` JSON-RPC to create, continue, read, archive, and rename threads. The executor stores only the local task index and the required `localTaskId -> threadId` mapping. +When Claude Code resumes an interactive-form session, the executor treats a defer as stale resume output only when it has the same `tool_use_id` as the form being answered and is still an interactive-form tool. A later form with a different `tool_use_id` is a new clarification request; even if that response also contains text, the executor must proxy it to the interactive MCP and wait for user input instead of discarding it as stale. + Before attachments enter Codex, the executor converts them by type: images become local image inputs, text attachments include a bounded preview and their complete local path, and binary attachments such as ZIP or PDF include their filename, MIME type, size, and local path. Codex can therefore locate a file even when the user sends an attachment without message text. These contexts are mutually exclusive by type so image and text attachments are not injected twice. Image conversion may create temporary `*.model-input.*` files that exist only for model consumption; those paths must not become persistent Wework message attachment URLs. When restoring user messages from the Codex transcript, the executor prefers the original attachment path retained in the file-mention context or local runtime handle. Temporary model inputs are used only during inference, so historical messages, task switching, and reopened tasks continue to render the original image after temporary files are cleaned up. diff --git a/docs/zh/wegent/developer-guide/architecture.md b/docs/zh/wegent/developer-guide/architecture.md index 417cee1d33..7bcad84d3d 100644 --- a/docs/zh/wegent/developer-guide/architecture.md +++ b/docs/zh/wegent/developer-guide/architecture.md @@ -366,6 +366,8 @@ EXECUTOR_IMAGE: wegent-executor:latest # 执行器镜像 Rust executor 是唯一的 executor 运行时实现。Backend 的 Chat shell 仍可走进程内路径，其他任务由 standalone/local executor 执行；Wework 打包 App 的 local-first 模式不启动本地 Backend，而是通过 Tauri app IPC 直接调用 executor。Codex 运行时通过 `codex app-server --stdio` 的 JSON-RPC 协议创建、继续、读取、归档和重命名线程，executor 只保存必要的本地任务索引和 `localTaskId -> threadId` 关联。 +Claude Code 恢复交互表单会话时，executor 只把与本次已回答表单具有相同 `tool_use_id`、且工具类型仍为交互表单的 defer 视为恢复阶段残留结果。模型随后返回不同 `tool_use_id` 的表单表示新的用户澄清，即使同一响应还包含文本，也必须继续代理到交互 MCP 并等待用户输入，不能按旧 defer 丢弃。 + 附件在进入 Codex 前由 executor 按类型转换：图片作为本地图片输入，文本附件附带受限预览和完整本地路径，ZIP、PDF 等二进制附件则附带文件名、MIME 类型、大小和本地路径。即使用户只发送附件而正文为空，Codex 仍能从输入上下文定位该文件；不同类型的上下文互斥生成，避免图片或文本附件被重复注入。 图片转换可能生成仅供模型读取的临时 `*.model-input.*` 文件，但该路径不能作为 Wework 消息附件的持久化地址。executor 从 Codex transcript 恢复用户消息时，优先使用文件提及上下文或本地 runtime handle 中保留的原始附件路径；临时模型输入仅用于推理阶段。这样临时文件清理后，历史消息、任务切换和重开任务仍能显示原始图片预览。 diff --git a/executor/src/process/mod.rs b/executor/src/process/mod.rs index 2532a40898..bff2ea4e64 100644 --- a/executor/src/process/mod.rs +++ b/executor/src/process/mod.rs @@ -611,18 +611,17 @@ async fn handle_deferred_mcp_loop( let Some(deferred_tool_use) = summary.deferred_tool_use.clone() else { return summary.outcome; }; - // After draining an already answered form, a non-empty final answer is - // authoritative; a leftover deferred form is stale Claude session state. - if stale_answer_defer_drained - && answered_interactive_form_tool_use_id(&request).is_some() + let answered_tool_use_id = answered_interactive_form_tool_use_id(&request); + if answered_tool_use_id.as_deref() == Some(deferred_tool_use.id.as_str()) + && crate::agents::interactive_mcp::is_interactive_form_tool(&deferred_tool_use.name) && completed_with_content(&summary.outcome) { - log_executor_event("ignoring stale deferred form after answered drain", &fields); + log_executor_event("ignoring stale deferred form after answer", &fields); return summary.outcome; } if !stale_answer_defer_drained - && answered_interactive_form_tool_use_id(&request) - .is_some_and(|tool_use_id| tool_use_id == deferred_tool_use.id) + && answered_tool_use_id.as_deref() == Some(deferred_tool_use.id.as_str()) + && crate::agents::interactive_mcp::is_interactive_form_tool(&deferred_tool_use.name) { stale_answer_defer_drained = true; log_executor_event("draining stale answered interactive form defer", &fields); diff --git a/executor/tests/agent_runtime_capabilities_contract.rs b/executor/tests/agent_runtime_capabilities_contract.rs index 1ac998572e..2a222c5478 100644 --- a/executor/tests/agent_runtime_capabilities_contract.rs +++ b/executor/tests/agent_runtime_capabilities_contract.rs @@ -898,6 +898,55 @@ async fn claude_runtime_streams_answer_drain_follow_up_output() { })); } +#[tokio::test] +async fn claude_runtime_preserves_new_deferred_form_after_answer_drain() { + let _lock = env_lock().await; + let workspace_root = unique_dir("claude-runtime-answer-new-defer-workspace"); + let marker = unique_dir("claude-runtime-answer-new-defer-marker").join("count"); + let fake_claude = write_fake_claude_answer_drain_with_new_defer(&marker); + let waiting_payload = json!({ + "__deferred_user_input__": true, + "success": true, + "status": "waiting_for_user_response" + }); + let mcp_url = spawn_mcp_server(vec![ + json!({"jsonrpc": "2.0", "id": 1, "result": {}}), + json!({"jsonrpc": "2.0", "result": {}}), + json!({ + "jsonrpc": "2.0", + "id": 2, + "result": { + "content": [{ + "type": "text", + "text": waiting_payload.to_string() + }] + } + }), + ]) + .await; + let _workspace = EnvGuard::set("WORKSPACE_ROOT", &workspace_root.display().to_string()); + let _mode = EnvGuard::set("EXECUTOR_MODE", "docker"); + let engine = AgentProcessEngine::new(AgentCommandPlanner::new( + fake_claude.display().to_string(), + "codex", + )); + let mut request = interactive_form_answer_request(7795, 106); + request.mcp_servers = vec![json!({ + "name": "interactive-wegent-interactive-form-question", + "type": "streamable-http", + "url": mcp_url + })]; + + let outcome = engine.run(request).await; + + assert_eq!( + outcome, + ExecutionOutcome::WaitingForUserInput { + stop_reason: "tool_deferred".to_owned() + } + ); +} + async fn env_lock() -> MutexGuard<'static, ()> { static LOCK: std::sync::OnceLock> = std::sync::OnceLock::new(); LOCK.get_or_init(|| Mutex::new(())).lock().await @@ -1045,7 +1094,35 @@ if ! grep -q 'tool-answered' >/dev/null 2>&1; then exit 9 fi printf '%s\n' '{{"type":"assistant","message":{{"content":[{{"type":"text","text":"published"}}]}}}}' -printf '%s\n' '{{"type":"result","subtype":"success","is_error":false,"session_id":"session-answer-stale","stop_reason":"tool_deferred","usage":{{}},"deferred_tool_use":{{"id":"tool-stale-followup","name":"mcp__interactive_wegent-interactive-form-question__interactive_form_question","input":{{"questions":[{{"id":"confirm","question":"Confirm?"}}]}}}}}}' +printf '%s\n' '{{"type":"result","subtype":"success","is_error":false,"session_id":"session-answer-stale","stop_reason":"tool_deferred","usage":{{}},"deferred_tool_use":{{"id":"tool-answered","name":"mcp__interactive_wegent-interactive-form-question__interactive_form_question","input":{{"questions":[]}}}}}}' +"#, + marker.display() + ); + fs::write(&path, content).unwrap(); + make_executable(&path); + path +} + +fn write_fake_claude_answer_drain_with_new_defer(marker: &Path) -> PathBuf { + if let Some(parent) = marker.parent() { + fs::create_dir_all(parent).unwrap(); + } + let path = unique_dir("fake-claude-answer-new-defer").join("claude"); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + let content = format!( + r#"#!/bin/sh +MARKER='{}' +if [ ! -f "$MARKER" ]; then + printf 1 > "$MARKER" + printf '%s\n' '{{"type":"system","subtype":"init","session_id":"session-answer-new-defer"}}' + printf '%s\n' '{{"type":"result","subtype":"success","is_error":false,"session_id":"session-answer-new-defer","stop_reason":"tool_deferred","usage":{{}},"deferred_tool_use":{{"id":"tool-answered","name":"mcp__interactive_wegent-interactive-form-question__interactive_form_question","input":{{"questions":[]}}}}}}' + exit 0 +fi +if ! grep -q 'tool-answered' >/dev/null 2>&1; then + exit 9 +fi +printf '%s\n' '{{"type":"assistant","message":{{"content":[{{"type":"text","text":"one verification decision remains"}}]}}}}' +printf '%s\n' '{{"type":"result","subtype":"success","is_error":false,"session_id":"session-answer-new-defer","stop_reason":"tool_deferred","usage":{{}},"deferred_tool_use":{{"id":"tool-new","name":"mcp__interactive_wegent-interactive-form-question__interactive_form_question","input":{{"questions":[{{"id":"verification_scope","question":"Which verification scope?"}}]}}}}}}' "#, marker.display() ); diff --git a/frontend/e2e/tests/tasks/agent-conversation-regression.spec.ts b/frontend/e2e/tests/tasks/agent-conversation-regression.spec.ts index 2e4e3970b3..4ca97239f1 100644 --- a/frontend/e2e/tests/tasks/agent-conversation-regression.spec.ts +++ b/frontend/e2e/tests/tasks/agent-conversation-regression.spec.ts @@ -54,6 +54,20 @@ type CapturedModelRequest = { body: unknown } +type SkillRefMeta = { + skill_id: number + namespace: string + is_public: boolean + content_hash?: string +} + +type InteractiveToolResponse = { + id: string + nameIncludes: string + input: Record + text?: string +} + test.describe.configure({ mode: 'serial', timeout: 180_000 }) test.describe('Agent conversation regression', () => { @@ -151,6 +165,224 @@ test.describe('Agent conversation regression', () => { expect(extractText(secondRequest.body)).toContain(firstPrompt) }) + test('ClaudeCode clarification submits the selected option to the resumed model turn', async ({ + page, + request, + }) => { + const prompt = `CLARIFICATION_DEFER_${makeContextToken('clarification')}` + const toolUseId = `tool_${makeContextToken('clarification_tool').toLowerCase()}` + const questionId = 'implementation_scope' + const selectedValue = 'complete' + + await configureInteractiveFormRule(request, prompt, { + id: toolUseId, + nameIncludes: 'interactive_form_question', + input: { + questions: [ + { + id: questionId, + question: 'Which implementation scope should be used?', + input_type: 'choice', + options: [ + { + label: 'Minimal', + value: 'minimal', + description: 'Make the smallest focused change.', + }, + { + label: 'Complete', + value: selectedValue, + description: 'Cover the full interaction flow.', + recommended: true, + }, + ], + required: true, + multi_select: false, + }, + ], + }, + }) + await openTaskPage(page, '/chat', claudeChatTeam.id, 'chat') + + await sendMessage(page, prompt) + const taskId = await waitForTaskId(page) + createdTaskIds.add(taskId) + + const initialRequest = await waitForCapturedModelRequest( + request, + capture => isAnthropicMessagesRequest(capture) && extractText(capture.body).includes(prompt), + `ClaudeCode model request containing ${prompt}` + ) + expect(requestOffersTool(initialRequest.body, 'interactive_form_question')).toBe(true) + + const form = page.getByTestId('ask-user-form') + await expect(form).toContainText('Which implementation scope should be used?', { + timeout: RESPONSE_TIMEOUT_MS, + }) + await page.getByTestId(`ask-user-option-${questionId}-1`).click() + await page.getByTestId('ask-user-submit').click() + + const resumedRequest = await waitForCapturedModelRequest( + request, + capture => { + return ( + isAnthropicMessagesRequest(capture) && + containsInteractiveFormAnswer(capture.body, toolUseId, questionId, selectedValue) + ) + }, + `ClaudeCode resumed request containing selected clarification option ${selectedValue}` + ) + expect( + containsInteractiveFormAnswer(resumedRequest.body, toolUseId, questionId, selectedValue) + ).toBe(true) + await waitForBackendTerminal(request, taskId) + + const captures = await loadCapturedModelRequests(request) + const clarificationRequests = captures.filter( + capture => isAnthropicMessagesRequest(capture) && extractText(capture.body).includes(prompt) + ) + expect(clarificationRequests).toHaveLength(2) + expect( + containsInteractiveFormAnswer( + clarificationRequests.at(-1)?.body, + toolUseId, + questionId, + selectedValue + ) + ).toBe(true) + }) + + test('ClaudeCode preserves a new clarification after answering the previous form', async ({ + page, + request, + }) => { + const prompt = `CONSECUTIVE_CLARIFICATION_${makeContextToken('consecutive_clarification')}` + const firstToolUseId = `tool_${makeContextToken('first_clarification').toLowerCase()}` + const secondToolUseId = `tool_${makeContextToken('second_clarification').toLowerCase()}` + const firstQuestionId = 'implementation_scope' + const secondQuestionId = 'verification_scope' + + await configureInteractiveFormRule(request, prompt, [ + { + id: firstToolUseId, + nameIncludes: 'interactive_form_question', + input: { + questions: [ + { + id: firstQuestionId, + question: 'Which implementation scope should be used?', + input_type: 'choice', + options: [ + { + label: 'Minimal', + value: 'minimal', + description: 'Make the smallest focused change.', + }, + { + label: 'Complete', + value: 'complete', + description: 'Cover the full interaction flow.', + }, + ], + required: true, + multi_select: false, + }, + ], + }, + }, + { + id: secondToolUseId, + nameIncludes: 'interactive_form_question', + text: 'The implementation scope is clear; one verification decision remains.', + input: { + questions: [ + { + id: secondQuestionId, + question: 'Which verification scope should be used?', + input_type: 'choice', + options: [ + { + label: 'Focused', + value: 'focused', + description: 'Run only the focused regression.', + }, + { + label: 'Full', + value: 'full', + description: 'Run the complete verification suite.', + }, + ], + required: true, + multi_select: false, + }, + ], + }, + }, + ]) + await openTaskPage(page, '/chat', claudeChatTeam.id, 'chat') + + await sendMessage(page, prompt) + const taskId = await waitForTaskId(page) + createdTaskIds.add(taskId) + + const firstForm = page + .getByTestId('ask-user-form') + .filter({ hasText: 'Which implementation scope should be used?' }) + await expect(firstForm).toBeVisible({ + timeout: RESPONSE_TIMEOUT_MS, + }) + await page.getByTestId(`ask-user-option-${firstQuestionId}-1`).click() + await firstForm.getByTestId('ask-user-submit').click() + + await waitForCapturedModelRequest( + request, + capture => + isAnthropicMessagesRequest(capture) && + containsInteractiveFormAnswer(capture.body, firstToolUseId, firstQuestionId, 'complete'), + 'ClaudeCode request containing the first clarification answer' + ) + + const secondForm = page + .getByTestId('ask-user-form') + .filter({ hasText: 'Which verification scope should be used?' }) + await expect(secondForm).toBeVisible({ + timeout: RESPONSE_TIMEOUT_MS, + }) + await page.getByTestId(`ask-user-option-${secondQuestionId}-1`).click() + await secondForm.getByTestId('ask-user-submit').click() + + await waitForCapturedModelRequest( + request, + capture => + isAnthropicMessagesRequest(capture) && + containsInteractiveFormAnswer(capture.body, secondToolUseId, secondQuestionId, 'full'), + 'ClaudeCode request containing the second clarification answer' + ) + await waitForBackendTerminal(request, taskId) + + const captures = await loadCapturedModelRequests(request) + const clarificationRequests = captures.filter( + capture => isAnthropicMessagesRequest(capture) && extractText(capture.body).includes(prompt) + ) + expect(clarificationRequests).toHaveLength(3) + expect( + containsInteractiveFormAnswer( + clarificationRequests[1]?.body, + firstToolUseId, + firstQuestionId, + 'complete' + ) + ).toBe(true) + expect( + containsInteractiveFormAnswer( + clarificationRequests[2]?.body, + secondToolUseId, + secondQuestionId, + 'full' + ) + ).toBe(true) + }) + test('coding mode ClaudeCode supports dialogue and follow-up', async ({ page, request }) => { const contextToken = makeContextToken('code') const firstPrompt = `Remember this code context token: ${contextToken}` @@ -392,6 +624,7 @@ test.describe('Agent conversation regression', () => { await createClaudeShell(request) + const interactiveSkillRef = await resolveSkillRef(request, 'interactive') chatShellTeam = await createTeam(request, { teamName: `${TEST_PREFIX}-chat-shell-team`, botName: `${TEST_PREFIX}-chat-shell-bot`, @@ -405,6 +638,10 @@ test.describe('Agent conversation regression', () => { shellName: CLAUDE_SHELL_NAME, bindMode: ['chat'], modelName: CLAUDE_MODEL_NAME, + skills: ['interactive'], + skillRefs: { interactive: interactiveSkillRef }, + preloadSkills: ['interactive'], + preloadSkillRefs: { interactive: interactiveSkillRef }, }) codeTeam = await createTeam(request, { teamName: `${TEST_PREFIX}-code-team`, @@ -465,6 +702,10 @@ test.describe('Agent conversation regression', () => { shellName: string bindMode: string[] modelName: string + skills?: string[] + skillRefs?: Record + preloadSkills?: string[] + preloadSkillRefs?: Record } ): Promise { const botResponse = await request.post(`${API_BASE_URL}/api/bots`, { @@ -477,6 +718,10 @@ test.describe('Agent conversation regression', () => { bind_model_type: 'user', }, system_prompt: 'You are a deterministic E2E regression assistant.', + skills: options.skills, + skill_refs: options.skillRefs, + preload_skills: options.preloadSkills, + preload_skill_refs: options.preloadSkillRefs, namespace: 'default', is_active: true, }, @@ -514,6 +759,43 @@ test.describe('Agent conversation regression', () => { } } + async function resolveSkillRef( + request: APIRequestContext, + skillName: string + ): Promise { + const response = await request.get( + `${API_BASE_URL}/api/v1/kinds/skills?name=${encodeURIComponent(skillName)}&namespace=default&exact_match=false`, + { headers: authHeaders() } + ) + expect(response.status()).toBe(200) + + const body = (await response.json()) as { + items?: Array<{ + metadata?: { + namespace?: string + labels?: Record + } + status?: { + fileHash?: string + } + }> + } + const skill = body.items?.[0] + const skillId = Number(skill?.metadata?.labels?.id) + expect( + skillId, + `Skill ${skillName} should expose a numeric metadata.labels.id` + ).toBeGreaterThan(0) + + const fileHash = skill?.status?.fileHash + return { + skill_id: skillId, + namespace: skill?.metadata?.namespace || 'default', + is_public: skill?.metadata?.labels?.user_id === '0', + content_hash: fileHash ? `sha256:${fileHash}` : undefined, + } + } + async function createPipelineTeam( request: APIRequestContext, options: { @@ -690,6 +972,95 @@ test.describe('Agent conversation regression', () => { expect(response.status()).toBe(200) } + async function configureInteractiveFormRule( + request: APIRequestContext, + matchText: string, + responseTool: InteractiveToolResponse | InteractiveToolResponse[] + ): Promise { + streamRuleMatchTexts.add(matchText) + const response = await request.post(`${MOCK_MODEL_SERVER_URL}/stream-rules`, { + ...MOCK_MODEL_CONTROL_REQUEST_OPTIONS, + data: { + matchText, + ...(Array.isArray(responseTool) ? { responseTools: responseTool } : { responseTool }), + }, + }) + expect(response.status()).toBe(200) + } + + function containsInteractiveFormAnswer( + value: unknown, + toolUseId: string, + questionId: string, + selectedValue: string + ): boolean { + if (Array.isArray(value)) { + return value.some(item => + containsInteractiveFormAnswer(item, toolUseId, questionId, selectedValue) + ) + } + if (!value || typeof value !== 'object') { + return false + } + + const record = value as Record + if ( + record.type === 'tool_result' && + record.tool_use_id === toolUseId && + answerPayloadContainsSelection(record.content, questionId, selectedValue) + ) { + return true + } + return Object.values(record).some(item => + containsInteractiveFormAnswer(item, toolUseId, questionId, selectedValue) + ) + } + + function requestOffersTool(value: unknown, nameIncludes: string): boolean { + if (!value || typeof value !== 'object') { + return false + } + const tools = (value as { tools?: unknown }).tools + if (!Array.isArray(tools)) { + return false + } + return tools.some(tool => { + if (!tool || typeof tool !== 'object') { + return false + } + const name = (tool as { name?: unknown }).name + return typeof name === 'string' && name.includes(nameIncludes) + }) + } + + function answerPayloadContainsSelection( + value: unknown, + questionId: string, + selectedValue: string + ): boolean { + if (typeof value === 'string') { + try { + return answerPayloadContainsSelection(JSON.parse(value), questionId, selectedValue) + } catch { + return false + } + } + if (Array.isArray(value)) { + return value.some(item => answerPayloadContainsSelection(item, questionId, selectedValue)) + } + if (!value || typeof value !== 'object') { + return false + } + + const record = value as Record + if (record[questionId] === selectedValue) { + return true + } + return Object.values(record).some(item => + answerPayloadContainsSelection(item, questionId, selectedValue) + ) + } + async function cleanupStreamRules(request: APIRequestContext): Promise { const rules = [...streamRuleMatchTexts] streamRuleMatchTexts.clear() @@ -809,19 +1180,23 @@ test.describe('Agent conversation regression', () => { : predicateOrText return waitForCapturedRequest( - async () => { - const response = await request.get( - `${MOCK_MODEL_SERVER_URL}/captured-requests`, - MOCK_MODEL_CONTROL_REQUEST_OPTIONS - ) - expect(response.status()).toBe(200) - return (await response.json()) as CapturedModelRequest[] - }, + () => loadCapturedModelRequests(request), predicate, label || `mock model request containing ${predicateOrText}` ) } + async function loadCapturedModelRequests( + request: APIRequestContext + ): Promise { + const response = await request.get( + `${MOCK_MODEL_SERVER_URL}/captured-requests`, + MOCK_MODEL_CONTROL_REQUEST_OPTIONS + ) + expect(response.status()).toBe(200) + return (await response.json()) as CapturedModelRequest[] + } + async function waitForCapturedRequest( load: () => Promise, predicate: (capture: T) => boolean, diff --git a/frontend/e2e/utils/mock-model-server.ts b/frontend/e2e/utils/mock-model-server.ts index b0855014f4..33c813380f 100644 --- a/frontend/e2e/utils/mock-model-server.ts +++ b/frontend/e2e/utils/mock-model-server.ts @@ -40,12 +40,21 @@ interface ModelRequest { messages?: VisionMessage[] system?: unknown stream?: boolean - tools?: unknown[] + tools?: Array> +} + +interface ToolResponse { + id: string + nameIncludes: string + input: Record + text?: string } interface StreamRule { matchText: string - responseContent: string + responseContent?: string + responseTool?: ToolResponse + responseTools?: ToolResponse[] chunkDelayMs?: number doneDelayMs?: number } @@ -53,6 +62,7 @@ interface StreamRule { // Store captured requests for verification const capturedRequests: CapturedRequest[] = [] const streamRules: StreamRule[] = [] +const servedToolRuleCounts = new Map() // Port for the mock server const PORT = parseInt(process.env.MOCK_MODEL_PORT || '9999') @@ -358,6 +368,113 @@ function writeAnthropicStreamingResponse( sendChunk() } +function findToolName(request: ModelRequest | null, nameIncludes: string): string | null { + const tool = request?.tools?.find(candidate => { + const name = candidate.name + return typeof name === 'string' && name.includes(nameIncludes) + }) + const name = tool?.name + return typeof name === 'string' ? name : null +} + +function writeAnthropicToolUseResponse( + res: http.ServerResponse, + request: ModelRequest | null, + tool: ToolResponse, + model: string +): void { + const toolName = findToolName(request, tool.nameIncludes) + if (!toolName) { + writeJson(res, 400, { + error: `Configured response tool was not offered: ${tool.nameIncludes}`, + }) + return + } + const input = JSON.stringify(tool.input) + + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }) + writeAnthropicSseEvent(res, 'message_start', { + type: 'message_start', + message: { + id: `msg_${Date.now()}`, + type: 'message', + role: 'assistant', + model, + content: [], + stop_reason: null, + stop_sequence: null, + usage: { + input_tokens: 100, + output_tokens: 0, + }, + }, + }) + let toolBlockIndex = 0 + if (tool.text) { + writeAnthropicSseEvent(res, 'content_block_start', { + type: 'content_block_start', + index: 0, + content_block: { + type: 'text', + text: '', + }, + }) + writeAnthropicSseEvent(res, 'content_block_delta', { + type: 'content_block_delta', + index: 0, + delta: { + type: 'text_delta', + text: tool.text, + }, + }) + writeAnthropicSseEvent(res, 'content_block_stop', { + type: 'content_block_stop', + index: 0, + }) + toolBlockIndex = 1 + } + writeAnthropicSseEvent(res, 'content_block_start', { + type: 'content_block_start', + index: toolBlockIndex, + content_block: { + type: 'tool_use', + id: tool.id, + name: toolName, + input: {}, + }, + }) + writeAnthropicSseEvent(res, 'content_block_delta', { + type: 'content_block_delta', + index: toolBlockIndex, + delta: { + type: 'input_json_delta', + partial_json: input, + }, + }) + writeAnthropicSseEvent(res, 'content_block_stop', { + type: 'content_block_stop', + index: toolBlockIndex, + }) + writeAnthropicSseEvent(res, 'message_delta', { + type: 'message_delta', + delta: { + stop_reason: 'tool_use', + stop_sequence: null, + }, + usage: { + output_tokens: 1, + }, + }) + writeAnthropicSseEvent(res, 'message_stop', { + type: 'message_stop', + }) + res.end() +} + function writeAnthropicJsonResponse( res: http.ServerResponse, content: string, @@ -508,7 +625,14 @@ const server = http.createServer((req, res) => { const isStreaming = parsedBody?.stream === true console.log(`Mock response content: ${truncateForLog(responseContent)}`) - if (isStreaming) { + const responseTools = + streamRule?.responseTools || (streamRule?.responseTool ? [streamRule.responseTool] : []) + const servedToolCount = streamRule ? servedToolRuleCounts.get(streamRule.matchText) || 0 : 0 + const responseTool = responseTools[servedToolCount] + if (isStreaming && streamRule && responseTool) { + servedToolRuleCounts.set(streamRule.matchText, servedToolCount + 1) + writeAnthropicToolUseResponse(res, parsedBody, responseTool, model) + } else if (isStreaming) { writeAnthropicStreamingResponse( res, responseContent, @@ -532,12 +656,20 @@ const server = http.createServer((req, res) => { writeJson(res, 200, streamRules) } else if (req.url === '/stream-rules' && req.method === 'POST') { const streamRule = parseJsonBody(body) - if (!streamRule?.matchText || !streamRule.responseContent) { - writeJson(res, 400, { error: 'matchText and responseContent are required' }) + if ( + !streamRule?.matchText || + (!streamRule.responseContent && + !streamRule.responseTool && + !streamRule.responseTools?.length) + ) { + writeJson(res, 400, { + error: 'matchText and responseContent, responseTool, or responseTools are required', + }) return } const existingIndex = streamRules.findIndex(rule => rule.matchText === streamRule.matchText) + servedToolRuleCounts.delete(streamRule.matchText) if (existingIndex >= 0) { streamRules[existingIndex] = streamRule } else { @@ -554,8 +686,10 @@ const server = http.createServer((req, res) => { if (ruleIndex >= 0) { streamRules.splice(ruleIndex, 1) } + servedToolRuleCounts.delete(matchText) } else { streamRules.length = 0 + servedToolRuleCounts.clear() } writeJson(res, 200, { message: 'Stream rules cleared', remainingCount: streamRules.length })