import { existsSync } from 'node:fs';
import { copyFile, mkdir, readdir, readFile, stat } from 'node:fs/promises';
import { writeTextFileAtomic } from './atomic-write.js';
import { join } from 'node:path';
import { spawn } from 'node:child_process';
import type { VideoExecutionPayload, VideoExecutionPollResult } from './types.js';
import { isOmniFirstFrameEnabled } from './provider-platform/route-capabilities.js';

interface VeoNativeJobState {
  externalJobId: string;
  routeId: 'veo-useapi';
  outputDir: string;
  createdAt: string;
  outputs: VideoExecutionPollResult['outputs'];
}

function readDotEnvLike(raw: string): Record<string, string> {
  const out: Record<string, string> = {};
  for (const line of raw.split('\n')) {
    const trimmed = line.trim();
    if (!trimmed || trimmed.startsWith('#') || !trimmed.includes('=')) continue;
    const [key, ...rest] = trimmed.split('=');
    out[key.trim()] = rest.join('=').trim().replace(/^['"]|['"]$/g, '');
  }
  return out;
}

async function loadWorkspaceEnv(workspaceRoot: string, env: NodeJS.ProcessEnv): Promise<NodeJS.ProcessEnv> {
  const envLocalPath = join(workspaceRoot, '.env.local');
  if (!existsSync(envLocalPath)) return env;
  return {
    ...readDotEnvLike(await readFile(envLocalPath, 'utf-8')),
    ...env,
  };
}

function veoCliRoot(workspaceRoot: string, env: NodeJS.ProcessEnv): string {
  return env.VCLAW_VEO_CLI_ROOT || join(workspaceRoot, 'vclaw-cli');
}

function veoOutputDir(workspaceRoot: string, env: NodeJS.ProcessEnv): string {
  return env.VCLAW_VEO_OUTPUT_DIR || join(veoCliRoot(workspaceRoot, env), 'output-videos');
}

function veoBunBin(env: NodeJS.ProcessEnv): string {
  return env.VCLAW_VEO_BUN_BIN || 'bun';
}

function ensureVeoCliEntry(cliRoot: string): void {
  const entryPath = join(cliRoot, 'flow.ts');
  if (existsSync(entryPath)) return;
  throw new Error(
    `veo-useapi native transport could not find flow.ts at ${entryPath}. ` +
    'Set VCLAW_VEO_CLI_ROOT to your vclaw-cli directory (for example, /path/to/videoclaw-v2/vclaw-cli).',
  );
}

// Default per-command timeout. 180s killed HEALTHY renders in production
// (hermes-do-launch 2026-07-04: fast-i2v renders took 2.5–4.5 min under queue
// load, and the sidecar's internal captcha-retry loop alone can grind past
// 3 min at "Starting video generation..."). 10 min keeps hang protection while
// clearing every observed healthy render; VCLAW_VEO_COMMAND_TIMEOUT_MS overrides.
export const DEFAULT_VEO_COMMAND_TIMEOUT_MS = 600_000;

function veoCommandTimeoutMs(env: NodeJS.ProcessEnv): number {
  const raw = env.VCLAW_VEO_COMMAND_TIMEOUT_MS;
  if (!raw) return DEFAULT_VEO_COMMAND_TIMEOUT_MS;
  const parsed = Number.parseInt(raw, 10);
  if (!Number.isFinite(parsed) || parsed <= 0) return DEFAULT_VEO_COMMAND_TIMEOUT_MS;
  return parsed;
}

function veoRatio(aspectRatio: VideoExecutionPayload['executionProfile']['aspectRatio']): 'landscape' | 'portrait' | 'square' {
  if (aspectRatio === '9:16') return 'portrait';
  if (aspectRatio === '1:1') return 'square';
  return 'landscape';
}

/**
 * The Veo CLI `-m` flag carries the Flow v1 model id. We prefer the explicit
 * `veoModel` override (which can select `omni-flash`) and fall back to
 * `quality` so legacy payloads pass `fast`/`quality` through verbatim.
 */
function veoModelFlag(profile: VideoExecutionPayload['executionProfile']): string {
  return profile.veoModel ?? profile.quality;
}

/**
 * Durations the Flow v1 backend accepts (`flow.ts --duration`). Anything else
 * is dropped rather than forwarded, mirroring the sidecar's own allowlist.
 */
const VEO_DURATIONS = new Set([4, 6, 8, 10]);

function isSessionRefreshFailure(detail: string): boolean {
  return /Failed to refresh session|Account status is \"error\"|setup-google-flow/i.test(detail);
}

/**
 * Transient infrastructure failures worth a retry — network/socket drops and
 * upstream 5xx/unavailable blips that the Flow sync path surfaces as an
 * exit-0-but-no-output (flow.ts prints the error and exits cleanly). These are
 * NOT content moderation (PUBLIC_ERROR_UNSAFE_GENERATION) — that is deterministic
 * for a given prompt and must NOT be auto-retried here (it would silently burn
 * credits), so it falls through to the normal fail path.
 */
function isTransientFailure(detail: string): boolean {
  return /socket connection was closed|socket hang ?up|ECONNRESET|ETIMEDOUT|EAI_AGAIN|ENOTFOUND|connection reset|fetch failed|network error|temporarily unavailable|\b50[234]\b/i.test(detail);
}

/**
 * Flow burst-throttle: the Google-Flow load balancer guards bursts behind a
 * reCAPTCHA challenge (`PUBLIC_ERROR_UNUSUAL_ACTIVITY`) and HTTP 429. Unlike
 * moderation (deterministic for a given prompt), a throttle clears after a
 * cooldown — so it IS worth retrying, just on a much longer adaptive delay than
 * a transient socket blip. A 429 / "too many requests" / reCAPTCHA / unusual-
 * activity is always a throttle; a BARE 403 or "permission_denied" is NOT —
 * auth, billing and quota failures are deterministic 403s that never clear, so a
 * 403 only counts as a throttle when the body also looks like a reCAPTCHA wall
 * (the same gating native-flow-r2v applies: `status===403 && RECAPTCHA_RE`).
 */
function isThrottleFailure(detail: string): boolean {
  // `captcha_quality` precedes `PUBLIC_ERROR_UNUSUAL_ACTIVITY` in the Flow
  // reject envelope ("Access denied: captcha_quality: PUBLIC_ERROR_…"), so it
  // survives truncation that cuts the later marker (hermes-do-launch
  // 2026-07-04: a 60-char summary cap ended the detail at "Reas." and a
  // retryable burst throttle classified as fatal — no cooldown, scene dead).
  return /\b429\b|too many requests|recaptcha|unusual_activity|captcha_quality/i.test(detail)
    || (/\b403\b|permission_denied/i.test(detail) && /recaptcha|unusual_activity|captcha_quality/i.test(detail));
}

/**
 * Classify a failed Veo CLI attempt. Moderation is checked FIRST: a moderation
 * response whose detail happens to also contain a transient- or throttle-looking
 * phrase (e.g. "network error" / "permission_denied" inside an UNSAFE_GENERATION
 * envelope) must never be retried — moderation is deterministic for a given
 * prompt and a retry only burns credits. Throttle is checked before transient so
 * a reCAPTCHA / 403 / 429 burst takes the long cooldown, not the 2s socket-blip
 * backoff. Exported for tests; the retry loop is the only runtime caller.
 */
export function classifyVeoFailure(detail: string): 'moderation' | 'throttle' | 'transient' | 'fatal' {
  if (/PUBLIC_ERROR_UNSAFE_GENERATION|UNSAFE_GENERATION|FINISH_REASON_INPUT|INPUT_SPEECH_EDIT|INPUT_VIDEO_EDIT/i.test(detail)) {
    return 'moderation';
  }
  if (isThrottleFailure(detail)) return 'throttle';
  return isTransientFailure(detail) ? 'transient' : 'fatal';
}

/** Max attempts per scene on transient/throttle failures (default 3, env override, capped at 6). */
function veoMaxAttempts(env: NodeJS.ProcessEnv): number {
  const raw = env.VCLAW_VEO_MAX_ATTEMPTS;
  const n = raw ? Number.parseInt(raw, 10) : 3;
  return Number.isFinite(n) && n >= 1 ? Math.min(n, 6) : 3;
}

/**
 * Cooldown (ms) before retrying a {@link isThrottleFailure} burst-throttle.
 * Escalates with the attempt number (attempt×base) and is capped at 180s — long
 * enough to clear Flow's reCAPTCHA window, which a 2s transient backoff never
 * does. `VCLAW_VEO_THROTTLE_COOLDOWN_MS` overrides the base (default 75s; set 0
 * in tests for an instant retry).
 */
function veoThrottleCooldownMs(env: NodeJS.ProcessEnv, attempt: number): number {
  const raw = env.VCLAW_VEO_THROTTLE_COOLDOWN_MS;
  const base = raw !== undefined ? Number.parseInt(raw, 10) : 75000;
  const safeBase = Number.isFinite(base) && base >= 0 ? base : 75000;
  return Math.min(safeBase * Math.max(attempt, 1), 180000);
}

/**
 * Curated aggression→benign substitutions for the opt-in safe-motion retry. Veo's
 * content filter deterministically rejects violent/graphic MOTION wording (the
 * Avatar panel-10 "wrathful third-eye + violent shockwave" reject); a benign
 * reword ("radiant pulse of light, awakening") cleared it on one retry.
 * Conservative and additive — only well-known trigger words are softened.
 */
const MOTION_SOFTEN_MAP: ReadonlyArray<readonly [RegExp, string]> = [
  [/\bviolently\b/gi, 'intensely'],
  [/\bviolent\b/gi, 'intense'],
  [/\bviolence\b/gi, 'intensity'],
  [/\bwrathful\b/gi, 'fierce'],
  [/\bwrath\b/gi, 'fierce resolve'],
  [/\bshockwaves?\b/gi, 'radiant pulse'],
  [/\bexplosions?\b/gi, 'burst of light'],
  [/\bexploding\b/gi, 'bursting with light'],
  [/\bbloody\b/gi, 'crimson'],
  [/\bblood\b/gi, 'crimson light'],
  [/\bgore\b/gi, 'intensity'],
  [/\bgory\b/gi, 'intense'],
  [/\bkilling\b/gi, 'defeating'],
  [/\bkills\b/gi, 'defeats'],
  [/\bkilled\b/gi, 'defeated'],
  [/\bkill\b/gi, 'defeat'],
  [/\bmurder(?:s|ing|ed)?\b/gi, 'defeat'],
  [/\bweapons?\b/gi, 'implement'],
  [/\battacking\b/gi, 'advancing'],
  [/\battacks?\b/gi, 'advance'],
  [/\bbrutally\b/gi, 'forcefully'],
  [/\bbrutal\b/gi, 'forceful'],
  [/\bslaughter\b/gi, 'rout'],
];

/**
 * Soften aggressive motion wording that Veo content-moderation rejects, returning
 * the rewritten prompt and whether anything changed. PURE; exported for tests and
 * the opt-in safe-motion retry. Retrying is only worthwhile when `changed` is true
 * — an unchanged prompt would re-trigger the same deterministic reject.
 */
export function softenMotionPrompt(prompt: string): { prompt: string; changed: boolean } {
  let out = prompt;
  for (const [pattern, replacement] of MOTION_SOFTEN_MAP) {
    out = out.replace(pattern, replacement);
  }
  return { prompt: out, changed: out !== prompt };
}

/** Opt-in (`VCLAW_VEO_SAFE_MOTION=1`): auto-soften a content-moderation reject and retry ONCE. Default off. */
function veoSafeMotionEnabled(env: NodeJS.ProcessEnv): boolean {
  return env.VCLAW_VEO_SAFE_MOTION === '1' || env.VCLAW_VEO_SAFE_MOTION === 'true';
}

const delay = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));

