/**
 * ZeTa-AI Background Job Queue
 *
 * Postgres-backed durable queue. Each job becomes a PlatformJob record.
 * enqueueJob() writes it QUEUED and, for same-process responsiveness, kicks
 * an immediate best-effort claim attempt — but durability comes from
 * startJobWorkerLoop()'s poll: it atomically claims QUEUED jobs (Postgres
 * `FOR UPDATE SKIP LOCKED`, safe across any number of concurrent worker
 * processes/containers) and requeues jobs whose heartbeat went stale,
 * so a crashed/restarted worker never silently loses a job.
 *
 * CRAWL-type jobs additionally support AWS SQS as an alternate transport
 * (JOB_QUEUE_BACKEND=sqs, see sqs-queue.ts) for horizontally-scaled crawler
 * fleets; every other job type always goes through the DB-backed queue.
 */

import crypto from 'node:crypto';
import { withTenant, prismaAdmin } from '@detiq/database';
import { crawlJobsTotal, crawlJobDurationSeconds } from '@detiq/core';
import { logger } from './logger.js';

async function notifyAdmins(tenantId: string, data: {
  type: string; title: string; body: string; metadata?: Record<string, unknown>
}) {
  try {
    // Notification/Membership both have FORCE ROW LEVEL SECURITY — a bare
    // prisma call with no app.current_tenant set gets silently filtered to
    // zero rows (SELECT) or rejected (INSERT), not an error, so this call
    // site had been a complete no-op with no visible failure until fixed.
    const members = await withTenant(tenantId, (tx) =>
      tx.membership.findMany({
        where: { tenantId, role: { in: ['OWNER', 'ADMIN'] } },
        select: { userId: true },
      })
    ) as any[];
    await withTenant(tenantId, (tx) =>
      tx.notification.createMany({
        data: members.map((m: { userId: string }) => ({
          id: `notif_${crypto.randomUUID()}`,
          tenantId,
          userId: m.userId,
          type: data.type,
          title: data.title,
          body: data.body,
          metadata: data.metadata as any,
        })),
      })
    );
  } catch (e) {
    console.error('[notify] failed:', e);
  }
}
import { AppBrain, getProjectStorageConfig, buildArtifactKey, uploadArtifact, getTenantStorageConfig, uploadScreenshotToTenantStorage, uploadFileToTenantStorage, notifyRunCompletion, recordLLMCall } from '@detiq/app-brain';
import { generateTaxonomySkillFiles, groupScreensWithAI, generateExecutionPlan, refreshTestCasesSkillFile, refreshTestDataSkillFile, refreshTestScriptSkillFile, refreshRequirementsCoverageSkillFile, refreshTestExecutionHistorySkillFile, refreshVisualRegressionHistorySkillFile, refreshLocatorStrategyWithHealingData, refreshFieldCatalogSkillFile, detectFieldsFromHtml, mergeIntoCatalog } from '@detiq/agents';
import type { ExecutionStep } from '@detiq/agents';
import { crawlProject, domFallbackElements } from './crawler.js';
import { httpCrawl } from './http-engine.js';
import { hashContent, normalizeText } from './incremental-diff.js';
import { discoverProjectUrls } from './sitemap.js';
import { executeExecutionStep, executeTestPlan } from './test-executor.js';
import { runNativeCrawl } from './native-crawler.js';
import { elementsToPlaywrightLocators } from './locator-extractor.js';
import { jira, confluence, figma } from '@detiq/connectors';
import { decrypt, config as platformConfig, dispatchWebhook, isFeatureEnabled, interpolatePlaceholders } from '@detiq/core';
import { createHmac } from 'node:crypto';
import fs from 'node:fs/promises';
import path from 'node:path';
import os from 'node:os';

// RFC 6238 TOTP — no external dependency, uses Node built-in crypto.
export function generateTOTP(secret: string): string {
  const BASE32 = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ234567';
  const clean = secret.toUpperCase().replace(/\s/g, '').replace(/=/g, '');
  let bits = 0, value = 0;
  const bytes: number[] = [];
  for (const ch of clean) {
    const idx = BASE32.indexOf(ch);
    if (idx === -1) continue;
    value = (value << 5) | idx;
    bits += 5;
    if (bits >= 8) { bits -= 8; bytes.push((value >> bits) & 0xff); }
  }
  const counter = Math.floor(Date.now() / 30000);
  const buf = Buffer.alloc(8);
  buf.writeBigInt64BE(BigInt(counter));
  const hmac = createHmac('sha1', Buffer.from(bytes)).update(buf).digest();
  const offset = hmac[hmac.length - 1] & 0xf;
  const code = ((hmac[offset] & 0x7f) << 24) | ((hmac[offset + 1] & 0xff) << 16) |
    ((hmac[offset + 2] & 0xff) << 8) | (hmac[offset + 3] & 0xff);
  return String(code % 1_000_000).padStart(6, '0');
}

function deriveFallbackTargetPath(ref: string, framework: string): string {
  const slug = (ref ?? 'test').toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-|-$/g, '') || 'test';
  const base = framework.startsWith('cypress') ? 'cypress/e2e/'
    : (framework === 'playwright-py' || framework === 'pytest' || framework === 'selenium-py') ? 'tests/'
      : (framework === 'playwright-java' || framework === 'selenium-java' || framework === 'junit') ? 'src/test/java/'
        : 'tests/';
  const ext = (framework === 'playwright-py' || framework === 'pytest' || framework === 'selenium-py') ? '_test.py'
    : (framework === 'playwright-java' || framework === 'selenium-java' || framework === 'junit') ? 'Test.java'
      : framework.includes('ts') ? '.spec.ts'
        : '.spec.js';
  return `${base}${slug}${ext}`;
}

function makeJobName(type: string): string {
  const now = new Date();
  const date = now.toISOString().slice(0, 10); // YYYY-MM-DD
  const ts = now.getTime();
  const label = type === 'CRAWL' ? 'Crawl'
    : type === 'CAPTURE_STEPS' ? 'CaptureSteps'
      : type === 'NATIVE_CRAWL' ? 'NativeCrawl'
        : type === 'GENERATE_TESTS' ? 'GenerateTests'
          : type === 'VISUAL_REGRESSION' ? 'VisualRegression'
            : type === 'AI_EDIT' ? 'AIEdit'
              : type === 'API_TESTS' ? 'ApiTests'
                : type === 'SECURITY_TESTS' ? 'SecurityTests'
                  : type === 'INTEGRATION_TESTS' ? 'IntegrationTests'
                    : type === 'GENERATE_TESTS_FROM_REQS' ? 'GenerateTestsFromReqs'
                      : type === 'GENERATE_DATA' ? 'GenerateData'
                        : type === 'GENERATE_SCRIPT' ? 'GenerateScript'
                          : type === 'GENERATE_TEST_PLAN' ? 'GenerateTestPlan'
                            : type === 'IMPORT_TESTS' ? 'ImportTests'
                              : type === 'GENERATE_SCREEN_TESTS' ? 'GenerateScreenTests'
                                : type === 'BATCH_GENERATE_ALIGNED' ? 'BatchGenerateAligned'
                                  : type === 'CREATE_PR' ? 'CreatePR'
                                    : type === 'NEW_PAGES_PIPELINE' ? 'NewPagesPipeline'
                                      : type.charAt(0) + type.slice(1).toLowerCase().replace(/_([a-z])/g, (_, c) => c.toUpperCase());
  return `${label}_${date}_${ts}`;
}

export interface PlatformJobConfig {
  tenantId: string;
  projectId: string;
  appUrl?: string; // Optional since generate jobs don't strictly need it
  credentials?: { username: string; password: string; oneTimeCode?: string };
  /** Decrypted Playwright storageState JSON — if present, session is restored without AI login. */
  storageState?: string;
  screenshotDir?: string;
  maxScreens?: number;
  /** Seed URLs added to the initial crawl queue — link discovery still runs from each page. */
  startUrls?: string[];
  /** When provided, crawl ONLY these exact URLs (no link-following). Used for selected-screen recrawl. */
  selectiveUrls?: string[];
  /** Device profiles for multi-viewport screenshot pass after main DESKTOP crawl. */
  deviceProfiles?: string[];
  /** HTTP Basic Auth credentials — injected as Authorization header on same-origin requests. */
  httpBasicUsername?: string;
  httpBasicPassword?: string;
  /** Bearer/API token — injected as a request header and into localStorage. Skips AI login. */
  authToken?: string;
  /** Header name for authToken (default: 'Authorization'). */
  authTokenHeader?: string;
  /** Prefix for authToken value (default: 'Bearer '). */
  authTokenPrefix?: string;
  /** CAPTCHA solving API key (2captcha or CapSolver). */
  captchaSolverApiKey?: string;
  /** CAPTCHA solving service provider (default: '2captcha'). */
  captchaSolverProvider?: '2captcha' | 'capsolver';
  /**
   * Path to a Chrome user data directory for profile inheritance (local deployments only).
   * When set, the crawler exports the existing session from this profile and injects it.
   */
  chromeProfilePath?: string;
  /** MailSlurp API key for email OTP auto-retrieval. */
  mailslurpApiKey?: string;
  /** MailSlurp inbox ID to poll for OTP emails. */
  mailslurpInboxId?: string;
  /** SSO provider ('okta' | 'azure_ad' | 'generic_oidc'). */
  ssoProvider?: string;
  /** SSO domain or tenant ID. */
  ssoDomain?: string;
  /** SSO OAuth2 client ID. */
  ssoClientId?: string;
  /** SSO OAuth2 client secret (decrypted). */
  ssoClientSecret?: string;
  /** SSO scopes (space-separated). Default: 'openid profile'. */
  ssoScope?: string;
  /** HTTP/HTTPS/SOCKS5 proxy URL — routes browser through this proxy for bot-protected sites. */
  proxyUrl?: string;
}

/** In-memory map of jobId → AbortController for RUNNING jobs. */
const runningControllers = new Map<string, AbortController>();

// ─── Durable job worker ──────────────────────────────────────────────────────
//
// A minimal Postgres-backed work queue: atomic claim via `FOR UPDATE SKIP
// LOCKED` (safe for any number of concurrent pollers, in-process or across
// containers), plus a stale-job sweep keyed off a liveness heartbeat rather
// than wall-clock job age, so a slow-but-alive job is never requeued and
// double-processed. See startJobWorkerLoop() below.

const WORKER_ID = `${os.hostname()}:${process.pid}`;
const STALE_JOB_TIMEOUT_MINUTES = Number(process.env.JOB_STALE_TIMEOUT_MINUTES) || 30;
const WORKER_POLL_INTERVAL_MS = Number(process.env.JOB_WORKER_POLL_INTERVAL_MS) || 15_000;
const WORKER_CONCURRENCY = Number(process.env.JOB_WORKER_CONCURRENCY) || 8;
// Max simultaneous DISCOVER jobs — each launches a Playwright browser and holds
// DB connections, so running many concurrently exhausts resources. Default 2.
const MAX_CONCURRENT_DISCOVER = Number(process.env.MAX_CONCURRENT_DISCOVER) || 2;
const JOB_HEARTBEAT_INTERVAL_MS = Number(process.env.JOB_HEARTBEAT_INTERVAL_MS) || 15_000;
// How many headless browsers a single RUN_TEST_SUITE job launches at once —
// distinct from WORKER_CONCURRENCY (how many *jobs* the poll loop claims at
// once). Bounded by default to avoid exhausting a container's memory; the
// primary scale-out path for running more tests at once is still horizontal
// (more worker containers / SQS), not raising this past what one container
// can safely hold — see startJobWorkerLoop's doc comment.
const TEST_EXEC_CONCURRENCY = Number(process.env.TEST_EXEC_CONCURRENCY) || 5;

type ClaimedJob = { id: string; tenantId: string; projectId: string; type: string; config: any };

/** Dispatch table populated at module bottom, once every process*Job function is declared. */
let jobDispatchTable: Record<string, (jobId: string, config: any) => Promise<void>> | null = null;

/** Atomically claims one specific QUEUED job for this worker. Returns null if it was already claimed. */
export async function claimSpecificJob(jobId: string): Promise<ClaimedJob | null> {
  const rows = await prismaAdmin.$queryRaw<ClaimedJob[]>`
    UPDATE "PlatformJob"
    SET status = 'RUNNING', "startedAt" = NOW(), "heartbeatAt" = NOW(), "workerId" = ${WORKER_ID}
    WHERE id = ${jobId} AND status = 'QUEUED'
    RETURNING id, "tenantId" AS "tenantId", "projectId" AS "projectId", type, config
  `;
  return rows[0] ?? null;
}

/** Atomically claims the oldest QUEUED job across all tenants. SKIP LOCKED makes this
 *  race-safe against any number of other pollers doing the same thing concurrently.
 *  DISCOVER jobs are subject to MAX_CONCURRENT_DISCOVER — skipped when that many are
 *  already RUNNING to prevent Playwright/DB pool exhaustion from simultaneous crawls. */
export async function claimNextQueuedJob(): Promise<ClaimedJob | null> {
  const maxDiscover = MAX_CONCURRENT_DISCOVER;
  const rows = await prismaAdmin.$queryRaw<ClaimedJob[]>`
    UPDATE "PlatformJob"
    SET status = 'RUNNING', "startedAt" = NOW(), "heartbeatAt" = NOW(), "workerId" = ${WORKER_ID}
    WHERE id = (
      SELECT id FROM "PlatformJob"
      WHERE status = 'QUEUED'
        AND (
          type != 'DISCOVER'
          OR (SELECT COUNT(*) FROM "PlatformJob" WHERE status = 'RUNNING' AND type = 'DISCOVER') < ${maxDiscover}
        )
      ORDER BY "queuedAt" ASC
      FOR UPDATE SKIP LOCKED
      LIMIT 1
    )
    RETURNING id, "tenantId" AS "tenantId", "projectId" AS "projectId", type, config
  `;
  return rows[0] ?? null;
}

/** Requeues RUNNING jobs whose heartbeat has gone stale — the worker that claimed them
 *  died (crash, restart, OOM) without ever marking them DONE/FAILED. Idempotent and
 *  safe to run from every worker process; only affects rows past the timeout.
 *  API-in-process job types (GENERATE_SCRIPT etc.) are failed directly — requeueing them
 *  would cause the worker to claim and fail them with a confusing "claimed by worker" error. */
export async function requeueStaleRunningJobs(): Promise<void> {
  // Directly fail stale API-in-process jobs — worker cannot process these types.
  const failedRows = await prismaAdmin.$queryRaw<Array<{ id: string }>>`
    UPDATE "PlatformJob"
    SET status = 'FAILED',
        "finishedAt" = NOW(),
        error = 'API process was restarted while this job was running — please retry from the UI.'
    WHERE status = 'RUNNING'
      AND type IN ('GENERATE_SCRIPT', 'GENERATE_DATA', 'GENERATE_TESTS', 'GENERATE_TEST_PLAN')
      AND COALESCE("heartbeatAt", "startedAt") < NOW() - (${STALE_JOB_TIMEOUT_MINUTES} * interval '1 minute')
    RETURNING id
  `;
  if (failedRows.length > 0) {
    logger.warn({ jobIds: failedRows.map((r: any) => r.id), staleTimeoutMinutes: STALE_JOB_TIMEOUT_MINUTES }, '[job-worker] failed stale API-in-process job(s) — API restarted while running');
  }

  const rows = await prismaAdmin.$queryRaw<Array<{ id: string }>>`
    UPDATE "PlatformJob"
    SET status = 'QUEUED',
        "workerId" = NULL,
        result = jsonb_set(
          COALESCE(result::jsonb, '{}'::jsonb),
          '{step}',
          to_jsonb('Worker heartbeat lost — queued for recovery'::text),
          true
        )
    WHERE status = 'RUNNING'
      AND type NOT IN ('GENERATE_SCRIPT', 'GENERATE_DATA', 'GENERATE_TESTS', 'GENERATE_TEST_PLAN')
      AND COALESCE("heartbeatAt", "startedAt") < NOW() - (${STALE_JOB_TIMEOUT_MINUTES} * interval '1 minute')
    RETURNING id
  `;
  if (rows.length > 0) {
    logger.warn({ jobIds: rows.map((r: any) => r.id), staleTimeoutMinutes: STALE_JOB_TIMEOUT_MINUTES }, '[job-worker] requeued stale RUNNING job(s)');
  }
}

/**
 * Requeues any RUNNING jobs claimed by a different worker (i.e., a previous
 * process instance). Call once at startup before the poll loop begins — this
 * provides immediate recovery instead of waiting the full STALE_JOB_TIMEOUT_MINUTES.
 * Safe to call from multiple replicas simultaneously (each uses its own WORKER_ID).
 */
export async function requeueOrphanedJobs(): Promise<void> {
  const rows = await prismaAdmin.$queryRaw<Array<{ id: string }>>`
    UPDATE "PlatformJob"
    SET status = 'QUEUED',
        "workerId" = NULL,
        result = jsonb_set(
          COALESCE(result::jsonb, '{}'::jsonb),
          '{step}',
          to_jsonb('Worker restarted — queued for recovery'::text),
          true
        )
    WHERE status = 'RUNNING'
      AND "workerId" IS NOT NULL
      AND "workerId" != ${WORKER_ID}
    RETURNING id
  `;
  if (rows.length > 0) {
    logger.warn({ jobIds: rows.map((r: any) => r.id), workerId: WORKER_ID }, '[job-worker] requeued orphaned RUNNING job(s) from dead worker');
  }
}

/** Dispatches a claimed job to its type's processor. Last-resort safety net — every
 *  process*Job function already catches and records its own failures. */
async function dispatchClaimedJob(job: ClaimedJob): Promise<void> {
  const dispatcher = jobDispatchTable?.[job.type];
  if (!dispatcher) {
    logger.error({ jobId: job.id, type: job.type }, '[job-worker] no dispatcher registered for job type');
    await prismaAdmin.platformJob.update({
      where: { id: job.id },
      data: { status: 'FAILED', error: `Unknown job type: ${job.type}`, finishedAt: new Date() },
    }).catch(() => { });
    return;
  }
  const stopTimer = crawlJobDurationSeconds.startTimer({ type: job.type });
  try {
    await dispatcher(job.id, job.config);
    stopTimer();
    // dispatcher() resolving doesn't mean the job succeeded — every process*Job
    // function catches and records its own failures internally without
    // rethrowing (see this function's doc comment). Read back the actual
    // final status so the metric reflects reality, not just "didn't throw".
    const finished = await prismaAdmin.platformJob.findUnique({ where: { id: job.id }, select: { status: true } }).catch(() => null);
    crawlJobsTotal.inc({ type: job.type, status: finished?.status === 'FAILED' ? 'failed' : 'success' });
  } catch (err: any) {
    stopTimer();
    crawlJobsTotal.inc({ type: job.type, status: 'failed' });
    logger.error({ jobId: job.id, type: job.type, err: String(err?.message ?? err) }, '[job-worker] job threw uncaught error');
    await prismaAdmin.platformJob.update({
      where: { id: job.id },
      data: { status: 'FAILED', error: String(err?.message ?? err), finishedAt: new Date() },
    }).catch(() => { });
  }
}

let workerLoopTimer: ReturnType<typeof setInterval> | null = null;

/**
 * Starts the durable job worker poll loop. This is what makes job processing
 * survive a restart — enqueueJob()'s immediate same-process claim attempt is
 * only a latency optimization; if the process dies before that fires (or
 * mid-job), this loop's next tick claims/requeues it. Call once at process
 * startup (API server and/or a dedicated crawler worker process — safe to
 * run in both, and safe to run in many container replicas at once).
 */
