/**
 * Overnight batch video queue.
 *
 * A "batch" is a JSON manifest the operator authors listing many independent
 * video jobs. It compiles into ONE {@link VideoExecutionPayload} with N tasks,
 * which is exactly the shape the native route transports (native-runway,
 * native-dreamina, native-seedance) already accept and loop over. This module
 * is pure/deterministic apart from the small fs helpers at the bottom:
 *
 *  - readBatchManifest()   — graceful parse, throws on malformed (no silent fallback)
 *  - buildBatchPayload()   — pure: manifest -> VideoExecutionPayload
 *  - queue-state helpers    — persist <outDir>/batch-queue.json, idempotent rollup
 *  - clip mapping helpers   — scene-<i>.mp4 -> clips/<jobId>.mp4
 *
 * The default route is the FREE runway-useapi explore mode, so a large queue
 * can run unattended overnight at zero credit cost (low-res/slow drafts).
 */

import { existsSync } from 'node:fs';
import { mkdir, readFile, writeFile } from 'node:fs/promises';
import { join } from 'node:path';
import type { VideoExecutionPayload, VideoExecutionTask } from './types.js';

const VALID_ASPECT_RATIOS = ['16:9', '9:16', '1:1'] as const;
const VALID_RESOLUTIONS = ['720p', '1080p'] as const;

/** Routes the batch queue can target. A subset of ProviderRouteId — the three
 * routes with in-process native submit/poll transports that loop payload.tasks. */
export type BatchRouteId = 'runway-useapi' | 'dreamina-useapi' | 'seedance-direct';

const SUPPORTED_ROUTES: readonly BatchRouteId[] = ['runway-useapi', 'dreamina-useapi', 'seedance-direct'];

const DEFAULT_ROUTE: BatchRouteId = 'runway-useapi';
// 10s is the free runway-useapi explore-mode ceiling (720p, ≤10s); default to it
// so free-lane batch clips use the full available length unless a job overrides.
const DEFAULT_SECONDS = 10;
const DEFAULT_ASPECT_RATIO: VideoExecutionPayload['executionProfile']['aspectRatio'] = '16:9';
const DEFAULT_RESOLUTION: VideoExecutionPayload['executionProfile']['resolution'] = '720p';

export interface BatchQueueJob {
  /** Stable, operator-chosen id. Used as the downloaded clip filename. */
  id: string;
  prompt: string;
  /** Optional first-frame keyframe — a local path or a public http(s) URL. */
  keyframe?: string;
  /**
   * Optional END keyframe — a local path or public http(s) URL. With `keyframe`
   * set, the clip animates from `keyframe` (first frame) to `endKeyframe` (last
   * frame) via Seedance-2 keyframe interpolation, turning two stills into one
   * continuous shot. Requires `keyframe`; mutually exclusive with
   * `characterRefs`.
   */
  endKeyframe?: string;
  /**
   * Optional character reference images (local paths or public http(s) URLs) —
   * one per character, delivered to the provider's reference slot (Runway
   * `imageAssetId1..N`, Dreamina `omni_N_imageRef`) via `referenceRole:
   * 'character'`. Unlike `keyframe`, these are NEVER used as the video's first
   * frame, so a lone character sheet does not open the clip on the character
   * grid (the grid-open bug fixed on main). Mutually exclusive with `keyframe`.
   */
  characterRefs?: string[];
  /** Per-job duration override; falls back to defaults.seconds, then 10. */
  seconds?: number;
  /** Per-job aspect-ratio override (recorded; provider uses the batch default). */
  aspectRatio?: string;
}

export interface BatchQueueManifest {
  schemaVersion: 1;
  route?: BatchRouteId;
  defaults?: {
    seconds?: number;
    aspectRatio?: string;
    resolution?: string;
  };
  jobs: BatchQueueJob[];
}

export type BatchJobStatus = 'pending' | 'done' | 'failed';