/** omni-flash R2V accepts up to 7 reference images. */
const VEO_MAX_R2V_REFS = 7;


/**
 * Image references are model-aware, and flow.ts reads them from the PROMPT text
 * (not a CLI flag): `image:<p>` → startImage (I2V), `ingredients:<p1,p2>` →
 * referenceImage_* (R2V).
 *  - Veo models: first-frame `image:` startImage (unchanged legacy behavior).
 *  - omni-flash (default): `ingredients:` (R2V) — the reliable voice path
 *    (referenceAudio needs an image or video reference).
 *  - omni-flash + `firstFrame` (default-on; VCLAW_OMNI_FIRST_FRAME is a
 *    kill-switch): `image:` startImage (I2V) using a single reference as the
 *    literal first frame — locks the opening frame. LIVE-VERIFIED 2026-06-06:
 *    omni-flash startImage routes to abra_i2v_8s / IMAGE_TO_VIDEO on useapi.net.
 * R2V is suppressed when a V2V edit is requested (referenceVideoMediaId) — the
 * provider treats referenceImage_* and referenceVideo_1 as mutually exclusive.
 */
function buildPrompt(
  task: VideoExecutionPayload['tasks'][number],
  omniFlash: boolean,
  firstFrame = false,
): string {
  const tag = `[scene_${task.sceneIndex}]`;
  const images = task.inputKind === 'image' ? task.referencePaths.filter(Boolean).slice(0, VEO_MAX_R2V_REFS) : [];
  if (images.length > 0) {
    // omni-flash First-Frame (gated): a single ref becomes the startImage (I2V),
    // never the loose R2V ingredients. V2V is mutually exclusive with startImage.
    if (omniFlash && firstFrame && !task.referenceVideoMediaId) {
      return `${tag} image:${images[0]} ${task.prompt}`.trim();
    }
    if (omniFlash && !task.referenceVideoMediaId) {
      return `${tag} ingredients:${images.join(',')} ${task.prompt}`.trim();
    }
    if (!omniFlash) {
      return `${tag} image:${images[0]} ${task.prompt}`.trim();
    }
  }
  return `${tag} ${task.prompt}`.trim();
}

