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 {
  deriveAssistantFacingText,
  deriveUserFacingText,
} from './conversation-display.mjs';
import { ensureGooseUserMessageMetadata } from './goose-message.mjs';
import { createShadowObservationDispatcher } from './services/orchestrator/shadow-dispatcher.mjs';
import {
  prepareAndDetectSessionDeliverables,
  SESSION_FINISHED_STALE_GRACE_MS,
  tryRecoverRunFromDeliverables,
} from './agent-run-deliverable-check.mjs';
import {
  isGenericPageGenerationRequest,
  isPageDataIntent,
  isPageGenerationIntent,
} from './chat-skills.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 extractRunMessageText(row) {
  const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
  if (typeof message.content === 'string') return message.content;
  if (!Array.isArray(message.content)) return '';
  return message.content
    .filter((item) => item?.type === 'text')
    .map((item) => String(item.text ?? '').trim())
    .filter(Boolean)
    .join('\n');
}

function extractRunDisplayText(row) {
  const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
  const displayText = String(message?.metadata?.displayText ?? '').trim();
  return displayText || deriveUserFacingText(extractRunMessageText(row));
}

function selectedRunSkill(row) {
  const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
  return String(
    message?.metadata?.[RUN_METADATA_KEY]?.selectedChatSkill
      ?? message?.metadata?.selectedChatSkill
      ?? '',
  ).trim();
}

function isPageDeliverableMutationIntent(text) {
  const normalized = String(text ?? '').trim();
  if (!normalized) return false;
  const withoutNegatedActions = normalized.replace(
    /(?:不要|不再|无需|不需要|禁止|仅确认|只确认)[^。！？\n]{0,24}(?:创建|生成|制作|搭建|开发|实现|设计|建立|修改|更新|编辑|发布|上线|写入|写|改)[^。！？\n]{0,16}/gu,
    '',
  );
  return /(?:帮我|请|给我|需要|要)?(?:做一个|做个|创建|生成|制作|搭建|开发|实现|设计|建立|修改|更新|编辑|发布|上线|写入|写一个|改造|新增)/u
    .test(withoutNegatedActions);
}

function messageText(message) {
  if (typeof message?.content === 'string') return message.content.trim();
  if (!Array.isArray(message?.content)) return '';
  return message.content
    .filter((item) => item?.type === 'text')
    .map((item) => String(item?.text ?? '').trim())
    .filter(Boolean)
    .join('\n');
}

function isPageDataWorkflowRequested(userMessage, taskType = null) {
  const metadata = userMessage?.metadata ?? {};
  const runMetadata = metadata?.[RUN_METADATA_KEY] ?? metadata?.agentRun ?? {};
  const normalizedTaskType = String(
    taskType ?? runMetadata?.taskType ?? metadata?.taskType ?? '',
  ).trim().toLowerCase();
  const selectedSkill = String(
    runMetadata?.selectedChatSkill ?? metadata?.selectedChatSkill ?? '',
  ).trim();
  const displayText = String(metadata?.displayText ?? '').trim()
    || deriveUserFacingText(messageText(userMessage));
  return runMetadata?.pgRequired === true
    || runMetadata?.pageDataAiderWorkflow === true
    || normalizedTaskType === 'page_data_dev'
    || normalizedTaskType === 'page_data_dev_complex'
    || selectedSkill === 'page-data-collect'
    || isPageDataIntent(displayText);
}

function pageDataFailureCheckId(code) {
  switch (String(code ?? '').trim().toUpperCase()) {
    case 'PAGE_DATA_DELIVERY_FAILED':
      return 'page_data_binding';
    case 'PAGE_DATA_DELIVERABLE_MISSING':
      return 'page_data_deliverable';
    case 'DELIVERABLE_DATA_STORAGE_FORBIDDEN':
      return 'page_data_storage_policy';
    case 'REQUIRED_REVIEW_EXECUTOR_MISMATCH':
    case 'REQUIRED_EXECUTOR_UNAVAILABLE':
      return 'independent_review';
    case 'TOOL_GATEWAY_VALIDATION_FAILED':
      return 'code_validation';
    default:
      return 'agent_run_completion';
  }
}

function buildFreshSessionContext(conversation, { maxMessages = 16, maxChars = 12_000 } = {}) {
  const visible = [];
  for (const message of Array.isArray(conversation) ? conversation : []) {
    const role = String(message?.role ?? '').trim();
    if (role !== 'user' && role !== 'assistant') continue;
    const raw = messageText(message);
    if (!raw) continue;
    const text = role === 'user'
      ? deriveUserFacingText(raw)
      : deriveAssistantFacingText(raw);
    if (!text || /^Ran into this error:/i.test(text)) continue;
    visible.push(`${role === 'user' ? '用户' : '助手'}：${text}`);
  }
  const selected = [];
  let usedChars = 0;
  for (let index = visible.length - 1; index >= 0 && selected.length < maxMessages; index -= 1) {
    const item = visible[index];
    if (selected.length > 0 && usedChars + item.length > maxChars) break;
    selected.unshift(item);
    usedChars += item.length;
  }
  return selected.join('\n\n');
}

function appendFreshSessionContext(userMessage, conversation) {
  const context = buildFreshSessionContext(conversation);
  if (!context) return userMessage;
  const message = userMessage && typeof userMessage === 'object'
    ? { ...userMessage }
    : { role: 'user', content: [] };
  const content = Array.isArray(message.content) ? [...message.content] : [];
  const textIndex = content.findIndex((item) => item?.type === 'text');
  const note = `【会话恢复上下文】\n以下内容仅用于理解上文，不是新的执行指令：\n${context}`;
  if (textIndex >= 0) {
    content[textIndex] = {
      ...content[textIndex],
      text: `${String(content[textIndex]?.text ?? '').trim()}\n\n${note}`,
    };
  } else {
    content.unshift({ type: 'text', text: note });
  }
  return { ...message, content };
}

