import { mkdir, readFile, writeFile } from "node:fs/promises";
import { basename } from "node:path";
import { resolve } from "node:path";
import { dbAll, dbGet, dbRun, withTransaction } from "./db.mjs";
import {
  addAudit,
  addUsage,
  hasPermission,
  httpError,
  parseModelRow,
  requireQuota,
  requirePermission,
  requireProjectWritable
} from "./tenant.mjs";
import { productionGraph } from "./production.mjs";
import { dispatchNotificationEvent } from "./notifications.mjs";
import { registerJobArtifacts } from "./media-artifacts.mjs";

const projectRoot = resolve(import.meta.dirname, "..");
const jobStorageRoot = resolve(projectRoot, "storage", "jobs");
const now = () => new Date().toISOString();
const makeId = (prefix) => `${prefix}-${Date.now()}-${Math.random().toString(16).slice(2, 8)}`;

function parseJson(value, fallback) {
  try {
    return JSON.parse(value);
  } catch {
    return fallback;
  }
}

function privateHost(hostname) {
  const host = String(hostname || "").toLowerCase();
  if (["localhost", "127.0.0.1", "::1"].includes(host) || host.endsWith(".local")) return true;
  const octets = host.split(".").map(Number);
  if (octets.length !== 4 || octets.some((value) => !Number.isInteger(value) || value < 0 || value > 255)) return false;
  return octets[0] === 10 || octets[0] === 127 || (octets[0] === 172 && octets[1] >= 16 && octets[1] <= 31) || (octets[0] === 192 && octets[1] === 168);
}

export function endpointInfo(endpoint) {
  let url;
  try {
    url = new URL(String(endpoint || ""));
  } catch {
    throw httpError(400, "endpoint_invalid", "模型连接器地址不是有效 HTTP URL");
  }
  if (!["http:", "https:"].includes(url.protocol)) throw httpError(400, "endpoint_protocol_invalid", "模型连接器只支持 HTTP/HTTPS");
  return { url, local: privateHost(url.hostname) };
}

function resolveAdapterId(context, adapterId, kind = "") {
  if (adapterId !== "owned-model-platform") return adapterId;
  const normalizedKind = String(kind).toLowerCase();
  if (normalizedKind.includes("视频") || normalizedKind.includes("i2v") || normalizedKind.includes("video")) return "owned-i2v";
  if (normalizedKind.includes("asr") || normalizedKind.includes("字幕")) {
    const preferred = dbGet("SELECT id FROM model_connectors WHERE id = 'newapi-audio-production' AND organization_id = ? AND (workspace_id IS NULL OR workspace_id = ?)", [context.organization.id, context.workspace.id]);
    if (preferred) return preferred.id;
  }
  if (normalizedKind.includes("tts") || normalizedKind.includes("配音")) return "local-tts";
  return "owned-image";
}

function modelForContext(context, adapterId, kind = "") {
  const resolvedAdapterId = resolveAdapterId(context, adapterId, kind);
  const row = dbGet(
    `SELECT * FROM model_connectors
     WHERE id = ? AND organization_id = ? AND (workspace_id IS NULL OR workspace_id = ?)`,
    [resolvedAdapterId, context.organization.id, context.workspace.id]
  );
  if (!row) {
    throw httpError(404, "adapter_not_found", "模型连接器不存在或不属于当前工作区", { adapterId: resolvedAdapterId });
  }
  return parseModelRow(row);
}

function jobRow(context, jobId) {
  if (!context.project) throw httpError(400, "project_required", "任务操作必须绑定项目");
  const row = dbGet(
    `SELECT j.*, p.name AS project_name, u.display_name AS creator_name
     FROM generation_jobs j
     JOIN projects p ON p.id = j.project_id
     LEFT JOIN users u ON u.id = j.created_by
     WHERE j.id = ? AND j.organization_id = ? AND j.workspace_id = ? AND j.project_id = ?`,
    [jobId, context.organization.id, context.workspace.id, context.project.id]
  );
  if (!row) throw httpError(404, "job_not_found", "任务不存在或不属于当前项目", { jobId });
  return row;
}