function jobStateDir(outputDir: string): string {
  return join(outputDir, '.vclaw-jobs');
}

function jobStatePath(outputDir: string, externalJobId: string): string {
  return join(jobStateDir(outputDir), `${externalJobId}.json`);
}

async function writeJobState(state: VeoNativeJobState): Promise<void> {
  await mkdir(jobStateDir(state.outputDir), { recursive: true });
  // Atomic (tmp + rename) so a kill mid-write never leaves a torn job-state file.
  await writeTextFileAtomic(jobStatePath(state.outputDir, state.externalJobId), `${JSON.stringify(state, null, 2)}\n`);
}

async function readJobState(outputDir: string, externalJobId: string): Promise<VeoNativeJobState> {
  const path = jobStatePath(outputDir, externalJobId);
  if (!existsSync(path)) {
    throw new Error(`Veo native job state not found for ${externalJobId}.`);
  }
  const raw = await readFile(path, 'utf-8');
  try {
    return JSON.parse(raw) as VeoNativeJobState;
  } catch (error) {
    throw new Error(
      `Veo native job state for ${externalJobId} is corrupt (invalid JSON at ${path}): ${error instanceof Error ? error.message : String(error)}`,
    );
  }
}

async function listFiles(dir: string): Promise<string[]> {
  if (!existsSync(dir)) return [];
  return (await readdir(dir)).map((entry) => join(dir, entry));
}

