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

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,
  systemDisclosurePolicyService = null,
  chatIntentRouter = null,
  sessionSnapshotService = null,
  conversationMemoryService = null,
  syncUserPagesOnSuccess = null,
  observePersonalMemoryOnSuccess = null,
  observeWorkflowRun = 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,
  ),
  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,
      );
    });
  }

  function dispatchShadowObservation(input) {
    if (typeof observeWorkflowRun !== 'function' || input.toolMode !== 'code') return;
    setImmediate(() => {
      const observationStartedAt = nowMs();
      void Promise.resolve()
        .then(() => observeWorkflowRun(input))
        .then(async (result) => {
          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() - observationStartedAt),
          });
        })
        .catch(async (error) => {
          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() - observationStartedAt),
          }).catch((appendError) => {
            console.warn(
              '[AgentRun] workflow shadow failure event skipped:',
              appendError instanceof Error ? appendError.message : appendError,
            );
          });
        });
    });
  }

  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 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;
    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' && 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, 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 (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);
    let toolEvidence = null;
    const awaitSessionFinish = envFlag(process.env.MEMIND_AGENT_RUN_AWAIT_SESSION_FINISH, true)
      && runOptions.toolMode === 'chat'
      && typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser === 'function';
    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: runOptions.toolMode,
              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: runOptions.toolMode,
          forceDeepReasoning: runOptions.forceDeepReasoning,
        },
      );
    }
    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;
    }
    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 = 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 marked = await markRun(runId, 'succeeded', {
      agent_session_id: sessionId,
      completed_at: nowMs(),
      error_message: null,
    }, { expectedStatus: 'running' });
    if (!marked) return false;
    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 && 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,
  };
}