import { safeErrorBody } from './http-error-safety.js';
import { existsSync } from 'node:fs';
import { mkdir, readFile, rename, unlink, writeFile } from 'node:fs/promises';
import { writeTextFileAtomic } from './atomic-write.js';
import { dirname, join } from 'node:path';
import {
  submitRunwayJob,
  pollRunwayJob,
  fetchRunwayResult,
  RUNWAY_MAX_IMAGE_REFS,
  RUNWAY_MAX_VIDEO_REFS,
  type RunwayAspectRatio,
  type RunwayDurationSeconds,
  type RunwayMode,
  type RunwayModel,
} from './providers/runway-useapi.js';
import { isExploreThrottled } from './batch-queue.js';
import type { VideoExecutionCancelResult, VideoExecutionPayload, VideoExecutionPollResult } from './types.js';

interface RunwayJobSceneState {
  sceneIndex: number;
  prompt: string;
  taskId: string;
  outputPath: string;
  status: 'submitted' | 'completed' | 'failed';
  error?: string;
}

interface RunwayNativeJobState {
  externalJobId: string;
  routeId: 'runway-useapi';
  outputDir: string;
  createdAt: string;
  scenes: RunwayJobSceneState[];
}

interface FetchLikeResponse {
  ok: boolean;
  status: number;
  text(): Promise<string>;
  json(): Promise<unknown>;
  arrayBuffer(): Promise<ArrayBuffer>;
}

type FetchLike = (input: string, init?: {
  method?: string;
  headers?: Record<string, string>;
  body?: string | Uint8Array;
}) => Promise<FetchLikeResponse>;

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 getUseApiToken(env: NodeJS.ProcessEnv): string {
  const token = env.USEAPI_API_TOKEN;
  if (!token || !token.trim()) {
    throw new Error('runway-useapi native transport requires USEAPI_API_TOKEN.');
  }
  return token.trim();
}

function defaultRunwayModel(env: NodeJS.ProcessEnv): RunwayModel {
  const raw = (env.VCLAW_RUNWAY_MODEL ?? '').trim().toLowerCase();
  if (raw === 'gen-4.5' || raw === 'gen-4' || raw === 'gen-4-turbo' || raw === 'seedance-2.0') {
    return raw;
  }
  // Seedance-2 via Runway is the production default — it supports keyframe
  // (image-to-video) plus all aspect ratios we ship, with the widest duration
  // set (5/8/10/15s).
  return 'seedance-2.0';
}

function defaultRunwayMode(env: NodeJS.ProcessEnv): RunwayMode {
  const raw = (env.VCLAW_RUNWAY_MODE ?? '').trim().toLowerCase();
  if (raw === 'credits' || raw === 'credit') return 'credits';
  // 'explore' is the safe, free-tier default. videoclaw's queue executor
  // defaulted to explore as well.
  return 'explore';
}

function aspectRatioFor(profileAspect: VideoExecutionPayload['executionProfile']['aspectRatio']): RunwayAspectRatio {
  if (profileAspect === '9:16') return '9:16';
  if (profileAspect === '1:1') return '1:1';
  return '16:9';
}

function clampDuration(seconds: number | undefined): RunwayDurationSeconds {
  const value = Number.isFinite(seconds) ? Number(seconds) : 8;
  if (value >= 15) return 15;
  if (value >= 10) return 10;
  if (value >= 8) return 8;
  return 5;
}

function classifyReferences(referencePaths: string[]): { images: string[]; videos: string[] } {
  const imageExt = new Set(['.jpg', '.jpeg', '.png', '.webp', '.gif']);
  const videoExt = new Set(['.mp4', '.mov', '.webm', '.avi', '.mkv']);
  const images: string[] = [];
  const videos: string[] = [];
  for (const path of referencePaths) {
    if (!path) continue;
    const ext = (path.split('?')[0]?.match(/\.[^.\\/]+$/)?.[0] ?? '').toLowerCase();
    if (videoExt.has(ext)) videos.push(path);
    else if (imageExt.has(ext)) images.push(path);
    else images.push(path); // unknown → treat as image to avoid silently dropping
  }
  return { images, videos };
}

