/**
 * native-modelark.ts — the `seedance-modelark` route's transport: BytePlus
 * ModelArk (Dreamina Seedance 2.5 / 2.0) as an in-process submit / poll /
 * cancel over the pure client in `providers/modelark.ts`.
 *
 * Spend discipline, in order:
 *   1. Every scene is planned AND its references materialised before the first
 *      paid create, so an over-budget scene N cannot leave scenes 0..N-1 billed.
 *   2. Each create is ONE POST, never retried (a replay could queue two charged
 *      tasks); job state is written after EACH successful create, so a failure
 *      mid-loop leaves the tasks already paid for reachable by `execute-status`.
 *   3. `content.video_url` lives 24 h / 100 downloads: a succeeded task is
 *      downloaded in the same poll that sees it.
 *   4. DELETE only ever targets a scene we are still waiting for: on a finished
 *      task it would erase the record, not the bill.
 *   5. A create is recorded as an INTENT before the POST. When the answer is
 *      lost (network, timeout, 5xx, a 2xx with no id) the scene is
 *      `submit-unknown`: the task may exist and be billing. The next poll
 *      looks for it in the vendor's task list by model and creation window
 *      and binds it only when exactly one unbound task is there; none after
 *      the window means the create was refused (not billed); several means
 *      it stays ambiguous, named, and nothing is re-submitted (ADR 0008).
 */
import { existsSync } from 'node:fs';
import { mkdir, readdir, readFile, rename, unlink, writeFile } from 'node:fs/promises';
import { dirname, join } from 'node:path';
import { writeTextFileAtomic } from './atomic-write.js';
import { safeErrorBody } from './http-error-safety.js';
import { preValidatePrompt, suggestRecovery } from './seedance-content-filter.js';
import type { VideoExecutionCancelResult, VideoExecutionPayload, VideoExecutionPollResult } from './types.js';
import {
  buildModelArkCreateBody,
  classifyModelArkError,
  MODELARK_MODELS,
  modelArkBaseUrl,
  modelArkCostUsd,
  modelArkCreatedAtSeconds,
  modelArkListTasks,
  isAmbiguousCreateError,
  modelArkCreateTask,
  modelArkDeleteTask,
  modelArkGetTask,
  ModelArkHttpError,
  planModelArkTask,
  resolveModelArkApiKey,
  resolveModelArkModel,
  modelArkChainOptions,
  MODELARK_CHAIN_MODE_ENV,
  type ModelArkClientContext,
  type ModelArkFetchLike,
  type ModelArkModelId,
  type ModelArkTaskPlan,
} from './providers/modelark.js';
import {
  defaultModelArkReferenceDeps,
  materializeModelArkReferences,
  type ModelArkReferenceDeps,
} from './providers/modelark-references.js';

export const SEEDANCE_MODELARK_ROUTE_ID = 'seedance-modelark' as const;

export interface ModelArkJobSceneState {
  sceneIndex: number;
  prompt: string;
  /** The vendor's task id. Absent while a create is in flight or its answer was lost. */
  taskId?: string;
  outputPath: string;
  /**
   * `submitting`: the intent is on file and the POST is in flight (a file
   * still saying so after the process died means the answer was lost).
   * `submit-unknown`: the create's answer was lost; the task may exist.
   */
  status: 'submitting' | 'submit-unknown' | 'submitted' | 'completed' | 'failed';
  error?: string;
  /**
   * When THIS scene's poll wrote `outputPath` (ISO). A file at that path this
   * job did not download belongs to something else — another job in the same
   * output dir renders to the same `scene-N.mp4` — so the poll may only skip
   * the download when it knows it made that file.
   */
  downloadedAt?: string;
  /** When the create was attempted (ISO); the anchor for finding a lost task. */
  intentAt?: string;
  /** What went wrong with the create whose answer was lost. */
  submitError?: string;
  lastFrameUrl?: string;
  /**
   * What was asked for, kept so the cost can be read off the vendor's usage
   * later. Optional only because a file written before cost accounting has
   * neither; every submit since writes both.
   */
  resolution?: '480p' | '720p' | '1080p';
  /** Whether the request carried a reference video (the vendor prices that differently). */
  videoInput?: boolean;
  /** The vendor's output token count on the succeeded task, when it sent one. */
  usageTokens?: number;
  /** `usageTokens` × the list rate, USD, unrounded. Absent when the cost is unknown. */
  costUsd?: number;
  /** Why `costUsd` is absent on a completed scene; repeated as an issue on every poll. */
  costUnknownReason?: string;
}

export interface ModelArkNativeJobState {
  externalJobId: string;
  routeId: typeof SEEDANCE_MODELARK_ROUTE_ID;
  model: ModelArkModelId;
  outputDir: string;
  createdAt: string;
  scenes: ModelArkJobSceneState[];
  /** How `costUsd` is derived; the acceptance audit prints it. Absent on a file written before cost accounting. */
  costSource?: typeof MODELARK_COST_SOURCE;
  /**
   * The job's list-price total, USD, once every completed scene has a cost —
   * on the file's top level because the acceptance audit reads it there
   * (as native-veo writes its own).
   */
  actualCost?: { currency: 'USD'; amount: number };
}

export const MODELARK_COST_SOURCE = 'vendor-usage-tokens-at-list-price' as const;

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;
}

/** `.env.local` sits UNDER the process environment, as on every other native route. */
export async function loadModelArkWorkspaceEnv(workspaceRoot: string, env: NodeJS.ProcessEnv): Promise<NodeJS.ProcessEnv> {
  const envLocalPath = join(workspaceRoot, '.env.local');
  if (!existsSync(envLocalPath)) return env;
  const raw = await readFile(envLocalPath, 'utf-8');
  return { ...readDotEnvLike(raw), ...env };
}

/**
 * True when a chained seedance-modelark scene should keep the previous clip as
 * a VIDEO (extend mode) instead of reducing it to its last frame. Read from the
 * same merged env as the transport (.env.local under the process env), so the
 * payload build and the submit agree; an invalid value keeps the default here
 * and is refused, loudly, by the planner.
 */
export async function modelArkChainKeepsVideo(workspaceRoot: string, env: NodeJS.ProcessEnv): Promise<boolean> {
  const merged = await loadModelArkWorkspaceEnv(workspaceRoot, env);
  return merged[MODELARK_CHAIN_MODE_ENV]?.trim() === 'extend';
}

