memind/agent-run-routes.mjs

import { normalizeAgentRunToolMode } from './agent-run-gateway.mjs';
import { createSessionAccess } from './session-broker.mjs';
import {
  extractRunFromStreamEvent,
  formatRunStreamSseChunk,
  isRunStreamReplayEnabled,
  isTerminalRunStatus,
  parseRunStreamLastEventId,
  shouldEmitRunUpdateForStreamEvent,
} from './agent-run-stream.mjs';
import { wrapRunStreamPayload, writeSseErrorAndEnd } from './sse-event-taxonomy.mjs';

function envFlag(value) {
  return ['1', 'true', 'yes', 'on'].includes(String(value ?? '').trim().toLowerCase());
}

function parseUserIdSet(value) {
  return new Set(
    String(value ?? '')
      .split(',')
      .map((item) => item.trim())
      .filter(Boolean),
  );
}

function parseTaskTypeSet(value) {
  return new Set(
    String(value ?? '')
      .split(',')
      .map((item) => item.trim().toLowerCase())
      .filter(Boolean),
  );
}

function hasExpectedFileValidation(userMessage) {
  const metadata = userMessage?.metadata;
  const runMetadata = metadata?.memindRun ?? metadata?.agentRun ?? {};
  const validation = runMetadata.validation ?? metadata?.toolGatewayValidation;
  if (!validation || typeof validation !== 'object' || Array.isArray(validation)) return false;
  const expectedFile = validation.expectedFile ?? validation.expectedPath;
  if (typeof expectedFile === 'string' && expectedFile.trim()) return true;
  if (expectedFile && typeof expectedFile === 'object' && !Array.isArray(expectedFile)) {
    const filePath = String(expectedFile.path ?? expectedFile.file ?? expectedFile.relativePath ?? '').trim();
    if (filePath) return true;
  }
  if (!Array.isArray(validation.expectedFiles)) return false;
  return validation.expectedFiles.some((item) => {
    if (typeof item === 'string') return Boolean(item.trim());
    if (!item || typeof item !== 'object' || Array.isArray(item)) return false;
    return Boolean(String(item.path ?? item.file ?? item.relativePath ?? '').trim());
  });
}