export function startJobWorkerLoop(opts?: { pollIntervalMs?: number; concurrency?: number }): void {
  if (workerLoopTimer) return; // already running in this process
  const pollIntervalMs = opts?.pollIntervalMs ?? WORKER_POLL_INTERVAL_MS;
  const concurrency = opts?.concurrency ?? WORKER_CONCURRENCY;
  logger.info({ workerId: WORKER_ID, pollIntervalMs, concurrency }, '[job-worker] starting');

  const tick = async () => {
    try {
      await requeueStaleRunningJobs();
    } catch (err) {
      logger.error({ err: String(err) }, '[job-worker] stale-job sweep failed');
    }
    try {
      const pauseRow = await prismaAdmin.systemConfig.findUnique({ where: { key: 'queue.paused' } });
      if (pauseRow?.value === 'true') {
        logger.info('[job-worker] queue is paused — skipping claim');
      } else {
        for (let i = 0; i < concurrency; i++) {
          const job = await claimNextQueuedJob();
          if (!job) break;
          void dispatchClaimedJob(job); // not awaited — runs concurrently while the loop keeps polling
        }
      }
    } catch (err) {
      logger.error({ err: String(err) }, '[job-worker] claim loop failed');
    }
  };

  workerLoopTimer = setInterval(tick, pollIntervalMs);
  // On startup: immediately requeue jobs orphaned by the previous process instance.
  // This gives instant recovery rather than waiting STALE_JOB_TIMEOUT_MINUTES (30 min).
  requeueOrphanedJobs().catch((err) =>
    logger.error({ err: String(err) }, '[job-worker] startup orphan sweep failed')
  );
  void tick();
}

/** Stops the poll loop. Mainly for tests. */
export function stopJobWorkerLoop(): void {
  if (workerLoopTimer) clearInterval(workerLoopTimer);
  workerLoopTimer = null;
}

/** Create a job record that tracks a synchronous operation (telemetry only, no background processor). */
export async function createTrackedJob(
  type: string,
  config: { tenantId: string; projectId: string;[k: string]: any }
): Promise<string> {
  const name = makeJobName(type);
  const job = await withTenant(config.tenantId, (tx) =>
    tx.platformJob.create({
      data: { tenantId: config.tenantId, projectId: config.projectId, type, name, status: 'RUNNING', startedAt: new Date(), config: config as any },
    })
  ) as { id: string };
  return job.id;
}

/** Mark a tracked job DONE or FAILED. Safe to call in finally — swallows errors.
 * Gap 30: When marking FAILED, sets nextRetryAt (1 min) so the retry worker picks it up promptly. */
export async function finishTrackedJob(tenantId: string, jobId: string, error?: string): Promise<void> {
  const nextRetryAt = error ? new Date(Date.now() + 60_000) : undefined;
  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({
      where: { id: jobId },
      data: {
        status: error ? 'FAILED' : 'DONE',
        finishedAt: new Date(),
        ...(error ? { error, nextRetryAt } : { progress: 100 }),
      },
    })
  ).catch(() => { });
}

/** Enqueue a generic platform job — returns the job ID immediately. */
export async function enqueueJob(type: string, config: any): Promise<string> {
  const { tenantId, projectId } = config;

  const name = makeJobName(type);
  const job = await withTenant(tenantId, (tx) =>
    tx.platformJob.create({
      data: {
        tenantId,
        projectId,
        type,
        name,
        status: 'QUEUED',
        config: config as any,
      },
    })
  ) as { id: string };

  // Durability comes from startJobWorkerLoop()'s poll — this is only a same-process
  // latency optimization so a job doesn't sit idle until the next poll tick. The
  // claim is atomic (WHERE status='QUEUED'), so it's safe even if the poll loop's
  // own tick fires concurrently and claims this same job first.
  if (type === 'CRAWL' && process.env.JOB_QUEUE_BACKEND === 'sqs') {
    const { sendJobToSqs } = await import('./sqs-queue.js');
    await sendJobToSqs(job.id, tenantId, config);
  } else if (process.env.DISABLE_WORKER !== 'true') {
    // Skip inline dispatch when DISABLE_WORKER=true — the dedicated crawler process
    // will pick this up via its poll loop. Inline dispatch in the API container
    // would launch Playwright browsers and crash the API.
    setImmediate(() => claimSpecificJob(job.id).then((claimed) => claimed && dispatchClaimedJob(claimed)));
  }

  return job.id;
}

/** Get the current state of a platform job. */
export async function getJob(tenantId: string, jobId: string) {
  return withTenant(tenantId, (tx) =>
    tx.platformJob.findFirst({ where: { id: jobId, tenantId } })
  );
}

/** List recent jobs for a project. */
export async function listJobs(tenantId: string, projectId: string) {
  return withTenant(tenantId, async (tx) => {
    const jobs = await tx.platformJob.findMany({
      where: { tenantId, projectId },
      orderBy: { queuedAt: 'desc' },
      take: 50,
    });
    // Attach the most-recently-used model for each job from LLMCallLog
    const jobIds = jobs.map((j) => j.id);
    if (jobIds.length === 0) return jobs;
    const modelRows = await (tx as any).lLMCallLog.groupBy({
      by: ['jobId'],
      where: { tenantId, jobId: { in: jobIds } },
      _max: { model: true, calledAt: true },
    });
    const modelByJobId = new Map<string, string>(
      modelRows
        .filter((r: any) => r.jobId && r._max?.model)
        .map((r: any) => [r.jobId as string, r._max.model as string])
    );
    return jobs.map((j) => ({ ...j, llmModel: modelByJobId.get(j.id) ?? null }));
  });
}

/** Cancel a RUNNING or QUEUED job. Aborts the in-process crawl and marks DB CANCELLED. */
export async function cancelJob(tenantId: string, jobId: string): Promise<void> {
  // Abort in-memory controller if job is running in the same process
  const controller = runningControllers.get(jobId);
  if (controller) {
    controller.abort();
    runningControllers.delete(jobId);
  }
  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({
      where: { id: jobId },
      data: { status: 'CANCELLED', finishedAt: new Date(), heartbeatAt: null, workerId: null },
    })
  );
}

/** Re-enqueue a job using the same config as a previous job. */
export async function rerunJob(tenantId: string, jobId: string): Promise<string> {
  const job = await withTenant(tenantId, (tx) =>
    tx.platformJob.findUnique({ where: { id: jobId } })
  ) as { type: string; config: any } | null;
  if (!job) throw new Error('Job not found');
  return enqueueJob(job.type, { ...(job.config as any), tenantId });
}

// In-memory checkpoint store: updated every 10 screens, saved to DB result on failure.
const jobCheckpoints = new Map<string, { visitedUrls: string[]; pendingQueue: string[] }>();

export async function resumeJob(tenantId: string, jobId: string): Promise<string> {
  const job = await withTenant(tenantId, (tx) =>
    tx.platformJob.findUnique({ where: { id: jobId } })
  ) as { type: string; config: any; result: any } | null;
  if (!job) throw new Error('Job not found');

  const checkpoint = (job.result as any)?.checkpoint;
  const visitedUrls: string[] = checkpoint?.visitedUrls ?? [];
  const pendingQueue: string[] = checkpoint?.pendingQueue ?? [];

  if (pendingQueue.length === 0 && visitedUrls.length === 0) {
    // No checkpoint saved — full rerun
    return enqueueJob(job.type, { ...(job.config as any), tenantId });
  }

  return enqueueJob(job.type, {
    ...(job.config as any),
    tenantId,
    skipUrls: visitedUrls.length > 0 ? visitedUrls : undefined,
    resumeUrls: pendingQueue.length > 0 ? pendingQueue : undefined,
  });
}

/** Update progress percentage and current step label on a running job (non-fatal). */
export async function updateJobProgress(tenantId: string, jobId: string, progress: number, step: string) {
  try {
    await withTenant(tenantId, async (tx) => {
      // DB-flag cancellation: works for both inline and SQS backends.
      // When a user cancels, status is set to 'CANCELLED' in DB.
      // This check fires at each progress update (every few seconds during crawl).
      const current = await tx.platformJob.findUnique({
        where: { id: jobId },
        select: { status: true },
      });
      if (current?.status === 'CANCELLED') {
        runningControllers.get(jobId)?.abort();
        runningControllers.delete(jobId);
        return; // abort signal propagates; crawlProject exits on next signal check
      }
      await tx.platformJob.update({
        where: { id: jobId },
        data: { progress, result: { step } as any, heartbeatAt: new Date() },
      });
    });
  } catch { /* non-fatal — progress updates best-effort */ }
}

/** Bump only worker liveness. This intentionally does not touch progress/result. */
async function heartbeatJob(tenantId: string, jobId: string) {
  try {
    await withTenant(tenantId, async (tx) => {
      const current = await tx.platformJob.findUnique({
        where: { id: jobId },
        select: { status: true },
      });
      if (current?.status === 'CANCELLED') {
        runningControllers.get(jobId)?.abort();
        runningControllers.delete(jobId);
        return;
      }
      if (current?.status !== 'RUNNING') return;
      await tx.platformJob.update({
        where: { id: jobId },
        data: { heartbeatAt: new Date(), workerId: WORKER_ID },
      });
    });
  } catch { /* non-fatal — heartbeat updates best-effort */ }
}

/** Background: for each changed screen, search Jira + Confluence for related tickets/pages. */
async function investigateScreenChanges(
  tenantId: string,
  projectId: string,
  changedScreens: Array<{ id: string; name: string; url?: string; addedElements: string[]; removedElements: string[] }>
) {
  const [jiraCfg, confluenceCfg] = await Promise.all([
    AppBrain.getConnectorConfig(tenantId, projectId, 'JIRA'),
    AppBrain.getConnectorConfig(tenantId, projectId, 'CONFLUENCE'),
  ]);

  for (const screen of changedScreens) {
    const searchTerm = screen.url ?? screen.name;
    const changeSummary = `Screen "${screen.name}" changed: +${screen.addedElements.length} elements, -${screen.removedElements.length} elements`;

    const detail: Record<string, any> = {
      screenId: screen.id,
      screenName: screen.name,
      url: screen.url,
      addedElements: screen.addedElements,
      removedElements: screen.removedElements,
      jiraTickets: [],
      confluencePages: [],
    };

    if (jiraCfg?.baseUrl && jiraCfg?.email && jiraCfg?.apiToken) {
      try {
        detail.jiraTickets = await jira.searchIssues(
          { baseUrl: jiraCfg.baseUrl, email: jiraCfg.email, apiToken: jiraCfg.apiToken, projectKey: jiraCfg.projectKey },
          searchTerm
        );
      } catch (e) {
        logger.warn({ searchTerm, err: e }, 'Jira search failed');
      }
    }

    if (confluenceCfg?.baseUrl && confluenceCfg?.email && confluenceCfg?.apiToken) {
      try {
        detail.confluencePages = await confluence.searchPages(
          { baseUrl: confluenceCfg.baseUrl, email: confluenceCfg.email, apiToken: confluenceCfg.apiToken, spaceKey: confluenceCfg.spaceKey },
          searchTerm
        );
      } catch (e) {
        logger.warn({ searchTerm, err: e }, 'Confluence search failed');
      }
    }

    await AppBrain.logChange(tenantId, projectId, 'MODIFIED', changeSummary, detail);
    logger.info({ screen: screen.name, jiraTickets: detail.jiraTickets.length, confluencePages: detail.confluencePages.length }, 'Logged screen change investigation');
  }
}

export interface ProcessJobOpts {
  onProgress?: (pct: number) => Promise<void>;
}

export async function processJob(jobId: string, config: any, opts?: ProcessJobOpts) {
  const { tenantId, projectId } = config;

  // Mark as RUNNING — skip if already CANCELLED (race: user cancelled before worker picked it up).
  const jobStartedAt = new Date();
  const claimed = await withTenant(tenantId, (tx) =>
    tx.platformJob.updateMany({
      where: { id: jobId, status: { not: 'CANCELLED' } },
      data: { status: 'RUNNING', startedAt: jobStartedAt, heartbeatAt: jobStartedAt, workerId: WORKER_ID },
    })
  );
  if (claimed.count === 0) return;

  const controller = new AbortController();
  runningControllers.set(jobId, controller);
  const heartbeatTimer = setInterval(() => {
    void heartbeatJob(tenantId, jobId);
  }, JOB_HEARTBEAT_INTERVAL_MS);

  const onProgress = async (pct: number, step: string) => {
    await updateJobProgress(tenantId, jobId, pct, step);
    await opts?.onProgress?.(pct);
  };

  const DEFAULT_MAX_SCREENS = parseInt(process.env.CRAWL_DEFAULT_MAX_SCREENS ?? '') || 150;

  try {
    // Per-project maxScreens + deviceProfiles + seedUrls + interactive probing + sitemap seeding + traversal strategy from crawlConfig override job config defaults.
    if (!config.maxScreens || !config.deviceProfiles || config.interactive === undefined || config.useSitemap === undefined || !config.traversalStrategy) {
      const proj: any = await withTenant(tenantId, (tx) =>
        (tx as any).project.findUnique({ where: { id: projectId }, select: { crawlConfig: true } })
      );
      const perProject = proj?.crawlConfig?.maxScreens;
      if (!config.maxScreens) config.maxScreens = perProject || DEFAULT_MAX_SCREENS;
      if (!config.deviceProfiles) config.deviceProfiles = proj?.crawlConfig?.deviceProfiles ?? ['DESKTOP'];
      if (config.interactive === undefined) config.interactive = proj?.crawlConfig?.interactive ?? false;
      if (config.useSitemap === undefined) config.useSitemap = proj?.crawlConfig?.useSitemap ?? false;
      if (!config.sitemapUrl) config.sitemapUrl = proj?.crawlConfig?.sitemapUrl ?? undefined;
      if (!config.traversalStrategy) config.traversalStrategy = proj?.crawlConfig?.traversalStrategy ?? 'BFS';
      if (!config.bestFirstKeywords) config.bestFirstKeywords = proj?.crawlConfig?.bestFirstKeywords ?? undefined;
      // segmentSize from crawl config — default unlimited (9_999_999) so normal sites run in one job;
      // only segment if the project explicitly sets a segmentSize in its crawl config.
      if (!config.segmentSize) config.segmentSize = proj?.crawlConfig?.segmentSize ?? 9_999_999;
      // Load saved seedUrls as non-selective seeds (link discovery still runs from them).
      // Skip when a selective recrawl is requested (selectiveUrls set) or seeds already provided.
      if (!config.selectiveUrls?.length && (!config.startUrls || config.startUrls.length === 0)) {
        const saved: string[] = proj?.crawlConfig?.seedUrls ?? [];
        if (saved.length > 0) config.startUrls = saved;
      }
      if (config.interactive === undefined) config.interactive = proj?.crawlConfig?.interactive ?? false;
    }
    // Build runtime credentials: convert stored totpSecret → getOneTimeCode function
    const runtimeCreds = config.credentials
      ? {
        ...config.credentials,
        ...(config.credentials.totpSecret
          ? { getOneTimeCode: async () => generateTOTP(config.credentials.totpSecret) }
          : {}),
      }
      : undefined;

    const storageConfigForCrawl = await getTenantStorageConfig(tenantId).catch(() => null);
    const result = await crawlProject({
      ...config, credentials: runtimeCreds, abortSignal: controller.signal, jobId,
      storageConfig: storageConfigForCrawl,
      onCheckpoint: async (visitedUrls, pendingQueue) => {
        jobCheckpoints.set(jobId, { visitedUrls, pendingQueue });
      },
    }, onProgress);
    jobCheckpoints.delete(jobId);

    const pendingUrls: string[] = (result as any)?.pendingUrls ?? [];
    let nextJobId: string | undefined;
    if (pendingUrls.length > 0) {
      // visitedUrls includes all attempted URLs (screens, soft-404s, API paths) — more complete than
      // discovered-with-screenId only, preventing the continuation job from re-attempting dead ends.
      const visitedForSkip: string[] = (result as any).visitedUrls ?? ((result as any).discovered ?? [])
        .filter((d: any) => !!d.screenId && !!d.url)
        .map((d: any) => { try { const _u = new URL(d.url as string); _u.hash = ''; return _u.href; } catch { return d.url as string; } });
      const reason = (result as any).wasTimeout ? 'timeout' : 'segment-limit';
      logger.info({ pendingUrls: pendingUrls.length, visited: visitedForSkip.length, reason }, '[crawl] auto-enqueuing continuation job');
      nextJobId = await enqueueJob('CRAWL', {
        ...config,
        startUrls: pendingUrls,
        skipUrls: [...(config.skipUrls ?? []), ...visitedForSkip],
      }).catch((autoErr: any) => { logger.warn({ err: autoErr?.message }, '[crawl] auto-segment enqueue failed (non-fatal)'); return undefined; });
    }

    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: {
          status: 'DONE',
          result: { ...(result as any), ...(nextJobId ? { nextJobId } : {}) } as any,
          progress: 100,
          finishedAt: new Date(),
        },
      })
    );
    clearInterval(heartbeatTimer);
    runningControllers.delete(jobId);

    {
      // Auto-seed: merge all successfully crawled URLs into crawlConfig.seedUrls so future
      // crawls and discovery agents start from a complete picture of the app.
      const crawledUrls: string[] = ((result as any).discovered ?? [])
        .filter((d: any) => !!d.screenId && !!d.url)
        .map((d: any) => d.url as string);
      if (crawledUrls.length > 0) {
        (async () => {
          const cfg = await AppBrain.getCrawlConfig(tenantId, projectId);
          const existing: string[] = cfg?.seedUrls ?? [];
          const merged = Array.from(new Set([...existing, ...crawledUrls]));
          if (merged.length > existing.length) {
            await AppBrain.saveCrawlConfig(tenantId, projectId, { seedUrls: merged });
            logger.info({ added: merged.length - existing.length, total: merged.length }, '[crawl] auto-seeded crawled URLs into crawlConfig.seedUrls');
          }
        })().catch((err: any) => logger.warn({ err: err?.message }, '[crawl] auto-seed seedUrls failed (non-fatal)'));
      }

      notifyAdmins(tenantId, {
        type: 'crawl.done',
        title: 'Crawl completed',
        body: `Crawl for project finished successfully.`,
        metadata: { projectId, jobId },
      }).catch(console.error);

      dispatchWebhook(tenantId, projectId, 'crawl.done', {
        jobId,
        screensFound: (result as any)?.screensFound ?? (result as any)?.screens?.length ?? 0,
      }).catch(console.error);
    }

    // Aggregate token usage from LLM calls made during this job.
    // Delay 8s to let concurrent recordLLMCall writes settle before querying LLMCallLog.
    setTimeout(() => {
      aggregateJobTokens(tenantId, jobId, projectId, jobStartedAt).catch(console.error);
    }, 8000);

    // Post-crawl: generate / refresh all taxonomy skill files + investigate changes.
    setImmediate(async () => {
      try {
        const projectMap = await AppBrain.getProjectMap(tenantId, projectId);
        if (projectMap?.screens?.length) {
          // Mark project oversized when > 300 screens — activates VectorSearchGrounding
          const screenCount = (projectMap.screens as any[]).length;
          if (screenCount > 300) {
            await withTenant(tenantId, (tx) =>
              (tx as any).project.update({ where: { id: projectId }, data: { oversized: true } })
            );
          }
          await generateTaxonomySkillFiles(tenantId, projectId, projectMap.screens as any, result.apiEndpoints, result.performanceMetrics);

          // Auto-detect AppField catalog from crawled DOM snapshots.
          // Runs before test generation so getFieldsForCase() returns typed fields (FORM mode).
          // Capped at 20 screens and only where domSnapshotPath exists to bound LLM cost.
          try {
            const { readFile } = await import('node:fs/promises');
            const { existsSync } = await import('node:fs');
            const screensWithDom = (projectMap.screens as any[]).filter((s: any) => s.domSnapshotPath);
            const detectLimit = Math.min(screensWithDom.length, 20);
            let fieldDetectAdded = 0;
            for (let si = 0; si < detectLimit; si++) {
              const screen = screensWithDom[si];
              try {
                let htmlContent: string | null = null;
                const domPath: string = screen.domSnapshotPath;
                if (domPath.startsWith('https://') || domPath.startsWith('http://')) {
                  const res = await fetch(domPath);
                  if (res.ok) htmlContent = await res.text();
                } else if (existsSync(domPath)) {
                  htmlContent = await readFile(domPath, 'utf8');
                }
                if (!htmlContent) continue;
                const detected = await detectFieldsFromHtml({ tenantId, projectId, html: htmlContent });
                if (detected.length > 0) {
                  const merged = await mergeIntoCatalog({ tenantId, projectId, detected });
                  fieldDetectAdded += merged.added;
                }
              } catch (sErr) {
                logger.warn({ screenId: screen.id, err: sErr }, '[field-detect] skipped screen');
              }
            }
            if (fieldDetectAdded > 0 || detectLimit > 0) {
              logger.info({ screens: detectLimit, added: fieldDetectAdded, projectId }, '[field-detect] catalog populated from crawl');
              // Refresh FIELD_CATALOG skill file after catalog is populated (GAP-SF-20)
              refreshFieldCatalogSkillFile(tenantId, projectId).catch(console.error);
            }
          } catch (fdErr) {
            logger.warn({ err: fdErr }, '[field-detect] post-crawl field detection failed');
          }

          // Cloud storage: upload screenshots to tenant S3/R2 if configured
          const tenantStorageCfg = await getTenantStorageConfig(tenantId).catch(() => null);
          if (tenantStorageCfg) {
            let uploaded = 0;
            for (const screen of projectMap.screens as any[]) {
              for (const shot of (screen.screenshots ?? []) as any[]) {
                if (!shot.storagePath || shot.cloudUrl) continue;
                try {
                  const filename = `${screen.id}-v${shot.version}.png`;
                  const key = buildArtifactKey(tenantId, projectId, 'screenshots', filename, tenantStorageCfg.prefix);
                  const cloudUrl = await uploadScreenshotToTenantStorage(tenantStorageCfg, shot.storagePath, key);
                  await AppBrain.updateScreenshotCloudUrl(tenantId, shot.id, cloudUrl);
                  uploaded++;
                } catch (uploadErr) {
                  logger.warn({ screenId: screen.id, err: uploadErr }, 'Failed to upload screenshot to cloud storage');
                }
              }
            }
            if (uploaded > 0) logger.info({ uploaded, provider: tenantStorageCfg.provider }, 'Uploaded screenshots to cloud storage');

            // Upload DOM snapshots and markdown to tenant storage
            const { existsSync } = await import('node:fs');
            let domUploaded = 0;
            let mdUploaded = 0;
            for (const screen of projectMap.screens as any[]) {
              // DOM snapshot
              const domPath = screen.domSnapshotPath;
              if (domPath && !domPath.startsWith('https://') && existsSync(domPath)) {
                try {
                  const key = buildArtifactKey(tenantId, projectId, 'dom-snapshots', `${screen.id}.json`, tenantStorageCfg.prefix);
                  const cloudUrl = await uploadFileToTenantStorage(tenantStorageCfg, domPath, key, 'application/json');
                  await AppBrain.updateScreenDomSnapshot(tenantId, screen.id, domPath, cloudUrl);
                  domUploaded++;
                } catch (err) {
                  logger.warn({ screenId: screen.id, err }, 'Failed to upload DOM snapshot to cloud storage');
                }
              }
              // Markdown extract
              const mdPath = screen.markdownPath;
              if (mdPath && !mdPath.startsWith('https://') && existsSync(mdPath)) {
                try {
                  const key = buildArtifactKey(tenantId, projectId, 'markdown', `${screen.id}.md`, tenantStorageCfg.prefix);
                  const cloudUrl = await uploadFileToTenantStorage(tenantStorageCfg, mdPath, key, 'text/markdown');
                  await AppBrain.updateScreenMarkdownPath(tenantId, screen.id, cloudUrl);
                  mdUploaded++;
                } catch (err) {
                  logger.warn({ screenId: screen.id, err }, 'Failed to upload markdown to cloud storage');
                }
              }
            }
            if (domUploaded > 0) logger.info({ domUploaded, provider: tenantStorageCfg.provider }, 'Uploaded DOM snapshots to cloud storage');
            if (mdUploaded > 0) logger.info({ mdUploaded, provider: tenantStorageCfg.provider }, 'Uploaded markdown to cloud storage');
          }

          // AI screen grouping — clusters screens into semantic feature areas
          await groupScreensWithAI(tenantId, projectId, projectMap.screens as any, jobId, jobStartedAt);

          // Auto-flag stale tests: mark PENDING_REVIEW when screen elements changed since test was generated
          try {
            await withTenant(tenantId, async (tx) => {
              const staleTests = await tx.$queryRaw<{ id: string }[]>`
                SELECT tc.id
                FROM "TestCase" tc
                JOIN "Screen" s ON tc."screenId" = s.id
                WHERE tc."tenantId" = ${tenantId}
                  AND s."projectId" = ${projectId}
                  AND s."elementHashChangedAt" IS NOT NULL
                  AND tc."genAt" IS NOT NULL
                  AND s."elementHashChangedAt" > tc."genAt"
                  AND tc.status != 'PENDING_REVIEW'
                  AND tc.status != 'RETIRED'
              `;
              if (staleTests.length > 0) {
                const ids = staleTests.map((t: { id: string }) => t.id);
                await tx.testCase.updateMany({
                  where: { id: { in: ids } },
                  data: { status: 'PENDING_REVIEW' },
                });
                logger.info({ count: staleTests.length, projectId }, 'Auto-flagged stale test cases as PENDING_REVIEW');
              }
            });
          } catch (staleErr) {
            logger.warn({ err: staleErr }, 'Failed to auto-flag stale tests');
          }

          const changed = (projectMap.screens as any[]).filter((s: any) => s.hasChange);
          if (changed.length > 0) {
            await investigateScreenChanges(tenantId, projectId, changed);
          }
        }
      } catch (skillErr) {
        logger.error({ projectId, err: skillErr }, 'Taxonomy generation failed');
      }
    });

  } catch (err: any) {
    clearInterval(heartbeatTimer);
    runningControllers.delete(jobId);
    const aborted = String(err).includes('CRAWL_ABORTED');
    const authFailure = !aborted && (String(err).includes('AUTH_FAILURE') || String(err).includes('SESSION_EXPIRED'));
    const checkpoint = jobCheckpoints.get(jobId);
    jobCheckpoints.delete(jobId);

    // Auth failed mid-crawl with work still pending — auto-requeue continuation job.
    // The crawler checkpoints the remaining queue before throwing, so checkpoint.pendingQueue
    // holds URLs not yet visited. skipUrls prevents re-crawling already-done pages.
    if (authFailure && (checkpoint?.pendingQueue?.length ?? 0) > 0) {
      const pendingAfterAuth: string[] = checkpoint!.pendingQueue!;
      logger.info({ pending: pendingAfterAuth.length }, '[crawl] auth failure mid-crawl — auto-enqueuing continuation with pending URLs');
      enqueueJob('CRAWL', {
        ...config,
        startUrls: pendingAfterAuth,
        skipUrls: [...(config.skipUrls ?? []), ...(checkpoint!.visitedUrls ?? [])],
      }).catch((enqErr: any) => logger.warn({ err: enqErr?.message }, '[crawl] auth-failure auto-segment enqueue failed (non-fatal)'));
    }

    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: {
          status: aborted ? 'CANCELLED' : 'FAILED',
          error: aborted ? 'Stopped by user' : String(err),
          finishedAt: new Date(),
          ...(checkpoint ? { result: { checkpoint } as any } : {}),
        },
      })
    );

    // Notify on crawl failure (fire-and-forget)
    if (!aborted) {
      notifyAdmins(tenantId, {
        type: 'crawl.failed',
        title: 'Crawl failed',
        body: `Crawl for project failed: ${String(err)}`,
        metadata: { projectId, jobId, error: String(err) },
      }).catch(console.error);

      // Dispatch outbound webhooks (fire-and-forget)
      dispatchWebhook(tenantId, projectId, 'crawl.failed', {
        jobId,
        error: String(err),
      }).catch(console.error);
    }
  } finally {
    runningControllers.delete(jobId);
  }
}