function clientContext(env: NodeJS.ProcessEnv, options: ModelArkTransportOptions): ModelArkClientContext {
  const fetchImpl = options.fetchImpl ?? (fetch as unknown as ModelArkFetchLike);
  return { fetchImpl, baseUrl: modelArkBaseUrl(env), apiKey: resolveModelArkApiKey(env), ...(options.retry ? { retry: options.retry } : {}) };
}

/** A vendor `error.message` as text, whatever shape the vendor sent. */
function vendorText(value: unknown, fallback: string): string {
  if (typeof value === 'string' && value) return value;
  if (value === undefined || value === null) return fallback;
  try {
    return JSON.stringify(value);
  } catch {
    return String(value);
  }
}

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

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

async function writeJobState(state: ModelArkNativeJobState): Promise<void> {
  await mkdir(jobStateDir(state.outputDir), { recursive: true });
  await writeTextFileAtomic(modelArkJobStatePath(state.outputDir, state.externalJobId), `${JSON.stringify(state, null, 2)}\n`);
}

const SCENE_STATUSES: ReadonlySet<string> = new Set<ModelArkJobSceneState['status']>(['submitting', 'submit-unknown', 'submitted', 'completed', 'failed']);
const SCENE_TEXT_FIELDS = ['error', 'intentAt', 'submitError', 'lastFrameUrl', 'costUnknownReason', 'downloadedAt'] as const;

function isPlainRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === 'object' && value !== null && !Array.isArray(value);
}

function isNonNegativeNumber(value: unknown): boolean {
  return typeof value === 'number' && Number.isFinite(value) && value >= 0;
}

/**
 * Every way `value` is not a job-state file this transport can act on, as
 * sentences (empty when it is one). Only this module writes the file, so a
 * problem means a bug or a hand edit, and acting on it anyway would poll a
 * task id that is not a string, download to a path that is not one, sum a
 * bill that is not a number, or write the job back under another id.
 *
 * It checks each field's type, and beyond that only what the poll relies on:
 * the file names its own job, the route and scene statuses are known values,
 * and each scene index is used once. A refusal strands whatever task the file
 * holds, so the states the poll handles on purpose stay valid (a lost create
 * with no task id, an intent time that does not parse), every field later
 * versions added is optional, because older files lack them (#610 cost, #611
 * intent), and values a later table edit could retire (the model, the
 * resolution) are only required to be text: the poll prices an unknown one as
 * "cost unknown". Fields this version does not know are ignored. Submit runs
 * the same check on the file it is about to write, before any upload or create.
 */
export function modelArkJobStateProblems(value: unknown, externalJobId: string): string[] {
  if (!isPlainRecord(value)) return ['the file does not hold a JSON object'];
  const problems: string[] = [];
  const shown = (field: unknown): string => (field === undefined ? 'missing'
    : typeof field === 'number' && !Number.isFinite(field) ? String(field) : JSON.stringify(field));
  if (value.externalJobId !== externalJobId) problems.push(`externalJobId is ${shown(value.externalJobId)}, but the file is named for ${externalJobId}`);
  if (value.routeId !== SEEDANCE_MODELARK_ROUTE_ID) problems.push(`routeId is ${shown(value.routeId)}, not ${SEEDANCE_MODELARK_ROUTE_ID}`);
  if (typeof value.model !== 'string' || !value.model) problems.push(`model is ${shown(value.model)}, not a model id`);
  if (typeof value.outputDir !== 'string' || !value.outputDir) problems.push(`outputDir is ${shown(value.outputDir)}, not a path`);
  if (typeof value.createdAt !== 'string') problems.push(`createdAt is ${shown(value.createdAt)}, not a string`);
  if (value.costSource !== undefined && value.costSource !== MODELARK_COST_SOURCE) problems.push(`costSource is ${shown(value.costSource)}, not ${MODELARK_COST_SOURCE}`);
  if (value.actualCost !== undefined
    && !(isPlainRecord(value.actualCost) && value.actualCost.currency === 'USD' && isNonNegativeNumber(value.actualCost.amount))) {
    problems.push(`actualCost is ${shown(value.actualCost)}, not { currency: "USD", amount: a number >= 0 }`);
  }
  if (!Array.isArray(value.scenes)) {
    problems.push(`scenes is ${shown(value.scenes)}, not a list`);
    return problems;
  }
  const seen = new Set<number>();
  value.scenes.forEach((scene: unknown, position: number) => {
    const at = `scenes[${position}]`;
    if (!isPlainRecord(scene)) {
      problems.push(`${at} is ${shown(scene)}, not an object`);
      return;
    }
    if (!Number.isInteger(scene.sceneIndex) || (scene.sceneIndex as number) < 0) {
      problems.push(`${at}.sceneIndex is ${shown(scene.sceneIndex)}, not a whole number >= 0`);
    } else if (seen.has(scene.sceneIndex as number)) {
      problems.push(`${at}.sceneIndex ${scene.sceneIndex} appears twice`);
    } else {
      seen.add(scene.sceneIndex as number);
    }
    if (typeof scene.prompt !== 'string') problems.push(`${at}.prompt is ${shown(scene.prompt)}, not a string`);
    if (typeof scene.outputPath !== 'string' || !scene.outputPath) problems.push(`${at}.outputPath is ${shown(scene.outputPath)}, not a path`);
    if (typeof scene.status !== 'string' || !SCENE_STATUSES.has(scene.status)) {
      problems.push(`${at}.status is ${shown(scene.status)}, not one of ${[...SCENE_STATUSES].join(', ')}`);
    }
    if (scene.taskId !== undefined && (typeof scene.taskId !== 'string' || !scene.taskId.trim())) problems.push(`${at}.taskId is ${shown(scene.taskId)}, not a task id`);
    for (const field of SCENE_TEXT_FIELDS) {
      if (scene[field] !== undefined && typeof scene[field] !== 'string') problems.push(`${at}.${field} is ${shown(scene[field])}, not a string`);
    }
    if (scene.resolution !== undefined && (typeof scene.resolution !== 'string' || !scene.resolution)) problems.push(`${at}.resolution is ${shown(scene.resolution)}, not a resolution`);
    if (scene.videoInput !== undefined && typeof scene.videoInput !== 'boolean') problems.push(`${at}.videoInput is ${shown(scene.videoInput)}, not true or false`);
    for (const field of ['usageTokens', 'costUsd'] as const) {
      if (scene[field] !== undefined && !isNonNegativeNumber(scene[field])) problems.push(`${at}.${field} is ${shown(scene[field])}, not a number >= 0`);
    }
  });
  return problems;
}