export interface BatchQueueJobState {
  id: string;
  sceneIndex: number;
  taskId: string;
  status: BatchJobStatus;
  /** Set once the clip has been copied to clips/<id>.mp4. */
  clipPath?: string;
  error?: string;
  /**
   * ISO timestamp of when this job's scene was (re)submitted. Used by wedge
   * detection to measure stall time. Optional for backward compat: queues
   * written before this field fall back to the batch-level state.submittedAt.
   */
  submittedAt?: string;
  /**
   * The real provider-issued task id captured from the native submit's
   * rawResult (e.g. a Runway useapi task id). Distinct from the synthetic
   * `taskId` (`<externalJobId>#<sceneIndex>`) the batch queue keys downloads on.
   * Optional/best-effort: absent when the transport did not surface a per-scene
   * id, so queues written before this field still load unchanged.
   */
  providerTaskId?: string;
  /**
   * How many times this scene has been auto-resubmitted (Phase 3 PR2). Absent /
   * 0 means it has never been resubmitted; bounded by --max-resubmits. Optional
   * for backward compat: queues written before this field load unchanged.
   */
  resubmitCount?: number;
  /**
   * When a scene is re-submitted it lives under a NEW externalJobId. This field
   * records that id so the monitor polls the scene's CURRENT job-state file
   * (`.vclaw-jobs/<activeExternalJobId>.json`) rather than the dead original.
   * Absent means the scene is still owned by the batch-level state.externalJobId
   * (the default for every job until it is resubmitted), so existing queues are
   * unchanged.
   */
  activeExternalJobId?: string;
  /**
   * The compiled single-scene execution task captured at submit time. Persisted
   * so an auto-resubmit can rebuild a byte-identical one-task payload (same
   * prompt / referencePaths / referenceRole / duration / inputKind) WITHOUT
   * re-reading the original manifest. Optional/best-effort: absent on queues
   * written before this field, which simply cannot be auto-resubmitted.
   */
  task?: VideoExecutionTask;
}

export interface BatchQueueState {
  schemaVersion: 1;
  externalJobId: string;
  route: BatchRouteId;
  outputDir: string;
  workspaceRoot: string;
  submittedAt: string;
  jobs: BatchQueueJobState[];
}

export interface BatchQueueRollup {
  total: number;
  done: number;
  pending: number;
  failed: number;
  /** True when no job is still pending (every job is done or failed). */
  terminal: boolean;
}

function isSupportedRoute(value: unknown): value is BatchRouteId {
  return typeof value === 'string' && (SUPPORTED_ROUTES as readonly string[]).includes(value);
}

/**
 * Reads + validates a batch manifest. Throws loudly on malformed JSON or
 * missing required fields — no silent fallback. The only default applied is
 * `route` -> "runway-useapi" when omitted (the free explore default).
 */