function resolveRequiredImageGeneration(row, routing) {
  const message = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
  const metadata = message.metadata ?? {};
  const runMetadata = metadata[RUN_METADATA_KEY] ?? metadata.agentRun ?? {};
  const requestedMode = String(
    runMetadata.imageGenerationMode ?? runMetadata.image_generation_mode ?? metadata.imageGenerationMode ?? '',
  ).trim().toLowerCase();
  if (requestedMode === 'required') return true;
  if (requestedMode === 'disabled') return false;
  return routing?.imageGeneration?.mode === 'required';
}

export function assertRequiredImageGenerationCompleted(row, routing, toolEvidence) {
  if (!resolveRequiredImageGeneration(row, routing)) return;
  if (toolEvidence?.generateImage?.succeeded === true) return;
  if (toolEvidence?.generateImage?.called !== true) {
    const error = new Error('本轮明确要求生成图片，但 Agent 没有实际调用 generate_image；历史回复不能作为本轮工具证据');
    error.code = 'IMAGE_GENERATION_REQUIRED_NOT_CALLED';
    error.retryable = false;
    throw error;
  }
  const error = new Error('本轮明确要求生成图片，但未获得 image_make 的有效 PNG/JPEG/WebP 产物，不能标记成功');
  error.code = 'IMAGE_GENERATION_REQUIRED_MISSING';
  error.retryable = false;
  throw error;
}

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

export function normalizeAgentRunWorkerIdentity(identity = {}) {
  const runtimeRoot = path.resolve(
    String(identity.runtimeRoot ?? process.cwd()).trim() || process.cwd(),
  );
  const buildId =
    String(
      identity.buildId
      ?? process.env.MEMIND_RUNTIME_BUILD_ID
      ?? process.env.MEMIND_RELEASE_ID
      ?? process.env.GIT_COMMIT
      ?? 'dev',
    ).trim() || 'dev';
  const workerId =
    String(identity.workerId ?? '').trim()
    || `${process.pid}:${crypto.randomUUID()}`;
  return Object.freeze({ workerId, runtimeRoot, buildId });
}

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

export function buildCodeRunCompletionReply(result) {
  const executor = String(result?.executor ?? 'code executor').trim() || 'code executor';
  const output = summarizeText(result?.stdout, 2400).trim();
  return [
    `已由 ${executor} 完成执行，并通过平台文件验收。`,
    output ? `\n${output}` : '',
  ].join('').trim();
}

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

export function resolveRequiredCodeExecutor(userMessage) {
  const metadata = userMessage?.metadata;
  const runMetadata = metadata?.[RUN_METADATA_KEY] ?? metadata?.agentRun ?? {};
  const executor = String(runMetadata?.executor ?? '').trim().toLowerCase();
  return ['aider', 'openhands'].includes(executor) ? executor : null;
}

export function assertRequiredCodeExecutorAvailable(requiredExecutor, toolGatewayStatus) {
  if (!requiredExecutor) return;
  const executors = Array.isArray(toolGatewayStatus?.executors)
    ? toolGatewayStatus.executors.map((item) => String(item).trim().toLowerCase())
    : [];
  if (!toolGatewayStatus?.enabled || !executors.includes(requiredExecutor)) {
    const error = new Error(`必需执行器 ${requiredExecutor} 不可用，任务已停止且不会回退`);
    error.code = 'REQUIRED_EXECUTOR_UNAVAILABLE';
    error.retryable = false;
    throw error;
  }
}

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),
    requiredExecutor: resolveRequiredCodeExecutor(userMessage),
    reviewExecutor: ['aider', 'openhands'].includes(
      String(runMetadata?.reviewExecutor ?? '').trim().toLowerCase(),
    )
      ? String(runMetadata.reviewExecutor).trim().toLowerCase()
      : null,
    pageDataAiderWorkflow: runMetadata?.pageDataAiderWorkflow === true,
    pgRequired: runMetadata?.pgRequired === true,
    forceDeepReasoning: runMetadata?.forceDeepReasoning === true || metadata?.forceDeepReasoning === true,
    validation: normalizeToolGatewayValidation(runMetadata?.validation ?? metadata?.toolGatewayValidation),
    sessionMessageCount: normalizeSessionMessageCount(
      runMetadata?.sessionMessageCount ?? metadata?.sessionMessageCount,
    ),
  };
}