export async function readModelArkJobState(outputDir: string, externalJobId: string): Promise<ModelArkNativeJobState> {
  const path = modelArkJobStatePath(outputDir, externalJobId);
  if (!existsSync(path)) throw new Error(`seedance-modelark job state not found for ${externalJobId}.`);
  const raw = await readFile(path, 'utf-8');
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch (error) {
    throw new Error(`seedance-modelark job state for ${externalJobId} is corrupt (invalid JSON at ${path}): ${error instanceof Error ? error.message : String(error)}`);
  }
  const problems = modelArkJobStateProblems(parsed, externalJobId);
  if (problems.length > 0) {
    throw new Error(
      `seedance-modelark job state for ${externalJobId} at ${path} cannot be acted on, so nothing was sent to ModelArk and the file was not changed: ${problems.join('; ')}. ` +
      'Only vclaw writes this file: restore it, or correct the fields named. Any task it started is listed in the ModelArk console.',
    );
  }
  return parsed as ModelArkNativeJobState;
}

async function downloadToFile(fetchImpl: ModelArkFetchLike, url: string, outputPath: string): Promise<void> {
  const response = await fetchImpl(url);
  if (!response.ok) {
    throw new Error(`seedance-modelark download failed with 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) {
    await unlink(tmpPath).catch(() => {});
    throw error;
  }
}

export interface ModelArkTransportOptions {
  env?: NodeJS.ProcessEnv;
  fetchImpl?: ModelArkFetchLike;
  referenceDeps?: ModelArkReferenceDeps;
  /** Retry policy for GET / DELETE; tests inject `{ retries: 0 }` so a 5xx does not sleep. */
  retry?: ModelArkClientContext['retry'];
  /** The clock; tests inject one so a lost-create window can be walked. */
  now?: () => Date;
}

/**
 * How long after the intent a task created by a lost create can appear in the
 * vendor's list. The create is one POST with the vendor's own timeout in front
 * of it; two minutes covers a slow accept, and a window this narrow is what
 * keeps a bind by creation time honest.
 */
export const MODELARK_LOST_CREATE_WINDOW_SECONDS = 120;
/**
 * Clock skew allowed between this machine and the vendor's `created_at`,
 * applied BEFORE the intent. Generous on purpose: a task the vendor stamped
 * earlier than our intent (their clock behind ours) missed by a tight bound
 * would be declared "never created", the one wrong answer that costs money.
 */
export const MODELARK_LOST_CREATE_SKEW_SECONDS = 60;
/** The vendor's maximum list page; a full page is the signal the window may lie beyond it. */
export const MODELARK_LIST_PAGE_SIZE = 500;


function describeCreateFailure(sceneIndex: number, error: unknown): Error {
  if (error instanceof ModelArkHttpError) {
    const kind = classifyModelArkError(error.code, error.vendorMessage);
    const remedy = kind === 'content-violation'
      ? ` ${suggestRecovery(error.vendorMessage ?? error.rawBody)}`
      : kind === 'task-constraint'
        ? ' The request mixed task types or broke a ratio/duration constraint the vendor enforces (an extension whose prompt did not read as one comes back as TaskTypeMismatch); see the scene plan in a dry run.'
        : '';
    return new Error(`seedance-modelark scene ${sceneIndex}: ${error.message}${remedy}`);
  }
  return new Error(`seedance-modelark scene ${sceneIndex}: ${error instanceof Error ? error.message : String(error)}`);
}

/** A scene as it goes on file before its create: the intent. */
function intentScene(plan: ModelArkTaskPlan, outputDir: string, intentAt: string): ModelArkJobSceneState {
  return {
    sceneIndex: plan.sceneIndex,
    prompt: plan.prompt,
    outputPath: join(outputDir, `scene-${plan.sceneIndex}.mp4`),
    status: 'submitting',
    intentAt,
    resolution: plan.resolution,
    videoInput: plan.refs.some((ref) => ref.kind === 'video'),
  };
}

export async function submitSeedanceModelArkNative(
  payload: VideoExecutionPayload,
  options: ModelArkTransportOptions = {},
): Promise<{ externalJobId: string; rawResult: unknown }> {
  const env = await loadModelArkWorkspaceEnv(payload.workspaceRoot, options.env ?? process.env);
  const ctx = clientContext(env, options);
  const model = resolveModelArkModel(env);
  const limits = MODELARK_MODELS[model];
  const referenceDeps = options.referenceDeps ?? defaultModelArkReferenceDeps(env);
  const externalJobId = `seedance-modelark-${Date.now()}`;
  const issues: string[] = [];

  // Preflight the WHOLE payload before the first paid create, in two passes:
  // plan every scene (pure, so a refusal costs nothing and uploads nothing),
  // then materialise every scene's references (which may upload). A refusal
  // in either pass leaves no task billed.
  const chainOptions = modelArkChainOptions(env);
  const plans = payload.tasks.map((task) => planModelArkTask(task, payload.executionProfile, model, chainOptions));
  for (const plan of plans) {
    issues.push(...plan.issues);
    for (const warning of preValidatePrompt(plan.prompt)) {
      if (warning.level === 'HIGH' || warning.level === 'MEDIUM') {
        issues.push(`scene ${plan.sceneIndex}: content-filter ${warning.level} risk: ${warning.reason} (match: ${warning.match})`);
      }
    }
  }
  const state: ModelArkNativeJobState = {
    externalJobId,
    routeId: SEEDANCE_MODELARK_ROUTE_ID,
    model,
    outputDir: payload.outputDir,
    createdAt: new Date().toISOString(),
    scenes: [],
    costSource: MODELARK_COST_SOURCE,
  };
  // The file this job writes must be one the poll can read back: a paid task
  // behind a file the reader refuses is stranded. Nothing upstream stops a
  // hand-edited storyboard repeating or dropping a scene index, so the file
  // with every scene's intent is checked here, before any upload or create.
  const fileProblems = modelArkJobStateProblems({ ...state, scenes: plans.map((plan) => intentScene(plan, payload.outputDir, state.createdAt)) }, externalJobId);
  if (fileProblems.length > 0) {
    const indexHint = fileProblems.some((problem) => problem.includes('.sceneIndex'))
      ? ' Scene indices come from the storyboard; each must be a whole number >= 0, used once.'
      : '';
    throw new Error(`seedance-modelark refused the payload before any upload or create, so nothing was billed: the job file it would write could not be read back (${fileProblems.join('; ')}).${indexHint}`);
  }
  const prepared: Array<{ plan: ModelArkTaskPlan; urls: string[] }> = [];
  for (const plan of plans) {
    const materialized = await materializeModelArkReferences(plan.sceneIndex, plan.refs, limits, referenceDeps);
    prepared.push({ plan, urls: materialized.map((ref) => ref.url) });
  }

  const rawResults: unknown[] = [];
  const now = options.now ?? (() => new Date());
  for (const { plan, urls } of prepared) {
    // The INTENT goes on file before the POST, so a create whose answer is
    // lost still leaves a scene the next poll can look for; a process that
    // dies here leaves it as `submitting`, which the poll treats the same way.
    const scene = intentScene(plan, payload.outputDir, now().toISOString());
    state.scenes.push(scene);
    await writeJobState(state);
    let created: { id: string; raw: unknown };
    try {
      created = await modelArkCreateTask(ctx, buildModelArkCreateBody(plan, urls));
    } catch (error) {
      const described = describeCreateFailure(plan.sceneIndex, error);
      if (isAmbiguousCreateError(error)) {
        // The vendor may have accepted it. Nothing is re-submitted: the scene
        // is recorded as ambiguous, the scenes not yet attempted are recorded
        // as such, and the job is RETURNED rather than thrown — a throw would
        // leave the run `blocked` with no job id, so nothing would ever poll
        // this file and the resolver below would have no caller.
        scene.status = 'submit-unknown';
        scene.submitError = described.message;
        const notice = `${described.message} — the create's answer was lost, so the task MAY exist and be billing. Nothing was re-submitted; ` +
          `\`vclaw video execute-status\` looks for it in ModelArk's list by model and creation time and binds it when exactly one is there.`;
        issues.push(notice);
        for (const later of prepared.slice(prepared.findIndex((entry) => entry.plan === plan) + 1)) {
          state.scenes.push({
            sceneIndex: later.plan.sceneIndex,
            prompt: later.plan.prompt,
            outputPath: join(payload.outputDir, `scene-${later.plan.sceneIndex}.mp4`),
            status: 'failed',
            error: `seedance-modelark scene ${later.plan.sceneIndex} was not attempted: scene ${plan.sceneIndex}'s create answer was lost first. Nothing billed for this scene.`,
            resolution: later.plan.resolution,
            videoInput: later.plan.refs.some((ref) => ref.kind === 'video'),
          });
        }
        await writeJobState(state);
        process.stderr.write(`[seedance-modelark] ${notice}\n`);
        break;
      }
      // A 4xx is the vendor refusing: nothing was created for THIS scene, and
      // nothing billed for it.
      scene.status = 'failed';
      scene.error = described.message;
      // But a refusal on scene N does not un-bill scenes 0..N-1. Throwing then
      // would leave the run `blocked` with no job id — nothing would poll this
      // file and nothing could abandon the tasks already running — so once an
      // earlier scene is in flight the job is RETURNED instead, with the
      // refusal named and the rest recorded as not attempted (ADR 0008). With
      // nothing in flight the throw stands: the caller must see the rejection
      // (`render-scenes`' ladder escalates to the next route on one).
      const billingAlready = state.scenes.some((other) => other !== scene && other.status === 'submitted' && other.taskId);
      if (!billingAlready) {
        await writeJobState(state);
        throw described;
      }
      issues.push(`${described.message} — the scenes after it were not attempted; the scenes already submitted keep running and are still billing.`);
      for (const later of prepared.slice(prepared.findIndex((entry) => entry.plan === plan) + 1)) {
        state.scenes.push({
          sceneIndex: later.plan.sceneIndex,
          prompt: later.plan.prompt,
          outputPath: join(payload.outputDir, `scene-${later.plan.sceneIndex}.mp4`),
          status: 'failed',
          error: `seedance-modelark scene ${later.plan.sceneIndex} was not attempted: scene ${plan.sceneIndex} was refused first. Nothing billed for this scene.`,
          resolution: later.plan.resolution,
          videoInput: later.plan.refs.some((ref) => ref.kind === 'video'),
        });
      }
      await writeJobState(state);
      process.stderr.write(`[seedance-modelark] ${described.message}\n`);
      break;
    }
    rawResults.push(created.raw);
    scene.taskId = created.id;
    scene.status = 'submitted';
    // After EVERY create, not once at the end: the task is already billed.
    await writeJobState(state);
  }

  return {
    externalJobId,
    rawResult: {
      externalJobId,
      model,
      submittedScenes: state.scenes.map((scene) => ({ sceneIndex: scene.sceneIndex, taskId: scene.taskId ?? null, status: scene.status })),
      ambiguousScenes: state.scenes.filter((scene) => scene.status === 'submit-unknown').map((scene) => scene.sceneIndex),
      issues,
      responses: rawResults,
    },
  };
}