function jobPayload(row) {
  const attempts = dbAll(
    "SELECT id, attempt_number, runner_id, status, error_message, started_at, finished_at, created_at FROM job_attempts WHERE job_id = ? ORDER BY attempt_number DESC",
    [row.id]
  );
  const dependencies = dbAll(
    `SELECT d.job_id, d.depends_on_job_id, d.dependency_type, d.created_at,
            j.kind, j.status, j.output_path
     FROM job_dependencies d
     JOIN generation_jobs j ON j.id = d.depends_on_job_id
     WHERE d.job_id = ?
     ORDER BY d.created_at`,
    [row.id]
  );
  return {
    ...row,
    shotId: row.shot_id || "E01",
    adapter: row.adapter_id,
    costPolicy: row.cost_policy,
    output: row.output_path,
    qa: row.qa_status,
    request: parseJson(row.request_json, {}),
    result: parseJson(row.result_json, {}),
    errorMessage: row.error_message || "",
    startedAt: row.started_at,
    finishedAt: row.finished_at,
    attempts: Number(row.attempts || attempts.length || 0),
    attemptLog: attempts,
    dependencies
  };
}

function dependencyRows(context, dependencyIds) {
  if (!dependencyIds.length) return [];
  const placeholders = dependencyIds.map(() => "?").join(",");
  const rows = dbAll(
    `SELECT id, kind, status, organization_id, workspace_id, project_id
     FROM generation_jobs
     WHERE id IN (${placeholders})`,
    dependencyIds
  );
  if (rows.length !== dependencyIds.length || rows.some((row) => row.organization_id !== context.organization.id || row.workspace_id !== context.workspace.id || row.project_id !== context.project.id)) {
    throw httpError(422, "job_dependency_scope_invalid", "任务前置依赖必须属于同一组织、工作区和项目");
  }
  return rows;
}

function assertNoDependencyCycle(jobId, dependencyIds) {
  const visiting = new Set();
  const visited = new Set();
  function walk(currentId) {
    if (currentId === jobId) throw httpError(422, "job_dependency_cycle", "任务前置依赖不能形成循环");
    if (visited.has(currentId)) return;
    if (visiting.has(currentId)) throw httpError(422, "job_dependency_cycle", "任务前置依赖不能形成循环");
    visiting.add(currentId);
    const parents = dbAll("SELECT depends_on_job_id FROM job_dependencies WHERE job_id = ?", [currentId]);
    for (const parent of parents) walk(parent.depends_on_job_id);
    visiting.delete(currentId);
    visited.add(currentId);
  }
  for (const dependencyId of dependencyIds) walk(dependencyId);
}

function unresolvedDependencies(jobId) {
  return dbAll(
    `SELECT d.depends_on_job_id, j.kind, j.status
     FROM job_dependencies d
     JOIN generation_jobs j ON j.id = d.depends_on_job_id
     WHERE d.job_id = ? AND j.status <> 'completed'
     ORDER BY d.created_at`,
    [jobId]
  );
}

function releaseReadyDependents(jobId) {
  const dependents = dbAll("SELECT DISTINCT job_id FROM job_dependencies WHERE depends_on_job_id = ?", [jobId]);
  for (const dependent of dependents) {
    const unresolved = unresolvedDependencies(dependent.job_id);
    if (unresolved.length) continue;
    const row = dbGet("SELECT j.*, m.status AS adapter_status FROM generation_jobs j LEFT JOIN model_connectors m ON m.id = j.adapter_id WHERE j.id = ?", [dependent.job_id]);
    if (row?.status === "blocked" && row.adapter_status === "ready" && /等待前置任务/.test(row.error_message || "")) {
      const timestamp = now();
      dbRun("UPDATE generation_jobs SET status = 'queued', error_message = '', updated_at = ? WHERE id = ?", [timestamp, dependent.job_id]);
      dbRun("UPDATE job_attempts SET status = 'queued', error_message = NULL WHERE job_id = ? AND attempt_number = (SELECT MAX(attempt_number) FROM job_attempts WHERE job_id = ?)", [dependent.job_id, dependent.job_id]);
    }
  }
}

function shotForJob(context, shotId) {
  if (!shotId) return null;
  const graph = productionGraph(context);
  return graph.shots.find((shot) => shot.id === shotId) || null;
}