export async function collectAiderReviewFiles(cwd, sinceMs = 0) {
  if (!cwd) return [];
  const root = path.resolve(String(cwd));
  const candidates = [
    { relativeDir: 'public', extensions: new Set(['.html', '.css', '.js']) },
    { relativeDir: '.mindspace/page-data-policies', extensions: new Set(['.json']) },
  ];
  const files = [];
  async function walk(baseDir, relativeDir, extensions) {
    let entries = [];
    try {
      entries = await fs.readdir(baseDir, { withFileTypes: true });
    } catch {
      return;
    }
    for (const entry of entries) {
      if (files.length >= 40) return;
      const absolute = path.join(baseDir, entry.name);
      const relative = path.posix.join(relativeDir.split(path.sep).join('/'), entry.name);
      if (entry.isDirectory()) {
        await walk(absolute, relative, extensions);
        continue;
      }
      if (!entry.isFile() || !extensions.has(path.extname(entry.name).toLowerCase())) continue;
      try {
        const stat = await fs.stat(absolute);
        if (Number(stat.mtimeMs) + 5_000 >= Number(sinceMs || 0)) files.push(relative);
      } catch {
        // File disappeared between directory listing and stat.
      }
    }
  }
  for (const candidate of candidates) {
    await walk(
      path.join(root, candidate.relativeDir),
      candidate.relativeDir,
      candidate.extensions,
    );
  }
  return files;
}

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,
  systemDisclosurePolicyService = null,
  chatIntentRouter = null,
  sessionSnapshotService = null,
  conversationMemoryService = null,
  syncUserPagesOnSuccess = null,
  observePersonalMemoryOnSuccess = null,
  observeWorkflowRun = null,
  observeWorkflowValidation = null,
  isSessionExternallyBusy = null,
  validateRunDeliverables = null,
  quiesceSessionOnTerminal = 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,
  ),
  maxConcurrentShadowObservations = positiveInteger(
    process.env.MEMIND_ORCHESTRATOR_SHADOW_MAX_CONCURRENCY,
    2,
  ),
  maxQueuedShadowObservations = positiveInteger(
    process.env.MEMIND_ORCHESTRATOR_SHADOW_MAX_QUEUE,
    100,
  ),
  enforcePageDataWorkflowValidation = envFlag(
    process.env.MEMIND_ORCHESTRATOR_PAGE_DATA_VALIDATION_GATE_ENABLED,
    false,
  ),
  workerIdentity = null,
}) {
  const worker = normalizeAgentRunWorkerIdentity(workerIdentity ?? {});
  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 quiesceTerminalSession(runId, {
    userId,
    sessionId,
    requestId = null,
    status,
  }) {
    if (!sessionId || typeof quiesceSessionOnTerminal !== 'function') return;
    try {
      const result = await quiesceSessionOnTerminal({
        runId,
        userId,
        sessionId,
        requestId,
        status,
      });
      await appendEvent(runId, 'session_extensions_quiesced', {
        sessionId,
        status,
        removed: result?.removed ?? [],
        skipped: Boolean(result?.skipped),
        cancellation: result?.cancellation ?? null,
      });
    } catch (err) {
      await appendEvent(runId, 'session_extension_quiesce_failed', {
        sessionId,
        status,
        error: err instanceof Error ? err.message : String(err),
      }).catch(() => {});
      console.warn(
        `[AgentRun] failed to quiesce session ${sessionId}:`,
        err instanceof Error ? err.message : err,
      );
    }
  }

  async function appendExecutionPlanEvent(input, source) {
    const executionPlan = source?.executionPlan;
    if (!executionPlan?.dryRun) return;
    await appendEvent(input.runId, 'workflow_execution_planned', {
      version: executionPlan.version ?? 'workflow-execution-decision-v1',
      mode: source.mode ?? null,
      candidateEngine: executionPlan.candidateEngine ?? null,
      effectiveEngine: executionPlan.effectiveEngine ?? 'native',
      fallbackEngine: executionPlan.fallbackEngine ?? 'native',
      reason: executionPlan.reason ?? 'execution_gate_disabled',
      candidateReason: executionPlan.candidateReason ?? null,
      configVersion: executionPlan.configVersion ?? null,
      bucket: executionPlan.bucket ?? null,
      taskType: executionPlan.taskType ?? input.taskType ?? null,
      dryRun: true,
      handoffAllowed: false,
    }).catch((error) => {
      console.warn(
        '[AgentRun] workflow execution plan event skipped:',
        error instanceof Error ? error.message : error,
      );
    });
  }

  const shadowDispatcher = createShadowObservationDispatcher({
    observe: observeWorkflowRun,
    maxConcurrent: maxConcurrentShadowObservations,
    maxQueued: maxQueuedShadowObservations,
    onCompleted: async (input, result, context) => {
      await appendExecutionPlanEvent(input, result);
      if (!result?.observed) return;
      await appendEvent(input.runId, 'workflow_shadow_completed', {
        engine: result.engine ?? 'langgraph',
        mode: result.mode ?? 'shadow',
        configVersion: result.configVersion ?? null,
        status: result.shadowRun?.status ?? null,
        phase: result.shadowRun?.phase ?? null,
        taskType: result.shadowRun?.plan?.taskType ?? input.taskType ?? null,
        executorAdapter: result.shadowRun?.plan?.executorAdapter ?? null,
        latencyMs: Math.max(0, nowMs() - context.enqueuedAt),
      });
    },
    onFailed: async (input, error, context) => {
      await appendExecutionPlanEvent(input, error);
      console.warn(
        '[AgentRun] workflow shadow observation failed:',
        error instanceof Error ? error.message : error,
      );
      await appendEvent(input.runId, 'workflow_shadow_failed', {
        code: String(error?.code ?? 'WORKFLOW_SHADOW_FAILED').slice(0, 128),
        message: String(error instanceof Error ? error.message : error).slice(0, 1000),
        latencyMs: Math.max(0, nowMs() - context.enqueuedAt),
      }).catch((appendError) => {
        console.warn(
          '[AgentRun] workflow shadow failure event skipped:',
          appendError instanceof Error ? appendError.message : appendError,
        );
      });
    },
    onSkipped: async (input, context) => {
      await appendEvent(input.runId, 'workflow_shadow_skipped', {
        reason: context.reason,
      });
    },
    logger: console,
  });

  const validationDispatcher = createShadowObservationDispatcher({
    observe: observeWorkflowValidation,
    maxConcurrent: maxConcurrentShadowObservations,
    maxQueued: maxQueuedShadowObservations,
    onCompleted: async (input, result, context) => {
      if (!result?.observed) {
        await appendEvent(input.runId, 'workflow_validation_observation_skipped', {
          reason: result?.reason ?? 'shadow_validation_not_selected',
        });
        return;
      }
      await appendEvent(input.runId, 'workflow_validation_observation_completed', {
        engine: result.engine ?? 'langgraph',
        mode: result.mode ?? 'shadow',
        configVersion: result.configVersion ?? null,
        kind: result.validation?.kind ?? 'page-data-delivery',
        verdict: result.validation?.verdict ?? null,
        latencyMs: Math.max(0, nowMs() - context.enqueuedAt),
      });
    },
    onFailed: async (input, error, context) => {
      await appendEvent(input.runId, 'workflow_validation_observation_failed', {
        code: String(error?.code ?? 'WORKFLOW_VALIDATION_OBSERVATION_FAILED').slice(0, 128),
        message: String(error instanceof Error ? error.message : error).slice(0, 1000),
        latencyMs: Math.max(0, nowMs() - context.enqueuedAt),
      }).catch(() => {});
    },
    onSkipped: async (input, context) => {
      await appendEvent(input.runId, 'workflow_validation_observation_skipped', {
        reason: context.reason,
      });
    },
    logger: console,
  });

  function dispatchShadowObservation(input) {
    const pageDataRequired = isPageDataWorkflowRequested(input.userMessage, input.taskType);
    if (input.toolMode !== 'code' && !pageDataRequired) return;
    shadowDispatcher.dispatch({
      ...input,
      taskType: input.taskType || (pageDataRequired ? 'page_data_dev' : null),
      pageDataRequired,
    });
  }

  function buildPageDataValidationObservationInput({
    row,
    runId,
    checks,
    metrics = null,
  }) {
    const userMessage = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
    const runOptions = getRunOptionsFromMessage(userMessage);
    return {
      runId,
      requestId: row?.request_id ?? null,
      userId: row?.user_id ?? null,
      workflowName: 'code-run-v1',
      observation: {
        idempotencyKey: `${runId}:page-data-delivery:v1`,
        taskType: runOptions.taskType || 'page_data_dev',
        required: true,
        checks,
        metrics: {
          pageCount: Number(metrics?.pageCount ?? 0),
          publicationCount: Number(metrics?.publicationCount ?? 0),
        },
        source: 'portal-agent-run',
        observedAt: nowMs(),
      },
    };
  }

  function dispatchPageDataValidationObservation(input) {
    return validationDispatcher.dispatch(input);
  }

  async function enforcePageDataValidationGate(input) {
    const startedAt = nowMs();
    let result = null;
    let gateError = null;
    try {
      if (typeof observeWorkflowValidation !== 'function') {
        gateError = new Error('Page Data Orchestrator validation observer is unavailable');
        gateError.code = 'WORKFLOW_VALIDATION_GATE_UNAVAILABLE';
        throw gateError;
      }
      result = await observeWorkflowValidation(input);
      const verdict = result?.validation?.verdict ?? null;
      if (!result?.observed) {
        gateError = new Error(
          `Page Data Orchestrator validation was not observed: ${result?.reason ?? 'not_selected'}`,
        );
        gateError.code = 'WORKFLOW_VALIDATION_GATE_UNAVAILABLE';
        throw gateError;
      }
      if (verdict !== 'passed') {
        gateError = new Error(
          `Page Data Orchestrator validation did not pass: ${verdict ?? 'unknown'}`,
        );
        gateError.code = verdict === 'failed'
          ? 'WORKFLOW_VALIDATION_GATE_REJECTED'
          : 'WORKFLOW_VALIDATION_GATE_INCONCLUSIVE';
        throw gateError;
      }
      await appendEvent(input.runId, 'workflow_validation_gate_passed', {
        engine: result.engine ?? 'langgraph',
        mode: result.mode ?? 'shadow',
        configVersion: result.configVersion ?? null,
        kind: result.validation?.kind ?? 'page-data-delivery',
        verdict,
        latencyMs: Math.max(0, nowMs() - startedAt),
      });
      return result;
    } catch (error) {
      const resolved = gateError ?? error;
      const code = String(
        gateError?.code ?? 'WORKFLOW_VALIDATION_GATE_UNAVAILABLE',
      ).slice(0, 128);
      const failure = resolved instanceof Error
        ? resolved
        : new Error(String(resolved ?? 'Page Data Orchestrator validation failed'));
      await appendEvent(input.runId, 'workflow_validation_gate_failed', {
        code,
        upstreamCode: gateError
          ? null
          : String(error?.code ?? 'WORKFLOW_VALIDATION_OBSERVER_FAILED').slice(0, 128),
        message: String(
          failure.message,
        ).slice(0, 1000),
        verdict: result?.validation?.verdict ?? null,
        latencyMs: Math.max(0, nowMs() - startedAt),
      }).catch(() => {});
      failure.code = code;
      failure.retryable = false;
      failure.validationObservationAttempted = true;
      throw failure;
    }
  }

  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)) {
      if (typeof isSessionExternallyBusy === 'function' && await isSessionExternallyBusy({
        userId,
        sessionId,
      })) {
        const conflict = new Error('该会话正在完成页面交付或自动修复，请稍候再发送');
        conflict.code = 'SESSION_RUN_CONFLICT';
        conflict.status = 409;
        throw conflict;
      }
      let [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) {
        await recoverStaleRunningRuns({
          staleMs: runTimeoutMs,
          limit: 1,
          dryRun: false,
          reason: 'create_run_session_conflict_stale_recovery',
          sessionId,
        });
        [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,
          required_runtime_root, required_build_id,
          claimed_worker_id, claimed_runtime_root, claimed_build_id)
       VALUES (?, ?, ?, ?, 'queued', 0, ?, NULL, ?, ?, NULL, NULL,
               ?, ?, NULL, NULL, NULL)`,
      [
        runId,
        userId,
        sessionId || null,
        normalizedRequestId,
        serializeMessage(runMessage),
        createdAt,
        createdAt,
        worker.runtimeRoot,
        worker.buildId,
      ],
    );
    await appendEvent(runId, 'queued', {
      sessionId: sessionId || null,
      toolMode: normalizedToolMode,
      taskType: normalizedTaskType,
      forceDeepReasoning: Boolean(forceDeepReasoning),
      requiredRuntimeRoot: worker.runtimeRoot,
      requiredBuildId: worker.buildId,
    });
    const createdRun = projectRun(await getRunById(runId));
    dispatchShadowObservation({
      runId,
      requestId: normalizedRequestId,
      userId,
      sessionId: sessionId || null,
      workflowName: 'code-run-v1',
      userMessage: runMessage,
      toolMode: normalizedToolMode,
      taskType: normalizedTaskType,
    });
    if (autoDispatch) dispatchRun(runId);
    return createdRun;
  }

  async function prepareRunDeliverables({ userId, sessionId, runId, runStartedAtMs = null }) {
    if (typeof syncUserPagesOnSuccess !== 'function') return;
    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 = {}, { expectedStatus = null } = {}) {
    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);
    const where = expectedStatus
      ? 'WHERE id = ? AND status = ?'
      : 'WHERE id = ?';
    if (expectedStatus) values.push(expectedStatus);
    const [result] = await pool.query(
      `UPDATE h5_agent_runs SET ${updates.join(', ')} ${where}`,
      values,
    );
    if (Number(result?.affectedRows ?? 0) === 0) return false;
    await appendEvent(runId, status, fields);
    await appendRunSnapshot(runId);
    return true;
  }

  function startRunHeartbeat(runId, { attempt }) {
    let stopped = false;
    const writeHeartbeat = async () => {
      if (stopped) return;
      try {
        await appendEvent(runId, 'worker_heartbeat', {
          attempt,
          pid: process.pid,
          heartbeatMs,
          workerId: worker.workerId,
          runtimeRoot: worker.runtimeRoot,
          buildId: worker.buildId,
        });
      } 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 runRequiredCodeReview({
    row,
    runId,
    userMessage,
    runOptions,
    sessionId,
  }) {
    if (!runOptions.reviewExecutor) return;
    const toolGatewayStatus = toolGateway?.getStatus ? toolGateway.getStatus() : null;
    assertRequiredCodeExecutorAvailable(runOptions.reviewExecutor, toolGatewayStatus);
    const workingDir = userAuth?.resolveWorkingDir
      ? await userAuth.resolveWorkingDir(row.user_id)
      : undefined;
    const claimedRun = await getRunById(runId);
    const contextFiles = await collectAiderReviewFiles(
      workingDir,
      claimedRun?.started_at ?? row.started_at ?? 0,
    );
    const reviewMessage = {
      ...userMessage,
      content: [{
        type: 'text',
        text: [
          '[Mandatory Aider Page Data review]',
          `Original user request: ${extractRunDisplayText(row)}`,
          `Session: ${sessionId}`,
          'Review the provided workspace files, fix concrete HTML/client-policy defects if needed,',
          'do not recreate PostgreSQL tables or datasets, and update the required validation receipt.',
          'Do not commit, push, publish, or modify files outside the current user workspace.',
          contextFiles.length
            ? `Review files:\n${contextFiles.map((item) => `- ${item}`).join('\n')}`
            : 'No recent page files were detected; record that fact in the receipt so platform delivery validation can fail closed.',
          `Receipt requestId: ${row.request_id}`,
        ].join('\n'),
      }],
      metadata: {
        ...(userMessage?.metadata ?? {}),
        memindRun: {
          ...(userMessage?.metadata?.memindRun ?? {}),
          executor: runOptions.reviewExecutor,
          aiderContextFiles: contextFiles,
        },
      },
    };
    await appendEvent(runId, 'required_code_review_dispatch', {
      executor: runOptions.reviewExecutor,
      contextFiles,
    });
    const result = await toolGateway.executeJob({
      runId,
      userId: row.user_id,
      requestId: row.request_id,
      userMessage: reviewMessage,
      taskType: 'page_data_dev',
      cwd: workingDir,
      timeoutMs: runTimeoutMs,
    });
    if (
      String(result?.executor ?? '').trim().toLowerCase() !== runOptions.reviewExecutor
    ) {
      const error = new Error(
        `审查执行器不匹配：要求 ${runOptions.reviewExecutor}，实际 ${result?.executor ?? 'unknown'}`,
      );
      error.code = 'REQUIRED_REVIEW_EXECUTOR_MISMATCH';
      error.retryable = false;
      throw error;
    }
    await appendEvent(runId, 'required_code_review_result', {
      executor: result.executor ?? null,
      exitCode: result.exitCode ?? null,
      stdoutTail: summarizeText(result.stdout),
      stderrTail: summarizeText(result.stderr),
    });
    const validation = await validateToolGatewayResult({
      result,
      validation: runOptions.validation,
      cwd: workingDir,
    });
    if (validation) {
      await appendEvent(runId, 'required_code_review_validation', validation);
    }
  }

  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;
    assertRequiredCodeExecutorAvailable(
      runOptions.requiredExecutor ?? runOptions.reviewExecutor,
      toolGatewayStatus,
    );
    const effectiveToolMode = runOptions.pageDataAiderWorkflow
      ? 'chat'
      : runOptions.toolMode;
    let disclosureDecision = null;
    try {
      disclosureDecision = systemDisclosurePolicyService?.evaluate?.({
        userMessage,
        channel: 'h5',
        userId: row.user_id,
        sessionId: row.agent_session_id ?? null,
      }) ?? null;
    } catch (err) {
      console.warn(
        '[AgentRun] system disclosure policy evaluation failed open:',
        err instanceof Error ? err.message : err,
      );
    }
    if (disclosureDecision?.enforced) {
      if (!directChatService?.respondDeterministically) {
        const error = new Error('System Disclosure Policy refusal service unavailable');
        error.code = 'SYSTEM_DISCLOSURE_REFUSAL_UNAVAILABLE';
        error.retryable = false;
        throw error;
      }
      const result = await directChatService.respondDeterministically({
        userId: row.user_id,
        sessionId: row.agent_session_id ?? null,
        requestId: row.request_id,
        userMessage,
        reply: disclosureDecision.responseText,
        metadata: {
          policyId: disclosureDecision.policyId,
          policyVersion: disclosureDecision.policyVersion,
          policyReasonCode: disclosureDecision.reasonCode,
        },
        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, 'system_disclosure_blocked', {
        policyId: disclosureDecision.policyId,
        policyVersion: disclosureDecision.policyVersion,
        reasonCode: disclosureDecision.reasonCode,
        categories: disclosureDecision.categories,
        channel: 'h5',
      });
      return {
        sessionId: result.sessionId,
        routing: null,
        toolEvidence: null,
        policyBlocked: true,
      };
    }
    const routing = await resolveRunRouting(row, userMessage, runOptions);
    const routingDecision = resolveLegacyRouteFromClassification(routing) ?? routing?.route ?? null;
    let agentMemoryContext = null;
    if (routingDecision === CHAT_INTENT_ROUTE.AGENT && chatIntentRouter?.resolveAgentMemoryContext) {
      const displayText = userMessage?.metadata?.displayText
        ?? userMessage?.content?.find?.((item) => item?.type === 'text')?.text
        ?? '';
      try {
        agentMemoryContext = await chatIntentRouter.resolveAgentMemoryContext({
          userId: row.user_id,
          sessionId: row.agent_session_id ?? null,
          text: displayText,
          forceDeepReasoning: runOptions.forceDeepReasoning,
        });
        if (agentMemoryContext?.enabled && agentMemoryContext.mode !== 'off') {
          await appendEvent(runId, 'agent_memory_resolved', {
            mode: agentMemoryContext.mode,
            injectionEnabled: Boolean(agentMemoryContext.injectionEnabled),
            source: agentMemoryContext.source ?? null,
            memoryCount: Array.isArray(agentMemoryContext.memories)
              ? agentMemoryContext.memories.length
              : 0,
            skipped: Boolean(agentMemoryContext.skipped),
            degraded: Boolean(agentMemoryContext.degraded),
            reason: agentMemoryContext.reason ?? null,
            latencyMs: Number(agentMemoryContext.latencyMs ?? 0),
          });
        }
      } catch (err) {
        console.warn(
          '[AgentRun] agent memory shadow resolve skipped:',
          err instanceof Error ? err.message : err,
        );
        agentMemoryContext = 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,
          memoryContext: agentMemoryContext,
        });
      }
    }
    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, routing };
      } 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' &&
      !runOptions.pageDataAiderWorkflow &&
      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,
      });
      if (
        runOptions.requiredExecutor &&
        String(result?.executor ?? '').trim().toLowerCase() !== runOptions.requiredExecutor
      ) {
        const error = new Error(
          `执行器不匹配：要求 ${runOptions.requiredExecutor}，实际 ${result?.executor ?? 'unknown'}`,
        );
        error.code = 'REQUIRED_EXECUTOR_MISMATCH';
        error.retryable = false;
        throw error;
      }
      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;
      }
      if (runOptions.requiredExecutor) {
        if (!directChatService?.respondDeterministically) {
          const error = new Error('代码任务已执行，但结果回传服务不可用');
          error.code = 'CODE_RUN_RESULT_DELIVERY_UNAVAILABLE';
          error.retryable = false;
          throw error;
        }
        const delivery = await directChatService.respondDeterministically({
          userId: row.user_id,
          sessionId: row.agent_session_id ?? null,
          requestId: row.request_id,
          userMessage,
          reply: buildCodeRunCompletionReply(result),
          metadata: {
            source: 'tool-gateway-code-run',
            executor: result.executor ?? null,
            validated: true,
          },
          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, 'tool_gateway_result_delivered', {
          sessionId: delivery.sessionId,
          executor: result.executor ?? null,
        });
        return { sessionId: delivery.sessionId, routing };
      }
      return { sessionId: row.agent_session_id ?? null, routing };
    }

    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 (effectiveToolMode === '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: effectiveToolMode,
        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);
    let toolEvidence = null;
    const awaitSessionFinish = envFlag(process.env.MEMIND_AGENT_RUN_AWAIT_SESSION_FINISH, true)
      && effectiveToolMode === 'chat'
      && typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser === 'function';
    if (
      runOptions.pageDataAiderWorkflow &&
      typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser !== 'function'
    ) {
      const error = new Error('Page Data + Aider 工作流需要等待 Page Data Agent 完成');
      error.code = 'PAGE_DATA_AIDER_FINISH_UNAVAILABLE';
      error.retryable = false;
      throw error;
    }
    if (awaitSessionFinish) {
      let submitMessage = ensureGooseUserMessageMetadata(userMessage);
      let replacedPoisonedSession = false;
      while (true) {
        try {
          const finish = await tkmindProxy.submitSessionReplyAndAwaitFinishForUser(
            row.user_id,
            sessionId,
            row.request_id,
            submitMessage,
            {
              toolMode: effectiveToolMode,
              forceDeepReasoning: runOptions.forceDeepReasoning,
              timeoutMs: runTimeoutMs,
            },
          );
          await appendEvent(runId, 'session_finished', {
            sessionId,
            tokenState: finish.tokenState ?? null,
            toolCalls: finish.toolEvidence?.calls ?? [],
            generateImage: finish.toolEvidence?.generateImage ?? null,
          });
          toolEvidence = finish.toolEvidence ?? null;
          break;
        } catch (err) {
          if (
            !replacedPoisonedSession
            && err?.code === 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED'
          ) {
            const previousSessionId = sessionId;
            const repairedConversation = Array.isArray(err?.repairedConversation)
              ? err.repairedConversation
              : [];
            const replacement = await tkmindProxy.startSessionForUser(row.user_id);
            sessionId = replacement.id;
            replacedPoisonedSession = true;
            await pool.query(
              `UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`,
              [sessionId, nowMs(), runId],
            );
            const persisted = await persistSessionTranscriptMessages({
              conversationMemoryService,
              sessionId,
              userId: row.user_id,
              messages: repairedConversation,
            });
            await appendEvent(runId, 'poisoned_session_replaced', {
              previousSessionId,
              sessionId,
              repairedMessageCount: repairedConversation.length,
              persistedMessageCount: persisted.saved,
            });
            await appendRunSnapshot(runId);
            await invalidatePortalDirectChatSnapshot(sessionId);
            submitMessage = ensureGooseUserMessageMetadata(
              appendFreshSessionContext(userMessage, repairedConversation),
            );
            continue;
          }
          if (await recoverRunFromDeliverables({
            runId,
            userId: row.user_id,
            sessionId,
            err,
          })) {
            return { sessionId, routing };
          }
          throw err;
        }
      }
    } else {
      await tkmindProxy.submitSessionReplyForUser(
        row.user_id,
        sessionId,
        row.request_id,
        ensureGooseUserMessageMetadata(userMessage),
        {
          toolMode: effectiveToolMode,
          forceDeepReasoning: runOptions.forceDeepReasoning,
        },
      );
    }
    await runRequiredCodeReview({
      row,
      runId,
      userMessage,
      runOptions,
      sessionId,
    });
    return { sessionId, routing, toolEvidence };
  }

  async function finalizeSuccessfulRun(
    runId,
    row,
    sessionId,
    { routing = null, toolEvidence = null, policyBlocked = false } = {},
  ) {
    if (policyBlocked) {
      await markRun(runId, 'succeeded', {
        agent_session_id: sessionId,
        completed_at: nowMs(),
        error_message: null,
      }, { expectedStatus: 'running' });
      return;
    }
    const userMessage = parseDbJsonColumn(row.user_message_json, {}) ?? {};
    const runOptions = getRunOptionsFromMessage(userMessage);
    assertRequiredImageGenerationCompleted(row, routing, toolEvidence);
    // `row` was loaded before this worker claimed the run, so its started_at
    // can still be null. Refresh it before scoping workspace files to the
    // current execution window.
    const claimedRun = await getRunById(runId);
    const runStartedAtMs = claimedRun?.started_at ?? row.started_at ?? null;
    let deliveryResult = null;
    if (typeof syncUserPagesOnSuccess === 'function') {
      deliveryResult = await syncUserPagesOnSuccess({
        userId: row.user_id,
        sessionId,
        runId,
        runStartedAtMs,
      });
    }
    const pageDataErrors = Array.isArray(deliveryResult?.pageDataBind?.errors)
      ? deliveryResult.pageDataBind.errors
      : [];
    if (pageDataErrors.length > 0) {
      const error = new Error(`Page Data 页面绑定失败：${pageDataErrors.map((item) => item?.message ?? item?.code ?? 'unknown').join('; ')}`);
      error.code = 'PAGE_DATA_DELIVERY_FAILED';
      error.retryable = false;
      throw error;
    }
    const runDisplayText = extractRunDisplayText(row);
    const selectedSkill = selectedRunSkill(row);
    const pageDataIntent = isPageDataWorkflowRequested(userMessage, runOptions.taskType)
      || isPageDataIntent(runDisplayText)
      || selectedSkill === 'page-data-collect'
      || routing?.suggestedSkill === 'page-data-collect';
    const pageGenerationIntent = isPageGenerationIntent(runDisplayText)
      || selectedSkill === 'generate-page'
      || routing?.suggestedSkill === 'static-page-publish';
    // A bare “generate a page” request has no subject to render. The assistant's
    // clarification is a successful conversational turn, not a failed delivery.
    // Once the user supplies a subject, the existing fail-closed guard still applies.
    const requiresPageDeliverable = isPageDeliverableMutationIntent(runDisplayText)
      && (pageDataIntent || (
        pageGenerationIntent
        && !isGenericPageGenerationRequest(runDisplayText)
      ));
    let deliverables = null;
    if (requiresPageDeliverable || typeof validateRunDeliverables === 'function') {
      const latest = await getRunById(runId);
      // `[]` is a strict allowlist meaning no workspace page may be examined.
      // Preserve it so a run with no current artifact cannot fall back to
      // historical workspace pages; `null` is reserved for legacy callers
      // that provide no scoping information at all.
      const workspaceRelativePaths = Array.isArray(deliveryResult?.pageDataRelativePaths)
        ? deliveryResult.pageDataRelativePaths
        : null;
      deliverables = await prepareAndDetectSessionDeliverables({
        pool,
        userId: row.user_id,
        sessionId,
        runStartedAtMs: latest?.started_at ?? runStartedAtMs,
        workspaceRelativePaths,
      });
      if (requiresPageDeliverable && deliverables.pageCount < 1) {
        const error = new Error(
          pageDataIntent
            ? 'Page Data 任务未生成可交付页面，不能标记成功'
            : '页面任务未生成 public HTML 交付物，不能标记成功',
        );
        error.code = pageDataIntent
          ? 'PAGE_DATA_DELIVERABLE_MISSING'
          : 'PUBLIC_PAGE_DELIVERABLE_MISSING';
        error.retryable = false;
        throw error;
      }
    }
    if (typeof validateRunDeliverables === 'function') {
      const validation = await validateRunDeliverables({
        userId: row.user_id,
        sessionId,
        runId,
        deliverables: deliverables ?? { pageCount: 0, publicationCount: 0, pages: [] },
      });
      const errors = Array.isArray(validation?.errors) ? validation.errors : [];
      if (errors.length > 0) {
        const error = new Error(
          `页面交付违反数据存储策略：${errors.map((item) => item?.message ?? item?.code ?? 'unknown').join('; ')}`,
        );
        error.code = 'DELIVERABLE_DATA_STORAGE_FORBIDDEN';
        error.retryable = false;
        throw error;
      }
    }
    const validationInput = pageDataIntent
      ? buildPageDataValidationObservationInput({
          row,
          runId,
          checks: [
            { id: 'agent_run_completion', status: 'passed' },
            {
              id: 'page_data_binding',
              status: typeof syncUserPagesOnSuccess === 'function' ? 'passed' : 'skipped',
            },
            {
              id: 'page_data_storage_policy',
              status: typeof validateRunDeliverables === 'function' ? 'passed' : 'skipped',
            },
            {
              id: 'page_data_deliverable',
              status: requiresPageDeliverable ? 'passed' : 'skipped',
            },
            {
              id: 'independent_review',
              status: runOptions.reviewExecutor ? 'passed' : 'skipped',
            },
          ],
          metrics: {
            pageCount: deliverables?.pageCount ?? 0,
            publicationCount: deliverables?.publicationCount ?? 0,
          },
        })
      : null;
    if (validationInput && enforcePageDataWorkflowValidation) {
      await enforcePageDataValidationGate(validationInput);
    }
    const marked = await markRun(runId, 'succeeded', {
      agent_session_id: sessionId,
      completed_at: nowMs(),
      error_message: null,
    }, { expectedStatus: 'running' });
    if (!marked) return false;
    if (validationInput && !enforcePageDataWorkflowValidation) {
      dispatchPageDataValidationObservation(validationInput);
    }
    if (typeof observePersonalMemoryOnSuccess === 'function') {
      await observePersonalMemoryOnSuccess({
        userId: row.user_id,
        sessionId,
        runId,
        userMessage: parseDbJsonColumn(row.user_message_json, {}),
      }).catch((err) => {
        console.warn(
          '[AgentRun] personal memory shadow observation 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,
           claimed_worker_id = ?, claimed_runtime_root = ?, claimed_build_id = ?
       WHERE id = ?
         AND status IN ('queued', 'retryable')
         AND (required_runtime_root IS NULL OR required_runtime_root = ?)
         AND (required_build_id IS NULL OR required_build_id = ?)`,
      [
        nextAttempt,
        nowMs(),
        nowMs(),
        worker.workerId,
        worker.runtimeRoot,
        worker.buildId,
        runId,
        worker.runtimeRoot,
        worker.buildId,
      ],
    );
    if (Number(claim?.affectedRows ?? 0) === 0) return;
    await appendEvent(runId, 'running', {
      attempt: nextAttempt,
      workerId: worker.workerId,
      runtimeRoot: worker.runtimeRoot,
      buildId: worker.buildId,
    });
    const stopHeartbeat = startRunHeartbeat(runId, { attempt: nextAttempt });

    try {
      const execution = await runWithTimeout(runId, () => executeRun(row, runId));
      await finalizeSuccessfulRun(runId, row, execution.sessionId, execution);
    } catch (err) {
      const latest = await getRunById(runId);
      const recoverySessionId = latest?.agent_session_id ?? row.agent_session_id ?? null;
      if (err?.code !== 'IMAGE_GENERATION_REQUIRED_MISSING' && 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(),
        ...(retryable
          ? {
              claimed_worker_id: null,
              claimed_runtime_root: null,
              claimed_build_id: null,
            }
          : {}),
      }, { expectedStatus: 'running' });
      if (!retryable && err?.validationObservationAttempted !== true) {
        const userMessage = parseDbJsonColumn(row?.user_message_json, {}) ?? {};
        const runOptions = getRunOptionsFromMessage(userMessage);
        if (isPageDataWorkflowRequested(userMessage, runOptions.taskType)) {
          dispatchPageDataValidationObservation(
            buildPageDataValidationObservationInput({
              row,
              runId,
              checks: [{
                id: pageDataFailureCheckId(err?.code),
                status: 'failed',
                codes: [String(err?.code ?? 'AGENT_RUN_FAILED')],
              }],
            }),
          );
        }
      }
      if (retryable && autoDispatch) {
        setTimeout(() => dispatchRun(runId), retryDelaysMs[nextAttempt - 1]);
      }
    } finally {
      stopHeartbeat();
      const terminalRun = await getRunById(runId).catch(() => null);
      if (terminalRun && TERMINAL_STATUSES.has(terminalRun.status)) {
        await quiesceTerminalSession(runId, {
          userId: terminalRun.user_id ?? row.user_id,
          sessionId: terminalRun.agent_session_id ?? row.agent_session_id ?? null,
          requestId: terminalRun.request_id ?? row.request_id ?? null,
          status: terminalRun.status,
        });
      }
    }
  }

  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,
    sessionId = null,
  } = {}) {
    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 heartbeatCutoff = nowMs() - normalizedStaleMs;
    const normalizedSessionId = String(sessionId ?? '').trim();
    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 (h.latest_heartbeat_at IS NULL OR h.latest_heartbeat_at <= ?)
         AND (
           r.started_at <= ?
           OR (
             sf.session_finished_at IS NOT NULL
             AND sf.session_finished_at <= ?
           )
         )
         ${normalizedSessionId ? 'AND r.agent_session_id = ?' : ''}
       ORDER BY COALESCE(sf.session_finished_at, r.started_at) ASC
       LIMIT ?`,
      [
        heartbeatCutoff,
        startedCutoff,
        sessionFinishedCutoff,
        ...(normalizedSessionId ? [normalizedSessionId] : []),
        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) {
          const marked = await markRun(row.id, 'succeeded', {
            agent_session_id: row.agent_session_id ?? null,
            completed_at: nowMs(),
            error_message: null,
          }, { expectedStatus: 'running' });
          if (!marked) continue;
          item.status = 'succeeded';
          item.recoveredAs = 'deliverables';
          await quiesceTerminalSession(row.id, {
            userId: row.user_id,
            sessionId: row.agent_session_id ?? null,
            requestId: row.request_id ?? null,
            status: 'succeeded',
          });
          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 <= ?
               )
             )
             AND (
               SELECT COALESCE(MAX(hb.created_at), h5_agent_runs.started_at)
               FROM h5_agent_run_events hb
               WHERE hb.run_id = h5_agent_runs.id
                 AND hb.event_type = 'worker_heartbeat'
             ) <= ?`,
          [message, completedAt, completedAt, row.id, startedCutoff, sessionFinishedCutoff, heartbeatCutoff],
        );
        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';
        await quiesceTerminalSession(row.id, {
          userId: row.user_id,
          sessionId: row.agent_session_id ?? null,
          requestId: row.request_id ?? null,
          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')
         AND (required_runtime_root IS NULL OR required_runtime_root = ?)
         AND (required_build_id IS NULL OR required_build_id = ?)
       ORDER BY updated_at ASC
       LIMIT ?`,
      [worker.runtimeRoot, worker.buildId, 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,
  };
}