/**
 * Every task id any job in this output dir already owns, so a lost create is
 * never bound to a sibling job's task. Never throws: an unreadable directory
 * means only this job's own ids are known, which `onIssue` hears about, rather
 * than one scene's lookup aborting every other scene's poll. Not covered:
 * another project on the same key within the same two minutes; its job files
 * live elsewhere, and the docs say so.
 */
async function taskIdsOnFile(outputDir: string, state: ModelArkNativeJobState, onIssue: (issue: string) => void): Promise<Set<string>> {
  const ids = new Set<string>();
  for (const scene of state.scenes) if (typeof scene.taskId === 'string') ids.add(scene.taskId);
  const dir = jobStateDir(outputDir);
  let files: string[] = [];
  try {
    files = existsSync(dir) ? await readdir(dir) : [];
  } catch (error) {
    onIssue(`the job directory could not be listed (${error instanceof Error ? error.message : String(error)}); a lost create is checked only against this job's own tasks.`);
  }
  for (const file of files) {
    if (!file.endsWith('.json')) continue;
    try {
      const other = JSON.parse(await readFile(join(dir, file), 'utf-8')) as { scenes?: Array<{ taskId?: unknown }> };
      for (const scene of other.scenes ?? []) if (typeof scene.taskId === 'string') ids.add(scene.taskId);
    } catch {
      // an unreadable sibling file cannot lend an id; nothing to exclude
    }
  }
  return ids;
}

/** A lost create resolved to `taskId`: from here the poll reads it as an ordinary submitted scene. */
function bindLostCreate(scene: ModelArkJobSceneState, taskId: string): void {
  scene.taskId = taskId;
  scene.status = 'submitted';
  delete scene.submitError;
}