/** After a job finishes, sum LLMCallLog entries for that project during the job window. */
export async function aggregateJobTokens(tenantId: string, jobId: string, projectId: string, startedAt: Date) {
  try {
    await withTenant(tenantId, async (tx) => {
      const rows = await tx.$queryRaw<any[]>`
        SELECT
          COALESCE(SUM("promptTokens"), 0)::INT     AS "totalPromptTokens",
          COALESCE(SUM("completionTokens"), 0)::INT AS "totalCompletionTokens",
          COALESCE(SUM("cacheReadTokens"), 0)::INT  AS "totalCacheReadTokens",
          COALESCE(SUM("estimatedCostUsd"), 0.0)    AS "totalEstimatedCostUsd",
          COUNT(*)::INT                             AS "llmCallCount"
        FROM "LLMCallLog"
        WHERE "tenantId" = ${tenantId}
          AND "jobId" = ${jobId}
      `;
      const row = rows[0];
      if (!row || !row.llmCallCount) return;
      await tx.platformJob.update({
        where: { id: jobId },
        data: {
          totalPromptTokens: row.totalPromptTokens,
          totalCompletionTokens: row.totalCompletionTokens,
          totalCacheReadTokens: row.totalCacheReadTokens,
          totalEstimatedCostUsd: Number(row.totalEstimatedCostUsd),
          llmCallCount: row.llmCallCount,
        },
      });
    });
  } catch (err) {
    console.warn('[job-tokens] aggregation failed:', err);
  }
}

async function processApiTestsJob(jobId: string, config: any) {
  const { tenantId, projectId } = config;
  const jobStartedAt = new Date();
  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: jobStartedAt } })
  );
  try {
    const { generateApiTests, refreshCoverageMatrix, refreshTestCaseInventory } = await import('@detiq/agents');
    await updateJobProgress(tenantId, jobId, 20, 'Loading API contracts and endpoints…');
    const result = await generateApiTests({ tenantId, projectId, apiEndpoints: config.apiEndpoints });
    await updateJobProgress(tenantId, jobId, 90, `Generated ${result.cases?.length ?? 0} API test cases`);
    refreshCoverageMatrix(tenantId, projectId).catch((e) => console.warn('[coverage-matrix]', e));
    refreshTestCaseInventory(tenantId, projectId).catch((e) => console.warn('[tc-inventory]', e));
    refreshTestCasesSkillFile(tenantId, projectId).catch(console.error);
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: {
          status: 'DONE', progress: 100, finishedAt: new Date(),
          result: { cases: result.cases?.length ?? 0, suiteName: result.suiteName } as any,
        },
      })
    );
    aggregateJobTokens(tenantId, jobId, projectId, jobStartedAt).catch(console.error);
  } catch (err: any) {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', error: String(err), finishedAt: new Date() } })
    );
  }
}

async function processSecurityTestsJob(jobId: string, config: any) {
  const { tenantId, projectId } = config;
  const jobStartedAt = new Date();
  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: jobStartedAt } })
  );
  try {
    const { generateSecurityTests, refreshCoverageMatrix, refreshTestCaseInventory } = await import('@detiq/agents');
    await updateJobProgress(tenantId, jobId, 20, 'Mapping OWASP attack surface…');
    const result = await generateSecurityTests({ tenantId, projectId, apiEndpoints: config.apiEndpoints });
    await updateJobProgress(tenantId, jobId, 90, `Generated ${result.cases?.length ?? 0} security test cases`);
    refreshCoverageMatrix(tenantId, projectId).catch((e) => console.warn('[coverage-matrix]', e));
    refreshTestCaseInventory(tenantId, projectId).catch((e) => console.warn('[tc-inventory]', e));
    refreshTestCasesSkillFile(tenantId, projectId).catch(console.error);
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: {
          status: 'DONE', progress: 100, finishedAt: new Date(),
          result: { cases: result.cases?.length ?? 0, suiteName: result.suiteName } as any,
        },
      })
    );
    aggregateJobTokens(tenantId, jobId, projectId, jobStartedAt).catch(console.error);
  } catch (err: any) {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', error: String(err), finishedAt: new Date() } })
    );
  }
}

async function processIntegrationTestsJob(jobId: string, config: any) {
  const { tenantId, projectId } = config;
  const jobStartedAt = new Date();
  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: jobStartedAt } })
  );
  try {
    const { generateIntegrationTests, refreshCoverageMatrix, refreshTestCaseInventory } = await import('@detiq/agents');
    await updateJobProgress(tenantId, jobId, 20, 'Analysing API contracts and data models…');
    const result = await generateIntegrationTests({ tenantId, projectId, apiEndpoints: config.apiEndpoints });
    await updateJobProgress(tenantId, jobId, 90, `Generated ${result.cases?.length ?? 0} integration test cases`);
    refreshCoverageMatrix(tenantId, projectId).catch((e) => console.warn('[coverage-matrix]', e));
    refreshTestCaseInventory(tenantId, projectId).catch((e) => console.warn('[tc-inventory]', e));
    refreshTestCasesSkillFile(tenantId, projectId).catch(console.error);
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: {
          status: 'DONE', progress: 100, finishedAt: new Date(),
          result: { cases: result.cases?.length ?? 0, suiteName: result.suiteName } as any,
        },
      })
    );
    aggregateJobTokens(tenantId, jobId, projectId, jobStartedAt).catch(console.error);
  } catch (err: any) {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', error: String(err), finishedAt: new Date() } })
    );
  }
}

async function processFigmaImportJob(jobId: string, config: any) {
  const { tenantId, projectId, fileKey } = config;
  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: new Date() } })
  );
  try {
    // Prefer OAuth token (connected via Figma OAuth flow), fall back to PAT connector.
    const oauthConn = await AppBrain.getOAuthConnection(tenantId, projectId, 'FIGMA');
    let creds: figma.FigmaCredentials;
    const resolvedFileKey = fileKey ?? '';
    if (!resolvedFileKey) throw new Error('No Figma file key. Paste the full Figma file URL as the App URL.');

    if (oauthConn?.accessToken && oauthConn.status === 'CONNECTED') {
      // OAuth token must use Authorization: Bearer header (oauthToken field), not X-Figma-Token
      creds = { accessToken: '', oauthToken: oauthConn.accessToken, fileKey: resolvedFileKey };
    } else {
      const connector = await withTenant(tenantId, (tx) =>
        (tx as any).connector.findFirst({ where: { projectId, kind: 'FIGMA', tenantId } })
      ) as any;
      if (!connector) throw new Error('Figma connector not configured. Add your Figma token in Project Settings → Integrations.');
      const cfg = connector.config as Record<string, string>;
      const pat = cfg.accessToken_enc ? decrypt(cfg.accessToken_enc) : '';
      if (!pat) throw new Error('Figma access token missing. Edit the Figma connector and add your Personal Access Token.');
      creds = { accessToken: pat, fileKey: resolvedFileKey };
    }

    await updateJobProgress(tenantId, jobId, 5, 'Listing frames from Figma file…');
    const frames = await figma.listScreens(creds, resolvedFileKey);
    if (frames.length === 0) {
      await withTenant(tenantId, (tx) =>
        tx.platformJob.update({ where: { id: jobId }, data: { status: 'DONE', progress: 100, result: { screensImported: 0 } as any, finishedAt: new Date() } })
      );
      return;
    }

    await updateJobProgress(tenantId, jobId, 15, `Fetching images for ${frames.length} frames…`);
    const nodeIds = frames.map((f: any) => f.id);
    const imageUrls = await figma.getScreenImages(creds, resolvedFileKey, nodeIds, 1);

    const screenshotDir = path.join(platformConfig.screenshotDir, projectId);
    await fs.mkdir(screenshotDir, { recursive: true });
    const tenantStorageCfg = await getTenantStorageConfig(tenantId).catch(() => null);

    let saved = 0;
    for (const frame of frames) {
      try {
        const pct = Math.round(15 + ((saved / frames.length) * 80));
        await updateJobProgress(tenantId, jobId, pct, `Saving frame ${saved + 1}/${frames.length}: ${(frame as any).name}`);
        const screen = await AppBrain.saveScreen(tenantId, projectId, (frame as any).name, `figma://file/${resolvedFileKey}/node/${(frame as any).id}`);
        const imgUrl = (imageUrls as any)[(frame as any).id];
        if (imgUrl) {
          const imgRes = await fetch(imgUrl);
          if (imgRes.ok) {
            const buf = Buffer.from(await imgRes.arrayBuffer());
            const filename = `${screen.id}-v1.png`;
            const storagePath = path.join(screenshotDir, filename);
            await fs.writeFile(storagePath, buf);
            let cloudUrl: string | undefined;
            if (tenantStorageCfg) {
              try {
                const key = buildArtifactKey(tenantId, projectId, 'screenshots', filename, tenantStorageCfg.prefix);
                cloudUrl = (await uploadScreenshotToTenantStorage(tenantStorageCfg, storagePath, key)) ?? undefined;
              } catch (e) {
                console.warn('[figma-import] cloud upload failed:', e);
              }
            }
            const w = (frame as any).boundingBox?.width ?? 1440;
            const viewport = w <= 480 ? 'MOBILE_PORTRAIT' : w <= 1024 ? 'TABLET' : 'DESKTOP';
            await AppBrain.saveScreenshot(tenantId, projectId, screen.id, storagePath, 1, viewport, cloudUrl);
          }
        }

        // Extract elements from Figma frame node → save for grounding + test generation
        if ((frame as any).node) {
          const figmaElements = figma.extractElements((frame as any).node);
          for (const el of figmaElements) {
            const role = el.type === 'COMPONENT' || el.type === 'COMPONENT_SET' ? 'COMPONENT'
              : el.type === 'INSTANCE' ? 'COMPONENT'
                : el.type === 'TEXT' ? 'TEXT'
                  : el.type === 'FRAME' || el.type === 'GROUP' ? 'REGION'
                    : el.type;
            const boundingRect = el.boundingBox
              ? { x: Math.round(el.boundingBox.x), y: Math.round(el.boundingBox.y), width: Math.round(el.boundingBox.width), height: Math.round(el.boundingBox.height) }
              : undefined;
            await AppBrain.saveElement(
              tenantId, projectId, screen.id,
              el.name, role, undefined,
              el.description ?? undefined, undefined,
              el.isInteractive,
              boundingRect, undefined, undefined,
            );
          }
        }

        saved++;
      } catch (err) {
        console.warn(`[figma-import] Failed to save frame:`, err);
      }
    }

    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'DONE', progress: 100, result: { screensImported: saved, totalFrames: frames.length } as any, finishedAt: new Date() },
      })
    );

    // Post-import: run AI grouping + taxonomy skill files (same as post-crawl)
    setImmediate(async () => {
      try {
        const projectMap = await AppBrain.getProjectMap(tenantId, projectId);
        if (projectMap?.screens?.length) {
          await generateTaxonomySkillFiles(tenantId, projectId, projectMap.screens as any);
          await groupScreensWithAI(tenantId, projectId, projectMap.screens as any, jobId, new Date());
        }
      } catch (err) {
        console.error('[figma-import] post-import grouping failed:', err);
      }
    });

  } catch (err: any) {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', error: String(err?.message ?? err), finishedAt: new Date() } })
    );
  }
}