function buildContract(context, body, adapter) {
  const shot = shotForJob(context, body.shotId);
  const kind = String(body.kind || "自定义生成任务").trim();
  const blockedTerms = ["split-screen", "comic panel", "collage", "contact sheet", "storyboard", "多格", "拼图", "分屏", "故事板拼图"];
  // Negative prompts intentionally name the forbidden layouts; only inspect positive generation text here.
  const promptText = [shot?.prompt, shot?.videoPrompt, shot?.action, shot?.camera, body.prompt].filter(Boolean).join(" ").toLowerCase();
  const blocked = blockedTerms.filter((term) => promptText.includes(term.toLowerCase()));
  if (blocked.length) throw httpError(422, "single_frame_contract_violation", "请求包含禁止的一图多画面表达", { blocked });
  if (shot && kind.includes("视频") && shot.transitionFromPrevious !== "episode-start" && (!shot.firstFrame || /pending|auto_previous/i.test(shot.firstFrame))) {
    throw httpError(422, "actual_last_frame_required", "连续视频镜头必须先登记上一段实际末帧作为首帧输入", { shotId: shot.id });
  }
  return {
    schema: "ai-drama.job.v1",
    createdAt: now(),
    project: { id: context.project.id, name: context.project.name },
    job: { kind, shotId: body.shotId || null, adapterId: adapter.id, costPolicy: "local-only" },
    shot,
    inputs: body.inputs || {},
    constraints: {
      singleFrameOnly: true,
      imageOutputCount: 1,
      requireActualLastFrame: true,
      localOnly: adapter.costMode === "local",
      blockedTerms
    }
  };
}

async function writeJobRequest(jobId, contract) {
  const directory = resolve(jobStorageRoot, jobId);
  await mkdir(directory, { recursive: true });
  const relative = `storage/jobs/${jobId}/request.json`;
  await writeFile(resolve(projectRoot, relative), `${JSON.stringify(contract, null, 2)}\n`, "utf8");
  return relative;
}

function assertExternalAllowed(context, adapter, body = {}) {
  if (adapter.costMode === "local") return;
  if (!adapter.approvalRequired) return;
  if (!body.approveExternal || !hasPermission(context, "model:manage")) {
    throw httpError(403, "external_connector_requires_approval", "外部或混合连接器必须由具备模型管理权限的用户显式批准后才能执行", { adapterId: adapter.id });
  }
}

export async function createGenerationJob(context, body) {
  requirePermission(context, "job:create");
  if (!context.project) throw httpError(400, "project_required", "生成任务必须绑定项目");
  requireQuota(context, "clip", 1);
  const requestedAdapterId = String(body.adapter || "").trim();
  if (!requestedAdapterId) throw httpError(400, "adapter_required", "生成任务必须选择模型连接器");
  const adapter = modelForContext(context, requestedAdapterId, body.kind);
  assertExternalAllowed(context, adapter, body);
  const dependencyIds = [...new Set((Array.isArray(body.dependsOnJobIds) ? body.dependsOnJobIds : []).map((id) => String(id || "").trim()).filter(Boolean))].slice(0, 12);
  dependencyRows(context, dependencyIds);
  assertNoDependencyCycle("__new_job__", dependencyIds);
  const selectedShot = body.shotId ? shotForJob(context, body.shotId) : null;
  if (body.shotId && !selectedShot) throw httpError(404, "shot_not_found", "镜头不存在或不属于当前项目", { shotId: body.shotId });
  const contract = buildContract(context, { ...body, shotId: selectedShot?.id || null }, adapter);
  const jobId = makeId("job");
  const timestamp = now();
  const outputPath = String(body.output || `storage/jobs/${jobId}/output`);
  if (outputPath.startsWith("/") || outputPath.includes("..")) throw httpError(400, "output_path_invalid", "输出路径必须是项目目录内的相对路径");
  const maxAttempts = Math.max(1, Math.min(10, Number(body.maxAttempts || 3)));
  const dependencyBlocked = dependencyIds.some((dependencyId) => dbGet("SELECT status FROM generation_jobs WHERE id = ?", [dependencyId])?.status !== "completed");
  const initialStatus = dependencyBlocked ? "blocked" : adapter.status === "ready" && adapter.costMode === "local" ? "queued" : "blocked";
  const errorMessage = dependencyBlocked
    ? `等待前置任务完成：${dependencyIds.join(", ")}`
    : initialStatus === "blocked" ? `连接器当前状态为 ${adapter.status}，请先在模型中台检测并启用连接器` : "";
  await writeJobRequest(jobId, contract);
  withTransaction(() => {
    dbRun(
      `INSERT INTO generation_jobs(
        id, organization_id, workspace_id, project_id, episode_id, shot_id, kind, adapter_id,
        status, priority, cost_policy, output_path, qa_status, request_json, result_json,
        error_message, max_attempts, next_run_at, leased_by, leased_at, created_by, created_at, updated_at
      ) VALUES (?, ?, ?, ?, (SELECT id FROM episodes WHERE season_id IN (SELECT id FROM seasons WHERE series_id = (SELECT id FROM series WHERE project_id = ?)) LIMIT 1), ?, ?, ?, ?, ?, 'local-only', ?, 'wait', ?, '{}', ?, ?, NULL, NULL, NULL, ?, ?, ?)`,
      [jobId, context.organization.id, context.workspace.id, context.project.id, context.project.id, selectedShot?.id || null, body.kind || "自定义生成任务", adapter.id, initialStatus, Math.max(1, Math.min(100, Number(body.priority || 50))), outputPath, JSON.stringify(contract), errorMessage, maxAttempts, context.user.id, timestamp, timestamp]
    );
    dbRun("INSERT INTO job_attempts(id, job_id, attempt_number, runner_id, status, error_message, created_at) VALUES (?, ?, 1, ?, ?, ?, ?)", [makeId("attempt"), jobId, adapter.id, initialStatus === "queued" ? "queued" : "blocked", errorMessage || null, timestamp]);
    for (const dependencyId of dependencyIds) {
      dbRun("INSERT INTO job_dependencies(job_id, depends_on_job_id, dependency_type, created_at) VALUES (?, ?, 'blocking', ?)", [jobId, dependencyId, timestamp]);
    }
  });
  addUsage({ context, kind: body.kind || "generation", units: 1, unitName: "job", estimatedCost: 0, metadata: { jobId, adapter: adapter.id, status: initialStatus } });
  addAudit({ context, action: "generation_job.created", targetType: "generation_job", targetId: jobId, result: initialStatus === "queued" ? "ok" : "blocked", metadata: { kind: body.kind, shotId: body.shotId || null, adapter: adapter.id, status: initialStatus } });
  return { job: jobPayload(dbGet("SELECT * FROM generation_jobs WHERE id = ?", [jobId])), jobs: listGenerationJobs(context) };
}