export async function pollSeedanceModelArkNative(
  input: { outputDir: string; externalJobId: string; workspaceRoot: string },
  options: ModelArkTransportOptions = {},
): Promise<VideoExecutionPollResult> {
  const env = await loadModelArkWorkspaceEnv(input.workspaceRoot, options.env ?? process.env);
  const ctx = clientContext(env, options);
  const state = await readModelArkJobState(input.outputDir, input.externalJobId);

  const outputs: VideoExecutionPollResult['outputs'] = [];
  const issues: string[] = [];
  const rawResults: unknown[] = [];
  let anyPending = false;
  let anyFailed = false;
  // A lost create still unresolved: the one case where the job must stay
  // pending even beside a failed sibling (an ordinary failure beside a queued
  // scene keeps the house rule, failed first, as on every other route).
  let anyUnresolvedAmbiguity = false;

  const pushOutput = (scene: ModelArkJobSceneState): void => {
    outputs.push({ id: `generated-scene-${scene.sceneIndex}`, kind: 'video', path: scene.outputPath, sceneIndex: scene.sceneIndex, backend: SEEDANCE_MODELARK_ROUTE_ID });
  };
  const now = options.now ?? (() => new Date());
  // The vendor's task list, fetched once per poll and only when a scene needs it.
  let listed: Awaited<ReturnType<typeof modelArkListTasks>> | null = null;
  // Includes ids THIS poll bound a moment ago, which are only in memory (on
  // `state`) until the finally writes the file.
  const boundTaskIds = () => taskIdsOnFile(input.outputDir, state, (issue) => issues.push(issue));

  // One scene's poll error (a 404 after the vendor purged its record, a key
  // rotated mid-run, a 5xx that outlives the retries) must not hide another
  // scene's progress: each scene is isolated, the error becomes an issue and
  // the scene stays pending, and the state is written whatever happens so a
  // clip already downloaded is recorded as completed.
  try {
    for (const scene of state.scenes) {
      if (scene.status === 'completed') {
        // A scene completed before cost accounting existed never transitioned
        // through the branch below, so it has neither a cost nor a reason: give
        // it the reason now, so this job does not go quiet about its cost.
        if (!scene.costUnknownReason && typeof scene.costUsd !== 'number') {
          scene.costUnknownReason = `ModelArk task ${scene.taskId}: this job predates cost accounting, so scene ${scene.sceneIndex}'s cost is unknown; the job reports no actualCost.`;
        }
        if (scene.costUnknownReason) issues.push(scene.costUnknownReason);
        pushOutput(scene);
        continue;
      }
      if (scene.status === 'failed') {
        if (scene.error) issues.push(scene.error);
        anyFailed = true;
        continue;
      }
      if (scene.status === 'submitting' || scene.status === 'submit-unknown' || !scene.taskId) {
        // A create whose answer was lost. Resolve it from the vendor's list by
        // model and creation window; bind only when exactly one unbound task
        // is there, and never re-submit. Pending only on the exits that leave
        // it unresolved: a settled "never accepted" must not keep the job open.
        let stillPending = true;
        const intentMs = scene.intentAt ? Date.parse(scene.intentAt) : Number.NaN;
        if (!Number.isFinite(intentMs)) {
          issues.push(`scene ${scene.sceneIndex}: its create's answer was lost and no intent time is on file, so it cannot be looked up. \`vclaw video execute-bind --task <id>\` binds the one you name; \`vclaw video execute-abandon\` stops waiting.`);
          anyPending = true;
          anyUnresolvedAmbiguity = true;
          continue;
        }
        const windowStart = intentMs / 1000 - MODELARK_LOST_CREATE_SKEW_SECONDS;
        const windowEnd = intentMs / 1000 + MODELARK_LOST_CREATE_WINDOW_SECONDS;
        try {
          // The vendor's maximum page: a busy key must not push the window off page 1.
          listed ??= await modelArkListTasks(ctx, { model: state.model, pageSize: MODELARK_LIST_PAGE_SIZE });
        } catch (error) {
          issues.push(`scene ${scene.sceneIndex}: looking for its lost create failed (${error instanceof Error ? error.message : String(error)}); it stays ambiguous until a poll can read ModelArk's task list.`);
          anyPending = true;
          anyUnresolvedAmbiguity = true;
          continue;
        }
        const bound = await boundTaskIds();
        const candidates = listed.filter((task) => typeof task.created_at === 'number' && task.created_at >= windowStart && task.created_at <= windowEnd && !bound.has(task.id));
        // "Never created" may only be said when the page provably reached
        // back past the window: a short page, or an oldest row older than the
        // window start. Otherwise the task may sit on a page never read.
        const oldestListed = listed.reduce((min, task) => (typeof task.created_at === 'number' && task.created_at < min ? task.created_at : min), Number.POSITIVE_INFINITY);
        const listCoversWindow = listed.length < MODELARK_LIST_PAGE_SIZE || oldestListed < windowStart;
        if (candidates.length === 1 && !candidates[0].id.trim()) {
          // A task with no id cannot be read or cancelled, and a blank id on
          // file is one the reader refuses. It is still a task, so the scene
          // stays ambiguous rather than "never accepted".
          issues.push(`scene ${scene.sceneIndex}: its create's answer was lost and the one task ModelArk lists in its window has no id, so it cannot be bound; it stays ambiguous — nothing is re-submitted. Check the ModelArk console. \`vclaw video execute-bind --task <id>\` binds the one you name; \`vclaw video execute-abandon\` stops waiting.`);
        } else if (candidates.length === 1 && existsSync(scene.outputPath)) {
          // Binding here would download this task's clip over a file another
          // run wrote to the same `scene-N.mp4`. The operator decides.
          issues.push(`scene ${scene.sceneIndex}: its create's answer was lost and ModelArk task ${candidates[0].id} is the one task in its window, but ${scene.outputPath} already holds a file this job did not download; it stays ambiguous. Move or delete that file, then \`vclaw video execute-bind --task ${candidates[0].id}\`; \`vclaw video execute-abandon\` stops waiting instead.`);
        } else if (candidates.length === 1) {
          bindLostCreate(scene, candidates[0].id);
          issues.push(`scene ${scene.sceneIndex}: bound to ModelArk task ${candidates[0].id}, the one task on this model created within ${MODELARK_LOST_CREATE_WINDOW_SECONDS} s of the lost create; the next poll reads it. If that is wrong, \`vclaw video execute-abandon\` stops waiting on it.`);
        } else if (candidates.length === 0) {
          const nowSeconds = now().getTime() / 1000;
          if (!listCoversWindow) {
            issues.push(`scene ${scene.sceneIndex}: its create's answer was lost and ModelArk's newest ${listed.length} tasks on ${state.model} are all newer than its window, so the window was not searched; it stays ambiguous. If the key is that busy, check the ModelArk console for a task created near ${scene.intentAt}. \`vclaw video execute-bind --task <id>\` binds the one you name; \`vclaw video execute-abandon\` stops waiting.`);
          } else if (nowSeconds > windowEnd + MODELARK_LOST_CREATE_SKEW_SECONDS) {
            scene.status = 'failed';
            scene.error = `scene ${scene.sceneIndex}: the create was never accepted — no task on ${state.model} was created within ${MODELARK_LOST_CREATE_WINDOW_SECONDS} s of ${scene.intentAt} (${scene.submitError ?? 'answer lost'}); nothing was billed.`;
            issues.push(scene.error);
            anyFailed = true;
            stillPending = false;
          } else {
            issues.push(`scene ${scene.sceneIndex}: its create's answer was lost; no task has appeared yet, and the window closes at ${new Date((windowEnd + MODELARK_LOST_CREATE_SKEW_SECONDS) * 1000).toISOString()}.`);
          }
        } else {
          issues.push(`scene ${scene.sceneIndex}: its create's answer was lost and ${candidates.length} unbound tasks on ${state.model} were created in its window (${candidates.map((task) => task.id).join(', ')}); it stays ambiguous — nothing is re-submitted. Check the ModelArk console. \`vclaw video execute-bind --task <id>\` binds the one you name; \`vclaw video execute-abandon\` stops waiting.`);
        }
        if (stillPending) {
          anyPending = true;
          // A bind is resolved: from the next poll it is an ordinary submitted
          // scene. Everything else here is still ambiguous.
          if (scene.status !== 'submitted') anyUnresolvedAmbiguity = true;
        }
        continue;
      }
      let task;
      try {
        task = await modelArkGetTask(ctx, scene.taskId);
      } catch (error) {
        issues.push(`ModelArk task ${scene.taskId}: poll failed (${error instanceof Error ? error.message : String(error)}); scene ${scene.sceneIndex} stays pending. If it never recovers, use \`vclaw video execute-abandon\`.`);
        anyPending = true;
        continue;
      }
      rawResults.push(task);
      const status = typeof task.status === 'string' ? task.status : '';
      if (status === 'succeeded') {
        const videoUrl = task.content?.video_url;
        if (typeof videoUrl !== 'string' || !videoUrl) {
          // A "succeeded" task with no video is a failure, not a still to save as an mp4.
          scene.status = 'failed';
          scene.error = `ModelArk task ${scene.taskId} succeeded without a video_url.`;
          issues.push(scene.error);
          anyFailed = true;
          continue;
        }
        try {
          // Only a file THIS scene downloaded may be kept: `scene-N.mp4` is
          // shared with every other job in this output dir, so an unclaimed
          // file there is another run's clip, and completing on it would
          // deliver the wrong video and bill this task for it.
          if (!scene.downloadedAt || !existsSync(scene.outputPath)) {
            await downloadToFile(ctx.fetchImpl, videoUrl, scene.outputPath);
            scene.downloadedAt = now().toISOString();
          }
        } catch (error) {
          issues.push(`ModelArk task ${scene.taskId}: the clip is ready but its download failed (${error instanceof Error ? error.message : String(error)}); scene ${scene.sceneIndex} stays pending and the next poll retries the download.`);
          anyPending = true;
          continue;
        }
        if (typeof task.content?.last_frame_url === 'string') scene.lastFrameUrl = task.content.last_frame_url;
        // The bill, from the vendor's own usage. A completed scene keeps these
        // in state, so a later poll sums a job's cost without re-fetching.
        // A job-state file written before cost accounting has no
        // resolution/videoInput; a vendor reply may carry no usage. Either way
        // the cost is unknown — recorded on the scene so every later poll says
        // so too — never a crash mid-poll and never a guess.
        const cost = modelArkCostUsd(state.model, scene.resolution, scene.videoInput, task.usage);
        if (cost) {
          scene.usageTokens = cost.tokens;
          scene.costUsd = cost.usd;
        } else {
          const noPrice = !Object.hasOwn(MODELARK_MODELS, state.model)
            || (scene.resolution !== undefined && !MODELARK_MODELS[state.model].usdPerMillionTokens[scene.resolution]);
          scene.costUnknownReason = scene.resolution === undefined || typeof scene.videoInput !== 'boolean'
            ? `ModelArk task ${scene.taskId}: this job predates cost accounting (no resolution/videoInput on file), so scene ${scene.sceneIndex}'s cost is unknown; the job reports no actualCost.`
            : noPrice
              ? `ModelArk task ${scene.taskId}: no list price is on file for ${state.model} at ${scene.resolution} (a model or resolution this version does not price), so scene ${scene.sceneIndex}'s cost is unknown; the job reports no actualCost.`
              : `ModelArk task ${scene.taskId}: the vendor returned no usable usage, so scene ${scene.sceneIndex}'s cost is unknown; the job reports no actualCost.`;
          issues.push(scene.costUnknownReason);
        }
        scene.status = 'completed';
        pushOutput(scene);
      } else if (status === 'failed' || status === 'expired' || status === 'cancelled') {
        const code = vendorText(task.error?.code, status);
        const message = vendorText(task.error?.message, `task ${status}`);
        scene.status = 'failed';
        scene.error = `ModelArk task ${scene.taskId} ${status} (${code}): ${message}`;
        if (classifyModelArkError(code, message) === 'content-violation') {
          scene.error += ` ${suggestRecovery(message)}`;
        }
        issues.push(scene.error);
        anyFailed = true;
      } else {
        if (status !== 'queued' && status !== 'running') {
          issues.push(`ModelArk task ${scene.taskId}: unexpected status ${JSON.stringify(status)} (treating as pending): ${safeErrorBody(JSON.stringify(task))}`);
        }
        anyPending = true;
      }
    }
    // actualCost is stated only when EVERY completed scene has a cost: a
    // partial sum would read as the whole bill. Scene costs are unrounded;
    // the total is rounded once here. Written onto the state so the file
    // carries it too.
    const completedScenes = state.scenes.filter((scene) => scene.status === 'completed');
    const costed = completedScenes.length > 0 && completedScenes.every((scene) => typeof scene.costUsd === 'number');
    const totalUsd = completedScenes.reduce((sum, scene) => sum + (scene.costUsd ?? 0), 0);
    if (costed) state.actualCost = { currency: 'USD', amount: Math.round(totalUsd * 10_000) / 10_000 };
    else delete state.actualCost;
  } finally {
    await writeJobState(state);
  }
  const actualCost = state.actualCost;
  // Failed outranks pending, as on every route — except while a lost create is
  // still unresolved: that job stays pending (its not-yet-attempted scenes are
  // on file as failed) until the ambiguity is settled or the operator abandons
  // it. Nothing times an unresolvable ambiguity out; the docs say so.
  return {
    status: anyUnresolvedAmbiguity ? 'pending' : anyFailed ? 'failed' : anyPending ? 'pending' : 'completed',
    externalJobId: input.externalJobId,
    outputs,
    issues,
    rawResult: { polls: rawResults, costSource: MODELARK_COST_SOURCE, ...(actualCost ? { actualCost } : {}) },
    ...(actualCost ? { actualCost } : {}),
  };
}

