From 21e03d86b5e8ec28a0bd511c484d0618476528a6 Mon Sep 17 00:00:00 2001 From: John Date: Thu, 2 Jul 2026 09:23:55 +0800 Subject: [PATCH] feat: add tool gateway protocol --- agent-run-gateway.mjs | 29 +++++ agent-run-gateway.test.mjs | 53 +++++++++ package.json | 2 +- scripts/agent-run-worker.mjs | 8 ++ scripts/check-tool-runtime.mjs | 13 +++ server.mjs | 4 + tool-gateway.mjs | 203 +++++++++++++++++++++++++++++++++ tool-gateway.test.mjs | 77 +++++++++++++ 8 files changed, 388 insertions(+), 1 deletion(-) create mode 100644 tool-gateway.mjs create mode 100644 tool-gateway.test.mjs diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 8e3efde..3729763 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -106,6 +106,7 @@ export function createAgentRunGateway({ pool, userAuth, tkmindProxy, + toolGateway = null, retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS, autoDispatch = envFlag(process.env.MEMIND_AGENT_RUN_AUTODISPATCH, true), maxConcurrentRuns = positiveInteger( @@ -251,6 +252,33 @@ export function createAgentRunGateway({ async function executeRun(row, runId) { const userMessage = safeJsonParse(row.user_message_json, {}); const runOptions = getRunOptionsFromMessage(userMessage); + const toolGatewayStatus = toolGateway?.getStatus ? toolGateway.getStatus() : null; + if (runOptions.toolMode === 'code' && toolGatewayStatus?.enabled) { + const workingDir = userAuth?.resolveWorkingDir + ? await userAuth.resolveWorkingDir(row.user_id) + : undefined; + await appendEvent(runId, 'tool_gateway_dispatch', { + protocol: toolGatewayStatus.protocol ?? 'agent-run-v1', + taskType: runOptions.taskType, + workingDir: workingDir ?? null, + }); + const result = await toolGateway.executeJob({ + runId, + userId: row.user_id, + requestId: row.request_id, + userMessage, + taskType: runOptions.taskType, + cwd: workingDir, + timeoutMs: runTimeoutMs, + }); + await appendEvent(runId, 'tool_gateway_result', { + executor: result.executor ?? null, + dryRun: Boolean(result.dryRun), + exitCode: result.exitCode ?? null, + }); + return { sessionId: row.agent_session_id ?? null }; + } + let sessionId = row.agent_session_id ?? null; if (!sessionId) { const sessionOptions = {}; @@ -353,6 +381,7 @@ export function createAgentRunGateway({ pendingDispatches: queuedDispatches.length, statusCounts, terminalStatuses: [...TERMINAL_STATUSES], + toolGateway: toolGateway?.getStatus ? toolGateway.getStatus() : null, }; } diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 51a866c..5c73fb3 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -223,6 +223,59 @@ test('agent run with code tool mode starts and submits with code policy', async assert.equal(submitted[0].userMessage.metadata.memindRun.toolMode, 'code'); }); +test('agent run with enabled tool gateway dispatches code runs outside goose session', async () => { + const pool = createFakePool(); + const jobs = []; + const gateway = createAgentRunGateway({ + pool, + userAuth: { + async resolveWorkingDir(userId) { + assert.equal(userId, 'user-1'); + return '/tmp/memind-user-1'; + }, + }, + tkmindProxy: { + async startSessionForUser() { + assert.fail('goose session should not start for external tool gateway run'); + }, + async submitSessionReplyForUser() { + assert.fail('goose reply should not be submitted for external tool gateway run'); + }, + }, + toolGateway: { + getStatus() { + return { enabled: true, protocol: 'agent-run-v1' }; + }, + async executeJob(job) { + jobs.push(job); + return { ok: true, dryRun: true, executor: 'aider' }; + }, + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-code-tool-gateway', + userMessage: { role: 'user', content: [{ type: 'text', text: 'run code task' }] }, + toolMode: 'code', + taskType: 'small_patch', + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + assert.equal(jobs.length, 1); + assert.equal(jobs[0].cwd, '/tmp/memind-user-1'); + assert.equal(jobs[0].taskType, 'small_patch'); + assert.equal(pool.runs.get(run.id).agent_session_id, null); + assert.equal( + pool.events.some((event) => event.runId === run.id && event.eventType === 'tool_gateway_dispatch'), + true, + ); + assert.equal( + pool.events.some((event) => event.runId === run.id && event.eventType === 'tool_gateway_result'), + true, + ); +}); + test('agent run retries transient failures and then becomes terminal', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ diff --git a/package.json b/package.json index 78e4f22..afc5f1b 100644 --- a/package.json +++ b/package.json @@ -38,7 +38,7 @@ "build:portal-runtime": "node scripts/build-portal-runtime.mjs", "check:mindspace-public-links": "node scripts/check-mindspace-public-links.mjs --downloads-only", "check:mindspace-public-links:all": "node scripts/check-mindspace-public-links.mjs --all-links", - "test": "node --test auth.test.mjs asr-proxy.test.mjs billing.test.mjs billing-recharge.test.mjs wechat-pay.test.mjs wechat-oauth.test.mjs wechat-mp.test.mjs schedule-intent.test.mjs schedule-reminder-worker.test.mjs capabilities.test.mjs policies.test.mjs chat-skills.test.mjs chat-finish-sync.test.mjs conversation-display.test.mjs user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-pages.test.mjs mindspace-html-download-links.test.mjs mindspace-page-purge.test.mjs mindspace-publications.test.mjs mindspace-public-links.test.mjs mindspace-chat-save.test.mjs mindspace-public-finish-sync.test.mjs mindspace-chat-context.test.mjs mindspace-content-scan.test.mjs mindspace-html-localize.test.mjs mindspace-visual-editor.test.mjs mindspace-cleanup.test.mjs mindspace-thumbnails.test.mjs mindspace-workspace-thumbnails.test.mjs mindspace-workspace-sync.test.mjs mindspace-asset-preview.test.mjs mindspace-agent-jobs.test.mjs mindspace-agent-runner.test.mjs mindspace-sandbox-mcp.test.mjs message-stream.test.mjs plaza-posts.test.mjs plaza-interactions.test.mjs plaza-algorithm.test.mjs plaza-seo.test.mjs plaza-ops.test.mjs user-auth.test.mjs llm-providers.test.mjs admin-guard.test.mjs user-feedback.test.mjs", + "test": "node --test auth.test.mjs asr-proxy.test.mjs billing.test.mjs billing-recharge.test.mjs wechat-pay.test.mjs wechat-oauth.test.mjs wechat-mp.test.mjs schedule-intent.test.mjs schedule-reminder-worker.test.mjs capabilities.test.mjs policies.test.mjs chat-skills.test.mjs chat-finish-sync.test.mjs conversation-display.test.mjs user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs tool-gateway.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-pages.test.mjs mindspace-html-download-links.test.mjs mindspace-page-purge.test.mjs mindspace-publications.test.mjs mindspace-public-links.test.mjs mindspace-chat-save.test.mjs mindspace-public-finish-sync.test.mjs mindspace-chat-context.test.mjs mindspace-content-scan.test.mjs mindspace-html-localize.test.mjs mindspace-visual-editor.test.mjs mindspace-cleanup.test.mjs mindspace-thumbnails.test.mjs mindspace-workspace-thumbnails.test.mjs mindspace-workspace-sync.test.mjs mindspace-asset-preview.test.mjs mindspace-agent-jobs.test.mjs mindspace-agent-runner.test.mjs mindspace-sandbox-mcp.test.mjs message-stream.test.mjs plaza-posts.test.mjs plaza-interactions.test.mjs plaza-algorithm.test.mjs plaza-seo.test.mjs plaza-ops.test.mjs user-auth.test.mjs llm-providers.test.mjs admin-guard.test.mjs user-feedback.test.mjs", "verify:chat-finish-sync": "node scripts/verify-chat-finish-sync.mjs", "verify:public-finish-sync-runtime": "node scripts/verify-public-finish-sync-runtime.mjs", "verify:mindspace-publish-guards": "node scripts/verify-mindspace-publish-guards.mjs", diff --git a/scripts/agent-run-worker.mjs b/scripts/agent-run-worker.mjs index c0ca476..9300a88 100644 --- a/scripts/agent-run-worker.mjs +++ b/scripts/agent-run-worker.mjs @@ -4,7 +4,9 @@ import path from 'node:path'; import { fileURLToPath } from 'node:url'; import { createAgentRunGateway } from '../agent-run-gateway.mjs'; import { createDbPool } from '../db.mjs'; +import { createLlmProviderService } from '../llm-providers.mjs'; import { createTkmindProxy } from '../tkmind-proxy.mjs'; +import { createToolGateway } from '../tool-gateway.mjs'; import { createUserAuth } from '../user-auth.mjs'; const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..'); @@ -74,6 +76,11 @@ const userAuth = createUserAuth(pool, { usersRoot: process.env.H5_USERS_ROOT ?? path.join(root, 'users'), h5Root: root, }); +const llmProviderService = createLlmProviderService(pool, { + apiTarget: process.env.TKMIND_API_TARGET ?? 'https://127.0.0.1:18006', + apiSecret: process.env.TKMIND_SERVER__SECRET_KEY ?? 'local-dev-secret', +}); +const toolGateway = createToolGateway({ llmProviderService }); const apiTargets = parseTargets(); const tkmindProxy = createTkmindProxy({ apiTarget: apiTargets[0], @@ -85,6 +92,7 @@ const gateway = createAgentRunGateway({ pool, userAuth, tkmindProxy, + toolGateway, autoDispatch: false, maxConcurrentRuns: Number(process.env.MEMIND_AGENT_RUN_QUEUE_CONCURRENCY ?? 1), runTimeoutMs: Number(process.env.MEMIND_AGENT_RUN_TIMEOUT_MS ?? 15 * 60 * 1000), diff --git a/scripts/check-tool-runtime.mjs b/scripts/check-tool-runtime.mjs index 9958de5..4a5b7ba 100644 --- a/scripts/check-tool-runtime.mjs +++ b/scripts/check-tool-runtime.mjs @@ -97,6 +97,19 @@ try { chatInjectsCodeTools: false, aiderTimeoutMs: Number(process.env.MEMIND_AIDER_TIMEOUT_MS ?? 600_000), openhandsTimeoutMs: Number(process.env.MEMIND_OPENHANDS_TIMEOUT_MS ?? 900_000), + toolGateway: { + enabled: ['1', 'true', 'yes', 'on'].includes( + String(process.env.MEMIND_TOOL_GATEWAY_ENABLED ?? '').trim().toLowerCase(), + ), + dryRun: ['1', 'true', 'yes', 'on'].includes( + String(process.env.MEMIND_TOOL_GATEWAY_DRY_RUN ?? '').trim().toLowerCase(), + ), + protocol: 'agent-run-v1', + defaultExecutor: process.env.MEMIND_TOOL_GATEWAY_DEFAULT_EXECUTOR ?? 'aider', + openhandsTaskTypes: + process.env.MEMIND_TOOL_GATEWAY_OPENHANDS_TASK_TYPES ?? + 'repo_refactor,multi_file,complex_repo', + }, }, }, null, 2)); process.exit(ok ? 0 : 1); diff --git a/server.mjs b/server.mjs index 3ddfae6..8ada420 100644 --- a/server.mjs +++ b/server.mjs @@ -14,6 +14,7 @@ import { } from './auth.mjs'; import { createDbPool, initSchema, isDatabaseConfigured } from './db.mjs'; import { createAgentRunGateway } from './agent-run-gateway.mjs'; +import { createToolGateway } from './tool-gateway.mjs'; import { createAgentRunEventsHandler, createGetAgentRunHandler, @@ -234,6 +235,7 @@ if (ACCESS_PASSWORD && !isDatabaseConfigured()) { let userAuth = null; let tkmindProxy = null; let agentRunGateway = null; +let toolGateway = null; let sessionSnapshotService = null; let conversationMemoryService = null; let mindSpace = null; @@ -568,10 +570,12 @@ async function bootstrapUserAuth() { } : null, }); + toolGateway = createToolGateway({ llmProviderService }); agentRunGateway = createAgentRunGateway({ pool, userAuth, tkmindProxy, + toolGateway, autoDispatch: ['1', 'true', 'yes', 'on'].includes( String(process.env.MEMIND_AGENT_RUN_AUTODISPATCH ?? '1').trim().toLowerCase(), ), diff --git a/tool-gateway.mjs b/tool-gateway.mjs new file mode 100644 index 0000000..b2ec43c --- /dev/null +++ b/tool-gateway.mjs @@ -0,0 +1,203 @@ +import { spawn as nodeSpawn } from 'node:child_process'; +import { EventEmitter } from 'node:events'; + +const CODE_EXECUTORS = new Set(['aider', 'openhands']); +const DEFAULT_STDIO_LIMIT = 64 * 1024; + +function envFlag(value, fallback = false) { + const raw = String(value ?? '').trim().toLowerCase(); + if (!raw) return fallback; + return ['1', 'true', 'yes', 'on'].includes(raw); +} + +function positiveInteger(value, fallback) { + const n = Number(value); + if (!Number.isFinite(n) || n <= 0) return fallback; + return Math.floor(n); +} + +function normalizeExecutor(value, fallback = 'aider') { + const normalized = String(value ?? fallback).trim().toLowerCase(); + if (CODE_EXECUTORS.has(normalized)) return normalized; + return fallback; +} + +function csvSet(value) { + return new Set( + String(value ?? '') + .split(',') + .map((item) => item.trim().toLowerCase()) + .filter(Boolean), + ); +} + +function appendLimited(current, chunk, limit) { + const next = `${current}${Buffer.isBuffer(chunk) ? chunk.toString('utf8') : String(chunk ?? '')}`; + if (next.length <= limit) return next; + return next.slice(next.length - limit); +} + +export function extractToolInstruction(userMessage) { + const content = userMessage?.content; + if (typeof content === 'string') return content.trim(); + if (!Array.isArray(content)) { + return String(userMessage?.text ?? userMessage?.value ?? '').trim(); + } + return content + .map((item) => { + if (typeof item === 'string') return item; + if (item?.type === 'text') return item.text ?? ''; + return ''; + }) + .map((item) => String(item ?? '').trim()) + .filter(Boolean) + .join('\n') + .trim(); +} + +export function createToolGateway({ + llmProviderService, + env = process.env, + spawnImpl = nodeSpawn, +} = {}) { + const enabled = envFlag(env.MEMIND_TOOL_GATEWAY_ENABLED, false); + const dryRun = envFlag(env.MEMIND_TOOL_GATEWAY_DRY_RUN, false); + const defaultExecutor = normalizeExecutor(env.MEMIND_TOOL_GATEWAY_DEFAULT_EXECUTOR, 'aider'); + const openhandsTaskTypes = csvSet( + env.MEMIND_TOOL_GATEWAY_OPENHANDS_TASK_TYPES ?? 'repo_refactor,multi_file,complex_repo', + ); + const stdioLimit = positiveInteger(env.MEMIND_TOOL_GATEWAY_STDIO_LIMIT, DEFAULT_STDIO_LIMIT); + + function getStatus() { + return { + enabled, + dryRun, + protocol: 'agent-run-v1', + executors: ['aider', 'openhands'], + defaultExecutor, + openhandsTaskTypes: [...openhandsTaskTypes], + }; + } + + function selectExecutor({ userMessage, taskType } = {}) { + const metadata = userMessage?.metadata ?? {}; + const runMetadata = metadata.memindRun ?? metadata.agentRun ?? {}; + const requested = normalizeExecutor(runMetadata.executor, ''); + if (requested) return requested; + const normalizedTaskType = String(taskType ?? runMetadata.taskType ?? '').trim().toLowerCase(); + if (openhandsTaskTypes.has(normalizedTaskType)) return 'openhands'; + return defaultExecutor; + } + + async function executeJob({ + runId, + requestId, + userId, + userMessage, + taskType, + cwd, + timeoutMs = 15 * 60 * 1000, + } = {}) { + if (!enabled) { + throw new Error('Tool Gateway is disabled'); + } + if (!llmProviderService?.getExecutorLaunchPlan) { + throw new Error('Tool Gateway missing llm provider service'); + } + const instruction = extractToolInstruction(userMessage); + if (!instruction) { + throw new Error('Tool Gateway job missing instruction'); + } + const executor = selectExecutor({ userMessage, taskType }); + const plan = await llmProviderService.getExecutorLaunchPlan(executor, { + cwd, + mode: 'headless', + instruction, + purpose: 'default', + includeSecret: true, + }); + if (!plan?.ok) { + throw new Error(plan?.message ?? `Tool Gateway launch plan unavailable for ${executor}`); + } + if (dryRun) { + return { + ok: true, + dryRun: true, + executor, + protocol: 'agent-run-v1', + cwd: plan.cwd, + command: plan.command, + args: plan.args ?? [], + runId, + requestId, + userId, + }; + } + + return await new Promise((resolve, reject) => { + let stdout = ''; + let stderr = ''; + const child = spawnImpl(plan.command, plan.args ?? [], { + cwd: plan.cwd, + env: { ...process.env, ...(plan.env ?? {}) }, + stdio: ['ignore', 'pipe', 'pipe'], + }); + const cleanup = () => { + if (timer) clearTimeout(timer); + }; + const timer = setTimeout(() => { + child.kill('SIGTERM'); + const err = new Error(`Tool Gateway job timed out after ${timeoutMs}ms`); + err.code = 'TOOL_GATEWAY_TIMEOUT'; + reject(err); + }, timeoutMs); + + if (child.stdout instanceof EventEmitter) { + child.stdout.on('data', (chunk) => { + stdout = appendLimited(stdout, chunk, stdioLimit); + }); + } + if (child.stderr instanceof EventEmitter) { + child.stderr.on('data', (chunk) => { + stderr = appendLimited(stderr, chunk, stdioLimit); + }); + } + child.on('error', (err) => { + cleanup(); + reject(err); + }); + child.on('exit', (code, signal) => { + cleanup(); + if (code === 0) { + resolve({ + ok: true, + executor, + protocol: 'agent-run-v1', + cwd: plan.cwd, + command: plan.command, + args: plan.args ?? [], + exitCode: code, + signal: signal ?? null, + stdout, + stderr, + }); + return; + } + const err = new Error(`Tool Gateway executor ${executor} exited with ${signal ?? code}`); + err.code = 'TOOL_GATEWAY_EXECUTOR_FAILED'; + err.executor = executor; + err.exitCode = code; + err.signal = signal ?? null; + err.stdout = stdout; + err.stderr = stderr; + reject(err); + }); + }); + } + + return { + getStatus, + selectExecutor, + executeJob, + }; +} diff --git a/tool-gateway.test.mjs b/tool-gateway.test.mjs new file mode 100644 index 0000000..adb3520 --- /dev/null +++ b/tool-gateway.test.mjs @@ -0,0 +1,77 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { createToolGateway, extractToolInstruction } from './tool-gateway.mjs'; + +test('tool gateway extracts text instructions from H5 message content', () => { + assert.equal( + extractToolInstruction({ + content: [ + { type: 'text', text: '第一步' }, + { type: 'image_url', image_url: { url: 'https://example.test/a.png' } }, + { type: 'text', text: '第二步' }, + ], + }), + '第一步\n第二步', + ); +}); + +test('tool gateway is disabled by default and reports protocol', () => { + const gateway = createToolGateway({ env: {} }); + assert.deepEqual(gateway.getStatus(), { + enabled: false, + dryRun: false, + protocol: 'agent-run-v1', + executors: ['aider', 'openhands'], + defaultExecutor: 'aider', + openhandsTaskTypes: ['repo_refactor', 'multi_file', 'complex_repo'], + }); +}); + +test('tool gateway selects openhands for configured task types', () => { + const gateway = createToolGateway({ + env: { + MEMIND_TOOL_GATEWAY_DEFAULT_EXECUTOR: 'aider', + MEMIND_TOOL_GATEWAY_OPENHANDS_TASK_TYPES: 'repo_refactor', + }, + }); + assert.equal(gateway.selectExecutor({ taskType: 'repo_refactor' }), 'openhands'); + assert.equal(gateway.selectExecutor({ taskType: 'small_patch' }), 'aider'); +}); + +test('tool gateway dry run builds executor launch plan without spawning', async () => { + const plans = []; + const gateway = createToolGateway({ + env: { + MEMIND_TOOL_GATEWAY_ENABLED: '1', + MEMIND_TOOL_GATEWAY_DRY_RUN: '1', + MEMIND_TOOL_GATEWAY_DEFAULT_EXECUTOR: 'aider', + }, + llmProviderService: { + async getExecutorLaunchPlan(executor, options) { + plans.push({ executor, options }); + return { + ok: true, + executor, + cwd: options.cwd, + command: '/usr/bin/true', + args: ['--noop'], + }; + }, + }, + }); + + const result = await gateway.executeJob({ + runId: 'run-1', + requestId: 'req-1', + userId: 'user-1', + cwd: '/tmp/work', + taskType: 'small_patch', + userMessage: { content: [{ type: 'text', text: 'fix it' }] }, + }); + + assert.equal(result.dryRun, true); + assert.equal(result.executor, 'aider'); + assert.equal(result.command, '/usr/bin/true'); + assert.equal(plans.length, 1); + assert.equal(plans[0].options.instruction, 'fix it'); +});