diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 929c724..b7d6fe3 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -12,9 +12,11 @@ import { } from './conversation-transcript-persist.mjs'; import { ensureGooseUserMessageMetadata } from './goose-message.mjs'; import { + prepareAndDetectSessionDeliverables, SESSION_FINISHED_STALE_GRACE_MS, tryRecoverRunFromDeliverables, } from './agent-run-deliverable-check.mjs'; +import { isPageDataIntent } from './chat-skills.mjs'; const DEFAULT_RUN_RETRY_DELAYS_MS = [1500, 5000, 15000]; const TERMINAL_STATUSES = new Set(['succeeded', 'failed']); @@ -53,6 +55,17 @@ function serializeMessage(message) { return JSON.stringify(message ?? {}); } +function extractRunMessageText(row) { + const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {}; + if (typeof message.content === 'string') return message.content; + if (!Array.isArray(message.content)) return ''; + return message.content + .filter((item) => item?.type === 'text') + .map((item) => String(item.text ?? '').trim()) + .filter(Boolean) + .join('\n'); +} + function positiveInteger(value, fallback) { const n = Number(value); if (!Number.isFinite(n) || n <= 0) return fallback; @@ -247,6 +260,8 @@ export function createAgentRunGateway({ sessionSnapshotService = null, conversationMemoryService = null, syncUserPagesOnSuccess = null, + observePersonalMemoryOnSuccess = null, + isSessionExternallyBusy = null, retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS, autoDispatch = envFlag(process.env.MEMIND_AGENT_RUN_AUTODISPATCH, true), maxConcurrentRuns = positiveInteger( @@ -348,6 +363,15 @@ export function createAgentRunGateway({ // Prevent multiple concurrent agent runs on the same Goose session, which would // cause replies to arrive out of order and appear garbled in the chat UI. if (sessionId && !isDirectChatSessionId(sessionId)) { + if (typeof isSessionExternallyBusy === 'function' && await isSessionExternallyBusy({ + userId, + sessionId, + })) { + const conflict = new Error('该会话正在完成页面交付或自动修复，请稍候再发送'); + conflict.code = 'SESSION_RUN_CONFLICT'; + conflict.status = 409; + throw conflict; + } const [activeRows] = await pool.query( `SELECT id FROM h5_agent_runs WHERE agent_session_id = ? AND status NOT IN ('succeeded', 'failed') @@ -765,19 +789,52 @@ export function createAgentRunGateway({ } async function finalizeSuccessfulRun(runId, row, sessionId) { + let deliveryResult = null; + if (typeof syncUserPagesOnSuccess === 'function') { + deliveryResult = await syncUserPagesOnSuccess({ + userId: row.user_id, + sessionId, + runId, + }); + } + const pageDataErrors = Array.isArray(deliveryResult?.pageDataBind?.errors) + ? deliveryResult.pageDataBind.errors + : []; + if (pageDataErrors.length > 0) { + const error = new Error(`Page Data 页面绑定失败：${pageDataErrors.map((item) => item?.message ?? item?.code ?? 'unknown').join('; ')}`); + error.code = 'PAGE_DATA_DELIVERY_FAILED'; + error.retryable = false; + throw error; + } + if (isPageDataIntent(extractRunMessageText(row))) { + const latest = await getRunById(runId); + const deliverables = await prepareAndDetectSessionDeliverables({ + pool, + userId: row.user_id, + sessionId, + runStartedAtMs: latest?.started_at ?? row.started_at ?? null, + }); + if (deliverables.pageCount < 1) { + const error = new Error('Page Data 任务未生成可交付页面，不能标记成功'); + error.code = 'PAGE_DATA_DELIVERABLE_MISSING'; + error.retryable = false; + throw error; + } + } await markRun(runId, 'succeeded', { agent_session_id: sessionId, completed_at: nowMs(), error_message: null, }); - if (typeof syncUserPagesOnSuccess === 'function') { - await syncUserPagesOnSuccess({ + if (typeof observePersonalMemoryOnSuccess === 'function') { + await observePersonalMemoryOnSuccess({ userId: row.user_id, sessionId, runId, + userMessage: parseDbJsonColumn(row.user_message_json, {}), }).catch((err) => { console.warn( - '[AgentRun] workspace page deliver failed:', + '[AgentRun] personal memory shadow observation failed:', err instanceof Error ? err.message : err, ); }); diff --git a/chat-agent-run-gate.mjs b/chat-agent-run-gate.mjs index d254091..2445459 100644 --- a/chat-agent-run-gate.mjs +++ b/chat-agent-run-gate.mjs @@ -10,14 +10,15 @@ */ /** - * @param {{ chatState?: string; finishedViaPortalDirectChat?: boolean }} input + * @param {{ chatState?: string; finishedViaPortalDirectChat?: boolean; agentRunSucceeded?: boolean }} input * @returns {'idle' | 'streaming'} */ export function resolvePostAgentRunChatState({ chatState = 'waiting', finishedViaPortalDirectChat = false, + agentRunSucceeded = false, } = {}) { - if (finishedViaPortalDirectChat) return 'idle'; + if (finishedViaPortalDirectChat || agentRunSucceeded) return 'idle'; if (chatState === 'idle') return 'idle'; return 'streaming'; } diff --git a/chat-agent-run-gate.test.mjs b/chat-agent-run-gate.test.mjs index 97b170d..51a0e85 100644 --- a/chat-agent-run-gate.test.mjs +++ b/chat-agent-run-gate.test.mjs @@ -25,6 +25,17 @@ test('resolvePostAgentRunChatState prefers portal direct chat completion', () => ); }); +test('resolvePostAgentRunChatState idles after worker-side agent run success', () => { + assert.equal( + resolvePostAgentRunChatState({ chatState: 'waiting', agentRunSucceeded: true }), + 'idle', + ); + assert.equal( + resolvePostAgentRunChatState({ chatState: 'streaming', agentRunSucceeded: true }), + 'idle', + ); +}); + test('shouldPromoteSessionIdToStreaming skips re-streaming after Finish', () => { assert.equal(shouldPromoteSessionIdToStreaming('idle'), false); assert.equal(shouldPromoteSessionIdToStreaming('waiting'), true); diff --git a/server.mjs b/server.mjs index 01134d2..31f87ad 100644 --- a/server.mjs +++ b/server.mjs @@ -192,6 +192,7 @@ import { createSessionSnapshotService } from './session-snapshot.mjs'; import { createConversationMemoryService } from './conversation-memory.mjs'; import { createManagedMemoryV2Runtime } from './memory-v2-runtime.mjs'; import { createMemoryV2AdminConfigService } from './memory-v2-admin-config.mjs'; +import { createSkillRuntimeAdminConfigService } from './skill-runtime-admin-config.mjs'; import { createWechatScheduleLlmConfigService } from './wechat-schedule-llm-config.mjs'; import { createExperienceService } from './experience-service.mjs'; import { attachAsrRoutes } from './asr-proxy.mjs'; @@ -202,24 +203,16 @@ import { syncPageDataPolicyAccessMode } from './page-data-publish-sync.mjs'; import { upsertPageDataPolicyIndex } from './page-data-policy-index.mjs'; import { attachShenmeiOpinionFormRoutes } from './shenmei-opinion-form-routes.mjs'; import { isNativeH5ApiPath } from './policies.mjs'; +import { + applyMemindRuntimeProfile, + describeMemindRuntimeProfile, + loadMemindEnvFiles, +} from './scripts/memind-runtime-profile.mjs'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); -function loadEnvFile(filePath) { - if (!fs.existsSync(filePath)) return; - for (const line of fs.readFileSync(filePath, 'utf8').split('\n')) { - const trimmed = line.trim(); - if (!trimmed || trimmed.startsWith('#')) continue; - const eq = trimmed.indexOf('='); - if (eq < 0) continue; - const key = trimmed.slice(0, eq).trim(); - const value = trimmed.slice(eq + 1).trim(); - if (!process.env[key]) process.env[key] = value; - } -} - -loadEnvFile(path.join(__dirname, '../../.env.local')); -loadEnvFile(path.join(__dirname, '.env')); +loadMemindEnvFiles(__dirname); +applyMemindRuntimeProfile({ rootDir: __dirname }); function parseApiTargets() { const csvTargets = (process.env.TKMIND_API_TARGETS ?? '') @@ -349,6 +342,15 @@ async function unregisterAgentSessionForUser(userId, sessionId) { await access.unregisterSession({ userId, sessionId }); } let agentRunGateway = null; +const sessionPageDeliveryLocks = new Map(); +function beginSessionPageDelivery(sessionId) { + sessionPageDeliveryLocks.set(sessionId, Number(sessionPageDeliveryLocks.get(sessionId) ?? 0) + 1); +} +function endSessionPageDelivery(sessionId) { + const remaining = Number(sessionPageDeliveryLocks.get(sessionId) ?? 0) - 1; + if (remaining > 0) sessionPageDeliveryLocks.set(sessionId, remaining); + else sessionPageDeliveryLocks.delete(sessionId); +} let chatIntentRouter = null; let toolGateway = null; let directChatService = null; @@ -356,6 +358,7 @@ let sessionSnapshotService = null; let conversationMemoryService = null; let memoryV2 = null; let memoryV2ConfigService = null; +let skillRuntimeConfigService = null; let wechatScheduleLlmConfigService = null; let mindSpace = null; let mindSpaceAssets = null; @@ -639,6 +642,7 @@ async function bootstrapUserAuth() { console.warn('LLM provider boot sync skipped:', err instanceof Error ? err.message : err); }); memoryV2ConfigService = createMemoryV2AdminConfigService(pool); + skillRuntimeConfigService = createSkillRuntimeAdminConfigService(pool, { h5Root: __dirname }); wechatScheduleLlmConfigService = createWechatScheduleLlmConfigService(pool); conversationMemoryService = createConversationMemoryService(pool, { llmProviderService, @@ -705,9 +709,16 @@ async function bootstrapUserAuth() { chatIntentRouter, sessionSnapshotService, conversationMemoryService, - syncUserPagesOnSuccess: async ({ userId }) => { - await syncUserGeneratedPages(userId); + observePersonalMemoryOnSuccess: async ({ userId, sessionId, userMessage }) => { + if (!memoryV2?.observePersonalMemory) return; + await memoryV2.observePersonalMemory({ + userId, + sessionId, + messages: [userMessage], + }); }, + syncUserPagesOnSuccess: async ({ userId }) => syncUserGeneratedPages(userId), + isSessionExternallyBusy: ({ sessionId }) => Number(sessionPageDeliveryLocks.get(sessionId) ?? 0) > 0, autoDispatch: ['1', 'true', 'yes', 'on'].includes( String(process.env.MEMIND_AGENT_RUN_AUTODISPATCH ?? '1').trim().toLowerCase(), ), @@ -851,6 +862,16 @@ app.use(attachUserSession); // ============ Legacy password auth ============ +async function resolveSkillRuntimeForClient() { + if (!skillRuntimeConfigService?.getPublicRuntimeConfig) return null; + try { + return await skillRuntimeConfigService.getPublicRuntimeConfig(); + } catch (err) { + console.warn('[SkillRuntime] public config unavailable:', err instanceof Error ? err.message : err); + return null; + } +} + app.get('/auth/status', async (req, res) => { await userAuthReady; if (userAuth) { @@ -859,6 +880,7 @@ app.get('/auth/status', async (req, res) => { if (!me) return res.json({ authenticated: false, mode: 'user' }); const row = await userAuth.getUserById(me.id); const capabilityState = await userAuth.resolveUserCapabilities(row); + const skillRuntime = await resolveSkillRuntimeForClient(); return res.json({ authenticated: true, user: me, @@ -866,6 +888,7 @@ app.get('/auth/status', async (req, res) => { capabilities: capabilityState.capabilities, grantedSkills: capabilityState.grantedSkills ?? [], unrestricted: capabilityState.unrestricted, + skillRuntime, }); } catch (err) { console.error('[Auth] status failed:', err instanceof Error ? err.message : err); @@ -1309,10 +1332,11 @@ app.get('/auth/me', async (req, res) => { if (!userAuth) return res.status(503).json({ message: '未启用用户系统' }); const me = await userAuth.getMe(userToken(req)); if (!me) return res.status(401).json({ message: '未登录' }); - const [paths, capabilityState, subscription] = await Promise.all([ + const [paths, capabilityState, subscription, skillRuntime] = await Promise.all([ userAuth.listPathGrants(me.id), userAuth.resolveUserCapabilities(await userAuth.getUserById(me.id)), subscriptionService ? subscriptionService.getActiveSubscription(me.id) : null, + resolveSkillRuntimeForClient(), ]); return res.json({ user: { ...me, subscription }, @@ -1320,6 +1344,7 @@ app.get('/auth/me', async (req, res) => { capabilities: capabilityState.capabilities, grantedSkills: capabilityState.grantedSkills ?? [], unrestricted: capabilityState.unrestricted, + skillRuntime, }); }); @@ -3209,11 +3234,10 @@ const SAVE_TARGET_CATEGORIES = new Set(['draft', 'oa', 'public']); async function syncUserGeneratedPages(userId) { if (!userId) return; if (workspacePageDeliver?.syncAndDeliver) { - await workspacePageDeliver.syncAndDeliver(userId); - return; + return await workspacePageDeliver.syncAndDeliver(userId); } if (!mindSpacePageSync) return; - await mindSpacePageSync.syncUserGeneratedPages(userId); + return await mindSpacePageSync.syncUserGeneratedPages(userId); } async function resolveChatSaveBundle(user, h5Root, input = {}) { @@ -5004,6 +5028,8 @@ api.get('/sessions/:sessionId/events', async (req, res, next) => { // workspace HTML into the asset store before a later restart rebuilds the // workspace from DB-backed assets only. const onAfterFinish = async (sid, uid) => { + beginSessionPageDelivery(sid); + try { const apiFetchFn = async (pathname, init) => { const target = await tkmindProxy.resolveTarget(sid); return tkmindProxy.apiFetchTo(target, pathname, init); @@ -5085,6 +5111,20 @@ api.get('/sessions/:sessionId/events', async (req, res, next) => { tkmindProxy, userText: lastUserText, }); + if (lastUserMessage && memoryV2?.observePersonalMemory) { + await memoryV2.observePersonalMemory({ + userId: uid, + sessionId: sid, + messages: [lastUserMessage], + }).catch((err) => { + console.warn( + `[memory-v2] finish shadow observation skipped for session ${sid}: ${err instanceof Error ? err.message : err}`, + ); + }); + } + } finally { + endSessionPageDelivery(sid); + } }; return tkmindProxy.proxySessionEvents(req, res, sessionId, { onAfterFinish, @@ -6233,6 +6273,7 @@ app.get('*', (_req, res) => { process.exit(1); } const server = app.listen(PORT, HOST, () => { + console.log(`[Portal] Runtime profile: ${describeMemindRuntimeProfile()}`); console.log(`TKMind H5 @ http://${HOST}:${PORT}`); console.log(`Proxy -> ${API_TARGETS.join(', ')}`); console.log(`Auth -> ${enabled ? 'multi-user (MySQL)' : legacyAuth ? 'legacy password' : 'disabled'}`);