memind/agent-run-gateway.mjs

import crypto from 'node:crypto';
import fs from 'node:fs/promises';
import path from 'node:path';
import { isRunStreamReplayEnabled } from './agent-run-stream.mjs';
import { isDirectChatSessionId } from './direct-chat-service.mjs';
import { CHAT_INTENT_ROUTE, resolveGatewayAgentSessionId, resolveLegacyRouteFromClassification, logRouterDecisionShadow } from './chat-intent-router.mjs';
import { resolveSessionAccess } from './session-broker.mjs';
import {
  loadSnapshotMessages,
  persistSessionTranscriptFromSnapshot,
  persistSessionTranscriptMessages,
} from './conversation-transcript-persist.mjs';
import { ensureGooseUserMessageMetadata } from './goose-message.mjs';
import {
  SESSION_FINISHED_STALE_GRACE_MS,
  tryRecoverRunFromDeliverables,
} from './agent-run-deliverable-check.mjs';

const DEFAULT_RUN_RETRY_DELAYS_MS = [1500, 5000, 15000];
const TERMINAL_STATUSES = new Set(['succeeded', 'failed']);
const CODE_TOOL_MODES = new Set(['code', 'code-task', 'code_task', 'code-tool', 'code_tool', 'code_tool_task']);
const RUN_METADATA_KEY = 'memindRun';
const DEFAULT_MAX_CONCURRENT_RUNS = 1;
const DEFAULT_RUN_TIMEOUT_MS = 15 * 60 * 1000;
const DEFAULT_RUN_HEARTBEAT_MS = 30 * 1000;
const TOOL_GATEWAY_SUMMARY_LIMIT = 4096;

function envFlag(value, fallback = false) {
  const raw = String(value ?? '').trim().toLowerCase();
  if (!raw) return fallback;
  return ['1', 'true', 'yes', 'on'].includes(raw);
}

function nowMs() {
  return Date.now();
}

function safeJsonParse(value, fallback = null) {
  try {
    return JSON.parse(value);
  } catch {
    return fallback;
  }
}

function parseDbJsonColumn(value, fallback = null) {
  if (value == null || value === '') return fallback;
  if (typeof value === 'object') return value;
  return safeJsonParse(value, fallback);
}

function serializeMessage(message) {
  return JSON.stringify(message ?? {});
}

function positiveInteger(value, fallback) {
  const n = Number(value);
  if (!Number.isFinite(n) || n <= 0) return fallback;
  return Math.floor(n);
}

function timeoutError(ms) {
  const err = new Error(`agent run timed out after ${ms}ms`);
  err.code = 'AGENT_RUN_TIMEOUT';
  return err;
}

export function normalizeAgentRunToolMode(value) {
  const normalized = String(value ?? 'chat').trim().toLowerCase();
  if (!normalized || normalized === 'chat') return 'chat';
  if (CODE_TOOL_MODES.has(normalized)) return 'code';
  throw new Error(`不支持的 tool_mode: ${value}`);
}

function normalizeTaskType(value) {
  const normalized = String(value ?? '').trim();
  return normalized || null;
}

function summarizeText(value, limit = TOOL_GATEWAY_SUMMARY_LIMIT) {
  const text = String(value ?? '');
  if (text.length <= limit) return text;
  return text.slice(text.length - limit);
}

function normalizeExpectedFileCheck(value) {
  if (typeof value === 'string') {
    const expectedPath = value.trim();
    return expectedPath ? { path: expectedPath } : null;
  }
  if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
  const expectedPath = String(value.path ?? value.file ?? value.relativePath ?? '').trim();
  if (!expectedPath) return null;
  const contains = value.contains ?? value.expectedContent ?? value.contentIncludes;
  return {
    path: expectedPath,
    ...(contains == null ? {} : { contains: String(contains) }),
  };
}

function normalizeToolGatewayValidation(value) {
  if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
  const checks = [];
  const single = normalizeExpectedFileCheck(value.expectedFile ?? value.expectedPath);
  if (single) checks.push(single);
  if (Array.isArray(value.expectedFiles)) {
    for (const item of value.expectedFiles) {
      const check = normalizeExpectedFileCheck(item);
      if (check) checks.push(check);
    }
  }
  return checks.length > 0 ? { expectedFiles: checks } : null;
}

function resolveValidationPath(cwd, expectedPath) {
  if (!String(cwd ?? '').trim()) {
    const err = new Error('Tool Gateway validation failed: missing working directory');
    err.code = 'TOOL_GATEWAY_VALIDATION_FAILED';
    err.retryable = false;
    throw err;
  }
  const root = path.resolve(String(cwd ?? ''));
  const target = path.resolve(root, String(expectedPath ?? ''));
  if (target !== root && !target.startsWith(`${root}${path.sep}`)) {
    const err = new Error(`Tool Gateway validation path escapes working directory: ${expectedPath}`);
    err.code = 'TOOL_GATEWAY_VALIDATION_FAILED';
    err.retryable = false;
    throw err;
  }
  return { root, target };
}