export async function readBatchManifest(path: string): Promise<BatchQueueManifest & { route: BatchRouteId }> {
  let raw: string;
  try {
    raw = await readFile(path, 'utf-8');
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    throw new Error(`batch manifest not readable at ${path}: ${message}`);
  }
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    throw new Error(`batch manifest is not valid JSON (${path}): ${message}`);
  }
  if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
    throw new Error(`batch manifest must be a JSON object: ${path}`);
  }
  const obj = parsed as Record<string, unknown>;
  if (obj.schemaVersion !== 1) {
    throw new Error(`batch manifest schemaVersion must be 1, got: ${JSON.stringify(obj.schemaVersion)}`);
  }
  const route = obj.route === undefined ? DEFAULT_ROUTE : obj.route;
  if (!isSupportedRoute(route)) {
    throw new Error(
      `batch manifest route must be one of ${SUPPORTED_ROUTES.join(', ')}, got: ${JSON.stringify(obj.route)}`,
    );
  }
  if (!Array.isArray(obj.jobs) || obj.jobs.length === 0) {
    throw new Error(`batch manifest requires at least one job in "jobs": ${path}`);
  }
  const seenIds = new Set<string>();
  const jobs: BatchQueueJob[] = obj.jobs.map((entry, index) => {
    if (!entry || typeof entry !== 'object' || Array.isArray(entry)) {
      throw new Error(`batch manifest job ${index} must be an object`);
    }
    const job = entry as Record<string, unknown>;
    if (typeof job.id !== 'string' || !job.id.trim()) {
      throw new Error(`batch manifest job ${index} requires a non-empty string "id"`);
    }
    if (typeof job.prompt !== 'string' || !job.prompt.trim()) {
      throw new Error(`batch manifest job "${job.id}" requires a non-empty string "prompt"`);
    }
    if (seenIds.has(job.id)) {
      throw new Error(`batch manifest has a duplicate job id: "${job.id}"`);
    }
    seenIds.add(job.id);
    if (job.keyframe !== undefined && (typeof job.keyframe !== 'string' || !job.keyframe.trim())) {
      throw new Error(`batch manifest job "${job.id}" keyframe must be a non-empty string when present`);
    }
    if (job.endKeyframe !== undefined) {
      if (typeof job.endKeyframe !== 'string' || !job.endKeyframe.trim()) {
        throw new Error(`batch manifest job "${job.id}" endKeyframe must be a non-empty string when present`);
      }
      if (typeof job.keyframe !== 'string' || !job.keyframe.trim()) {
        throw new Error(`batch manifest job "${job.id}" endKeyframe requires a "keyframe" (first frame) to interpolate from`);
      }
      if (Array.isArray(job.characterRefs) && job.characterRefs.length > 0) {
        throw new Error(`batch manifest job "${job.id}" endKeyframe is mutually exclusive with characterRefs`);
      }
    }
    if (job.characterRefs !== undefined) {
      if (!Array.isArray(job.characterRefs) || job.characterRefs.some((ref: unknown) => typeof ref !== 'string' || !ref.trim())) {
        throw new Error(`batch manifest job "${job.id}" characterRefs must be an array of non-empty strings when present`);
      }
    }
    if (job.seconds !== undefined && (typeof job.seconds !== 'number' || !Number.isFinite(job.seconds))) {
      throw new Error(`batch manifest job "${job.id}" seconds must be a number when present`);
    }
    if (job.aspectRatio !== undefined && typeof job.aspectRatio !== 'string') {
      throw new Error(`batch manifest job "${job.id}" aspectRatio must be a string when present`);
    }
    return {
      id: job.id,
      prompt: job.prompt,
      ...(typeof job.keyframe === 'string' ? { keyframe: job.keyframe } : {}),
      ...(typeof job.endKeyframe === 'string' ? { endKeyframe: job.endKeyframe } : {}),
      ...(Array.isArray(job.characterRefs) ? { characterRefs: job.characterRefs as string[] } : {}),
      ...(typeof job.seconds === 'number' ? { seconds: job.seconds } : {}),
      ...(typeof job.aspectRatio === 'string' ? { aspectRatio: job.aspectRatio } : {}),
    };
  });

  let defaults: BatchQueueManifest['defaults'];
  if (obj.defaults !== undefined) {
    if (!obj.defaults || typeof obj.defaults !== 'object' || Array.isArray(obj.defaults)) {
      throw new Error(`batch manifest defaults must be an object when present`);
    }
    const d = obj.defaults as Record<string, unknown>;
    if (d.seconds !== undefined && (typeof d.seconds !== 'number' || !Number.isFinite(d.seconds))) {
      throw new Error(`batch manifest defaults.seconds must be a number when present`);
    }
    if (d.aspectRatio !== undefined) {
      if (typeof d.aspectRatio !== 'string' || !(VALID_ASPECT_RATIOS as readonly string[]).includes(d.aspectRatio)) {
        throw new Error(
          `batch manifest defaults.aspectRatio must be one of ${VALID_ASPECT_RATIOS.join(', ')}, got: ${JSON.stringify(d.aspectRatio)}`,
        );
      }
    }
    if (d.resolution !== undefined) {
      if (typeof d.resolution !== 'string' || !(VALID_RESOLUTIONS as readonly string[]).includes(d.resolution)) {
        throw new Error(
          `batch manifest defaults.resolution must be one of ${VALID_RESOLUTIONS.join(', ')}, got: ${JSON.stringify(d.resolution)}`,
        );
      }
    }
    defaults = d as BatchQueueManifest['defaults'];
  }

  return { schemaVersion: 1, route, ...(defaults ? { defaults } : {}), jobs };
}

function resolveAspectRatio(value: string | undefined): VideoExecutionPayload['executionProfile']['aspectRatio'] {
  if (value === undefined) return DEFAULT_ASPECT_RATIO;
  if (value === '9:16' || value === '1:1' || value === '16:9') return value;
  throw new Error(
    `batch manifest defaults.aspectRatio must be one of ${VALID_ASPECT_RATIOS.join(', ')}, got: ${JSON.stringify(value)}`,
  );
}

function resolveResolution(value: string | undefined): VideoExecutionPayload['executionProfile']['resolution'] {
  if (value === undefined) return DEFAULT_RESOLUTION;
  if (value === '1080p' || value === '720p') return value;
  throw new Error(
    `batch manifest defaults.resolution must be one of ${VALID_RESOLUTIONS.join(', ')}, got: ${JSON.stringify(value)}`,
  );
}