/**
 * Cancel what CAN be cancelled: a queued task. ModelArk cannot stop a running
 * task, so a scene the vendor refuses to cancel keeps its `submitted` status
 * (it is still billing and its clip will still arrive), and the answer says so
 * rather than reporting it stopped. Finished scenes are never touched.
 */
export async function cancelSeedanceModelArkNative(
  input: { outputDir: string; externalJobId: string; workspaceRoot: string },
  options: ModelArkTransportOptions = {},
): Promise<VideoExecutionCancelResult> {
  const env = await loadModelArkWorkspaceEnv(input.workspaceRoot, options.env ?? process.env);
  const ctx = clientContext(env, options);
  const state = await readModelArkJobState(input.outputDir, input.externalJobId);

  const issues: string[] = [];
  const rawResults: unknown[] = [];
  let cancelled = 0;
  for (const scene of state.scenes) {
    if (scene.status === 'submitting' || scene.status === 'submit-unknown' || (scene.status === 'submitted' && !scene.taskId)) {
      issues.push(`scene ${scene.sceneIndex}: its create's answer was lost and it has no task id yet, so there is nothing to cancel; run \`vclaw video execute-status\` to find the task, \`vclaw video execute-bind --task <id>\` to name it, or \`vclaw video execute-abandon\` to stop waiting.`);
      continue;
    }
    if (scene.status !== 'submitted' || !scene.taskId) continue;
    const result = await modelArkDeleteTask(ctx, scene.taskId);
    rawResults.push({ taskId: scene.taskId, ...result });
    if (result.ok) {
      scene.status = 'failed';
      scene.error = 'Execution cancelled by operator (ModelArk task cancelled while queued).';
      cancelled += 1;
    } else {
      issues.push(
        `scene ${scene.sceneIndex}: ModelArk refused to cancel task ${scene.taskId} (HTTP ${result.status}): a running task cannot be stopped and is still billing; ` +
        '`vclaw video execute-status` will still collect its clip. To stop waiting for it, use `vclaw video execute-abandon`.',
      );
    }
  }
  if (cancelled > 0) await writeJobState(state);

  return {
    status: cancelled > 0 ? 'cancelled' : 'unsupported',
    externalJobId: input.externalJobId,
    issues,
    rawResult: rawResults,
  };
}

