import { Readable } from 'node:stream';
import { Agent, fetch as undiciFetch } from 'undici';
import { appendBalanceEvent, createSseBillingTransform } from './sse-billing.mjs';
import { developerToolsFromPolicy } from './capabilities.mjs';
import { evaluateProxyRequest, isNativeH5ApiPath } from './policies.mjs';
import { buildSandboxSessionConstraints } from './user-publish.mjs';
import { buildTaskRoutingAgentText } from './user-memory-profile.mjs';
import { reconcileAgentSession } from './session-reconcile.mjs';
import { createImgproxySigner } from './imgproxy-signer.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, '后端');
}

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,
  });
}

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 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.+)$/;

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 injectTaskRoutingHint(body, sessionPolicy) {
  const originalText = firstUserText(body?.user_message);
  if (!originalText?.trim()) return body;

  const agentText = buildTaskRoutingAgentText(originalText, sessionPolicy);
  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,
      },
    },
  };
}

export async function buildVisionPayload({
  userMessage,
  userId,
  publishLayout,
  localFetchAsset,
  llmProviderService,
  imgproxySigner = null,
}) {
  if (!localFetchAsset || !llmProviderService || !userId) return null;
  const rawImageUrls = extractImageUrlsFromMessage(userMessage);
  if (rawImageUrls.length === 0) return null;

  const imageItems = [];
  for (const rawUrl of rawImageUrls) {
    try {
      const match = rawUrl.match(/\/mindspace\/v1\/assets\/([^/?#]+)(?:\/download)?/);
      if (!match) continue;
      const assetId = decodeURIComponent(match[1]);

      let buffer;
      let mimeType = 'image/jpeg';

      if (imgproxySigner) {
        try {
          const visionUrl = imgproxySigner.buildUrl(
            process.env.IMGPROXY_BASE_URL || 'https://img.tkmind.cn',
            `users/${userId}/assets/${assetId}`,
            'vision'
          );
          const visionResponse = await undiciFetch(visionUrl);
          if (visionResponse.ok) {
            buffer = Buffer.from(await visionResponse.arrayBuffer());
            mimeType = visionResponse.headers.get('content-type') || 'image/jpeg';
          } else {
            throw new Error(`Vision fetch failed: ${visionResponse.status}`);
          }
        } catch (visionErr) {
          console.warn(`Vision fetch fallback for asset ${assetId}:`, visionErr instanceof Error ? visionErr.message : visionErr);
          const asset = await localFetchAsset(userId, assetId);
          buffer = asset.buffer;
          mimeType = asset.mimeType;
        }
      } else {
        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 */ }
      imageItems.push({ mimeType, data: buffer.toString('base64'), relativePath, rawUrl });
    } 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.relativePath}" 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`
      : '') +
    `图片 HTML 嵌入路径（直接写入 <img> 标签，浏览器有 cookie 可直接加载，无需 fetch）：\n${pathList}\n` +
    '执行要求：必须先调用 load_skill → static-page-publish（每次生成页面都要调用，不可省略），' +
    '再用 write_file 写入 public/页面.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.relativePath);
      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 },
    billableImageCount: visionDescription ? 1 : 0,
  };
}