/**
 * Pure: compiles a manifest into a single VideoExecutionPayload whose `tasks`
 * map jobs 1:1. Job index becomes sceneIndex; job.keyframe becomes the task's
 * sole referencePath; duration resolves job.seconds -> defaults.seconds -> 10.
 *
 * The resulting payload is byte-compatible with what the native route
 * transports already consume, so submit/poll need no batch-specific code path.
 */
export function buildBatchPayload(
  manifest: BatchQueueManifest,
  opts: { workspaceRoot: string; outputDir: string },
): VideoExecutionPayload {
  const route = isSupportedRoute(manifest.route) ? manifest.route : DEFAULT_ROUTE;
  const defaultSeconds =
    typeof manifest.defaults?.seconds === 'number' && Number.isFinite(manifest.defaults.seconds)
      ? manifest.defaults.seconds
      : DEFAULT_SECONDS;
  const aspectRatio = resolveAspectRatio(manifest.defaults?.aspectRatio);
  const resolution = resolveResolution(manifest.defaults?.resolution);

  const tasks = manifest.jobs.map((job, index) => {
    // characterRefs take priority: delivered as identity references (never the
    // first frame). A keyframe stays a genuine first-frame seed. A job uses one
    // or the other; characterRefs wins if both are somehow present.
    const characterRefs = (job.characterRefs ?? []).filter((ref) => typeof ref === 'string' && ref.trim());
    const referencePaths = characterRefs.length > 0
      ? characterRefs
      : job.keyframe
        ? [job.keyframe]
        : [];
    const referenceRole: 'character' | undefined = characterRefs.length > 0 ? 'character' : undefined;
    // End keyframe only applies to the keyframe path (not characterRefs) and is
    // carried separately so it never counts as a second referencePaths image.
    const endKeyframePath =
      characterRefs.length === 0 && job.keyframe && typeof job.endKeyframe === 'string'
        ? job.endKeyframe
        : undefined;
    const durationSeconds =
      typeof job.seconds === 'number' && Number.isFinite(job.seconds) ? job.seconds : defaultSeconds;
    return {
      sceneIndex: index,
      prompt: job.prompt,
      inputKind: (referencePaths.length > 0 ? 'image' : 'text') as 'image' | 'text',
      referencePaths,
      ...(referenceRole ? { referenceRole } : {}),
      ...(endKeyframePath ? { endKeyframePath } : {}),
      sourceAssetIds: [],
      backendHints: ['batch-queue'],
      characters: [],
      durationSeconds,
    };
  });

  return {
    workspaceRoot: opts.workspaceRoot,
    projectSlug: 'batch-queue',
    productionMode: 'storyboard',
    routeId: route,
    operationKind: 'text-to-video',
    executionProfile: {
      aspectRatio,
      quality: 'fast',
      resolution,
      generateAudio: false,
      outputCount: 1,
    },
    generatedAt: new Date().toISOString(),
    outputDir: opts.outputDir,
    tasks,
    promptGuidance: [],
  };
}

/** The path the native transport writes a completed scene to. */
export function sceneOutputPathFor(outputDir: string, sceneIndex: number): string {
  return join(outputDir, `scene-${sceneIndex}.mp4`);
}

/** The stable per-job clip path the batch monitor copies completed scenes to. */
export function clipPathForJob(outputDir: string, jobId: string): string {
  return join(outputDir, 'clips', `${jobId}.mp4`);
}

export function batchQueueStatePath(outputDir: string): string {
  return join(outputDir, 'batch-queue.json');
}

export function batchStatusPath(outputDir: string): string {
  return join(outputDir, 'batch-status.json');
}

export async function writeBatchQueueState(state: BatchQueueState): Promise<void> {
  await mkdir(state.outputDir, { recursive: true });
  await writeFile(batchQueueStatePath(state.outputDir), `${JSON.stringify(state, null, 2)}\n`);
}

export async function readBatchQueueState(outputDir: string): Promise<BatchQueueState> {
  const path = batchQueueStatePath(outputDir);
  if (!existsSync(path)) {
    throw new Error(`batch-queue.json not found in ${outputDir}; run "vclaw video batch-submit" first.`);
  }
  return JSON.parse(await readFile(path, 'utf-8')) as BatchQueueState;
}