async function runVeoCommand(
  args: string[],
  options: {
    cwd: string;
    env: NodeJS.ProcessEnv;
  },
): Promise<{ stdout: string; stderr: string }> {
  const result = await new Promise<{
    stdout: string;
    stderr: string;
    code: number | null;
    signal: NodeJS.Signals | null;
    timedOut: boolean;
  }>((resolve, reject) => {
    const child = spawn(veoBunBin(options.env), args, {
      cwd: options.cwd,
      env: options.env,
      stdio: ['ignore', 'pipe', 'pipe'],
    });
    let didTimeout = false;
    const timeoutHandle = setTimeout(() => {
      didTimeout = true;
      child.kill('SIGTERM');
    }, veoCommandTimeoutMs(options.env));
    let stdout = '';
    let stderr = '';
    child.stdout.on('data', (chunk) => {
      stdout += String(chunk);
    });
    child.stderr.on('data', (chunk) => {
      stderr += String(chunk);
    });
    child.on('error', reject);
    child.on('close', (code, signal) => {
      clearTimeout(timeoutHandle);
      resolve({ stdout, stderr, code, signal, timedOut: didTimeout });
    });
  });
  if (result.timedOut) {
    const detail = result.stderr.trim() || result.stdout.trim();
    if (detail && isSessionRefreshFailure(detail)) {
      throw new Error(
        `veo-useapi native command timed out: session refresh failed. ` +
        `Update Google-flow cookies for the Veo CLI and retry (cookie hint: ${join(options.cwd, 'cookie.json')}, docs: https://useapi.net/docs/start-here/setup-google-flow). ` +
        `Original output: ${detail}`,
      );
    }
    throw new Error(
      `veo-native command timed out. ` +
      `Check Veo CLI runtime health and refresh Google-flow cookies if needed (cookie hint: ${join(options.cwd, 'cookie.json')}, docs: https://useapi.net/docs/start-here/setup-google-flow)` +
      `${detail ? `: ${detail}` : ''}`,
    );
  }
  if (result.code !== 0) {
    const detail = result.stderr.trim() || result.stdout.trim() || `exit ${result.code}${result.signal ? ` (${result.signal})` : ''}`;
    if (isSessionRefreshFailure(detail)) {
      throw new Error(
        `veo-useapi native command failed: session refresh failed. ` +
        `Update Google-flow cookies for the Veo CLI and retry (cookie hint: ${join(options.cwd, 'cookie.json')}, docs: https://useapi.net/docs/start-here/setup-google-flow). ` +
        `Original output: ${detail}`,
      );
    }
    throw new Error(`veo-useapi native command failed: ${detail}`);
  }
  return {
    stdout: result.stdout,
    stderr: result.stderr,
  };
}