async function validateToolGatewayResult({ result, validation, cwd }) {
  if (!validation?.expectedFiles?.length || result?.dryRun) {
    return null;
  }
  const checks = [];
  for (const expected of validation.expectedFiles) {
    const { root, target } = resolveValidationPath(cwd ?? result?.cwd, expected.path);
    let content = '';
    let stat = null;
    try {
      stat = await fs.stat(target);
      content = await fs.readFile(target, 'utf8');
    } catch (cause) {
      const err = new Error(`Tool Gateway validation failed: expected file not found: ${expected.path}`);
      err.code = 'TOOL_GATEWAY_VALIDATION_FAILED';
      err.retryable = false;
      err.validation = {
        expectedFile: expected.path,
        cwd: root,
        reason: 'missing_file',
      };
      err.cause = cause;
      throw err;
    }
    if (expected.contains != null && !content.includes(expected.contains)) {
      const err = new Error(`Tool Gateway validation failed: expected content not found in ${expected.path}`);
      err.code = 'TOOL_GATEWAY_VALIDATION_FAILED';
      err.retryable = false;
      err.validation = {
        expectedFile: expected.path,
        cwd: root,
        reason: 'missing_content',
      };
      throw err;
    }
    checks.push({
      path: expected.path,
      sizeBytes: Number(stat?.size ?? 0),
      contains: expected.contains == null ? null : true,
    });
  }
  return { expectedFiles: checks };
}

function withRunMetadata(userMessage, { toolMode = 'chat', taskType = null, forceDeepReasoning = false } = {}) {
  const message = (userMessage && typeof userMessage === 'object' && !Array.isArray(userMessage))
    ? { ...userMessage }
    : { value: userMessage };
  const metadata = (message.metadata && typeof message.metadata === 'object' && !Array.isArray(message.metadata))
    ? { ...message.metadata }
    : {};
  metadata[RUN_METADATA_KEY] = {
    ...(metadata[RUN_METADATA_KEY] && typeof metadata[RUN_METADATA_KEY] === 'object'
      ? metadata[RUN_METADATA_KEY]
      : {}),
    toolMode: normalizeAgentRunToolMode(toolMode),
    ...(taskType ? { taskType } : {}),
    ...(forceDeepReasoning ? { forceDeepReasoning: true } : {}),
  };
  return { ...message, metadata };
}

function normalizeSessionMessageCount(value) {
  if (value == null || value === '') return null;
  const parsed = Number(value);
  if (!Number.isFinite(parsed) || parsed < 0) return null;
  return Math.floor(parsed);
}

function getRunOptionsFromMessage(userMessage) {
  const metadata = userMessage?.metadata;
  const runMetadata = metadata?.[RUN_METADATA_KEY] ?? metadata?.agentRun ?? {};
  let toolMode = 'chat';
  try {
    toolMode = normalizeAgentRunToolMode(runMetadata?.toolMode ?? metadata?.toolMode ?? 'chat');
  } catch {
    toolMode = 'chat';
  }
  return {
    toolMode,
    taskType: normalizeTaskType(runMetadata?.taskType ?? metadata?.taskType),
    forceDeepReasoning: runMetadata?.forceDeepReasoning === true || metadata?.forceDeepReasoning === true,
    validation: normalizeToolGatewayValidation(runMetadata?.validation ?? metadata?.toolGatewayValidation),
    sessionMessageCount: normalizeSessionMessageCount(
      runMetadata?.sessionMessageCount ?? metadata?.sessionMessageCount,
    ),
  };
}

function projectRun(row) {
  if (!row) return null;
  return {
    id: row.id,
    userId: row.user_id,
    sessionId: row.agent_session_id ?? null,
    requestId: row.request_id,
    status: row.status,
    attempts: Number(row.attempts ?? 0),
    error: row.error_message ?? null,
    createdAt: Number(row.created_at ?? 0),
    updatedAt: Number(row.updated_at ?? 0),
    startedAt: row.started_at == null ? null : Number(row.started_at),
    completedAt: row.completed_at == null ? null : Number(row.completed_at),
  };
}