/**
 * Preflight the Runway/Seedance-2 reference budget before any upload or submit.
 * Mirrors `assertReferenceBudget` (native-seedance): fail-fast with a clear
 * error rather than letting the gateway reject a partially-uploaded payload.
 * Runway caps: ≤11 images, ≤3 videos per submission.
 */
function assertRunwayReferenceBudget(sceneIndex: number, images: string[], videos: string[]): void {
  if (images.length > RUNWAY_MAX_IMAGE_REFS) {
    throw new Error(
      `runway-useapi scene ${sceneIndex}: ${images.length} image references exceed the Runway cap of ${RUNWAY_MAX_IMAGE_REFS}.`,
    );
  }
  if (videos.length > RUNWAY_MAX_VIDEO_REFS) {
    throw new Error(
      `runway-useapi scene ${sceneIndex}: ${videos.length} video references exceed the Runway cap of ${RUNWAY_MAX_VIDEO_REFS}.`,
    );
  }
}

/**
 * Best-effort prompt pre-validation against the shared Seedance content filter.
 * Seedance moderation is identical across all three gateways (ARK / Runway /
 * Dreamina), so the Runway transport reuses `seedance-content-filter` to surface
 * HIGH/MEDIUM-risk warnings. Loaded defensively so the transport never hard-fails
 * if the module's surface changes; warnings are advisory (returned, not thrown).
 */
async function preValidateRunwayPrompt(prompt: string): Promise<string[]> {
  try {
    const mod: Record<string, unknown> = await import('./seedance-content-filter.js');
    const fn = mod.preValidatePrompt as
      | ((p: string) => { warnings?: string[]; messages?: string[]; reasons?: string[] } | string[])
      | undefined;
    if (typeof fn !== 'function') return [];
    const result = fn(prompt);
    if (Array.isArray(result)) {
      // preValidatePrompt returns ContentFilterWarning[] ({level,reason,match}).
      // Format objects the same way native-dreamina/native-seedance do (HIGH/MEDIUM
      // only) instead of String(obj) -> "[object Object]". A plain string[] (the
      // defensive alt-shape) passes through unchanged.
      return result
        .filter((w) =>
          typeof w === 'string' ||
          (w as { level?: string }).level === 'HIGH' ||
          (w as { level?: string }).level === 'MEDIUM',
        )
        .map((w) =>
          typeof w === 'string'
            ? w
            : `${(w as { level?: string }).level} risk: ${(w as { reason?: string }).reason} (match: ${(w as { match?: string }).match})`,
        );
    }
    if (result && typeof result === 'object') {
      const out = result.warnings ?? result.messages ?? result.reasons ?? [];
      return Array.isArray(out) ? out.map(String) : [];
    }
    return [];
  } catch {
    // Content filter is advisory only — never block submission on its absence.
    return [];
  }
}

async function readReferenceBytes(path: string, fetchImpl: FetchLike): Promise<Buffer | null> {
  if (!path) return null;
  if (path.startsWith('http://') || path.startsWith('https://')) {
    const response = await fetchImpl(path);
    if (!response.ok) {
      throw new Error(`runway-useapi reference fetch failed (HTTP ${response.status}): ${path}`);
    }
    return Buffer.from(await response.arrayBuffer());
  }
  if (path.startsWith('Asset://')) {
    // Asset:// is a videoclaw asset-library URI we can't resolve here; the
    // pipeline must pre-resolve it before invoking the runway transport.
    return null;
  }
  if (!existsSync(path)) return null;
  return readFile(path);
}

/**
 * MIME type for a reference upload, inferred from the file extension. Image
 * refs default to `image/jpeg`; video refs must carry their real video MIME so
 * UseAPI stores them as a video asset (a `.mp4` uploaded as `image/jpeg` is
 * rejected / mis-typed). Mirrors `referenceContentType` in native-dreamina.
 */