export function listGenerationJobs(context, options = {}) {
  if (!context.project) return [];
  const status = String(options.status || "").trim();
  const params = [context.organization.id, context.workspace.id, context.project.id];
  let where = "j.organization_id = ? AND j.workspace_id = ? AND j.project_id = ?";
  if (status) { where += " AND j.status = ?"; params.push(status); }
  const rows = dbAll(
    `SELECT j.*, p.name AS project_name, u.display_name AS creator_name,
            (SELECT COUNT(*) FROM job_attempts a WHERE a.job_id = j.id) AS attempts
     FROM generation_jobs j
     JOIN projects p ON p.id = j.project_id
     LEFT JOIN users u ON u.id = j.created_by
     WHERE ${where}
     ORDER BY j.created_at DESC LIMIT ?`,
    [...params, Math.max(1, Math.min(500, Number(options.limit || 100)))]
  );
  return rows.map(jobPayload);
}

export function getGenerationJob(context, jobId) {
  return jobPayload(jobRow(context, jobId));
}

async function fetchWithTimeout(url, options, timeoutMs = 20000) {
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), timeoutMs);
  try {
    return await fetch(url, { ...options, signal: controller.signal });
  } finally {
    clearTimeout(timer);
  }
}

function normalizeResult(value, fallbackPath) {
  if (!value || typeof value !== "object") return { raw: value, outputPath: fallbackPath };
  const outputPath = value.outputPath || value.output_path || value.path || value.file || fallbackPath;
  return { ...value, outputPath };
}

function adapterProtocol(adapter) {
  const configured = adapter.protocol && typeof adapter.protocol === "object"
    ? adapter.protocol
    : parseJson(adapter.protocol_json, {});
  return configured && typeof configured === "object" ? configured : {};
}