async function captureNewOutputs(
  outputDir: string,
  before: string[],
  startedAt: number,
): Promise<string[]> {
  const after = await listFiles(outputDir);
  const beforeSet = new Set(before);
  const added = after.filter((path) => !beforeSet.has(path));
  if (added.length > 0) return added;

  const recent: string[] = [];
  for (const path of after) {
    const fileStat = await stat(path);
    if (fileStat.mtimeMs >= startedAt) {
      recent.push(path);
    }
  }
  return recent;
}

/**
 * From the files that appeared during a scene's render window, pick the one
 * that actually belongs to THIS scene.
 *
 * flow.ts names every output `<timestamp>_scene_<i>.mp4` (the tag comes from the
 * `[scene_<i>]` prompt prefix {@link buildPrompt} sets). When a SECOND project
 * renders into the same `output-videos/` concurrently — e.g. two sessions
 * sharing one `VCLAW_VEO_CLI_ROOT` — the window can contain that project's
 * fresh clips too, so blindly taking the first new file collects the WRONG
 * video (a cross-project contamination that silently ships another project's
 * footage). Match by the `scene[-_]<i>.mp4` tag instead; the trailing `.mp4`
 * anchor stops `scene_1` from matching `scene_10`, and among tag-matches the
 * newest by mtime is this scene's clip (it just finished rendering). The
 * separator class `[-_]` tolerates both flow.ts's `_scene_<i>` and any
 * `-scene-<i>` variant. Falls back to the newest untagged file only when
 * nothing carries the tag, so single-project runs keep working even if flow.ts
 * ever changes its naming.
 */
async function pickSceneOutput(candidates: string[], sceneIndex: number): Promise<string | null> {
  if (candidates.length === 0) return null;
  const tagPattern = new RegExp(`scene[-_]${sceneIndex}\\.mp4$`);
  const withMtime = await Promise.all(
    candidates.map(async (path) => ({ path, mtimeMs: (await stat(path)).mtimeMs })),
  );
  const tagged = withMtime.filter((entry) => tagPattern.test(entry.path));
  const pool = tagged.length > 0 ? tagged : withMtime;
  pool.sort((a, b) => b.mtimeMs - a.mtimeMs);
  return pool[0]?.path ?? null;
}