function referenceContentType(path: string): string {
  const ext = (path.split('?')[0]?.match(/\.[^.\\/]+$/)?.[0] ?? '').toLowerCase();
  if (ext === '.png') return 'image/png';
  if (ext === '.webp') return 'image/webp';
  if (ext === '.gif') return 'image/gif';
  if (ext === '.mp4') return 'video/mp4';
  if (ext === '.mov') return 'video/quicktime';
  if (ext === '.webm') return 'video/webm';
  if (ext === '.avi') return 'video/x-msvideo';
  if (ext === '.mkv') return 'video/x-matroska';
  return 'image/jpeg';
}

async function uploadRunwayAsset(
  apiToken: string,
  bytes: Buffer,
  name: string,
  fetchImpl: FetchLike,
  contentType = 'image/jpeg',
): Promise<string> {
  const url = `https://api.useapi.net/v1/runwayml/assets/?name=${encodeURIComponent(name)}`;
  const response = await fetchImpl(url, {
    method: 'POST',
    headers: { Authorization: `Bearer ${apiToken}`, 'Content-Type': contentType },
    body: bytes,
  });
  if (!response.ok) {
    throw new Error(`runway-useapi asset upload failed (HTTP ${response.status}): ${safeErrorBody(await response.text().catch(() => ''))}`);
  }
  const json = (await response.json()) as { assetId?: string; asset?: { assetId?: string }; id?: string };
  const assetId = json.assetId ?? json.asset?.assetId ?? json.id;
  if (!assetId) {
    throw new Error(`runway-useapi asset upload returned no assetId: ${JSON.stringify(json)}`);
  }
  return assetId;
}

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: RunwayNativeJobState): 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<RunwayNativeJobState> {
  const path = jobStatePath(outputDir, externalJobId);
  if (!existsSync(path)) {
    throw new Error(`Runway native job state not found for ${externalJobId}.`);
  }
  const raw = await readFile(path, 'utf-8');
  try {
    return JSON.parse(raw) as RunwayNativeJobState;
  } catch (error) {
    // Uniform with seedance/veo: a present-but-unparseable file is a clear
    // blocker, never silently treated as absent (which would risk a re-submit).
    throw new Error(
      `Runway native job state for ${externalJobId} is corrupt (invalid JSON at ${path}): ${error instanceof Error ? error.message : String(error)}`,
    );
  }
}

async function downloadToFile(fetchImpl: FetchLike, url: string, outputPath: string): Promise<void> {
  const response = await fetchImpl(url);
  if (!response.ok) {
    throw new Error(`runway-useapi download failed (HTTP ${response.status}): ${safeErrorBody(await response.text().catch(() => ''))}`);
  }
  await mkdir(dirname(outputPath), { recursive: true });
  const tmpPath = `${outputPath}.tmp`;
  try {
    await writeFile(tmpPath, Buffer.from(await response.arrayBuffer()));
    await rename(tmpPath, outputPath);
  } catch (error) {
    // Clean the partial .tmp but let the REAL failure (disk full, perms,
    // rename onto a locked path) propagate instead of being masked.
    await unlink(tmpPath).catch(() => {});
    throw error;
  }
}