/**
 * Reads the native transport's per-scene job-state file
 * (`<outputDir>/.vclaw-jobs/<externalJobId>.json`) and returns a map from
 * sceneIndex to `{ status, error }`. All three transports (runway, dreamina,
 * seedance) write this file with an identical `scenes[]` shape.
 *
 * Returns an empty map when the file is absent (e.g. a very early poll pass
 * before the transport has flushed state). The caller treats absent scenes as
 * still-pending.
 */
export async function readNativeJobSceneStates(
  outputDir: string,
  externalJobId: string,
): Promise<Map<number, { status: 'submitted' | 'completed' | 'failed'; error?: string; outputPath: string }>> {
  const path = join(outputDir, '.vclaw-jobs', `${externalJobId}.json`);
  if (!existsSync(path)) return new Map();
  let raw: string;
  try {
    raw = await readFile(path, 'utf-8');
  } catch {
    return new Map();
  }
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch {
    return new Map();
  }
  const jobState = parsed as {
    scenes?: Array<{ sceneIndex?: number; status?: string; error?: string; outputPath?: string }>;
  };
  const result = new Map<number, { status: 'submitted' | 'completed' | 'failed'; error?: string; outputPath: string }>();
  for (const scene of jobState.scenes ?? []) {
    if (typeof scene.sceneIndex !== 'number') continue;
    const status =
      scene.status === 'completed' || scene.status === 'failed' || scene.status === 'submitted'
        ? scene.status
        : 'submitted';
    result.set(scene.sceneIndex, {
      status,
      outputPath: typeof scene.outputPath === 'string' ? scene.outputPath : '',
      ...(scene.error ? { error: scene.error } : {}),
    });
  }
  return result;
}

/** Counts done/pending/failed and reports whether the queue is fully terminal. */
export function rollupBatchQueueState(state: BatchQueueState): BatchQueueRollup {
  let done = 0;
  let pending = 0;
  let failed = 0;
  for (const job of state.jobs) {
    if (job.status === 'done') done += 1;
    else if (job.status === 'failed') failed += 1;
    else pending += 1;
  }
  return { total: state.jobs.length, done, pending, failed, terminal: pending === 0 };
}

/**
 * Runway free explore mode allows ~1 concurrent job and returns
 * canUseExploreMode:false / HTTP 429 / rate-limit text when saturated. Detect
 * that signal from a monitor pass's combined stdout+stderr so the loop can back
 * off instead of hammering at the normal interval. Pure/deterministic.
 */