export function createTkmindProxy({
  apiTarget,
  apiTargets,
  apiSecret,
  userAuth,
  llmProviderService,
  localFetchAsset,
  subscriptionService,
  sessionSnapshotService,
  conversationMemoryService,
}) {
  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;
    }
  }

  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) {
    if (!sessionSnapshotService?.isEnabled?.()) return;
    const needsSummary = sessions.filter(
      (session) => hasDefaultSessionName(session) || Number(session?.message_count ?? 0) === 0,
    );
    if (needsSummary.length === 0) return;
    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 the client-side label.
    }
  }

  async function pickTarget() {
    if (targets.length <= 1) return primaryTarget;
    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;
  }

  async function startSessionForUser(
    userId,
    {
      workingDir,
      sessionPolicy = null,
      recipe = null,
    } = {},
  ) {
    const startTarget = await pickTarget();
    const upstream = await apiFetch(startTarget, apiSecret, '/agent/start', {
      method: 'POST',
      body: JSON.stringify({
        ...(workingDir ? { working_dir: workingDir } : {}),
        enable_context_memory: sessionPolicy?.enableContextMemory,
        ...(sessionPolicy?.extensionOverrides
          ? { extension_overrides: sessionPolicy.extensionOverrides }
          : {}),
        ...(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 userAuth.registerAgentSession(userId, 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(() => '');
        throw new Error(modeText || `设置会话模式失败 (${modeRes.status})`);
      }
    }
    return session;
  }

  async function resolveTarget(sessionId) {
    if (targets.length <= 1 || !sessionId) return primaryTarget;
    try {
      const { target, node } = await userAuth.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;
      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 reconcileSessionPolicyForUser(userId, sessionId) {
    if (!userId || !sessionId) return;
    const target = await resolveTarget(sessionId);
    const workingDir = await userAuth.resolveWorkingDir(userId);
    const sessionPolicy = await userAuth.getAgentSessionPolicy(userId);
    const publishLayout = await userAuth.getUserPublishLayout(userId);
    const userMemories = conversationMemoryService?.listMemories
      ? await conversationMemoryService.listMemories(userId, { limit: 40 }).catch(() => [])
      : [];
    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,
      },
    );
  }

  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 userAuth.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 = conversationMemoryService?.listMemories
              ? await conversationMemoryService
                  .listMemories(req.currentUser.id, { limit: 40 })
                  .catch(() => [])
              : [];
            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 userAuth.ownsSession(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 = conversationMemoryService?.listMemories
              ? await conversationMemoryService
                  .listMemories(req.currentUser.id, { limit: 40 })
                  .catch(() => [])
              : [];
            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 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(3000),
              });
              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 sessions = [...sessionsById.values()];
          await enrichSessionHistory(sessions);
          res.json({ sessions });
        } 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 userAuth.ownsSession(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 } = {}) => {
    try {
      const pathname = `/sessions/${sessionId}/events`;
      const sessionTarget = await resolveTarget(sessionId);
      const upstream = await apiFetch(sessionTarget, apiSecret, pathname, {
        method: 'GET',
        headers: {
          Accept: 'text/event-stream',
          'Last-Event-ID': req.get('last-event-id') ?? '',
        },
      });

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

      res.status(upstream.status);
      res.setHeader('Content-Type', 'text/event-stream');
      res.setHeader('Cache-Control', 'no-cache');
      res.setHeader('Connection', 'keep-alive');

      let pendingBalance = null;
      const billingTransform = createSseBillingTransform({
        onFinish: async (event) => {
          const result = await userAuth.billSessionUsage(
            req.currentUser.id,
            sessionId,
            event.token_state,
            null,
          );
          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);

      // 每 20s 发一个注释行保持连接，防止 nginx/代理因静默超时切断 SSE
      const keepalive = setInterval(() => {
        if (!res.writableEnded) res.write(': keepalive\n\n');
      }, 20000);
      const stopKeepalive = () => clearInterval(keepalive);

      billingTransform.on('data', (chunk) => {
        res.write(chunk);
        if (pendingBalance != null) {
          res.write(appendBalanceEvent(pendingBalance));
          pendingBalance = null;
        }
      });
      billingTransform.on('end', () => { stopKeepalive(); res.end(); });
      billingTransform.on('error', () => { stopKeepalive(); res.end(); });
      source.on('error', () => { stopKeepalive(); res.end(); });
      req.on('close', stopKeepalive);
      source.pipe(billingTransform);
    } catch (err) {
      res.status(502).json({
        message: 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 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\/([^/]+)/);

      let baseBody = req.body;
      if (isReplyPath && !policyState.unrestricted) {
        baseBody = injectTaskRoutingHint(
          req.body,
          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') ?? '',
        },
      });

      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,
    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;
}