feat(billing): bill Cursor executor usage via stream-json token parsing

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>

agent-run-gateway.mjs

@@ -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();
@@ -1778,8 +1849,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 +1900,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;