import crypto from 'node:crypto';
import fs from 'node:fs/promises';
import path from 'node:path';

const JOB_TYPES = new Set(['generate_page', 'analyze_asset', 'summarize']);
const OUTPUT_TYPES = new Set(['page_draft', 'html_page', 'markdown']);
const ACTIVE_JOB_STATUSES = new Set(['queued', 'running']);
const RETRYABLE_ERROR_CODES = new Set([
  'worker_crashed',
  'worker_unavailable',
  'model_rate_limited',
  'network_temporary',
  'storage_temporary',
  'job_timed_out',
  'invalid_agent_job_output',
]);

function asNumber(value) {
  return Number(value ?? 0);
}

function agentJobError(message, code, details) {
  return Object.assign(new Error(message), { code, details });
}

function dedupeStrings(values) {
  return [...new Set((Array.isArray(values) ? values : []).map((value) => String(value).trim()).filter(Boolean))];
}

function normalizeInstruction(value) {
  const instruction = String(value ?? '').normalize('NFKC').trim();
  if (!instruction) throw agentJobError('任务说明不能为空', 'invalid_agent_job_input');
  if (instruction.length > 4000) {
    throw agentJobError('任务说明超过长度限制', 'invalid_agent_job_input');
  }
  return instruction;
}

function normalizeJobInput(input) {
  const jobType = String(input.jobType ?? '').trim();
  if (!JOB_TYPES.has(jobType)) {
    throw agentJobError('不支持的任务类型', 'invalid_agent_job_input');
  }
  const outputType = String(input.outputType ?? '').trim();
  if (!OUTPUT_TYPES.has(outputType)) {
    throw agentJobError('不支持的输出类型', 'invalid_agent_job_input');
  }
  const allowedAssetIds = dedupeStrings(input.allowedAssetIds);
  if (allowedAssetIds.length === 0) {
    throw agentJobError('至少需要授权一个输入资产', 'invalid_agent_job_input');
  }
  return {
    jobType,
    instruction: normalizeInstruction(input.instruction),
    outputType,
    allowedAssetIds,
    idempotencyKey: String(input.idempotencyKey ?? '').trim() || null,
    outputCategoryId: input.outputCategoryId ? String(input.outputCategoryId) : null,
    locale: String(input.locale ?? 'zh-CN').trim() || 'zh-CN',
    timezone: String(input.timezone ?? 'Asia/Shanghai').trim() || 'Asia/Shanghai',
    capabilities: {
      network: Boolean(input.capabilities?.network),
      shell: Boolean(input.capabilities?.shell),
      createPage: input.capabilities?.createPage !== false,
    },
  };
}

function hashJobToken(token) {
  return crypto.createHash('sha256').update(String(token)).digest('hex');
}

function jsonValue(value, fallback) {
  if (value == null) return fallback;
  if (typeof value === 'object') return value;
  try {
    return JSON.parse(String(value));
  } catch {
    return fallback;
  }
}

function jobResponse(row, assets = []) {
  if (!row) return null;
  return {
    id: row.id,
    sessionId: row.session_id ?? null,
    jobType: row.job_type,
    instruction: row.instruction,
    status: row.status,
    outputType: row.output_type,
    outputCategoryId: row.output_category_id,
    outputCategoryCode: row.output_category_code ?? null,
    progress: jsonValue(row.progress_json, { stage: row.status }),
    permissionScope: jsonValue(row.permission_scope, {}),
    userContext: jsonValue(row.user_context_json, {
      locale: 'zh-CN',
      timezone: 'Asia/Shanghai',
    }),
    resultPageId: row.result_page_id ?? null,
    resultAssetId: row.result_asset_id ?? null,
    errorCode: row.error_code ?? null,
    errorMessage: row.error_message ?? null,
    retryable: RETRYABLE_ERROR_CODES.has(row.error_code),
    queuedAt: asNumber(row.queued_at),
    startedAt: row.started_at == null ? null : asNumber(row.started_at),
    completedAt: row.completed_at == null ? null : asNumber(row.completed_at),
    expiresAt: row.expires_at == null ? null : asNumber(row.expires_at),
    assets,
  };
}