export function createAgentRunGateway({
  pool,
  userAuth,
  sessionAccess = null,
  tkmindProxy,
  toolGateway = null,
  directChatService = null,
  chatIntentRouter = null,
  sessionSnapshotService = null,
  conversationMemoryService = null,
  syncUserPagesOnSuccess = null,
  retryDelaysMs = DEFAULT_RUN_RETRY_DELAYS_MS,
  autoDispatch = envFlag(process.env.MEMIND_AGENT_RUN_AUTODISPATCH, true),
  maxConcurrentRuns = positiveInteger(
    process.env.MEMIND_AGENT_RUN_QUEUE_CONCURRENCY,
    DEFAULT_MAX_CONCURRENT_RUNS,
  ),
  runTimeoutMs = positiveInteger(
    process.env.MEMIND_AGENT_RUN_TIMEOUT_MS,
    DEFAULT_RUN_TIMEOUT_MS,
  ),
  heartbeatMs = positiveInteger(
    process.env.MEMIND_AGENT_RUN_HEARTBEAT_MS,
    DEFAULT_RUN_HEARTBEAT_MS,
  ),
}) {
  const sessionStore = resolveSessionAccess({ userAuth, sessionAccess });
  const inFlight = new Set();
  const queuedDispatches = [];
  const queuedDispatchSet = new Set();

  function enqueueRun(runId) {
    if (!runId || inFlight.has(runId) || queuedDispatchSet.has(runId)) return false;
    queuedDispatchSet.add(runId);
    queuedDispatches.push(runId);
    return true;
  }

  function nextQueuedRunId() {
    const runId = queuedDispatches.shift();
    if (runId) queuedDispatchSet.delete(runId);
    return runId ?? null;
  }

  async function appendEvent(runId, type, data = null) {
    await pool.query(
      `INSERT INTO h5_agent_run_events (id, run_id, event_type, data_json, created_at)
       VALUES (?, ?, ?, ?, ?)`,
      [
        crypto.randomUUID(),
        runId,
        type,
        data == null ? null : JSON.stringify(data),
        nowMs(),
      ],
    );
  }

  async function appendRunSnapshot(runId) {
    if (!isRunStreamReplayEnabled()) return;
    const row = await getRunById(runId);
    if (!row) return;
    await appendEvent(runId, 'run_snapshot', { run: projectRun(row) });
  }

  async function getRunById(runId) {
    const [rows] = await pool.query(
      `SELECT * FROM h5_agent_runs WHERE id = ? LIMIT 1`,
      [runId],
    );
    return rows[0] ?? null;
  }

  async function getRunForUser(userId, runId) {
    const [rows] = await pool.query(
      `SELECT * FROM h5_agent_runs WHERE id = ? AND user_id = ? LIMIT 1`,
      [runId, userId],
    );
    return projectRun(rows[0] ?? null);
  }

  async function getRunByRequest(userId, requestId) {
    const [rows] = await pool.query(
      `SELECT * FROM h5_agent_runs WHERE user_id = ? AND request_id = ? LIMIT 1`,
      [userId, requestId],
    );
    return rows[0] ?? null;
  }

  async function createRun(userId, {
    sessionId = null,
    requestId,
    userMessage,
    toolMode = 'chat',
    taskType = null,
    forceDeepReasoning = false,
  }) {
    const normalizedRequestId = String(requestId ?? '').trim();
    if (!normalizedRequestId) {
      throw new Error('缺少 request_id');
    }
    const normalizedToolMode = normalizeAgentRunToolMode(toolMode);
    const normalizedTaskType = normalizeTaskType(taskType);
    const existing = await getRunByRequest(userId, normalizedRequestId);
    if (existing) {
      if (autoDispatch && !TERMINAL_STATUSES.has(existing.status)) dispatchRun(existing.id);
      return projectRun(existing);
    }

    // Prevent multiple concurrent agent runs on the same Goose session, which would
    // cause replies to arrive out of order and appear garbled in the chat UI.
    if (sessionId && !isDirectChatSessionId(sessionId)) {
      const [activeRows] = await pool.query(
        `SELECT id FROM h5_agent_runs
         WHERE agent_session_id = ? AND status NOT IN ('succeeded', 'failed')
         LIMIT 1`,
        [sessionId],
      );
      if (activeRows.length > 0) {
        const conflict = new Error('该会话有正在处理的任务，请等待完成后再发送');
        conflict.code = 'SESSION_RUN_CONFLICT';
        conflict.status = 409;
        throw conflict;
      }
    }

    const runId = crypto.randomUUID();
    const createdAt = nowMs();
    const runMessage = withRunMetadata(userMessage, {
      toolMode: normalizedToolMode,
      taskType: normalizedTaskType,
      forceDeepReasoning,
    });
    await pool.query(
      `INSERT INTO h5_agent_runs
         (id, user_id, agent_session_id, request_id, status, attempts,
          user_message_json, error_message, created_at, updated_at, started_at, completed_at)
       VALUES (?, ?, ?, ?, 'queued', 0, ?, NULL, ?, ?, NULL, NULL)`,
      [
        runId,
        userId,
        sessionId || null,
        normalizedRequestId,
        serializeMessage(runMessage),
        createdAt,
        createdAt,
      ],
    );
    await appendEvent(runId, 'queued', {
      sessionId: sessionId || null,
      toolMode: normalizedToolMode,
      taskType: normalizedTaskType,
      forceDeepReasoning: Boolean(forceDeepReasoning),
    });
    if (autoDispatch) dispatchRun(runId);
    return projectRun(await getRunById(runId));
  }

  async function prepareRunDeliverables({ userId, sessionId, runId, runStartedAtMs = null }) {
    if (typeof syncUserPagesOnSuccess !== 'function') return;
    await syncUserPagesOnSuccess({
      userId,
      sessionId,
      runId,
      runStartedAtMs,
    }).catch((err) => {
      console.warn(
        '[AgentRun] prepare deliverables sync failed:',
        err instanceof Error ? err.message : err,
      );
    });
  }

  async function recoverRunFromDeliverables({
    runId,
    userId,
    sessionId,
    err,
    runStartedAtMs = null,
    requireRecoverableError = true,
  }) {
    const recovered = await tryRecoverRunFromDeliverables({
      pool,
      userId,
      sessionId,
      error: err,
      requireRecoverableError,
      runStartedAtMs,
      prepareDeliverables: ({ userId: uid, sessionId: sid }) =>
        prepareRunDeliverables({ userId: uid, sessionId: sid, runId, runStartedAtMs }),
    });
    if (!recovered) return false;
    await appendEvent(runId, 'run_recovered_from_deliverables', {
      sessionId,
      deliverables: recovered.deliverables,
      originalError: recovered.originalError,
    });
    return true;
  }

  async function markRun(runId, status, fields = {}) {
    const updates = ['status = ?', 'updated_at = ?'];
    const values = [status, nowMs()];
    for (const [key, value] of Object.entries(fields)) {
      updates.push(`${key} = ?`);
      values.push(value);
    }
    values.push(runId);
    await pool.query(
      `UPDATE h5_agent_runs SET ${updates.join(', ')} WHERE id = ?`,
      values,
    );
    await appendEvent(runId, status, fields);
    await appendRunSnapshot(runId);
  }

  function startRunHeartbeat(runId, { attempt }) {
    let stopped = false;
    const writeHeartbeat = async () => {
      if (stopped) return;
      try {
        await appendEvent(runId, 'worker_heartbeat', {
          attempt,
          pid: process.pid,
          heartbeatMs,
        });
      } catch (err) {
        console.error('[AgentRun] worker heartbeat failed:', err instanceof Error ? err.message : err);
      }
    };
    void writeHeartbeat();
    const timer = setInterval(() => {
      void writeHeartbeat();
    }, heartbeatMs);
    timer.unref?.();
    return () => {
      stopped = true;
      clearInterval(timer);
    };
  }

  async function runWithTimeout(runId, task) {
    let timer = null;
    const timeout = new Promise((_, reject) => {
      timer = setTimeout(() => reject(timeoutError(runTimeoutMs)), runTimeoutMs);
    });
    try {
      return await Promise.race([task(), timeout]);
    } finally {
      if (timer) clearTimeout(timer);
    }
  }

  async function resolveGrantedSkills(userId) {
    if (!userId) return [];
    if (userAuth?.getUserSkills) {
      const skills = await userAuth.getUserSkills(userId).catch(() => null);
      if (Array.isArray(skills?.grantedSkills)) return skills.grantedSkills;
    }
    if (!userAuth?.getUserCapabilities) return [];
    const caps = await userAuth.getUserCapabilities(userId).catch(() => null);
    return Array.isArray(caps?.grantedSkills) ? caps.grantedSkills : [];
  }

  async function invalidatePortalDirectChatSnapshot(sessionId) {
    if (!sessionId || isDirectChatSessionId(sessionId) || !sessionSnapshotService?.remove) return;
    await sessionSnapshotService.remove(sessionId).catch(() => {});
  }

  async function assertOwnedAgentSession(userId, sessionId) {
    if (!sessionStore.enabled || !sessionId || isDirectChatSessionId(sessionId)) return;
    const owns = await sessionStore.validateOwnership(userId, sessionId);
    if (!owns) {
      const err = new Error('无权访问该会话');
      err.code = 'SESSION_FORBIDDEN';
      err.status = 403;
      throw err;
    }
  }

  async function resolveRunRouting(row, userMessage, runOptions) {
    if (!chatIntentRouter?.classify) return null;
    const enabled = chatIntentRouter.isEnabled
      ? await Promise.resolve(chatIntentRouter.isEnabled()).catch(() => false)
      : true;
    if (!enabled) return null;
    const grantedSkills = await resolveGrantedSkills(row.user_id);
    return chatIntentRouter.classify({
      userId: row.user_id,
      userMessage,
      sessionId: row.agent_session_id ?? null,
      sessionMessageCount: runOptions.sessionMessageCount,
      toolMode: runOptions.toolMode,
      forceDeepReasoning: runOptions.forceDeepReasoning,
      grantedSkills,
    });
  }

  async function executeRun(row, runId) {
    let userMessage = safeJsonParse(row.user_message_json, {});
    const runOptions = getRunOptionsFromMessage(userMessage);
    const toolGatewayStatus = toolGateway?.getStatus ? toolGateway.getStatus() : null;
    const routing = await resolveRunRouting(row, userMessage, runOptions);
    const routingDecision = resolveLegacyRouteFromClassification(routing) ?? routing?.route ?? null;
    if (routing) {
      logRouterDecisionShadow(routing, {
        requestId: row.request_id ?? null,
        sessionId: row.agent_session_id ?? null,
      });
      await appendEvent(runId, 'intent_routed', routing);
      if (routingDecision === CHAT_INTENT_ROUTE.AGENT && chatIntentRouter?.applyAgentOrchestration) {
        const grantedSkills = await resolveGrantedSkills(row.user_id);
        userMessage = chatIntentRouter.applyAgentOrchestration(userMessage, routing, { grantedSkills });
      }
    }
    const preferDirectChat =
      routingDecision === CHAT_INTENT_ROUTE.DIRECT_CHAT ||
      (isDirectChatSessionId(row.agent_session_id ?? null) && !runOptions.forceDeepReasoning);
    const directChatInput = {
      sessionId: row.agent_session_id ?? null,
      toolMode: runOptions.toolMode,
      userMessage,
      routingDecision,
    };
    const canDirectChat = Boolean(
      preferDirectChat && directChatService?.canHandle?.(directChatInput),
    );
    if (preferDirectChat && !canDirectChat) {
      const rejection = directChatService?.explainCanHandle?.(directChatInput);
      await appendEvent(runId, 'direct_chat_skipped', {
        sessionId: row.agent_session_id ?? null,
        routingDecision,
        reason: rejection?.reason ?? 'can_handle_false',
      });
    }
    if (canDirectChat) {
      try {
        const result = await directChatService.run({
          userId: row.user_id,
          sessionId: row.agent_session_id ?? null,
          requestId: row.request_id,
          userMessage,
          routingDecision,
          routingMemory: routing?.memory ?? null,
          onSessionReady: async (activeSessionId) => {
            await pool.query(
              `UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
              [activeSessionId, nowMs(), runId],
            );
            await appendRunSnapshot(runId);
          },
        });
        await appendEvent(runId, 'direct_chat_completed', {
          sessionId: result.sessionId,
          providerId: result.providerId ?? null,
          model: result.model ?? null,
          billed: Boolean(result.billing?.ok),
        });
        await appendRunSnapshot(runId);
        return { sessionId: result.sessionId };
      } catch (err) {
        await appendEvent(runId, 'direct_chat_failed', {
          code: err?.code ?? null,
          message: err instanceof Error ? err.message : String(err),
        });
        if (isDirectChatSessionId(row.agent_session_id)) {
          throw err;
        }
      }
    }
    if (runOptions.toolMode === 'code' && toolGatewayStatus?.enabled) {
      const workingDir = userAuth?.resolveWorkingDir
        ? await userAuth.resolveWorkingDir(row.user_id)
        : undefined;
      await invalidatePortalDirectChatSnapshot(row.agent_session_id ?? null);
      await appendEvent(runId, 'tool_gateway_dispatch', {
        protocol: toolGatewayStatus.protocol ?? 'agent-run-v1',
        taskType: runOptions.taskType,
        workingDir: workingDir ?? null,
      });
      const result = await toolGateway.executeJob({
        runId,
        userId: row.user_id,
        requestId: row.request_id,
        userMessage,
        taskType: runOptions.taskType,
        cwd: workingDir,
        timeoutMs: runTimeoutMs,
      });
      await appendEvent(runId, 'tool_gateway_result', {
        executor: result.executor ?? null,
        dryRun: Boolean(result.dryRun),
        exitCode: result.exitCode ?? null,
        stdoutTail: summarizeText(result.stdout),
        stderrTail: summarizeText(result.stderr),
      });
      try {
        const validation = await validateToolGatewayResult({
          result,
          validation: runOptions.validation,
          cwd: workingDir,
        });
        if (validation) {
          await appendEvent(runId, 'tool_gateway_validation', validation);
        }
      } catch (err) {
        await appendEvent(runId, 'tool_gateway_validation_failed', {
          code: err?.code ?? null,
          message: err instanceof Error ? err.message : String(err),
          validation: err?.validation ?? null,
        });
        throw err;
      }
      return { sessionId: row.agent_session_id ?? null };
    }

    let sessionId = resolveGatewayAgentSessionId({
      agentSessionId: row.agent_session_id ?? null,
      classification: routing,
      forceDeepReasoning: runOptions.forceDeepReasoning,
    });
    let escalatedDirectSessionId = null;
    if (isDirectChatSessionId(sessionId)) {
      escalatedDirectSessionId = sessionId;
      await appendEvent(runId, 'direct_session_escalated_to_deep_reasoning', {
        previousSessionId: sessionId,
      });
      sessionId = null;
    }
    if (!sessionId) {
      const sessionOptions = {};
      if (runOptions.toolMode === 'code' && userAuth?.getCodeAgentSessionPolicy) {
        sessionOptions.sessionPolicy = await userAuth.getCodeAgentSessionPolicy(row.user_id);
      }
      const session = await tkmindProxy.startSessionForUser(row.user_id, sessionOptions);
      sessionId = session.id;
      await pool.query(
        `UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
        [sessionId, nowMs(), runId],
      );
      await appendEvent(runId, 'session_started', {
        sessionId,
        toolMode: runOptions.toolMode,
        taskType: runOptions.taskType,
      });
      await appendRunSnapshot(runId);
      if (escalatedDirectSessionId) {
        const priorMessages = await loadSnapshotMessages(
          sessionSnapshotService,
          escalatedDirectSessionId,
        );
        const persisted = await persistSessionTranscriptMessages({
          conversationMemoryService,
          sessionId,
          userId: row.user_id,
          messages: priorMessages,
        });
        if (persisted.saved > 0) {
          await appendEvent(runId, 'direct_session_transcript_persisted', {
            previousSessionId: escalatedDirectSessionId,
            sessionId,
            saved: persisted.saved,
          });
        }
      }
    } else {
      await assertOwnedAgentSession(row.user_id, sessionId);
    }

    const transcriptPersisted = await persistSessionTranscriptFromSnapshot({
      sessionSnapshotService,
      conversationMemoryService,
      sessionId,
      userId: row.user_id,
    });
    if (transcriptPersisted.saved > 0) {
      await appendEvent(runId, 'portal_direct_transcript_persisted', {
        sessionId,
        saved: transcriptPersisted.saved,
      });
    }
    await invalidatePortalDirectChatSnapshot(sessionId);
    const awaitSessionFinish = envFlag(process.env.MEMIND_AGENT_RUN_AWAIT_SESSION_FINISH, true)
      && runOptions.toolMode === 'chat'
      && typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser === 'function';
    if (awaitSessionFinish) {
      try {
        const finish = await tkmindProxy.submitSessionReplyAndAwaitFinishForUser(
          row.user_id,
          sessionId,
          row.request_id,
          ensureGooseUserMessageMetadata(userMessage),
          {
            toolMode: runOptions.toolMode,
            forceDeepReasoning: runOptions.forceDeepReasoning,
            timeoutMs: runTimeoutMs,
          },
        );
        await appendEvent(runId, 'session_finished', {
          sessionId,
          tokenState: finish.tokenState ?? null,
        });
      } catch (err) {
        if (await recoverRunFromDeliverables({
          runId,
          userId: row.user_id,
          sessionId,
          err,
        })) {
          return { sessionId };
        }
        throw err;
      }
    } else {
      await tkmindProxy.submitSessionReplyForUser(
        row.user_id,
        sessionId,
        row.request_id,
        ensureGooseUserMessageMetadata(userMessage),
        {
          toolMode: runOptions.toolMode,
          forceDeepReasoning: runOptions.forceDeepReasoning,
        },
      );
    }
    return { sessionId };
  }

  async function finalizeSuccessfulRun(runId, row, sessionId) {
    await markRun(runId, 'succeeded', {
      agent_session_id: sessionId,
      completed_at: nowMs(),
      error_message: null,
    });
    if (typeof syncUserPagesOnSuccess === 'function') {
      await syncUserPagesOnSuccess({
        userId: row.user_id,
        sessionId,
        runId,
      }).catch((err) => {
        console.warn(
          '[AgentRun] workspace page deliver failed:',
          err instanceof Error ? err.message : err,
        );
      });
    }
  }

  async function processRun(runId) {
    const row = await getRunById(runId);
    if (!row || TERMINAL_STATUSES.has(row.status)) return;
    const nextAttempt = Number(row.attempts ?? 0) + 1;
    const [claim] = await pool.query(
      `UPDATE h5_agent_runs
       SET status = 'running', attempts = ?, started_at = COALESCE(started_at, ?), updated_at = ?, error_message = NULL
       WHERE id = ? AND status IN ('queued', 'retryable')`,
      [nextAttempt, nowMs(), nowMs(), runId],
    );
    if (Number(claim?.affectedRows ?? 0) === 0) return;
    await appendEvent(runId, 'running', { attempt: nextAttempt });
    const stopHeartbeat = startRunHeartbeat(runId, { attempt: nextAttempt });

    try {
      const { sessionId } = await runWithTimeout(runId, () => executeRun(row, runId));
      await finalizeSuccessfulRun(runId, row, sessionId);
    } catch (err) {
      const latest = await getRunById(runId);
      const recoverySessionId = latest?.agent_session_id ?? row.agent_session_id ?? null;
      if (await recoverRunFromDeliverables({
        runId,
        userId: row.user_id,
        sessionId: recoverySessionId,
        err,
        runStartedAtMs: latest?.started_at ?? row.started_at ?? null,
      })) {
        await finalizeSuccessfulRun(runId, row, recoverySessionId);
        return;
      }
      const message = err instanceof Error ? err.message : String(err);
      const timedOut = err?.code === 'AGENT_RUN_TIMEOUT';
      if (timedOut) {
        await appendEvent(runId, 'timeout', { timeoutMs: runTimeoutMs });
      }
      const retryable = !timedOut && err?.retryable !== false && nextAttempt < retryDelaysMs.length;
      await markRun(runId, retryable ? 'retryable' : 'failed', {
        error_message: message,
        completed_at: retryable ? null : nowMs(),
      });
      if (retryable && autoDispatch) {
        setTimeout(() => dispatchRun(runId), retryDelaysMs[nextAttempt - 1]);
      }
    } finally {
      stopHeartbeat();
    }
  }

  function drainQueue() {
    while (inFlight.size < maxConcurrentRuns) {
      const runId = nextQueuedRunId();
      if (!runId) break;
      inFlight.add(runId);
      void processRun(runId)
        .finally(() => {
          inFlight.delete(runId);
          drainQueue();
        });
    }
  }

  function dispatchRun(runId) {
    if (!enqueueRun(runId)) return;
    drainQueue();
  }

  async function getQueueStatus() {
    const [rows] = await pool.query(
      `SELECT status, COUNT(*) AS count
       FROM h5_agent_runs
       WHERE status IN ('queued', 'running', 'retryable')
       GROUP BY status`,
    );
    const statusCounts = {};
    for (const row of rows) {
      statusCounts[row.status] = Number(row.count ?? 0);
    }
    const [oldestRunningRows] = await pool.query(
      `SELECT
         r.id,
         r.request_id,
         r.started_at,
         r.updated_at,
         r.attempts,
         h.latest_heartbeat_at
       FROM h5_agent_runs r
       LEFT JOIN (
         SELECT run_id, MAX(created_at) AS latest_heartbeat_at
         FROM h5_agent_run_events
         WHERE event_type = 'worker_heartbeat'
         GROUP BY run_id
       ) h ON h.run_id = r.id
       WHERE r.status = 'running'
       ORDER BY COALESCE(h.latest_heartbeat_at, r.started_at) ASC
       LIMIT 1`,
    );
    const oldestRunningStartedAt = oldestRunningRows[0]?.started_at == null
      ? null
      : Number(oldestRunningRows[0].started_at);
    const oldestRunningAgeMs = oldestRunningStartedAt == null
      ? 0
      : Math.max(0, nowMs() - oldestRunningStartedAt);
    const oldestRunningHeartbeatAt = oldestRunningRows[0]?.latest_heartbeat_at == null
      ? null
      : Number(oldestRunningRows[0].latest_heartbeat_at);
    const oldestRunningHeartbeatAgeMs = oldestRunningRows[0]
      ? Math.max(0, nowMs() - Number(oldestRunningRows[0].latest_heartbeat_at ?? oldestRunningRows[0].started_at ?? nowMs()))
      : 0;
    const [runningWithoutHeartbeatRows] = await pool.query(
      `SELECT COUNT(*) AS count
       FROM h5_agent_runs r
       LEFT JOIN (
         SELECT run_id, MAX(created_at) AS latest_heartbeat_at
         FROM h5_agent_run_events
         WHERE event_type = 'worker_heartbeat'
         GROUP BY run_id
       ) h ON h.run_id = r.id
       WHERE r.status = 'running' AND h.latest_heartbeat_at IS NULL`,
    );
    const chatIntentRouterStatus = chatIntentRouter?.getStatus
      ? await Promise.resolve(chatIntentRouter.getStatus()).catch((err) => ({
          enabled: false,
          error: err instanceof Error ? err.message : String(err),
        }))
      : null;
    return {
      autoDispatch,
      maxConcurrentRuns,
      runTimeoutMs,
      heartbeatMs,
      inFlight: inFlight.size,
      pendingDispatches: queuedDispatches.length,
      statusCounts,
      oldestRunningStartedAt,
      oldestRunningAgeMs,
      oldestRunningHeartbeatAt,
      oldestRunningHeartbeatAgeMs,
      runningWithoutHeartbeatCount: Number(runningWithoutHeartbeatRows[0]?.count ?? 0),
      latestRunningRun: oldestRunningRows[0] ? {
        id: oldestRunningRows[0].id,
        requestId: oldestRunningRows[0].request_id,
        startedAt: oldestRunningStartedAt,
        updatedAt: oldestRunningRows[0].updated_at == null ? null : Number(oldestRunningRows[0].updated_at),
        attempts: Number(oldestRunningRows[0].attempts ?? 0),
        heartbeatAt: oldestRunningHeartbeatAt,
        heartbeatAgeMs: oldestRunningHeartbeatAgeMs,
      } : null,
      terminalStatuses: [...TERMINAL_STATUSES],
      toolGateway: toolGateway?.getStatus ? toolGateway.getStatus() : null,
      directChat: directChatService?.getStatus ? directChatService.getStatus() : null,
      chatIntentRouter: chatIntentRouterStatus,
    };
  }

  async function recoverStaleRunningRuns({
    staleMs = runTimeoutMs,
    limit = maxConcurrentRuns,
    dryRun = true,
    reason = 'stale_running_timeout',
    sessionFinishedGraceMs = SESSION_FINISHED_STALE_GRACE_MS,
  } = {}) {
    const normalizedStaleMs = positiveInteger(staleMs, runTimeoutMs);
    const normalizedLimit = positiveInteger(limit, maxConcurrentRuns);
    const normalizedSessionFinishedGraceMs = positiveInteger(
      sessionFinishedGraceMs,
      SESSION_FINISHED_STALE_GRACE_MS,
    );
    const startedCutoff = nowMs() - normalizedStaleMs;
    const sessionFinishedCutoff = nowMs() - normalizedSessionFinishedGraceMs;
    const [rows] = await pool.query(
      `SELECT
         r.id,
         r.user_id,
         r.agent_session_id,
         r.request_id,
         r.started_at,
         r.updated_at,
         r.attempts,
         h.latest_heartbeat_at,
         sf.session_finished_at
       FROM h5_agent_runs r
       LEFT JOIN (
         SELECT run_id, MAX(created_at) AS latest_heartbeat_at
         FROM h5_agent_run_events
         WHERE event_type = 'worker_heartbeat'
         GROUP BY run_id
       ) h ON h.run_id = r.id
       LEFT JOIN (
         SELECT run_id, MAX(created_at) AS session_finished_at
         FROM h5_agent_run_events
         WHERE event_type = 'session_finished'
         GROUP BY run_id
       ) sf ON sf.run_id = r.id
       WHERE r.status = 'running'
         AND r.started_at IS NOT NULL
         AND (
           r.started_at <= ?
           OR (
             sf.session_finished_at IS NOT NULL
             AND sf.session_finished_at <= ?
           )
         )
       ORDER BY COALESCE(sf.session_finished_at, r.started_at) ASC
       LIMIT ?`,
      [startedCutoff, sessionFinishedCutoff, normalizedLimit],
    );
    const recovered = [];
    for (const row of rows) {
      const ageMs = Math.max(0, nowMs() - Number(row.started_at ?? 0));
      const heartbeatAt = row.latest_heartbeat_at == null ? null : Number(row.latest_heartbeat_at);
      const heartbeatAgeMs = Math.max(0, nowMs() - Number(row.latest_heartbeat_at ?? row.started_at ?? 0));
      const sessionFinishedAt = row.session_finished_at == null ? null : Number(row.session_finished_at);
      const item = {
        id: row.id,
        requestId: row.request_id,
        startedAt: Number(row.started_at ?? 0),
        updatedAt: Number(row.updated_at ?? 0),
        attempts: Number(row.attempts ?? 0),
        heartbeatAt,
        heartbeatAgeMs,
        sessionFinishedAt,
        ageMs,
        reason,
      };
      if (!dryRun) {
        const staleError = Object.assign(
          new Error(`agent run stale after ${ageMs}ms`),
          { code: 'AGENT_RUN_STALE_RECOVERY' },
        );
        const deliverableRecovered = await recoverRunFromDeliverables({
          runId: row.id,
          userId: row.user_id,
          sessionId: row.agent_session_id ?? null,
          err: staleError,
          runStartedAtMs: row.started_at ?? null,
          requireRecoverableError: false,
        });
        if (deliverableRecovered) {
          await markRun(row.id, 'succeeded', {
            agent_session_id: row.agent_session_id ?? null,
            completed_at: nowMs(),
            error_message: null,
          });
          item.status = 'succeeded';
          item.recoveredAs = 'deliverables';
          recovered.push(item);
          continue;
        }

        const message = sessionFinishedAt != null
          ? `agent run recovered from stale running state after session finished ${Math.max(0, nowMs() - sessionFinishedAt)}ms ago`
          : heartbeatAt == null
            ? `agent run recovered from stale running state after ${ageMs}ms without heartbeat`
            : `agent run recovered from stale running state after running for ${ageMs}ms`;
        const completedAt = nowMs();
        const [update] = await pool.query(
          `UPDATE h5_agent_runs
           SET status = 'failed', error_message = ?, updated_at = ?, completed_at = ?
           WHERE id = ?
             AND status = 'running'
             AND started_at IS NOT NULL
             AND (
               started_at <= ?
               OR EXISTS (
                 SELECT 1
                 FROM h5_agent_run_events sf
                 WHERE sf.run_id = h5_agent_runs.id
                   AND sf.event_type = 'session_finished'
                   AND sf.created_at <= ?
               )
             )`,
          [message, completedAt, completedAt, row.id, startedCutoff, sessionFinishedCutoff],
        );
        if (Number(update?.affectedRows ?? 0) === 0) continue;
        await appendEvent(row.id, 'stale_recovered', {
          reason,
          staleMs: normalizedStaleMs,
          sessionFinishedGraceMs: normalizedSessionFinishedGraceMs,
          ageMs,
          heartbeatAt,
          heartbeatAgeMs,
          sessionFinishedAt,
          status: 'failed',
          error: message,
        });
        item.status = 'failed';
      }
      recovered.push(item);
    }
    return {
      dryRun,
      staleMs: normalizedStaleMs,
      startedCutoff,
      sessionFinishedCutoff,
      considered: rows.length,
      recovered: dryRun ? 0 : recovered.filter((item) => item.status).length,
      runs: recovered,
    };
  }

  async function dispatchQueuedRuns({ limit = maxConcurrentRuns } = {}) {
    const staleRecovery = await recoverStaleRunningRuns({
      staleMs: runTimeoutMs,
      limit: Math.max(maxConcurrentRuns, Number(limit) || maxConcurrentRuns),
      dryRun: false,
      reason: 'worker_dispatch_stale_recovery',
    });
    const dispatchLimit = Math.max(1, Number(limit) || maxConcurrentRuns);
    const available = Math.max(0, maxConcurrentRuns - inFlight.size - queuedDispatches.length);
    if (available <= 0) {
      return {
        considered: 0,
        dispatched: 0,
        staleRecovery,
        queue: await getQueueStatus(),
      };
    }
    const take = Math.min(dispatchLimit, available);
    const [rows] = await pool.query(
      `SELECT id
       FROM h5_agent_runs
       WHERE status IN ('queued', 'retryable')
       ORDER BY updated_at ASC
       LIMIT ?`,
      [take],
    );
    for (const row of rows) {
      dispatchRun(row.id);
    }
    return {
      considered: rows.length,
      dispatched: rows.length,
      staleRecovery,
      queue: await getQueueStatus(),
    };
  }

  async function listRunEventsForUser(userId, runId, { afterEventId = null, limit = 500 } = {}) {
    const run = await getRunForUser(userId, runId);
    if (!run) return null;

    let afterCreatedAt = null;
    let cursorMiss = false;
    if (afterEventId) {
      const [cursorRows] = await pool.query(
        `SELECT e.created_at
         FROM h5_agent_run_events e
         INNER JOIN h5_agent_runs r ON r.id = e.run_id
         WHERE e.id = ? AND e.run_id = ? AND r.user_id = ?
         LIMIT 1`,
        [afterEventId, runId, userId],
      );
      if (!cursorRows[0]) {
        cursorMiss = true;
      } else {
        afterCreatedAt = Number(cursorRows[0].created_at);
      }
    }

    const [rows] = await pool.query(
      afterCreatedAt == null
        ? `SELECT id, event_type, data_json, created_at
           FROM h5_agent_run_events
           WHERE run_id = ?
           ORDER BY created_at ASC, id ASC
           LIMIT ?`
        : `SELECT id, event_type, data_json, created_at
           FROM h5_agent_run_events
           WHERE run_id = ? AND created_at > ?
           ORDER BY created_at ASC, id ASC
           LIMIT ?`,
      afterCreatedAt == null ? [runId, limit] : [runId, afterCreatedAt, limit],
    );

    return {
      run,
      events: rows.map((row) => ({
        id: row.id,
        eventType: row.event_type,
        data: parseDbJsonColumn(row.data_json, null),
        createdAt: Number(row.created_at),
      })),
      cursorMiss,
    };
  }

  return {
    createRun,
    getRunForUser,
    listRunEventsForUser,
    dispatchRun,
    getQueueStatus,
    dispatchQueuedRuns,
    recoverStaleRunningRuns,
  };
}