async function processNativeCrawlJob(jobId: string, config: any) {
  const { tenantId, projectId } = config;
  const jobStartedAt = new Date();
  try {
    const build = await withTenant(tenantId, (tx) =>
      (tx as any).nativeAppBuild.findFirst({
        where: { tenantId, projectId },
        orderBy: { uploadedAt: 'desc' },
      })
    ) as { platform: string; fileFormat: string; s3Path: string; packageName: string | null } | null;

    if (!build) throw new Error('No native app build found for project');

    // AAB requires bundletool conversion — fail early with a clear message
    if (build.fileFormat === 'aab') {
      throw new Error(
        'AAB format cannot be installed directly by Appium. Convert to APK using bundletool first, then re-upload the APK.'
      );
    }

    // Persist device targeting info on the job record
    const deviceFarm = (config.deviceFarm ?? 'local_emulator') as 'local_emulator' | 'aws_device_farm' | 'browserstack';
    await withTenant(tenantId, (tx) =>
      (tx as any).platformJob.update({
        where: { id: jobId },
        data: {
          deviceModel: config.deviceModel ?? null,
          osVersion: config.osVersion ?? null,
          deviceFarm,
        },
      })
    ).catch(() => { });

    const workspaceDeviceFarm = await AppBrain.getTenantDeviceFarmConfig(tenantId).catch(() => null);

    await runNativeCrawl({
      tenantId,
      projectId,
      jobId,
      platform: build.platform as 'android' | 'ios' | 'windows' | 'macos',
      appS3Path: build.s3Path,
      packageName: build.packageName ?? undefined,
      deviceModel: config.deviceModel,
      osVersion: config.osVersion,
      deviceFarm,
      appiumServerUrl: workspaceDeviceFarm?.appiumServerUrl,
      browserstackUsername: workspaceDeviceFarm?.browserstackUsername,
      browserstackAccessKey: workspaceDeviceFarm?.browserstackAccessKey,
      awsAccessKeyId: workspaceDeviceFarm?.awsAccessKeyId,
      awsSecretAccessKey: workspaceDeviceFarm?.awsSecretAccessKey,
      awsRegion: workspaceDeviceFarm?.awsRegion,
      awsDeviceFarmProjectArn: workspaceDeviceFarm?.awsDeviceFarmProjectArn,
    });

    // Post-crawl: generate taxonomy skill files + AI screen grouping (mirrors web crawl processJob)
    setImmediate(async () => {
      try {
        const projectMap = await AppBrain.getProjectMap(tenantId, projectId);
        const screens = (projectMap?.screens ?? []) as any[];
        if (screens.length > 0) {
          await generateTaxonomySkillFiles(tenantId, projectId, screens);
          await groupScreensWithAI(tenantId, projectId, screens, jobId, jobStartedAt);

          // Auto-flag stale tests: mark PENDING_REVIEW when screen elements changed since test was generated
          try {
            await withTenant(tenantId, async (tx) => {
              const staleTests = await tx.$queryRaw<{ id: string }[]>`
                SELECT tc.id
                FROM "TestCase" tc
                JOIN "Screen" s ON tc."screenId" = s.id
                WHERE tc."tenantId" = ${tenantId}
                  AND s."projectId" = ${projectId}
                  AND s."elementHashChangedAt" IS NOT NULL
                  AND tc."genAt" IS NOT NULL
                  AND s."elementHashChangedAt" > tc."genAt"
                  AND tc.status != 'PENDING_REVIEW'
                  AND tc.status != 'RETIRED'
              `;
              if (staleTests.length > 0) {
                const ids = staleTests.map((t: { id: string }) => t.id);
                await tx.testCase.updateMany({
                  where: { id: { in: ids } },
                  data: { status: 'PENDING_REVIEW' },
                });
                logger.info({ count: staleTests.length, projectId }, 'Auto-flagged stale test cases as PENDING_REVIEW');
              }
            });
          } catch (staleErr) {
            logger.warn({ err: staleErr }, 'Failed to auto-flag stale tests');
          }
        }
      } catch (e) {
        logger.error({ err: e }, 'post-native-crawl processing failed');
      }
    });
  } catch (err: any) {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', error: String(err?.message ?? err), finishedAt: new Date() } })
    ).catch(() => { });
  }
}

async function processGenerateTestsFromReqsJob(jobId: string, config: any) {
  const { tenantId, projectId, reqIds, suiteId, suiteName } = config;
  try {
    await updateJobProgress(tenantId, jobId, 5, 'Loading requirements');

    const { askTracked, deterministicSeed, getTaskAgentLLMConfig } = await import('@detiq/agents');
    const llmCfg = await getTaskAgentLLMConfig(tenantId, 'testGeneration');

    const all = await AppBrain.listRequirements(tenantId, projectId, {}) as any[];
    const reqs = reqIds?.length
      ? all.filter((r: any) => reqIds.includes(r.reqId))
      : all.filter((r: any) => r.status === 'unbuilt' || r.status === 'pending');

    if (reqs.length === 0) {
      await updateJobProgress(tenantId, jobId, 100, 'No pending requirements found');
      await withTenant(tenantId, (tx) =>
        tx.platformJob.update({ where: { id: jobId }, data: { status: 'DONE', finishedAt: new Date(), result: { created: 0, skipped: 0 } as any } })
      );
      return;
    }

    // Resolve or create suite
    let resolvedSuiteId = suiteId;
    if (!resolvedSuiteId) {
      const name = suiteName?.trim() || 'Requirements Tests';
      const suite = await withTenant(tenantId, async (tx) => {
        const existing = await (tx as any).testSuite.findFirst({ where: { tenantId, projectId, name } });
        if (existing) return existing;
        return (tx as any).testSuite.create({ data: { tenantId, projectId, name } });
      }) as any;
      resolvedSuiteId = suite.id;
    }

    await updateJobProgress(tenantId, jobId, 20, `Generating tests for ${reqs.length} requirements`);

    const reqBlock = reqs.map((r: any) => [
      `## ${r.reqId}: ${r.title}`,
      `Priority: ${r.priority} | Type: ${r.type}`,
      r.acceptanceCriteria?.length ? `Acceptance Criteria:\n${r.acceptanceCriteria.map((c: string) => `- ${c}`).join('\n')}` : '',
      r.businessRules?.length ? `Business Rules:\n${r.businessRules.map((b: string) => `- ${b}`).join('\n')}` : '',
    ].filter(Boolean).join('\n')).join('\n\n');

    const system = `You are ZeTa-AI's test case generator. Convert requirements into structured test cases.
Return ONLY valid JSON: { "cases": [ { "title": string, "kind": string, "priority": string, "preconditions": string, "steps": [{"action": string, "data": string, "expected": string}], "expectedResult": string, "tags": string[], "reqRef": string } ] }
Kind must be one of: POSITIVE, NEGATIVE, EDGE, SECURITY, ACCESSIBILITY, INTEGRATION, E2E, SMOKE, REGRESSION.
Priority must be one of: CRITICAL, HIGH, MEDIUM, LOW.
Generate 1-3 test cases per requirement (positive + negative + edge if applicable).
Each test case must reference its reqRef (the requirement ID like REQ-001).`;

    const user = `Generate test cases for these requirements:\n\n${reqBlock}`;

    let parsed: { cases: any[] } = { cases: [] };
    try {
      const reqSeed = deterministicSeed(`${tenantId}:${projectId}:req-cases:${reqs.map((r: any) => r.reqId).sort().join('|').slice(0, 300)}`);
      const { text, usage } = await askTracked(system, user, llmCfg, undefined, undefined, { temperature: 0, seed: reqSeed });
      if (usage) recordLLMCall({ tenantId, projectId, jobId, agentType: 'generate_tests_from_reqs', label: 'test_case_generation', usage }).catch(() => { });
      const json = text.match(/\{[\s\S]*\}/)?.[0];
      if (json) parsed = JSON.parse(json);
    } catch (e: any) {
      throw new Error('AI failed to generate test cases: ' + (e.message ?? 'unknown'));
    }

    await updateJobProgress(tenantId, jobId, 70, `Saving ${parsed.cases?.length ?? 0} test cases`);

    const VALID_KINDS = new Set(['POSITIVE', 'NEGATIVE', 'EDGE', 'STATE_TRANSITION', 'ACCESSIBILITY', 'SECURITY', 'INTEGRATION', 'E2E', 'API', 'EXPLORATORY', 'UAT', 'PERFORMANCE', 'COMPATIBILITY', 'L10N', 'USABILITY', 'LOAD', 'UNIT', 'MUTATION', 'AB_TEST', 'ALPHA_BETA', 'SYSTEM', 'SMOKE', 'REGRESSION', 'SANITY']);
    const VALID_PRIORITIES = new Set(['CRITICAL', 'HIGH', 'MEDIUM', 'LOW']);
    const normalized = (parsed.cases ?? []).map((c: any) => ({
      ref: c.reqRef ?? null,
      traceTo: c.reqRef ?? null,
      title: c.title || 'Untitled',
      kind: VALID_KINDS.has((c.kind ?? '').toUpperCase()) ? c.kind.toUpperCase() : 'POSITIVE',
      priority: VALID_PRIORITIES.has((c.priority ?? '').toUpperCase()) ? c.priority.toUpperCase() : 'MEDIUM',
      preconditions: c.preconditions ?? '',
      steps: Array.isArray(c.steps) ? c.steps : [],
      expectedResult: c.expectedResult ?? '',
      tags: Array.isArray(c.tags) ? c.tags : [],
    }));

    const { created, skipped } = await AppBrain.importTestCasesFromArtifact(
      tenantId, projectId, resolvedSuiteId, normalized, { duplicateMode: 'skip' }
    );

    // Mark requirements as covered
    const coveredReqIds = new Set(normalized.map((c: any) => c.ref).filter(Boolean));
    await Promise.all(
      reqs
        .filter((r: any) => coveredReqIds.has(r.reqId))
        .map((r: any) => AppBrain.updateRequirement(tenantId, projectId, r.reqId, { status: 'covered' }).catch(() => { }))
    );

    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'DONE', finishedAt: new Date(), result: { created: created.length, skipped, suiteId: resolvedSuiteId } as any, progress: 100 },
      })
    );
  } catch (e: any) {
    const msg = e?.message ?? 'Unknown error';
    logger.error({ jobId, err: msg }, 'processGenerateTestsFromReqsJob failed');
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', finishedAt: new Date(), error: msg } })
    ).catch(() => { });
  }
}

/**
 * Runs every APPROVED test case in a project (optionally scoped to one suite)
 * against the agent-free Playwright executor (test-executor.ts) and aggregates
 * the results into the TestRun row created synchronously by POST /run-tests
 * (see ci.ts) — this is what makes the CI gate (GET /test-results/latest)
 * finally reflect what CI itself triggered, instead of a stale or nonexistent
 * result from an unrelated, manually-triggered, agent-connected suite run.
 */
async function processRunTestSuiteJob(jobId: string, config: any) {
  const { tenantId, projectId, testRunId, suiteId, storageState, browser, shardIndex, totalShards } = config as any;
  const runStart = Date.now();

  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: new Date() } })
  );

  try {
    let testCases = await AppBrain.listTestCases(tenantId, projectId, {
      status: 'APPROVED',
      ...(suiteId ? { suiteId } : {}),
    }) as Array<{ id: string }>;

    // Shard filtering: when running a distributed matrix (CI-triggered multi-browser/shard),
    // each worker only executes its own slice — testCases[shardIndex::totalShards].
    if (typeof totalShards === 'number' && totalShards > 1 && typeof shardIndex === 'number') {
      testCases = testCases.filter((_, i) => i % totalShards === shardIndex);
    }

    if (testCases.length === 0) {
      // Nothing approved to run isn't a CI failure — same as any test runner
      // reporting "0 tests, 0 failures" when a suite is empty.
      await AppBrain.completeTestRun(tenantId, testRunId, { total: 0, passed: 0, failed: 0, skipped: 0, duration: 0 });
      await withTenant(tenantId, (tx) =>
        tx.platformJob.update({
          where: { id: jobId },
          data: { status: 'DONE', progress: 100, finishedAt: new Date(), result: { total: 0 } as any },
        })
      );
      return;
    }

    const aiHealingEnabled = await isFeatureEnabled(tenantId, 'AI_SELF_HEALING');
    const SCREENSHOT_DIR = process.env.SCREENSHOT_DIR ?? '/tmp/zeta-screenshots';
    const screenshotDir = config.screenshotDir || platformConfig.screenshotDir || SCREENSHOT_DIR;
    const VALID_BROWSERS = ['chromium', 'firefox', 'webkit'];
    const validBrowser = VALID_BROWSERS.includes(browser) ? browser : 'chromium';

    let passed = 0;
    let failed = 0;
    const runResults: Array<{ testCaseId: string; status: string }> = [];
    for (let i = 0; i < testCases.length; i += TEST_EXEC_CONCURRENCY) {
      const batch = testCases.slice(i, i + TEST_EXEC_CONCURRENCY);
      await Promise.all(batch.map(async (tc) => {
        try {
          const full = await AppBrain.getTestCaseForExecution(tenantId, tc.id) as any;
          const baseUrl = full?.screen?.project?.appUrl ?? '';
          const execProjectId = full?.screen?.project?.id ?? projectId;
          const dataRows: Array<{ id: string; rowIndex: number; data: Record<string, string> }> =
            full?.dataSet?.rows ?? [];

          // Data-driven: run once per row (substituting placeholders). If no data set, run once with no substitution.
          const runs = dataRows.length > 0 ? dataRows : [null];

          for (const row of runs) {
            let tcForPlan = full ?? tc;
            if (row !== null) {
              // Clone the test case with interpolated steps and title
              const rawSteps = Array.isArray(tcForPlan.steps) ? tcForPlan.steps as string[] : [];
              tcForPlan = {
                ...tcForPlan,
                steps: rawSteps.map((s: string) => interpolatePlaceholders(s, row.data)),
                title: interpolatePlaceholders(tcForPlan.title ?? '', row.data),
                preconditions: tcForPlan.preconditions
                  ? interpolatePlaceholders(tcForPlan.preconditions, row.data)
                  : tcForPlan.preconditions,
              };
            }

            const plan = await generateExecutionPlan(tenantId, tcForPlan, baseUrl);
            const result = await executeTestPlan(plan, {
              screenshotDir,
              testRunId,
              maxRuns: 1,
              browser: validBrowser as 'chromium' | 'firefox' | 'webkit',
              storageState,
              healCtx: aiHealingEnabled ? { tenantId, projectId: execProjectId } : undefined,
            });
            const status = result.status === 'PASSED' ? 'PASSED' : result.status === 'ERROR' ? 'ERROR' : 'FAILED';
            if (status === 'PASSED') passed++; else failed++;
            runResults.push({ testCaseId: tc.id, status });
            await AppBrain.saveTestResult(tenantId, {
              runId: testRunId,
              testCaseId: tc.id,
              status,
              duration: result.totalDuration,
              ...(result.rootCauseSummary
                ? { failureAnalysis: result.rootCauseSummary, failureAnalysisAt: new Date() }
                : {}),
              ...(row !== null ? { dataRowId: row.id, dataRowIndex: row.rowIndex } : {}),
              ...((result.consoleLogs?.length || result.networkErrors?.length) ? {
                logs: {
                  consoleLogs: result.consoleLogs ?? [],
                  networkErrors: result.networkErrors ?? [],
                },
              } : {}),
            });
          }
          await AppBrain.recordFlakinessOutcome(tenantId, tc.id).catch(() => { });
        } catch (err: any) {
          failed++;
          runResults.push({ testCaseId: tc.id, status: 'ERROR' });
          await AppBrain.saveTestResult(tenantId, {
            runId: testRunId,
            testCaseId: tc.id,
            status: 'ERROR',
            error: String(err?.message ?? err).slice(0, 500),
          }).catch(() => { });
        }
      }));
      await updateJobProgress(
        tenantId, jobId,
        Math.round((Math.min(i + TEST_EXEC_CONCURRENCY, testCases.length) / testCases.length) * 100),
        `Ran ${Math.min(i + TEST_EXEC_CONCURRENCY, testCases.length)}/${testCases.length} test cases`,
      );
    }

    await AppBrain.completeTestRun(tenantId, testRunId, {
      total: testCases.length,
      passed,
      failed,
      skipped: 0,
      duration: Date.now() - runStart,
    });

    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'DONE', progress: 100, finishedAt: new Date(), result: { total: testCases.length, passed, failed } as any },
      })
    );

    // Same full fan-out as the interactive web-driven run path (in-app,
    // generic webhook, suite-owner failure emails, GitHub check-run, Slack,
    // Teams, alert emails on failure) — a CI-triggered run has nobody
    // watching the SSE stream, so this is the only place these fire from.
    notifyRunCompletion(
      tenantId, projectId, testRunId,
      failed === 0 ? 'COMPLETED' : 'FAILED',
      { passed, failed, total: testCases.length },
      runResults,
    ).catch(console.error);
    refreshTestExecutionHistorySkillFile(tenantId, projectId).catch(console.error);
  } catch (err: any) {
    const msg = String(err?.message ?? err).slice(0, 500);
    logger.error({ jobId, testRunId, err: msg }, 'processRunTestSuiteJob failed');
    await AppBrain.failTestRun(tenantId, testRunId, msg).catch(() => { });
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', finishedAt: new Date(), error: msg } })
    ).catch(() => { });
  }
}

async function processGenerateSkillFilesJob(jobId: string, config: any) {
  const { tenantId, projectId } = config;
  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: new Date() } })
  );
  try {
    await updateJobProgress(tenantId, jobId, 10, 'Loading project data…');
    const projectMap = await AppBrain.getProjectMap(tenantId, projectId);
    if (!projectMap) throw new Error('Project not found.');
    const screens = (projectMap.screens ?? []) as any[];
    // Manual regeneration has no fresh crawl result — reuse the metrics and
    // endpoints persisted on the last completed crawl job so the Performance
    // Baseline and API Contract aren't wiped to "no data captured".
    const lastCrawl = await withTenant(tenantId, (tx) =>
      tx.platformJob.findFirst({
        where: { projectId, type: 'CRAWL', status: 'DONE' },
        orderBy: { finishedAt: 'desc' },
        select: { result: true },
      })
    );
    const lastResult = ((lastCrawl as any)?.result ?? {}) as any;
    await updateJobProgress(tenantId, jobId, 25, `Generating ${screens.length} app intelligence skill files…`);
    await generateTaxonomySkillFiles(tenantId, projectId, screens, lastResult.apiEndpoints, lastResult.performanceMetrics);
    await updateJobProgress(tenantId, jobId, 85, 'Refreshing testing intelligence skill files…');
    await Promise.allSettled([
      refreshTestCasesSkillFile(tenantId, projectId),
      refreshTestDataSkillFile(tenantId, projectId),
      refreshTestScriptSkillFile(tenantId, projectId),
      refreshRequirementsCoverageSkillFile(tenantId, projectId),
      refreshTestExecutionHistorySkillFile(tenantId, projectId),
      refreshVisualRegressionHistorySkillFile(tenantId, projectId),
      refreshLocatorStrategyWithHealingData(tenantId, projectId),
      refreshFieldCatalogSkillFile(tenantId, projectId),
    ]);
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'DONE', progress: 100, finishedAt: new Date() } })
    );
  } catch (err: any) {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', error: String(err?.message ?? err), finishedAt: new Date() } })
    );
  }
}

// DATA_RETENTION_PURGE — deletes AuditLog rows older than each tenant's
// dataRetentionDays setting (default 365), and LLMCallLog rows older than 90
// days (fixed, no per-tenant override). Runs once daily via the cron in
// worker.ts. The job has no tenantId/projectId — it sweeps all tenants so
// enqueueJob() is not used; the worker.ts cron calls this directly.
export async function processDataRetentionPurge(): Promise<void> {
  logger.info('[data-retention] starting purge run');

  // LLMCallLog: fixed 90-day retention across all tenants (uses calledAt, not createdAt)
  const llmCutoff = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
  const llmResult = await prismaAdmin.lLMCallLog.deleteMany({
    where: { calledAt: { lt: llmCutoff } },
  });
  logger.info({ deleted: llmResult.count }, '[data-retention] LLMCallLog purged');

  // AuditLog: per-tenant retention via TenantSetting.dataRetentionDays
  const tenants = await prismaAdmin.tenant.findMany({ select: { id: true } });
  let auditTotal = 0;

  for (const { id: tenantId } of tenants) {
    // Find the most restrictive (lowest) dataRetentionDays across this tenant's
    // settings rows, falling back to 365 if no setting row exists.
    const setting = await prismaAdmin.tenantSetting.findFirst({
      where: { tenantId },
      orderBy: { dataRetentionDays: 'asc' },
      select: { dataRetentionDays: true },
    });
    const retentionDays = setting?.dataRetentionDays ?? (parseInt(process.env.DEFAULT_DATA_RETENTION_DAYS ?? '') || 365);
    const cutoff = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000);

    const result = await prismaAdmin.auditLog.deleteMany({
      where: { tenantId, createdAt: { lt: cutoff } },
    });
    auditTotal += result.count;
  }

  logger.info({ deleted: auditTotal, tenants: tenants.length }, '[data-retention] AuditLog purged');

  // AuthSession: fixed 12-month retention — IP addresses are PII under GDPR.
  // Only purge revoked sessions; active (non-revoked) sessions are kept.
  const SESSION_RETENTION_DAYS = 365;
  const sessionCutoff = new Date(Date.now() - SESSION_RETENTION_DAYS * 24 * 60 * 60_000);
  const sessionResult = await prismaAdmin.authSession.deleteMany({
    where: {
      createdAt: { lt: sessionCutoff },
      revokedAt: { not: null },
    },
  });
  logger.info({ deleted: sessionResult.count }, '[data-retention] AuthSession purged');
}