export async function submitRunwayUseApiNative(
  payload: VideoExecutionPayload,
  options: {
    env?: NodeJS.ProcessEnv;
    fetchImpl?: FetchLike;
    /**
     * Stable job id to resume into. When provided, existing per-scene state at
     * `.vclaw-jobs/<externalJobId>.json` is loaded and already-submitted scenes
     * are skipped (never re-submitted). Omit for a fresh submission.
     */
    externalJobId?: string;
  } = {},
): Promise<{
  externalJobId: string;
  rawResult: unknown;
}> {
  const env = await loadWorkspaceEnv(payload.workspaceRoot, options.env ?? process.env);
  const apiToken = getUseApiToken(env);
  const fetchImpl = options.fetchImpl ?? (fetch as unknown as FetchLike);
  const model = defaultRunwayModel(env);
  const mode = defaultRunwayMode(env);
  const aspectRatio = aspectRatioFor(payload.executionProfile.aspectRatio);
  // Output resolution ('720p' | '1080p') from the execution profile. The
  // provider only emits a `resolution` field when this is set; without it the
  // gateway falls back to the model default (Seedance 2.0 = 480p), which is why
  // free-lane clips came back ~496×864 even when 720p/1080p was requested.
  // explore mode does NOT cap resolution on Seedance 2.0 (480p/720p/1080p all
  // valid), so forwarding this is what actually unlocks 720p/1080p output.
  const resolution = payload.executionProfile.resolution;
  // Provider-generated audio (Seedance-2 only): the user's `--audio on` intent
  // (executionProfile.generateAudio) or an explicit VCLAW_RUNWAY_AUDIO override.
  // With a voice-reference video present, this is what makes the character speak
  // in the cloned voice. Default off (generateAudio defaults false) → byte-identical.
  const wantAudio = payload.executionProfile.generateAudio === true
    || env.VCLAW_RUNWAY_AUDIO === '1' || env.VCLAW_RUNWAY_AUDIO === 'true';
  const externalJobId = options.externalJobId ?? `runway-useapi-${Date.now()}`;
  const createdAt = new Date().toISOString();

  // Resume: seed from any existing job state so already-submitted scenes are
  // skipped (not re-fired) when this is called again after an explore throttle.
  const scenes: RunwayJobSceneState[] = [];
  if (options.externalJobId && existsSync(jobStatePath(payload.outputDir, externalJobId))) {
    try {
      scenes.push(...(await readJobState(payload.outputDir, externalJobId)).scenes);
    } catch {
      // Corrupt/partial state file → start fresh rather than crash.
    }
  }
  const alreadySubmitted = new Set(
    scenes.filter((s) => s.taskId && (s.status === 'submitted' || s.status === 'completed')).map((s) => s.sceneIndex),
  );
  const persist = () =>
    writeJobState({ externalJobId, routeId: 'runway-useapi', outputDir: payload.outputDir, createdAt, scenes });

  const rawResponses: unknown[] = [];

  const warnings: string[] = [];
  let throttled = false;

  // Preflight EVERY pending task's reference budget before uploading anything,
  // so a scene-5 violation cannot land after scenes 1-4 already uploaded assets
  // and submitted (parity with the seedance/dreamina whole-payload fail-fast).
  // The per-task assert below stays as a cheap invariant re-check.
  for (const task of payload.tasks) {
    if (alreadySubmitted.has(task.sceneIndex)) continue;
    const { images, videos } = classifyReferences(task.referencePaths);
    assertRunwayReferenceBudget(
      task.sceneIndex,
      images.filter((path) => !path.startsWith('Asset://')),
      videos.filter((path) => !path.startsWith('Asset://')),
    );
  }

  for (const task of payload.tasks) {
    // Resume-skip: a scene already submitted on a prior pass must not be re-sent.
    if (alreadySubmitted.has(task.sceneIndex)) continue;
    const { images, videos } = classifyReferences(task.referencePaths);
    // Asset:// URIs are ARK Asset-Library avatar references — they are NOT
    // valid Runway asset ids and cannot be resolved against UseAPI's Runway
    // proxy, so they are skipped here (with a logged note). Only real
    // file-path / HTTP-URL refs are uploadable as Runway assets.
    const dropAssetUris = (paths: string[], kind: 'image' | 'video') =>
      paths.filter((path) => {
        if (path.startsWith('Asset://')) {
          warnings.push(
            `runway-useapi scene ${task.sceneIndex}: skipped Asset:// ${kind} reference (ARK avatar, not a Runway asset id): ${path}`,
          );
          return false;
        }
        return true;
      });
    const uploadableImages = dropAssetUris(images, 'image');
    const uploadableVideos = dropAssetUris(videos, 'video');

    // Fail-fast on the reference budget BEFORE uploading anything.
    assertRunwayReferenceBudget(task.sceneIndex, uploadableImages, uploadableVideos);

    // Surface Seedance content-moderation warnings (shared across gateways).
    for (const note of await preValidateRunwayPrompt(task.prompt)) {
      warnings.push(`runway-useapi scene ${task.sceneIndex}: content-filter: ${note}`);
    }

    // Upload ALL usable images in referencePaths order. Each successful upload
    // yields a Runway asset id; unresolved/missing paths are skipped.
    const imageAssetIds: string[] = [];
    for (let i = 0; i < uploadableImages.length; i += 1) {
      const bytes = await readReferenceBytes(uploadableImages[i], fetchImpl);
      if (!bytes) continue;
      const assetId = await uploadRunwayAsset(
        apiToken,
        bytes,
        `scene-${task.sceneIndex}-ref-${i + 1}`,
        fetchImpl,
        referenceContentType(uploadableImages[i]),
      );
      imageAssetIds.push(assetId);
    }

    // Upload video references (Seedance-2 motion/style refs) the same way,
    // with their real video MIME so UseAPI stores them as video assets. The
    // resulting ids feed `videoAssetIds` → `videoAssetId`/`videoAssetIdN`
    // (capped at RUNWAY_MAX_VIDEO_REFS). Dreamina/Seedance already wired this;
    // Runway previously dropped videos on the floor.
    const videoAssetIds: string[] = [];
    for (let i = 0; i < uploadableVideos.length; i += 1) {
      const bytes = await readReferenceBytes(uploadableVideos[i], fetchImpl);
      if (!bytes) continue;
      const assetId = await uploadRunwayAsset(
        apiToken,
        bytes,
        `scene-${task.sceneIndex}-vref-${i + 1}`,
        fetchImpl,
        referenceContentType(uploadableVideos[i]),
      );
      videoAssetIds.push(assetId);
    }

    // Routing (Seedance-2):
    //  - CHARACTER references (task.referenceRole === 'character') → ALWAYS the
    //    multi-reference slot (imageAssetId1..N), even for a single sheet, so a
    //    lone character sheet is never the literal first frame (that made solo
    //    clips open on the character grid).
    //  - KEYFRAME (default) + exactly 1 → single keyframe (startFrameAssetId).
    //  - >1 image → multi-reference regardless (can't be one keyframe).
    //  - VOICE VIDEOS present → a single image must be a REFERENCE image
    //    (imageAssetId1), never a keyframe: Seedance-2 rejects mixing a keyframe
    //    (startFrameAssetId) with video references ("Cannot mix keyframe with
    //    video references"). The talking-cartoon scene (identity still + voice
    //    video) hits exactly this, so force multi-ref when voices are present.
    //  - Gen-4.x + ≥1 image → firstImageAssetId i2v (uses first image only).
    const asCharacterRefs = task.referenceRole === 'character';
    const hasVoiceVideos = videoAssetIds.length > 0;
    const useMultiRef = imageAssetIds.length > 1
      || ((asCharacterRefs || hasVoiceVideos) && imageAssetIds.length === 1);
    const singleKeyframe = !asCharacterRefs && !hasVoiceVideos && imageAssetIds.length === 1
      ? imageAssetIds[0]
      : null;

    // Optional END keyframe (Seedance-2 keyframe interpolation: the clip
    // animates startFrame → endFrame). Uploaded separately from referencePaths
    // so it never counts as a second image (which would trip multi-ref). Only
    // meaningful on seedance-2.0 alongside a single start keyframe.
    let endFrameAssetId: string | null = null;
    if (task.endKeyframePath && singleKeyframe && model === 'seedance-2.0') {
      const endBytes = await readReferenceBytes(task.endKeyframePath, fetchImpl);
      if (endBytes) {
        endFrameAssetId = await uploadRunwayAsset(
          apiToken,
          endBytes,
          `scene-${task.sceneIndex}-endframe`,
          fetchImpl,
          referenceContentType(task.endKeyframePath),
        );
      } else {
        warnings.push(
          `runway-useapi scene ${task.sceneIndex}: end keyframe could not be read, rendering single-frame: ${task.endKeyframePath}`,
        );
      }
    }

    const seconds = clampDuration(task.durationSeconds);
    let submit: { taskId: string };
    try {
      submit = await submitRunwayJob({
        apiToken,
        model,
        textPrompt: task.prompt,
        mode,
        seconds,
        aspectRatio,
        ...(resolution ? { resolution } : {}),
        ...(model === 'seedance-2.0' && useMultiRef ? { imageAssetIds } : {}),
        // A lone CHARACTER ref must take the multi-ref slot, not the keyframe
        // convenience path in submitRunwayJob — force it (the grid-open fix).
        ...(model === 'seedance-2.0' && useMultiRef && imageAssetIds.length === 1 ? { forceMultiRef: true } : {}),
        ...(model === 'seedance-2.0' && singleKeyframe ? { startFrameAssetId: singleKeyframe } : {}),
        ...(model === 'seedance-2.0' && singleKeyframe && endFrameAssetId ? { endFrameAssetId } : {}),
        ...(model !== 'seedance-2.0' && singleKeyframe ? { firstImageAssetId: singleKeyframe } : {}),
        ...(model !== 'seedance-2.0' && useMultiRef ? { firstImageAssetId: imageAssetIds[0] } : {}),
        // Video references map to videoAssetId/videoAssetIdN in BOTH the
        // multi-ref and single-ref provider branches, so pass them whenever
        // present regardless of the image-routing mode above.
        ...(videoAssetIds.length > 0 ? { videoAssetIds } : {}),
        // Provider-generated audio (seedance-2.0): speaks dialogue, and locks the
        // cloned voice when a voice-reference video rode in via videoAssetIds.
        ...(model === 'seedance-2.0' && wantAudio ? { audio: true } : {}),
        fetchImpl,
      });
    } catch (error) {
      // Explore-mode throttle (1 concurrent): NOT a failure. Stop submitting,
      // keep the scenes already persisted this pass, and report the remainder
      // as pending so a later pass resumes them — never re-firing earlier scenes.
      if (isExploreThrottled(error instanceof Error ? error.message : String(error))) {
        throttled = true;
        break;
      }
      // A real submit failure: persist progress so far, then surface it.
      await persist();
      throw error;
    }
    rawResponses.push({ sceneIndex: task.sceneIndex, taskId: submit.taskId, imageAssetIds, videoAssetIds, ...(endFrameAssetId ? { endFrameAssetId } : {}) });

    scenes.push({
      sceneIndex: task.sceneIndex,
      prompt: task.prompt,
      taskId: submit.taskId,
      outputPath: join(payload.outputDir, `scene-${task.sceneIndex}.mp4`),
      status: 'submitted',
    });
    // Incremental persist: each submitted scene is durable immediately, so a
    // throttle on a *later* scene cannot lose this one's state.
    await persist();
  }

  await persist();

  const submittedSceneIndexes = scenes
    .filter((s) => s.status === 'submitted' || s.status === 'completed')
    .map((s) => s.sceneIndex);
  const submittedSet = new Set(submittedSceneIndexes);
  const pendingSceneIndexes = payload.tasks
    .map((t) => t.sceneIndex)
    .filter((i) => !submittedSet.has(i));

  return {
    externalJobId,
    rawResult: {
      externalJobId,
      model,
      mode,
      throttled,
      submittedSceneIndexes,
      pendingSceneIndexes,
      submittedScenes: scenes.map((scene) => ({ sceneIndex: scene.sceneIndex, taskId: scene.taskId })),
      responses: rawResponses,
      ...(warnings.length > 0 ? { warnings } : {}),
    },
  };
}