function joinEndpoint(baseEndpoint, route) {
  if (!route) return new URL(baseEndpoint);
  if (/^https?:\/\//i.test(route)) return new URL(route);
  const base = String(baseEndpoint || "").replace(/\/+$/, "");
  const suffix = `/${String(route).replace(/^\/+/, "")}`;
  return new URL(`${base}${suffix}`);
}

function operationForJob(job) {
  const kind = String(job.kind || "").toLowerCase();
  if (kind.includes("tts") || kind.includes("配音") || kind.includes("声音")) return "tts";
  if (kind.includes("asr") || kind.includes("字幕") || kind.includes("对齐")) return "asr";
  if (kind.includes("视频") || kind.includes("i2v") || kind.includes("video")) return "video";
  if (kind.includes("图") || kind.includes("关键帧") || kind.includes("image")) return "image";
  return "chat";
}

function contractText(contract) {
  return [
    contract?.shot?.prompt,
    contract?.shot?.videoPrompt,
    contract?.shot?.action,
    contract?.inputs?.prompt,
    contract?.inputs?.text,
    contract?.inputs?.input
  ].filter(Boolean).join("\n");
}

function authHeaders(adapter) {
  const envKey = String(adapter.auth_env || adapter.authEnv || adapter.protocol?.authEnv || "").trim();
  const token = envKey ? process.env[envKey] : "";
  return token ? { authorization: `Bearer ${token}` } : {};
}

async function openAiMultipartRequest(url, headers, protocol, contract, job, timestamp, attemptNumber) {
  const inputPath = String(contract?.inputs?.inputFilePath || contract?.inputs?.audioPath || "").trim();
  if (!inputPath) {
    return {
      url,
      options: {
        method: "POST",
        headers: { ...headers, "content-type": "application/json", accept: "application/json" },
        body: JSON.stringify({
          model: protocol.models?.asr || protocol.model || contract?.inputs?.model || "local-asr",
          language: contract?.inputs?.language || "zh",
          response_format: contract?.inputs?.response_format || "verbose_json",
          metadata: { jobId: job.id, attemptNumber, requestedAt: timestamp },
          input: contractText(contract)
        })
      }
    };
  }
  if (inputPath.startsWith("/") || inputPath.includes("..")) throw httpError(400, "input_path_invalid", "音频输入必须是 storage/ 下的相对路径");
  const absolutePath = resolve(projectRoot, inputPath);
  const audio = await readFile(absolutePath);
  const form = new FormData();
  form.append("model", protocol.models?.asr || protocol.model || contract?.inputs?.model || "local-asr");
  form.append("language", contract?.inputs?.language || "zh");
  form.append("response_format", contract?.inputs?.response_format || "verbose_json");
  form.append("file", new Blob([audio], { type: contract?.inputs?.mimeType || "audio/wav" }), basename(absolutePath));
  return { url, options: { method: "POST", headers: { ...headers, accept: "application/json" }, body: form } };
}

async function buildAdapterRequest(adapter, job, contract, attemptNumber, timestamp) {
  const protocol = adapterProtocol(adapter);
  const kind = String(adapter.kind || "http-json").toLowerCase();
  const operation = operationForJob(job);
  const headers = { ...authHeaders(adapter), accept: "application/json", "x-ai-drama-job-id": job.id };
  const execution = { jobId: job.id, attemptNumber, requestedAt: timestamp, operation };

  if (kind === "comfyui") {
    const url = joinEndpoint(adapter.endpoint, protocol.promptRoute || "prompt");
    return {
      url,
      options: {
        method: "POST",
        headers: { ...headers, "content-type": "application/json" },
        body: JSON.stringify({
          prompt: contract?.inputs?.workflow || contract?.workflow || contract,
          client_id: protocol.clientId || "ai-drama-platform",
          extra_data: { aiDrama: execution }
        })
      }
    };
  }

  if (kind.includes("openai-compatible")) {
    const defaultRoutes = {
      tts: "audio/speech",
      asr: "audio/transcriptions",
      image: "images/generations",
      video: "videos/generations",
      chat: "chat/completions"
    };
    const url = joinEndpoint(adapter.endpoint, protocol.routes?.[operation] || defaultRoutes[operation]);
    if (operation === "asr") return openAiMultipartRequest(url, headers, protocol, contract, job, timestamp, attemptNumber);
    const model = protocol.models?.[operation] || protocol.models?.default || protocol.model || contract?.inputs?.model || "local-model";
    let body;
    if (operation === "tts") {
      body = {
        model,
        input: contract?.inputs?.text || contract?.inputs?.input || contractText(contract),
        voice: contract?.inputs?.voice || contract?.inputs?.voiceId || "locked-voice",
        response_format: contract?.inputs?.response_format || "wav",
        speed: contract?.inputs?.speed || 1,
        metadata: execution
      };
    } else if (operation === "image") {
      body = {
        model,
        prompt: contract?.shot?.prompt || contract?.inputs?.prompt || contractText(contract),
        negative_prompt: contract?.shot?.negativePrompt || contract?.inputs?.negativePrompt || "",
        n: 1,
        size: contract?.inputs?.size || "928x1664",
        response_format: contract?.inputs?.response_format || "b64_json",
        metadata: execution
      };
    } else if (operation === "video") {
      body = {
        model,
        prompt: contract?.shot?.videoPrompt || contract?.inputs?.prompt || contractText(contract),
        image: contract?.inputs?.firstFrame || contract?.shot?.firstFrame || undefined,
        first_frame: contract?.inputs?.firstFrame || contract?.shot?.firstFrame || undefined,
        last_frame: contract?.inputs?.lastFrame || contract?.shot?.lastFrame || undefined,
        duration: contract?.shot?.durationSec,
        n: 1,
        metadata: execution
      };
    } else {
      body = {
        model,
        messages: [{ role: "user", content: contractText(contract) || JSON.stringify(contract) }],
        temperature: 0.2,
        metadata: execution
      };
    }
    return { url, options: { method: "POST", headers: { ...headers, "content-type": "application/json" }, body: JSON.stringify(body) } };
  }

  return {
    url: new URL(adapter.endpoint),
    options: {
      method: "POST",
      headers: { ...headers, "content-type": "application/json" },
      body: JSON.stringify({ ...contract, execution })
    }
  };
}

async function readAdapterResponse(response, jobId) {
  const contentType = String(response.headers.get("content-type") || "").toLowerCase();
  if (contentType.includes("json") || contentType.startsWith("text/")) {
    const raw = await response.text();
    let parsed;
    try { parsed = raw ? JSON.parse(raw) : {}; } catch { parsed = { raw }; }
    return { parsed, contentType };
  }
  const buffer = Buffer.from(await response.arrayBuffer());
  const resultRelative = `storage/jobs/${jobId}/output${contentType.includes("wav") ? ".wav" : contentType.includes("mp4") ? ".mp4" : ".bin"}`;
  await mkdir(resolve(projectRoot, "storage", "jobs", jobId), { recursive: true });
  await writeFile(resolve(projectRoot, resultRelative), buffer);
  return { parsed: { outputPath: resultRelative, mimeType: contentType || "application/octet-stream", bytes: buffer.length, binary: true }, contentType };
}

function assertSingleFrameResult(job, result) {
  const kind = String(job.kind || "").toLowerCase();
  if (!(kind.includes("图") || kind.includes("关键帧") || kind.includes("image"))) return;
  const candidates = [result?.data, result?.images, result?.outputs].filter(Array.isArray);
  for (const values of candidates) {
    if (values.length !== 1) throw new Error(`一图一画面校验失败：模型返回了 ${values.length} 个图像结果，要求恰好 1 个`);
  }
}

export async function executeGenerationJob(context, jobId, body = {}) {
  const job = jobRow(context, jobId);
  if (!hasPermission(context, "job:create") && !hasPermission(context, "queue:manage")) throw httpError(403, "permission_denied", "没有执行生成任务的权限");
  requireProjectWritable(context);
  if (!["queued", "blocked", "failed", "cancelled"].includes(job.status)) throw httpError(409, "job_not_runnable", "当前任务状态不能执行", { status: job.status });
  const unresolved = unresolvedDependencies(jobId);
  if (unresolved.length) {
    const message = `等待前置任务完成：${unresolved.map((item) => item.depends_on_job_id).join(", ")}`;
    dbRun("UPDATE generation_jobs SET status = 'blocked', error_message = ?, updated_at = ? WHERE id = ?", [message, now(), jobId]);
    addAudit({ context, action: "generation_job.blocked_by_dependencies", targetType: "generation_job", targetId: jobId, result: "blocked", metadata: { dependencies: unresolved } });
    throw httpError(409, "job_dependencies_unresolved", message, { dependencies: unresolved });
  }
  const adapter = modelForContext(context, job.adapter_id);
  assertExternalAllowed(context, adapter, body);
  if (body.approveExternal && adapter.costMode !== "local") {
    addAudit({ context, action: "generation_job.external_approved", targetType: "generation_job", targetId: jobId, metadata: { adapter: adapter.id, costMode: adapter.costMode, approvalRequired: adapter.approvalRequired } });
  }
  const { local } = endpointInfo(adapter.endpoint);
  if (!local && adapter.costMode === "local") throw httpError(403, "local_only_endpoint_required", "local-only 任务只能调用本机或私有局域网 HTTP 连接器");
  if (adapter.status !== "ready") {
    const message = `连接器当前状态为 ${adapter.status}，请先完成探活并启用连接器`;
    dbRun("UPDATE generation_jobs SET status = 'blocked', error_message = ?, updated_at = ? WHERE id = ?", [message, now(), jobId]);
    throw httpError(409, "adapter_not_ready", message, { adapterId: adapter.id });
  }
  const timestamp = now();
  const attemptNumber = Number(dbGet("SELECT MAX(attempt_number) AS attempt_number FROM job_attempts WHERE job_id = ?", [jobId])?.attempt_number || 0) + 1;
  const attemptId = makeId("attempt");
  dbRun("UPDATE generation_jobs SET status = 'running', error_message = '', started_at = ?, finished_at = NULL, updated_at = ? WHERE id = ?", [timestamp, timestamp, jobId]);
  dbRun("INSERT INTO job_attempts(id, job_id, attempt_number, runner_id, status, started_at, created_at) VALUES (?, ?, ?, ?, 'running', ?, ?)", [attemptId, jobId, attemptNumber, adapter.id, timestamp, timestamp]);
  const contract = parseJson(job.request_json, {});
  try {
    const request = await buildAdapterRequest(adapter, job, contract, attemptNumber, timestamp);
    const response = await fetchWithTimeout(request.url.toString(), request.options);
    const { parsed } = await readAdapterResponse(response, jobId);
    if (!response.ok) throw new Error(`模型连接器返回 HTTP ${response.status}: ${String(parsed?.detail || parsed?.error || parsed?.raw || "无响应内容").slice(0, 500)}`);
    const result = normalizeResult(parsed, job.output_path);
    assertSingleFrameResult(job, result);
    const finishedAt = now();
    const resultRelative = `storage/jobs/${jobId}/result.json`;
    await mkdir(resolve(projectRoot, "storage", "jobs", jobId), { recursive: true });
    const artifacts = await registerJobArtifacts(context, job, result);
    const storedResult = { ...result, resultFile: resultRelative, artifacts };
    await writeFile(resolve(projectRoot, resultRelative), `${JSON.stringify(storedResult, null, 2)}\n`, "utf8");
    dbRun("UPDATE generation_jobs SET status = 'completed', result_json = ?, output_path = ?, finished_at = ?, updated_at = ? WHERE id = ?", [JSON.stringify(storedResult), result.outputPath || job.output_path, finishedAt, finishedAt, jobId]);
    dbRun("UPDATE job_attempts SET status = 'completed', finished_at = ? WHERE id = ?", [finishedAt, attemptId]);
    dbRun("UPDATE model_connectors SET status = 'ready', last_probe_at = ?, latency_ms = ?, error_message = '' WHERE id = ?", [finishedAt, Math.max(0, Date.parse(finishedAt) - Date.parse(timestamp)), adapter.id]);
    releaseReadyDependents(jobId);
    addUsage({ context, kind: `${job.kind}:executed`, units: 1, unitName: "execution", estimatedCost: 0, metadata: { jobId, adapter: adapter.id, attemptNumber } });
    addAudit({ context, action: "generation_job.completed", targetType: "generation_job", targetId: jobId, metadata: { adapter: adapter.id, attemptNumber, resultFile: resultRelative, artifactCount: artifacts.length } });
    void dispatchNotificationEvent({ context, eventKey: "job.completed", payload: { jobId, kind: job.kind, adapterId: adapter.id, resultFile: resultRelative, artifactCount: artifacts.length } });
    return { job: jobPayload(dbGet("SELECT * FROM generation_jobs WHERE id = ?", [jobId])), jobs: listGenerationJobs(context) };
  } catch (error) {
    const finishedAt = now();
    const message = error.name === "AbortError" ? "模型连接器请求超时" : String(error.message || error).slice(0, 1000);
    dbRun("UPDATE generation_jobs SET status = 'failed', error_message = ?, finished_at = ?, updated_at = ? WHERE id = ?", [message, finishedAt, finishedAt, jobId]);
    dbRun("UPDATE job_attempts SET status = 'failed', error_message = ?, finished_at = ? WHERE id = ?", [message, finishedAt, attemptId]);
    dbRun("UPDATE model_connectors SET status = 'error', last_probe_at = ?, error_message = ? WHERE id = ?", [finishedAt, message, adapter.id]);
    addAudit({ context, action: "generation_job.failed", targetType: "generation_job", targetId: jobId, result: "error", metadata: { adapter: adapter.id, attemptNumber, error: message } });
    void dispatchNotificationEvent({ context, eventKey: "job.failed", payload: { jobId, kind: job.kind, adapterId: adapter.id, attemptNumber, error: message } });
    throw httpError(502, "adapter_execution_failed", message, { jobId, adapterId: adapter.id });
  }
}

export async function probeModelConnector(context, modelId) {
  requirePermission(context, "model:manage");
  const adapter = modelForContext(context, modelId);
  const { local } = endpointInfo(adapter.endpoint);
  if (!local && adapter.costMode === "local") throw httpError(403, "local_only_endpoint_required", "local-only 连接器只能指向本机或私有局域网地址");
  const protocol = adapterProtocol(adapter);
  const probeRoute = protocol.healthRoute || (String(adapter.kind || "").toLowerCase().includes("openai-compatible") ? "models" : String(adapter.kind || "").toLowerCase() === "comfyui" ? "system_stats" : "");
  const probeUrl = joinEndpoint(adapter.endpoint, probeRoute);
  const started = Date.now();
  let status = "error";
  let message = "";
  let httpStatus = null;
  try {
    const response = await fetchWithTimeout(probeUrl.toString(), { method: "GET", headers: { ...authHeaders(adapter), accept: "application/json" } }, 8000);
    httpStatus = response.status;
    if (response.ok || response.status === 401 || response.status === 403 || response.status === 405) status = "ready";
    else message = `探活返回 HTTP ${response.status}`;
  } catch (error) {
    message = error.name === "AbortError" ? "探活超时" : String(error.message || error).slice(0, 500);
  }
  const timestamp = now();
  dbRun("UPDATE model_connectors SET status = ?, last_probe_at = ?, latency_ms = ?, error_message = ?, updated_at = ? WHERE id = ?", [status, timestamp, Date.now() - started, message, timestamp, modelId]);
  addAudit({ context, action: "model_connector.probed", targetType: "model_connector", targetId: modelId, result: status === "ready" ? "ok" : "error", metadata: { status, httpStatus, latencyMs: Date.now() - started, message } });
  return { model: parseModelRow(dbGet("SELECT * FROM model_connectors WHERE id = ?", [modelId])), httpStatus, message };
}

export function updateModelConnector(context, modelId, body) {
  requirePermission(context, "model:manage");
  const current = dbGet("SELECT * FROM model_connectors WHERE id = ? AND organization_id = ? AND (workspace_id IS NULL OR workspace_id = ?)", [modelId, context.organization.id, context.workspace.id]);
  if (!current) throw httpError(404, "adapter_not_found", "模型连接器不存在或不属于当前工作区");
  const label = String(body.label ?? current.label).trim();
  const endpoint = String(body.endpoint ?? current.endpoint).trim();
  const kind = String(body.kind ?? current.kind).trim();
  const costMode = String(body.costMode ?? current.cost_mode).trim();
  const status = String(body.status ?? current.status).trim();
  if (!label || !endpoint) throw httpError(400, "adapter_fields_required", "连接器名称和地址不能为空");
  if (!["local", "mixed", "cloud"].includes(costMode)) throw httpError(400, "cost_mode_invalid", "成本策略无效");
  if (!["ready", "not-connected", "planned", "optional", "paused", "error"].includes(status)) throw httpError(400, "adapter_status_invalid", "连接器状态无效");
  const { local } = endpointInfo(endpoint);
  if (costMode === "local" && !local) throw httpError(403, "local_only_endpoint_required", "local-only 连接器只能指向本机或私有局域网地址");
  const requestedApproval = body.approvalRequired === undefined ? Boolean(current.approval_required) : Boolean(body.approvalRequired);
  if (costMode !== "local" && !requestedApproval) throw httpError(400, "external_connector_approval_required", "混合或外部连接器必须开启审批策略");
  const approvalRequired = costMode === "local" ? requestedApproval : true;
  const capabilities = Array.isArray(body.capability) ? body.capability : parseJson(current.capabilities_json, []);
  const currentProtocol = parseJson(current.protocol_json, {});
  const protocol = body.protocol && typeof body.protocol === "object" ? body.protocol : currentProtocol;
  const authEnv = String(body.authEnv ?? current.auth_env ?? protocol.authEnv ?? "").trim();
  const timestamp = now();
  dbRun("UPDATE model_connectors SET label = ?, kind = ?, capabilities_json = ?, endpoint = ?, status = ?, cost_mode = ?, approval_required = ?, protocol_json = ?, auth_env = ?, error_message = ?, updated_at = ? WHERE id = ?", [label, kind, JSON.stringify(capabilities), endpoint, status, costMode, approvalRequired ? 1 : 0, JSON.stringify(protocol), authEnv, status === "error" ? current.error_message : "", timestamp, modelId]);
  addAudit({ context, action: "model_connector.updated", targetType: "model_connector", targetId: modelId, metadata: { label, endpoint, status, costMode, approvalRequired } });
  return parseModelRow(dbGet("SELECT * FROM model_connectors WHERE id = ?", [modelId]));
}