// ── URL Discovery Job ─────────────────────────────────────────────────────────
// Three-phase URL discovery: sitemap seed → unauthenticated BFS spider → authenticated BFS
// spider (when crawl credentials exist). Persists discovered URLs to crawlConfig and marks the
// job DONE with the full URL list so the UI can poll for results.

async function processDiscoverJob(jobId: string, config: any, opts?: ProcessJobOpts) {
  const { tenantId, projectId } = config;

  const controller = new AbortController();
  runningControllers.set(jobId, controller);

  // Independent heartbeat timer — keeps heartbeatAt fresh even when no URLs are
  // arriving (e.g. during auth login, page-load waits, 2FA handling).
  const heartbeatTimer = setInterval(() => {
    void heartbeatJob(tenantId, jobId);
  }, JOB_HEARTBEAT_INTERVAL_MS);

  // Shared accumulator — updated by onUrlsDiscovered callback as each page is visited.
  // This lets the frontend poll and see a growing list before the job finishes.
  const allUrls = new Set<string>();
  let lastPartialFlush = 0;
  let lastFlushedPct = 0; // high-water mark — progress never decreases

  const flushPartialUrls = async (pct: number, step: string) => {
    const now = Date.now();
    // Flush to DB at most every 5 seconds to avoid excessive writes
    if (now - lastPartialFlush < 5_000 && allUrls.size < 5) return;
    const monotonic = Math.max(pct, lastFlushedPct);
    lastFlushedPct = monotonic;
    lastPartialFlush = now;
    try {
      await withTenant(tenantId, (tx) =>
        tx.platformJob.update({
          where: { id: jobId },
          data: { progress: monotonic, result: { step, partialUrls: Array.from(allUrls) } as any, heartbeatAt: new Date() },
        })
      );
    } catch { /* non-fatal */ }
    await opts?.onProgress?.(monotonic);
  };

  const onProgress = async (pct: number, step: string) => {
    await updateJobProgress(tenantId, jobId, pct, step);
    await opts?.onProgress?.(pct);
  };

  try {
    // Mark RUNNING inside the try block so any DB error is caught by the internal
    // handler below (which writes result.error). If this were before the try block,
    // a DB failure would bubble up to dispatchClaimedJob's catch, which writes to
    // the `error` column instead of `result.error` — the frontend would then show
    // "Unknown error" rather than the actual message.
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: new Date() } })
    );
    // Load decrypted project crawl config for credentials/proxy/sitemap settings.
    // Reading Project.crawlConfig directly leaves encrypted fields as *_enc, so
    // discovery would see username but miss password/storageState/authToken.
    const crawlCfg = await AppBrain.getCrawlConfig(tenantId, projectId).catch(() => null) ?? {};
    const appUrl: string = config.appUrl ?? crawlCfg.appUrl;
    if (!appUrl) throw new Error('No app URL configured. Save the crawl settings first.');

    const { getSystemConfig } = await import('@detiq/core');
    const crawlTuning = await getSystemConfig('crawl_tuning');
    const maxDiscoverUrls: number = config.maxDiscoverUrls ?? parseInt(String(crawlTuning?.maxDiscoverUrls ?? '2000'), 10);
    const proxyUrl: string | undefined = config.proxyUrl ?? crawlCfg.proxyUrl;
    const screenshotDir = os.tmpdir();
    const discoverStorageConfig = await getTenantStorageConfig(tenantId).catch(() => null);

    await onProgress(3, 'Seeding from sitemap/robots.txt…');

    // Phase 1 — sitemap + robots.txt seed (fast, no browser)
    const sitemapResult = await discoverProjectUrls(appUrl, {
      sitemapUrl: config.sitemapUrl ?? crawlCfg.sitemapUrl,
      proxyUrl,
    }).catch(() => ({ candidates: [] as Array<{ url: string; source: string }>, sitemapFound: false }));
    const sitemapUrls = sitemapResult.candidates.map((c: any) => c.url as string);
    // Seed accumulator with sitemap URLs so they show immediately
    for (const u of sitemapUrls) allUrls.add(u);
    logger.info({ count: sitemapUrls.length, sitemapFound: sitemapResult.sitemapFound }, '[discover] Phase 1 sitemap done');

    // Phase 2 — unauthenticated BFS spider
    await onProgress(8, `Unauthenticated spider from ${appUrl}…`);
    const unauthResult = await crawlProject({
      tenantId,
      projectId,
      appUrl,
      discoveryMode: true,
      startUrls: sitemapUrls.length > 0 ? sitemapUrls : undefined,
      screenshotDir,
      maxScreens: maxDiscoverUrls,
      proxyUrl,
      useFirefox: crawlCfg.useFirefox,
      storageConfig: discoverStorageConfig,
      abortSignal: controller.signal,
      jobId,
      onUrlsDiscovered: async (urls: string[]) => {
        for (const u of urls) allUrls.add(u);
        const pct = 8 + Math.round((allUrls.size / Math.max(1, maxDiscoverUrls)) * 42);
        await flushPartialUrls(Math.min(pct, 49), `Discovered ${allUrls.size} URLs…`);
      },
    }, async (pct: number, step: string) => {
      await flushPartialUrls(8 + Math.round(pct * 0.42), `Unauthenticated: ${step}`);
    }).catch((err: any) => {
      if (err?.message === 'CRAWL_ABORTED') throw err;
      logger.warn({ err: err?.message }, '[discover] Phase 2 unauthenticated spider failed — continuing with partial results');
      return { discoveredUrls: [] as string[] };
    });

    for (const u of ((unauthResult as any).discoveredUrls ?? [] as string[])) allUrls.add(u);
    logger.info({ count: allUrls.size }, '[discover] Phase 2 unauth spider done');

    // Phase 3 — authenticated BFS spider (only when credentials configured)
    // getCrawlConfig returns flat fields (username/password/totpSecret) — NOT nested under .credentials
    let resolvedSso: {
      ssoProvider?: string;
      ssoDomain?: string;
      ssoClientId?: string;
      ssoClientSecret?: string;
      ssoScope?: string;
    } = {};
    if (crawlCfg.ssoProviderId) {
      const ssoProviders = await AppBrain.getWorkspaceSsoProviders(tenantId).catch(() => []);
      const ssoProvider = ssoProviders.find((p: any) => p.id === crawlCfg.ssoProviderId);
      if (ssoProvider) {
        resolvedSso = {
          ssoProvider: ssoProvider.provider,
          ssoDomain: ssoProvider.domain,
          ssoClientId: ssoProvider.clientId,
          ssoClientSecret: ssoProvider.clientSecret,
          ssoScope: ssoProvider.scope,
        };
      }
    }

    const hasCreds = !!(
      crawlCfg.username ||
      crawlCfg.storageState ||
      crawlCfg.authToken ||
      crawlCfg.chromeProfilePath ||
      (crawlCfg.httpBasicUsername && crawlCfg.httpBasicPassword)
    );
    if (hasCreds && !controller.signal.aborted) {
      await onProgress(52, 'Authenticated spider…');
      const runtimeCreds = crawlCfg.username
        ? {
          username: crawlCfg.username,
          password: crawlCfg.password ?? '',
          ...(crawlCfg.totpSecret
            ? { getOneTimeCode: async () => generateTOTP(crawlCfg.totpSecret!) }
            : {}),
        }
        : undefined;

      const authResult = await crawlProject({
        tenantId,
        projectId,
        appUrl,
        discoveryMode: true,
        credentials: runtimeCreds,
        storageState: crawlCfg.storageState,
        httpBasicUsername: crawlCfg.httpBasicUsername,
        httpBasicPassword: crawlCfg.httpBasicPassword,
        authToken: crawlCfg.authToken,
        authTokenHeader: crawlCfg.authTokenHeader,
        authTokenPrefix: crawlCfg.authTokenPrefix,
        startUrls: crawlCfg.postLoginUrl
          ? [crawlCfg.postLoginUrl, ...(sitemapUrls.length > 0 ? sitemapUrls : [])]
          : (sitemapUrls.length > 0 ? sitemapUrls : undefined),
        screenshotDir,
        maxScreens: maxDiscoverUrls,
        proxyUrl,
        chromeProfilePath: crawlCfg.chromeProfilePath,
        useFirefox: crawlCfg.useFirefox,
        storageConfig: discoverStorageConfig,
        captchaSolverApiKey: crawlCfg.captchaSolverApiKey,
        captchaSolverProvider: crawlCfg.captchaSolverProvider,
        mailslurpApiKey: crawlCfg.mailslurpApiKey,
        mailslurpInboxId: crawlCfg.mailslurpInboxId,
        loginInstructions: crawlCfg.loginInstructions,
        ...resolvedSso,
        abortSignal: controller.signal,
        jobId,
        onUrlsDiscovered: async (urls: string[]) => {
          for (const u of urls) allUrls.add(u);
          const pct = 52 + Math.round((allUrls.size / Math.max(1, maxDiscoverUrls)) * 38);
          await flushPartialUrls(Math.min(pct, 89), `Discovered ${allUrls.size} URLs (authenticated)…`);
        },
      }, async (pct: number, step: string) => {
        await flushPartialUrls(52 + Math.round(pct * 0.38), `Authenticated: ${step}`);
      }).catch((err: any) => {
        if (err?.message === 'CRAWL_ABORTED') throw err;
        logger.warn({ err: err?.message }, '[discover] Phase 3 authenticated spider failed — using unauthenticated results');
        return { discoveredUrls: [] as string[] };
      });

      for (const u of (((authResult as any).discoveredUrls) ?? [] as string[])) {
        allUrls.add(u);
      }
      logger.info({ count: allUrls.size }, '[discover] Phase 3 auth spider done');
    }

    // Persist discovered URLs to project crawl config
    // Filter sentinel strings (_pending_, _auth_wall_, etc.) that may have leaked in
    // via onUrlsDiscovered callbacks fired before visited-map entries were updated.
    const discoveredUrls = Array.from(allUrls).filter(u => u.startsWith('http'));
    const existingCfg = await AppBrain.getCrawlConfig(tenantId, projectId).catch(() => ({}));
    await AppBrain.saveCrawlConfig(tenantId, projectId, {
      ...(existingCfg as any),
      discoveredUrls,
      discoveredAt: new Date().toISOString(),
    });

    await onProgress(100, `Done — ${discoveredUrls.length} URLs discovered`);

    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: {
          status: 'DONE',
          result: {
            discoveredUrls,
            count: discoveredUrls.length,
            sitemapFound: sitemapResult.sitemapFound,
          } as any,
          progress: 100,
          finishedAt: new Date(),
        },
      })
    );

    notifyAdmins(tenantId, {
      type: 'discovery.done',
      title: 'URL Discovery completed',
      body: `Found ${discoveredUrls.length} URLs for the project.`,
      metadata: { projectId, jobId },
    }).catch(console.error);

  } catch (err: any) {
    logger.error({ err: err?.message, jobId }, '[discover] Job failed');
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: {
          status: err?.message === 'CRAWL_ABORTED' ? 'CANCELLED' : 'FAILED',
          result: { error: err?.message ?? String(err) } as any,
          finishedAt: new Date(),
        },
      })
    ).catch(console.error);
  } finally {
    clearInterval(heartbeatTimer);
    runningControllers.delete(jobId);
  }
}

// ── Site Monitor Job ─────────────────────────────────────────────────────────
// Lightweight scheduled content check: fetches configured URLs, diffs content
// hashes against stored baselines, fires content.changed webhooks on drift.
// Baselines are stored in project.crawlConfig.monitorBaselines (url→hash map).

async function processMonitorJob(jobId: string, config: any, opts?: ProcessJobOpts) {
  const { tenantId, projectId } = config;
  const controller = new AbortController();
  runningControllers.set(jobId, controller);

  try {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: new Date() } })
    );

    const proj: any = await withTenant(tenantId, (tx) =>
      (tx as any).project.findUnique({ where: { id: projectId }, select: { crawlConfig: true } })
    );
    const crawlCfg = proj?.crawlConfig ?? {};
    const monitorUrls: string[] = config.monitorUrls ?? crawlCfg.monitorUrls ?? [];

    if (monitorUrls.length === 0) {
      throw new Error('No monitor URLs configured. Add monitorUrls to crawl config or job config.');
    }

    await updateJobProgress(tenantId, jobId, 10, `Monitoring ${monitorUrls.length} URL(s)...`);

    // Baseline: stored baselines from previous monitor run take priority;
    // project screens' elementHash is the fallback for first-run.
    const storedBaselines: Record<string, string> = crawlCfg.monitorBaselines ?? {};
    const projectScreens: Array<{ url: string | null; elementHash: string | null }> =
      await withTenant(tenantId, (tx) =>
        tx.screen.findMany({ where: { projectId }, select: { url: true, elementHash: true } })
      ).catch(() => []) as any;
    const screenHashMap: Record<string, string> = {};
    for (const s of projectScreens) {
      if (s.url && s.elementHash) screenHashMap[s.url] = s.elementHash;
    }

    const results: Array<{
      url: string; changed: boolean; reason: string;
      currentHash: string; previousHash: string | null;
    }> = [];
    const newBaselines: Record<string, string> = { ...storedBaselines };

    for (let i = 0; i < monitorUrls.length; i++) {
      if (controller.signal.aborted) break;
      const url = monitorUrls[i];
      const pct = 10 + Math.floor((i / monitorUrls.length) * 80);
      await updateJobProgress(tenantId, jobId, pct, `Checking ${url}...`);
      opts?.onProgress?.(pct);

      try {
        const crawlResult = await httpCrawl({ appUrl: url, maxScreens: 1, respectRobotsTxt: false, timeoutMs: parseInt(process.env.CRAWL_TIMEOUT_MS ?? '20000', 10) });
        const screen = crawlResult.screens[0];
        if (!screen) {
          results.push({ url, changed: false, reason: 'fetch_empty', currentHash: '', previousHash: null });
          continue;
        }
        const currentHash = hashContent(normalizeText(screen.markdown));
        const previousHash = storedBaselines[url] ?? screenHashMap[url] ?? null;
        newBaselines[url] = currentHash;

        if (!previousHash) {
          results.push({ url, changed: false, reason: 'new_baseline', currentHash, previousHash: null });
        } else if (previousHash === currentHash) {
          results.push({ url, changed: false, reason: 'unchanged', currentHash, previousHash });
        } else {
          results.push({ url, changed: true, reason: 'content_changed', currentHash, previousHash });
        }
      } catch (fetchErr: any) {
        results.push({ url, changed: false, reason: `error:${fetchErr?.message ?? String(fetchErr)}`, currentHash: '', previousHash: null });
      }
    }

    // Persist updated baselines so next run has fresh reference points.
    await withTenant(tenantId, (tx) =>
      (tx as any).project.update({
        where: { id: projectId },
        data: { crawlConfig: { ...crawlCfg, monitorBaselines: newBaselines } as any },
      })
    ).catch((e: any) => logger.warn({ err: e }, '[monitor] baseline persist failed'));

    const changedUrls = results.filter(r => r.changed);

    if (changedUrls.length > 0) {
      dispatchWebhook(tenantId, projectId, 'content.changed', {
        jobId,
        changedUrls: changedUrls.map(r => ({ url: r.url, previousHash: r.previousHash, currentHash: r.currentHash })),
        totalChecked: results.length,
        changedCount: changedUrls.length,
      }).catch(console.error);
    }

    await updateJobProgress(tenantId, jobId, 100, 'Monitor check complete');
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: {
          status: 'DONE',
          progress: 100,
          finishedAt: new Date(),
          result: { results, changedCount: changedUrls.length, totalChecked: results.length } as any,
        },
      })
    );
  } catch (err: any) {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'FAILED', error: String(err?.message ?? err), finishedAt: new Date() },
      })
    );
  } finally {
    runningControllers.delete(jobId);
  }
}

// ─── IMPORT_TESTS job processor ──────────────────────────────────────────────
// Async counterpart to the synchronous POST /artifacts/:id/import-tests route.
// Enqueued by POST /modernization/bulk-import for each artifact in a batch.
// Config: { tenantId, projectId, artifactId, suiteId?, autoLabel?, sessionId? }

async function processImportTestsJob(jobId: string, config: any) {
  const { tenantId, projectId, artifactId, autoLabel, sourcePlatform } = config;
  const jobStart = new Date();
  let suiteId: string = config.suiteId ?? 'new';

  await updateJobProgress(tenantId, jobId, 5, 'Loading artifact');

  const artifact = await withTenant(tenantId, (tx) =>
    (tx as any).projectArtifact.findUnique({ where: { id: artifactId } })
  );
  if (!artifact || (artifact as any).tenantId !== tenantId) {
    throw new Error(`Artifact ${artifactId} not found`);
  }

  const content: string = (artifact as any).content ?? '';
  const filename: string = (artifact as any).originalName ?? (artifact as any).name ?? '';
  const kind: string = (artifact as any).kind ?? '';

  // Detect format using same logic as the synchronous route
  function detectFormat(fn: string, k: string, c: string): string {
    if (fn.endsWith('.feature')) return 'GHERKIN';
    if (fn.match(/\.(xlsx|xls|csv)$/i) || k === 'SPREADSHEET') return 'EXCEL';
    if (fn.endsWith('.txt') && c && c.split('\n').some((l) => l.trim().startsWith('|'))) return 'EXCEL';
    return 'AI_EXTRACT';
  }
  const format = detectFormat(filename, kind, content);

  await updateJobProgress(tenantId, jobId, 15, `Parsing ${filename}`);

  const { parseTestCases, labelTestCasesBatch } = await import('@detiq/agents');
  let res = await parseTestCases(content, format, tenantId, (artifact as any).columnMapping ?? undefined, undefined, undefined, sourcePlatform);
  if (res.cases.length === 0 && format !== 'AI_EXTRACT') {
    await updateJobProgress(tenantId, jobId, 25, 'Retrying with AI extraction');
    try {
      res = await parseTestCases(content, 'AI_EXTRACT', tenantId, undefined, undefined, undefined, sourcePlatform);
    } catch {
      // AI unavailable (no LLM configured) — proceed with 0 cases
    }
  }

  const parsedCases = res.cases;
  if (parsedCases.length === 0) {
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'DONE', progress: 100, finishedAt: new Date(), result: { imported: 0, skipped: 0, labeled: 0 } as any },
      })
    );
    return;
  }

  await updateJobProgress(tenantId, jobId, 35, `Parsed ${parsedCases.length} cases — creating suite`);

  // Auto-create suite from filename when suiteId === 'new'
  if (!suiteId || suiteId === 'new') {
    const suiteName = filename.replace(/\.[^/.]+$/, '') || 'Imported Tests';
    const suite = await withTenant(tenantId, (tx) =>
      (tx as any).testSuite.create({
        data: { tenantId, projectId, name: suiteName },
      })
    );
    suiteId = (suite as any).id;
  }

  await updateJobProgress(tenantId, jobId, 50, `Importing ${parsedCases.length} cases`);

  const { created, skipped } = await AppBrain.importTestCasesFromArtifact(
    tenantId, projectId, suiteId, parsedCases, { duplicateMode: 'skip', artifactId }
  );

  let labeled = 0;
  if (autoLabel && created.length > 0) {
    await updateJobProgress(tenantId, jobId, 75, `AI labeling ${created.length} cases`);
    try {
      const labelInput = created.map((c: any) => ({
        id: c.id, title: c.title,
        description: c.description ?? '',
        stepsText: (Array.isArray(c.steps) ? c.steps.join(' ') : String(c.steps ?? '')).slice(0, 200),
      }));
      const labelResults = await labelTestCasesBatch(labelInput, tenantId);
      // Persist labels back to each test case
      await Promise.all(
        labelResults.map((lr: any) =>
          withTenant(tenantId, (tx) =>
            tx.testCase.update({
              where: { id: lr.id },
              data: {
                priority: lr.priority ?? undefined,
                kind: lr.kind ?? undefined,
                tags: lr.tags ?? undefined,
                status: lr.needsReview ? 'PENDING_REVIEW' : undefined,
              },
            })
          ).catch(() => { }) // non-fatal per-case
        )
      );
      labeled = labelResults.length;
    } catch {
      // non-fatal — import succeeded even if labeling fails
    }
  }

  // Auto-generate AI test data for imported cases that have recorded form fields (testDataInline)
  const casesNeedingData = created.filter((c: any) => {
    const inline = (c as any).testDataInline;
    return inline && typeof inline === 'object' && Object.keys(inline).length > 0;
  });
  if (casesNeedingData.length > 0) {
    const cap = Math.min(casesNeedingData.length, 20);
    await updateJobProgress(tenantId, jobId, 90, `Generating test data for ${cap} cases`);
    const { generateTestData, refreshTestDataSkillFile } = await import('@detiq/agents');
    for (let i = 0; i < cap; i++) {
      const c = casesNeedingData[i];
      await generateTestData({ tenantId, projectId, testCaseId: c.id, jobId }).catch((e: any) => {
        logger.warn({ err: e?.message, testCaseId: c.id }, '[import] test data generation failed');
      });
    }
    refreshTestDataSkillFile(tenantId, projectId).catch(console.error);
  }

  if (created.length > 0) {
    refreshTestCasesSkillFile(tenantId, projectId).catch(console.error);
  }

  await updateJobProgress(tenantId, jobId, 98, `Done — ${created.length} imported`);

  // Write final result and mark job DONE
  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({
      where: { id: jobId },
      data: {
        status: 'DONE',
        progress: 100,
        finishedAt: new Date(),
        result: {
          imported: created.length,
          skipped: skipped ?? 0,
          labeled,
          suiteId,
          artifactId,
          artifactName: filename,
        } as any,
      },
    })
  );
  aggregateJobTokens(tenantId, jobId, projectId, jobStart).catch(console.error);
}