export async function pollRunwayUseApiNative(
  input: {
    outputDir: string;
    externalJobId: string;
    workspaceRoot: string;
  },
  options: {
    env?: NodeJS.ProcessEnv;
    fetchImpl?: FetchLike;
  } = {},
): Promise<VideoExecutionPollResult> {
  const env = await loadWorkspaceEnv(input.workspaceRoot, options.env ?? process.env);
  const apiToken = getUseApiToken(env);
  const fetchImpl = options.fetchImpl ?? (fetch as unknown as FetchLike);
  const state = await readJobState(input.outputDir, input.externalJobId);

  const outputs: VideoExecutionPollResult['outputs'] = [];
  const issues: string[] = [];
  const rawResults: unknown[] = [];
  let anyPending = false;
  let anyFailed = false;

  for (const scene of state.scenes) {
    if (scene.status === 'completed' && existsSync(scene.outputPath)) {
      outputs.push({
        id: `generated-scene-${scene.sceneIndex}`,
        kind: 'video',
        path: scene.outputPath,
        sceneIndex: scene.sceneIndex,
        backend: 'runway-useapi',
      });
      continue;
    }
    if (scene.status === 'failed') {
      anyFailed = true;
      if (scene.error) issues.push(scene.error);
      continue;
    }
    const polled = await pollRunwayJob({ apiToken, taskId: scene.taskId, fetchImpl });
    rawResults.push({ sceneIndex: scene.sceneIndex, status: polled.status, progress: polled.progress });
    if (polled.status === 'completed') {
      const fetched = await fetchRunwayResult({ apiToken, taskId: scene.taskId, fetchImpl });
      if (!fetched.videoUrl) {
        scene.status = 'failed';
        scene.error = `runway-useapi scene ${scene.sceneIndex} completed without a video URL.`;
        issues.push(scene.error);
        anyFailed = true;
        continue;
      }
      if (!existsSync(scene.outputPath)) {
        await downloadToFile(fetchImpl, fetched.videoUrl, scene.outputPath);
      }
      scene.status = 'completed';
      outputs.push({
        id: `generated-scene-${scene.sceneIndex}`,
        kind: 'video',
        path: scene.outputPath,
        sceneIndex: scene.sceneIndex,
        backend: 'runway-useapi',
      });
    } else if (polled.status === 'failed') {
      scene.status = 'failed';
      scene.error = `runway-useapi scene ${scene.sceneIndex} failed: ${JSON.stringify(polled.raw).slice(0, 400)}`;
      issues.push(scene.error);
      anyFailed = true;
    } else {
      anyPending = true;
    }
  }

  await writeJobState(state);

  return {
    status: anyFailed ? 'failed' : anyPending ? 'pending' : 'completed',
    externalJobId: input.externalJobId,
    outputs,
    issues,
    rawResult: rawResults,
  };
}

export async function cancelRunwayUseApiNative(
  input: {
    outputDir: string;
    externalJobId: string;
    workspaceRoot: string;
  },
): Promise<VideoExecutionCancelResult> {
  // UseAPI's Runway endpoints don't expose a cancel verb — best we can do is
  // mark local state failed so subsequent polls don't keep waiting. Pending
  // server-side tasks will continue and consume credits/quota until they
  // resolve on their own.
  const state = await readJobState(input.outputDir, input.externalJobId);
  for (const scene of state.scenes) {
    if (scene.status === 'submitted') {
      scene.status = 'failed';
      scene.error = 'Execution cancelled by operator (runway-useapi has no server-side cancel; remote task continues).';
    }
  }
  await writeJobState(state);
  return {
    status: 'cancelled',
    externalJobId: input.externalJobId,
    issues: [
      'runway-useapi has no UseAPI cancel endpoint; remote tasks may continue to run and consume quota.',
    ],
    rawResult: null,
  };
}