export async function submitVeoUseApiNative(
  payload: VideoExecutionPayload,
  options: {
    env?: NodeJS.ProcessEnv;
  } = {},
): Promise<{
  externalJobId: string;
  rawResult: unknown;
}> {
  const env = await loadWorkspaceEnv(payload.workspaceRoot, options.env ?? process.env);
  const cliRoot = veoCliRoot(payload.workspaceRoot, env);
  ensureVeoCliEntry(cliRoot);
  const outputDir = veoOutputDir(payload.workspaceRoot, env);
  await mkdir(outputDir, { recursive: true });
  const externalJobId = `veo-useapi-${Date.now()}`;
  const outputs: VideoExecutionPollResult['outputs'] = [];

  const maxAttempts = veoMaxAttempts(env);

  const model = veoModelFlag(payload.executionProfile);
  const isOmni = model === 'omni-flash';
  // omni-flash First-Frame is gated; off → task.firstFrame is ignored and the
  // prompt is byte-identical to today (R2V ingredients).
  const omniFirstFrameEnabled = isOmni && isOmniFirstFrameEnabled(env);

  for (const task of payload.tasks) {
    const args: string[] = [
      'run',
      'flow.ts',
      '-p',
      buildPrompt(task, isOmni, omniFirstFrameEnabled && !!task.firstFrame),
      '-n',
      String(payload.executionProfile.outputCount),
      '-r',
      veoRatio(payload.executionProfile.aspectRatio),
      '-m',
      model,
    ];
    // (R2V image references are encoded in the prompt via buildPrompt's
    // `ingredients:` syntax — flow.ts reads refs from the prompt, not a flag.)
    // Optional Flow v1 / omni-flash flags appended strictly BETWEEN -m and the
    // trailing --backend/--yes, so the legacy tail (and its assertions) hold.
    if (task.durationSeconds !== undefined && VEO_DURATIONS.has(task.durationSeconds)) {
      args.push('--duration', String(task.durationSeconds));
    }
    if (task.voicePreset) {
      args.push('--voice', task.voicePreset);
    }
    // Saved Google Flow character refs → repeated `--character` flags (R2V entity
    // mode: locked identity + bundled voice). Absent → byte-identical legacy.
    if (task.characterRefs && task.characterRefs.length > 0) {
      for (const ref of task.characterRefs.slice(0, 7)) {
        if (ref) args.push('--character', ref);
      }
    }
    // V2V edit reference (omni-flash). Triggered ONLY by the explicit
    // referenceVideoMediaId field — never by inputKind 'video' (the
    // scene-chaining seed), so chained payloads stay byte-identical.
    if (task.referenceVideoMediaId) {
      args.push('--ref-video', task.referenceVideoMediaId);
    }
    args.push('--backend', 'useapi', '--yes');

    // Retry on TRANSIENT infra failures (socket drops / 5xx, short backoff) and
    // on burst-THROTTLE (reCAPTCHA / 429, long adaptive cooldown). The first
    // attempt is byte-identical to the legacy single-shot; moderation and other
    // non-retryable failures still throw immediately (no wasted credits).
    let produced = false;
    let softened = false;
    let lastDetail = '';
    for (let attempt = 1; attempt <= maxAttempts; attempt++) {
      const before = await listFiles(outputDir);
      const startedAt = Date.now();
      const commandResult = await runVeoCommand(args, { cwd: cliRoot, env });
      const newOutputs = await captureNewOutputs(outputDir, before, startedAt);
      // Pick THIS scene's clip by its `scene_<i>` tag — never just the first new
      // file, which under a shared output-videos/ could be a concurrent
      // project's render (see {@link pickSceneOutput}).
      const sceneOutput = await pickSceneOutput(newOutputs, task.sceneIndex);
      if (sceneOutput) {
        // flow.ts renders synchronously into the Veo CLI's `output-videos/` dir
        // with a timestamped name. COLLECT it into the project's outputs as
        // `scene-<i>.mp4` — the standard per-scene clip location every consumer
        // (execute-status, scene-candidates, assemble/clip-stitch, preview)
        // looks for, matching the genuinely-async routes (native-runway /
        // native-seedance download straight to `<outputDir>/scene-<i>.mp4`).
        // Without this the rendered clips are invisible to the rest of the
        // pipeline. The original stays in `output-videos/` as a cache.
        await mkdir(payload.outputDir, { recursive: true });
        const collectedPath = join(payload.outputDir, `scene-${task.sceneIndex}.mp4`);
        await copyFile(sceneOutput, collectedPath);
        outputs.push({
          id: `generated-scene-${task.sceneIndex}`,
          kind: 'video',
          path: collectedPath,
          sceneIndex: task.sceneIndex,
          backend: 'veo-useapi',
        });
        produced = true;
        break;
      }
      lastDetail = commandResult.stderr.trim() || commandResult.stdout.trim();
      if (lastDetail && isSessionRefreshFailure(lastDetail)) {
        throw new Error(
          `veo-useapi native transport did not produce an output file for scene ${task.sceneIndex} because session refresh failed. ` +
          `Update Google-flow cookies for the Veo CLI and retry (cookie hint: ${join(cliRoot, 'cookie.json')}, docs: https://useapi.net/docs/start-here/setup-google-flow). ` +
          `Original output: ${lastDetail}`,
        );
      }
      const failureClass = classifyVeoFailure(lastDetail);
      if ((failureClass === 'transient' || failureClass === 'throttle') && attempt < maxAttempts) {
        // Throttle (reCAPTCHA / 429 / 403) clears only after a long cooldown; a
        // transient socket blip retries on a short escalating backoff. Moderation
        // and fatal failures fall through and throw (no wasted credits).
        await delay(failureClass === 'throttle' ? veoThrottleCooldownMs(env, attempt) : 2000 * attempt);
        continue;
      }
      // Opt-in safe-motion: a content-moderation reject is deterministic, so a
      // plain retry is futile — but softening the aggressive wording ONCE (only
      // when it actually changes the prompt) often clears the filter (the film
      // panel-10 reword). Default off → moderation still fails fast, no wasted
      // credits and no silent prompt rewrites.
      if (failureClass === 'moderation' && !softened && attempt < maxAttempts && veoSafeMotionEnabled(env)) {
        const promptIdx = args.indexOf('-p') + 1;
        const soft = promptIdx > 0 ? softenMotionPrompt(args[promptIdx]) : { prompt: '', changed: false };
        if (soft.changed) {
          args[promptIdx] = soft.prompt;
          softened = true;
          continue;
        }
      }
      break;
    }
    if (!produced) {
      // The Flow sync path often masks a content-moderation reject as an opaque
      // "All operations failed" (the real PUBLIC_ERROR_UNSAFE_GENERATION only
      // surfaces on the async/raw-jobId path). Surface the likely cause + the
      // safe-motion escape hatch instead of leaving the operator guessing.
      const looksModerated =
        classifyVeoFailure(lastDetail) === 'moderation' || /all operations failed/i.test(lastDetail);
      const moderationHint = looksModerated
        ? ' This often indicates Veo content moderation on the motion prompt — soften aggressive wording'
          + ' (e.g. "violent"/"wrath"/"shockwave") or set VCLAW_VEO_SAFE_MOTION=1 to auto-soften and retry once.'
        : '';
      throw new Error(
        `veo-useapi native transport did not produce an output file for scene ${task.sceneIndex} (cliRoot=${cliRoot}, outputDir=${outputDir})${lastDetail ? `; command output: ${lastDetail}` : ''}.${moderationHint}`,
      );
    }
  }

  await writeJobState({
    externalJobId,
    routeId: 'veo-useapi',
    outputDir: payload.outputDir,
    createdAt: new Date().toISOString(),
    outputs,
  });

  return {
    externalJobId,
    rawResult: {
      externalJobId,
      outputs,
    },
  };
}

export async function pollVeoUseApiNative(
  input: {
    outputDir: string;
    externalJobId: string;
  },
): Promise<VideoExecutionPollResult> {
  const state = await readJobState(input.outputDir, input.externalJobId);
  return {
    status: 'completed',
    externalJobId: input.externalJobId,
    outputs: state.outputs,
    issues: [],
    rawResult: state,
  };
}