export function isExploreThrottled(passOutput: string): boolean {
  // Allow an optional closing quote/whitespace between the key and the colon so a
  // JSON fragment like `"canUseExploreMode": false` is matched, not just bare `canUseExploreMode:false`.
  return /canUseExploreMode["'\s]*[:=]\s*false|\b429\b|too many requests|rate.?limit/i.test(passOutput);
}

/**
 * Pure: extracts the provider-issued per-scene task ids from a native submit's
 * `rawResult`. Defensively reads `rawResult.submittedScenes` (an array of
 * `{ sceneIndex: number, taskId: string }`) and returns a Map sceneIndex ->
 * taskId. Tolerant by design — a missing/non-object rawResult, a missing or
 * non-array `submittedScenes`, or partial/wrong-typed entries are skipped
 * rather than throwing. Returns an empty Map when nothing usable is present.
 */
export function extractProviderTaskIds(rawResult: unknown): Map<number, string> {
  const result = new Map<number, string>();
  if (!rawResult || typeof rawResult !== 'object' || Array.isArray(rawResult)) return result;
  const submittedScenes = (rawResult as { submittedScenes?: unknown }).submittedScenes;
  if (!Array.isArray(submittedScenes)) return result;
  for (const entry of submittedScenes) {
    if (!entry || typeof entry !== 'object' || Array.isArray(entry)) continue;
    const { sceneIndex, taskId } = entry as { sceneIndex?: unknown; taskId?: unknown };
    if (typeof sceneIndex !== 'number' || !Number.isFinite(sceneIndex)) continue;
    if (typeof taskId !== 'string' || !taskId) continue;
    result.set(sceneIndex, taskId);
  }
  return result;
}

/**
 * Pure: exponential backoff for the explore-throttle case. Returns
 * `baseMs * 2^consecutiveThrottles`, clamped to `[baseMs, capMs]`. A
 * `consecutiveThrottles <= 0` returns `baseMs` (the normal interval). Used by
 * the monitor loop to widen the poll interval while Runway explore mode is
 * saturated instead of hammering at the flat interval.
 */
export function nextBackoffMs(baseMs: number, consecutiveThrottles: number, capMs: number): number {
  if (!(consecutiveThrottles > 0)) return Math.min(baseMs, capMs);
  const scaled = baseMs * 2 ** consecutiveThrottles;
  return Math.min(Math.max(scaled, baseMs), capMs);
}

export interface WedgedScene {
  id: string;
  sceneIndex: number;
  stalledMinutes: number;
}

/**
 * A scene is "wedged" when its job is still pending, the native transport still
 * reports it 'submitted' (never reached completed/failed — e.g. a Runway explore
 * queue that never returns it), and it has made no progress for more than
 * stallMinutes. `stallMinutes <= 0` disables detection. Per-job submittedAt falls
 * back to the batch-level state.submittedAt for queues written before that field
 * existed, so old `batch-queue.json` files still work. Pure/deterministic.
 */
export function detectWedgedScenes<S extends { status: 'submitted' | 'completed' | 'failed' }>(
  state: BatchQueueState,
  sceneStates: Map<number, S>,
  nowMs: number,
  stallMinutes: number,
): WedgedScene[] {
  if (!(stallMinutes > 0)) return [];
  const wedged: WedgedScene[] = [];
  for (const job of state.jobs) {
    if (job.status !== 'pending' || !job.taskId) continue;
    const scene = sceneStates.get(job.sceneIndex);
    if (!scene || scene.status !== 'submitted') continue;
    const submittedIso = job.submittedAt ?? state.submittedAt;
    const submittedMs = Date.parse(submittedIso);
    if (Number.isNaN(submittedMs)) continue;
    const stalledMinutes = (nowMs - submittedMs) / 60000;
    if (stalledMinutes > stallMinutes) {
      wedged.push({ id: job.id, sceneIndex: job.sceneIndex, stalledMinutes });
    }
  }
  return wedged;
}

/**
 * Pure: the distinct set of job-state ids the monitor must poll. A scene that
 * has never been resubmitted is owned by the batch-level `state.externalJobId`;
 * a re-submitted scene is owned by its `activeExternalJobId`. The monitor reads
 * `.vclaw-jobs/<id>.json` for each id in this set and merges the per-scene
 * states. Order is preserved (first occurrence wins); deduped. When no job has
 * been resubmitted this returns exactly `[state.externalJobId]`, so the default
 * monitor behaviour is unchanged.
 */
export function activeJobIdsFor(state: BatchQueueState): string[] {
  const ids: string[] = [];
  const seen = new Set<string>();
  for (const job of state.jobs) {
    const id = job.activeExternalJobId ?? state.externalJobId;
    if (seen.has(id)) continue;
    seen.add(id);
    ids.push(id);
  }
  // Defensive: a state with zero jobs still yields the batch-level id so the
  // monitor has something to poll (matches pre-resubmit behaviour).
  if (ids.length === 0 && state.externalJobId) {
    ids.push(state.externalJobId);
  }
  return ids;
}

/**
 * Pure: plans which wedged scenes are eligible for an auto-resubmit this pass.
 * Delegates wedge detection to {@link detectWedgedScenes} (same stall policy),
 * then keeps only the wedged scenes whose job has been resubmitted fewer than
 * `maxResubmits` times (`resubmitCount ?? 0 < maxResubmits`). Returns the planned
 * `{ sceneIndex, jobId }` list — the side-effecting submit + state mutation lives
 * in the CLI, not here. Empty when `stallMinutes <= 0`, when nothing is wedged,
 * or when every wedged scene has hit its resubmit cap. Never throws.
 */
export function planResubmits(
  state: BatchQueueState,
  sceneStates: Map<number, { status: 'submitted' | 'completed' | 'failed' }>,
  nowMs: number,
  opts: { stallMinutes: number; maxResubmits: number },
): { sceneIndex: number; jobId: string }[] {
  let wedged: WedgedScene[];
  try {
    wedged = detectWedgedScenes(state, sceneStates, nowMs, opts.stallMinutes);
  } catch {
    return [];
  }
  if (wedged.length === 0) return [];
  const byIndex = new Map(state.jobs.map((job) => [job.sceneIndex, job] as const));
  const planned: { sceneIndex: number; jobId: string }[] = [];
  for (const scene of wedged) {
    const job = byIndex.get(scene.sceneIndex);
    if (!job) continue;
    if ((job.resubmitCount ?? 0) < opts.maxResubmits) {
      planned.push({ sceneIndex: scene.sceneIndex, jobId: job.id });
    }
  }
  return planned;
}

/**
 * Pure: builds a single-task VideoExecutionPayload for ONE scene of an existing
 * batch state, reusing that scene's persisted `task` (same prompt /
 * referencePaths / referenceRole / duration / inputKind) so a resubmit
 * reproduces the same job. The task keeps its ORIGINAL `sceneIndex` so the
 * transport writes `scene-<i>.mp4` into the same outputDir the batch monitor
 * reads. The payload's routeId/outputDir/workspaceRoot/profile come from the
 * batch state (the route is the credit-safety-guarded runway-useapi).
 *
 * Throws if the scene is unknown or its `task` was not captured at submit time
 * (an old queue) — the CLI guards against the latter before calling this, so a
 * resubmit is simply skipped rather than crashing the monitor.
 */
export function buildSingleSceneResubmitPayload(
  state: BatchQueueState,
  sceneIndex: number,
): VideoExecutionPayload {
  const job = state.jobs.find((j) => j.sceneIndex === sceneIndex);
  if (!job) {
    throw new Error(`buildSingleSceneResubmitPayload: no job for sceneIndex ${sceneIndex}`);
  }
  if (!job.task) {
    throw new Error(
      `buildSingleSceneResubmitPayload: job "${job.id}" (scene ${sceneIndex}) has no captured task to resubmit`,
    );
  }
  // Preserve the original sceneIndex so the transport writes scene-<i>.mp4 to the
  // same place; everything else is the captured task verbatim.
  const task: VideoExecutionTask = { ...job.task, sceneIndex };
  return {
    workspaceRoot: state.workspaceRoot,
    projectSlug: 'batch-queue',
    productionMode: 'storyboard',
    routeId: state.route,
    operationKind: 'text-to-video',
    executionProfile: {
      aspectRatio: DEFAULT_ASPECT_RATIO,
      quality: 'fast',
      resolution: DEFAULT_RESOLUTION,
      generateAudio: false,
      outputCount: 1,
    },
    generatedAt: new Date().toISOString(),
    outputDir: state.outputDir,
    tasks: [task],
    promptGuidance: [],
  };
}

/**
 * Returns a copy of `state` with the given scene's taskId/submittedAt cleared so
 * a subsequent submit pass re-queues it. Pure — does not mutate the input.
 */
export function clearWedgedScene(state: BatchQueueState, sceneIndex: number): BatchQueueState {
  return {
    ...state,
    jobs: state.jobs.map((job) =>
      job.sceneIndex === sceneIndex ? { ...job, taskId: '', submittedAt: undefined } : job,
    ),
  };
}

/**
 * Applies wedge handling to a freshly-polled batch state. Pure/deterministic:
 * given the current jobs, the transport's per-scene states, the clock, and the
 * stall policy, returns the wedged scenes and — when `failWedged` is set — a new
 * state with those still-pending wedged jobs marked 'failed'. Failing them lets
 * the queue reach `terminal` so the monitor can stop instead of polling a stuck
 * scene until its deadline. Does NOT mutate the input.
 *
 * `stallMinutes <= 0` disables detection (returns the state unchanged + an empty
 * list), so the monitor's default behaviour is byte-identical to before.
 */
export function applyWedgeHandling(
  state: BatchQueueState,
  sceneStates: Map<number, { status: 'submitted' | 'completed' | 'failed' }>,
  nowMs: number,
  opts: { stallMinutes: number; failWedged: boolean },
): { state: BatchQueueState; wedged: WedgedScene[] } {
  const wedged = detectWedgedScenes(state, sceneStates, nowMs, opts.stallMinutes);
  if (!opts.failWedged || wedged.length === 0) return { state, wedged };
  const wedgedIndexes = new Set(wedged.map((scene) => scene.sceneIndex));
  const next: BatchQueueState = {
    ...state,
    jobs: state.jobs.map((job) =>
      wedgedIndexes.has(job.sceneIndex) && job.status === 'pending'
        ? { ...job, status: 'failed' as const, error: `wedged: no provider progress in >${opts.stallMinutes}m` }
        : job,
    ),
  };
  return { state: next, wedged };
}