export function createPostAgentRunsHandler({
  userAuth,
  sessionAccess = null,
  agentRunGateway,
  mindSpaceAssetAgent = null,
  codeRunPolicyService = null,
  codeRunsEnabled = envFlag(process.env.MEMIND_AGENT_CODE_RUNS_ENABLED),
  codeRunUserIds = parseUserIdSet(process.env.MEMIND_AGENT_CODE_RUNS_USER_IDS),
  codeRunTaskTypes = parseTaskTypeSet(process.env.MEMIND_AGENT_CODE_RUN_TASK_TYPES),
  requireCodeRunValidation = envFlag(process.env.MEMIND_AGENT_CODE_RUNS_REQUIRE_VALIDATION),
}) {
  const sessionStore = sessionAccess ?? createSessionAccess({ userAuth, enabled: false });

  async function resolveCodeRunPolicy(userId) {
    if (codeRunPolicyService?.getEffectivePolicy) {
      return codeRunPolicyService.getEffectivePolicy(userId);
    }
    return {
      source: 'env',
      enabled: codeRunsEnabled,
      clientEnabled: codeRunsEnabled,
      generalAutodetect: false,
      pageDataDevAutodetect: false,
      requireValidation: requireCodeRunValidation,
      userAllowlist: [...codeRunUserIds],
      taskTypeAllowlist: [...codeRunTaskTypes],
      userAllowed:
        !codeRunUserIds.size || codeRunUserIds.has(String(userId ?? '').trim()),
    };
  }

  return async function postAgentRuns(request, response) {
    try {
      const sessionId = String(request.body?.session_id ?? '').trim() || null;
      const requestId = String(request.body?.request_id ?? '').trim();
      const userMessage = request.body?.user_message ?? null;
      const rawToolMode = request.body?.tool_mode ?? request.body?.toolMode ?? 'chat';
      const taskType = String(request.body?.task_type ?? request.body?.taskType ?? '').trim() || null;
      const forceDeepReasoning = request.body?.force_deep_reasoning === true || request.body?.forceDeepReasoning === true;
      if (!requestId) {
        response.status(400).json({ message: '缺少 request_id' });
        return;
      }
      if (!userMessage) {
        response.status(400).json({ message: '缺少 user_message' });
        return;
      }
      let toolMode = 'chat';
      try {
        toolMode = normalizeAgentRunToolMode(rawToolMode);
      } catch (err) {
        response.status(400).json({
          message: err instanceof Error ? err.message : '不支持的 tool_mode',
        });
        return;
      }
      if (toolMode === 'code') {
        const codeRunPolicy = await resolveCodeRunPolicy(request.currentUser.id);
        if (!codeRunPolicy.enabled) {
          response.status(403).json({ message: '代码任务灰度未开启' });
          return;
        }
        if (!codeRunPolicy.userAllowed) {
          response.status(403).json({ message: '当前用户未开启代码任务灰度' });
          return;
        }
        const taskTypeAllowlist = codeRunPolicy.taskTypeAllowlist ?? [];
        if (
          taskTypeAllowlist.length > 0 &&
          (!taskType || !taskTypeAllowlist.map((item) => String(item).toLowerCase()).includes(taskType.toLowerCase()))
        ) {
          response.status(403).json({ message: '当前代码任务类型未开启灰度' });
          return;
        }
        if (codeRunPolicy.requireValidation && !hasExpectedFileValidation(userMessage)) {
          response.status(400).json({ message: '代码任务必须声明产物校验规则' });
          return;
        }
      }
      if (sessionId) {
        const owns = await sessionStore.validateOwnership(request.currentUser.id, sessionId);
        if (!owns) {
          response.status(403).json({ message: '无权访问该会话' });
          return;
        }
      }
      const selectedAssetIds = [
        ...new Set(
          (Array.isArray(request.body?.selected_asset_ids)
            ? request.body.selected_asset_ids
            : Array.isArray(request.body?.selectedAssetIds)
              ? request.body.selectedAssetIds
              : []
          )
            .map((item) => String(item ?? '').trim())
            .filter(Boolean),
        ),
      ];
      if (
        selectedAssetIds.length > 0 &&
        mindSpaceAssetAgent?.materializeAssetsForSession
      ) {
        try {
          await mindSpaceAssetAgent.materializeAssetsForSession({
            sessionId,
            userId: request.currentUser.id,
            assetIds: selectedAssetIds,
          });
        } catch (err) {
          response.status(400).json({
            message: err instanceof Error ? err.message : '勾选资料落盘失败',
            code: err?.code ?? 'asset_materialize_failed',
          });
          return;
        }
      }
      const run = await agentRunGateway.createRun(request.currentUser.id, {
        sessionId,
        requestId,
        userMessage,
        toolMode,
        taskType,
        ...(forceDeepReasoning ? { forceDeepReasoning: true } : {}),
      });
      response.status(202).json({ run });
    } catch (err) {
      if (err?.code === 'SESSION_RUN_CONFLICT') {
        response.status(409).json({
          message: err.message,
          code: 'SESSION_RUN_CONFLICT',
        });
        return;
      }
      response.status(500).json({
        message: err instanceof Error ? err.message : '创建任务失败',
      });
    }
  };
}

export function createGetAgentRunHandler({ agentRunGateway }) {
  return async function getAgentRun(request, response) {
    try {
      const run = await agentRunGateway.getRunForUser(
        request.currentUser.id,
        request.params.runId,
      );
      if (!run) {
        response.status(404).json({ message: '任务不存在' });
        return;
      }
      if (run.status !== 'succeeded' && run.status !== 'failed') {
        agentRunGateway.dispatchRun(run.id);
      }
      response.json({ run });
    } catch (err) {
      response.status(500).json({
        message: err instanceof Error ? err.message : '读取任务失败',
      });
    }
  };
}

function createRunStreamWriter(response, { wrapPayload = true } = {}) {
  let lastSentEventId = null;
  let lastRunPayload = null;

  const writeChunk = (eventId, eventName, data) => {
    const payload = wrapPayload ? wrapRunStreamPayload(eventName, data) : data;
    response.write(formatRunStreamSseChunk({ id: eventId, event: eventName, data: payload }));
    if (eventId) lastSentEventId = eventId;
  };

  const emitRunIfChanged = (run, eventId = null) => {
    const nextPayload = JSON.stringify(run);
    if (nextPayload === lastRunPayload) return false;
    lastRunPayload = nextPayload;
    writeChunk(eventId, 'run', { run });
    return true;
  };

  return {
    getLastSentEventId: () => lastSentEventId,
    setLastSentEventId: (value) => {
      lastSentEventId = value;
    },
    emitRunIfChanged,
    writeError: (message) => {
      writeChunk(null, 'error', { message });
    },
  };
}

