Compare commits

..

2 Commits

Author SHA1 Message Date
john aecde46ff6 fix(agent): restore active-task routing and active-mode validation observation
Memind CI / Test, build, and release guards (push) Successful in 5m23s
Keep correction follow-ups on Agent using session transcript and active task
context, and let Page Data validation reach Orchestrator when mode is active
instead of requiring shadowEngine.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-27 12:15:15 +08:00
john f95d766f69 feat(billing): bill Cursor executor usage via stream-json token parsing
Memind CI / Test, build, and release guards (push) Failing after 14m36s
Parse Cursor CLI usage from stream-json output and charge through billSessionUsage so subscription quota is consumed first and wallet overage applies when needed.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-27 10:16:38 +08:00
13 changed files with 1006 additions and 28 deletions
+207 -5
View File
@@ -3,7 +3,7 @@ import fs from 'node:fs/promises';
import path from 'node:path';
import { isRunStreamReplayEnabled } from './agent-run-stream.mjs';
import { isDirectChatSessionId } from './direct-chat-service.mjs';
import { CHAT_INTENT_ROUTE, resolveGatewayAgentSessionId, resolveLegacyRouteFromClassification, logRouterDecisionShadow } from './chat-intent-router.mjs';
import { CHAT_INTENT_ROUTE, formatRouterTranscript, resolveGatewayAgentSessionId, resolveLegacyRouteFromClassification, logRouterDecisionShadow } from './chat-intent-router.mjs';
import { resolveSessionAccess } from './session-broker.mjs';
import {
loadSnapshotMessages,
@@ -44,6 +44,7 @@ import {
} from './executor-display-label.mjs';
import { applyCursorFirstAgentExecution } from './cursor-page-routing.mjs';
import { cursorDeepseekFallbackEnabled } from './cursor-agent-launch.mjs';
import { buildCursorBillingTokenState } from './cursor-agent-usage.mjs';
const DEFAULT_RUN_RETRY_DELAYS_MS = [1500, 5000, 15000];
const TERMINAL_STATUSES = new Set(['succeeded', 'failed']);
@@ -369,13 +370,83 @@ function summarizeText(value, limit = TOOL_GATEWAY_SUMMARY_LIMIT) {
export function buildCodeRunCompletionReply(result) {
const executor = resolveExecutorDisplayLabel(result?.executor, result?.executorLabel);
const output = sanitizeUserFacingBrandText(summarizeText(result?.stdout, 2400).trim());
const rawOutput = result?.displayStdout ?? result?.stdout;
const output = sanitizeUserFacingBrandText(summarizeText(rawOutput, 2400).trim());
return [
`已由 ${executor} 完成执行,并通过平台文件验收。`,
output ? `\n${output}` : '',
].join('').trim();
}
async function billCursorToolGatewayUsage({
userAuth,
userId,
sessionId,
requestId,
usage,
}) {
if (!usage || !sessionId || !userAuth?.billSessionUsage) return null;
const prior = typeof userAuth.getBillingState === 'function'
? await userAuth.getBillingState(sessionId).catch(() => null)
: null;
const tokenState = buildCursorBillingTokenState(usage, prior);
if (!tokenState) return null;
return userAuth.billSessionUsage(userId, sessionId, tokenState, requestId);
}
function resolveCursorBillingSessionId({ deliverySessionId, agentSessionId, runId }) {
const delivery = String(deliverySessionId ?? '').trim();
if (delivery) return delivery;
const session = String(agentSessionId ?? '').trim();
if (session) return session;
const run = String(runId ?? '').trim();
return run ? `h5run_${run}` : null;
}
async function recordCursorToolGatewayBilling({
appendEvent,
runId,
row,
result,
userAuth,
deliverySessionId = null,
}) {
if (String(result?.executor ?? '').trim().toLowerCase() !== 'cursor' || !result?.usage) {
return null;
}
const billingSessionId = resolveCursorBillingSessionId({
deliverySessionId,
agentSessionId: row.agent_session_id,
runId,
});
try {
const billing = await billCursorToolGatewayUsage({
userAuth,
userId: row.user_id,
sessionId: billingSessionId,
requestId: row.request_id,
usage: result.usage,
});
await appendEvent(runId, 'tool_gateway_billed', {
sessionId: billingSessionId,
executor: result.executor ?? null,
usage: result.usage,
billed: Boolean(billing?.ok),
costCents: billing?.costCents ?? 0,
deltaInputTokens: billing?.deltaInputTokens ?? 0,
deltaOutputTokens: billing?.deltaOutputTokens ?? 0,
});
return billing;
} catch (err) {
await appendEvent(runId, 'tool_gateway_billing_failed', {
sessionId: billingSessionId,
executor: result.executor ?? null,
message: err instanceof Error ? err.message : String(err),
});
return null;
}
}
function normalizeExpectedFileCheck(value) {
if (typeof value === 'string') {
const expectedPath = value.trim();
@@ -546,6 +617,103 @@ function resolveEffectiveToolMode(runOptions) {
return runOptions?.pageDataAiderWorkflow ? 'chat' : (runOptions?.toolMode ?? 'chat');
}
const ACTIVE_TASK_CONTEXT_MAX_AGE_MS = 4 * 60 * 60 * 1000;
function parseIntentRoutedEvent(dataJson) {
if (!dataJson) return null;
const payload = typeof dataJson === 'string'
? (() => {
try {
return JSON.parse(dataJson);
} catch {
return null;
}
})()
: dataJson;
if (!payload || typeof payload !== 'object') return null;
return {
route: payload.route ?? payload.decision?.route ?? null,
suggestedSkill: payload.suggestedSkill ?? payload.suggested_skill ?? null,
reason: payload.reason ?? null,
};
}
export async function resolveActiveTaskContext(pool, {
userId,
sessionId,
excludeRunId = null,
nowMs: nowMsFn = Date.now,
} = {}) {
if (!pool || !userId || !sessionId || isDirectChatSessionId(sessionId)) {
return null;
}
const params = [userId, sessionId];
let excludeClause = '';
if (excludeRunId) {
excludeClause = ' AND r.id <> ?';
params.push(excludeRunId);
}
const [rows] = await pool.query(
`SELECT r.id, r.status, r.error_message, r.created_at, r.updated_at,
(
SELECT e.data_json
FROM h5_agent_run_events e
WHERE e.run_id = r.id AND e.event_type = 'intent_routed'
ORDER BY e.created_at ASC
LIMIT 1
) AS intent_json
FROM h5_agent_runs r
WHERE r.user_id = ? AND r.agent_session_id = ?${excludeClause}
ORDER BY r.created_at DESC
LIMIT 1`,
params,
);
const row = rows?.[0];
if (!row) return null;
const intent = parseIntentRoutedEvent(row.intent_json);
const legacyRoute = String(intent?.route ?? '').trim();
const agentRoute = legacyRoute === 'agent'
|| legacyRoute === CHAT_INTENT_ROUTE.AGENT
|| legacyRoute === 'agent_orchestration';
if (!agentRoute) return null;
const referenceMs = Math.max(Number(row.updated_at) || 0, Number(row.created_at) || 0);
if (referenceMs > 0 && nowMsFn() - referenceMs > ACTIVE_TASK_CONTEXT_MAX_AGE_MS) {
return null;
}
return {
hasRecentAgentTask: true,
lastRunId: row.id,
lastRunStatus: String(row.status ?? '').trim() || null,
lastRunFailed: String(row.status ?? '').trim() === 'failed',
lastSuggestedSkill: intent?.suggestedSkill ?? null,
lastRoute: legacyRoute,
lastIntentReason: intent?.reason ?? null,
lastErrorMessage: row.error_message
? String(row.error_message).slice(0, 500)
: null,
};
}
export async function resolveRouterTranscript(sessionSnapshotService, {
sessionId,
userId = null,
excludeLatestUserText = null,
} = {}) {
if (!sessionSnapshotService?.get || !sessionId || isDirectChatSessionId(sessionId)) {
return '';
}
const snapshot = await sessionSnapshotService.get(sessionId).catch(() => null);
if (!snapshot) return '';
if (userId && snapshot?.session?.user_id && snapshot.session.user_id !== userId) {
return '';
}
const messages = Array.isArray(snapshot?.messages) ? snapshot.messages : [];
return formatRouterTranscript(messages, { excludeLatestUserText });
}
function restoreCursorFallbackUserMessage(userMessage) {
const message = (userMessage && typeof userMessage === 'object' && !Array.isArray(userMessage))
? { ...userMessage }
@@ -1476,13 +1644,29 @@ export function createAgentRunGateway({
}
}
async function resolveRunRouting(row, userMessage, runOptions) {
async function resolveRunRouting(row, userMessage, runOptions, { runId = null } = {}) {
if (!chatIntentRouter?.classify) return null;
const enabled = chatIntentRouter.isEnabled
? await Promise.resolve(chatIntentRouter.isEnabled()).catch(() => false)
: true;
if (!enabled) return null;
const grantedSkills = await resolveGrantedSkills(row.user_id);
const activeTaskContext = await resolveActiveTaskContext(pool, {
userId: row.user_id,
sessionId: row.agent_session_id ?? null,
excludeRunId: runId,
}).catch(() => null);
const displayText = userMessage?.metadata?.displayText
?? deriveUserFacingText(
typeof userMessage?.content?.find === 'function'
? (userMessage.content.find((item) => item?.type === 'text')?.text ?? '')
: '',
);
const recentTranscript = await resolveRouterTranscript(sessionSnapshotService, {
sessionId: row.agent_session_id ?? null,
userId: row.user_id,
excludeLatestUserText: displayText,
}).catch(() => '');
return chatIntentRouter.classify({
userId: row.user_id,
userMessage,
@@ -1491,6 +1675,8 @@ export function createAgentRunGateway({
toolMode: runOptions.toolMode,
forceDeepReasoning: runOptions.forceDeepReasoning,
grantedSkills,
activeTaskContext,
recentTranscript,
});
}
@@ -1553,7 +1739,7 @@ export function createAgentRunGateway({
policyBlocked: true,
};
}
const routing = await resolveRunRouting(row, userMessage, runOptions);
const routing = await resolveRunRouting(row, userMessage, runOptions, { runId });
const routingDecision = resolveLegacyRouteFromClassification(routing) ?? routing?.route ?? null;
const cursorFirstApplied = applyCursorFirstAgentExecution(userMessage, runOptions, {
routingDecision,
@@ -1778,8 +1964,9 @@ export function createAgentRunGateway({
executor: result.executor ?? null,
dryRun: Boolean(result.dryRun),
exitCode: result.exitCode ?? null,
stdoutTail: summarizeText(result.stdout),
stdoutTail: summarizeText(result.displayStdout ?? result.stdout),
stderrTail: summarizeText(result.stderr),
usage: result.usage ?? null,
});
try {
const validation = await validateToolGatewayResult({
@@ -1828,8 +2015,23 @@ export function createAgentRunGateway({
sessionId: delivery.sessionId,
executor: result.executor ?? null,
});
await recordCursorToolGatewayBilling({
appendEvent,
runId,
row,
result,
userAuth,
deliverySessionId: delivery.sessionId,
});
return { sessionId: delivery.sessionId, routing };
}
await recordCursorToolGatewayBilling({
appendEvent,
runId,
row,
result,
userAuth,
});
return { sessionId: row.agent_session_id ?? null, routing };
} catch (err) {
if (!(await fallbackCursorExecutorToDeepseek(err))) throw err;
+149
View File
@@ -9,9 +9,66 @@ import {
assertRequiredImageGenerationCompleted,
createAgentRunGateway,
normalizeAgentRunWorkerIdentity,
resolveActiveTaskContext,
resolveRouterTranscript,
resolveRequiredCodeExecutor,
} from './agent-run-gateway.mjs';
test('resolveActiveTaskContext returns failed agent task metadata for same session', async () => {
const now = Date.now();
const pool = {
async query() {
return [[{
id: 'run-failed-1',
status: 'failed',
error_message: 'delivery failed',
created_at: now - 60_000,
updated_at: now - 30_000,
intent_json: {
route: 'agent_orchestration',
suggestedSkill: 'page-data-collect',
reason: '页面需要数据交互与持久化',
},
}]];
},
};
const context = await resolveActiveTaskContext(pool, {
userId: 'user-1',
sessionId: '20260827_3',
excludeRunId: 'run-current',
nowMs: () => now,
});
assert.equal(context?.hasRecentAgentTask, true);
assert.equal(context?.lastRunFailed, true);
assert.equal(context?.lastSuggestedSkill, 'page-data-collect');
});
test('resolveRouterTranscript formats recent portal snapshot messages', async () => {
const transcript = await resolveRouterTranscript({
async get(sessionId) {
assert.equal(sessionId, '20260827_3');
return {
messages: [
{
role: 'user',
metadata: { displayText: '帮我做台账' },
content: [{ type: 'text', text: '帮我做台账' }],
},
{
role: 'assistant',
content: [{ type: 'text', text: '正在创建页面。' }],
},
],
};
},
}, {
sessionId: '20260827_3',
excludeLatestUserText: '继续完善',
});
assert.match(transcript, /用户:帮我做台账/);
assert.match(transcript, /助手:正在创建页面/);
});
test('required code executor is read from run metadata', () => {
assert.equal(resolveRequiredCodeExecutor({
metadata: { memindRun: { executor: 'AIDER' } },
@@ -3131,6 +3188,98 @@ test('required Aider run persists a validated result into a chat session', async
);
});
test('required Cursor run bills usage after validated delivery', async () => {
const pool = createFakePool();
const workdir = await fs.mkdtemp(path.join(os.tmpdir(), 'memind-cursor-billing-'));
const deliveries = [];
const billingCalls = [];
const gateway = createAgentRunGateway({
pool,
userAuth: {
async resolveWorkingDir() {
return workdir;
},
async getBillingState() {
return { lastInputTokens: 100, lastOutputTokens: 20 };
},
async billSessionUsage(userId, sessionId, tokenState, requestId) {
billingCalls.push({ userId, sessionId, tokenState, requestId });
return {
ok: true,
costCents: 3,
deltaInputTokens: tokenState.accumulatedInputTokens - 100,
deltaOutputTokens: tokenState.accumulatedOutputTokens - 20,
};
},
},
tkmindProxy: {},
directChatService: {
async respondDeterministically(options) {
deliveries.push(options);
await options.onSessionReady('h5direct_cursor_result');
return { sessionId: 'h5direct_cursor_result' };
},
},
toolGateway: {
getStatus() {
return {
enabled: true,
protocol: 'agent-run-v1',
executors: ['cursor', 'aider', 'openhands'],
};
},
async executeJob() {
return {
ok: true,
dryRun: false,
executor: 'cursor',
exitCode: 0,
cwd: workdir,
stdout: '{"type":"result","subtype":"success","result":"done","usage":{"inputTokens":1000,"outputTokens":200}}',
displayStdout: 'done',
usage: { inputTokens: 1000, outputTokens: 200 },
stderr: '',
};
},
},
retryDelaysMs: [],
});
const run = await gateway.createRun('user-1', {
requestId: 'req-cursor-billing',
userMessage: {
role: 'user',
content: [{ type: 'text', text: 'build the page' }],
metadata: {
displayText: 'build the page',
memindRun: {
requiredExecutor: 'cursor',
},
},
},
toolMode: 'code',
taskType: 'h5_chat_code_task',
});
await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
assert.equal(deliveries.length, 1);
assert.match(deliveries[0].reply, /done/);
assert.equal(billingCalls.length, 1);
assert.equal(billingCalls[0].userId, 'user-1');
assert.equal(billingCalls[0].sessionId, 'h5direct_cursor_result');
assert.equal(billingCalls[0].requestId, 'req-cursor-billing');
assert.deepEqual(billingCalls[0].tokenState, {
accumulatedInputTokens: 1100,
accumulatedOutputTokens: 220,
});
assert.equal(
pool.events.some(
(event) => event.runId === run.id && event.eventType === 'tool_gateway_billed',
),
true,
);
});
test('agent run fails non-retryably when tool gateway artifact validation fails', async () => {
const pool = createFakePool();
const workdir = await fs.mkdtemp(path.join(os.tmpdir(), 'memind-tool-validation-missing-'));
+234 -15
View File
@@ -16,6 +16,10 @@ import {
} from './memory-intervention.mjs';
import { filterMemoriesByQuery } from './memory-legacy-fallback.mjs';
import { matchDirectChatFaqRule, isExplicitTextOnlyRequest } from './chat-intent-router-rules.mjs';
import {
deriveAssistantFacingText,
deriveUserFacingText,
} from './conversation-display.mjs';
import { isGoalRunIntent } from './goal-run-intent.mjs';
import { pgvectorMemoryBackendInternals } from './memory-v2-pgvector.mjs';
@@ -84,6 +88,17 @@ const AGENT_SESSION_CONTINUE_PATTERNS = [
/^(?:开始吧|按默认做|默认方案|继续|就这样|就这样吧)[!!。.\s]*$/iu,
];
/** Corrections and refinements during an active Agent task must stay on Agent. */
const AGENT_TASK_FOLLOWUP_PATTERNS = [
/(?:还是|仍然|依然|并没|没有|并未).{0,16}(?:按照|按|理解|改|做|对|听)/u,
/(?:不对|错了|有误|有问题|不行|不好|不太对)/u,
/(?:更正|修正|修改|改一下|调整|优化|重新|再来|重做|再试)/u,
/(?:不听|没听|不按|没按|不照).{0,16}(?:指令|要求|说的|逻辑|意思)/u,
/(?:继续|接着).{0,16}(?:做|改|完善|优化|调整|处理|执行)/u,
/(?:按我(?:的|说)|照我(?:的|说)|我说的|我讲的|我上面)/u,
/(?:没(?:有)?按照|没有按).{0,20}(?:逻辑|要求|方案|口径|框架)/u,
];
/** Pure text chat/creative prompts — fast-path even when LLM router is enabled. */
const OBVIOUS_DIRECT_CHAT_PATTERNS = [
/(?:讲|说|来|编).{0,10}(?:个|一段|一首|一个)?(?:睡前故事|故事|笑话|段子)/u,
@@ -310,6 +325,84 @@ export function isAgentSessionContinueText(text) {
return AGENT_SESSION_CONTINUE_PATTERNS.some((pattern) => pattern.test(normalized));
}
export function isAgentTaskFollowUpText(text) {
const normalized = String(text ?? '').trim();
if (!normalized) return false;
return AGENT_TASK_FOLLOWUP_PATTERNS.some((pattern) => pattern.test(normalized));
}
function isExplicitDirectChatOnlyText(text) {
const normalized = String(text ?? '').trim();
if (!normalized) return false;
if (isMemoryRecallQuestion(normalized)) return true;
if (isExplicitTextOnlyRequest(normalized)) return true;
if (OBVIOUS_DIRECT_PATTERNS.some((pattern) => pattern.test(normalized))) return true;
if (OBVIOUS_DIRECT_CHAT_PATTERNS.some((pattern) => pattern.test(normalized))) return true;
const faqMatch = matchDirectChatFaqRule(normalized);
return Boolean(faqMatch);
}
export function shouldForceActiveAgentTaskContinuation({
activeTaskContext = null,
text = '',
sessionId = null,
sessionMessageCount = null,
} = {}) {
if (!activeTaskContext?.hasRecentAgentTask) return false;
if (!hasPriorAgentConversation(sessionId, sessionMessageCount)) return false;
const normalized = String(text ?? '').trim();
if (!normalized || isExplicitDirectChatOnlyText(normalized)) return false;
if (activeTaskContext.lastRunFailed === true) return true;
return isAgentTaskFollowUpText(normalized);
}
export function shouldDeferActiveTaskRoutingToLlm({
activeTaskContext = null,
text = '',
sessionId = null,
sessionMessageCount = null,
} = {}) {
if (!activeTaskContext?.hasRecentAgentTask) return false;
if (!hasPriorAgentConversation(sessionId, sessionMessageCount)) return false;
const normalized = String(text ?? '').trim();
if (!normalized || isExplicitDirectChatOnlyText(normalized)) return false;
if (shouldForceActiveAgentTaskContinuation({
activeTaskContext,
text: normalized,
sessionId,
sessionMessageCount,
})) {
return false;
}
return activeTaskContext.lastRoute !== 'direct_chat'
&& activeTaskContext.lastRoute !== 'chat';
}
/** @deprecated use shouldForceActiveAgentTaskContinuation */
export function shouldContinueActiveAgentTask(input = {}) {
return shouldForceActiveAgentTaskContinuation(input)
|| shouldDeferActiveTaskRoutingToLlm(input);
}
export function buildActiveTaskFollowUpClassification(activeTaskContext, { grantedSkills = [] } = {}) {
let suggestedSkill = String(activeTaskContext?.lastSuggestedSkill ?? '').trim() || null;
if (suggestedSkill && grantedSkills.length > 0 && !grantedSkills.includes(suggestedSkill)) {
suggestedSkill = null;
}
const agentBrief = suggestedSkill
? `继续完成未完成的 ${suggestedSkill} 任务;用户正在纠正或补充要求,必须调用工具执行,不要只回复文字。`
: '用户正在纠正或补充上一轮 Agent 任务,必须继续执行并产出结果,不要只回复文字。';
return normalizeClassification({
route: CHAT_INTENT_ROUTE.AGENT,
confidence: 0.92,
reason: activeTaskContext?.lastRunFailed
? '上一轮任务未完成,延续 Agent 执行'
: '活跃任务会话中的纠正或补充',
suggested_skill: suggestedSkill,
agent_brief: agentBrief,
}, { source: 'rule' });
}
export function isRealtimeInfoQuestion(text) {
const normalized = String(text ?? '').trim();
if (!normalized) return false;
@@ -541,6 +634,8 @@ function buildRouterSystemPrompt(grantedSkills = []) {
'- 用户要产出可访问页面、文件、链接,或需要工具/skills → agent_orchestration',
'- 用户询问实时赛况、新闻、天气、行情等需要联网查询的问题 → agent_orchestrationsuggested_skill 填 web(不要填 search',
'- 不确定时优先 agent_orchestration,避免漏执行',
'- 若 [Active Task] 显示上一轮 Agent 任务失败或进行中,用户当前消息是在纠正、补充、追问进度或要求按前文继续 → agent_orchestration,不要判成 direct_chat',
'- [Recent Conversation] 是对话摘要,当前消息可能引用上文(如「不对」「按我说的」「还是没有改」);结合上下文判断,不要孤立看当前一句',
'- 记忆线索只用于辅助判断本轮意图,不能替用户扩写新需求',
'',
skills.length ? `当前用户已授权 skills${skills.join(', ')}` : '当前用户未授权额外 skills。',
@@ -626,15 +721,83 @@ export function buildRouterContext(resolveResult, {
};
}
function buildRouterUserPrompt({ text, routerContext }) {
const context = String(routerContext ?? '').trim() || '无';
return [
'[Router Context]',
context,
'',
'[User]',
text || '(empty)',
].join('\n');
export function formatRouterTranscript(messages = [], {
maxTurns = 8,
maxChars = 4_000,
excludeLatestUserText = null,
} = {}) {
const normalizedExclude = String(excludeLatestUserText ?? '').trim();
const lines = [];
let usedChars = 0;
const visible = [];
for (const message of Array.isArray(messages) ? messages : []) {
const role = String(message?.role ?? '').trim();
if (role !== 'user' && role !== 'assistant') continue;
const raw = Array.isArray(message?.content)
? message.content
.map((item) => (typeof item === 'string' ? item : item?.text ?? ''))
.join('\n')
: String(message?.content ?? message?.text ?? '');
const displayText = String(message?.metadata?.displayText ?? '').trim();
const text = role === 'user'
? (displayText || deriveUserFacingText(raw))
: deriveAssistantFacingText(raw);
const trimmed = String(text ?? '').trim();
if (!trimmed) continue;
if (role === 'user' && normalizedExclude && trimmed === normalizedExclude) continue;
visible.push({ role, text: trimmed });
}
for (let index = visible.length - 1; index >= 0 && lines.length < maxTurns; index -= 1) {
const item = visible[index];
const line = `${item.role === 'user' ? '用户' : '助手'}${item.text}`;
if (lines.length > 0 && usedChars + line.length > maxChars) break;
lines.unshift(line);
usedChars += line.length;
}
return lines.join('\n');
}
function formatActiveTaskContextForRouter(activeTaskContext) {
if (!activeTaskContext?.hasRecentAgentTask) return '';
const parts = [
`- 上一轮 Agent 任务状态:${activeTaskContext.lastRunStatus ?? 'unknown'}`,
];
if (activeTaskContext.lastRunFailed) {
parts.push('- 上一轮任务失败,用户可能在纠正、补充或要求重试');
}
if (activeTaskContext.lastSuggestedSkill) {
parts.push(`- 上一轮建议 skill${activeTaskContext.lastSuggestedSkill}`);
}
if (activeTaskContext.lastIntentReason) {
parts.push(`- 上一轮路由原因:${activeTaskContext.lastIntentReason}`);
}
if (activeTaskContext.lastErrorMessage) {
parts.push(`- 失败摘要:${String(activeTaskContext.lastErrorMessage).slice(0, 200)}`);
}
return parts.join('\n');
}
function buildRouterUserPrompt({
text,
routerContext,
recentTranscript = '',
activeTaskContext = null,
} = {}) {
const sections = [];
const memoryContext = String(routerContext ?? '').trim();
if (memoryContext) {
sections.push('[Memory Context]', memoryContext, '');
}
const taskContext = formatActiveTaskContextForRouter(activeTaskContext);
if (taskContext) {
sections.push('[Active Task]', taskContext, '');
}
const transcript = String(recentTranscript ?? '').trim();
if (transcript) {
sections.push('[Recent Conversation]', transcript, '');
}
sections.push('[Current User Message]', text || '(empty)');
return sections.join('\n');
}
function parseRouterJson(reply) {
@@ -1043,6 +1206,8 @@ export function classifyWithRules({
sessionMessageCount = null,
userMessage = null,
includeIntentPatterns = true,
activeTaskContext = null,
grantedSkills = [],
} = {}) {
const decisionContext = {
text,
@@ -1123,6 +1288,21 @@ export function classifyWithRules({
reason: 'Agent 会话确认/续聊',
}, { source: 'rule' }), decisionContext);
}
if (
includeIntentPatterns
&& normalized
&& shouldForceActiveAgentTaskContinuation({
activeTaskContext,
text: normalized,
sessionId,
sessionMessageCount,
})
) {
return finalizeRouterClassification(
buildActiveTaskFollowUpClassification(activeTaskContext, { grantedSkills }),
decisionContext,
);
}
if (includeIntentPatterns && normalized && isPageDataDevIntent(normalized)) {
return finalizeRouterClassification(normalizeClassification({
route: CHAT_INTENT_ROUTE.AGENT,
@@ -1173,6 +1353,16 @@ export function classifyWithRules({
return finalizeRouterClassification(buildRealtimeInfoClassification(), decisionContext);
}
if (includeIntentPatterns && normalized) {
if (
shouldDeferActiveTaskRoutingToLlm({
activeTaskContext,
text: normalized,
sessionId,
sessionMessageCount,
})
) {
return null;
}
return finalizeRouterClassification(normalizeClassification({
route: CHAT_INTENT_ROUTE.DIRECT_CHAT,
confidence: 0.72,
@@ -1229,6 +1419,8 @@ export function createChatIntentRouter(options = {}) {
text,
grantedSkills = [],
routerContext = null,
recentTranscript = '',
activeTaskContext = null,
} = {}) {
if (typeof llmProviderService?.createChatCompletion !== 'function') return null;
try {
@@ -1244,6 +1436,8 @@ export function createChatIntentRouter(options = {}) {
content: buildRouterUserPrompt({
text,
routerContext: routerContext?.content ?? null,
recentTranscript,
activeTaskContext,
}),
},
],
@@ -1496,6 +1690,8 @@ export function createChatIntentRouter(options = {}) {
toolMode = 'chat',
forceDeepReasoning = false,
grantedSkills = [],
activeTaskContext = null,
recentTranscript = '',
} = {}) {
const text = messageDisplayText(userMessage);
const decisionContext = {
@@ -1535,10 +1731,28 @@ export function createChatIntentRouter(options = {}) {
sessionMessageCount,
userMessage,
includeIntentPatterns: true,
activeTaskContext,
grantedSkills,
});
if (ruleResult) return finalizeWithCoercion(ruleResult);
const baseline = buildFallbackClassification();
const activeTaskFallback = (
shouldForceActiveAgentTaskContinuation({
activeTaskContext,
text,
sessionId,
sessionMessageCount,
})
|| shouldDeferActiveTaskRoutingToLlm({
activeTaskContext,
text,
sessionId,
sessionMessageCount,
})
)
? buildActiveTaskFollowUpClassification(activeTaskContext, { grantedSkills })
: null;
const baseline = activeTaskFallback ?? buildFallbackClassification();
if (!llmRouterEligible) {
return finalizeWithCoercion(baseline);
}
@@ -1554,14 +1768,19 @@ export function createChatIntentRouter(options = {}) {
.catch(() => buildRouterContext(null))
: Promise.resolve(buildRouterContext(null));
// Memory is optional routing hint; run LLM in parallel to cut serial latency.
const [routerContext, llmResult] = await Promise.all([
memoryPromise,
const llmPromise = memoryPromise.then((routerContext) =>
classifyWithLlm({
text,
grantedSkills,
routerContext: null,
routerContext,
recentTranscript,
activeTaskContext,
}),
);
const [routerContext, llmResult] = await Promise.all([
memoryPromise,
llmPromise,
]);
if (policy.shadowMode) {
@@ -1592,7 +1811,7 @@ export function createChatIntentRouter(options = {}) {
if (!llmResult || llmResult.confidence < policy.minConfidence) {
return finalizeWithCoercion({
...baseline,
...(activeTaskFallback ?? baseline),
llmSuggestion: llmResult
? {
route: llmResult.route,
+138
View File
@@ -11,6 +11,9 @@ import {
coercePageGenerationSkill,
createChatIntentRouter,
createManagedChatIntentRouter,
formatRouterTranscript,
shouldDeferActiveTaskRoutingToLlm,
shouldForceActiveAgentTaskContinuation,
isNormalizedRouterDecisionEnabled,
isNormalizedRouterDecisionShadow,
resolveGatewayAgentSessionId,
@@ -469,6 +472,141 @@ test('classifyWithRules keeps agent session confirmation on agent path', () => {
assert.match(fresh.reason, /记忆/);
});
test('classifyWithRules continues agent task after failed run in same session', () => {
const activeTaskContext = {
hasRecentAgentTask: true,
lastRunFailed: true,
lastSuggestedSkill: 'page-data-collect',
lastRunStatus: 'failed',
lastRoute: 'agent_orchestration',
};
const result = classifyWithRules({
text: '还是没有按照我的逻辑更正',
sessionId: '20260827_3',
sessionMessageCount: 50,
activeTaskContext,
grantedSkills: ['page-data-collect'],
});
assert.equal(result.route, CHAT_INTENT_ROUTE.AGENT);
assert.equal(result.suggestedSkill, 'page-data-collect');
assert.match(result.reason, /上一轮任务未完成/);
});
test('classifyWithRules routes correction follow-up to agent when recent agent task exists', () => {
const activeTaskContext = {
hasRecentAgentTask: true,
lastRunFailed: false,
lastSuggestedSkill: 'page-data-collect',
lastRunStatus: 'succeeded',
lastRoute: 'agent_orchestration',
};
const result = classifyWithRules({
text: '不听指令了吗',
sessionId: '20260827_3',
sessionMessageCount: 51,
activeTaskContext,
grantedSkills: ['page-data-collect'],
});
assert.equal(result.route, CHAT_INTENT_ROUTE.AGENT);
assert.equal(result.suggestedSkill, 'page-data-collect');
assert.match(result.reason, /纠正或补充/);
});
test('classifyWithRules defers ambiguous active-task follow-up to LLM router', () => {
const activeTaskContext = {
hasRecentAgentTask: true,
lastRunFailed: false,
lastSuggestedSkill: 'page-data-collect',
lastRunStatus: 'succeeded',
lastRoute: 'agent_orchestration',
};
const result = classifyWithRules({
text: '招商管理端可以展开招商漏斗,明确准入门槛',
sessionId: '20260827_3',
sessionMessageCount: 52,
activeTaskContext,
grantedSkills: ['page-data-collect'],
});
assert.equal(result, null);
});
test('formatRouterTranscript keeps recent turns and strips skill prefixes from user text', () => {
const transcript = formatRouterTranscript([
{
role: 'user',
metadata: { displayText: '帮我做商户沟通台账' },
content: [{ type: 'text', text: '请使用 page-data-collect 技能:帮我做商户沟通台账' }],
},
{
role: 'assistant',
content: [{ type: 'text', text: '好的,我先创建问卷页面。' }],
},
], {
excludeLatestUserText: '招商管理端展开漏斗',
});
assert.match(transcript, /用户:帮我做商户沟通台账/);
assert.match(transcript, /助手:好的,我先创建问卷页面/);
assert.doesNotMatch(transcript, /page-data-collect 技能/);
});
test('createChatIntentRouter uses active-task fallback when LLM router is unavailable', async () => {
const router = createChatIntentRouter({
llmProviderService: {
async createChatCompletion() {
return null;
},
},
env: {
...process.env,
MEMIND_CHAT_LLM_ROUTER_ENABLED: '1',
MEMIND_CHAT_LLM_ROUTER_SHADOW: '0',
MEMIND_CHAT_ROUTER_CANARY_USER_IDS: '',
},
});
const result = await router.classify({
userId: 'user-1',
sessionId: '20260827_3',
sessionMessageCount: 40,
grantedSkills: ['page-data-collect'],
activeTaskContext: {
hasRecentAgentTask: true,
lastRunFailed: true,
lastSuggestedSkill: 'page-data-collect',
lastRunStatus: 'failed',
lastRoute: 'agent_orchestration',
},
userMessage: {
role: 'user',
content: [{ type: 'text', text: '还是没有按照我的逻辑更正' }],
metadata: { displayText: '还是没有按照我的逻辑更正' },
},
});
assert.equal(result.route, CHAT_INTENT_ROUTE.AGENT);
assert.equal(result.suggestedSkill, 'page-data-collect');
});
test('classifyWithRules keeps memory recall on direct chat even with failed agent task context', () => {
const activeTaskContext = {
hasRecentAgentTask: true,
lastRunFailed: true,
lastSuggestedSkill: 'page-data-collect',
lastRunStatus: 'failed',
};
const result = classifyWithRules({
text: '你记得我说想去哪儿吗',
sessionId: '20260827_3',
sessionMessageCount: 50,
activeTaskContext,
userMessage: {
role: 'user',
content: [{ type: 'text', text: '你记得我说想去哪儿吗' }],
metadata: { displayText: '你记得我说想去哪儿吗' },
},
});
assert.equal(result.route, CHAT_INTENT_ROUTE.DIRECT_CHAT);
assert.match(result.reason, /记忆|历史对话/);
});
test('classifyWithRules keeps memory recall on direct chat even when session already active', () => {
const result = classifyWithRules({
text: '你记得我说想去哪儿吗',
+1 -1
View File
@@ -95,7 +95,7 @@ export function buildCursorExecutorLaunchPlan({
'--force',
'--approve-mcps',
'--output-format',
'text',
'stream-json',
'--workspace',
workspace,
prompt,
+2
View File
@@ -42,6 +42,8 @@ test('buildCursorExecutorLaunchPlan includes workspace and mindspace hints', asy
assert.equal(plan.executor, 'cursor');
assert.equal(plan.cwd, '/tmp/mindspace/user-1');
assert.equal(plan.command, agentStub);
assert.ok(plan.args.includes('--output-format'));
assert.ok(plan.args.includes('stream-json'));
assert.ok(plan.args.includes('--workspace'));
assert.ok(plan.args.includes('/tmp/mindspace/user-1'));
assert.match(plan.args.at(-1) ?? '', /public\/.*\.html/);
+90
View File
@@ -0,0 +1,90 @@
function parseStreamJsonLine(line) {
const text = String(line ?? '').trim();
if (!text) return null;
try {
const parsed = JSON.parse(text);
return parsed && typeof parsed === 'object' ? parsed : null;
} catch {
return null;
}
}
export function parseCursorAgentStreamJsonEvents(stdout) {
const events = [];
for (const line of String(stdout ?? '').split('\n')) {
const parsed = parseStreamJsonLine(line);
if (parsed) events.push(parsed);
}
return events;
}
export function normalizeCursorAgentUsage(raw) {
if (!raw || typeof raw !== 'object') return null;
const inputTokens = Number(raw.inputTokens ?? raw.input_tokens ?? 0);
const outputTokens = Number(raw.outputTokens ?? raw.output_tokens ?? 0);
const cacheReadTokens = Number(raw.cacheReadTokens ?? raw.cache_read_tokens ?? 0);
const cacheWriteTokens = Number(raw.cacheWriteTokens ?? raw.cache_write_tokens ?? 0);
const billableInput = inputTokens + cacheWriteTokens;
const billableOutput = outputTokens;
if (billableInput <= 0 && billableOutput <= 0) return null;
return {
inputTokens: billableInput,
outputTokens: billableOutput,
cacheReadTokens: cacheReadTokens > 0 ? cacheReadTokens : 0,
cacheWriteTokens: cacheWriteTokens > 0 ? cacheWriteTokens : 0,
};
}
export function parseCursorAgentUsage(stdout) {
const events = parseCursorAgentStreamJsonEvents(stdout);
for (let index = events.length - 1; index >= 0; index -= 1) {
const event = events[index];
if (event?.type !== 'result') continue;
const usage = normalizeCursorAgentUsage(event.usage);
if (usage) return usage;
}
return null;
}
function extractAssistantText(message) {
if (!message || typeof message !== 'object') return '';
const content = message.content ?? message.message?.content;
if (typeof content === 'string') return content.trim();
if (!Array.isArray(content)) return '';
return content
.map((part) => {
if (!part || typeof part !== 'object') return '';
if (part.type === 'text') return String(part.text ?? '').trim();
return '';
})
.filter(Boolean)
.join('\n')
.trim();
}
export function extractCursorAgentDisplayText(stdout) {
const events = parseCursorAgentStreamJsonEvents(stdout);
for (let index = events.length - 1; index >= 0; index -= 1) {
const event = events[index];
if (event?.type === 'result') {
const resultText = String(event.result ?? '').trim();
if (resultText) return resultText;
}
if (event?.type === 'assistant') {
const assistantText = extractAssistantText(event.message ?? event);
if (assistantText) return assistantText;
}
}
return '';
}
export function buildCursorBillingTokenState(usage, priorBillingState = null) {
const normalized = normalizeCursorAgentUsage(usage);
if (!normalized) return null;
const priorIn = Number(priorBillingState?.lastInputTokens ?? 0);
const priorOut = Number(priorBillingState?.lastOutputTokens ?? 0);
return {
accumulatedInputTokens: priorIn + normalized.inputTokens,
accumulatedOutputTokens: priorOut + normalized.outputTokens,
};
}
+54
View File
@@ -0,0 +1,54 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import {
buildCursorBillingTokenState,
extractCursorAgentDisplayText,
normalizeCursorAgentUsage,
parseCursorAgentUsage,
} from './cursor-agent-usage.mjs';
const SAMPLE_STREAM_JSON = [
'{"type":"system","subtype":"init","cwd":"/tmp","session_id":"abc","model":"Auto"}',
'{"type":"user","message":{"role":"user","content":[{"type":"text","text":"reply with exactly: ok"}]}}',
'{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"ok"}]}}',
'{"type":"result","subtype":"success","duration_ms":13064,"result":"ok","usage":{"inputTokens":7765,"outputTokens":64,"cacheReadTokens":5427,"cacheWriteTokens":0}}',
].join('\n');
test('parseCursorAgentUsage reads result usage from stream-json stdout', () => {
const usage = parseCursorAgentUsage(SAMPLE_STREAM_JSON);
assert.deepEqual(usage, {
inputTokens: 7765,
outputTokens: 64,
cacheReadTokens: 5427,
cacheWriteTokens: 0,
});
});
test('normalizeCursorAgentUsage includes cache write tokens in billable input', () => {
const usage = normalizeCursorAgentUsage({
inputTokens: 1000,
outputTokens: 200,
cacheWriteTokens: 300,
});
assert.deepEqual(usage, {
inputTokens: 1300,
outputTokens: 200,
cacheReadTokens: 0,
cacheWriteTokens: 300,
});
});
test('extractCursorAgentDisplayText prefers result text', () => {
assert.equal(extractCursorAgentDisplayText(SAMPLE_STREAM_JSON), 'ok');
});
test('buildCursorBillingTokenState accumulates on prior session billing', () => {
const tokenState = buildCursorBillingTokenState(
{ inputTokens: 1000, outputTokens: 200 },
{ lastInputTokens: 5000, lastOutputTokens: 800 },
);
assert.deepEqual(tokenState, {
accumulatedInputTokens: 6000,
accumulatedOutputTokens: 1000,
});
});
+5 -4
View File
@@ -15,13 +15,14 @@ test('buildCodeRunCompletionReply never exposes cursor brand to users', () => {
assert.doesNotMatch(reply, /cursor/i);
});
test('buildCodeRunCompletionReply sanitizes cursor wording in stdout', () => {
test('buildCodeRunCompletionReply prefers displayStdout over raw stream-json stdout', () => {
const reply = buildCodeRunCompletionReply({
executor: 'cursor',
stdout: 'cursor agent finished editing public/page.html',
stdout: '{"type":"result","result":"hidden"}\n',
displayStdout: '已写入 public/spring-poem.html',
});
assert.match(reply, /TKMind 智趣 finished editing public\/page\.html/);
assert.doesNotMatch(reply, /\bcursor\b/i);
assert.match(reply, /已写入 public\/spring-poem\.html/);
assert.doesNotMatch(reply, /"type":"result"/);
});
test('sanitizeUserFacingBrandText replaces cursor wording', () => {
+1 -1
View File
@@ -113,7 +113,7 @@
"verify:page-data-delivery": "node scripts/repair-page-data-workspace-bindings.mjs --dry-run",
"verify:template-catalog-portal": "node scripts/verify-template-catalog-portal.mjs",
"verify:template-catalog-e2e": "node scripts/verify-template-catalog-e2e.mjs",
"verify:cursor-executor": "node --test cursor-agent-launch.test.mjs cursor-page-routing.test.mjs tool-gateway.test.mjs llm-providers.test.mjs help-escalation.test.mjs executor-display-label.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs wechat-cursor-executor-policy.test.mjs",
"verify:cursor-executor": "node --test cursor-agent-launch.test.mjs cursor-agent-usage.test.mjs cursor-page-routing.test.mjs tool-gateway.test.mjs llm-providers.test.mjs help-escalation.test.mjs executor-display-label.test.mjs agent-run-gateway.test.mjs agent-run-routes.test.mjs wechat-cursor-executor-policy.test.mjs",
"repair:page-data-bindings": "node scripts/repair-page-data-workspace-bindings.mjs",
"repair:page-data:103": "node scripts/ensure-page-data-datasets.mjs && node scripts/repair-page-data-workspace-bindings.mjs",
"verify:wechat-channel-isolation": "node scripts/check-wechat-channel-isolation.mjs",
+43 -1
View File
@@ -28,6 +28,19 @@ async function recordValidationWithRetry(engine, runId, observation) {
}
}
export function resolveValidationEngineSelection(selection) {
if (selection.shadowEngine === WORKFLOW_ENGINE.LANGGRAPH) {
return selection;
}
if (
selection.engine === WORKFLOW_ENGINE.LANGGRAPH
&& ['active', 'canary', 'shadow'].includes(String(selection.mode ?? ''))
) {
return selection;
}
return null;
}
export function createWorkflowShadowObserver({
configService,
serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN,
@@ -66,6 +79,34 @@ export function createWorkflowShadowObserver({
};
}
async function selectValidationEngine({
runId,
requestId,
userId,
workflowName,
}) {
const selection = await configService.selectEngine({
runId,
requestId,
userId,
workflowName,
});
if (!resolveValidationEngineSelection(selection)) {
return { selection, engine: null };
}
const state = await configService.getRuntimeState();
return {
selection,
engine: createRemoteWorkflowEngine({
id: WORKFLOW_ENGINE.LANGGRAPH,
baseUrl: state.config.serviceUrl,
serviceToken,
timeoutMs: state.config.requestTimeoutMs,
fetchImpl,
}),
};
}
async function observeWorkflowRun({
runId,
requestId,
@@ -152,7 +193,7 @@ export function createWorkflowShadowObserver({
workflowName = 'code-run-v1',
observation,
} = {}) {
const { selection, engine } = await selectShadowEngine({
const { selection, engine } = await selectValidationEngine({
runId,
requestId,
userId,
@@ -187,5 +228,6 @@ export function createWorkflowShadowObserver({
export const workflowShadowObserverInternals = {
recordValidationWithRetry,
resolveValidationEngineSelection,
safeError,
};
+72 -1
View File
@@ -3,7 +3,7 @@ import test from 'node:test';
import { MemorySaver } from '@langchain/langgraph';
import { createOrchestratorApp } from './app.mjs';
import { createLangGraphOrchestratorRuntime } from './runtime.mjs';
import { createWorkflowShadowObserver } from './shadow-observer.mjs';
import { createWorkflowShadowObserver, resolveValidationEngineSelection } from './shadow-observer.mjs';
function jsonResponse(body, status = 200) {
return new Response(JSON.stringify(body), {
@@ -25,6 +25,77 @@ async function listen(app) {
};
}
test('resolveValidationEngineSelection accepts active-mode LangGraph engine', () => {
assert.ok(resolveValidationEngineSelection({
engine: 'langgraph',
shadowEngine: null,
mode: 'active',
reason: 'active',
}));
assert.equal(resolveValidationEngineSelection({
engine: 'native',
shadowEngine: null,
mode: 'active',
reason: 'active',
}), null);
});
test('validation observer records observations in active mode without shadowEngine', async () => {
let capturedUrl = null;
const observer = createWorkflowShadowObserver({
configService: {
async selectEngine() {
return {
engine: 'langgraph',
candidateEngine: 'langgraph',
shadowEngine: null,
fallbackEngine: 'native',
reason: 'active',
candidateReason: 'active',
mode: 'active',
configVersion: 6,
dryRun: false,
};
},
async getRuntimeState() {
return {
config: {
serviceUrl: 'http://orchestrator.internal:8093',
requestTimeoutMs: 1200,
},
};
},
},
serviceToken: 'internal-token',
fetchImpl: async (url) => {
capturedUrl = url;
return jsonResponse({
runId: 'run-active-1',
validation: { verdict: 'passed', kind: 'page-data-delivery' },
});
},
});
const result = await observer.observeValidation({
runId: 'run-active-1',
requestId: 'request-active-1',
userId: 'user-1',
observation: {
idempotencyKey: 'run-active-1:page-data-delivery:v1',
taskType: 'page_data_dev',
required: true,
checks: [{ id: 'page_data_binding', status: 'passed' }],
source: 'portal-agent-run',
observedAt: 123,
},
});
assert.equal(result.observed, true);
assert.equal(result.mode, 'active');
assert.equal(result.validation.verdict, 'passed');
assert.match(String(capturedUrl), /\/v1\/runs\/run-active-1\/validation-observations$/);
});
test('shadow observer skips without creating a remote client when mode does not select shadow', async () => {
let fetchCalls = 0;
const observer = createWorkflowShadowObserver({
+10
View File
@@ -3,6 +3,10 @@ import { EventEmitter } from 'node:events';
import fs from 'node:fs/promises';
import path from 'node:path';
import { buildCursorExecutorLaunchPlan } from './cursor-agent-launch.mjs';
import {
extractCursorAgentDisplayText,
parseCursorAgentUsage,
} from './cursor-agent-usage.mjs';
import { resolveExecutorDisplayLabel } from './executor-display-label.mjs';
const BASE_CODE_EXECUTORS = ['aider', 'openhands'];
@@ -308,6 +312,10 @@ export function createToolGateway({
child.on('exit', (code, signal) => {
cleanup();
if (code === 0) {
const usage = executor === 'cursor' ? parseCursorAgentUsage(stdout) : null;
const displayStdout = executor === 'cursor'
? extractCursorAgentDisplayText(stdout)
: '';
resolve({
ok: true,
executor,
@@ -319,6 +327,8 @@ export function createToolGateway({
exitCode: code,
signal: signal ?? null,
stdout,
displayStdout: displayStdout || undefined,
usage: usage ?? undefined,
stderr,
});
return;