// Generate test cases for a single screen, enqueued by NEW_PAGES_PIPELINE.
// Uses the same DualContextOrchestrator path as the in-process GENERATE_TESTS route.
async function processGenerateScreenTestsJob(jobId: string, config: any) {
  const { tenantId, projectId, screenId, repoConnectionId, triggerAlignedScripts } = config;
  const jobStartedAt = new Date();
  try {
    await withTenant(tenantId, (tx: any) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: jobStartedAt } })
    );

    const { DualContextOrchestrator } = await import('@detiq/agents');
    await updateJobProgress(tenantId, jobId, 10, 'Generating test cases for screen');

    const crawlCfg = await AppBrain.getCrawlConfig(tenantId, projectId).catch(() => null);
    const deviceProfiles: string[] = (crawlCfg as any)?.deviceProfiles ?? ['DESKTOP'];

    for (const viewport of deviceProfiles) {
      await DualContextOrchestrator.execute({ tenantId, projectId, targetScreenId: screenId, jobId, viewport });
    }

    await updateJobProgress(tenantId, jobId, 80, 'Test cases generated');

    // Optionally kick framework-aligned script generation for this screen's test cases
    if (triggerAlignedScripts && repoConnectionId) {
      const conn: any = await withTenant(tenantId, (tx: any) =>
        tx.gitHubRepoConnection.findFirst({ where: { id: repoConnectionId, tenantId } })
      );
      if (conn?.styleProfile) {
        const { generateScriptAligned, mapWithConcurrency, SCRIPT_GEN_CONCURRENCY } = await import('@detiq/agents');
        const cases = await withTenant(tenantId, (tx: any) =>
          tx.testCase.findMany({ where: { tenantId, projectId, screenId, status: 'APPROVED' }, select: { id: true }, take: 50 })
        );
        await mapWithConcurrency(cases as any[], SCRIPT_GEN_CONCURRENCY, (tc: any) =>
          generateScriptAligned({
            tenantId,
            projectId,
            testCaseId: tc.id,
            styleProfile: conn.styleProfile,
            pageObjectInventory: conn.pageObjectInventory ?? undefined,
          }).catch(() => { })
        );
      }
    }

    await withTenant(tenantId, (tx: any) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'DONE', progress: 100, finishedAt: new Date() } })
    );
    aggregateJobTokens(tenantId, jobId, projectId, jobStartedAt).catch(console.error);
  } catch (err: any) {
    const msg = err?.message ?? String(err);
    logger.error({ jobId, screenId, err: msg }, 'processGenerateScreenTestsJob failed');
    await withTenant(tenantId, (tx: any) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', error: msg.slice(0, 500), finishedAt: new Date() } })
    ).catch(() => { });
  }
}

type CaptureStepStatus = 'PASSED' | 'FAILED' | 'HEALED' | 'SKIPPED';
type CaptureStepResult = {
  stepNumber: number;
  description: string;
  status: CaptureStepStatus;
  screenshotPath?: string;
  screenshotUrl?: string;
  screenId?: string;
  url?: string;
  elementsCount?: number;
  healed?: boolean;
  healTier?: 'mechanical' | 'ai';
  healReasoning?: string;
  error?: string;
};

function captureStepText(step: unknown): string {
  if (typeof step === 'string') return step;
  if (!step || typeof step !== 'object') return '';
  const obj = step as Record<string, unknown>;
  return [obj.action, obj.text, obj.label, obj.data, obj.expected]
    .filter((v): v is string => typeof v === 'string' && v.trim().length > 0)
    .join(' -> ');
}

function firstQuoted(text: string): string | undefined {
  return (text.match(/"([^"]+)"/)?.[1] ?? text.match(/'([^']+)'/)?.[1])?.trim();
}

function fallbackExecutionPlan(rawSteps: unknown[]): ExecutionStep[] {
  const plan: ExecutionStep[] = [];
  for (const raw of rawSteps) {
    const text = captureStepText(raw).trim();
    if (!text) continue;
    const stepNumber = plan.length + 1;
    const url = text.match(/https?:\/\/[^\s"']+/)?.[0];

    if (/^navigate\b|^go to\b|^open\b/i.test(text) && url) {
      plan.push({ stepNumber, description: text, action: 'navigate', url });
      continue;
    }

    const urlContains = text.match(/verify url contains:\s*["']?([^"']+)["']?/i);
    if (urlContains) {
      plan.push({ stepNumber, description: text, action: 'assert_url', value: urlContains[1].trim() });
      continue;
    }

    const inputMatch = text.match(/(?:enter|type|fill)\s+["']([^"']+)["']\s+(?:in|into)\s+(.+?)(?:\s+field|\s+input|\s+box|\s+area)?(?:\s*→.*)?$/i);
    if (inputMatch) {
      const selector = inputMatch[2].replace(/^(the\s+)?/i, '').replace(/\s+(field|input|box|area)$/i, '').trim();
      plan.push({ stepNumber, description: text, action: 'fill', selectorStrategy: 'label', selector, value: inputMatch[1].trim() });
      continue;
    }

    const quoted = firstQuoted(text);
    if (/^(click|tap|navigate menu|check|select)\b/i.test(text) && quoted) {
      plan.push({ stepNumber, description: text, action: 'click', selectorStrategy: 'text', selector: quoted });
      continue;
    }
    if (/^(verify|assert|expect)\b/i.test(text) && quoted) {
      plan.push({ stepNumber, description: text, action: 'assert_visible', selectorStrategy: 'text', selector: quoted });
      continue;
    }
    if (/execute script/i.test(text)) {
      const script = text.split(':').slice(1).join(':').replace(/\s*→.*$/, '').trim();
      if (script) plan.push({ stepNumber, description: text, action: 'script', value: script });
    }
  }
  return plan;
}

function screenNameFromUrl(url: string, fallback: string): string {
  try {
    const u = new URL(url);
    return u.pathname.replace(/\/+$/, '').split('/').filter(Boolean).pop() || u.hostname || fallback;
  } catch {
    return fallback;
  }
}

async function processCaptureStepsJob(jobId: string, config: any) {
  const { tenantId, projectId, testCaseId } = config as { tenantId: string; projectId: string; testCaseId: string };
  const screenshotDir = config.screenshotDir || platformConfig.screenshotDir || process.env.SCREENSHOT_DIR || '/tmp/zeta-screenshots';
  await fs.mkdir(screenshotDir, { recursive: true });

  await withTenant(tenantId, (tx) =>
    tx.platformJob.update({ where: { id: jobId }, data: { status: 'RUNNING', startedAt: new Date(), progress: 2 } })
  );

  let browser: any;
  try {
    await updateJobProgress(tenantId, jobId, 5, 'Loading test case');
    const [tc, crawlCfg, tenantStorageCfg] = await Promise.all([
      AppBrain.getTestCaseForExecution(tenantId, testCaseId) as Promise<any>,
      AppBrain.getCrawlConfig(tenantId, projectId).catch(() => null),
      getTenantStorageConfig(tenantId).catch(() => null),
    ]);
    if (!tc) throw new Error('Test case not found');

    const rawSteps = Array.isArray(tc.steps) ? tc.steps : [];
    if (rawSteps.length === 0) throw new Error('Test case has no steps to capture');

    // Resolve env tokens from testDataInline so {{fze.env.URL}} → concrete URL before execution plan
    const _tdi = ((tc as any).testDataInline ?? {}) as Record<string, unknown>;
    const _tdiStartUrl = typeof _tdi['startUrl'] === 'string' && (_tdi['startUrl'] as string).startsWith('http')
      ? (_tdi['startUrl'] as string) : '';
    const _recordedUrls = Array.isArray(_tdi['recordedUrls']) ? (_tdi['recordedUrls'] as unknown[]).map(String) : [];
    const _recordedStartUrl = _recordedUrls.find((u) => u.startsWith('http')) ?? '';

    const baseUrl = config.baseUrl || _tdiStartUrl || _recordedStartUrl || tc?.screen?.project?.appUrl || crawlCfg?.appUrl || '';

    // Build env token → concrete value map from testDataInline for step text substitution
    const _envTokenMap: Record<string, string> = {};
    for (const [k, v] of Object.entries(_tdi)) {
      if (typeof v === 'string' && v.startsWith('http')) _envTokenMap[k] = v;
    }
    if (_recordedStartUrl && !_envTokenMap['startUrl']) _envTokenMap['startUrl'] = _recordedStartUrl;

    // Replace {{fze.env.X}} and [env:X] tokens in step texts with concrete values from testDataInline.
    // IMPORTANT: only substitute if we have a concrete value for that key.
    // Fall back to [key] placeholder — never fall back to baseUrl (would inject URL into credential fields).
    const resolveStepText = (text: string): string =>
      text
        .replace(/\{\{\s*fze\.env\.([A-Za-z_][A-Za-z0-9_]*)\s*\}\}/g, (_m, k) => _envTokenMap[k] ?? `[${k}]`)
        .replace(/\[env:([A-Za-z_][A-Za-z0-9_]*)\]/g, (_m, k) => _envTokenMap[k] ?? `[${k}]`);

    const stepTexts = rawSteps.map(captureStepText).filter(Boolean).map(resolveStepText);
    if (stepTexts.length === 0) throw new Error('Test case has no steps to capture');

    let plan = await generateExecutionPlan(tenantId, { ...tc, steps: stepTexts }, baseUrl).catch(() => []);
    plan = Array.isArray(plan)
      ? plan.filter(Boolean).map((step: any, idx: number) => ({ ...step, stepNumber: idx + 1 }))
      : [];
    if (plan.length === 0) plan = fallbackExecutionPlan(rawSteps);
    if (plan.length === 0) throw new Error('Could not generate execution plan from steps');

    const { chromium } = await import('playwright');
    browser = await chromium.launch({
      headless: true,
      ...(process.env.PLAYWRIGHT_EXECUTABLE_PATH ? { executablePath: process.env.PLAYWRIGHT_EXECUTABLE_PATH } : {}),
    });
    let parsedStorageState: any;
    try {
      parsedStorageState = config.storageState ? JSON.parse(config.storageState) : crawlCfg?.storageState ? JSON.parse(crawlCfg.storageState) : undefined;
    } catch {
      parsedStorageState = undefined;
    }
    const context = await browser.newContext({ viewport: { width: 1280, height: 800 }, ...(parsedStorageState ? { storageState: parsedStorageState } : {}) });
    const page = await context.newPage();
    const aiHealingEnabled = await isFeatureEnabled(tenantId, 'AI_SELF_HEALING').catch(() => false);

    const stepResults: CaptureStepResult[] = [];
    const screens: Array<{ stepNumber: number; screenId: string; url: string; screenshotPath?: string; screenshotUrl?: string; isNewScreen: boolean }> = [];
    const locators: any[] = [];
    const seenLocatorExpressions = new Set<string>();
    const seenScreenKeys = new Set<string>();
    let lastStateKey = '';

    for (let i = 0; i < plan.length; i++) {
      const step = plan[i];
      const description = step.description || stepTexts[i] || `Step ${i + 1}`;
      const stepNumber = i + 1;
      const result: CaptureStepResult = { stepNumber, description, status: 'FAILED' };
      await withTenant(tenantId, (tx) =>
        tx.platformJob.update({
          where: { id: jobId },
          data: {
            progress: Math.min(90, Math.round((i / Math.max(plan.length, 1)) * 90)),
            result: { currentStep: { stepNumber, description } } as any,
            heartbeatAt: new Date(),
          },
        })
      ).catch(() => { });

      try {
        const heal = await executeExecutionStep(page, step, aiHealingEnabled ? { healCtx: { tenantId, projectId } } : {});
        if (heal.healed) {
          result.healed = true;
          result.healTier = heal.healedVia;
          result.healReasoning = heal.healReasoning;
        }
        await page.waitForLoadState('networkidle', { timeout: parseInt(process.env.TEST_NETWORK_IDLE_TIMEOUT_MS ?? '2000', 10) }).catch(() => { });
        result.status = result.healed ? 'HEALED' : 'PASSED';
      } catch (err: any) {
        result.status = step.optional ? 'SKIPPED' : 'FAILED';
        result.error = String(err?.message ?? err).slice(0, 300);
      }

      const currentUrl = page.url();
      result.url = currentUrl;
      const ssFilename = `${testCaseId}-capture-step-${stepNumber}${result.status === 'FAILED' ? '-err' : ''}.png`;
      const ssPath = path.join(screenshotDir, ssFilename);
      await page.screenshot({ path: ssPath, fullPage: false }).catch(() => { });
      result.screenshotPath = ssPath;

      let cloudUrl: string | undefined;
      if (tenantStorageCfg) {
        const key = buildArtifactKey(tenantId, projectId, 'screenshots', ssFilename, tenantStorageCfg.prefix);
        cloudUrl = (await uploadScreenshotToTenantStorage(tenantStorageCfg, ssPath, key).catch(() => undefined)) ?? undefined;
        if (cloudUrl) result.screenshotUrl = cloudUrl;
      }

      const elements = await domFallbackElements(page).catch(() => []);
      result.elementsCount = elements.length;
      const stateHash = hashContent(elements.map((el) => `${el.role}:${el.meaning}:${el.notes ?? ''}`).join('|'));
      const screenKey = `${currentUrl || 'about:blank'}:${stateHash}`;
      const shouldPersistScreen = currentUrl && currentUrl !== 'about:blank' && (i === 0 || screenKey !== lastStateKey || !seenScreenKeys.has(screenKey));
      lastStateKey = screenKey;

      if (shouldPersistScreen) {
        const screenName = screenNameFromUrl(currentUrl, tc.title ?? `Step ${stepNumber}`);
        const screen = await AppBrain.saveScreen(tenantId, projectId, screenName, currentUrl, jobId, i === 0 ? 'ENTRY_POINT' : 'LINK_FOLLOW').catch(() => null) as any;
        if (screen?.id) {
          result.screenId = screen.id;
          await AppBrain.saveScreenshot(tenantId, projectId, screen.id, ssPath, screen.version ?? 1, 'DESKTOP', cloudUrl).catch(() => { });
          for (const el of elements) {
            await AppBrain.saveElement(
              tenantId, projectId, screen.id,
              el.meaning, el.role, el.expectedData, el.notes,
              undefined, true, el.boundingRect, el.ariaState, el.parentLandmark,
            ).catch(() => { });
          }
          for (const loc of elementsToPlaywrightLocators(elements as any)) {
            if (seenLocatorExpressions.has(loc.expression)) continue;
            seenLocatorExpressions.add(loc.expression);
            locators.push(loc);
          }
          if (!seenScreenKeys.has(screenKey)) {
            screens.push({ stepNumber, screenId: screen.id, url: currentUrl, screenshotPath: ssPath, screenshotUrl: cloudUrl, isNewScreen: true });
            seenScreenKeys.add(screenKey);
          }
        }
      }

      stepResults.push(result);
      if (result.status === 'FAILED') break;
    }

    if (locators.length > 0) {
      await AppBrain.saveLocatorsJson(tenantId, testCaseId, locators as any).catch(() => { });
    }

    const firstScreen = screens[0];
    await withTenant(tenantId, (tx) =>
      (tx as any).testCase.update({
        where: { id: testCaseId },
        data: {
          ...(firstScreen?.screenId ? { screenId: firstScreen.screenId } : {}),
          captureStepScreenshots: stepResults as any,
        },
      })
    ).catch(() => { });

    const finalResult = {
      stepResults,
      screens,
      locators,
      totalSteps: plan.length,
      capturedScreens: screens.length,
    };
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'DONE', progress: 100, result: finalResult as any, finishedAt: new Date(), heartbeatAt: new Date() },
      })
    );
  } catch (err: any) {
    const error = String(err?.message ?? err);
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', error, finishedAt: new Date(), heartbeatAt: new Date() } })
    ).catch(() => { });
    logger.error({ jobId, testCaseId, err: error }, 'processCaptureStepsJob failed');
    throw err;
  } finally {
    await browser?.close?.().catch(() => { });
  }
}

// Registers the type→processor dispatch table used by the durable worker (both the
// enqueueJob() fast-path kick and startJobWorkerLoop()'s poll). Declared at module
// bottom so it can reference every process*Job function by name; function
// declarations are hoisted, so this could technically live anywhere in the file,
// but keeping it here alongside the last definition keeps the mapping easy to audit
// against the actual functions above.
jobDispatchTable = {
  CRAWL: processJob,
  CAPTURE_STEPS: processCaptureStepsJob,
  DISCOVER: processDiscoverJob,
  MONITOR: processMonitorJob,
  NATIVE_CRAWL: processNativeCrawlJob,
  API_TESTS: processApiTestsJob,
  SECURITY_TESTS: processSecurityTestsJob,
  INTEGRATION_TESTS: processIntegrationTestsJob,
  FIGMA_IMPORT: processFigmaImportJob,
  GENERATE_SKILL_FILES: processGenerateSkillFilesJob,
  GENERATE_TESTS_FROM_REQS: processGenerateTestsFromReqsJob,
  GENERATE_SCREEN_TESTS: processGenerateScreenTestsJob,
  RUN_TEST_SUITE: processRunTestSuiteJob,
  REPO_SCAN: processRepoScanJob,
  IMPORT_TESTS: processImportTestsJob,
  BATCH_GENERATE_ALIGNED: processBatchGenerateAlignedJob,
  CREATE_PR: processCreatePRJob,
  NEW_PAGES_PIPELINE: processNewPagesPipelineJob,
  // The following job types are processed in-process by the API (never QUEUED).
  // If the worker claims one, the API crashed after creating it — mark failed so it's visible.
  GENERATE_TESTS: async (jobId: string, _config: any) => {
    logger.error({ jobId }, '[job-worker] GENERATE_TESTS claimed by worker — API did not process it in-process');
    await prismaAdmin.platformJob.update({
      where: { id: jobId },
      data: { status: 'FAILED', error: 'Test generation was not completed by the API — please retry from the UI.', finishedAt: new Date() },
    }).catch(() => { });
  },
  GENERATE_DATA: async (jobId: string, _config: any) => {
    logger.error({ jobId }, '[job-worker] GENERATE_DATA claimed by worker — API did not process it in-process');
    await prismaAdmin.platformJob.update({
      where: { id: jobId },
      data: { status: 'FAILED', error: 'Data generation was not completed by the API — please retry from the UI.', finishedAt: new Date() },
    }).catch(() => { });
  },
  GENERATE_SCRIPT: async (jobId: string, _config: any) => {
    logger.error({ jobId }, '[job-worker] GENERATE_SCRIPT claimed by worker — API did not process it in-process');
    await prismaAdmin.platformJob.update({
      where: { id: jobId },
      data: { status: 'FAILED', error: 'Script generation was not completed by the API — please retry from the UI.', finishedAt: new Date() },
    }).catch(() => { });
  },
  GENERATE_TEST_PLAN: async (jobId: string, _config: any) => {
    logger.error({ jobId }, '[job-worker] GENERATE_TEST_PLAN claimed by worker — API did not process it in-process');
    await prismaAdmin.platformJob.update({
      where: { id: jobId },
      data: { status: 'FAILED', error: 'Test plan generation was not completed by the API — please retry from the UI.', finishedAt: new Date() },
    }).catch(() => { });
  },
};