export function createAgentRunEventsHandler({
  agentRunGateway,
  pollIntervalMs = 1000,
  keepaliveIntervalMs = 20000,
  replayEnabled = isRunStreamReplayEnabled(),
}) {
  return async function getAgentRunEvents(request, response) {
    const userId = request.currentUser.id;
    const runId = request.params.runId;
    let closed = false;
    let pollTimer = null;
    let keepaliveTimer = null;
    const initialLastEventId = parseRunStreamLastEventId(request.get?.('last-event-id'));

    const cleanup = () => {
      closed = true;
      if (pollTimer) clearInterval(pollTimer);
      if (keepaliveTimer) clearInterval(keepaliveTimer);
      pollTimer = null;
      keepaliveTimer = null;
    };

    const firstRun = await agentRunGateway.getRunForUser(userId, runId);
    if (!firstRun) {
      response.status(404).json({ message: '任务不存在' });
      return;
    }

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

    request.on('close', cleanup);
    keepaliveTimer = setInterval(() => {
      if (!closed) response.write(': keepalive\n\n');
    }, keepaliveIntervalMs);

    const stream = createRunStreamWriter(response);

    const publishLegacy = async (prefetchedRun = null) => {
      if (closed) return;
      try {
        const run = prefetchedRun ?? await agentRunGateway.getRunForUser(userId, runId);
        if (!run) {
          stream.writeError('任务不存在');
          cleanup();
          response.end();
          return;
        }
        stream.emitRunIfChanged(run);
        if (!isTerminalRunStatus(run.status)) {
          agentRunGateway.dispatchRun(run.id);
          return;
        }
        cleanup();
        response.end();
      } catch (err) {
        if (response.headersSent) {
          writeSseErrorAndEnd(response, err instanceof Error ? err.message : '读取任务失败');
        } else {
          stream.writeError(err instanceof Error ? err.message : '读取任务失败');
          response.end();
        }
        cleanup();
      }
    };

    const publishReplay = async () => {
      if (closed) return;
      try {
        const cursor = stream.getLastSentEventId() ?? initialLastEventId;
        const batch = await agentRunGateway.listRunEventsForUser(userId, runId, {
          afterEventId: cursor,
        });
        if (!batch?.run) {
          stream.writeError('任务不存在');
          cleanup();
          response.end();
          return;
        }

        let latestRun = batch.run;
        for (const event of batch.events) {
          if (closed) return;
          const snapshotRun = extractRunFromStreamEvent(event);
          if (snapshotRun) {
            latestRun = snapshotRun;
            stream.emitRunIfChanged(snapshotRun, event.id);
            continue;
          }
          if (shouldEmitRunUpdateForStreamEvent(event.eventType)) {
            latestRun = await agentRunGateway.getRunForUser(userId, runId) ?? latestRun;
            stream.emitRunIfChanged(latestRun, event.id);
            continue;
          }
          response.write(formatRunStreamSseChunk({
            id: event.id,
            event: 'run_event',
            data: wrapRunStreamPayload('run_event', {
              eventId: event.id,
              eventType: event.eventType,
              data: event.data,
              createdAt: event.createdAt,
            }),
          }));
          stream.setLastSentEventId(event.id);
        }

        if (batch.cursorMiss && !cursor) {
          stream.emitRunIfChanged(batch.run);
        } else if (batch.cursorMiss && cursor) {
          latestRun = await agentRunGateway.getRunForUser(userId, runId) ?? latestRun;
          stream.emitRunIfChanged(latestRun);
        } else {
          stream.emitRunIfChanged(latestRun);
        }

        if (!isTerminalRunStatus(latestRun.status)) {
          agentRunGateway.dispatchRun(runId);
          return;
        }
        cleanup();
        response.end();
      } catch (err) {
        if (response.headersSent) {
          writeSseErrorAndEnd(response, err instanceof Error ? err.message : '读取任务失败');
        } else {
          stream.writeError(err instanceof Error ? err.message : '读取任务失败');
          response.end();
        }
        cleanup();
      }
    };

    const publish = replayEnabled && agentRunGateway.listRunEventsForUser
      ? publishReplay
      : publishLegacy;

    pollTimer = setInterval(() => {
      void publish();
    }, pollIntervalMs);
    if (replayEnabled && agentRunGateway.listRunEventsForUser) {
      await publishReplay();
    } else {
      await publishLegacy(firstRun);
    }
  };
}