memind/tkmind-proxy.mjs

import fs from 'node:fs';
import crypto from 'node:crypto';
import path from 'node:path';
import { Readable, Transform, Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { Agent, fetch as undiciFetch } from 'undici';
import { appendBalanceEvent, createSseBillingTransform } from './sse-billing.mjs';
import { finalizeSessionStreamEvent, writeSseErrorAndEnd } from './sse-event-taxonomy.mjs';
import { developerToolsFromPolicy } from './capabilities.mjs';
import { evaluateProxyRequest, isNativeH5ApiPath } from './policies.mjs';
import {
  buildPublicUrl,
  buildSandboxSessionConstraints,
  PUBLISH_ROOT_DIR,
  resolvePublicBaseUrl,
} from './user-publish.mjs';
import { buildCurrentTimeAgentPrefix, buildTaskRoutingAgentText } from './user-memory-profile.mjs';
import { reconcileAgentSession } from './session-reconcile.mjs';
import { createSessionAccess, isSessionBrokerEnabled } from './session-broker.mjs';
import { createSessionBrokerMetrics, isSessionBrokerMetricsEnabled } from './session-broker-metrics.mjs';
import { createImgproxySigner } from './imgproxy-signer.mjs';
import { isDirectChatSessionId } from './direct-chat-service.mjs';
import { ensureGooseUserMessageMetadata } from './goose-message.mjs';
import { consumeSessionEventsUntilFinish } from './session-reply-wait.mjs';
import {
  memoryLimitForIntervention,
  resolveMemoryInterventionMode,
} from './memory-intervention.mjs';
import {
  formatSessionStreamSseChunk,
  isSessionStreamReplayEnabled,
  parseSessionSseBlock,
  parseSessionStreamLastEventId,
  shouldPersistSessionStreamEvent,
  shouldSkipUpstreamAfterSessionReplay,
} from './session-stream.mjs';

const insecureDispatcher = new Agent({
  connect: { rejectUnauthorized: false },
});

function isHttpsTarget(target) {
  return target.startsWith('https://');
}

function sanitizeUserFacingProxyMessage(message, fallback = '后端连接失败，请稍后重试') {
  const normalized = String(message ?? '').trim();
  if (!normalized) return fallback;
  if (!/goose|goosed/i.test(normalized)) return normalized;
  if (/超时|timeout/i.test(normalized)) {
    return '后端连接超时，请确认后端服务正常后重试';
  }
  if (/不可用|连接失败|failed to fetch|networkerror|fetch failed|upstream|econn|enotfound/i.test(normalized)) {
    return fallback;
  }
  return normalized
    .replace(/\bgoosed\b/gi, '后端服务')
    .replace(/\bgoose\b/gi, '后端');
}

const PUBLIC_IMAGE_STORAGE_KEY_PATTERN = /^users\/([^/]+)\/images\/(\d{4}-\d{2}-\d{2}\/[^?#"'<>@\s]+)$/i;

function extractPublicImageStorageKey(rawUrl) {
  const value = String(rawUrl ?? '').trim();
  if (!value) return null;
  const directMatch = value.match(/(^|\/)(users\/[^?#"'<>@\s]+\/images\/\d{4}-\d{2}-\d{2}\/[^?#"'<>@\s]+)/i);
  if (directMatch?.[2]) return directMatch[2];
  const plainIndex = value.indexOf('plain/local:///');
  if (plainIndex >= 0) {
    const tail = value.slice(plainIndex + 'plain/local:///'.length);
    const storageKey = tail.replace(/@[a-z0-9]+(?:[?#].*)?$/i, '').replace(/[?#].*$/, '');
    return storageKey || null;
  }
  return null;
}

function resolvePublicBaseUrlFromLayout(publishLayout) {
  const publicUrl = String(publishLayout?.publicUrl ?? '').trim();
  if (!publicUrl) return resolvePublicBaseUrl();
  try {
    return new URL(publicUrl).origin;
  } catch {
    return resolvePublicBaseUrl();
  }
}

function buildPublicStandardImageUrl(rawUrl, userId, publishLayout = null) {
  const storageKey = extractPublicImageStorageKey(rawUrl);
  if (!storageKey) return null;
  const match = storageKey.match(PUBLIC_IMAGE_STORAGE_KEY_PATTERN);
  if (!match || match[1] !== String(userId ?? '')) return null;
  return buildPublicUrl(resolvePublicBaseUrlFromLayout(publishLayout), userId, `public/images/${match[2]}`);
}

async function apiFetch(target, apiSecret, pathname, init = {}) {
  const url = new URL(pathname, target);
  const headers = {
    ...(init.headers ?? {}),
    'X-Secret-Key': apiSecret,
  };
  if (init.body && !headers['Content-Type']) {
    headers['Content-Type'] = 'application/json';
  }

  return undiciFetch(url, {
    ...init,
    headers,
    dispatcher: isHttpsTarget(target) ? insecureDispatcher : undefined,
  });
}

function createRuntimeRouter({
  targets,
  targetHealthy,
  redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL,
  namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE ?? 'memind:runtime',
}) {
  if (!redisUrl || targets.length <= 1) return null;
  let clientPromise = null;
  let warned = false;
  const workerIdForTarget = (target) => {
    const idx = targets.indexOf(target);
    return idx >= 0 ? `goosed-${idx + 1}` : `goosed-${Buffer.from(String(target)).toString('base64url')}`;
  };
  const key = (...parts) => [namespace, ...parts].join(':');
  const getClient = async () => {
    if (!clientPromise) {
      clientPromise = import('redis')
        .then(async ({ createClient }) => {
          const client = createClient({ url: redisUrl });
          client.on('error', (err) => {
            if (!warned) {
              warned = true;
              console.warn('[RuntimeRouter] Redis error:', err instanceof Error ? err.message : err);
            }
          });
          await client.connect();
          console.log('[RuntimeRouter] Redis scheduler enabled');
          return client;
        })
        .catch((err) => {
          clientPromise = null;
          if (!warned) {
            warned = true;
            console.warn('[RuntimeRouter] Redis disabled:', err instanceof Error ? err.message : err);
          }
          return null;
        });
    }
    return clientPromise;
  };
  const readNumber = (value) => {
    const n = Number(value ?? 0);
    return Number.isFinite(n) ? n : 0;
  };
  const parseFirstTokenSample = (value) => {
    const parts = String(value ?? '').split(':');
    return readNumber(parts[1]);
  };
  const percentile = (values, pct) => {
    const sorted = values
      .map((value) => readNumber(value))
      .filter((value) => value > 0)
      .sort((a, b) => a - b);
    if (sorted.length === 0) return 0;
    const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil((pct / 100) * sorted.length) - 1));
    return sorted[idx];
  };
  const firstTokenWindowStats = async (client, workerId, windowMs) => {
    if (!client) return { count: 0, p50Ms: 0, p95Ms: 0 };
    const now = Date.now();
    const samples = await client
      .sendCommand([
        'ZRANGEBYSCORE',
        key('worker', workerId, 'first_token_samples'),
        String(now - windowMs),
        String(now),
      ])
      .catch(() => []);
    const values = samples.map(parseFirstTokenSample).filter((value) => value > 0);
    return {
      count: values.length,
      p50Ms: percentile(values, 50),
      p95Ms: percentile(values, 95),
    };
  };
  const scoreWorker = async (client, target) => {
    const workerId = workerIdForTarget(target);
    const values = await client.mGet([
      key('worker', workerId, 'active_streams'),
      key('worker', workerId, 'active_sessions'),
      key('worker', workerId, 'ewma_first_token_ms'),
      key('worker', workerId, 'error_rate'),
      key('worker', workerId, 'memory_pressure'),
      key('worker', workerId, 'cpu_load'),
      key('worker', workerId, 'fd_pressure'),
      key('worker', workerId, 'drain'),
    ]);
    if (/^(1|true|yes)$/i.test(String(values[7] ?? ''))) return Number.POSITIVE_INFINITY;
    return (
      readNumber(values[0]) * 3 +
      readNumber(values[1]) +
      readNumber(values[2]) * 0.01 +
      readNumber(values[3]) * 5 +
      readNumber(values[4]) * 2 +
      readNumber(values[5]) * 2 +
      readNumber(values[6]) * 2
    );
  };
  const workerScoreFromValues = (values = []) => (
    readNumber(values[0]) * 3 +
    readNumber(values[1]) +
    readNumber(values[2]) * 0.01 +
    readNumber(values[3]) * 5 +
    readNumber(values[4]) * 2 +
    readNumber(values[12]) * 2 +
    readNumber(values[13]) * 2
  );
  return {
    async pickTarget(fallbackPick, orderedTargets = targets) {
      const client = await getClient();
      if (!client) return fallbackPick();
      let winner = null;
      for (const target of orderedTargets) {
        if (!await targetHealthy(target)) continue;
        const score = await scoreWorker(client, target).catch(() => Number.POSITIVE_INFINITY);
        if (!Number.isFinite(score)) continue;
        if (!winner || score < winner.score) winner = { target, score };
      }
      return winner?.target ?? fallbackPick();
    },
    async registerSession(sessionId, target) {
      const client = await getClient();
      if (!client || !sessionId || !target) return;
      const workerId = workerIdForTarget(target);
      await client
        .multi()
        .set(key('session', sessionId, 'worker'), workerId, { EX: 60 * 60 * 24 * 30 })
        .set(key('session', sessionId, 'target'), target, { EX: 60 * 60 * 24 * 30 })
        .exec()
        .catch(() => null);
    },
    async resolveSessionTarget(sessionId) {
      const client = await getClient();
      if (!client || !sessionId) return null;
      const target = await client.get(key('session', sessionId, 'target')).catch(() => null);
      return target && targets.includes(target) ? target : null;
    },
    async streamStarted(sessionId, target) {
      const client = await getClient();
      if (!client || !target) return;
      const workerId = workerIdForTarget(target);
      const streamKey = sessionId ? key('stream', sessionId, 'status') : null;
      const now = String(Date.now());
      const multi = client
        .multi()
        .incr(key('worker', workerId, 'active_streams'))
        .incr(key('worker', workerId, 'stream_open_count'))
        .set(key('worker', workerId, 'last_stream_started_at'), now)
        .set(key('worker', workerId, 'heartbeat'), now, { EX: 30 });
      if (streamKey) multi.set(streamKey, 'active', { EX: 60 * 60 });
      await multi.exec().catch(() => null);
    },
    async streamEnded(sessionId, target, { status = 'closed' } = {}) {
      const client = await getClient();
      if (!client || !target) return;
      const workerId = workerIdForTarget(target);
      const streamKey = sessionId ? key('stream', sessionId, 'status') : null;
      const activeKey = key('worker', workerId, 'active_streams');
      const now = String(Date.now());
      const nextValue = await client.decr(activeKey).catch(() => null);
      const multi = client
        .multi()
        .set(key('worker', workerId, 'last_stream_ended_at'), now)
        .set(key('worker', workerId, 'heartbeat'), now, { EX: 30 });
      if (Number(nextValue ?? 0) < 0) multi.set(activeKey, '0');
      if (status === 'aborted') multi.incr(key('worker', workerId, 'stream_abort_count'));
      if (status === 'error') multi.incr(key('worker', workerId, 'stream_error_count'));
      if (streamKey) multi.set(streamKey, status, { EX: 600 });
      await multi.exec().catch(() => null);
    },
    async firstTokenObserved(target, latencyMs) {
      const client = await getClient();
      if (!client || !target) return;
      const workerId = workerIdForTarget(target);
      const sample = Math.max(0, Math.round(Number(latencyMs) || 0));
      const ewmaKey = key('worker', workerId, 'ewma_first_token_ms');
      const samplesKey = key('worker', workerId, 'first_token_samples');
      const prev = readNumber(await client.get(ewmaKey).catch(() => null));
      const next = prev > 0 ? Math.round(prev * 0.8 + sample * 0.2) : sample;
      const nowMs = Date.now();
      const now = String(nowMs);
      await client
        .multi()
        .set(ewmaKey, String(next))
        .set(key('worker', workerId, 'last_first_token_ms'), String(sample))
        .set(key('worker', workerId, 'last_first_token_at'), now)
        .incr(key('worker', workerId, 'first_token_count'))
        .set(key('worker', workerId, 'heartbeat'), now, { EX: 30 })
        .exec()
        .catch(() => null);
      await client
        .sendCommand([
          'ZADD',
          samplesKey,
          String(nowMs),
          `${now}:${sample}:${Math.random().toString(36).slice(2)}`,
        ])
        .catch(() => null);
      await client
        .sendCommand(['ZREMRANGEBYSCORE', samplesKey, '0', String(nowMs - 60 * 60 * 1000)])
        .catch(() => null);
      await client.sendCommand(['EXPIRE', samplesKey, String(2 * 60 * 60)]).catch(() => null);
    },
    async getStatus() {
      const client = await getClient();
      const workers = [];
      for (const target of targets) {
        const workerId = workerIdForTarget(target);
        const values = client
          ? await client
              .mGet([
                key('worker', workerId, 'active_streams'),
                key('worker', workerId, 'active_sessions'),
                key('worker', workerId, 'ewma_first_token_ms'),
                key('worker', workerId, 'error_rate'),
                key('worker', workerId, 'memory_pressure'),
                key('worker', workerId, 'heartbeat'),
                key('worker', workerId, 'drain'),
                key('worker', workerId, 'stream_open_count'),
                key('worker', workerId, 'stream_abort_count'),
                key('worker', workerId, 'stream_error_count'),
                key('worker', workerId, 'last_stream_started_at'),
                key('worker', workerId, 'last_stream_ended_at'),
                key('worker', workerId, 'cpu_load'),
                key('worker', workerId, 'fd_pressure'),
                key('worker', workerId, 'fd_count'),
                key('worker', workerId, 'container_pids'),
                key('worker', workerId, 'container_health'),
                key('worker', workerId, 'metrics_sampled_at'),
                key('worker', workerId, 'last_first_token_ms'),
                key('worker', workerId, 'last_first_token_at'),
                key('worker', workerId, 'first_token_count'),
                key('worker', workerId, 'heartbeat_source'),
                key('worker', workerId, 'heartbeat_ok'),
                key('worker', workerId, 'heartbeat_status_code'),
                key('worker', workerId, 'heartbeat_latency_ms'),
                key('worker', workerId, 'heartbeat_error'),
              ])
              .catch(() => [])
          : [];
        const firstToken5m = await firstTokenWindowStats(client, workerId, 5 * 60 * 1000);
        const firstToken1h = await firstTokenWindowStats(client, workerId, 60 * 60 * 1000);
        workers.push({
          id: workerId,
          target,
          activeStreams: readNumber(values?.[0]),
          activeSessions: readNumber(values?.[1]),
          ewmaFirstTokenMs: readNumber(values?.[2]),
          errorRate: readNumber(values?.[3]),
          memoryPressure: readNumber(values?.[4]),
          heartbeat: values?.[5] ? Number(values[5]) : null,
          drain: /^(1|true|yes)$/i.test(String(values?.[6] ?? '')),
          streamOpenCount: readNumber(values?.[7]),
          streamAbortCount: readNumber(values?.[8]),
          streamErrorCount: readNumber(values?.[9]),
          lastStreamStartedAt: values?.[10] ? Number(values[10]) : null,
          lastStreamEndedAt: values?.[11] ? Number(values[11]) : null,
          cpuLoad: readNumber(values?.[12]),
          fdPressure: readNumber(values?.[13]),
          fdCount: readNumber(values?.[14]),
          containerPids: readNumber(values?.[15]),
          containerHealth: values?.[16] ?? null,
          metricsSampledAt: values?.[17] ? Number(values[17]) : null,
          lastFirstTokenMs: readNumber(values?.[18]),
          lastFirstTokenAt: values?.[19] ? Number(values[19]) : null,
          firstTokenCount: readNumber(values?.[20]),
          heartbeatSource: values?.[21] ?? null,
          heartbeatOk: values?.[22] == null ? null : /^(1|true|yes)$/i.test(String(values[22])),
          heartbeatStatusCode: readNumber(values?.[23]),
          heartbeatLatencyMs: readNumber(values?.[24]),
          heartbeatError: values?.[25] || null,
          firstToken5m,
          firstToken1h,
          score: workerScoreFromValues(values),
        });
      }
      return {
        enabled: Boolean(client),
        namespace,
        workers,
      };
    },
  };
}

async function readJsonBody(req) {
  if (req.method === 'GET' || req.method === 'HEAD') return null;
  const chunks = [];
  for await (const chunk of req) {
    chunks.push(chunk);
  }
  if (chunks.length === 0) return null;
  const raw = Buffer.concat(chunks).toString('utf8');
  if (!raw.trim()) return null;
  return JSON.parse(raw);
}

function sendProxyResponse(res, upstream) {
  res.status(upstream.status);
  upstream.headers.forEach((value, key) => {
    if (key === 'transfer-encoding') return;
    res.setHeader(key, value);
  });
  if (!upstream.body) {
    res.end();
    return;
  }
  Readable.fromWeb(upstream.body).pipe(res);
}

function writeSseProxyErrorAndEnd(res, message) {
  writeSseErrorAndEnd(res, message, { status: 502 });
}

function extractSessionId(req, body) {
  const fromParams = req.params?.sessionId ?? req.params?.id;
  if (fromParams) return fromParams;
  if (body?.session_id) return body.session_id;
  if (body?.sessionId) return body.sessionId;
  return null;
}

function firstUserText(message) {
  return (
    message?.content?.find?.((item) => item?.type === 'text' && typeof item.text === 'string')?.text ??
    ''
  );
}

const IMAGE_URL_LINE_RE = /^\[图片\d+]:\s*(\S.+)$/;
const PUBLIC_HTML_LINK_PATTERN =
  /https?:\/\/[^\s<>"')\]]+\/MindSpace\/([0-9a-f-]{36}|[a-z0-9._-]+)\/([^\s<>"')\]]+\.html)/gi;
const PUBLIC_HTML_MARKDOWN_LINK_PATTERN =
  /\[([^\]\n]*)\]\((https?:\/\/[^\s<>"')\]]+\/MindSpace\/([0-9a-f-]{36}|[a-z0-9._-]+)\/([^\s<>"')\]]+\.html))\)/gi;
const USER_IDENTITY_BLOCK_PATTERN = /^\[用户身份\][\s\S]*?(?:\n{2,}|$)/;
const IMAGE_URL_LINES_PATTERN = /\n*\[图片\d+]: [^\n]+/g;
const TKMIND_VISION_NOTE_PATTERN = /\n*【TKMind 图片分析结果[\s\S]*$/;

function decodePathSegment(segment) {
  try {
    return decodeURIComponent(segment);
  } catch {
    return segment;
  }
}

function encodeUrlPath(relativePath) {
  return String(relativePath ?? '')
    .split('/')
    .filter(Boolean)
    .map((part) => encodeURIComponent(part))
    .join('/');
}

function normalizeStaticHtmlRelativePath(relativePath) {
  const parts = String(relativePath ?? '')
    .replace(/^\/+/, '')
    .split('/')
    .filter((part) => part && part !== '.' && part !== '..');
  if (parts.length === 0) return '';
  if (parts[0].toLowerCase() === 'public') return ['public', ...parts.slice(1)].join('/');
  if (parts.length === 1 && parts[0].toLowerCase().endsWith('.html')) return `public/${parts[0]}`;
  return parts.join('/');
}

function canonicalizeStaticPageUrl(publicUrl, originalRelativePath, canonicalRelativePath) {
  const originalClean = String(originalRelativePath ?? '').replace(/^\/+/, '');
  if (!canonicalRelativePath || canonicalRelativePath === originalClean) return publicUrl;
  const suffix = encodeUrlPath(originalClean).replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
  return String(publicUrl).replace(new RegExp(`${suffix}$`), encodeUrlPath(canonicalRelativePath));
}

function buildMissingPublicHtmlNotice(filename) {
  return `（页面生成未完成，已阻止显示失效链接：${filename || '页面'}。）`;
}

function publicHtmlExistsForUser(owner, relativePath, currentUser) {
  const normalizedOwner = String(owner ?? '').trim().toLowerCase();
  const normalizedUserId = String(currentUser?.id ?? '').trim().toLowerCase();
  const normalizedUsername = String(currentUser?.username ?? '').trim().toLowerCase();
  if (!normalizedOwner) return false;
  if (normalizedOwner !== normalizedUserId && normalizedOwner !== normalizedUsername) return true;
  const normalizedRelativePath = normalizeStaticHtmlRelativePath(relativePath);
  if (!normalizedRelativePath || !normalizedRelativePath.toLowerCase().endsWith('.html')) return false;
  const root = path.resolve(process.cwd(), PUBLISH_ROOT_DIR, normalizedOwner);
  const target = path.resolve(root, normalizedRelativePath);
  if (target !== root && !target.startsWith(`${root}${path.sep}`)) return false;
  return fs.existsSync(target) && fs.statSync(target).isFile();
}

function sanitizeOwnPublicHtmlUrl(publicUrl, owner, rawRelativePath, currentUser) {
  const relativePath = decodePathSegment(rawRelativePath);
  const canonicalRelativePath = normalizeStaticHtmlRelativePath(relativePath);
  if (!publicHtmlExistsForUser(owner, canonicalRelativePath, currentUser)) {
    return {
      ok: false,
      notice: buildMissingPublicHtmlNotice(path.posix.basename(canonicalRelativePath || relativePath)),
    };
  }
  return {
    ok: true,
    url: buildPublicUrl(resolvePublicBaseUrl(), owner, canonicalRelativePath),
  };
}

export function sanitizePublicHtmlLinksInText(text, currentUser) {
  let next = String(text ?? '').replace(
    PUBLIC_HTML_MARKDOWN_LINK_PATTERN,
    (match, label, url, owner, rawRelativePath) => {
      const result = sanitizeOwnPublicHtmlUrl(url, owner, rawRelativePath, currentUser);
      if (!result.ok) return result.notice;
      return label ? `[${label}](${result.url})` : result.url;
    },
  );
  return next.replace(PUBLIC_HTML_LINK_PATTERN, (match, owner, rawRelativePath) => {
    const result = sanitizeOwnPublicHtmlUrl(match, owner, rawRelativePath, currentUser);
    return result.ok ? result.url : result.notice;
  });
}

export function sanitizeUserVisibleMessageText(text, currentUser) {
  return sanitizePublicHtmlLinksInText(
    String(text ?? '')
      .replace(USER_IDENTITY_BLOCK_PATTERN, '')
      .replace(TKMIND_VISION_NOTE_PATTERN, '')
      .replace(IMAGE_URL_LINES_PATTERN, '')
      .trim(),
    currentUser,
  );
}

export function sanitizeSessionMessagePublicHtmlLinks(message, currentUser) {
  if (!message || !Array.isArray(message.content)) return message;
  let changed = false;
  const imageUrls = extractImageUrlsFromMessage(message);
  const content = message.content.map((item) => {
    if (item?.type !== 'text' || typeof item.text !== 'string') return item;
    const nextText = sanitizeUserVisibleMessageText(item.text, currentUser);
    if (nextText === item.text) return item;
    changed = true;
    return { ...item, text: nextText };
  });
  const currentDisplayText =
    typeof message.metadata?.displayText === 'string' ? message.metadata.displayText : null;
  const nextDisplayText =
    currentDisplayText == null ? null : sanitizeUserVisibleMessageText(currentDisplayText, currentUser);
  const metadataChanged =
    (currentDisplayText != null && nextDisplayText !== currentDisplayText) ||
    (!Array.isArray(message.metadata?.imageUrls) && imageUrls.length > 0);
  if (!changed && !metadataChanged) return message;
  return {
    ...message,
    content,
    metadata: {
      ...(message.metadata ?? {}),
      ...(nextDisplayText != null ? { displayText: nextDisplayText } : {}),
      ...(!Array.isArray(message.metadata?.imageUrls) && imageUrls.length > 0
        ? { imageUrls }
        : {}),
    },
  };
}

export function sanitizeSessionConversationPublicHtmlLinks(conversation, currentUser) {
  if (!Array.isArray(conversation)) return conversation;
  return conversation.map((message) => sanitizeSessionMessagePublicHtmlLinks(message, currentUser));
}

function createSessionEventSanitizer(currentUser, { onEvent, onPersistFrame, normalizeReplayIds = false } = {}) {
  let buffer = '';
  const flushChunk = (controller, block) => {
    if (!block) return;
    const parsed = parseSessionSseBlock(block);
    if (!parsed.data) {
      controller.push(block);
      return;
    }
    let event;
    try {
      event = JSON.parse(parsed.data);
    } catch {
      controller.push(block);
      return;
    }
    event = finalizeSessionStreamEvent(event);
    if (typeof onEvent === 'function') {
      try {
        onEvent(event);
      } catch {
        // Ignore side-effect failures and keep SSE flowing to the client.
      }
    }
    if (event?.type === 'Message' && event.message) {
      event.message = sanitizeSessionMessagePublicHtmlLinks(event.message, currentUser);
    } else if (event?.type === 'UpdateConversation' && Array.isArray(event.conversation)) {
      event.conversation = sanitizeSessionConversationPublicHtmlLinks(event.conversation, currentUser);
    }
    if (normalizeReplayIds) {
      const frameId = parsed.id || crypto.randomUUID();
      if (typeof onPersistFrame === 'function' && shouldPersistSessionStreamEvent(event)) {
        try {
          onPersistFrame({ id: frameId, upstreamEventId: parsed.id ?? null, payload: event });
        } catch {
          // Ignore persistence failures and keep SSE flowing to the client.
        }
      }
      controller.push(formatSessionStreamSseChunk({
        id: frameId,
        event: parsed.eventName,
        data: event,
      }));
      return;
    }
    const lines = block.split('\n');
    const sanitizedLines = lines.map((line) => {
      if (!line.startsWith('data: ')) return line;
      return `data: ${JSON.stringify(event)}`;
    });
    controller.push(`${sanitizedLines.join('\n')}\n`);
  };

  return new Transform({
    transform(chunk, _encoding, callback) {
      buffer += chunk.toString('utf8');
      let boundary = buffer.indexOf('\n\n');
      while (boundary >= 0) {
        const block = buffer.slice(0, boundary + 2);
        buffer = buffer.slice(boundary + 2);
        flushChunk(this, block);
        boundary = buffer.indexOf('\n\n');
      }
      callback();
    },
    flush(callback) {
      flushChunk(this, buffer);
      buffer = '';
      callback();
    },
  });
}

export function extractImageUrlsFromMessage(userMessage) {
  const urls = [];
  const imageUrls = userMessage?.metadata?.imageUrls;
  if (Array.isArray(imageUrls)) {
    urls.push(...imageUrls);
  }

  const content = userMessage?.content;
  if (Array.isArray(content)) {
    for (const item of content) {
      if (item?.type === 'image_url' && item.image_url?.url) {
        urls.push(item.image_url.url);
        continue;
      }
      if (item?.type !== 'text' || typeof item.text !== 'string') continue;
      for (const line of item.text.split('\n')) {
        const match = line.trim().match(IMAGE_URL_LINE_RE);
        if (match?.[1]) urls.push(match[1].trim());
      }
    }
  }

  return [...new Set(urls.filter((url) => typeof url === 'string' && url.trim()))];
}

function messageHasImages(userMessage) {
  return extractImageUrlsFromMessage(userMessage).length > 0;
}

function prependAgentTextToUserMessage(body, transformText) {
  const originalText = firstUserText(body?.user_message);
  if (!originalText?.trim()) return body;

  const agentText = transformText(originalText);
  if (!agentText || agentText === originalText) return body;

  const userMessage = body.user_message ?? {};
  const content = Array.isArray(userMessage.content) ? [...userMessage.content] : [];
  const firstTextIndex = content.findIndex(
    (item) => item?.type === 'text' && typeof item.text === 'string',
  );
  if (firstTextIndex >= 0) {
    content[firstTextIndex] = { ...content[firstTextIndex], text: agentText };
  } else {
    content.unshift({ type: 'text', text: agentText });
  }

  return {
    ...body,
    user_message: {
      ...userMessage,
      content,
      metadata: {
        ...(userMessage.metadata ?? {}),
        displayText:
          userMessage.metadata?.displayText && String(userMessage.metadata.displayText).trim()
            ? userMessage.metadata.displayText
            : originalText,
      },
    },
  };
}

function injectCurrentTimeAnchor(body, timezone) {
  const tz = String(timezone ?? process.env.H5_DEFAULT_TIMEZONE ?? 'Asia/Shanghai').trim() || 'Asia/Shanghai';
  return prependAgentTextToUserMessage(body, (originalText) => {
    if (originalText.includes('TKMind 当前时间基准')) return originalText;
    return `${buildCurrentTimeAgentPrefix({ timezone: tz })}${originalText}`;
  });
}

function injectTaskRoutingHint(body, sessionPolicy) {
  return prependAgentTextToUserMessage(body, (originalText) =>
    buildTaskRoutingAgentText(originalText, sessionPolicy),
  );
}

export async function buildVisionPayload({
  userMessage,
  userId,
  publishLayout,
  localFetchAsset,
  llmProviderService,
  imgproxySigner = null,
  fetchImpl = undiciFetch,
}) {
  void imgproxySigner;
  if (!llmProviderService || !userId) return null;
  const rawImageUrls = extractImageUrlsFromMessage(userMessage);
  if (rawImageUrls.length === 0) return null;
  const originalDisplayText =
    typeof userMessage?.metadata?.displayText === 'string' &&
    userMessage.metadata.displayText.trim()
      ? userMessage.metadata.displayText
      : Array.isArray(userMessage?.content)
        ? userMessage.content
            .filter((item) => item?.type === 'text' && typeof item.text === 'string')
            .map((item) => item.text)
            .join('\n')
            .trim()
        : '';

  const imageItems = [];
  for (const rawUrl of rawImageUrls) {
    try {
      const match = rawUrl.match(/\/mindspace\/v1\/assets\/([^/?#]+)(?:\/download)?/);
      let buffer;
      let mimeType = 'image/jpeg';
      const fetchableUrl = (() => {
        if (/^https?:\/\//i.test(rawUrl)) return rawUrl;
        if (rawUrl.startsWith('/')) {
          const baseUrl = process.env.H5_PUBLIC_BASE_URL || 'http://127.0.0.1:8081';
          return new URL(rawUrl, baseUrl).toString();
        }
        return rawUrl;
      })();
      try {
        const response = await fetchImpl(fetchableUrl);
        if (!response.ok) {
          throw new Error(`Vision fetch failed: ${response.status}`);
        }
        buffer = Buffer.from(await response.arrayBuffer());
        mimeType = response.headers.get('content-type') || 'image/jpeg';
      } catch (fetchErr) {
        if (!match || !localFetchAsset) {
          throw fetchErr;
        }
        const assetId = decodeURIComponent(match[1]);
        console.warn(`Vision fetch fallback for asset ${assetId}:`, fetchErr instanceof Error ? fetchErr.message : fetchErr);
        const asset = await localFetchAsset(userId, assetId);
        buffer = asset.buffer;
        mimeType = asset.mimeType;
      }

      let relativePath = rawUrl;
      try {
        const parsed = new URL(rawUrl);
        relativePath = parsed.pathname + parsed.search;
      } catch { /* keep rawUrl */ }
      const publicStandardUrl = buildPublicStandardImageUrl(rawUrl, userId, publishLayout);
      imageItems.push({
        mimeType,
        data: buffer.toString('base64'),
        relativePath,
        rawUrl,
        embedUrl: publicStandardUrl ?? relativePath,
      });
    } catch (err) {
      console.warn('Vision image fetch skipped:', err instanceof Error ? err.message : err);
    }
  }
  if (imageItems.length === 0) return null;

  const visionDescription = await llmProviderService
    .analyzeImagesWithVision(imageItems, '请详细描述图片的视觉内容：主体、颜色、风格、构图，不要生成代码或页面方案')
    .catch(() => null);

  const pathList = imageItems
    .map((item, i) => `图片${i + 1}: <img src="${item.embedUrl}" alt="图片${i + 1}">`)
    .join('  ');

  const publicUrlPrefix = publishLayout?.publicUrl
    ? `${String(publishLayout.publicUrl).replace(/\/$/, '')}/public/`
    : null;

  const injectedNote =
    '\n\n【TKMind 图片分析结果 — 仅供执行参考，不要向用户复述此段内容】\n' +
    (visionDescription
      ? `Qwen VL 图片描述：\n${visionDescription}\n\n`
      : '') +
    '写作约束：不得改写图片里人物的年龄、性别、人数或主体关系；如果用户明确要求儿童语气或童趣风格，也只能调整表达方式，不能把图片主体改写成儿童场景。\n' +
    `图片 HTML 嵌入路径（直接写入 <img> 标签；以下都是无需 cookie 的公开压缩标准图片，禁止使用 local://、/users/ 私有路径、原图地址或需要登录态的下载链接）：\n${pathList}\n` +
    '执行要求：必须先调用 load_skill → static-page-publish（每次生成页面都要调用，不可省略），' +
    '再用 write_file 写入 public/页面.html（禁止用 shell/cat/heredoc 写 HTML，容器内文件不会出现在公网链接）；' +
    (publicUrlPrefix
      ? `完成后向用户给出 Markdown 可点击链接，格式：[页面标题](${publicUrlPrefix}<文件名>.html)；`
      : '完成后按技能说明里的链接格式给用户一个 Markdown 可点击链接；') +
    '不要向用户展示 HTML 代码块。';

  let updatedContent = Array.isArray(userMessage?.content) ? [...userMessage.content] : [];
  for (const item of imageItems) {
    updatedContent = updatedContent.map((c) => {
      if (c?.type !== 'text' || typeof c.text !== 'string') return c;
      const updated = c.text.replaceAll(item.rawUrl, item.embedUrl);
      return updated === c.text ? c : { ...c, text: updated };
    });
  }

  const lastTextIdx = updatedContent.reduceRight(
    (found, item, i) => (found === -1 && item?.type === 'text' ? i : found),
    -1,
  );
  if (lastTextIdx >= 0) {
    updatedContent[lastTextIdx] = {
      ...updatedContent[lastTextIdx],
      text: updatedContent[lastTextIdx].text + injectedNote,
    };
  } else {
    updatedContent.push({ type: 'text', text: injectedNote });
  }

  return {
    userMessage: {
      ...userMessage,
      content: updatedContent,
      metadata: {
        ...(userMessage.metadata ?? {}),
        ...(originalDisplayText ? { displayText: originalDisplayText } : {}),
      },
    },
    billableImageCount: visionDescription ? 1 : 0,
  };
}

export function createTkmindProxy({
  // GOOSED PROXY BOUNDARY: canonical H5 chat → goosed adapter (Patch 5, goosed-proxy-boundary.mjs)
  apiTarget,
  apiTargets,
  apiSecret,
  userAuth,
  sessionAccess = null,
  sessionStreamStore = null,
  llmProviderService,
  localFetchAsset,
  subscriptionService,
  sessionSnapshotService,
  conversationMemoryService,
  memoryV2,
}) {
  const sessionStore =
    sessionAccess ?? createSessionAccess({ userAuth, enabled: isSessionBrokerEnabled() });
  const brokerMetrics = isSessionBrokerMetricsEnabled() && sessionStore.enabled
    ? createSessionBrokerMetrics({ logger: console })
    : null;
  const targets = apiTargets?.length ? apiTargets : apiTarget ? [apiTarget] : [];
  const primaryTarget = targets[0] ?? apiTarget ?? '';
  let rrIdx = 0;

  let imgproxySigner = null;
  if (process.env.IMGPROXY_BASE_URL && process.env.IMGPROXY_SIGNING_KEY && process.env.IMGPROXY_SIGNING_SALT) {
    try {
      imgproxySigner = createImgproxySigner(
        process.env.IMGPROXY_SIGNING_KEY,
        process.env.IMGPROXY_SIGNING_SALT,
      );
      console.log('[TKMindProxy] imgproxy signer initialized');
    } catch (err) {
      console.warn('[TKMindProxy] imgproxy signer init failed:', err instanceof Error ? err.message : err);
    }
  }

  async function targetHealthy(target) {
    try {
      const upstream = await apiFetch(target, apiSecret, '/status', {
        method: 'GET',
        signal: AbortSignal.timeout(1500),
      });
      return upstream.ok;
    } catch {
      return false;
    }
  }
  const runtimeRouter = createRuntimeRouter({ targets, targetHealthy });

  function isGeneratedSessionName(name) {
    const normalized = typeof name === 'string' ? name.trim() : '';
    return /^20\d{6}(?:[_-]\d+)?$/.test(normalized);
  }

  // A session "name" that carries no real topic: empty, or Goose's default
  // placeholder before its describe step has run.
  function hasDefaultSessionName(session) {
    if (session?.user_set_name) return false;
    const name = typeof session?.name === 'string' ? session.name.trim() : '';
    return name === '' || name === 'New Chat' || name === '新对话' || isGeneratedSessionName(name);
  }

  // For sessions still carrying a default name, fill in a title derived from the
  // first user message (kept in the snapshot cache) so the history list shows a
  // meaningful label instead of "会话 <id>". Best-effort: never blocks the list.
  async function enrichSessionHistory(sessions, userId) {
    const needsSummary = sessions.filter(
      (session) => hasDefaultSessionName(session) || Number(session?.message_count ?? 0) === 0,
    );
    if (needsSummary.length === 0) return;

    if (sessionSnapshotService?.isEnabled?.()) {
      try {
        const summaries = await sessionSnapshotService.getHistorySummaries(needsSummary.map((s) => s.id));
        for (const session of needsSummary) {
          const summary = summaries.get(session.id);
          if (!summary) continue;
          if (summary.title && hasDefaultSessionName(session)) {
            session.name = summary.title;
          }
          if (Number(session?.message_count ?? 0) === 0 && Number(summary.messageCount ?? 0) > 0) {
            session.message_count = Number(summary.messageCount);
          }
        }
      } catch {
        // Non-fatal: fall back to DB summaries below.
      }
    }

    const stillNeedsSummary = needsSummary.filter(
      (session) => hasDefaultSessionName(session) || Number(session?.message_count ?? 0) === 0,
    );
    if (stillNeedsSummary.length === 0 || !conversationMemoryService?.loadSessionListSummaries) return;
    try {
      const dbSummaries = await conversationMemoryService.loadSessionListSummaries(
        userId,
        stillNeedsSummary.map((session) => session.id),
      );
      for (const session of stillNeedsSummary) {
        const summary = dbSummaries.get(session.id);
        if (!summary) continue;
        if (summary.title && hasDefaultSessionName(session)) {
          session.name = summary.title;
        }
        if (Number(session?.message_count ?? 0) === 0 && Number(summary.messageCount ?? 0) > 0) {
          session.message_count = Number(summary.messageCount);
        }
        if (!session.updated_at && summary.updatedAt) {
          session.updated_at = summary.updatedAt;
        }
      }
    } catch {
      // Non-fatal: fall back to the client-side label.
    }
  }

  function projectSessionSummary(session) {
    return {
      id: session?.id,
      name: typeof session?.name === 'string' ? session.name : '',
      message_count: Number(session?.message_count ?? 0),
      created_at: session?.created_at ?? undefined,
      updated_at: session?.updated_at ?? undefined,
      user_set_name: Boolean(session?.user_set_name),
      recipe: session?.recipe ?? null,
    };
  }

  function sortSessionsByRecent(left, right) {
    const leftTime = Date.parse(left?.updated_at ?? left?.created_at ?? '') || 0;
    const rightTime = Date.parse(right?.updated_at ?? right?.created_at ?? '') || 0;
    return rightTime - leftTime;
  }

  function matchesSessionQuery(summary, rawQuery) {
    const query = String(rawQuery ?? '').trim().toLowerCase();
    if (!query) return true;
    const haystacks = [
      summary?.name,
      summary?.recipe?.title,
      summary?.id,
    ]
      .filter((value) => typeof value === 'string' && value.trim())
      .map((value) => value.toLowerCase());
    return haystacks.some((value) => value.includes(query));
  }

  function paginateSessionConversation(conversation, beforeValue, limitValue) {
    const visibleConversation = Array.isArray(conversation)
      ? conversation.filter((message) => message?.metadata?.userVisible)
      : [];
    const total = visibleConversation.length;
    const before = Math.max(Number(beforeValue ?? 0) || 0, 0);
    const requestedLimit = Number(limitValue ?? 0) || 0;
    if (requestedLimit <= 0) {
      return {
        conversation: visibleConversation,
        page: {
          total,
          before: 0,
          limit: total,
          returned: total,
          has_more_before: false,
        },
      };
    }

    const limit = Math.min(Math.max(requestedLimit, 1), 200);
    const end = Math.max(total - before, 0);
    const start = Math.max(end - limit, 0);
    const paged = visibleConversation.slice(start, end);
    return {
      conversation: paged,
      page: {
        total,
        before,
        limit,
        returned: paged.length,
        has_more_before: start > 0,
      },
    };
  }

  async function pickTarget() {
    if (targets.length <= 1) return primaryTarget;
    const fallbackPick = async () => {
      for (let i = 0; i < targets.length; i += 1) {
        const target = targets[rrIdx];
        rrIdx = (rrIdx + 1) % targets.length;
        if (await targetHealthy(target)) return target;
      }
      return primaryTarget;
    };
    if (runtimeRouter) {
      const orderedTargets = targets.map((_, i) => targets[(rrIdx + i) % targets.length]);
      rrIdx = (rrIdx + 1) % targets.length;
      return runtimeRouter.pickTarget(fallbackPick, orderedTargets);
    }
    return fallbackPick();
  }

  async function rememberSessionTarget(sessionId, target) {
    if (!runtimeRouter || !sessionId || !target) return;
    await runtimeRouter.registerSession(sessionId, target).catch(() => null);
  }

  async function markStreamStarted(sessionId, target) {
    if (!runtimeRouter || !sessionId || !target) return;
    await runtimeRouter.streamStarted(sessionId, target).catch(() => null);
  }

  async function markStreamEnded(sessionId, target, options = {}) {
    if (!runtimeRouter || !sessionId || !target) return;
    await runtimeRouter.streamEnded(sessionId, target, options).catch(() => null);
  }

  function markFirstTokenObserved(target, latencyMs) {
    if (!runtimeRouter || !target) return;
    void runtimeRouter.firstTokenObserved(target, latencyMs).catch(() => null);
  }

  async function getRuntimeStatus() {
    const targetStatuses = [];
    for (const target of targets) {
      targetStatuses.push({
        target,
        healthy: await targetHealthy(target),
      });
    }
    const routerStatus = runtimeRouter
      ? await runtimeRouter.getStatus().catch((err) => ({
          enabled: false,
          error: err instanceof Error ? err.message : String(err),
        }))
      : { enabled: false, reason: 'MEMIND_RUNTIME_REDIS_URL not configured' };
    return {
      publicBaseUrl: process.env.H5_PUBLIC_BASE_URL ?? null,
      router: routerStatus,
      toolRuntime: {
        defaultMode: 'chat',
        codeToolMode: 'code',
        chatInjectsCodeTools: false,
        codeRunsEnabled: ['1', 'true', 'yes', 'on'].includes(
          String(process.env.MEMIND_AGENT_CODE_RUNS_ENABLED ?? '').trim().toLowerCase(),
        ),
        aiderTimeoutMs: Number(process.env.MEMIND_AIDER_TIMEOUT_MS ?? 600_000),
        openhandsTimeoutMs: Number(process.env.MEMIND_OPENHANDS_TIMEOUT_MS ?? 900_000),
      },
      memory: memoryV2?.getStatus
        ? await Promise.resolve(memoryV2.getStatus())
        : {
            enabled: Boolean(conversationMemoryService?.isEnabled?.()),
            backend: 'legacy',
            selectedBackend: conversationMemoryService ? 'legacy-conversation-memory' : null,
            profileEnabled: false,
            eventLogEnabled: Boolean(conversationMemoryService?.isEnabled?.()),
            vectorEnabled: false,
            failOpen: true,
            backends: [
              {
                name: 'legacy-conversation-memory',
                available: Boolean(conversationMemoryService),
                supports: {
                  resolve: Boolean(conversationMemoryService?.listMemories),
                  write: Boolean(conversationMemoryService?.saveAndAnalyze),
                  compact: Boolean(conversationMemoryService?.analyzeUser),
                },
              },
            ],
          },
      targets: targetStatuses,
    };
  }

  async function resolveUserMemories(
    userId,
    { sessionId = null, query = null, limit = null, forceDeepReasoning = false, recallQuestion = false } = {},
  ) {
    if (!userId) return [];
    const intervention = resolveMemoryInterventionMode({
      forceDeepReasoning,
      recallQuestion,
      context: 'agent',
    });
    const resolvedLimit = Number.isFinite(Number(limit)) && Number(limit) > 0
      ? Number(limit)
      : memoryLimitForIntervention(intervention, { context: 'agent' });
    if (resolvedLimit <= 0) return [];
    if (memoryV2?.resolve) {
      const resolved = await memoryV2.resolve({
        userId,
        sessionId,
        query,
        limit: resolvedLimit,
      }).catch(() => null);
      return Array.isArray(resolved?.memories) ? resolved.memories : [];
    }
    return conversationMemoryService?.listMemories
      ? conversationMemoryService.listMemories(userId, { limit: resolvedLimit }).catch(() => [])
      : [];
  }

  async function startSessionForUser(
    userId,
    {
      workingDir,
      sessionPolicy = null,
      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({
        ...(resolvedWorkingDir ? { working_dir: resolvedWorkingDir } : {}),
        enable_context_memory: resolvedSessionPolicy?.enableContextMemory,
        ...(resolvedSessionPolicy?.extensionOverrides
          ? { extension_overrides: resolvedSessionPolicy.extensionOverrides }
          : {}),
        ...(recipe ? { recipe } : {}),
      }),
    });
    const text = await upstream.text();
    if (!upstream.ok) {
      throw new Error(text || `创建会话失败 (${upstream.status})`);
    }
    const session = JSON.parse(text);
    if (!session?.id) {
      throw new Error('创建会话失败：缺少 session id');
    }
    await rememberSessionTarget(session.id, startTarget);
    await sessionStore.registerAgentSession(userId, session.id, startTarget);
    if (resolvedSessionPolicy?.gooseMode) {
      const modeRes = await apiFetch(startTarget, apiSecret, '/agent/update_session', {
        method: 'POST',
        body: JSON.stringify({
          session_id: session.id,
          goose_mode: resolvedSessionPolicy.gooseMode,
        }),
      });
      if (!modeRes.ok) {
        const modeText = await modeRes.text().catch(() => '');
        throw new Error(modeText || `设置会话模式失败 (${modeRes.status})`);
      }
    }
    const publishLayout = await userAuth.getUserPublishLayout(userId);
    const userMemories = await resolveUserMemories(userId, { sessionId: session.id });
    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;
  }

  async function resolveTarget(sessionId) {
    if (targets.length <= 1 || !sessionId) return primaryTarget;
    try {
      const routedTarget = await runtimeRouter?.resolveSessionTarget(sessionId);
      if (routedTarget) return routedTarget;
      const { target, node } = await sessionStore.getSessionTarget(sessionId);
      // Prefer the pinned URL: stable across reordering/resizing the target list.
      // Only honor it if that upstream is still configured; otherwise fall back to
      // the legacy integer index, then to primary.
      if (target && targets.includes(target)) return target;
      if (brokerMetrics) {
        if (target && !targets.includes(target)) {
          brokerMetrics.resolveTargetMiss({
            sessionId,
            target,
            node,
            reason: 'stale_pinned_target',
          });
        }
      }
      return targets[node] ?? primaryTarget;
    } catch {
      return primaryTarget;
    }
  }

  async function applySessionLlmProvider(sessionId) {
    if (!llmProviderService || !sessionId) return null;
    try {
      const target = await resolveTarget(sessionId);
      return await llmProviderService.applyBestProviderForSession(
        sessionId,
        (url, init) => apiFetch(target, apiSecret, `${url.pathname}${url.search}`, init),
      );
    } catch (err) {
      console.warn(
        'LLM provider apply skipped:',
        err instanceof Error ? err.message : err,
      );
      return null;
    }
  }

  async function applyLocalFallbackForSession(sessionId) {
    if (!llmProviderService || !sessionId) return null;
    const target = await resolveTarget(sessionId);
    return llmProviderService.applyLocalFallbackForSession(
      sessionId,
      (url, init) => apiFetch(target, apiSecret, `${url.pathname}${url.search}`, init),
    );
  }

  // Tracks which sessions are currently using the vision provider so we only
  // issue a switch-back call when the session was actually on vision.
  const visionActiveSessions = new Set();

  async function applyVisionProviderForSession(sessionId) {
    if (!llmProviderService || !sessionId) return null;
    const target = await resolveTarget(sessionId);
    return llmProviderService.applyVisionProviderForSession(
      sessionId,
      (url, init) => apiFetch(target, apiSecret, `${url.pathname}${url.search}`, init),
    );
  }

  // Two-step vision approach:
  //   Step 1 — Call Qwen VL directly (not through Goose) to analyze the image.
  //   Step 2 — Inject Qwen's text description + server-relative image paths into the
  //             user_message that goes to DeepSeek via Goose.  DeepSeek retains full
  //             tool-calling capability (write_file, etc.) and creates the page properly.
  async function buildVisionBody(userMessage, userId, publishLayout) {
    return buildVisionPayload({
      userMessage,
      userId,
      publishLayout,
      localFetchAsset,
      llmProviderService,
      imgproxySigner,
    });
  }

  async function getSessionPolicyForToolMode(userId, toolMode = 'chat') {
    if (toolMode === 'code' && userAuth.getCodeAgentSessionPolicy) {
      return userAuth.getCodeAgentSessionPolicy(userId);
    }
    return userAuth.getAgentSessionPolicy(userId);
  }

  async function reconcileSessionPolicyForUser(
    userId,
    sessionId,
    { toolMode = 'chat', query = null, forceDeepReasoning = false } = {},
  ) {
    if (!userId || !sessionId) return;
    const target = await resolveTarget(sessionId);
    const workingDir = await userAuth.resolveWorkingDir(userId);
    const sessionPolicy = await getSessionPolicyForToolMode(userId, toolMode);
    const publishLayout = await userAuth.getUserPublishLayout(userId);
    const userMemories = await resolveUserMemories(userId, {
      sessionId,
      query,
      forceDeepReasoning,
    });
    await reconcileAgentSession(
      (pathname, init) => apiFetch(target, apiSecret, pathname, init),
      sessionId,
      {
        workingDir,
        sessionPolicy,
        sandboxConstraints: publishLayout?.constraints ?? null,
        tolerateInvalidWorkingDir: true,
        userMemories,
        userContext: publishLayout
          ? {
              userId,
              displayName: publishLayout.displayName,
              username: publishLayout.username,
              slug: publishLayout.slug,
            }
          : null,
      },
    );
  }

  async function prepareSessionReplyBody(
    userId,
    sessionId,
    requestId,
    userMessage,
    { toolMode = 'chat', forceDeepReasoning = false } = {},
  ) {
    if (!userId || !sessionId) throw new Error('缺少会话信息');
    const owns = await sessionStore.validateOwnership(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, {
      toolMode,
      query: firstUserText(userMessage),
      forceDeepReasoning,
    });
    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),
      );
    }
    body = {
      ...body,
      user_message: ensureGooseUserMessageMetadata(body.user_message),
    };
    const target = await resolveTarget(sessionId);
    return { body, target };
  }

  async function submitSessionReplyForUser(
    userId,
    sessionId,
    requestId,
    userMessage,
    options = {},
  ) {
    const { body, target } = await prepareSessionReplyBody(
      userId,
      sessionId,
      requestId,
      userMessage,
      options,
    );
    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 };
  }

  async function submitSessionReplyAndAwaitFinishForUser(
    userId,
    sessionId,
    requestId,
    userMessage,
    { toolMode = 'chat', forceDeepReasoning = false, timeoutMs = 15 * 60 * 1000 } = {},
  ) {
    const { body, target } = await prepareSessionReplyBody(
      userId,
      sessionId,
      requestId,
      userMessage,
      { toolMode, forceDeepReasoning },
    );
    const eventsResponse = await apiFetch(
      target,
      apiSecret,
      `/sessions/${encodeURIComponent(sessionId)}/events`,
      {
        method: 'GET',
        headers: { Accept: 'text/event-stream' },
      },
    );
    if (!eventsResponse.ok || !eventsResponse.body) {
      const text = await eventsResponse.text().catch(() => '');
      throw new Error(text || `无法建立 session 事件流 (${eventsResponse.status})`);
    }

    const finishPromise = consumeSessionEventsUntilFinish(eventsResponse.body, {
      requestId,
      timeoutMs,
    });
    const replyResponse = await apiFetch(
      target,
      apiSecret,
      `/sessions/${encodeURIComponent(sessionId)}/reply`,
      {
        method: 'POST',
        body: JSON.stringify(body),
      },
    );
    if (!replyResponse.ok) {
      const text = await replyResponse.text().catch(() => '');
      throw new Error(text || `发送失败 (${replyResponse.status})`);
    }
    replyResponse.body?.cancel?.().catch?.(() => {});
    const finish = await finishPromise;
    return { ok: true, ...finish };
  }

  const requireUser = async (req, res, next) => {
    try {
      const session = req.userSession;
      if (!session) {
        res.status(401).json({ message: '未登录' });
        return;
      }
      const me = await userAuth.getMe(req.userToken);
      if (!me) {
        res.status(401).json({ message: '登录已过期' });
        return;
      }
      req.currentUser = me;
      next();
    } catch (err) {
      res.status(500).json({ message: err instanceof Error ? err.message : '认证失败' });
    }
  };

  const ensureChatAllowed = async (req, res, next) => {
    const gate = await userAuth.canUseChat(req.currentUser.id);
    if (!gate.ok) {
      res.status(402).json({
        message: gate.message,
        code: gate.code,
        balanceCents: gate.balanceCents,
        minRechargeCents: gate.minRechargeCents,
        suggestedTiers: gate.suggestedTiers,
      });
      return;
    }
    next();
  };

  const handlers = {
    'POST /agent/start': [
      requireUser,
      ensureChatAllowed,
      async (req, res) => {
        try {
          const workingDir = await userAuth.resolveWorkingDir(req.currentUser.id);
          const sessionPolicy = await userAuth.getAgentSessionPolicy(req.currentUser.id);
          const startTarget = await pickTarget();
          const upstream = await apiFetch(startTarget, apiSecret, '/agent/start', {
            method: 'POST',
            body: JSON.stringify({
              working_dir: workingDir,
              enable_context_memory: sessionPolicy.enableContextMemory,
              ...(sessionPolicy.extensionOverrides
                ? { extension_overrides: sessionPolicy.extensionOverrides }
                : {}),
              ...(req.body?.recipe ? { recipe: req.body.recipe } : {}),
            }),
          });
          const text = await upstream.text();
          if (!upstream.ok) {
            res.status(upstream.status).send(text);
            return;
          }
          const session = JSON.parse(text);
          if (session?.id) {
            await rememberSessionTarget(session.id, startTarget);
            await sessionStore.registerAgentSession(
              req.currentUser.id,
              session.id,
              startTarget,
            );
            if (sessionPolicy.gooseMode) {
              const modeRes = await apiFetch(startTarget, apiSecret, '/agent/update_session', {
                method: 'POST',
                body: JSON.stringify({
                  session_id: session.id,
                  goose_mode: sessionPolicy.gooseMode,
                }),
              });
              if (!modeRes.ok) {
                const modeText = await modeRes.text().catch(() => '');
                res.status(modeRes.status).send(modeText || '设置会话模式失败');
                return;
              }
            }
            const publishLayout = await userAuth.getUserPublishLayout(req.currentUser.id);
            const userMemories = await resolveUserMemories(req.currentUser.id, {
              sessionId: session.id,
            });
            const api = (pathname, init) => apiFetch(startTarget, apiSecret, pathname, init);
            try {
              await reconcileAgentSession(api, session.id, {
                workingDir,
                sessionPolicy,
                sandboxConstraints: publishLayout?.constraints ?? null,
                userMemories,
                userContext: publishLayout
                  ? {
                      userId: req.currentUser.id,
                      displayName: publishLayout.displayName,
                      username: publishLayout.username,
                      slug: publishLayout.slug,
                    }
                  : null,
              });
            } catch (reconcileErr) {
              res.status(500).json({
                message:
                  reconcileErr instanceof Error
                    ? `会话策略同步失败：${reconcileErr.message}`
                    : '会话策略同步失败',
              });
              return;
            }
            await applySessionLlmProvider(session.id);
          }
          res.status(upstream.status).json(session);
        } catch (err) {
          res.status(500).json({ message: err instanceof Error ? err.message : '启动会话失败' });
        }
      },
    ],

    'POST /agent/resume': [
      requireUser,
      ensureChatAllowed,
      async (req, res) => {
        try {
          const sessionId = req.body?.session_id;
          if (!sessionId) {
            res.status(400).json({ message: '缺少 session_id' });
            return;
          }
          const owns = await sessionStore.validateOwnership(req.currentUser.id, sessionId);
          if (!owns) {
            res.status(403).json({ message: '无权访问该会话' });
            return;
          }

          const resumeTarget = await resolveTarget(sessionId);
          const upstream = await apiFetch(resumeTarget, apiSecret, '/agent/resume', {
            method: 'POST',
            body: JSON.stringify(req.body ?? {}),
          });
          const text = await upstream.text();
          if (!upstream.ok) {
            res.status(upstream.status).send(text);
            return;
          }

          const payload = JSON.parse(text);
          const skipReconcile = req.body?.skip_reconcile === true;
          if (!skipReconcile) {
            const workingDir = await userAuth.resolveWorkingDir(req.currentUser.id);
            const sessionPolicy = await userAuth.getAgentSessionPolicy(req.currentUser.id);
            const publishLayout = await userAuth.getUserPublishLayout(req.currentUser.id);
            const userMemories = await resolveUserMemories(req.currentUser.id, {
              sessionId,
            });
            try {
              await reconcileAgentSession(
                (pathname, init) => apiFetch(resumeTarget, apiSecret, pathname, init),
                sessionId,
                {
                  workingDir,
                  sessionPolicy,
                  sandboxConstraints: publishLayout?.constraints ?? null,
                  tolerateInvalidWorkingDir: true,
                  userMemories,
                  userContext: publishLayout
                    ? {
                        userId: req.currentUser.id,
                        displayName: publishLayout.displayName,
                        username: publishLayout.username,
                        slug: publishLayout.slug,
                      }
                    : null,
                },
              );
            } catch (reconcileErr) {
              res.status(500).json({
                message:
                  reconcileErr instanceof Error
                    ? `会话恢复后策略同步失败：${reconcileErr.message}`
                    : '会话恢复后策略同步失败',
              });
              return;
            }
          }

          await applySessionLlmProvider(sessionId);

          res.status(upstream.status).json(payload);
        } catch (err) {
          res.status(500).json({
            message: sanitizeUserFacingProxyMessage(
              err instanceof Error ? err.message : '恢复会话失败',
              '恢复会话失败',
            ),
          });
        }
      },
    ],

    'GET /sessions': [
      requireUser,
      async (req, res) => {
        try {
          const offset = Math.max(Number(req.query?.offset ?? 0) || 0, 0);
          const requestedLimit = Number(req.query?.limit ?? 20) || 20;
          const limit = Math.min(Math.max(requestedLimit, 1), 100);
          const query = String(req.query?.query ?? '').trim();
          const owned = await userAuth.listOwnedSessionIds(req.currentUser.id);
          const sessionsById = new Map();
          let healthyTargets = 0;
          let lastFailure = null;

          for (const target of targets) {
            try {
              const upstream = await apiFetch(target, apiSecret, '/sessions', {
                method: 'GET',
                signal: AbortSignal.timeout(12_000),
              });
              const text = await upstream.text();
              if (!upstream.ok) {
                lastFailure = text || `upstream ${upstream.status}`;
                continue;
              }
              healthyTargets += 1;
              const payload = JSON.parse(text);
              for (const item of payload.sessions ?? []) {
                if (owned.has(item.id)) sessionsById.set(item.id, item);
              }
            } catch (err) {
              lastFailure = sanitizeUserFacingProxyMessage(
                err instanceof Error ? err.message : '读取会话失败',
                '读取会话失败',
              );
            }
          }

          if (healthyTargets === 0) {
            res.status(502).json({ message: lastFailure ?? '后端连接失败，请稍后重试' });
            return;
          }
          if (healthyTargets < targets.length) {
            res.setHeader('X-TKMind-Degraded', '1');
          }

          const directSessionIds = [...owned].filter(isDirectChatSessionId);
          const ownedMissingFromGoose = [...owned].filter((sessionId) => !sessionsById.has(sessionId));
          const snapshotFallbackIds = [...new Set([...directSessionIds, ...ownedMissingFromGoose])];
          if (snapshotFallbackIds.length > 0 && sessionSnapshotService?.isEnabled?.()) {
            for (const sessionId of snapshotFallbackIds) {
              if (sessionsById.has(sessionId)) continue;
              const snapshot = await sessionSnapshotService.get(sessionId).catch(() => null);
              if (!snapshot?.session) continue;
              sessionsById.set(sessionId, {
                ...snapshot.session,
                message_count: snapshot.meta?.synced_msg_count ?? snapshot.messages?.length ?? 0,
                updated_at: snapshot.meta?.source_updated_at ?? snapshot.session.updated_at,
                conversation: undefined,
              });
            }
          }

          const stillMissingFromGoose = [...owned].filter((sessionId) => !sessionsById.has(sessionId));
          if (stillMissingFromGoose.length > 0 && conversationMemoryService?.loadSessionListSummaries) {
            const dbSummaries = await conversationMemoryService
              .loadSessionListSummaries(req.currentUser.id, stillMissingFromGoose)
              .catch(() => new Map());
            for (const sessionId of stillMissingFromGoose) {
              const summary = dbSummaries.get(sessionId);
              if (!summary) continue;
              sessionsById.set(sessionId, {
                id: sessionId,
                name: summary.title || sessionId,
                message_count: Number(summary.messageCount ?? 0),
                updated_at: summary.updatedAt,
                user_set_name: false,
                recipe: null,
              });
            }
          }

          const sessions = [...sessionsById.values()].sort(sortSessionsByRecent);
          await enrichSessionHistory(sessions, req.currentUser.id);
          if (typeof userAuth.getSessionOrigins === 'function' && sessions.length > 0) {
            try {
              const origins = await userAuth.getSessionOrigins(sessions.map((s) => s.id));
              for (const session of sessions) {
                session.origin = origins.get(session.id) ?? 'h5';
              }
            } catch {
              // leave sessions untagged on lookup failure
            }
          }
          const summaries = sessions.map(projectSessionSummary).filter((item) => matchesSessionQuery(item, query));
          const paged = summaries.slice(offset, offset + limit);
          res.json({
            sessions: paged,
            page: {
              total: summaries.length,
              offset,
              limit,
              has_more: offset + paged.length < summaries.length,
              query,
            },
          });
        } catch (err) {
          res.status(500).json({
            message: sanitizeUserFacingProxyMessage(
              err instanceof Error ? err.message : '读取会话失败',
              '读取会话失败',
            ),
          });
        }
      },
    ],
  };

  const sessionScoped = (build) => [
    requireUser,
    async (req, res, next) => {
      try {
        const body = req.body ?? (await readJsonBody(req));
        req.body = body;
        const sessionId = extractSessionId(req, body);
        if (!sessionId) {
          res.status(400).json({ message: '缺少 session_id' });
          return;
        }
        const owns = await sessionStore.validateOwnership(req.currentUser.id, sessionId);
        if (!owns) {
          res.status(403).json({ message: '无权访问该会话' });
          return;
        }
        req.agentSessionId = sessionId;
        await build(req, res, next);
      } catch (err) {
        res.status(500).json({
          message: sanitizeUserFacingProxyMessage(
            err instanceof Error ? err.message : '请求失败',
            '请求失败',
          ),
        });
      }
    },
  ];

  const proxySessionEvents = async (req, res, sessionId, { onAfterFinish, onEvent } = {}) => {
    const upstreamAbort = new AbortController();
    const streamRequestedAt = Date.now();
    let clientClosed = false;
    const replayEnabled = isSessionStreamReplayEnabled() && sessionStreamStore;
    const initialLastEventId = parseSessionStreamLastEventId(req.get('last-event-id'));
    const abortUpstream = () => {
      clientClosed = true;
      upstreamAbort.abort();
    };
    req.once('close', abortUpstream);

    const writeReplayToClient = async (events) => {
      for (const event of events) {
        if (clientClosed || res.writableEnded) return;
        await writeClientChunk(formatSessionStreamSseChunk({
          id: event.id,
          data: event.payload,
        }));
      }
    };

    let writeClientChunk = async () => {};
    let pendingBalance = null;
    const waitForDrain = () => new Promise((resolve) => res.once('drain', resolve));
    writeClientChunk = async (chunk) => {
      if (res.writableEnded || clientClosed) return;
      let needsDrain = !res.write(chunk);
      if (pendingBalance != null && !res.writableEnded && !clientClosed) {
        needsDrain = !res.write(appendBalanceEvent(pendingBalance)) || needsDrain;
        pendingBalance = null;
      }
      if (needsDrain && !res.writableEnded && !clientClosed) {
        await waitForDrain();
      }
    };

    try {
      if (replayEnabled && initialLastEventId) {
        res.status(200);
        res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
        res.setHeader('Cache-Control', 'no-cache, no-transform');
        res.setHeader('Connection', 'keep-alive');
        res.setHeader('X-Accel-Buffering', 'no');
        res.flushHeaders?.();

        let batch = await sessionStreamStore.listEventsForUser(req.currentUser.id, sessionId, {
          afterEventId: initialLastEventId,
        });
        if (!batch) {
          res.status(404).end();
          return;
        }
        if (batch.cursorMiss) {
          batch = await sessionStreamStore.listEventsForUser(req.currentUser.id, sessionId);
        }
        await writeReplayToClient(batch?.events ?? []);
        if (shouldSkipUpstreamAfterSessionReplay(batch?.events ?? [], { cursorMiss: batch?.cursorMiss })) {
          req.off('close', abortUpstream);
          if (!res.writableEnded) res.end();
          return;
        }
      }

      const pathname = `/sessions/${sessionId}/events`;
      const sessionTarget = await resolveTarget(sessionId);
      const upstream = await apiFetch(sessionTarget, apiSecret, pathname, {
        method: 'GET',
        signal: upstreamAbort.signal,
        headers: {
          Accept: 'text/event-stream',
          'Last-Event-ID': req.get('last-event-id') ?? '',
        },
      });
      if (clientClosed) return;

      if (!upstream.ok || !upstream.body) {
        const text = await upstream.text().catch(() => '');
        if (!res.headersSent) {
          res.status(upstream.status).send(text);
        } else if (!res.writableEnded) {
          res.end();
        }
        return;
      }

      if (!res.headersSent) {
        res.status(upstream.status);
        res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
        res.setHeader('Cache-Control', 'no-cache, no-transform');
        res.setHeader('Connection', 'keep-alive');
        res.setHeader('X-Accel-Buffering', 'no');
        res.flushHeaders?.();
      }

      const billingTransform = createSseBillingTransform({
        onFinish: async (event) => {
          const billingRequestId = event.request_id ?? event.chat_request_id ?? null;
          const result = await userAuth.billSessionUsage(
            req.currentUser.id,
            sessionId,
            event.token_state,
            billingRequestId,
          );
          if (result.ok && result.costCents > 0 && result.balanceCents != null) {
            pendingBalance = {
              balanceCents: result.balanceCents,
              tokensUsed: result.tokensUsed ?? undefined,
              lastUsage: {
                inputTokens: result.deltaInputTokens ?? 0,
                outputTokens: result.deltaOutputTokens ?? 0,
                costCents: result.costCents,
              },
            };
          }
          // Fire-and-forget snapshot refresh after conversation finishes.
          if (typeof onAfterFinish === 'function') {
            void Promise.resolve().then(() => onAfterFinish(sessionId, req.currentUser.id)).catch(() => {});
          }
        },
      });

      const source = Readable.fromWeb(upstream.body);
      let firstChunkSeen = false;
      const firstTokenProbe = new Transform({
        transform(chunk, _encoding, callback) {
          if (!firstChunkSeen && chunk?.length > 0) {
            firstChunkSeen = true;
            markFirstTokenObserved(sessionTarget, Date.now() - streamRequestedAt);
          }
          callback(null, chunk);
        },
      });
      const linkSanitizer = createSessionEventSanitizer(req.currentUser, {
        onEvent,
        normalizeReplayIds: replayEnabled,
        onPersistFrame: replayEnabled
          ? ({ id, upstreamEventId, payload }) => {
              void sessionStreamStore.appendEvent({
                userId: req.currentUser.id,
                sessionId,
                id,
                upstreamEventId,
                payload,
              }).catch(() => {});
            }
          : undefined,
      });
      const clientSink = new Writable({
        write(chunk, _encoding, callback) {
          writeClientChunk(chunk).then(() => callback(), callback);
        },
      });

      // 每 20s 发一个注释行保持连接，防止 nginx/代理因静默超时切断 SSE
      const keepalive = setInterval(() => {
        if (!res.writableEnded && !clientClosed) {
          const ok = res.write(': keepalive\n\n');
          if (!ok) source.pause();
        }
      }, 20000);
      res.on('drain', () => source.resume());
      await markStreamStarted(sessionId, sessionTarget);
      let streamCloseStatus = 'closed';
      try {
        try {
          await pipeline(source, firstTokenProbe, linkSanitizer, billingTransform, clientSink);
        } catch (err) {
          streamCloseStatus = clientClosed || upstreamAbort.signal.aborted ? 'aborted' : 'error';
          throw err;
        }
      } finally {
        clearInterval(keepalive);
        req.off('close', abortUpstream);
        await markStreamEnded(sessionId, sessionTarget, { status: streamCloseStatus });
        if (!res.writableEnded) res.end();
      }
    } catch (err) {
      req.off('close', abortUpstream);
      if (clientClosed || upstreamAbort.signal.aborted) {
        if (!res.writableEnded) res.end();
        return;
      }
      writeSseProxyErrorAndEnd(
        res,
        sanitizeUserFacingProxyMessage(
          err instanceof Error ? err.message : 'SSE 代理失败',
          'SSE 代理失败',
        ),
      );
    }
  };

  const proxyFallback = async (req, res) => {
    try {
      const pathname = req.originalUrl.replace(/^\/api/, '') || '/';
      if (isNativeH5ApiPath(pathname)) {
        res.status(404).json({
          message: 'H5 本地接口未找到，请确认服务端已更新并重启',
          code: 'not_found',
        });
        return;
      }
      const isAgentRunPath =
        (req.method === 'POST' && pathname === '/agent/runs')
        || (req.method === 'GET' && /^\/agent\/runs\/[^/]+$/.test(pathname))
        || (req.method === 'GET' && /^\/agent\/runs\/[^/]+\/events$/.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, {
        unrestricted: policyState.unrestricted,
      });
      if (!gate.allowed) {
        res.status(403).json({ message: gate.reason ?? '该 API 已被策略禁止' });
        return;
      }
      if (
        /^\/agent\/harness_(bootstrap|remember)$/.test(pathname) &&
        !capabilityState.unrestricted &&
        !capabilityState.capabilities.context_memory
      ) {
        res.status(403).json({ message: '当前账户未开通项目记忆，无法访问该 API' });
        return;
      }
      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) {
        const publishLayout = await userAuth
          .getUserPublishLayout(req.currentUser.id)
          .catch(() => null);
        baseBody = injectCurrentTimeAnchor(req.body, publishLayout?.timezone);
        baseBody = injectTaskRoutingHint(
          baseBody,
          await userAuth.getAgentSessionPolicy(req.currentUser.id),
        );
      }

      // Vision pre-processing: call Qwen VL directly for image analysis, then inject
      // the description + image paths into the message text for DeepSeek/Goose.
      // We do NOT switch the Goose session provider — DeepSeek retains full tool-calling
      // capability so it can write files and return real links.
      if (isReplyPath && llmProviderService) {
        const sessionId = sessionMatch?.[1];
        if (sessionId) {
          const hasImages = messageHasImages(req.body?.user_message);
          if (hasImages && await llmProviderService.hasVisionKey()) {
            const publishLayout = await userAuth
              .getUserPublishLayout(req.currentUser?.id)
              .catch(() => null);
            const visionResult = await buildVisionBody(
              baseBody?.user_message ?? req.body?.user_message,
              req.currentUser?.id,
              publishLayout,
            ).catch(() => null);
            if (visionResult?.userMessage) {
              baseBody = { ...(baseBody ?? req.body), user_message: visionResult.userMessage };
            }
            if (visionResult?.billableImageCount > 0 && subscriptionService) {
              await subscriptionService.consumeImageQuota(
                req.currentUser.id,
                visionResult.billableImageCount,
              ).catch((err) => {
                console.warn(
                  'Subscription image quota consume skipped:',
                  err instanceof Error ? err.message : err,
                );
              });
            }
          }
        }
      }

      const body =
        req.method === 'GET' || req.method === 'HEAD'
          ? undefined
          : baseBody
            ? JSON.stringify(baseBody)
            : undefined;

      const fallbackTarget =
        req.goosedTarget ??
        (sessionMatch ? await resolveTarget(sessionMatch[1]) : primaryTarget);
      const upstream = await apiFetch(fallbackTarget, apiSecret, pathname, {
        method: req.method,
        body,
        headers: {
          Accept: req.get('accept') ?? '*/*',
          'Last-Event-ID': req.get('last-event-id') ?? '',
        },
      });

      if (req.method === 'GET' && /^\/sessions\/[^/]+$/.test(pathname) && upstream.ok) {
        const payload = await upstream.json().catch(() => null);
        if (payload && Array.isArray(payload.conversation)) {
          const sanitizedConversation = sanitizeSessionConversationPublicHtmlLinks(
            payload.conversation,
            req.currentUser,
          );
          const { conversation, page } = paginateSessionConversation(
            sanitizedConversation,
            req.query?.history_before,
            req.query?.history_limit,
          );
          payload.conversation = conversation;
          payload.conversation_page = page;
        }
        res.status(upstream.status).json(payload ?? {});
        return;
      }

      sendProxyResponse(res, upstream);
    } catch (err) {
      res.status(502).json({
        message: sanitizeUserFacingProxyMessage(
          err instanceof Error ? err.message : '代理失败',
          '代理失败',
        ),
      });
    }
  };

  return {
    requireUser,
    ensureChatAllowed,
    applySessionLlmProvider,
    applyLocalFallbackForSession,
    applyVisionProviderForSession,
    reconcileSessionPolicyForUser,
    handlers,
    sessionScoped,
    proxyFallback,
    proxySessionEvents,
    resolveTarget,
    startSessionForUser,
    getRuntimeStatus,
    submitSessionReplyForUser,
    submitSessionReplyAndAwaitFinishForUser,
    apiFetch: async (pathname, init) => apiFetch(await pickTarget(), apiSecret, pathname, init),
    apiFetchTo: (target, pathname, init) => apiFetch(target, apiSecret, pathname, init),
  };
}

export function matchHandler(handlers, method, path) {
  const key = `${method.toUpperCase()} ${path}`;
  return handlers[key] ?? null;
}