function outputContentFormat(outputType, input) {
  if (input.contentFormat === 'html' || outputType === 'html_page') return 'html';
  return 'markdown';
}

function progressPayload(input, fallbackStage) {
  const stage = String(input?.stage ?? fallbackStage ?? '').trim() || fallbackStage || 'running';
  const payload = { stage };
  const message = String(input?.message ?? '').trim();
  if (message) payload.message = message.slice(0, 500);
  return payload;
}

function hasPathAccess(asset) {
  return asset.status !== 'deleted' && asset.status !== 'quarantined' && asset.scan_status !== 'blocked';
}

function verifyTokenHash(expectedHash, token) {
  if (!expectedHash || !token) return false;
  const actualHash = hashJobToken(token);
  const expected = Buffer.from(String(expectedHash), 'hex');
  const actual = Buffer.from(actualHash, 'hex');
  return expected.length === actual.length && crypto.timingSafeEqual(expected, actual);
}

export function createAgentJobService(pool, options = {}) {
  const idFactory = options.idFactory ?? (() => crypto.randomUUID());
  const nowFactory = options.nowFactory ?? (() => Date.now());
  const tokenTtlMs = Number(options.tokenTtlMs ?? 30 * 60 * 1000);
  const maxOutputBytes = Number(options.maxOutputBytes ?? 2 * 1024 * 1024);
  const pageService = options.pageService;
  const storageRoot = path.resolve(options.storageRoot ?? path.join(process.cwd(), 'data', 'mindspace'));

  const absoluteStoragePath = (storageKey) => {
    const resolved = path.resolve(storageRoot, storageKey);
    if (resolved !== storageRoot && !resolved.startsWith(`${storageRoot}${path.sep}`)) {
      throw new Error('存储路径越界');
    }
    return resolved;
  };

  const resolveStoragePathCandidates = (storageKey) => {
    const normalized = String(storageKey ?? '').replace(/\\/g, '/');
    const withoutMd = normalized.endsWith('.md') ? normalized.slice(0, -3) : normalized;
    const match = withoutMd.match(/^users\/([^/]+)\/(assets|pages)\/([^/]+)\/(v\d+)$/);
    if (!match) return [normalized];
    const [, userId, scope, entityId, versionTag] = match;
    return [
      normalized,
      `users/${userId}/${scope}/${entityId}/versions/${versionTag}.md`,
      `users/${userId}/${scope}/${entityId}/versions/${versionTag}`,
    ].filter((candidate, index, list) => list.indexOf(candidate) === index);
  };

  const resolveReadableStoragePath = async (storageKey) => {
    let lastError = null;
    for (const candidate of resolveStoragePathCandidates(storageKey)) {
      const absolutePath = absoluteStoragePath(candidate);
      try {
        await fs.stat(absolutePath);
        return absolutePath;
      } catch (error) {
        if (error?.code === 'ENOENT') {
          lastError = error;
          continue;
        }
        throw error;
      }
    }
    throw lastError ?? Object.assign(new Error('存储文件不存在'), { code: 'storage_not_found' });
  };

  const fetchJobAssets = async (jobId) => {
    const [rows] = await pool.query(
      `SELECT ja.asset_id, ja.asset_version_id, ja.permission, a.display_name, a.mime_type,
              a.status, v.scan_status
       FROM h5_agent_job_assets ja
       JOIN h5_assets a ON a.id = ja.asset_id
       JOIN h5_asset_versions v ON v.id = ja.asset_version_id
       WHERE ja.job_id = ?
       ORDER BY ja.created_at ASC`,
      [jobId],
    );
    return rows.map((row) => ({
      assetId: row.asset_id,
      assetVersionId: row.asset_version_id,
      permission: row.permission,
      displayName: row.display_name,
      mimeType: row.mime_type,
      status: row.status,
      scanStatus: row.scan_status,
    }));
  };

  const fetchJobRow = async (jobId, userId = null) => {
    const params = [jobId];
    let sql =
      `SELECT j.*, c.category_code AS output_category_code
       FROM h5_agent_jobs j
       JOIN h5_space_categories c ON c.id = j.output_category_id
       WHERE j.id = ?`;
    if (userId) {
      sql += ' AND j.user_id = ?';
      params.push(userId);
    }
    sql += ' LIMIT 1';
    const [rows] = await pool.query(sql, params);
    return rows[0] ?? null;
  };

  const createJob = async (userId, input) => {
    const normalized = normalizeJobInput(input);
    if (!pageService) {
      throw agentJobError('页面服务未初始化', 'internal_error');
    }
    if (normalized.idempotencyKey) {
      const [existingRows] = await pool.query(
        `SELECT j.*, c.category_code AS output_category_code
         FROM h5_agent_jobs j
         JOIN h5_space_categories c ON c.id = j.output_category_id
         WHERE j.user_id = ? AND j.idempotency_key = ?
         LIMIT 1`,
        [userId, normalized.idempotencyKey],
      );
      if (existingRows[0]) {
        const assets = await fetchJobAssets(existingRows[0].id);
        return jobResponse(existingRows[0], assets);
      }
    }

    const conn = await pool.getConnection();
    try {
      await conn.beginTransaction();
      const categoryParams = normalized.outputCategoryId
        ? [normalized.outputCategoryId, userId]
        : [userId];
      const [categoryRows] = await conn.query(
        normalized.outputCategoryId
          ? `SELECT id, category_code
             FROM h5_space_categories
             WHERE id = ? AND user_id = ?
             LIMIT 1
             FOR UPDATE`
          : `SELECT id, category_code
             FROM h5_space_categories
             WHERE user_id = ? AND category_code = 'draft'
             LIMIT 1
             FOR UPDATE`,
        categoryParams,
      );
      const category = categoryRows[0];
      if (!category) throw agentJobError('输出分类不存在', 'category_not_found');
      if (!['draft', 'oa'].includes(category.category_code)) {
        throw agentJobError('该分类不允许写入 Agent 输出', 'invalid_agent_job_input');
      }

      const placeholders = normalized.allowedAssetIds.map(() => '?').join(', ');
      const [assetRows] = await conn.query(
        `SELECT a.id, a.current_version_id, a.display_name, a.mime_type, a.status, v.scan_status
         FROM h5_assets a
         JOIN h5_asset_versions v ON v.id = a.current_version_id
         WHERE a.user_id = ? AND a.id IN (${placeholders}) AND a.status <> 'deleted'
         FOR UPDATE`,
        [userId, ...normalized.allowedAssetIds],
      );
      const assetMap = new Map(assetRows.map((row) => [row.id, row]));
      const missingIds = normalized.allowedAssetIds.filter((assetId) => !assetMap.has(assetId));
      if (missingIds.length > 0) {
        throw agentJobError('存在未授权或不存在的资产', 'asset_not_found', {
          assetIds: missingIds,
        });
      }
      const blockedAssets = assetRows.filter((row) => !hasPathAccess(row));
      if (blockedAssets.length > 0) {
        throw agentJobError('有输入资产尚未通过安全检查', 'security_scan_required', {
          assetIds: blockedAssets.map((row) => row.id),
        });
      }

      const jobId = idFactory();
      const now = nowFactory();
      await conn.query(
        `INSERT INTO h5_agent_jobs
         (id, user_id, job_type, instruction, permission_scope, user_context_json,
          output_category_id, output_type, status, idempotency_key, progress_json,
          queued_at, expires_at, updated_at, max_output_bytes)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?)`,
        [
          jobId,
          userId,
          normalized.jobType,
          normalized.instruction,
          JSON.stringify({ capabilities: normalized.capabilities }),
          JSON.stringify({ locale: normalized.locale, timezone: normalized.timezone }),
          category.id,
          normalized.outputType,
          normalized.idempotencyKey,
          JSON.stringify({ stage: 'queued' }),
          now,
          now + tokenTtlMs,
          now,
          maxOutputBytes,
        ],
      );
      for (const assetId of normalized.allowedAssetIds) {
        const asset = assetMap.get(assetId);
        await conn.query(
          `INSERT INTO h5_agent_job_assets
           (id, job_id, asset_id, asset_version_id, permission, created_at)
           VALUES (?, ?, ?, ?, 'read', ?)`,
          [idFactory(), jobId, asset.id, asset.current_version_id, now],
        );
      }
      await conn.commit();
      const created = await fetchJobRow(jobId, userId);
      const assets = await fetchJobAssets(jobId);
      return jobResponse(created, assets);
    } catch (error) {
      await conn.rollback();
      throw error;
    } finally {
      conn.release();
    }
  };

  const getJob = async (userId, jobId) => {
    const row = await fetchJobRow(jobId, userId);
    if (!row) throw agentJobError('任务不存在', 'agent_job_not_found');
    const assets = await fetchJobAssets(jobId);
    return jobResponse(row, assets);
  };

  const listJobs = async (userId, { limit = 20, offset = 0 } = {}) => {
    const safeLimit = Math.min(Math.max(Number(limit) || 20, 1), 100);
    const safeOffset = Math.max(Number(offset) || 0, 0);
    const [countRows] = await pool.query(
      `SELECT COUNT(*) AS total FROM h5_agent_jobs WHERE user_id = ?`,
      [userId],
    );
    const total = Number(countRows[0]?.total ?? 0);
    const [rows] = await pool.query(
      `SELECT j.*, c.category_code AS output_category_code
       FROM h5_agent_jobs j
       JOIN h5_space_categories c ON c.id = j.output_category_id
       WHERE j.user_id = ?
       ORDER BY COALESCE(j.completed_at, j.started_at, j.queued_at) DESC
       LIMIT ${safeLimit} OFFSET ${safeOffset}`,
      [userId],
    );
    const jobs = await Promise.all(
      rows.map(async (row) => jobResponse(row, await fetchJobAssets(row.id))),
    );
    return {
      items: jobs,
      total,
      offset: safeOffset,
      limit: safeLimit,
      hasMore: safeOffset + jobs.length < total,
    };
  };

  const cancelJob = async (userId, jobId) => {
    const conn = await pool.getConnection();
    try {
      await conn.beginTransaction();
      const [rows] = await conn.query(
        `SELECT * FROM h5_agent_jobs
         WHERE id = ? AND user_id = ?
         LIMIT 1
         FOR UPDATE`,
        [jobId, userId],
      );
      const job = rows[0];
      if (!job) throw agentJobError('任务不存在', 'agent_job_not_found');
      if (!ACTIVE_JOB_STATUSES.has(job.status)) {
        throw agentJobError('当前状态不允许取消任务', 'invalid_state_transition');
      }
      const now = nowFactory();
      await conn.query(
        `UPDATE h5_agent_jobs
         SET status = 'cancelled', completed_at = ?, expires_at = ?, updated_at = ?,
             job_token_hash = NULL, heartbeat_at = NULL, progress_json = ?
         WHERE id = ?`,
        [now, now, now, JSON.stringify({ stage: 'cancelled' }), jobId],
      );
      await conn.commit();
      return getJob(userId, jobId);
    } catch (error) {
      await conn.rollback();
      throw error;
    } finally {
      conn.release();
    }
  };

  const retryJob = async (userId, jobId) => {
    const conn = await pool.getConnection();
    try {
      await conn.beginTransaction();
      const [rows] = await conn.query(
        `SELECT * FROM h5_agent_jobs
         WHERE id = ? AND user_id = ?
         LIMIT 1
         FOR UPDATE`,
        [jobId, userId],
      );
      const job = rows[0];
      if (!job) throw agentJobError('任务不存在', 'agent_job_not_found');
      if (!['failed', 'timed_out', 'cancelled'].includes(job.status)) {
        throw agentJobError('当前状态不允许重试任务', 'invalid_state_transition');
      }
      if (job.status === 'failed' && !RETRYABLE_ERROR_CODES.has(job.error_code)) {
        throw agentJobError('该任务错误不可自动重试', 'invalid_state_transition');
      }
      const now = nowFactory();
      await conn.query(
        `UPDATE h5_agent_jobs
         SET status = 'queued', progress_json = ?, started_at = NULL, completed_at = NULL,
             error_code = NULL, error_message = NULL, result_page_id = NULL, result_asset_id = NULL,
             job_token_hash = NULL, heartbeat_at = NULL, expires_at = ?, updated_at = ?
         WHERE id = ?`,
        [JSON.stringify({ stage: 'queued' }), now + tokenTtlMs, now, jobId],
      );
      await conn.commit();
      return getJob(userId, jobId);
    } catch (error) {
      await conn.rollback();
      throw error;
    } finally {
      conn.release();
    }
  };

  // Marks a locked, queued job row as running and returns the claim payload.
  // Caller must hold a row lock on `job` inside `conn`'s open transaction; this
  // commits the transaction.
  const finalizeClaim = async (conn, job) => {
    const jobId = job.id;
    const token = crypto.randomBytes(24).toString('base64url');
    const now = nowFactory();
    await conn.query(
      `UPDATE h5_agent_jobs
       SET status = 'running', started_at = COALESCE(started_at, ?), heartbeat_at = ?,
           updated_at = ?, expires_at = ?, progress_json = ?, job_token_hash = ?
       WHERE id = ?`,
      [
        now,
        now,
        now,
        now + tokenTtlMs,
        JSON.stringify({ stage: 'preparing_files' }),
        hashJobToken(token),
        jobId,
      ],
    );
    await conn.commit();
    const assets = await fetchJobAssets(jobId);
    const permissionScope = jsonValue(job.permission_scope, {});
    const userContext = jsonValue(job.user_context_json, {
      locale: 'zh-CN',
      timezone: 'Asia/Shanghai',
    });
    return {
      jobId,
      userId: job.user_id,
      jobToken: token,
      instruction: job.instruction,
      userContext,
      allowedAssets: assets.map((asset) => ({
        assetId: asset.assetId,
        versionId: asset.assetVersionId,
        permission: asset.permission,
        displayName: asset.displayName,
        mimeType: asset.mimeType,
        downloadEndpoint: `/api/internal/agent/jobs/${jobId}/assets/${asset.assetId}`,
      })),
      output: {
        categoryId: job.output_category_id,
        categoryCode: job.output_category_code,
        allowedTypes: [job.output_type],
        maxBytes: asNumber(job.max_output_bytes),
      },
      capabilities: permissionScope.capabilities ?? {
        network: false,
        shell: false,
        createPage: true,
      },
      expiresAt: now + tokenTtlMs,
    };
  };

  // Atomically claim the oldest queued, unexpired job. Uses SKIP LOCKED so that
  // multiple worker loops (across portal instances) pull distinct jobs without
  // blocking each other. Returns null when the queue is empty.
  const claimNextJob = async () => {
    const conn = await pool.getConnection();
    try {
      await conn.beginTransaction();
      const [rows] = await conn.query(
        `SELECT j.*, c.category_code AS output_category_code
         FROM h5_agent_jobs j
         JOIN h5_space_categories c ON c.id = j.output_category_id
         WHERE j.status = 'queued'
           AND (j.expires_at IS NULL OR j.expires_at >= ?)
         ORDER BY j.queued_at ASC
         LIMIT 1
         FOR UPDATE SKIP LOCKED`,
        [nowFactory()],
      );
      const job = rows[0];
      if (!job) {
        await conn.commit();
        return null;
      }
      return await finalizeClaim(conn, job);
    } catch (error) {
      await conn.rollback();
      throw error;
    } finally {
      conn.release();
    }
  };

  const claimJob = async (jobId) => {
    const conn = await pool.getConnection();
    try {
      await conn.beginTransaction();
      const [rows] = await conn.query(
        `SELECT j.*, c.category_code AS output_category_code
         FROM h5_agent_jobs j
         JOIN h5_space_categories c ON c.id = j.output_category_id
         WHERE j.id = ?
         LIMIT 1
         FOR UPDATE`,
        [jobId],
      );
      const job = rows[0];
      if (!job) throw agentJobError('任务不存在', 'agent_job_not_found');
      if (job.status !== 'queued') {
        throw agentJobError('任务当前不可领取', 'invalid_state_transition');
      }
      if (job.expires_at != null && asNumber(job.expires_at) < nowFactory()) {
        throw agentJobError('任务已过期', 'agent_job_expired');
      }
      return await finalizeClaim(conn, job);
    } catch (error) {
      await conn.rollback();
      throw error;
    } finally {
      conn.release();
    }
  };


  const requireJobToken = async (jobId, token) => {
    const [rows] = await pool.query(
      `SELECT j.*, c.category_code AS output_category_code
       FROM h5_agent_jobs j
       JOIN h5_space_categories c ON c.id = j.output_category_id
       WHERE j.id = ?
       LIMIT 1`,
      [jobId],
    );
    const job = rows[0];
    if (!job) throw agentJobError('任务不存在', 'agent_job_not_found');
    if (job.status !== 'running') {
      throw agentJobError('任务未处于执行中状态', 'invalid_state_transition');
    }
    if (job.expires_at != null && asNumber(job.expires_at) < nowFactory()) {
      throw agentJobError('任务 token 已过期', 'agent_job_expired');
    }
    if (!verifyTokenHash(job.job_token_hash, token)) {
      throw agentJobError('任务 token 无效', 'agent_job_token_invalid');
    }
    return job;
  };

  const getAssetForJob = async (jobId, token, assetId) => {
    await requireJobToken(jobId, token);
    const [rows] = await pool.query(
      `SELECT a.id, a.display_name, a.mime_type, a.status, v.id AS asset_version_id,
              v.storage_key, v.scan_status
       FROM h5_agent_job_assets ja
       JOIN h5_assets a ON a.id = ja.asset_id
       JOIN h5_asset_versions v ON v.id = ja.asset_version_id
       WHERE ja.job_id = ? AND ja.asset_id = ?
       LIMIT 1`,
      [jobId, assetId],
    );
    const asset = rows[0];
    if (!asset) throw agentJobError('任务资产不存在', 'asset_not_found');
    if (!hasPathAccess(asset)) {
      throw agentJobError('任务资产尚未通过安全检查', 'security_scan_required');
    }
    return {
      assetId: asset.id,
      assetVersionId: asset.asset_version_id,
      displayName: asset.display_name,
      mimeType: asset.mime_type,
      path: await resolveReadableStoragePath(asset.storage_key),
    };
  };

  const heartbeat = async (jobId, token, input) => {
    const job = await requireJobToken(jobId, token);
    const now = nowFactory();
    const payload = progressPayload(input, 'running');
    await pool.query(
      `UPDATE h5_agent_jobs
       SET heartbeat_at = ?, updated_at = ?, expires_at = ?, progress_json = ?
       WHERE id = ?`,
      [now, now, now + tokenTtlMs, JSON.stringify(payload), jobId],
    );
    return jobResponse({ ...job, progress_json: JSON.stringify(payload), heartbeat_at: now });
  };

  const setSessionId = async (jobId, token, sessionId) => {
    await requireJobToken(jobId, token);
    const normalizedSessionId = String(sessionId ?? '').trim();
    if (!normalizedSessionId) {
      throw agentJobError('Agent 会话不能为空', 'invalid_agent_job_input');
    }
    await pool.query(
      `UPDATE h5_agent_jobs
       SET session_id = ?, updated_at = ?
       WHERE id = ?`,
      [normalizedSessionId, nowFactory(), jobId],
    );
  };

  const completeJob = async (jobId, token, input) => {
    const job = await requireJobToken(jobId, token);
    const now = nowFactory();
    if (input?.status === 'failed') {
      const errorCode = String(input.errorCode ?? 'worker_crashed').trim() || 'worker_crashed';
      const errorMessage = String(input.errorMessage ?? '任务执行失败').trim() || '任务执行失败';
      await pool.query(
        `UPDATE h5_agent_jobs
         SET status = 'failed', error_code = ?, error_message = ?, completed_at = ?,
             updated_at = ?, heartbeat_at = NULL, job_token_hash = NULL, progress_json = ?
         WHERE id = ?`,
        [errorCode, errorMessage.slice(0, 1000), now, now, JSON.stringify({ stage: 'failed' }), jobId],
      );
      return getJob(job.user_id, jobId);
    }

    const content = String(input?.content ?? '');
    const outputBytes = Buffer.byteLength(content, 'utf8');
    if (!content.trim()) {
      throw agentJobError('任务输出内容不能为空', 'invalid_agent_job_output');
    }
    if (outputBytes > asNumber(job.max_output_bytes)) {
      throw agentJobError('任务输出超过大小限制', 'page_content_too_large', {
        maxBytes: asNumber(job.max_output_bytes),
      });
    }
    const sourceAssetIds = dedupeStrings(input?.sourceAssetIds);
    const boundAssets = await fetchJobAssets(jobId);
    const boundAssetIdSet = new Set(boundAssets.map((asset) => asset.assetId));
    const invalidSourceIds = sourceAssetIds.filter((assetId) => !boundAssetIdSet.has(assetId));
    if (invalidSourceIds.length > 0) {
      throw agentJobError('输出引用了未授权的输入资产', 'invalid_agent_job_output', {
        assetIds: invalidSourceIds,
      });
    }

    const page = await pageService.createFromAgent(
      job.user_id,
      {
        title: input?.title,
        summary: input?.summary,
        content,
        contentFormat: outputContentFormat(job.output_type, input),
        pageType: input?.pageType,
        templateId: input?.templateId,
        categoryCode: job.output_category_code,
        changeNote: `Agent job ${jobId}`,
      },
      {
        jobId,
        assetId: sourceAssetIds[0] ?? null,
        assetIds: sourceAssetIds,
      },
    );
    await pool.query(
      `UPDATE h5_agent_jobs
       SET status = 'completed', result_page_id = ?, completed_at = ?, updated_at = ?,
           heartbeat_at = NULL, job_token_hash = NULL, progress_json = ?
       WHERE id = ?`,
      [page.id, now, now, JSON.stringify({ stage: 'completed' }), jobId],
    );
    return getJob(job.user_id, jobId);
  };

  // Restart resilience: a job left in 'running' by a crashed/restarted worker
  // stops heartbeating. Mark such jobs failed with the retryable worker_crashed
  // code so they surface to the user (and can be retried) instead of being stuck
  // forever. We fail rather than auto-requeue to avoid poison-job loops, since
  // there is no per-job attempt counter.
  const reapStaleJobs = async (maxStaleMs = 5 * 60 * 1000) => {
    const now = nowFactory();
    const cutoff = now - Math.max(1, maxStaleMs);
    const [result] = await pool.query(
      `UPDATE h5_agent_jobs
       SET status = 'failed', error_code = 'worker_crashed',
           error_message = 'worker heartbeat timed out',
           completed_at = ?, updated_at = ?, heartbeat_at = NULL, job_token_hash = NULL,
           progress_json = ?
       WHERE status = 'running'
         AND heartbeat_at IS NOT NULL
         AND heartbeat_at < ?`,
      [now, now, JSON.stringify({ stage: 'failed' }), cutoff],
    );
    return result?.affectedRows ?? 0;
  };

  return {
    createJob,
    getJob,
    listJobs,
    cancelJob,
    retryJob,
    claimJob,
    claimNextJob,
    reapStaleJobs,
    getAssetForJob,
    heartbeat,
    setSessionId,
    completeJob,
  };
}

export const agentJobInternals = {
  normalizeJobInput,
  hashJobToken,
  verifyTokenHash,
  progressPayload,
};