/**
 * Stop WAITING for scenes still in flight without touching the provider: the
 * remote task keeps running and keeps billing, and the recorded reason says so.
 * Mirrors `abandonDreaminaUseApiNativeJob` for `execute-abandon`.
 */
export async function abandonSeedanceModelArkNativeJob(
  input: { outputDir: string; externalJobId: string; dryRun?: boolean },
): Promise<{ abandonedScenes: number[]; untouchedScenes: Array<{ sceneIndex: number; status: string }> }> {
  const state = await readModelArkJobState(input.outputDir, input.externalJobId);
  const abandonedScenes: number[] = [];
  const untouchedScenes: Array<{ sceneIndex: number; status: string }> = [];
  for (const scene of state.scenes) {
    if (scene.status === 'submitted' || scene.status === 'submitting' || scene.status === 'submit-unknown') {
      const which = scene.taskId
        ? `ModelArk task ${scene.taskId}`
        : `a ModelArk task that may or may not exist (its create's answer was lost; look in the ModelArk console for a ${state.model} task created near ${scene.intentAt ?? 'an unknown time'})`;
      scene.status = 'failed';
      scene.error = `seedance-modelark scene ${scene.sceneIndex} was abandoned: nobody is waiting for ${which} any more. It was NOT cancelled; if it exists it keeps running and keeps billing, and its clip will not be collected.`;
      abandonedScenes.push(scene.sceneIndex);
    } else {
      untouchedScenes.push({ sceneIndex: scene.sceneIndex, status: scene.status });
    }
  }
  if (abandonedScenes.length > 0 && !input.dryRun) await writeJobState(state);
  return { abandonedScenes, untouchedScenes };
}

export interface ModelArkBindResult {
  externalJobId: string;
  sceneIndex: number;
  taskId: string;
  /** false = the plan only; nothing was written. */
  applied: boolean;
  /** What ModelArk says about the task, read before anything is written. */
  task: { status: string; model: string; createdAt: string | null; resolution: string | null };
  /** Whether the task was created in the lost create's window; null when either time is unknown. Reported, not enforced: the operator is overriding the window on purpose. */
  createdInsideWindow: boolean | null;
  issues: string[];
}

/**
 * Bind a lost create to the task the operator names — the exit for the cases
 * the poll will not decide: several unbound tasks in the window, a window it
 * could not search, or no intent time to search by. Nothing is submitted: the
 * task is read with one GET, and only a task that exists, is on this job's
 * model and is not already some job's is bound. The scene then polls as an
 * ordinary submitted one. `dryRun` reports the plan and writes nothing.
 */
