From 8f620fa714593bf7ec8d9c8931a2de13cd18a06a Mon Sep 17 00:00:00 2001 From: john Date: Mon, 29 Jun 2026 22:50:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E8=81=8A=E5=A4=A9=E6=8F=90?= =?UTF-8?q?=E4=BA=A4=E7=BB=9F=E4=B8=80=E8=B5=B0=20POST=20/agent/runs=20?= =?UTF-8?q?=E5=BC=82=E6=AD=A5=E7=BD=91=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 引入 Agent Run 网关替代直连 /sessions/:id/reply，并在 api_lockdown 白名单中放行新入口，避免策略拦截导致聊天不可用。 Co-authored-by: Cursor --- agent-run-gateway.mjs | 194 ++++++++++++++++++++++++++++++++ agent-run-gateway.test.mjs | 168 +++++++++++++++++++++++++++ db.mjs | 34 ++++++ package.json | 2 +- policies.mjs | 2 + policies.test.mjs | 10 ++ schema.sql | 30 +++++ server.mjs | 128 +++++++++++++++------ src/api/client.ts | 30 ++++- src/components/ChatPanel.tsx | 8 +- src/components/ChatView.tsx | 6 +- src/hooks/usePageEditSubChat.ts | 35 +++++- src/hooks/useTKMindChat.ts | 109 ++++++++++++------ tkmind-proxy.mjs | 117 ++++++++++++++++++- 14 files changed, 784 insertions(+), 89 deletions(-) create mode 100644 agent-run-gateway.mjs create mode 100644 agent-run-gateway.test.mjs diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs new file mode 100644 index 0000000..d89e9b3 --- /dev/null +++ b/agent-run-gateway.mjs @@ -0,0 +1,194 @@ +import crypto from 'node:crypto'; + +const DEFAULT_RUN_RETRY_DELAYS_MS = [1500, 5000, 15000]; +const TERMINAL_STATUSES = new Set(['succeeded', 'failed']); + +function nowMs() { + return Date.now(); +} + +function safeJsonParse(value, fallback = null) { + try { + return JSON.parse(value); + } catch { + return fallback; + } +} + +function serializeMessage(message) { + return JSON.stringify(message ?? {}); +} + +function projectRun(row) { + if (!row) return null; + return { + id: row.id, + userId: row.user_id, + sessionId: row.agent_session_id ?? null, + requestId: row.request_id, + status: row.status, + attempts: Number(row.attempts ?? 0), + error: row.error_message ?? null, + createdAt: Number(row.created_at ?? 0), + updatedAt: Number(row.updated_at ?? 0), + startedAt: row.started_at == null ? null : Number(row.started_at), + completedAt: row.completed_at == null ? null : Number(row.completed_at), + }; +} + +export function createAgentRunGateway({ + pool, + userAuth, + tkmindProxy, + retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS, + autoDispatch = true, +}) { + const inFlight = new Set(); + + async function appendEvent(runId, type, data = null) { + await pool.query( + `INSERT INTO h5_agent_run_events (id, run_id, event_type, data_json, created_at) + VALUES (?, ?, ?, ?, ?)`, + [ + crypto.randomUUID(), + runId, + type, + data == null ? null : JSON.stringify(data), + nowMs(), + ], + ); + } + + async function getRunById(runId) { + const [rows] = await pool.query( + `SELECT * FROM h5_agent_runs WHERE id = ? LIMIT 1`, + [runId], + ); + return rows[0] ?? null; + } + + async function getRunForUser(userId, runId) { + const [rows] = await pool.query( + `SELECT * FROM h5_agent_runs WHERE id = ? AND user_id = ? LIMIT 1`, + [runId, userId], + ); + return projectRun(rows[0] ?? null); + } + + async function getRunByRequest(userId, requestId) { + const [rows] = await pool.query( + `SELECT * FROM h5_agent_runs WHERE user_id = ? AND request_id = ? LIMIT 1`, + [userId, requestId], + ); + return rows[0] ?? null; + } + + async function createRun(userId, { sessionId = null, requestId, userMessage }) { + const normalizedRequestId = String(requestId ?? '').trim(); + if (!normalizedRequestId) { + throw new Error('缺少 request_id'); + } + const existing = await getRunByRequest(userId, normalizedRequestId); + if (existing) { + if (autoDispatch && !TERMINAL_STATUSES.has(existing.status)) dispatchRun(existing.id); + return projectRun(existing); + } + + const runId = crypto.randomUUID(); + const createdAt = nowMs(); + await pool.query( + `INSERT INTO h5_agent_runs + (id, user_id, agent_session_id, request_id, status, attempts, + user_message_json, error_message, created_at, updated_at, started_at, completed_at) + VALUES (?, ?, ?, ?, 'queued', 0, ?, NULL, ?, ?, NULL, NULL)`, + [ + runId, + userId, + sessionId || null, + normalizedRequestId, + serializeMessage(userMessage), + createdAt, + createdAt, + ], + ); + await appendEvent(runId, 'queued', { sessionId: sessionId || null }); + if (autoDispatch) dispatchRun(runId); + return projectRun(await getRunById(runId)); + } + + async function markRun(runId, status, fields = {}) { + const updates = ['status = ?', 'updated_at = ?']; + const values = [status, nowMs()]; + for (const [key, value] of Object.entries(fields)) { + updates.push(`${key} = ?`); + values.push(value); + } + values.push(runId); + await pool.query( + `UPDATE h5_agent_runs SET ${updates.join(', ')} WHERE id = ?`, + values, + ); + await appendEvent(runId, status, fields); + } + + async function processRun(runId) { + const row = await getRunById(runId); + if (!row || TERMINAL_STATUSES.has(row.status)) return; + const nextAttempt = Number(row.attempts ?? 0) + 1; + const [claim] = await pool.query( + `UPDATE h5_agent_runs + SET status = 'running', attempts = ?, started_at = COALESCE(started_at, ?), updated_at = ?, error_message = NULL + WHERE id = ? AND status IN ('queued', 'retryable')`, + [nextAttempt, nowMs(), nowMs(), runId], + ); + if (Number(claim?.affectedRows ?? 0) === 0) return; + await appendEvent(runId, 'running', { attempt: nextAttempt }); + + try { + let sessionId = row.agent_session_id ?? null; + if (!sessionId) { + const session = await tkmindProxy.startSessionForUser(row.user_id); + sessionId = session.id; + await pool.query( + `UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`, + [sessionId, nowMs(), runId], + ); + await appendEvent(runId, 'session_started', { sessionId }); + } + + const userMessage = safeJsonParse(row.user_message_json, {}); + await tkmindProxy.submitSessionReplyForUser( + row.user_id, + sessionId, + row.request_id, + userMessage, + ); + await markRun(runId, 'succeeded', { + agent_session_id: sessionId, + completed_at: nowMs(), + }); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + const retryable = nextAttempt < retryDelaysMs.length; + await markRun(runId, retryable ? 'retryable' : 'failed', { + error_message: message, + completed_at: retryable ? null : nowMs(), + }); + if (retryable) { + setTimeout(() => dispatchRun(runId), retryDelaysMs[nextAttempt - 1]); + } + } + } + + function dispatchRun(runId) { + if (!runId || inFlight.has(runId)) return; + inFlight.add(runId); + void processRun(runId).finally(() => inFlight.delete(runId)); + } + + return { + createRun, + getRunForUser, + dispatchRun, + }; +} diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs new file mode 100644 index 0000000..612626f --- /dev/null +++ b/agent-run-gateway.test.mjs @@ -0,0 +1,168 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { createAgentRunGateway } from './agent-run-gateway.mjs'; + +function createFakePool() { + const runs = new Map(); + const events = []; + + return { + runs, + events, + async query(sql, params = []) { + if (sql.includes('SELECT * FROM h5_agent_runs WHERE id = ? AND user_id = ?')) { + const [id, userId] = params; + const row = runs.get(id); + return [[row && row.user_id === userId ? row : undefined].filter(Boolean)]; + } + if (sql.includes('SELECT * FROM h5_agent_runs WHERE user_id = ? AND request_id = ?')) { + const [userId, requestId] = params; + return [[...runs.values()].filter((row) => row.user_id === userId && row.request_id === requestId)]; + } + if (sql.includes('SELECT * FROM h5_agent_runs WHERE id = ? LIMIT 1')) { + return [[runs.get(params[0])].filter(Boolean)]; + } + if (sql.includes('INSERT INTO h5_agent_runs')) { + const [ + id, + userId, + sessionId, + requestId, + userMessageJson, + createdAt, + updatedAt, + ] = params; + runs.set(id, { + id, + user_id: userId, + agent_session_id: sessionId, + request_id: requestId, + status: 'queued', + attempts: 0, + user_message_json: userMessageJson, + error_message: null, + created_at: createdAt, + updated_at: updatedAt, + started_at: null, + completed_at: null, + }); + return [{ affectedRows: 1 }]; + } + if (sql.includes('INSERT INTO h5_agent_run_events')) { + const [id, runId, eventType, dataJson, createdAt] = params; + events.push({ id, runId, eventType, dataJson, createdAt }); + return [{ affectedRows: 1 }]; + } + if (sql.includes("SET status = 'running'")) { + const [attempts, startedAt, updatedAt, id] = params; + const row = runs.get(id); + if (!row || !['queued', 'retryable'].includes(row.status)) { + return [{ affectedRows: 0 }]; + } + Object.assign(row, { + status: 'running', + attempts, + started_at: row.started_at ?? startedAt, + updated_at: updatedAt, + error_message: null, + }); + return [{ affectedRows: 1 }]; + } + if (sql.includes('UPDATE h5_agent_runs SET')) { + const id = params.at(-1); + const row = runs.get(id); + if (!row) return [{ affectedRows: 0 }]; + const columns = [...sql.matchAll(/([a-z_]+) = \?/g)].map((match) => match[1]); + for (let i = 0; i < columns.length; i += 1) { + row[columns[i]] = params[i]; + } + return [{ affectedRows: 1 }]; + } + throw new Error(`Unhandled SQL: ${sql}`); + }, + }; +} + +async function waitFor(predicate) { + for (let i = 0; i < 50; i += 1) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + assert.fail('condition was not met'); +} + +test('agent run creation is idempotent by user and request id', async () => { + const pool = createFakePool(); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + }); + + const first = await gateway.createRun('user-1', { + requestId: 'req-1', + userMessage: { role: 'user', content: [] }, + }); + const second = await gateway.createRun('user-1', { + requestId: 'req-1', + userMessage: { role: 'user', content: [] }, + }); + + assert.equal(first.id, second.id); + assert.equal(pool.runs.size, 1); + assert.equal(second.status, 'queued'); +}); + +test('agent run starts a session and marks submitted reply as succeeded', async () => { + const pool = createFakePool(); + const submitted = []; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-1' }; + }, + async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { + submitted.push({ userId, sessionId, requestId, userMessage }); + }, + }, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-1', + userMessage: { role: 'user', content: [{ type: 'text', text: 'hi' }] }, + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + assert.deepEqual(submitted.map((item) => item.sessionId), ['session-1']); + assert.equal(pool.runs.get(run.id).attempts, 1); +}); + +test('agent run retries transient failures and then becomes terminal', async () => { + const pool = createFakePool(); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-1' }; + }, + async submitSessionReplyForUser() { + throw new Error('upstream unavailable'); + }, + }, + retryDelaysMs: [0, 0], + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-1', + userMessage: { role: 'user', content: [] }, + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'failed'); + assert.equal(pool.runs.get(run.id).attempts, 2); + assert.match(pool.runs.get(run.id).error_message, /upstream unavailable/); +}); diff --git a/db.mjs b/db.mjs index 1f9ff44..f6857b7 100644 --- a/db.mjs +++ b/db.mjs @@ -255,6 +255,40 @@ export async function migrateSchema(pool) { ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci `); + await pool.query(` + CREATE TABLE IF NOT EXISTS h5_agent_runs ( + id CHAR(36) PRIMARY KEY, + user_id CHAR(36) NOT NULL, + agent_session_id VARCHAR(128) NULL, + request_id VARCHAR(128) NOT NULL, + status ENUM('queued', 'running', 'retryable', 'succeeded', 'failed') NOT NULL DEFAULT 'queued', + attempts INT NOT NULL DEFAULT 0, + user_message_json LONGTEXT NOT NULL, + error_message TEXT NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + started_at BIGINT NULL, + completed_at BIGINT NULL, + UNIQUE KEY uq_h5_agent_run_request (user_id, request_id), + KEY idx_h5_agent_run_user_status (user_id, status, updated_at), + KEY idx_h5_agent_run_session (agent_session_id, updated_at), + CONSTRAINT fk_h5_agent_run_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE, + CONSTRAINT fk_h5_agent_run_session FOREIGN KEY (agent_session_id) REFERENCES h5_user_sessions(agent_session_id) ON DELETE SET NULL + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + `); + + await pool.query(` + CREATE TABLE IF NOT EXISTS h5_agent_run_events ( + id CHAR(36) PRIMARY KEY, + run_id CHAR(36) NOT NULL, + event_type VARCHAR(64) NOT NULL, + data_json JSON NULL, + created_at BIGINT NOT NULL, + KEY idx_h5_agent_run_event_run (run_id, created_at), + CONSTRAINT fk_h5_agent_run_event_run FOREIGN KEY (run_id) REFERENCES h5_agent_runs(id) ON DELETE CASCADE + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + `); + await pool.query(` CREATE TABLE IF NOT EXISTS h5_user_memory_items ( id CHAR(64) PRIMARY KEY, diff --git a/package.json b/package.json index 87e15a2..f4eb4a7 100644 --- a/package.json +++ b/package.json @@ -34,7 +34,7 @@ "enrich:plaza": "node scripts/enrich-plaza-posts.mjs", "build": "vite build", "build:portal-runtime": "node scripts/build-portal-runtime.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 user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-pages.test.mjs mindspace-publications.test.mjs mindspace-chat-save.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", + "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 user-publish.test.mjs user-memory-profile.test.mjs skills-registry.test.mjs agent-run-gateway.test.mjs mindspace.test.mjs mindspace-scan.test.mjs mindspace-assets.test.mjs mindspace-pages.test.mjs mindspace-publications.test.mjs mindspace-chat-save.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", "test:mindspace-e2e": "node scripts/mindspace-e2e.mjs", "test:mindspace-pages-e2e": "node scripts/mindspace-pages-e2e.mjs", "test:mindspace-publications-e2e": "node scripts/mindspace-publications-e2e.mjs", diff --git a/policies.mjs b/policies.mjs index 264fd4d..e7fb538 100644 --- a/policies.mjs +++ b/policies.mjs @@ -205,6 +205,8 @@ const USER_API_ALLOWLIST = [ { method: 'GET', pattern: /^\/status$/ }, { method: 'POST', pattern: /^\/agent\/start$/ }, { method: 'POST', pattern: /^\/agent\/resume$/ }, + { method: 'POST', pattern: /^\/agent\/runs$/ }, + { method: 'GET', pattern: /^\/agent\/runs\/[^/]+$/ }, { method: 'GET', pattern: /^\/sessions$/ }, { method: 'GET', pattern: /^\/sessions\/[^/]+$/ }, { method: 'DELETE', pattern: /^\/sessions\/[^/]+$/ }, diff --git a/policies.test.mjs b/policies.test.mjs index 6559d98..61c8b0c 100644 --- a/policies.test.mjs +++ b/policies.test.mjs @@ -60,6 +60,16 @@ test('api lockdown blocks extension admin routes', () => { }); assert.equal(allowed.allowed, true); + const agentRunCreate = evaluateProxyRequest('POST', '/agent/runs', { + api_lockdown: true, + }); + assert.equal(agentRunCreate.allowed, true); + + const agentRunGet = evaluateProxyRequest('GET', '/agent/runs/run-1', { + api_lockdown: true, + }); + assert.equal(agentRunGet.allowed, true); + const deleteAllowed = evaluateProxyRequest('DELETE', '/sessions/abc', { api_lockdown: true, }); diff --git a/schema.sql b/schema.sql index e408cd4..9715ea6 100644 --- a/schema.sql +++ b/schema.sql @@ -353,6 +353,36 @@ CREATE TABLE IF NOT EXISTS h5_user_sessions ( CONSTRAINT fk_h5_session_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +CREATE TABLE IF NOT EXISTS h5_agent_runs ( + id CHAR(36) PRIMARY KEY, + user_id CHAR(36) NOT NULL, + agent_session_id VARCHAR(128) NULL, + request_id VARCHAR(128) NOT NULL, + status ENUM('queued', 'running', 'retryable', 'succeeded', 'failed') NOT NULL DEFAULT 'queued', + attempts INT NOT NULL DEFAULT 0, + user_message_json LONGTEXT NOT NULL, + error_message TEXT NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + started_at BIGINT NULL, + completed_at BIGINT NULL, + UNIQUE KEY uq_h5_agent_run_request (user_id, request_id), + KEY idx_h5_agent_run_user_status (user_id, status, updated_at), + KEY idx_h5_agent_run_session (agent_session_id, updated_at), + CONSTRAINT fk_h5_agent_run_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE, + CONSTRAINT fk_h5_agent_run_session FOREIGN KEY (agent_session_id) REFERENCES h5_user_sessions(agent_session_id) ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS h5_agent_run_events ( + id CHAR(36) PRIMARY KEY, + run_id CHAR(36) NOT NULL, + event_type VARCHAR(64) NOT NULL, + data_json JSON NULL, + created_at BIGINT NOT NULL, + KEY idx_h5_agent_run_event_run (run_id, created_at), + CONSTRAINT fk_h5_agent_run_event_run FOREIGN KEY (run_id) REFERENCES h5_agent_runs(id) ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + CREATE TABLE IF NOT EXISTS h5_session_billing_state ( agent_session_id VARCHAR(128) NOT NULL PRIMARY KEY, user_id CHAR(36) NOT NULL, diff --git a/server.mjs b/server.mjs index 6fa81a2..b18535b 100644 --- a/server.mjs +++ b/server.mjs @@ -13,6 +13,7 @@ import { sessionCookie, } from './auth.mjs'; import { createDbPool, initSchema, isDatabaseConfigured } from './db.mjs'; +import { createAgentRunGateway } from './agent-run-gateway.mjs'; import { createTkmindProxy } from './tkmind-proxy.mjs'; import { clearUserSessionCookie, @@ -216,6 +217,7 @@ if (ACCESS_PASSWORD) { let userAuth = null; let tkmindProxy = null; +let agentRunGateway = null; let sessionSnapshotService = null; let conversationMemoryService = null; let mindSpace = null; @@ -540,6 +542,11 @@ async function bootstrapUserAuth() { } : null, }); + agentRunGateway = createAgentRunGateway({ + pool, + userAuth, + tkmindProxy, + }); wechatMpService = createWechatMpService({ config: WECHAT_MP_CONFIG, userAuth, @@ -3668,6 +3675,85 @@ api.post('/agent/start', async (req, res, next) => { return runHandlerChain(tkmindProxy.handlers['POST /agent/start'], req, res, next); }); +api.post('/agent/runs', async (req, res, next) => { + await userAuthReady; + if (!userAuth || !tkmindProxy || !agentRunGateway) return next(); + return runHandlerChain( + [ + tkmindProxy.requireUser, + tkmindProxy.ensureChatAllowed, + async (request, response) => { + try { + const sessionId = String(request.body?.session_id ?? '').trim() || null; + const requestId = String(request.body?.request_id ?? '').trim(); + const userMessage = request.body?.user_message ?? null; + if (!requestId) { + response.status(400).json({ message: '缺少 request_id' }); + return; + } + if (!userMessage) { + response.status(400).json({ message: '缺少 user_message' }); + return; + } + if (sessionId) { + const owns = await userAuth.ownsSession(request.currentUser.id, sessionId); + if (!owns) { + response.status(403).json({ message: '无权访问该会话' }); + return; + } + } + const run = await agentRunGateway.createRun(request.currentUser.id, { + sessionId, + requestId, + userMessage, + }); + response.status(202).json({ run }); + } catch (err) { + response.status(500).json({ + message: err instanceof Error ? err.message : '创建任务失败', + }); + } + }, + ], + req, + res, + next, + ); +}); + +api.get('/agent/runs/:runId', async (req, res, next) => { + await userAuthReady; + if (!userAuth || !tkmindProxy || !agentRunGateway) return next(); + return runHandlerChain( + [ + tkmindProxy.requireUser, + async (request, response) => { + try { + const run = await agentRunGateway.getRunForUser( + request.currentUser.id, + request.params.runId, + ); + if (!run) { + response.status(404).json({ message: '任务不存在' }); + return; + } + if (run.status !== 'succeeded' && run.status !== 'failed') { + agentRunGateway.dispatchRun(run.id); + } + response.json({ run }); + } catch (err) { + response.status(500).json({ + message: err instanceof Error ? err.message : '读取任务失败', + }); + } + }, + ], + req, + res, + next, + ); +}); + api.post('/agent/resume', async (req, res, next) => { await userAuthReady; if (!userAuth || !tkmindProxy) return next(); @@ -3802,46 +3888,16 @@ api.use(async (req, res, next) => { if (req.path.endsWith('/events') && req.method === 'GET') { return next(); } + if (req.path.endsWith('/reply') && req.method === 'POST') { + return res.status(410).json({ + message: '聊天提交入口已统一为 POST /agent/runs', + code: 'AGENT_RUNS_REQUIRED', + }); + } const owns = await userAuth.ownsSession(req.currentUser.id, sessionId); if (!owns) { return res.status(403).json({ message: '无权访问该会话' }); } - if (req.path.endsWith('/reply') && req.method === 'POST') { - const gate = await userAuth.canUseChat(req.currentUser.id); - if (!gate.ok) { - return res.status(402).json({ - message: gate.message, - code: gate.code, - balanceCents: gate.balanceCents, - minRechargeCents: gate.minRechargeCents, - suggestedTiers: gate.suggestedTiers, - }); - } - try { - await tkmindProxy.reconcileSessionPolicyForUser(req.currentUser.id, sessionId); - } catch (err) { - console.warn( - 'Session policy sync before reply failed:', - err instanceof Error ? err.message : err, - ); - return res.status(500).json({ - message: - err instanceof Error - ? `会话策略同步失败：${err.message}` - : '会话策略同步失败', - }); - } - if (llmProviderService) { - try { - await tkmindProxy.applySessionLlmProvider(sessionId); - } catch (err) { - console.warn( - 'LLM provider sync before reply skipped:', - err instanceof Error ? err.message : err, - ); - } - } - } } if (req.method === 'POST' && req.path === '/agent/resume' && req.body?.session_id) { diff --git a/src/api/client.ts b/src/api/client.ts index a15cc7b..7929685 100644 --- a/src/api/client.ts +++ b/src/api/client.ts @@ -51,6 +51,21 @@ import { normalizeConversationMessages, normalizeUserMessageForApi } from '../ut const API = '/api'; const DEFAULT_API_TIMEOUT_MS = 20_000; +const AGENT_RUNS_PATH = '/agent/runs'; + +export type AgentRun = { + id: string; + userId: string; + sessionId: string | null; + requestId: string; + status: 'queued' | 'running' | 'retryable' | 'succeeded' | 'failed'; + attempts: number; + error: string | null; + createdAt: number; + updatedAt: number; + startedAt: number | null; + completedAt: number | null; +}; export class ApiError extends Error { readonly status: number; @@ -2038,18 +2053,25 @@ export async function loadSessionDetail( return { session: detail, messages }; } -export async function sendReply( - sessionId: string, +export async function createAgentRun( + sessionId: string | null, requestId: string, userMessage: Message, -): Promise { - await apiFetch(`${sessionPath(sessionId)}/reply`, { +): Promise { + const result = await apiFetch<{ run: AgentRun }>(AGENT_RUNS_PATH, { method: 'POST', body: JSON.stringify({ + session_id: sessionId, request_id: requestId, user_message: normalizeUserMessageForApi(userMessage), }), }); + return result.run; +} + +export async function getAgentRun(runId: string): Promise { + const result = await apiFetch<{ run: AgentRun }>(`${AGENT_RUNS_PATH}/${encodeURIComponent(runId)}`); + return result.run; } export async function cancelRequest(sessionId: string, requestId: string): Promise { diff --git a/src/components/ChatPanel.tsx b/src/components/ChatPanel.tsx index 4f32a80..c1c4511 100644 --- a/src/components/ChatPanel.tsx +++ b/src/components/ChatPanel.tsx @@ -134,7 +134,11 @@ export function ChatPanel({ bottomRef.current?.scrollIntoView({ behavior: 'smooth' }); }, [messages, chatState, pendingTool]); - const busy = chatState === 'streaming' || chatState === 'loading' || chatState === 'connecting'; + const busy = + chatState === 'streaming' || + chatState === 'loading' || + chatState === 'connecting' || + chatState === 'waiting'; const offlineBlocked = !online; const publishSkillName = user?.publishSkillName ?? 'static-page-publish'; const hasPublishSkill = grantedSkills?.includes(publishSkillName) ?? false; @@ -147,6 +151,8 @@ export function ChatPanel({ ? '网络断开，恢复连接后可继续输入' : chatState === 'connecting' ? '正在连接会话…' + : chatState === 'waiting' + ? '消息已收到，正在连接后台…' : compact ? '在这里继续和 Agent 对话…' : randomPrompt; diff --git a/src/components/ChatView.tsx b/src/components/ChatView.tsx index c5e120f..84d7be9 100644 --- a/src/components/ChatView.tsx +++ b/src/components/ChatView.tsx @@ -71,7 +71,11 @@ export function ChatView({ } }, [sidebarOpen, onSidebarOpen]); - const busy = chatState === 'streaming' || chatState === 'loading' || chatState === 'connecting'; + const busy = + chatState === 'streaming' || + chatState === 'loading' || + chatState === 'connecting' || + chatState === 'waiting'; const showHomeWelcome = messages.length === 0; const handleSelectSession = (sessionId: string) => { diff --git a/src/hooks/usePageEditSubChat.ts b/src/hooks/usePageEditSubChat.ts index d1a6480..802849a 100644 --- a/src/hooks/usePageEditSubChat.ts +++ b/src/hooks/usePageEditSubChat.ts @@ -4,12 +4,13 @@ import { cancelRequest, closeMindSpacePageEditSession, confirmTool, + createAgentRun, forkMindSpacePageEditSession, + getAgentRun, getMindSpace, loadSessionDetail, resumeSession, uploadMindSpaceAsset, - sendReply, subscribeSessionEvents, } from '../api/client'; import type { ChatState, Message, MindSpaceChatContext, PortalUser, Session, SessionEvent } from '../types'; @@ -30,6 +31,21 @@ import { pushMessage, } from '../utils/message'; +const AGENT_RUN_POLL_DELAY_MS = 1200; +const AGENT_RUN_MAX_POLLS = 100; + +async function waitForAgentRun(runId: string) { + for (let attempt = 0; attempt < AGENT_RUN_MAX_POLLS; attempt += 1) { + const run = await getAgentRun(runId); + if (run.status === 'succeeded') return run; + if (run.status === 'failed') { + throw new Error(run.error || '后台任务失败，请稍后重试'); + } + await new Promise((resolve) => window.setTimeout(resolve, AGENT_RUN_POLL_DELAY_MS)); + } + throw new Error('后台任务仍在排队，请稍后刷新会话查看结果'); +} + function buildPageEditSummary(messages: Message[], pageTitle: string): string { const lines = messages .map((message) => { @@ -288,7 +304,12 @@ export function usePageEditSubChat({ (value) => typeof value === 'string' && value.trim(), ); if (!currentSession || (!text.trim() && normalizedImageUrls.length === 0)) return; - if (chatState === 'streaming' || chatState === 'loading' || chatState === 'connecting') return; + if ( + chatState === 'streaming' || + chatState === 'loading' || + chatState === 'connecting' || + chatState === 'waiting' + ) return; const trimmed = text.trim(); const mindspacePrefix = buildContextPrefix(context); @@ -304,12 +325,18 @@ export function usePageEditSubChat({ activeRequestId.current = requestId; messagesRef.current = [...messagesRef.current, userMessage]; setMessages(messagesRef.current); - setChatState('streaming'); + setChatState('waiting'); setError(null); setPendingTool(null); try { - await sendReply(currentSession.id, requestId, userMessage); + const createdRun = await createAgentRun(currentSession.id, requestId, userMessage); + const finishedRun = + createdRun.status === 'succeeded' ? createdRun : await waitForAgentRun(createdRun.id); + if (!finishedRun.sessionId) { + throw new Error('后台任务已提交，但未返回会话'); + } + setChatState('streaming'); } catch (err) { if (err instanceof ApiError && err.status === 402) { setNotice('余额不足，请充值后继续使用'); diff --git a/src/hooks/useTKMindChat.ts b/src/hooks/useTKMindChat.ts index 2163eb3..db3e21f 100644 --- a/src/hooks/useTKMindChat.ts +++ b/src/hooks/useTKMindChat.ts @@ -5,7 +5,9 @@ import { bootstrapProjectMemory, cancelRequest, confirmTool, + createAgentRun, deleteChatSession, + getAgentRun, getMindSpace, getMe, listNotifications, @@ -17,8 +19,6 @@ import { rememberProjectContext, uploadMindSpaceAsset, resumeSession, - sendReply, - startSession, subscribeNotificationEvents, subscribeSessionEvents, updateProvider, @@ -69,9 +69,23 @@ import { const INSUFFICIENT_BALANCE_NOTICE = '余额不足，请充值后继续使用'; const REPLY_RECOVERY_SYNC_DELAYS_MS = [1500, 5000, 12000]; +const AGENT_RUN_POLL_DELAY_MS = 1200; +const AGENT_RUN_MAX_POLLS = 100; export { INSUFFICIENT_BALANCE_NOTICE }; +async function waitForAgentRun(runId: string) { + for (let attempt = 0; attempt < AGENT_RUN_MAX_POLLS; attempt += 1) { + const run = await getAgentRun(runId); + if (run.status === 'succeeded') return run; + if (run.status === 'failed') { + throw new Error(run.error || '后台任务失败，请稍后重试'); + } + await new Promise((resolve) => window.setTimeout(resolve, AGENT_RUN_POLL_DELAY_MS)); + } + throw new Error('后台任务仍在排队，请稍后刷新会话查看结果'); +} + function isAmbiguousReplySubmitError(err: unknown) { if (!(err instanceof ApiError)) return true; return err.status === 0 || err.status === 409 || err.status >= 500; @@ -444,9 +458,15 @@ export function useTKMindChat( } const retryRequestId = crypto.randomUUID(); activeRequestId.current = retryRequestId; - setChatState('streaming'); + setChatState('waiting'); setError(null); - await sendReply(sessionRef.current!.id, retryRequestId, lastUser); + const retryRun = await createAgentRun(sessionRef.current!.id, retryRequestId, lastUser); + const finishedRetryRun = + retryRun.status === 'succeeded' ? retryRun : await waitForAgentRun(retryRun.id); + if (!finishedRetryRun.sessionId) { + throw new Error('后台任务已提交，但未返回会话'); + } + setChatState('streaming'); } catch (err) { setError(err instanceof Error ? err.message : String(err)); setChatState('error'); @@ -741,7 +761,7 @@ export function useTKMindChat( const switchSession = useCallback( (sessionId: string) => { - if (session?.id === sessionId || chatState === 'streaming') return; + if (session?.id === sessionId || chatState === 'streaming' || chatState === 'waiting') return; resetSessionView(); const token = connectTokenRef.current; @@ -776,7 +796,12 @@ export function useTKMindChat( (value) => typeof value === 'string' && value.trim(), ); if (!text.trim() && normalizedImageUrls.length === 0) return; - if (chatState === 'streaming' || chatState === 'loading' || chatState === 'connecting') return; + if ( + chatState === 'streaming' || + chatState === 'loading' || + chatState === 'connecting' || + chatState === 'waiting' + ) return; const trimmed = text.trim(); const mindspacePrefix = options?.mindspaceContext @@ -795,41 +820,48 @@ export function useTKMindChat( activeRequestId.current = requestId; messagesRef.current = [...messagesRef.current, userMessage]; setMessages(messagesRef.current); - setChatState('streaming'); + setChatState('waiting'); setError(null); setPendingTool(null); let activeSessionId = session?.id ?? null; - if (!activeSessionId) { - try { - const created = await startSession(); - activeSessionId = created.id; - writeStoredSessionId(userRef.current?.id, created.id); - setSession(created); - setSessions((prev) => prependUnique(prev, created)); - subscribeToSession(created.id); - void ensureProvider(created.id); - void loadProjectMemory(created.id, false); - void refreshSessions(); - } catch (err) { - if (err instanceof ApiError && err.status === 402) { - notifyInsufficientBalance(); - } else { - setError(err instanceof Error ? err.message : String(err)); - setChatState('error'); - } - activeRequestId.current = null; - return; - } - } else { + if (activeSessionId) { setSessions((prev) => touchSession(prev, activeSessionId!, 1)); } try { - await sendReply(activeSessionId, requestId, userMessage); + const createdRun = await createAgentRun(activeSessionId, requestId, userMessage); + const finishedRun = + createdRun.status === 'succeeded' ? createdRun : await waitForAgentRun(createdRun.id); + activeSessionId = finishedRun.sessionId; + if (!activeSessionId) { + throw new Error('后台任务已提交，但未返回会话'); + } + if (!session?.id || session.id !== activeSessionId) { + const nextSession: Session = { + id: activeSessionId, + name: 'New Chat', + message_count: messagesRef.current.length, + working_dir: '', + }; + writeStoredSessionId(userRef.current?.id, activeSessionId); + setSession(nextSession); + setSessions((prev) => prependUnique(prev, nextSession)); + void loadSessionDetail(activeSessionId) + .then(({ session: loaded }) => { + setSession(loaded); + setSessions((prev) => prependUnique(prev, loaded)); + }) + .catch(() => {}); + void ensureProvider(activeSessionId); + void loadProjectMemory(activeSessionId, false); + void refreshSessions(); + } + subscribeToSession(activeSessionId); + setChatState('streaming'); } catch (err) { - if (isAmbiguousReplySubmitError(err)) { + if (activeSessionId && isAmbiguousReplySubmitError(err)) { subscribeToSession(activeSessionId); setChatState('streaming'); setError(null); @@ -842,7 +874,7 @@ export function useTKMindChat( } return; } - if (session) setSessions((prev) => touchSession(prev, activeSessionId!, -1)); + if (session && activeSessionId) setSessions((prev) => touchSession(prev, activeSessionId!, -1)); if (err instanceof ApiError && err.status === 402) { notifyInsufficientBalance(); } else { @@ -886,7 +918,7 @@ export function useTKMindChat( ); const newSession = useCallback(() => { - if (chatState === 'streaming') return; + if (chatState === 'streaming' || chatState === 'waiting') return; connectTokenRef.current += 1; unsubscribeRef.current?.(); @@ -902,12 +934,17 @@ export function useTKMindChat( }, [chatState]); const refreshProjectMemory = useCallback(async () => { - if (!session || chatState === 'streaming') return; + if (!session || chatState === 'streaming' || chatState === 'waiting') return; await loadProjectMemory(session.id, true); }, [chatState, loadProjectMemory, session]); const rememberCurrentContext = useCallback(async () => { - if (chatState === 'streaming' || chatState === 'loading' || chatState === 'connecting') return; + if ( + chatState === 'streaming' || + chatState === 'loading' || + chatState === 'connecting' || + chatState === 'waiting' + ) return; await rememberRecentContext({ silent: false }); }, [chatState, rememberRecentContext]); @@ -917,7 +954,7 @@ export function useTKMindChat( const deleteSession = useCallback( (sessionId: string) => { - if (chatState === 'streaming') return; + if (chatState === 'streaming' || chatState === 'waiting') return; void (async () => { try { diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index 0ebf587..a7c2875 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -363,14 +363,16 @@ export function createTkmindProxy({ recipe = null, } = {}, ) { + const resolvedWorkingDir = workingDir ?? await userAuth.resolveWorkingDir(userId); + const resolvedSessionPolicy = sessionPolicy ?? await userAuth.getAgentSessionPolicy(userId); const startTarget = await pickTarget(); const upstream = await apiFetch(startTarget, apiSecret, '/agent/start', { method: 'POST', body: JSON.stringify({ - ...(workingDir ? { working_dir: workingDir } : {}), - enable_context_memory: sessionPolicy?.enableContextMemory, - ...(sessionPolicy?.extensionOverrides - ? { extension_overrides: sessionPolicy.extensionOverrides } + ...(resolvedWorkingDir ? { working_dir: resolvedWorkingDir } : {}), + enable_context_memory: resolvedSessionPolicy?.enableContextMemory, + ...(resolvedSessionPolicy?.extensionOverrides + ? { extension_overrides: resolvedSessionPolicy.extensionOverrides } : {}), ...(recipe ? { recipe } : {}), }), @@ -384,12 +386,12 @@ export function createTkmindProxy({ throw new Error('创建会话失败：缺少 session id'); } await userAuth.registerAgentSession(userId, session.id, startTarget); - if (sessionPolicy?.gooseMode) { + if (resolvedSessionPolicy?.gooseMode) { const modeRes = await apiFetch(startTarget, apiSecret, '/agent/update_session', { method: 'POST', body: JSON.stringify({ session_id: session.id, - goose_mode: sessionPolicy.gooseMode, + goose_mode: resolvedSessionPolicy.gooseMode, }), }); if (!modeRes.ok) { @@ -397,6 +399,29 @@ export function createTkmindProxy({ throw new Error(modeText || `设置会话模式失败 (${modeRes.status})`); } } + const publishLayout = await userAuth.getUserPublishLayout(userId); + const userMemories = conversationMemoryService?.listMemories + ? await conversationMemoryService.listMemories(userId, { limit: 40 }).catch(() => []) + : []; + await reconcileAgentSession( + (pathname, init) => apiFetch(startTarget, apiSecret, pathname, init), + session.id, + { + workingDir: resolvedWorkingDir, + sessionPolicy: resolvedSessionPolicy, + sandboxConstraints: publishLayout?.constraints ?? null, + userMemories, + userContext: publishLayout + ? { + userId, + displayName: publishLayout.displayName, + username: publishLayout.username, + slug: publishLayout.slug, + } + : null, + }, + ); + await applySessionLlmProvider(session.id); return session; } @@ -499,6 +524,68 @@ export function createTkmindProxy({ ); } + async function submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { + if (!userId || !sessionId) throw new Error('缺少会话信息'); + const owns = await userAuth.ownsSession(userId, sessionId); + if (!owns) throw new Error('无权访问该会话'); + const gate = await userAuth.canUseChat(userId); + if (!gate.ok) { + const err = new Error(gate.message ?? '余额不足，请充值后继续使用'); + err.code = gate.code; + err.status = 402; + throw err; + } + + await reconcileSessionPolicyForUser(userId, sessionId); + await applySessionLlmProvider(sessionId); + + const user = await userAuth.getUserById(userId); + if (!user) throw new Error('用户不存在'); + let finalUserMessage = userMessage; + if (llmProviderService && messageHasImages(userMessage) && await llmProviderService.hasVisionKey()) { + const publishLayout = await userAuth.getUserPublishLayout(userId).catch(() => null); + const visionResult = await buildVisionBody(userMessage, userId, publishLayout).catch(() => null); + if (visionResult?.userMessage) { + finalUserMessage = visionResult.userMessage; + } + if (visionResult?.billableImageCount > 0 && subscriptionService) { + await subscriptionService.consumeImageQuota(userId, visionResult.billableImageCount).catch((err) => { + console.warn( + 'Subscription image quota consume skipped:', + err instanceof Error ? err.message : err, + ); + }); + } + } + const policyState = await userAuth.resolveUserPolicies(user); + let body = { + request_id: requestId, + user_message: finalUserMessage, + }; + if (!policyState.unrestricted) { + body = injectTaskRoutingHint( + body, + await userAuth.getAgentSessionPolicy(userId), + ); + } + + const target = await resolveTarget(sessionId); + const upstream = await apiFetch( + target, + apiSecret, + `/sessions/${encodeURIComponent(sessionId)}/reply`, + { + method: 'POST', + body: JSON.stringify(body), + }, + ); + if (!upstream.ok) { + const text = await upstream.text().catch(() => ''); + throw new Error(text || `发送失败 (${upstream.status})`); + } + return { ok: true }; + } + const requireUser = async (req, res, next) => { try { const session = req.userSession; @@ -877,6 +964,16 @@ export function createTkmindProxy({ }); return; } + const isAgentRunPath = + (req.method === 'POST' && pathname === '/agent/runs') + || (req.method === 'GET' && /^\/agent\/runs\/[^/]+$/.test(pathname)); + if (isAgentRunPath) { + res.status(503).json({ + message: 'Agent Run 接口需在 Portal 本地处理，请确认 server.mjs 已更新并重启后端', + code: 'AGENT_RUNS_NATIVE_REQUIRED', + }); + return; + } const policyState = await userAuth.resolveUserPolicies(req.currentUser); const capabilityState = await userAuth.resolveUserCapabilities(req.currentUser); const gate = evaluateProxyRequest(req.method, pathname, policyState.policies, { @@ -896,6 +993,13 @@ export function createTkmindProxy({ } const isReplyPath = pathname.match(/^\/sessions\/[^/]+\/reply$/) && req.method === 'POST'; const sessionMatch = pathname.match(/^\/sessions\/([^/]+)/); + if (isReplyPath) { + res.status(410).json({ + message: '聊天提交入口已统一为 POST /agent/runs', + code: 'AGENT_RUNS_REQUIRED', + }); + return; + } let baseBody = req.body; if (isReplyPath && !policyState.unrestricted) { @@ -983,6 +1087,7 @@ export function createTkmindProxy({ proxySessionEvents, resolveTarget, startSessionForUser, + submitSessionReplyForUser, apiFetch: async (pathname, init) => apiFetch(await pickTarget(), apiSecret, pathname, init), apiFetchTo: (target, pathname, init) => apiFetch(target, apiSecret, pathname, init), };