async function processRepoScanJob(jobId: string, config: any) {
  const { repoConnectionId, tenantId, projectId } = config;
  const jobStart = new Date();
  try {
    await updateJobProgress(tenantId, jobId, 5, 'Starting repo scan');

    const {
      fetchRepoTree, fetchFileContent, detectFramework,
      extractTestIds, extractStyleProfile,
      buildPageObjectInventory, resolveFixtureAliasesFromImports,
    } = await import('@detiq/agents');

    const conn = await withTenant(tenantId, (tx) =>
      tx.gitHubRepoConnection.findFirst({ where: { id: repoConnectionId, tenantId } })
    ) as any;
    if (!conn) throw new Error(`RepoConnection ${repoConnectionId} not found`);

    // Decrypt token
    const { decrypt } = await import('@detiq/core');
    const token = decrypt(conn.tokenEnc);

    // Mark scanning
    await withTenant(tenantId, (tx) =>
      tx.gitHubRepoConnection.update({ where: { id: repoConnectionId }, data: { scanStatus: 'SCANNING', scanError: null } })
    );

    await updateJobProgress(tenantId, jobId, 15, 'Fetching repo tree');
    // Use HEAD so it always resolves to the actual default branch regardless of stored value
    const treeResult = await fetchRepoTree(token, conn.repoFullName, 'HEAD');
    if (treeResult.truncated) {
      console.warn(`[repo-scan] ${conn.repoFullName}: tree truncated (>100k blobs) — detection may be incomplete`);
    }

    const frameworks = detectFramework(treeResult.paths);
    const primary = frameworks[0];
    await updateJobProgress(tenantId, jobId, 30, `Detected framework: ${primary?.framework ?? 'unknown'}`);

    // Prefer framework-detected glob when stored glob is the TS default (never explicitly set)
    const storedGlob = conn.testGlob || '';
    const testGlob = (storedGlob && storedGlob !== '**/*.spec.ts')
      ? storedGlob
      : (primary?.testGlob || storedGlob || '**/*.spec.ts');
    const framework = primary?.framework ?? 'unknown';
    const language = primary?.language ?? 'typescript';

    // Filter test files
    const { minimatch } = await import('minimatch').catch(() => ({ minimatch: null }));
    const isPythonTestFile = (p: string) =>
      p.endsWith('.py') && !p.endsWith('__init__.py') && !p.endsWith('conftest.py') &&
      (p.endsWith('_test.py') || /\/test_[^/]+\.py$/.test(p) || /^test_[^/]+\.py$/.test(p) ||
        p.includes('/tests/') || p.includes('/test/'));
    // Java: testGlob matches all *.java — further filter to actual test classes only
    // (excludes helpers, base classes, utilities that don't follow *Test.java/*Tests.java naming)
    const isJavaTestFile = (p: string) =>
      p.endsWith('.java') && (
        /[A-Z]\w*Tests?\.java$/.test(p) ||
        (/\/tests?\//.test(p) && /[A-Z][a-z]/.test(p.split('/').pop() ?? ''))
      );
    const testPaths = treeResult.paths.filter((p) => {
      if (language === 'java') return isJavaTestFile(p);
      if (language === 'python') return isPythonTestFile(p);
      if (minimatch) return minimatch(p, testGlob, { matchBase: true });
      return p.endsWith('.spec.ts') || p.endsWith('.spec.js') || p.endsWith('.test.ts') || p.endsWith('.test.js') || p.endsWith('_test.py') || p.includes('/test_');
    }).slice(0, 500); // cap at 500 files

    await updateJobProgress(tenantId, jobId, 40, `Found ${testPaths.length} test files`);

    if (testPaths.length === 0) {
      logger.warn({ jobId, repoConnectionId, framework }, '[repo-scan] no test files matched glob — scan complete with 0 results');
      await withTenant(tenantId, (tx) =>
        tx.gitHubRepoConnection.update({
          where: { id: repoConnectionId },
          data: { scanStatus: 'DONE', testFileCount: 0, extractedIdCount: 0, lastScannedAt: new Date(), lastSyncAt: new Date() },
        })
      );
      await updateJobProgress(tenantId, jobId, 100, 'Done: 0 test files found');
      await withTenant(tenantId, (tx) =>
        tx.platformJob.update({ where: { id: jobId }, data: { status: 'DONE', finishedAt: new Date(), result: { testFileCount: 0, extractedIdCount: 0 } as any } })
      );
      return;
    }

    // Extract test IDs from each file
    const allExtracted: any[] = [];
    const styleSampleFiles: Array<{ path: string; content: string }> = [];
    let processed = 0;

    // Delete old test files for this connection
    await withTenant(tenantId, (tx) => tx.repoTestFile.deleteMany({ where: { repoConnectionId } }));

    for (const filePath of testPaths) {
      const content = await fetchFileContent(token, conn.repoFullName, filePath);
      if (!content) continue;

      // G-07: multi-file fixture tracing — resolve non-obvious aliases before extraction
      const extraAliases = await resolveFixtureAliasesFromImports(
        content, token, conn.repoFullName, filePath,
      ).catch(() => new Set<string>());
      const extracted = extractTestIds(content, filePath, framework, extraAliases);
      allExtracted.push(...extracted);

      await withTenant(tenantId, (tx) =>
        tx.repoTestFile.create({
          data: {
            tenantId,
            projectId,
            repoConnectionId,
            filePath,
            framework,
            extractedIds: extracted.map((e) => ({ name: e.name, normalizedName: e.normalizedName, lineNumber: e.lineNumber })) as any,
            rawSample: content.slice(0, 2000),
            lastScannedAt: new Date(),
          },
        })
      );

      if (styleSampleFiles.length < 5) styleSampleFiles.push({ path: filePath, content: content.slice(0, 3000) });
      processed++;

      if (processed % 10 === 0) {
        await updateJobProgress(tenantId, jobId, 40 + Math.floor((processed / testPaths.length) * 40), `Processed ${processed}/${testPaths.length} files`);
      }
    }

    // Fetch and parse framework config files (playwright.config.ts, cypress.config.ts, tsconfig.json)
    // Fetch and parse framework config files for all supported frameworks.
    // Pure regex/JSON — no LLM. Results override LLM inference for path detection.
    // Covers: playwright, cypress, jest, pytest, junit/maven/gradle, selenium-java/py,
    //         webdriverio, cucumber — all frameworks in DetectedFramework.
    const configPaths: {
      testDir?: string; fixturesFolder?: string; baseURL?: string;
      tsconfigPaths?: Record<string, string[]>; tsconfigBaseUrl?: string;
      javaTestSourceDir?: string; jestModuleNameMapper?: Record<string, string>;
    } = {};
    try {
      const pathSet = new Set(treeResult.paths as string[]);
      const lang = framework.includes('java') ? 'java'
        : (framework.includes('py') || framework === 'pytest') ? 'python' : 'js';

      const configFilesToFetch: Array<{ path: string; type: string; dirPrefix: string }> = [];
      const tryAdd = (candidates: string[], type: string) => {
        for (const c of candidates) { if (pathSet.has(c)) { configFilesToFetch.push({ path: c, type, dirPrefix: '' }); break; } }
      };
      // Monorepo fallback: when root-level config is absent, find shallowest match anywhere in tree.
      // The dirPrefix (e.g. 'packages/e2e/') is prepended to testDir so paths are repo-root-relative.
      const scanTree = (pattern: RegExp, type: string) => {
        if (configFilesToFetch.some(f => f.type === type)) return;
        const all = (treeResult.paths as string[]).filter(p => pattern.test(p)).sort((a, b) => a.split('/').length - b.split('/').length);
        if (all[0]) {
          const lastSlash = all[0].lastIndexOf('/');
          const dirPrefix = lastSlash > 0 ? all[0].slice(0, lastSlash + 1) : '';
          configFilesToFetch.push({ path: all[0], type, dirPrefix });
        }
      };

      if (lang === 'js') {
        if (framework.startsWith('playwright')) {
          tryAdd(['playwright.config.ts', 'playwright.config.js'], 'playwright');
          scanTree(/(?:^|\/)playwright\.config\.(ts|js|mts)$/, 'playwright');
        }
        if (framework.startsWith('cypress')) {
          tryAdd(['cypress.config.ts', 'cypress.config.js'], 'cypress');
          scanTree(/(?:^|\/)cypress\.config\.(ts|js)$/, 'cypress');
        }
        if (framework === 'jest-ts' || framework === 'jest-js') {
          tryAdd(['jest.config.ts', 'jest.config.js', 'jest.config.mjs'], 'jest');
          scanTree(/(?:^|\/)jest\.config\.(ts|js|mjs|cjs)$/, 'jest');
          tryAdd(['package.json'], 'package-json');
        }
        if (framework === 'webdriverio') {
          tryAdd(['wdio.conf.ts', 'wdio.conf.js', 'wdio.config.ts', 'wdio.config.js'], 'wdio');
          scanTree(/(?:^|\/)wdio\.(?:conf|config)\.(ts|js|mjs)$/, 'wdio');
        }
        if (framework === 'cucumber' || framework === 'gherkin') {
          tryAdd(['.cucumber.js', '.cucumber.cjs', 'cucumber.json', '.cucumber.yaml', '.cucumber.yml'], 'cucumber');
          tryAdd(['package.json'], 'package-json');
        }
        tryAdd(['tsconfig.json'], 'tsconfig');
        scanTree(/(?:^|\/)tsconfig(?:\.base)?\.json$/, 'tsconfig');
      } else if (lang === 'python') {
        tryAdd(['pytest.ini'], 'pytest-ini');
        scanTree(/(?:^|\/)pytest\.ini$/, 'pytest-ini');
        tryAdd(['pyproject.toml'], 'pyproject-toml');
        scanTree(/(?:^|\/)pyproject\.toml$/, 'pyproject-toml');
        tryAdd(['setup.cfg'], 'setup-cfg');
      } else if (lang === 'java') {
        tryAdd(['pom.xml'], 'pom-xml');
        scanTree(/(?:^|\/)pom\.xml$/, 'pom-xml');
        tryAdd(['build.gradle.kts', 'build.gradle'], 'build-gradle');
        scanTree(/(?:^|\/)build\.gradle(?:\.kts)?$/, 'build-gradle');
        tryAdd(['testng.xml', 'src/test/resources/testng.xml'], 'testng-xml');
      }

      // Inline parsers — mirror parseFrameworkConfig from @detiq/agents (kept inline to avoid build dep)
      const normDir = (s: string) => s.replace(/^\.\//, '').replace(/\/$/, '');
      const globRoot = (pattern: string): string | null => {
        const root = pattern.replace(/\/\*\*.*$/, '').replace(/\/\*[^*].*$/, '');
        return (root && !root.startsWith('*') && !root.startsWith('{')) ? normDir(root) : null;
      };

      const fetched = await Promise.all(
        configFilesToFetch.map(async ({ path: cfgPath, type, dirPrefix }) => {
          const content = await fetchFileContent(token, conn.repoFullName, cfgPath).catch(() => null);
          if (!content) return null;
          const r: typeof configPaths = {};

          if (type === 'playwright') {
            const m = content.match(/\btestDir\s*:\s*['"`]([^'"`]+)['"`]/);
            if (m) r.testDir = normDir(m[1]);
            const bm = content.match(/\bbaseURL\s*:\s*['"`]([^'"`]+)['"`]/);
            if (bm) r.baseURL = bm[1];
          } else if (type === 'cypress') {
            const sm = content.match(/\bspecPattern\s*:\s*['"`]([^'"`]+)['"`]/);
            if (sm) { const gr = globRoot(sm[1]); if (gr) r.testDir = gr; }
            const fm = content.match(/\bfixturesFolder\s*:\s*['"`]([^'"`]+)['"`]/);
            if (fm) r.fixturesFolder = normDir(fm[1]);
            const bm = content.match(/\bbaseUrl\s*:\s*['"`]([^'"`]+)['"`]/);
            if (bm) r.baseURL = bm[1];
          } else if (type === 'jest') {
            const rm = content.match(/\broots\s*:\s*\[([^\]]+)\]/);
            if (rm) { const fr = rm[1].match(/['"`]([^'"`]+)['"`]/)?.[1]; if (fr) r.testDir = fr.replace(/<rootDir>\/?/, '').replace(/\/$/, '') || 'src'; }
            if (!r.testDir) { const tm = content.match(/\btestMatch\s*:\s*\[([^\]]+)\]/); if (tm) { const fp = tm[1].match(/['"`]([^'"`]+)['"`]/)?.[1]; if (fp) { const gr = globRoot(fp.replace(/<rootDir>\/?/, '')); if (gr) r.testDir = gr; } } }
            const mm = content.match(/\bmoduleNameMapper\s*:\s*\{([^}]+)\}/s);
            if (mm) { const map: Record<string, string> = {}; for (const [, k, v] of mm[1].matchAll(/['"`]([^'"`]+)['"`]\s*:\s*['"`]([^'"`]+)['"`]/g)) map[k] = v; if (Object.keys(map).length) r.jestModuleNameMapper = map; }
          } else if (type === 'package-json') {
            try { const p = JSON.parse(content) as any; const j = p?.jest ?? {}; if (Array.isArray(j.roots) && j.roots[0]) r.testDir = String(j.roots[0]).replace(/<rootDir>\/?/, '').replace(/\/$/, '') || 'src'; if (!r.testDir && Array.isArray(j.testMatch) && j.testMatch[0]) { const gr = globRoot(String(j.testMatch[0]).replace(/<rootDir>\/?/, '')); if (gr) r.testDir = gr; } if (j.moduleNameMapper && typeof j.moduleNameMapper === 'object') r.jestModuleNameMapper = j.moduleNameMapper; } catch { /* ignore */ }
          } else if (type === 'tsconfig') {
            try { const stripped = content.replace(/\/\/[^\n]*/g, '').replace(/,(\s*[}\]])/g, '$1'); const p = JSON.parse(stripped) as any; const opts = p?.compilerOptions ?? {}; if (opts.paths && typeof opts.paths === 'object') r.tsconfigPaths = opts.paths; if (typeof opts.baseUrl === 'string') r.tsconfigBaseUrl = opts.baseUrl; } catch { /* ignore */ }
          } else if (type === 'pytest-ini') {
            const m = content.match(/^\s*testpaths\s*=\s*(.+)$/m);
            if (m) { const first = m[1].trim().split(/\s+/)[0]; if (first) r.testDir = normDir(first); }
          } else if (type === 'pyproject-toml') {
            const m = content.match(/testpaths\s*=\s*\[([^\]]+)\]/);
            if (m) { const first = m[1].match(/['"]([^'"]+)['"]/)?.[1]; if (first) r.testDir = normDir(first); }
            if (!r.testDir) { const m2 = content.match(/testpaths\s*=\s*['"]([^'"]+)['"]/); if (m2) r.testDir = normDir(m2[1]); }
          } else if (type === 'setup-cfg') {
            const m = content.match(/^\s*testpaths\s*=\s*(.+)$/m);
            if (m) { const first = m[1].trim().split(/[\s,]+/)[0]; if (first) r.testDir = normDir(first); }
          } else if (type === 'pom-xml') {
            const m = content.match(/<testSourceDirectory>\s*([^<]+)\s*<\/testSourceDirectory>/);
            if (m) r.javaTestSourceDir = normDir(m[1].trim());
          } else if (type === 'build-gradle') {
            const m = content.match(/srcDir[s]?\s*[=([]?\s*['"`]([^'"`]+)['"`]/) ?? content.match(/srcDir\s*\(\s*['"`]([^'"`]+)['"`]\s*\)/);
            if (m) r.javaTestSourceDir = normDir(m[1]);
          } else if (type === 'wdio') {
            const sm = content.match(/\bspecs\s*:\s*\[([^\]]+)\]/);
            if (sm) { const fp = sm[1].match(/['"`]([^'"`]+)['"`]/)?.[1]; if (fp) { const gr = globRoot(fp); if (gr) r.testDir = gr; } }
            const bm = content.match(/\bbaseUrl\s*:\s*['"`]([^'"`]+)['"`]/);
            if (bm) r.baseURL = bm[1];
          } else if (type === 'cucumber') {
            const pm = content.match(/\bpaths\s*:\s*\[([^\]]+)\]/);
            if (pm) { const fp = pm[1].match(/['"`]([^'"`]+)['"`]/)?.[1]; if (fp) { const gr = globRoot(fp); if (gr) r.testDir = gr; } }
            if (!r.testDir) { try { const p = JSON.parse(content) as any; const paths = p?.default?.paths ?? p?.paths; if (Array.isArray(paths) && paths[0]) { const gr = globRoot(String(paths[0])); if (gr) r.testDir = gr; } } catch { /* not JSON */ } }
          }
          // Monorepo workspace prefix: config found at e.g. packages/frontend/playwright.config.ts.
          // testDir from config is relative to that file ('tests' or 'e2e'). Prepend the directory
          // so the final path is repo-root-relative ('packages/frontend/tests').
          if (dirPrefix && r.testDir && !r.testDir.startsWith(dirPrefix)) {
            r.testDir = dirPrefix + r.testDir;
          }
          if (dirPrefix && r.javaTestSourceDir && !r.javaTestSourceDir.startsWith(dirPrefix)) {
            r.javaTestSourceDir = dirPrefix + r.javaTestSourceDir;
          }
          return r;
        })
      );

      for (const parsed of fetched) {
        if (!parsed) continue;
        if (parsed.testDir && !configPaths.testDir) configPaths.testDir = parsed.testDir;
        if (parsed.fixturesFolder && !configPaths.fixturesFolder) configPaths.fixturesFolder = parsed.fixturesFolder;
        if (parsed.baseURL && !configPaths.baseURL) configPaths.baseURL = parsed.baseURL;
        if (parsed.tsconfigPaths && !configPaths.tsconfigPaths) configPaths.tsconfigPaths = parsed.tsconfigPaths;
        if (parsed.tsconfigBaseUrl && !configPaths.tsconfigBaseUrl) configPaths.tsconfigBaseUrl = parsed.tsconfigBaseUrl;
        if (parsed.javaTestSourceDir && !configPaths.javaTestSourceDir) configPaths.javaTestSourceDir = parsed.javaTestSourceDir;
        if (parsed.jestModuleNameMapper && !configPaths.jestModuleNameMapper) configPaths.jestModuleNameMapper = parsed.jestModuleNameMapper;
      }
    } catch { /* non-fatal */ }

    // Extract style profile for aligned generation
    // For Java, include main source paths so LLM can detect page object location in src/main/java/
    // Fall back to existing styleProfile from DB if extraction fails or returns null.
    let styleProfile: any = conn.styleProfile ?? null;
    if (styleSampleFiles.length > 0) {
      try {
        const mainSourcePaths = language === 'java'
          ? treeResult.paths.filter((p: string) => /^src\/main\/java\//.test(p)).slice(0, 50)
          : [];
        // For TS/JS, also include non-test source files so LLM can detect pageObjectLocation
        // (e.g. src/advantage/pages/ in playwright-sample-project)
        const testPathSet = new Set(testPaths);
        const sourceFilePaths = (language === 'typescript' || language === 'javascript')
          ? treeResult.paths
            .filter((p: string) =>
              (p.endsWith('.ts') || p.endsWith('.js')) &&
              !testPathSet.has(p) &&
              !p.endsWith('.d.ts') &&
              !p.includes('node_modules/') &&
              !p.includes('dist/') &&
              !p.includes('.config.'))
            .slice(0, 60)
          : [];
        const allPathsForProfile = [...testPaths, ...mainSourcePaths, ...sourceFilePaths].slice(0, 200);
        // Pass full tree paths separately so detectDataDirs can find testdata/*.json, fixtures/*.yaml etc.
        const freshProfile = await extractStyleProfile(styleSampleFiles, framework, language, tenantId, projectId, jobId, allPathsForProfile, treeResult.paths);
        if (freshProfile) styleProfile = freshProfile;
      } catch { /* non-fatal */ }
    }

    // Merge config file ground-truth into styleProfile (Priority 0 — highest, no LLM inference needed).
    // Java: javaTestSourceDir also sets testLocation (equivalent field for Java).
    if (styleProfile && Object.keys(configPaths).length > 0) {
      if (configPaths.testDir) styleProfile.configTestDir = configPaths.testDir;
      if (configPaths.javaTestSourceDir) {
        styleProfile.configTestDir = configPaths.javaTestSourceDir;
        // Also set testLocation if not already present from LLM scan
        if (!styleProfile.testLocation) styleProfile.testLocation = configPaths.javaTestSourceDir;
      }
      if (configPaths.fixturesFolder) styleProfile.configFixturesFolder = configPaths.fixturesFolder;
      if (configPaths.baseURL) styleProfile.configBaseURL = configPaths.baseURL;
      if (configPaths.tsconfigPaths) styleProfile.tsconfigPaths = configPaths.tsconfigPaths;
      if (configPaths.tsconfigBaseUrl) styleProfile.tsconfigBaseUrl = configPaths.tsconfigBaseUrl;
      if (configPaths.jestModuleNameMapper) styleProfile.jestModuleNameMapper = configPaths.jestModuleNameMapper;
    }

    // Extract page object inventory — finds POM/page classes in non-test files
    // Fall back to existing inventory from DB if fetch fails.
    let pageObjectInventory: any[] = (conn.pageObjectInventory as any[]) ?? [];
    try {
      const freshInventory = await buildPageObjectInventory(token, conn.repoFullName, treeResult.paths, framework);
      if (freshInventory.length > 0) {
        pageObjectInventory = freshInventory;
        await updateJobProgress(tenantId, jobId, 82, `Found ${pageObjectInventory.length} page objects`);
      }
    } catch { /* non-fatal */ }

    // Run coverage mapping
    await updateJobProgress(tenantId, jobId, 85, 'Running coverage mapping');
    const { buildCoverageMap } = await import('@detiq/agents');
    const testCases = await withTenant(tenantId, (tx) =>
      tx.testCase.findMany({ where: { projectId, tenantId, status: { not: 'RETIRED' } }, select: { id: true, title: true, ref: true } })
    ) as any[];

    const mappings = await buildCoverageMap(testCases, allExtracted, tenantId);

    // Upsert CoverageMapping records (skip manualOverride=true)
    for (const m of mappings) {
      const existing = await withTenant(tenantId, (tx) =>
        tx.coverageMapping.findUnique({ where: { testCaseId: m.testCaseId } })
      ) as any;
      if (existing?.manualOverride) continue;

      await withTenant(tenantId, (tx) =>
        tx.coverageMapping.upsert({
          where: { testCaseId: m.testCaseId },
          create: { tenantId, projectId, testCaseId: m.testCaseId, repoConnectionId, status: m.status, matchedFilePath: m.matchedFilePath ?? null, scriptTestName: m.scriptTestName ?? null, lineNumber: m.lineNumber ?? null, confidence: m.confidence, matchMethod: m.matchMethod },
          update: { status: m.status, matchedFilePath: m.matchedFilePath ?? null, scriptTestName: m.scriptTestName ?? null, lineNumber: m.lineNumber ?? null, confidence: m.confidence, matchMethod: m.matchMethod },
        })
      );
    }

    const mapped = mappings.filter((m) => m.status === 'MAPPED').length;
    const unmapped = mappings.filter((m) => m.status === 'UNMAPPED').length;
    const ambiguous = mappings.filter((m) => m.status === 'AMBIGUOUS').length;
    const coveragePercent = testCases.length > 0 ? (mapped / testCases.length) * 100 : 0;

    // Write coverage snapshot
    await withTenant(tenantId, (tx) =>
      tx.coverageSnapshot.create({
        data: { tenantId, projectId, totalCases: testCases.length, mappedCases: mapped, unmappedCases: unmapped, ambiguousCases: ambiguous, coveragePercent },
      })
    );

    // For Java, if styleProfile still missing testStructurePattern, infer two-tier from tree paths.
    // This covers the case where both extractStyleProfile and buildPageObjectInventory failed.
    const resolvedStyleProfile: any = (() => {
      const sp = styleProfile ?? {};
      if (framework === 'selenium-java' && !sp.testStructurePattern) {
        const hasMavenMain = treeResult.paths.some((p: string) => /^src\/main\/java\//.test(p));
        if (hasMavenMain) return { ...sp, testStructurePattern: 'two-tier' };
      }
      return sp;
    })();

    // Update repo connection with results
    await withTenant(tenantId, (tx) =>
      tx.gitHubRepoConnection.update({
        where: { id: repoConnectionId },
        data: {
          scanStatus: 'DONE',
          detectedFramework: framework,
          configFilePath: primary?.configPath ?? null,
          testGlob: testGlob,
          testFileCount: testPaths.length,
          extractedIdCount: allExtracted.length,
          styleProfile: resolvedStyleProfile as any,
          pageObjectInventory: pageObjectInventory.length > 0 ? pageObjectInventory as any : undefined,
          treeSnapshot: JSON.stringify(treeResult.paths).slice(0, 100000),
          lastScannedAt: new Date(),
          lastSyncAt: new Date(),
        },
      })
    );
    try {
      const { generateRepoSkillFiles } = await import('@detiq/agents');
      await generateRepoSkillFiles(tenantId, projectId, {
        repoFullName: conn.repoFullName,
        framework,
        language,
        testGlob,
        styleProfile: resolvedStyleProfile,
        pageObjectInventory,
        testPaths,
        extractedTestCount: allExtracted.length,
        treeWasTruncated: treeResult.truncated,
        extractedTests: allExtracted,
      });
    } catch (e: any) {
      logger.warn({ jobId, repoConnectionId, err: e?.message }, '[repo-scan] repo skill files generation failed (non-fatal)');
    }

    await updateJobProgress(tenantId, jobId, 100, `Done: ${mapped} mapped, ${unmapped} unmapped`);
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'DONE', finishedAt: new Date(), result: { testFileCount: testPaths.length, extractedIdCount: allExtracted.length, mapped, unmapped, ambiguous, coveragePercent } as any } })
    );
    aggregateJobTokens(tenantId, jobId, projectId, jobStart).catch(console.error);
  } catch (err: any) {
    const cause = err?.cause?.message ?? (typeof err?.cause === 'string' ? err.cause : '');
    const msg = `${String(err?.message ?? err)}${cause ? ` — ${cause}` : ''}`;
    logger.error({ jobId, repoConnectionId, err: msg }, '[repo-scan] failed');
    await withTenant(tenantId, (tx) =>
      tx.gitHubRepoConnection.update({ where: { id: repoConnectionId }, data: { scanStatus: 'FAILED', scanError: msg.slice(0, 500) } })
    ).catch(() => { });
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', finishedAt: new Date(), error: msg.slice(0, 1000) } })
    ).catch(() => { });
  }
}

// ── BATCH_GENERATE_ALIGNED — generate framework-aligned scripts for UNMAPPED cases ──

async function processBatchGenerateAlignedJob(jobId: string, config: any) {
  const { tenantId, projectId, repoConnectionId, scope = 'UNMAPPED', testCaseIds, force = false } = config;
  try {
    await updateJobProgress(tenantId, jobId, 5, 'Loading repo connection');

    const conn = await withTenant(tenantId, (tx) =>
      (tx as any).gitHubRepoConnection.findFirst({ where: { id: repoConnectionId, tenantId } })
    ) as any;
    if (!conn) throw new Error(`RepoConnection ${repoConnectionId} not found`);

    const { decrypt } = await import('@detiq/core');

    const {
      generateScriptAligned, deriveRepoTargetPath, validateGeneratedScript,
      mapWithConcurrency, SCRIPT_GEN_CONCURRENCY,
    } = await import('@detiq/agents');

    const styleProfile = conn.styleProfile as any;
    const pageObjectInventory = (conn.pageObjectInventory as any[]) ?? [];
    const framework = conn.detectedFramework ?? 'playwright-ts';
    const language = styleProfile?.language ?? 'typescript';

    // Resolve which test cases to process
    let caseIds: string[];
    if (testCaseIds && testCaseIds.length > 0) {
      caseIds = testCaseIds;
    } else if (scope === 'UNMAPPED') {
      const mappings = await withTenant(tenantId, (tx) =>
        (tx as any).coverageMapping.findMany({
          where: { tenantId, projectId, repoConnectionId, status: 'UNMAPPED' },
          select: { testCaseId: true },
        })
      ) as any[];
      caseIds = mappings.map((m: any) => m.testCaseId);
    } else {
      const cases = await withTenant(tenantId, (tx) =>
        tx.testCase.findMany({
          where: { tenantId, projectId, status: { not: 'RETIRED' } },
          select: { id: true },
        })
      ) as any[];
      caseIds = cases.map((c: any) => c.id);
    }

    await updateJobProgress(tenantId, jobId, 15, `Generating scripts for ${caseIds.length} test cases`);

    let done = 0;
    const errors: string[] = [];

    // One LLM call per case — bounded worker pool instead of a strictly serial loop,
    // which made large batches take an hour or more.
    await mapWithConcurrency(caseIds, SCRIPT_GEN_CONCURRENCY, async (tcId: string) => {
      try {
        // Skip if script already exists and not forcing
        if (!force) {
          const existing = await withTenant(tenantId, (tx) =>
            (tx as any).testScript.findFirst({ where: { testCaseId: tcId } })
          );
          if (existing) { done++; return; }
        }

        // Get screen URL and ref for target path derivation
        const tc = await withTenant(tenantId, (tx) =>
          tx.testCase.findUnique({
            where: { id: tcId },
            select: { id: true, ref: true, title: true, screen: { select: { url: true } } },
          })
        );

        const screenUrl = (tc as any)?.screen?.url ?? '';
        const repoTargetPath = styleProfile
          ? deriveRepoTargetPath(screenUrl, styleProfile, framework)
          : deriveFallbackTargetPath((tc as any)?.ref ?? (tc as any)?.title ?? tcId, framework);

        const result = await generateScriptAligned({
          tenantId, projectId, testCaseId: tcId,
          styleProfile,
          pageObjectInventory,
          repoTargetPath,
          save: true,
        });

        // Validate the generated script
        if (result.code) {
          const validation = await validateGeneratedScript(result.code, framework, language);
          await withTenant(tenantId, (tx) =>
            (tx as any).testScript.updateMany({
              where: { testCaseId: tcId },
              data: {
                validationStatus: validation.ok ? 'PASS' : 'FAIL',
                validationErrors: validation.ok ? null : validation.errors.slice(0, 2000),
              },
            })
          );
        }

        done++;
      } catch (e: any) {
        errors.push(`${tcId}: ${String(e.message).slice(0, 100)}`);
      }

      if (done % 5 === 0) {
        await updateJobProgress(tenantId, jobId, 15 + Math.floor((done / caseIds.length) * 80), `Generated ${done}/${caseIds.length}`);
      }
    });

    refreshTestScriptSkillFile(tenantId, projectId).catch(console.error);

    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'DONE', finishedAt: new Date(), result: { done, errors: errors.slice(0, 20), total: caseIds.length } as any },
      })
    );
  } catch (err: any) {
    const msg = String(err?.message ?? err);
    logger.error({ jobId, err: msg }, '[batch-generate-aligned] failed');
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', finishedAt: new Date(), error: msg.slice(0, 1000) } })
    ).catch(() => { });
  }
}

// ── CREATE_PR — push generated scripts to GitHub as a PR ─────────────────────

async function processCreatePRJob(jobId: string, config: any) {
  const { tenantId, projectId, repoConnectionId, testScriptIds, crawlJobId } = config;
  try {
    await updateJobProgress(tenantId, jobId, 5, 'Loading connection');

    const conn = await withTenant(tenantId, (tx) =>
      (tx as any).gitHubRepoConnection.findFirst({ where: { id: repoConnectionId, tenantId } })
    ) as any;
    if (!conn) throw new Error(`RepoConnection ${repoConnectionId} not found`);

    const { decrypt } = await import('@detiq/core');
    const token = decrypt(conn.tokenEnc);

    const { pushScriptsAsPR, getFileContent, mergeSpecContent } = await import('@detiq/agents');

    const framework = conn.detectedFramework ?? 'playwright-ts';

    // Collect scripts + their target paths
    let scripts: any[];
    if (testScriptIds && testScriptIds.length > 0) {
      scripts = await withTenant(tenantId, (tx) =>
        (tx as any).testScript.findMany({
          where: { id: { in: testScriptIds }, tenantId },
          include: { testCase: { select: { title: true, ref: true, repoTargetPath: true } } },
        })
      );
    } else {
      // All scripts that haven't been pushed yet
      scripts = await withTenant(tenantId, (tx) =>
        (tx as any).testScript.findMany({
          where: { tenantId, projectId, pushedToPR: false, validationStatus: { in: ['PASS', 'PENDING'] } },
          include: { testCase: { select: { title: true, ref: true, repoTargetPath: true } } },
        })
      );
    }

    if (scripts.length === 0) throw new Error('No generated scripts found to push.');

    // For each file, check if it already exists in the repo and merge if so
    const files = await Promise.all(scripts.map(async (s: any) => {
      const path = s.testCase?.repoTargetPath
        ?? deriveFallbackTargetPath(s.testCase?.ref ?? s.testCase?.title ?? s.id, framework);
      let content = s.code;

      // Spec files: attempt to merge with existing content
      const isSpecFile = /\.(spec|test|cy)\.(ts|js|py|java)$/.test(path) || /_test\.py$/.test(path);
      if (isSpecFile) {
        const existing = await getFileContent(token, conn.repoFullName, path, conn.defaultBranch).catch(() => null);
        if (existing && existing.trim().length > 0) {
          logger.info({ path }, '[create-pr] merging new tests into existing spec file');
          content = mergeSpecContent(existing, s.code);
        }
      }

      return { path, content };
    }));

    if (files.length === 0) throw new Error('No scripts to push.');

    await updateJobProgress(tenantId, jobId, 20, `Pushing ${files.length} scripts to GitHub`);

    const now = new Date().toISOString().slice(0, 10);
    const branchName = `zeta/new-coverage-${now}-${jobId.slice(0, 8)}`;

    const prBody = buildPRBody(scripts, projectId);

    const prResult = await pushScriptsAsPR({
      token,
      repo: conn.repoFullName,
      baseBranch: conn.defaultBranch,
      branchName,
      files,
      prTitle: `ZeTa: Add test coverage for ${files.length} new scenario${files.length === 1 ? '' : 's'}`,
      prBody,
      draft: false,
    });

    // Record the PR
    const pr = await withTenant(tenantId, (tx) =>
      (tx as any).gitHubPR.create({
        data: {
          tenantId, projectId,
          repoConnectionId,
          prNumber: prResult.prNumber,
          prUrl: prResult.prUrl,
          branch: prResult.branch,
          headSha: prResult.headSha,
          status: 'OPEN',
          crawlJobId: crawlJobId ?? null,
          scriptCount: files.length,
          prBody: prBody.slice(0, 10000),
        },
      })
    ) as any;

    // Mark scripts as pushed
    const scriptIds = scripts.map((s: any) => s.id);
    await withTenant(tenantId, (tx) =>
      (tx as any).testScript.updateMany({
        where: { id: { in: scriptIds } },
        data: { pushedToPR: true, pushedPRId: pr.id },
      })
    );

    await notifyAdmins(tenantId, {
      type: 'GITHUB_PR_CREATED',
      title: 'ZeTa created a GitHub PR',
      body: `${files.length} new test script${files.length === 1 ? '' : 's'} pushed. PR #${prResult.prNumber}: ${prResult.prUrl}`,
      metadata: { prUrl: prResult.prUrl, prNumber: prResult.prNumber, scriptCount: files.length },
    });

    await updateJobProgress(tenantId, jobId, 100, `PR #${prResult.prNumber} created`);
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'DONE', finishedAt: new Date(), result: { prUrl: prResult.prUrl, prNumber: prResult.prNumber, scriptCount: files.length } as any },
      })
    );
  } catch (err: any) {
    const msg = String(err?.message ?? err);
    logger.error({ jobId, err: msg }, '[create-pr] failed');
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', finishedAt: new Date(), error: msg.slice(0, 1000) } })
    ).catch(() => { });
  }
}

function buildPRBody(scripts: any[], projectId: string): string {
  const lines = [
    '## ZeTa — Automated Test Coverage',
    '',
    `Generated by ZeTa Quality Intelligence Platform for project \`${projectId}\`.`,
    '',
    '### Scripts Added',
    '',
    '| Test Case | Target Path |',
    '|---|---|',
  ];
  for (const s of scripts.slice(0, 30)) {
    const title = s.testCase?.title ?? s.id;
    const path = s.testCase?.repoTargetPath ?? '(unknown)';
    lines.push(`| ${title.replace(/\|/g, '\\|').slice(0, 60)} | \`${path}\` |`);
  }
  if (scripts.length > 30) lines.push(`| ... and ${scripts.length - 30} more | |`);
  lines.push('', '---', '_Review each test carefully before merging. ZeTa generates scripts aligned to your existing framework conventions._');
  return lines.join('\n');
}

// ── NEW_PAGES_PIPELINE — post-crawl: detect new screens, generate tests + scripts ──

async function processNewPagesPipelineJob(jobId: string, config: any) {
  const { tenantId, projectId, crawlJobId, repoConnectionId, autoPR = false } = config;
  try {
    await updateJobProgress(tenantId, jobId, 5, 'Detecting new screens');

    // Find repo connection (use provided or find first active)
    const conn = await withTenant(tenantId, (tx) =>
      (tx as any).gitHubRepoConnection.findFirst({
        where: repoConnectionId
          ? { id: repoConnectionId, tenantId }
          : { tenantId, projectId, status: 'ACTIVE' },
      })
    ) as any;

    // Get all screens last seen after the lastSyncAt of the repo connection
    const sinceDate = conn?.lastSyncAt ?? new Date(0);
    const newScreens = await withTenant(tenantId, (tx) =>
      tx.screen.findMany({
        where: {
          tenantId,
          projectId,
          lastSeenAt: { gt: sinceDate },
          status: { not: 'DELETED' },
        },
        select: { id: true, url: true, name: true },
        take: 100,
      })
    ) as any[];

    if (newScreens.length === 0) {
      await updateJobProgress(tenantId, jobId, 100, 'No new screens to process');
      await withTenant(tenantId, (tx) =>
        tx.platformJob.update({ where: { id: jobId }, data: { status: 'DONE', finishedAt: new Date(), result: { newScreens: 0 } as any } })
      );
      return;
    }

    await updateJobProgress(tenantId, jobId, 15, `Found ${newScreens.length} new screens`);

    // For each new screen, enqueue GENERATE_TESTS if no test cases exist
    const generationJobs: string[] = [];
    for (const screen of newScreens) {
      const caseCount = await withTenant(tenantId, (tx) =>
        tx.testCase.count({ where: { tenantId, screenId: screen.id } })
      );
      if (caseCount === 0) {
        const jid = await enqueueJob('GENERATE_SCREEN_TESTS', {
          tenantId, projectId, screenId: screen.id,
          triggerAlignedScripts: conn != null,
          repoConnectionId: (conn as any)?.id ?? null,
        });
        generationJobs.push(jid);
      }
    }

    await updateJobProgress(tenantId, jobId, 60, `Queued ${generationJobs.length} test generation jobs`);

    // If repo connected and autoPR enabled, schedule BATCH_GENERATE_ALIGNED after generation
    if (conn && autoPR) {
      await enqueueJob('BATCH_GENERATE_ALIGNED', {
        tenantId, projectId,
        repoConnectionId: conn.id,
        scope: 'UNMAPPED',
      });
    }

    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({
        where: { id: jobId },
        data: { status: 'DONE', finishedAt: new Date(), result: { newScreens: newScreens.length, generationJobs: generationJobs.length } as any },
      })
    );
  } catch (err: any) {
    const msg = String(err?.message ?? err);
    logger.error({ jobId, err: msg }, '[new-pages-pipeline] failed');
    await withTenant(tenantId, (tx) =>
      tx.platformJob.update({ where: { id: jobId }, data: { status: 'FAILED', finishedAt: new Date(), error: msg.slice(0, 1000) } })
    ).catch(() => { });
  }
}