export async function bindSeedanceModelArkLostCreate(
  input: { outputDir: string; externalJobId: string; workspaceRoot: string; taskId: string; sceneIndex?: number; dryRun?: boolean },
  options: ModelArkTransportOptions = {},
): Promise<ModelArkBindResult> {
  const taskId = input.taskId.trim();
  if (!taskId) throw new Error('seedance-modelark: execute-bind needs the id of a ModelArk task (--task <id>).');
  const state = await readModelArkJobState(input.outputDir, input.externalJobId);
  const lost = state.scenes.filter((scene) => !scene.taskId && scene.status !== 'completed' && scene.status !== 'failed');
  let scene: ModelArkJobSceneState | undefined;
  if (input.sceneIndex !== undefined) {
    scene = state.scenes.find((candidate) => candidate.sceneIndex === input.sceneIndex);
    if (!scene) throw new Error(`seedance-modelark job ${input.externalJobId} has no scene ${input.sceneIndex} (scenes: ${state.scenes.map((s) => s.sceneIndex).join(', ')}).`);
    if (!lost.includes(scene)) {
      throw new Error(`seedance-modelark job ${input.externalJobId} scene ${scene.sceneIndex} is not a lost create waiting for a task: it is ${scene.status}${scene.taskId ? ` on ModelArk task ${scene.taskId}` : ''}. Nothing was bound.`);
    }
  } else if (lost.length === 1) {
    scene = lost[0];
  } else {
    throw new Error(lost.length === 0
      ? `seedance-modelark job ${input.externalJobId} has no lost create waiting for a task; every scene already has one or has ended. Nothing was bound.`
      : `seedance-modelark job ${input.externalJobId} has ${lost.length} lost creates (scenes ${lost.map((s) => s.sceneIndex).join(', ')}); name one with --scene <i>. Nothing was bound.`);
  }

  const issues: string[] = [];
  // The clip would land on a file this job did not write: another run's
  // `scene-N.mp4`. The poll would overwrite it, so the operator decides.
  if (!scene.downloadedAt && existsSync(scene.outputPath)) {
    throw new Error(`seedance-modelark: scene ${scene.sceneIndex}'s output path ${scene.outputPath} already holds a file this job did not download, and the bound task's clip would replace it. Move or delete that file, then bind. Nothing was bound.`);
  }
  const owned = await taskIdsOnFile(input.outputDir, state, (issue) => issues.push(issue));
  if (owned.has(taskId)) {
    throw new Error(`seedance-modelark: ModelArk task ${taskId} already belongs to a job in ${input.outputDir}; binding it to scene ${scene.sceneIndex} too would collect one clip twice. Nothing was bound.`);
  }

  const env = await loadModelArkWorkspaceEnv(input.workspaceRoot, options.env ?? process.env);
  const ctx = clientContext(env, options);
  let task: Awaited<ReturnType<typeof modelArkGetTask>>;
  try {
    task = await modelArkGetTask(ctx, taskId);
  } catch (error) {
    if (error instanceof ModelArkHttpError && error.status === 404) {
      throw new Error(`seedance-modelark: ModelArk has no task ${taskId} on this key (HTTP 404). Nothing was bound.`);
    }
    throw new Error(`seedance-modelark: ModelArk task ${taskId} could not be read, so it was not bound (${error instanceof Error ? error.message : String(error)}).`);
  }
  // ModelArk's task answer carries the model (live-checked 2026-09-22 against
  // task cgt-20260922213730-rmzl9), so an answer without one is not one this
  // bind can check — and an unchecked bind is what collects the wrong clip.
  const model = typeof task.model === 'string' && task.model ? task.model : null;
  if (model === null) {
    throw new Error(`seedance-modelark: ModelArk's answer for task ${taskId} does not name its model, so it cannot be checked against job ${input.externalJobId}'s ${state.model}. Nothing was bound.`);
  }
  if (model !== state.model) {
    throw new Error(`seedance-modelark: ModelArk task ${taskId} is on ${model}, but job ${input.externalJobId} submitted to ${state.model}; it is not this scene's task. Nothing was bound.`);
  }
  // The bill is priced from the scene's resolution, which so far is the one
  // the lost create INTENDED. The task's own is what was rendered and billed —
  // but only a spelling this version prices may replace it: taking an unpriced
  // one (a `720P`, a resolution a later model sells) would turn a bill this
  // version can state into "cost unknown", for good.
  const taskResolution = typeof task.resolution === 'string' && task.resolution ? task.resolution : null;
  const pricedResolution = taskResolution !== null
    && Object.hasOwn(MODELARK_MODELS, state.model)
    && Object.hasOwn(MODELARK_MODELS[state.model].usdPerMillionTokens, taskResolution);
  if (taskResolution !== null && taskResolution !== scene.resolution) {
    issues.push(pricedResolution
      ? `ModelArk task ${taskId} was rendered at ${taskResolution}, not the ${scene.resolution ?? 'unrecorded resolution'} this scene asked for; its bill is priced at ${taskResolution}.`
      : `ModelArk reports task ${taskId} at ${taskResolution}, which this version does not price; the bill stays at the intent's ${scene.resolution ?? 'unrecorded resolution'}.`);
  }

  const createdSeconds = modelArkCreatedAtSeconds(task.created_at);
  const intentMs = scene.intentAt ? Date.parse(scene.intentAt) : Number.NaN;
  let createdInsideWindow: boolean | null = null;
  if (createdSeconds !== undefined && Number.isFinite(intentMs)) {
    createdInsideWindow = createdSeconds >= intentMs / 1000 - MODELARK_LOST_CREATE_SKEW_SECONDS
      && createdSeconds <= intentMs / 1000 + MODELARK_LOST_CREATE_WINDOW_SECONDS;
    if (!createdInsideWindow) {
      issues.push(`ModelArk task ${taskId} was created at ${new Date(createdSeconds * 1000).toISOString()}, outside scene ${scene.sceneIndex}'s window (${MODELARK_LOST_CREATE_SKEW_SECONDS} s before to ${MODELARK_LOST_CREATE_WINDOW_SECONDS} s after its create at ${scene.intentAt}); make sure it is this scene's task.`);
    }
  } else {
    issues.push(`scene ${scene.sceneIndex}'s window could not be checked (${createdSeconds === undefined ? 'ModelArk gave no creation time for the task' : 'no intent time is on file'}).`);
  }

  if (!input.dryRun) {
    bindLostCreate(scene, taskId);
    if (pricedResolution) scene.resolution = taskResolution as ModelArkJobSceneState['resolution'];
    await writeJobState(state);
    // A poll running in another terminal writes back the whole state it read
    // before this bind, which would silently undo it. Read the file again —
    // from the directory the write went to — and say so rather than reporting
    // a bind that is no longer on disk.
    const written = await readModelArkJobState(state.outputDir, input.externalJobId);
    if (written.scenes.find((candidate) => candidate.sceneIndex === scene.sceneIndex)?.taskId !== taskId) {
      throw new Error(`seedance-modelark: scene ${scene.sceneIndex} was bound to ModelArk task ${taskId}, but the job state no longer says so — another vclaw process wrote it at the same time. Nothing else was changed; run the bind again.`);
    }
  }
  return {
    externalJobId: input.externalJobId,
    sceneIndex: scene.sceneIndex,
    taskId,
    applied: !input.dryRun,
    task: {
      status: typeof task.status === 'string' ? task.status : '',
      model,
      createdAt: createdSeconds === undefined ? null : new Date(createdSeconds * 1000).toISOString(),
      resolution: taskResolution,
    },
    createdInsideWindow,
    issues,
  };
}