harn-stdlib 0.9.17

/**
 * `harn models batch` renderer.
 *
 * Batch APIs are for latency-tolerant, provider-billed work. Interactive
 * agent turns stay synchronous; this planner only reports catalog routes
 * whose provider capabilities advertise async batch support, then derives
 * manifests and provider-native request artifacts from that catalog data.
 *
 * Inputs (from the dispatch shim):
 *   HARN_MODELS_BATCH_PROVIDER                  - optional provider filter.
 *   HARN_MODELS_BATCH_MODEL                     - optional model/alias filter.
 *   HARN_MODELS_BATCH_WORKLOAD                  - eval, judge, corpus, or generic.
 *   HARN_MODELS_BATCH_MIN_DISCOUNT_PERCENT      - optional integer filter.
 *   HARN_MODELS_BATCH_MAX_TURNAROUND_HOURS      - optional integer filter.
 *   HARN_MODELS_BATCH_MODE                      - plan, manifest, prepare, submit, status, or download.
 *   HARN_MODELS_BATCH_REQUESTS                  - manifest-mode input JSONL path.
 *   HARN_MODELS_BATCH_OUT                       - manifest-mode output JSON path.
 *   HARN_MODELS_BATCH_MANIFEST                  - prepare-mode manifest JSON path.
 *   HARN_MODELS_BATCH_OUT_DIR                   - prepare-mode artifact directory.
 *   HARN_MODELS_BATCH_RECEIPT                   - submit-mode prepare receipt path.
 *   HARN_MODELS_BATCH_SUBMIT_OUT                - submit-mode submission receipt path.
 *   HARN_MODELS_BATCH_SUBMISSION                - status-mode submission receipt path.
 *   HARN_MODELS_BATCH_STATUS_OUT                - status-mode status receipt path.
 *   HARN_MODELS_BATCH_STATUS                    - download-mode status receipt path.
 *   HARN_MODELS_BATCH_RESULTS_OUT_DIR           - download-mode result artifact directory.
 *   HARN_MODELS_BATCH_MAX_DOWNLOAD_BYTES        - download-mode per-file response cap.
 *   HARN_MODELS_BATCH_DRY_RUN                   - "1" to validate without provider calls.
 *   HARN_MODELS_BATCH_TOOL_FORMAT               - manifest-mode default tool format.
 *   HARN_MODELS_BATCH_ID_PREFIX                 - manifest-mode generated id prefix.
 *   HARN_OUTPUT_JSON                            - "1" for JSON, else human text.
 */
import {
  cli_json_envelope,
  print_list,
  safe_bool,
  safe_dict,
  safe_int_string,
  safe_list,
  safe_string,
} from "std/cli/render"

const SUPPORTED_WORKLOADS: list = ["eval", "judge", "corpus", "generic"]

const BATCH_ROW_META_FIELDS: list = [
  "provider",
  "model",
  "tool_format",
  "endpoint",
  "custom_id",
  "id",
  "metadata",
  "body",
  "params",
  "request",
]

fn __safe_int(value) -> int? {
  if type_of(value) == "int" {
    return value
  }
  return to_int(value)
}

fn __has_string(items: list, needle: string) -> bool {
  for item in items {
    if type_of(item) == "string" && item == needle {
      return true
    }
  }
  return false
}

fn __normalize_workload(value: string) -> string {
  let workload = trim(value)
  if __has_string(SUPPORTED_WORKLOADS, workload) {
    return workload
  }
  return "generic"
}

fn __filter_int(value: string) -> int? {
  let text = trim(value)
  if text == "" {
    return nil
  }
  return to_int(text)
}

fn __model_matches(entry: dict, filter: string) -> bool {
  let model_filter = trim(filter)
  if model_filter == "" {
    return true
  }
  let provider = safe_string(entry["provider"], "")
  let id = safe_string(entry["id"], "")
  let name = safe_string(entry["name"], "")
  if model_filter == id || model_filter == name || model_filter == provider + "/" + id {
    return true
  }
  let provider_prefix = provider + "/"
  if len(id) >= len(provider_prefix) && id[0:len(provider_prefix)] == provider_prefix {
    let unprefixed = id[len(provider_prefix):len(id)]
    if model_filter == unprefixed || model_filter == provider + "/" + unprefixed {
      return true
    }
  }
  return false
}

fn __sorted_strings(items: list) -> list {
  var out = []
  for item in items {
    if type_of(item) != "string" {
      continue
    }
    var idx = 0
    while idx < len(out) && out[idx] < item {
      idx = idx + 1
    }
    out = out[0:idx].push(item) + out[idx:len(out)]
  }
  return out
}

fn __sort_models_by_provider_id(models: list) -> list {
  var out = []
  for model in models {
    let m = safe_dict(model)
    let key = safe_string(m["provider"], "") + "/" + safe_string(m["id"], "")
    var idx = 0
    while idx < len(out) {
      let other = safe_dict(out[idx])
      let other_key = safe_string(other["provider"], "") + "/" + safe_string(other["id"], "")
      if other_key >= key {
        break
      }
      idx = idx + 1
    }
    out = out[0:idx].push(model) + out[idx:len(out)]
  }
  return out
}

fn __workload_recommendation(workload: string) -> string {
  if workload == "eval" {
    return "Use for lower-priority eval gauntlets, judge calibration sweeps, and nightly comparisons."
  }
  if workload == "judge" {
    return "Use for asynchronous rubric calibration, transcript mining, and label generation."
  }
  if workload == "corpus" {
    return "Use for corpus refresh, distillation, and large synthetic-data review passes."
  }
  return "Use for latency-tolerant bulk inference with stable request ids."
}

fn __live_batch_adapter_key(provider: string, wire: string) -> string {
  if provider == "anthropic" && wire == "anthropic_messages" {
    return "anthropic"
  }
  if provider == "openai" && wire == "openai" {
    return "openai"
  }
  if provider == "groq" && wire == "openai" {
    return "groq"
  }
  if provider == "together" && wire == "openai" {
    return "together"
  }
  if provider == "parasail" && wire == "openai" {
    return "parasail"
  }
  if provider == "gemini" && wire == "gemini" {
    return "gemini"
  }
  if provider == "fireworks" && wire == "fireworks" {
    return "fireworks"
  }
  if provider == "mistral" && wire == "mistral" {
    return "mistral"
  }
  if provider == "xai" && wire == "xai" {
    return "xai"
  }
  return ""
}

fn __live_batch_adapter(provider: string, wire: string) -> dict {
  let supported = __live_batch_adapter_key(provider, wire) != ""
  return {submit: supported, status: supported, download: supported}
}

fn __is_openai_compatible_live_adapter(adapter: string) -> bool {
  return adapter == "openai" || adapter == "groq" || adapter == "together" || adapter == "parasail"
}

fn __live_batch_operation_error(operation: string, provider: string, wire: string) -> string {
  return "live batch "
    + operation
    + " is not implemented for provider="
    + provider
    + " wire="
    + wire
    + "; use --dry-run or add a provider adapter"
}

fn __constraints(workload: string) -> list {
  return [
    "Use batch only for offline work; interactive agent turns and escalation repairs stay synchronous.",
    "Group requests by provider, model, batch wire format, and tool-call convention before submission.",
    "Persist a stable custom id per request so provider results can rejoin eval/corpus rows deterministically.",
    "Treat provider subscription-plan CLI auth as separate from API/batch credentials; batch jobs need provider API billing.",
    __workload_recommendation(workload),
  ]
}

fn __provider_group(models: list) -> dict {
  var groups = {}
  for model in models {
    let m = safe_dict(model)
    let provider = safe_string(m["provider"], "")
    if provider == "" {
      continue
    }
    let prior = if groups[provider] == nil {
      []
    } else {
      groups[provider]
    }
    groups = groups + {[provider]: prior.push(m)}
  }
  return groups
}

fn __candidate(entry: dict, workload: string) -> dict {
  let provider = safe_string(entry["provider"], "")
  let id = safe_string(entry["id"], "")
  let discount = __safe_int(entry["batch_discount_percent"])
  let turnaround = __safe_int(entry["batch_turnaround_hours"])
  let wire_format = safe_string(entry["batch_wire_format"], "")
  let input_mode = safe_string(entry["batch_input_mode"], "")
  let live_adapter = __live_batch_adapter(provider, wire_format)
  let recommended = discount != nil && discount > 0 && (turnaround == nil || turnaround <= 24)
  var constraints = []
  if wire_format == "" {
    constraints = constraints.push("Provider catalog omits batch wire format; submission must probe before enqueue.")
  }
  if input_mode == "" {
    constraints = constraints
      .push(
      "Provider catalog omits batch input mode; submission must choose a provider-native packer.",
    )
  }
  if discount == nil {
    constraints = constraints
      .push(
      "Provider catalog does not publish a discount; cost planner must price from live usage tables.",
    )
  }
  if turnaround == nil {
    constraints = constraints
      .push(
      "Provider catalog does not publish a turnaround bound; schedule as best-effort offline work.",
    )
  }
  if !safe_bool(live_adapter["submit"], false)
    || !safe_bool(live_adapter["status"], false)
    || !safe_bool(live_adapter["download"], false) {
    constraints = constraints
      .push(
      "Harn can prepare this provider's batch request files, but live submit/status/download is not implemented; use --dry-run or add a provider adapter.",
    )
  }
  return {
    provider: provider,
    id: id,
    name: safe_string(entry["name"], ""),
    capabilities: safe_list(entry["capabilities"]),
    batch: {
      api: true,
      wire_format: wire_format,
      input_mode: input_mode,
      discount_percent: discount,
      turnaround_hours: turnaround,
      harn_live_adapter: live_adapter,
    },
    workload: workload,
    recommended: recommended,
    recommendation: __workload_recommendation(workload),
    constraints: constraints,
  }
}

fn __collect(
  catalog: list,
  provider_filter: string,
  model_filter: string,
  workload: string,
  min_discount: int?,
  max_turnaround: int?,
) -> list {
  var out = []
  for raw_entry in catalog {
    let entry = safe_dict(raw_entry)
    let provider = safe_string(entry["provider"], "")
    let id = safe_string(entry["id"], "")
    if provider == "" || id == "" {
      continue
    }
    if trim(provider_filter) != "" && provider != trim(provider_filter) {
      continue
    }
    if !__model_matches(entry, model_filter) {
      continue
    }
    let capabilities = safe_list(entry["capabilities"])
    if !safe_bool(entry["batch_api"], false) && !__has_string(capabilities, "batch") {
      continue
    }
    let discount = __safe_int(entry["batch_discount_percent"])
    if min_discount != nil && (discount == nil || discount < min_discount) {
      continue
    }
    let turnaround = __safe_int(entry["batch_turnaround_hours"])
    if max_turnaround != nil && (turnaround == nil || turnaround > max_turnaround) {
      continue
    }
    out = out.push(__candidate(entry, workload))
  }
  return __sort_models_by_provider_id(out)
}

fn __render_batch_value(value, suffix: string, fallback: string) -> string {
  let parsed = __safe_int(value)
  if parsed == nil {
    return fallback
  }
  return to_string(parsed) + suffix
}

fn __render_model_line(model: dict) -> string {
  let batch = safe_dict(model["batch"])
  let wire = safe_string(batch["wire_format"], "unknown_wire")
  let input = safe_string(batch["input_mode"], "unknown_input")
  let discount = __render_batch_value(batch["discount_percent"], "%", "unknown discount")
  let turnaround = __render_batch_value(batch["turnaround_hours"], "h", "unknown turnaround")
  let live = if safe_bool(safe_dict(batch["harn_live_adapter"])["submit"], false) {
    "live submit"
  } else {
    "dry-run only"
  }
  return "  "
    + safe_string(model["id"], "")
    + "  "
    + wire
    + "/"
    + input
    + "  discount "
    + discount
    + "  turnaround "
    + turnaround
    + "  "
    + live
}

fn __render_human(harness: Harness, report: dict) {
  let workload = safe_string(report["workload"], "generic")
  let models = safe_list(report["models"])
  if len(models) == 0 {
    harness.stdio.println("(no batch-capable models match)")
  } else {
    harness.stdio.println("Batch-capable models for " + workload)
    let groups = __provider_group(models)
    for provider in __sorted_strings(keys(groups)) {
      harness.stdio.println(provider)
      for model in groups[provider] {
        harness.stdio.println(__render_model_line(safe_dict(model)))
        print_list(harness, "constraints", safe_list(safe_dict(model)["constraints"]))
      }
      harness.stdio.println("")
    }
  }
  print_list(harness, "planner constraints", safe_list(report["constraints"]))
  print_list(harness, "warnings", safe_list(report["warnings"]))
}

fn __render_json(report: dict) -> string {
  return json_stringify_pretty(cli_json_envelope({schema_version: 1, ok: true, data: report}))
}

fn __resolve_candidate(
  catalog: list,
  provider_filter: string,
  model_filter: string,
  workload: string,
) -> dict {
  let matches = __collect(catalog, provider_filter, model_filter, workload, nil, nil)
  if len(matches) == 0 {
    return {
      ok: false,
      error: "no batch-capable model matches provider='" + provider_filter + "' model='" + model_filter
        + "'",
    }
  }
  if len(matches) > 1 && trim(provider_filter) == "" {
    return {
      ok: false,
      error: "model '" + model_filter + "' matches multiple batch-capable providers; set provider",
    }
  }
  return {ok: true, candidate: safe_dict(matches[0])}
}

fn __row_string(row: dict, key: string, fallback: string) -> string {
  let value = safe_string(row[key], "")
  if trim(value) == "" {
    return fallback
  }
  return trim(value)
}

fn __custom_id(row: dict, line_no: int, id_prefix: string) -> string {
  let explicit = __row_string(row, "custom_id", __row_string(row, "id", ""))
  if explicit != "" {
    return explicit
  }
  return id_prefix + "-" + to_string(line_no) + "-" + sha256(json_stringify(row)).substring(0, 12)
}

fn __request_entry(row: dict, line_no: int, id_prefix: string) -> dict {
  let custom_id = __custom_id(row, line_no, id_prefix)
  return {
    custom_id: custom_id,
    source_line: line_no,
    source_sha256: sha256(json_stringify(row)),
    metadata: safe_dict(row["metadata"]),
    request: row,
  }
}

fn __group_key(provider: string, model: string, batch: dict, tool_format: string, endpoint: string) -> string {
  return provider
    + "\t"
    + model
    + "\t"
    + safe_string(batch["wire_format"], "")
    + "\t"
    + safe_string(batch["input_mode"], "")
    + "\t"
    + tool_format
    + "\t"
    + endpoint
}

fn __group_summary(group: dict) -> dict {
  return {
    id: safe_string(group["id"], ""),
    provider: safe_string(group["provider"], ""),
    model: safe_string(group["model"], ""),
    tool_format: safe_string(group["tool_format"], ""),
    endpoint: safe_string(group["endpoint"], ""),
    request_count: len(safe_list(group["requests"])),
  }
}

fn __read_request_rows(harness: Harness, requests_path: string) -> dict {
  var rows = []
  var errors = []
  if requests_path == "" {
    return {rows: rows, errors: ["--requests is required"]}
  }
  if !harness.fs.exists(requests_path) {
    return {rows: rows, errors: ["request JSONL not found: " + requests_path]}
  }
  let source = harness.fs.read_text(requests_path)
  var line_no = 0
  for line in source.split("\n") {
    line_no = line_no + 1
    let trimmed = trim(line)
    if trimmed == "" {
      continue
    }
    let parsed = try {
      json_parse(trimmed)
    } catch (e) {
      errors = errors.push("line " + to_string(line_no) + ": invalid JSON: " + to_string(e))
      continue
    }
    if type_of(parsed) != "dict" {
      errors = errors.push("line " + to_string(line_no) + ": expected a JSON object")
      continue
    }
    rows = rows.push({line_no: line_no, row: parsed})
  }
  return {rows: rows, errors: errors, source_sha256: sha256(source)}
}

fn __read_manifest(harness: Harness, manifest_path: string) -> dict {
  var errors = []
  if manifest_path == "" {
    return {manifest: {}, source_sha256: "", errors: ["--manifest is required"]}
  }
  if !harness.fs.exists(manifest_path) {
    return {manifest: {}, source_sha256: "", errors: ["batch manifest not found: " + manifest_path]}
  }
  let source = harness.fs.read_text(manifest_path)
  let parsed = try {
    json_parse(source)
  } catch (e) {
    return {
      manifest: {},
      source_sha256: sha256(source),
      errors: ["invalid batch manifest JSON: " + to_string(e)],
    }
  }
  if type_of(parsed) != "dict" {
    errors = errors.push("batch manifest must be a JSON object")
    return {manifest: {}, source_sha256: sha256(source), errors: errors}
  }
  let manifest = safe_dict(parsed)
  if safe_string(manifest["kind"], "") != "harn.model_batch_manifest" {
    errors = errors.push("unsupported batch manifest kind: " + safe_string(manifest["kind"], ""))
  }
  if len(safe_list(manifest["groups"])) == 0 {
    errors = errors.push("batch manifest contains no groups")
  }
  return {manifest: manifest, source_sha256: sha256(source), errors: errors}
}

fn __read_prepare_receipt(harness: Harness, receipt_path: string) -> dict {
  var errors = []
  if receipt_path == "" {
    return {receipt: {}, source_sha256: "", errors: ["--receipt is required"]}
  }
  if !harness.fs.exists(receipt_path) {
    return {receipt: {}, source_sha256: "", errors: ["batch prepare receipt not found: " + receipt_path]}
  }
  let source = harness.fs.read_text(receipt_path)
  let parsed = try {
    json_parse(source)
  } catch (e) {
    return {
      receipt: {},
      source_sha256: sha256(source),
      errors: ["invalid batch prepare receipt JSON: " + to_string(e)],
    }
  }
  if type_of(parsed) != "dict" {
    errors = errors.push("batch prepare receipt must be a JSON object")
    return {receipt: {}, source_sha256: sha256(source), errors: errors}
  }
  let receipt = safe_dict(parsed)
  if safe_string(receipt["kind"], "") != "harn.model_batch_prepare_receipt" {
    errors = errors.push("unsupported batch prepare receipt kind: " + safe_string(receipt["kind"], ""))
  }
  if len(safe_list(receipt["jobs"])) == 0 {
    errors = errors.push("batch prepare receipt contains no jobs")
  }
  return {receipt: receipt, source_sha256: sha256(source), errors: errors}
}

fn __read_submission_receipt(harness: Harness, receipt_path: string) -> dict {
  var errors = []
  if receipt_path == "" {
    return {receipt: {}, source_sha256: "", errors: ["--submission is required"]}
  }
  if !harness.fs.exists(receipt_path) {
    return {receipt: {}, source_sha256: "", errors: ["batch submission receipt not found: " + receipt_path]}
  }
  let source = harness.fs.read_text(receipt_path)
  let parsed = try {
    json_parse(source)
  } catch (e) {
    return {
      receipt: {},
      source_sha256: sha256(source),
      errors: ["invalid batch submission receipt JSON: " + to_string(e)],
    }
  }
  if type_of(parsed) != "dict" {
    errors = errors.push("batch submission receipt must be a JSON object")
    return {receipt: {}, source_sha256: sha256(source), errors: errors}
  }
  let receipt = safe_dict(parsed)
  if safe_string(receipt["kind"], "") != "harn.model_batch_submission_receipt" {
    errors = errors.push("unsupported batch submission receipt kind: " + safe_string(receipt["kind"], ""))
  }
  if len(safe_list(receipt["jobs"])) == 0 {
    errors = errors.push("batch submission receipt contains no jobs")
  }
  return {receipt: receipt, source_sha256: sha256(source), errors: errors}
}

fn __read_status_receipt(harness: Harness, receipt_path: string) -> dict {
  var errors = []
  if receipt_path == "" {
    return {receipt: {}, source_sha256: "", errors: ["--status is required"]}
  }
  if !harness.fs.exists(receipt_path) {
    return {receipt: {}, source_sha256: "", errors: ["batch status receipt not found: " + receipt_path]}
  }
  let source = harness.fs.read_text(receipt_path)
  let parsed = try {
    json_parse(source)
  } catch (e) {
    return {
      receipt: {},
      source_sha256: sha256(source),
      errors: ["invalid batch status receipt JSON: " + to_string(e)],
    }
  }
  if type_of(parsed) != "dict" {
    errors = errors.push("batch status receipt must be a JSON object")
    return {receipt: {}, source_sha256: sha256(source), errors: errors}
  }
  let receipt = safe_dict(parsed)
  if safe_string(receipt["kind"], "") != "harn.model_batch_status_receipt" {
    errors = errors.push("unsupported batch status receipt kind: " + safe_string(receipt["kind"], ""))
  }
  if len(safe_list(receipt["jobs"])) == 0 {
    errors = errors.push("batch status receipt contains no jobs")
  }
  return {receipt: receipt, source_sha256: sha256(source), errors: errors}
}

fn __mkdirp(harness: Harness, path: string) {
  if trim(path) == "" {
    return
  }
  let absolute = starts_with(path, "/")
  var current = if absolute {
    "/"
  } else {
    ""
  }
  for part in split(path, "/") {
    if part == "" {
      continue
    }
    current = if current == "" || current == "/" {
      current + part
    } else {
      path_join(current, part)
    }
    try {
      harness.fs.mkdir(current)
    } catch (_e) {
      nil
    }
  }
}

fn __dict_without_batch_fields(row: dict) -> dict {
  var out = {}
  for key in row.keys().sort() {
    if !__has_string(BATCH_ROW_META_FIELDS, key) {
      out = out + {[key]: row[key]}
    }
  }
  return out
}

fn __body_from_row(row: dict, model: string, include_model: bool) -> dict {
  var body = safe_dict(row["params"])
  if body == {} {
    body = safe_dict(row["body"])
  }
  if body == {} {
    body = safe_dict(row["request"])
  }
  if body == {} {
    body = __dict_without_batch_fields(row)
  }
  if include_model && body["model"] == nil && model != "" {
    body = body + {model: model}
  }
  return body
}

fn __batch_endpoint(group: dict) -> string {
  let endpoint = trim(safe_string(group["endpoint"], ""))
  if starts_with(endpoint, "/") {
    return endpoint
  }
  let wire = safe_string(safe_dict(group["batch"])["wire_format"], "")
  if wire == "anthropic_messages" {
    return "/v1/messages/batches"
  }
  if wire == "openai" || wire == "mistral" || wire == "fireworks" || wire == "xai" {
    return "/v1/chat/completions"
  }
  if wire == "gemini" {
    return "batchGenerateContent"
  }
  if endpoint != "" {
    return endpoint
  }
  return "provider_default"
}

fn __prepare_file_stem(group: dict) -> string {
  let seed = safe_string(group["id"], "") + ":" + safe_string(group["provider"], "") + ":"
    + safe_string(group["model"], "")
  return "batch_group_" + substring(sha256(seed), 0, 16)
}

fn __provider_request_record(group: dict, request: dict, endpoint: string) -> dict {
  let wire = safe_string(safe_dict(group["batch"])["wire_format"], "")
  let raw_row = safe_dict(request["request"])
  let model = safe_string(group["model"], "")
  let custom_id = safe_string(request["custom_id"], "")
  if wire == "anthropic_messages" {
    return {custom_id: custom_id, params: __body_from_row(raw_row, model, true)}
  }
  if wire == "gemini" {
    return {key: custom_id, request: __body_from_row(raw_row, model, false)}
  }
  if wire == "fireworks" {
    return {custom_id: custom_id, body: __body_from_row(raw_row, model, false)}
  }
  return {custom_id: custom_id, method: "POST", url: endpoint, body: __body_from_row(raw_row, model, true)}
}

fn __supported_prepare_wire(wire: string) -> bool {
  return wire == "openai"
    || wire == "mistral"
    || wire == "fireworks"
    || wire == "xai"
    || wire == "gemini"
    || wire == "anthropic_messages"
}

fn __request_file_text(wire: string, records: list) -> string {
  if wire == "anthropic_messages" {
    return json_stringify_pretty({requests: records}) + "\n"
  }
  var lines = []
  for record in records {
    lines = lines.push(json_stringify(record))
  }
  return join(lines, "\n") + "\n"
}

fn __submit_contract(group: dict, request_path: string, endpoint: string) -> dict {
  let provider = safe_string(group["provider"], "")
  let model = safe_string(group["model"], "")
  let batch = safe_dict(group["batch"])
  let wire = safe_string(batch["wire_format"], "")
  if wire == "anthropic_messages" {
    return {
      provider: provider,
      operation: "POST /v1/messages/batches",
      request_body: "Use the prepared JSON file as the request body.",
      request_file: request_path,
      result_stream: "GET results_url after processing_status becomes ended",
    }
  }
  if wire == "gemini" {
    return {
      provider: provider,
      operation: "batches.create",
      model: model,
      input: {mode: "file_api_jsonl", file: request_path},
      request_line_shape: "{key, request}",
    }
  }
  if wire == "fireworks" {
    return {
      provider: provider,
      operation: "POST /v1/accounts/{account_id}/batchInferenceJobs",
      model: model,
      upload: {
        create_dataset: "POST /v1/accounts/{account_id}/datasets",
        upload_dataset: "POST /v1/accounts/{account_id}/datasets/{dataset_id}:upload",
        file: request_path,
      },
      create_batch: {input_dataset_id: "<input-dataset-id>", output_dataset_id: "<output-dataset-id>"},
      request_line_shape: "{custom_id, body}",
      result_stream: "GET /v1/accounts/{account_id}/datasets/{output_dataset_id}:getDownloadEndpoint",
    }
  }
  if wire == "mistral" {
    return {
      provider: provider,
      operation: "POST /v1/batch/jobs",
      upload: {purpose: "batch", file: request_path},
      create_job: {endpoint: endpoint, model: model, input_files: ["<uploaded-file-id>"]},
    }
  }
  if wire == "xai" {
    return {
      provider: provider,
      operation: "POST /v1/batches",
      upload: {file: request_path},
      create_batch: {name: safe_string(group["id"], ""), input_file_id: "<uploaded-file-id>"},
      result_stream: "GET /v1/batches/{batch_id}/results after state.num_pending reaches 0",
    }
  }
  if provider == "together" && wire == "openai" {
    return {
      provider: provider,
      operation: "POST /v1/batches",
      upload: {purpose: "batch-api", file: request_path},
      create_batch: {endpoint: endpoint, input_file_id: "<uploaded-file-id>"},
    }
  }
  return {
    provider: provider,
    operation: "POST /v1/batches",
    upload: {purpose: "batch", file: request_path},
    create_batch: {endpoint: endpoint, completion_window: "24h", input_file_id: "<uploaded-file-id>"},
  }
}

fn __prepare_artifact(out_dir: string, group: dict) -> dict {
  let batch = safe_dict(group["batch"])
  let wire = safe_string(batch["wire_format"], "")
  var errors = []
  if !__supported_prepare_wire(wire) {
    errors = errors.push("unsupported batch wire format for prepare: " + wire)
    return {ok: false, errors: errors}
  }
  let requests = safe_list(group["requests"])
  if len(requests) == 0 {
    errors = errors.push("group " + safe_string(group["id"], "") + " contains no requests")
    return {ok: false, errors: errors}
  }
  let endpoint = __batch_endpoint(group)
  var records = []
  for request in requests {
    records = records.push(__provider_request_record(group, safe_dict(request), endpoint))
  }
  let ext = if wire == "anthropic_messages" {
    ".json"
  } else {
    ".jsonl"
  }
  let request_path = path_join(out_dir, __prepare_file_stem(group) + ext)
  let text = __request_file_text(wire, records)
  let job = {
    id: safe_string(group["id"], ""),
    provider: safe_string(group["provider"], ""),
    model: safe_string(group["model"], ""),
    workload: safe_string(group["workload"], ""),
    endpoint: endpoint,
    tool_format: safe_string(group["tool_format"], ""),
    batch: batch,
    input_mode: safe_string(batch["input_mode"], ""),
    request_count: len(records),
    request_file: request_path,
    request_file_sha256: sha256(text),
    request_format: if wire == "anthropic_messages" {
      "json_requests"
    } else {
      "jsonl"
    },
    status: "prepared",
    submit: __submit_contract(group, request_path, endpoint),
  }
  return {ok: true, job: job, path: request_path, text: text, errors: []}
}

fn __build_manifest(
  catalog: list,
  request_rows: list,
  source_path: string,
  source_sha256: string,
  out_path: string,
  provider_filter: string,
  model_filter: string,
  workload: string,
  default_tool_format: string,
  id_prefix: string,
  min_discount: int?,
  max_turnaround: int?,
) -> dict {
  var errors = []
  var warnings = []
  var groups = {}
  var group_keys = []
  var request_count = 0
  for item in request_rows {
    let entry = safe_dict(item)
    let line_no = __safe_int(entry["line_no"]) ?? 0
    let row = safe_dict(entry["row"])
    let provider = __row_string(row, "provider", provider_filter)
    let model = __row_string(row, "model", model_filter)
    if provider == "" && model == "" {
      errors = errors.push("line " + to_string(line_no) + ": provider/model missing")
      continue
    }
    if model == "" {
      errors = errors.push("line " + to_string(line_no) + ": model missing")
      continue
    }
    let resolved = __resolve_candidate(catalog, provider, model, workload)
    if !safe_bool(resolved["ok"], false) {
      errors = errors
        .push(
        "line " + to_string(line_no) + ": " + safe_string(resolved["error"], "unresolved model"),
      )
      continue
    }
    let candidate = safe_dict(resolved["candidate"])
    let batch = safe_dict(candidate["batch"])
    let discount = __safe_int(batch["discount_percent"])
    if min_discount != nil && (discount == nil || discount < min_discount) {
      errors = errors
        .push(
        "line "
          + to_string(line_no)
          + ": model discount below --min-discount-percent for "
          + safe_string(candidate["provider"], "")
          + "/"
          + safe_string(candidate["id"], ""),
      )
      continue
    }
    let turnaround = __safe_int(batch["turnaround_hours"])
    if max_turnaround != nil && (turnaround == nil || turnaround > max_turnaround) {
      errors = errors
        .push(
        "line "
          + to_string(line_no)
          + ": model turnaround exceeds --max-turnaround-hours for "
          + safe_string(candidate["provider"], "")
          + "/"
          + safe_string(candidate["id"], ""),
      )
      continue
    }
    let tool_format = __row_string(row, "tool_format", default_tool_format)
    let endpoint = __row_string(row, "endpoint", safe_string(batch["wire_format"], "provider_default"))
    let key = __group_key(
      safe_string(candidate["provider"], ""),
      safe_string(candidate["id"], ""),
      batch,
      tool_format,
      endpoint,
    )
    let request = __request_entry(row, line_no, id_prefix)
    let current = safe_dict(groups[key])
    if current == {} {
      let group_id = "batch_group_" + sha256(key).substring(0, 16)
      let group = {
        id: group_id,
        provider: safe_string(candidate["provider"], ""),
        model: safe_string(candidate["id"], ""),
        model_name: safe_string(candidate["name"], ""),
        workload: workload,
        endpoint: endpoint,
        tool_format: tool_format,
        batch: batch,
        requests: [request],
      }
      groups = groups + {[key]: group}
      group_keys = group_keys.push(key)
    } else {
      let updated = current + {requests: safe_list(current["requests"]).push(request)}
      groups = groups + {[key]: updated}
    }
    request_count = request_count + 1
  }
  var group_list = []
  for key in group_keys {
    let group = safe_dict(groups[key])
    group_list = group_list.push(group + {request_count: len(safe_list(group["requests"]))})
  }
  if len(request_rows) == 0 {
    errors = errors.push("request JSONL contains no request rows")
  }
  if out_path == "" {
    errors = errors.push("--out is required")
  }
  let manifest = {
    schemaVersion: 1,
    kind: "harn.model_batch_manifest",
    producer: "harn models batch manifest",
    workload: workload,
    source: {path: source_path, sha256: source_sha256, row_count: len(request_rows)},
    constraints: __constraints(workload),
    requestCount: request_count,
    groupCount: len(group_list),
    groups: group_list,
    warnings: warnings,
  }
  return {manifest: manifest, errors: errors, warnings: warnings}
}

fn __render_manifest_human(harness: Harness, report: dict) {
  if !safe_bool(report["ok"], false) {
    harness.stdio.eprintln("Batch manifest failed")
    print_list(harness, "errors", safe_list(report["errors"]))
    print_list(harness, "warnings", safe_list(report["warnings"]))
    return
  }
  harness.stdio.println("Batch manifest written: " + safe_string(report["out"], ""))
  harness.stdio.println("  requests: " + safe_int_string(report["request_count"], "0"))
  harness.stdio.println("  groups: " + safe_int_string(report["group_count"], "0"))
  harness.stdio.println("  manifest sha256: " + safe_string(report["manifest_sha256"], ""))
  for group in safe_list(report["groups"]) {
    let g = safe_dict(group)
    harness.stdio
      .println(
      "  "
        + safe_string(g["id"], "")
        + " "
        + safe_string(g["provider"], "")
        + "/"
        + safe_string(g["model"], "")
        + " "
        + safe_string(g["tool_format"], "")
        + " requests="
        + safe_int_string(g["request_count"], "0"),
    )
  }
  print_list(harness, "warnings", safe_list(report["warnings"]))
}

fn __render_manifest_json(report: dict) -> string {
  if safe_bool(report["ok"], false) {
    return json_stringify_pretty(cli_json_envelope({schema_version: 1, ok: true, data: report}))
  }
  return json_stringify_pretty(
    cli_json_envelope(
      {
        schema_version: 1,
        ok: false,
        error: {
          code: "batch_manifest_failed",
          message: "Batch manifest request rows could not be resolved into provider batch groups.",
          details: report,
        },
        warnings: safe_list(report["warnings"]),
      },
    ),
  )
}

fn __run_manifest(
  harness: Harness,
  catalog: list,
  provider_filter: string,
  model_filter: string,
  workload: string,
  min_discount: int?,
  max_turnaround: int?,
  planner_warnings: list,
) -> int {
  let requests_path = trim(harness.env.get_or("HARN_MODELS_BATCH_REQUESTS", ""))
  let out_path = trim(harness.env.get_or("HARN_MODELS_BATCH_OUT", ""))
  let default_tool_format = trim(harness.env.get_or("HARN_MODELS_BATCH_TOOL_FORMAT", "auto"))
  let id_prefix_raw = trim(harness.env.get_or("HARN_MODELS_BATCH_ID_PREFIX", "harn-batch"))
  let id_prefix = if id_prefix_raw == "" {
    "harn-batch"
  } else {
    id_prefix_raw
  }
  let loaded = __read_request_rows(harness, requests_path)
  let built = __build_manifest(
    catalog,
    safe_list(loaded["rows"]),
    requests_path,
    safe_string(loaded["source_sha256"], ""),
    out_path,
    provider_filter,
    model_filter,
    workload,
    default_tool_format,
    id_prefix,
    min_discount,
    max_turnaround,
  )
  let errors = safe_list(loaded["errors"]) + safe_list(built["errors"])
  let warnings = safe_list(planner_warnings) + safe_list(built["warnings"])
  let manifest = safe_dict(built["manifest"]) + {warnings: warnings}
  var report = {
    schema_version: 1,
    ok: len(errors) == 0,
    out: out_path,
    source: safe_dict(manifest["source"]),
    workload: workload,
    request_count: __safe_int(manifest["requestCount"]) ?? 0,
    group_count: __safe_int(manifest["groupCount"]) ?? 0,
    groups: safe_list(manifest["groups"]).map({ group -> __group_summary(safe_dict(group)) }),
    errors: errors,
    warnings: warnings,
  }
  if len(errors) == 0 {
    let manifest_text = json_stringify_pretty(manifest) + "\n"
    harness.fs.write_text(out_path, manifest_text)
    report = report + {manifest_sha256: sha256(manifest_text)}
  }
  if harness.env.get_or("HARN_OUTPUT_JSON", "0") == "1" {
    harness.stdio.println(__render_manifest_json(report))
  } else {
    __render_manifest_human(harness, report)
  }
  if len(errors) == 0 {
    return 0
  }
  return 1
}

fn __render_prepare_human(harness: Harness, report: dict) {
  if !safe_bool(report["ok"], false) {
    harness.stdio.eprintln("Batch prepare failed")
    print_list(harness, "errors", safe_list(report["errors"]))
    print_list(harness, "warnings", safe_list(report["warnings"]))
    return
  }
  harness.stdio.println("Batch prepare receipt written: " + safe_string(report["receipt"], ""))
  harness.stdio.println("  jobs: " + safe_int_string(report["job_count"], "0"))
  harness.stdio.println("  requests: " + safe_int_string(report["request_count"], "0"))
  harness.stdio.println("  receipt sha256: " + safe_string(report["receipt_sha256"], ""))
  for job in safe_list(report["jobs"]) {
    let j = safe_dict(job)
    harness.stdio
      .println(
      "  "
        + safe_string(j["id"], "")
        + " "
        + safe_string(j["provider"], "")
        + "/"
        + safe_string(j["model"], "")
        + " "
        + safe_string(safe_dict(j["batch"])["wire_format"], "")
        + " requests="
        + safe_int_string(j["request_count"], "0")
        + " file="
        + safe_string(j["request_file"], ""),
    )
  }
  print_list(harness, "warnings", safe_list(report["warnings"]))
}

fn __render_prepare_json(report: dict) -> string {
  if safe_bool(report["ok"], false) {
    return json_stringify_pretty(cli_json_envelope({schema_version: 1, ok: true, data: report}))
  }
  return json_stringify_pretty(
    cli_json_envelope(
      {
        schema_version: 1,
        ok: false,
        error: {
          code: "batch_prepare_failed",
          message: "Batch manifest could not be prepared into provider request files.",
          details: report,
        },
        warnings: safe_list(report["warnings"]),
      },
    ),
  )
}

fn __run_prepare(harness: Harness) -> int {
  let manifest_path = trim(harness.env.get_or("HARN_MODELS_BATCH_MANIFEST", ""))
  let out_dir = trim(harness.env.get_or("HARN_MODELS_BATCH_OUT_DIR", ""))
  let loaded = __read_manifest(harness, manifest_path)
  var errors = safe_list(loaded["errors"])
  var warnings = []
  if out_dir == "" {
    errors = errors.push("--out-dir is required")
  }
  let manifest = safe_dict(loaded["manifest"])
  var artifacts = []
  var jobs = []
  var request_count = 0
  if len(errors) == 0 {
    for group in safe_list(manifest["groups"]) {
      let artifact = __prepare_artifact(out_dir, safe_dict(group))
      errors = errors + safe_list(artifact["errors"])
      if safe_bool(artifact["ok"], false) {
        artifacts = artifacts.push(artifact)
        let job = safe_dict(artifact["job"])
        jobs = jobs.push(job)
        request_count = request_count + (__safe_int(job["request_count"]) ?? 0)
      }
    }
  }
  let receipt_path = if out_dir == "" {
    ""
  } else {
    path_join(out_dir, "receipt.json")
  }
  var receipt = {
    schemaVersion: 1,
    kind: "harn.model_batch_prepare_receipt",
    producer: "harn models batch prepare",
    manifest: {
      path: manifest_path,
      sha256: safe_string(loaded["source_sha256"], ""),
      source: safe_dict(manifest["source"]),
      request_count: __safe_int(manifest["requestCount"]) ?? 0,
      group_count: __safe_int(manifest["groupCount"]) ?? 0,
    },
    outDir: out_dir,
    jobCount: len(jobs),
    requestCount: request_count,
    jobs: jobs,
    status: if len(errors) == 0 {
      "prepared"
    } else {
      "failed"
    },
    warnings: warnings,
    errors: errors,
  }
  var report = {
    schema_version: 1,
    ok: len(errors) == 0,
    manifest: receipt["manifest"],
    out_dir: out_dir,
    receipt: receipt_path,
    job_count: len(jobs),
    request_count: request_count,
    jobs: jobs,
    errors: errors,
    warnings: warnings,
  }
  if len(errors) == 0 {
    __mkdirp(harness, out_dir)
    for artifact in artifacts {
      harness.fs.write_text(safe_string(artifact["path"], ""), safe_string(artifact["text"], ""))
    }
    let receipt_text = json_stringify_pretty(receipt) + "\n"
    harness.fs.write_text(receipt_path, receipt_text)
    receipt = receipt + {receipt_sha256: sha256(receipt_text)}
    report = report + {receipt_sha256: sha256(receipt_text)}
  }
  if harness.env.get_or("HARN_OUTPUT_JSON", "0") == "1" {
    harness.stdio.println(__render_prepare_json(report))
  } else {
    __render_prepare_human(harness, report)
  }
  if len(errors) == 0 {
    return 0
  }
  return 1
}

fn __truthy_env(value: string) -> bool {
  let text = lowercase(trim(value))
  return text == "1" || text == "true" || text == "yes" || text == "on"
}

fn __provider_auth_env(provider: string) -> string {
  if provider == "openai" {
    return "OPENAI_API_KEY"
  }
  if provider == "anthropic" {
    return "ANTHROPIC_API_KEY"
  }
  if provider == "mistral" {
    return "MISTRAL_API_KEY"
  }
  if provider == "xai" {
    return "XAI_API_KEY"
  }
  if provider == "together" {
    return "TOGETHER_AI_API_KEY"
  }
  return uppercase(provider) + "_API_KEY"
}

fn __provider_auth_env_candidates(provider: string) -> list {
  if provider == "together" {
    return ["TOGETHER_AI_API_KEY", "TOGETHER_API_KEY"]
  }
  if provider == "gemini" {
    return ["GEMINI_API_KEY", "GOOGLE_API_KEY"]
  }
  if provider == "fireworks" {
    return ["FIREWORKS_API_KEY"]
  }
  return [__provider_auth_env(provider)]
}

fn __provider_base_url(harness: Harness, provider: string) -> string {
  if provider == "openai" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_OPENAI_BASE_URL",
        harness.env.get_or("OPENAI_BASE_URL", "https://api.openai.com/v1"),
      ),
    )
  }
  if provider == "anthropic" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_ANTHROPIC_BASE_URL",
        harness.env.get_or("ANTHROPIC_BASE_URL", "https://api.anthropic.com"),
      ),
    )
  }
  if provider == "mistral" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_MISTRAL_BASE_URL",
        harness.env.get_or("MISTRAL_BASE_URL", "https://api.mistral.ai/v1"),
      ),
    )
  }
  if provider == "xai" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_XAI_BASE_URL",
        harness.env.get_or("XAI_BASE_URL", "https://api.x.ai/v1"),
      ),
    )
  }
  if provider == "gemini" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_GEMINI_BASE_URL",
        harness.env
          .get_or(
          "GEMINI_BASE_URL",
          harness.env
            .get_or(
            "GOOGLE_GENERATIVE_LANGUAGE_BASE_URL",
            "https://generativelanguage.googleapis.com",
          ),
        ),
      ),
    )
  }
  if provider == "groq" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_GROQ_BASE_URL",
        harness.env.get_or("GROQ_BASE_URL", "https://api.groq.com/openai/v1"),
      ),
    )
  }
  if provider == "together" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_TOGETHER_BASE_URL",
        harness.env
          .get_or(
          "TOGETHER_AI_BASE_URL",
          harness.env.get_or("TOGETHER_BASE_URL", "https://api.together.xyz/v1"),
        ),
      ),
    )
  }
  if provider == "fireworks" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_FIREWORKS_BASE_URL",
        harness.env.get_or("FIREWORKS_BASE_URL", "https://api.fireworks.ai/v1"),
      ),
    )
  }
  if provider == "parasail" {
    return trim(
      harness.env
        .get_or(
        "HARN_BATCH_PARASAIL_BASE_URL",
        harness.env.get_or("PARASAIL_BASE_URL", "https://api.saas.parasail.io/v1"),
      ),
    )
  }
  return ""
}

fn __join_url(base: string, path: string) -> string {
  let left = trim(base)
  let right = trim(path)
  if right == "" {
    return left
  }
  if ends_with(left, "/") && starts_with(right, "/") {
    return left[0:len(left) - 1] + right
  }
  if !ends_with(left, "/") && !starts_with(right, "/") {
    return left + "/" + right
  }
  return left + right
}

fn __parse_response_body(body: string) {
  let text = trim(body)
  if text == "" {
    return nil
  }
  return try {
    json_parse(text)
  } catch (_e) {
    nil
  }
}

fn __http_result(harness: Harness, method: string, url: string, options: dict) -> dict {
  let result = try {
    harness.net.request(method, url, options)
  }
  if is_err(result) {
    return {ok: false, error: "network error: " + to_string(unwrap_err(result))}
  }
  let response = unwrap(result)
  let body_text = to_string(response?.body ?? "")
  let parsed = __parse_response_body(body_text)
  return {
    ok: response?.ok ?? false,
    status: response?.status ?? nil,
    headers: response?.headers ?? {},
    body: parsed,
    body_text: body_text,
    final_url: response?.final_url ?? url,
  }
}

fn __request_file_payload(harness: Harness, job: dict) -> dict {
  let path = safe_string(job["request_file"], "")
  if path == "" {
    return {ok: false, error: "job " + safe_string(job["id"], "") + " has no request_file"}
  }
  if !harness.fs.exists(path) {
    return {ok: false, error: "request file not found for job " + safe_string(job["id"], "") + ": " + path}
  }
  let text = harness.fs.read_text(path)
  let actual_sha = sha256(text)
  let expected_sha = safe_string(job["request_file_sha256"], "")
  if expected_sha != "" && expected_sha != actual_sha {
    return {
      ok: false,
      error: "request file sha256 mismatch for job " + safe_string(job["id"], ""),
      expected_sha256: expected_sha,
      actual_sha256: actual_sha,
    }
  }
  return {ok: true, path: path, text: text, sha256: actual_sha}
}

fn __redacted_job_operation(job: dict, base_url: string, credential_env: string) -> dict {
  let submit = safe_dict(job["submit"])
  return {
    provider: safe_string(job["provider"], ""),
    operation: safe_string(submit["operation"], ""),
    base_url: base_url,
    credential_env: credential_env,
    auth: if credential_env == "" {
      "none"
    } else {
      credential_env + "=<redacted>"
    },
  }
}

fn __submission_job_base(harness: Harness, job: dict, payload: dict, dry_run: bool) -> dict {
  let provider = safe_string(job["provider"], "")
  let credential_env = __provider_auth_env(provider)
  return {
    id: safe_string(job["id"], ""),
    provider: provider,
    model: safe_string(job["model"], ""),
    workload: safe_string(job["workload"], ""),
    endpoint: safe_string(job["endpoint"], ""),
    tool_format: safe_string(job["tool_format"], ""),
    batch: safe_dict(job["batch"]),
    request_count: __safe_int(job["request_count"]) ?? 0,
    request_file: safe_string(job["request_file"], ""),
    request_file_sha256: safe_string(payload["sha256"], safe_string(job["request_file_sha256"], "")),
    request_format: safe_string(job["request_format"], ""),
    dry_run: dry_run,
    provider_operation: __redacted_job_operation(job, __provider_base_url(harness, provider), credential_env),
  }
}

fn __dry_run_submission(harness: Harness, job: dict, payload: dict) -> dict {
  return {
    ok: true,
    job: __submission_job_base(harness, job, payload, true)
      + {status: "ready", provider_batch_id: nil, response: nil},
    errors: [],
  }
}

fn __api_key_for_job(harness: Harness, provider: string) -> dict {
  let candidates = __provider_auth_env_candidates(provider)
  for env_name in candidates {
    let value = trim(harness.env.get_or(env_name, ""))
    if value != "" {
      return {ok: true, env: env_name, value: value}
    }
  }
  let primary = __provider_auth_env(provider)
  let aliases = join(candidates, ", ")
  return {
    ok: false,
    env: primary,
    error: "missing provider API key env " + primary + " (accepted: " + aliases + ")",
  }
}

fn __fireworks_account_id(harness: Harness) -> dict {
  let candidates = ["HARN_BATCH_FIREWORKS_ACCOUNT_ID", "FIREWORKS_ACCOUNT_ID"]
  for env_name in candidates {
    var value = trim(harness.env.get_or(env_name, ""))
    if starts_with(value, "accounts/") {
      value = value[9:len(value)]
    }
    if value != "" {
      return {ok: true, env: env_name, value: value}
    }
  }
  return {
    ok: false,
    env: "HARN_BATCH_FIREWORKS_ACCOUNT_ID",
    error: "missing Fireworks account id env HARN_BATCH_FIREWORKS_ACCOUNT_ID (accepted: HARN_BATCH_FIREWORKS_ACCOUNT_ID, FIREWORKS_ACCOUNT_ID)",
  }
}

fn __fireworks_account_id_or_placeholder(harness: Harness) -> string {
  let account = __fireworks_account_id(harness)
  if safe_bool(account["ok"], false) {
    return safe_string(account["value"], "")
  }
  return "{account_id}"
}

fn __fireworks_dataset_resource(account_id: string, dataset_id: string) -> string {
  let trimmed = trim(dataset_id)
  if starts_with(trimmed, "accounts/") {
    return trimmed
  }
  return "accounts/" + account_id + "/datasets/" + trimmed
}

fn __fireworks_dataset_id(job: dict, suffix: string) -> string {
  let seed = safe_string(job["id"], "")
    + ":"
    + safe_string(job["request_file_sha256"], "")
    + ":"
    + suffix
  return "harn-" + suffix + "-" + substring(sha256(seed), 0, 20)
}

fn __fireworks_batch_job_id(job: dict) -> string {
  let seed = safe_string(job["id"], "")
    + ":"
    + safe_string(job["model"], "")
    + ":"
    + safe_string(job["request_file_sha256"], "")
  return "harn-job-" + substring(sha256(seed), 0, 24)
}

fn __submit_anthropic(harness: Harness, job: dict, payload: dict) -> dict {
  let auth = __api_key_for_job(harness, "anthropic")
  let base = __provider_base_url(harness, "anthropic")
  let base_job = __submission_job_base(harness, job, payload, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let response = __http_result(
    harness,
    "POST",
    __join_url(base, "/v1/messages/batches"),
    {
      headers: {
        "x-api-key": safe_string(auth["value"], ""),
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
      },
      body: safe_string(payload["text"], ""),
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", response_status: response["status"], response: body},
      errors: ["anthropic batch submit failed for job " + safe_string(job["id"], "")],
    }
  }
  return {
    ok: true,
    job: base_job
      + {
      status: "submitted",
      provider_batch_id: safe_string(body["id"], ""),
      provider_status: body["processing_status"] ?? nil,
      results_url: body["results_url"] ?? nil,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __fireworks_create_dataset(
  harness: Harness,
  account_id: string,
  dataset_id: string,
  api_key: string,
) -> dict {
  let response = __http_result(
    harness,
    "POST",
    __join_url(__provider_base_url(harness, "fireworks"), "/accounts/" + account_id + "/datasets"),
    {
      headers: {Authorization: "Bearer " + api_key, "content-type": "application/json"},
      body: json_stringify({datasetId: dataset_id, dataset: {displayName: dataset_id, userUploaded: {}}}),
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  let status_code = __safe_int(response["status"]) ?? 0
  if safe_bool(response["ok"], false) || status_code == 409 {
    return {
      ok: true,
      dataset_id: dataset_id,
      dataset_resource: __fireworks_dataset_resource(account_id, dataset_id),
      status: response["status"],
      response: body,
      already_exists: status_code == 409,
    }
  }
  return {
    ok: false,
    error: "fireworks batch dataset create failed for " + dataset_id,
    status: response["status"],
    response: body,
  }
}

fn __fireworks_upload_dataset(
  harness: Harness,
  account_id: string,
  dataset_id: string,
  api_key: string,
  file_path: string,
) -> dict {
  let response = __http_result(
    harness,
    "POST",
    __join_url(
      __provider_base_url(harness, "fireworks"),
      "/accounts/" + account_id + "/datasets/" + dataset_id + ":upload",
    ),
    {
      headers: {Authorization: "Bearer " + api_key},
      multipart: [{name: "file", path: file_path, filename: basename(file_path), content_type: "application/jsonl"}],
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      error: "fireworks batch dataset upload failed for " + dataset_id,
      status: response["status"],
      response: body,
    }
  }
  return {ok: true, dataset_id: dataset_id, status: response["status"], response: body}
}

fn __submit_fireworks(harness: Harness, job: dict, payload: dict) -> dict {
  let auth = __api_key_for_job(harness, "fireworks")
  let account = __fireworks_account_id(harness)
  let base_job = __submission_job_base(harness, job, payload, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  if !safe_bool(account["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(account["error"], "")]}
  }
  let account_id = safe_string(account["value"], "")
  let input_dataset_id = __fireworks_dataset_id(job, "input")
  let output_dataset_id = __fireworks_dataset_id(job, "output")
  let input_dataset = __fireworks_create_dataset(harness, account_id, input_dataset_id, safe_string(auth["value"], ""))
  if !safe_bool(input_dataset["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", input_dataset: input_dataset},
      errors: [safe_string(input_dataset["error"], "")],
    }
  }
  let upload = __fireworks_upload_dataset(
    harness,
    account_id,
    input_dataset_id,
    safe_string(auth["value"], ""),
    safe_string(payload["path"], ""),
  )
  if !safe_bool(upload["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", input_dataset: input_dataset, upload: upload},
      errors: [safe_string(upload["error"], "")],
    }
  }
  let batch_job_id = __fireworks_batch_job_id(job)
  let create_body = {
    model: safe_string(job["model"], ""),
    inputDatasetId: __fireworks_dataset_resource(account_id, input_dataset_id),
    outputDatasetId: __fireworks_dataset_resource(account_id, output_dataset_id),
  }
  let response = __http_result(
    harness,
    "POST",
    __join_url(
      __provider_base_url(harness, "fireworks"),
      "/accounts/" + account_id + "/batchInferenceJobs?batchInferenceJobId=" + batch_job_id,
    ),
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], ""), "content-type": "application/json"},
      body: json_stringify(create_body),
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job
        + {
        status: "failed",
        input_dataset: input_dataset,
        upload: upload,
        response_status: response["status"],
        response: body,
      },
      errors: ["fireworks batch job create failed for job " + safe_string(job["id"], "")],
    }
  }
  let provider_batch_id = safe_string(body["name"], batch_job_id)
  return {
    ok: true,
    job: base_job
      + {
      status: "submitted",
      provider_batch_id: provider_batch_id,
      provider_status: body["state"] ?? nil,
      input_dataset_id: input_dataset_id,
      input_dataset_resource: __fireworks_dataset_resource(account_id, input_dataset_id),
      output_dataset_id: output_dataset_id,
      output_dataset_resource: __fireworks_dataset_resource(account_id, output_dataset_id),
      job_progress: body["jobProgress"] ?? nil,
      input_dataset: input_dataset,
      upload: upload,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __upload_jsonl_file(
  harness: Harness,
  provider: string,
  url: string,
  credential_env: string,
  api_key: string,
  file_path: string,
) -> dict {
  let purpose = if provider == "together" {
    "batch-api"
  } else {
    "batch"
  }
  let response = __http_result(
    harness,
    "POST",
    url,
    {
      headers: {Authorization: "Bearer " + api_key},
      multipart: [
        {name: "purpose", value: purpose},
        {name: "file", path: file_path, filename: basename(file_path), content_type: "application/jsonl"},
      ],
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  let file_id = safe_string(body["id"], "")
  if !safe_bool(response["ok"], false) || file_id == "" {
    return {
      ok: false,
      error: provider + " batch file upload failed using " + credential_env,
      status: response["status"],
      response: body,
    }
  }
  return {ok: true, file_id: file_id, status: response["status"], response: body}
}

fn __openai_compatible_create_body(provider: string, upload: dict, endpoint: string) -> dict {
  let body = {input_file_id: safe_string(upload["file_id"], ""), endpoint: endpoint}
  if provider == "together" {
    return body
  }
  return body + {completion_window: "24h"}
}

fn __openai_compatible_batch_body(provider: string, body: dict) -> dict {
  if provider == "together" {
    let wrapped = safe_dict(body["job"])
    if wrapped != {} {
      return wrapped
    }
  }
  return body
}

fn __submit_openai_compatible(harness: Harness, job: dict, payload: dict, provider: string) -> dict {
  let auth = __api_key_for_job(harness, provider)
  let base = __provider_base_url(harness, provider)
  let base_job = __submission_job_base(harness, job, payload, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let upload = __upload_jsonl_file(
    harness,
    provider,
    __join_url(base, "/files"),
    safe_string(auth["env"], ""),
    safe_string(auth["value"], ""),
    safe_string(payload["path"], ""),
  )
  if !safe_bool(upload["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", upload: upload},
      errors: [safe_string(upload["error"], "")],
    }
  }
  let create_body = __openai_compatible_create_body(
    provider,
    upload,
    safe_string(job["endpoint"], "/v1/chat/completions"),
  )
  let response = __http_result(
    harness,
    "POST",
    __join_url(base, "/batches"),
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], ""), "content-type": "application/json"},
      body: json_stringify(create_body),
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  let batch_body = __openai_compatible_batch_body(provider, body)
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job
        + {status: "failed", upload: upload, response_status: response["status"], response: body},
      errors: [provider + " batch create failed for job " + safe_string(job["id"], "")],
    }
  }
  return {
    ok: true,
    job: base_job
      + {
      status: "submitted",
      provider_batch_id: safe_string(batch_body["id"], ""),
      provider_status: batch_body["status"] ?? nil,
      input_file_id: safe_string(upload["file_id"], ""),
      output_file_id: batch_body["output_file_id"] ?? nil,
      error_file_id: batch_body["error_file_id"] ?? nil,
      upload: upload,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __submit_mistral(harness: Harness, job: dict, payload: dict) -> dict {
  let auth = __api_key_for_job(harness, "mistral")
  let base = __provider_base_url(harness, "mistral")
  let base_job = __submission_job_base(harness, job, payload, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let upload = __upload_jsonl_file(
    harness,
    "mistral",
    __join_url(base, "/files"),
    safe_string(auth["env"], ""),
    safe_string(auth["value"], ""),
    safe_string(payload["path"], ""),
  )
  if !safe_bool(upload["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", upload: upload},
      errors: [safe_string(upload["error"], "")],
    }
  }
  let create_body = {
    input_files: [safe_string(upload["file_id"], "")],
    endpoint: safe_string(job["endpoint"], "/v1/chat/completions"),
    model: safe_string(job["model"], ""),
    timeout_hours: 24,
  }
  let response = __http_result(
    harness,
    "POST",
    __join_url(base, "/batch/jobs"),
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], ""), "content-type": "application/json"},
      body: json_stringify(create_body),
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job
        + {status: "failed", upload: upload, response_status: response["status"], response: body},
      errors: ["mistral batch job create failed for job " + safe_string(job["id"], "")],
    }
  }
  return {
    ok: true,
    job: base_job
      + {
      status: "submitted",
      provider_batch_id: safe_string(body["id"], ""),
      provider_status: body["status"] ?? nil,
      input_files: body["input_files"] ?? [safe_string(upload["file_id"], "")],
      output_file: body["output_file"] ?? nil,
      error_file: body["error_file"] ?? nil,
      upload: upload,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __upload_xai_jsonl_file(harness: Harness, file_path: string, api_key: string) -> dict {
  let response = __http_result(
    harness,
    "POST",
    __join_url(__provider_base_url(harness, "xai"), "/files"),
    {
      headers: {Authorization: "Bearer " + api_key},
      multipart: [{name: "file", path: file_path, filename: basename(file_path), content_type: "application/jsonl"}],
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  let file_id = safe_string(body["id"], "")
  if !safe_bool(response["ok"], false) || file_id == "" {
    return {
      ok: false,
      error: "xai batch file upload failed using XAI_API_KEY",
      status: response["status"],
      response: body,
    }
  }
  return {ok: true, file_id: file_id, status: response["status"], response: body}
}

fn __submit_xai(harness: Harness, job: dict, payload: dict) -> dict {
  let auth = __api_key_for_job(harness, "xai")
  let base_job = __submission_job_base(harness, job, payload, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let upload = __upload_xai_jsonl_file(harness, safe_string(payload["path"], ""), safe_string(auth["value"], ""))
  if !safe_bool(upload["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", upload: upload},
      errors: [safe_string(upload["error"], "")],
    }
  }
  let create_body = {name: safe_string(job["id"], "harn-batch"), input_file_id: safe_string(upload["file_id"], "")}
  let response = __http_result(
    harness,
    "POST",
    __join_url(__provider_base_url(harness, "xai"), "/batches"),
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], ""), "content-type": "application/json"},
      body: json_stringify(create_body),
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job
        + {status: "failed", upload: upload, response_status: response["status"], response: body},
      errors: ["xai batch create failed for job " + safe_string(job["id"], "")],
    }
  }
  return {
    ok: true,
    job: base_job
      + {
      status: "submitted",
      provider_batch_id: safe_string(body["batch_id"], ""),
      provider_status: __xai_provider_status(body),
      input_file_id: safe_string(upload["file_id"], ""),
      results_url: __xai_results_url(harness, safe_string(body["batch_id"], ""), ""),
      upload: upload,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __header_value(headers: dict, name: string) -> string {
  let exact = headers[name] ?? headers[lowercase(name)] ?? nil
  if exact != nil {
    return trim(to_string(exact))
  }
  let target = lowercase(name)
  for key in headers.keys() {
    if lowercase(to_string(key)) == target {
      return trim(to_string(headers[key]))
    }
  }
  return ""
}

fn __gemini_model_path(model: string) -> string {
  let trimmed = trim(model)
  if starts_with(trimmed, "models/") {
    return trimmed
  }
  return "models/" + trimmed
}

fn __gemini_provider_status(body: dict) -> string {
  let metadata = safe_dict(body["metadata"])
  let response = safe_dict(body["response"])
  let status = body["state"] ?? metadata["state"] ?? response["state"] ?? nil
  let text = safe_string(status, "")
  if text != "" {
    return text
  }
  if safe_bool(body["done"], false) {
    if body["error"] != nil {
      return "JOB_STATE_FAILED"
    }
    return "JOB_STATE_SUCCEEDED"
  }
  return ""
}

fn __gemini_responses_file(body: dict) -> string {
  let response = safe_dict(body["response"])
  let output = safe_dict(body["output"])
  return safe_string(response["responsesFile"], safe_string(output["responsesFile"], ""))
}

fn __upload_gemini_jsonl_file(
  harness: Harness,
  job: dict,
  payload: dict,
  credential_env: string,
  api_key: string,
) -> dict {
  let base = __provider_base_url(harness, "gemini")
  let file_path = safe_string(payload["path"], "")
  let text = safe_string(payload["text"], "")
  let start = __http_result(
    harness,
    "POST",
    __join_url(base, "/upload/v1beta/files"),
    {
      headers: {
        "x-goog-api-key": api_key,
        "X-Goog-Upload-Protocol": "resumable",
        "X-Goog-Upload-Command": "start",
        "X-Goog-Upload-Header-Content-Length": to_string(len(text)),
        "X-Goog-Upload-Header-Content-Type": "application/jsonl",
        "Content-Type": "application/json",
      },
      body: json_stringify({file: {displayName: safe_string(job["id"], "harn-batch")}}),
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let upload_url = __header_value(safe_dict(start["headers"]), "x-goog-upload-url")
  if !safe_bool(start["ok"], false) || upload_url == "" {
    return {
      ok: false,
      error: "gemini batch file upload start failed using " + credential_env,
      status: start["status"],
      response: start["body"] ?? nil,
    }
  }
  let upload = __http_result(
    harness,
    "POST",
    upload_url,
    {
      headers: {
        "Content-Length": to_string(len(text)),
        "X-Goog-Upload-Offset": "0",
        "X-Goog-Upload-Command": "upload, finalize",
        "Content-Type": "application/jsonl",
      },
      body: text,
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(upload["body"])
  let file = safe_dict(body["file"])
  let file_name = safe_string(file["name"], "")
  if !safe_bool(upload["ok"], false) || file_name == "" {
    return {
      ok: false,
      error: "gemini batch file upload finalize failed using " + credential_env,
      status: upload["status"],
      response: body,
    }
  }
  return {
    ok: true,
    file_name: file_name,
    file_uri: file["uri"] ?? nil,
    path: file_path,
    start_status: start["status"],
    status: upload["status"],
    response: body,
  }
}

fn __submit_gemini(harness: Harness, job: dict, payload: dict) -> dict {
  let auth = __api_key_for_job(harness, "gemini")
  let base = __provider_base_url(harness, "gemini")
  let base_job = __submission_job_base(harness, job, payload, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let upload = __upload_gemini_jsonl_file(
    harness,
    job,
    payload,
    safe_string(auth["env"], ""),
    safe_string(auth["value"], ""),
  )
  if !safe_bool(upload["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", upload: upload},
      errors: [safe_string(upload["error"], "")],
    }
  }
  let create_body = {
    batch: {
      displayName: safe_string(job["id"], "harn-batch"),
      inputConfig: {fileName: safe_string(upload["file_name"], "")},
    },
  }
  let response = __http_result(
    harness,
    "POST",
    __join_url(
      base,
      "/v1beta/" + __gemini_model_path(safe_string(job["model"], "")) + ":batchGenerateContent",
    ),
    {
      headers: {"x-goog-api-key": safe_string(auth["value"], ""), "content-type": "application/json"},
      body: json_stringify(create_body),
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job
        + {status: "failed", upload: upload, response_status: response["status"], response: body},
      errors: ["gemini batch create failed for job " + safe_string(job["id"], "")],
    }
  }
  return {
    ok: true,
    job: base_job
      + {
      status: "submitted",
      provider_batch_id: safe_string(body["name"], ""),
      provider_status: __gemini_provider_status(body),
      input_file_name: safe_string(upload["file_name"], ""),
      responses_file: __gemini_responses_file(body),
      upload: upload,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __submit_live(harness: Harness, job: dict, payload: dict) -> dict {
  let provider = safe_string(job["provider"], "")
  let wire = safe_string(safe_dict(job["batch"])["wire_format"], "")
  let adapter = __live_batch_adapter_key(provider, wire)
  if adapter == "anthropic" {
    return __submit_anthropic(harness, job, payload)
  }
  if __is_openai_compatible_live_adapter(adapter) {
    return __submit_openai_compatible(harness, job, payload, adapter)
  }
  if adapter == "gemini" {
    return __submit_gemini(harness, job, payload)
  }
  if adapter == "fireworks" {
    return __submit_fireworks(harness, job, payload)
  }
  if adapter == "mistral" {
    return __submit_mistral(harness, job, payload)
  }
  if adapter == "xai" {
    return __submit_xai(harness, job, payload)
  }
  let base_job = __submission_job_base(harness, job, payload, false)
  return {
    ok: false,
    job: base_job + {status: "failed"},
    errors: [__live_batch_operation_error("submit", provider, wire)],
  }
}

fn __submit_one_job(harness: Harness, job: dict, dry_run: bool) -> dict {
  let payload = __request_file_payload(harness, job)
  if !safe_bool(payload["ok"], false) {
    return {
      ok: false,
      job: {
        id: safe_string(job["id"], ""),
        provider: safe_string(job["provider"], ""),
        model: safe_string(job["model"], ""),
        status: "failed",
        dry_run: dry_run,
      },
      errors: [safe_string(payload["error"], "request file validation failed")],
    }
  }
  if dry_run {
    return __dry_run_submission(harness, job, payload)
  }
  return __submit_live(harness, job, payload)
}

fn __submission_status(jobs: list, errors: list, dry_run: bool) -> string {
  if len(errors) > 0 {
    for job in jobs {
      if safe_string(safe_dict(job)["status"], "") == "submitted" {
        return "partial"
      }
    }
    return "failed"
  }
  if dry_run {
    return "dry_run"
  }
  return "submitted"
}

fn __count_jobs_with_status(jobs: list, status: string) -> int {
  var count = 0
  for job in jobs {
    if safe_string(safe_dict(job)["status"], "") == status {
      count = count + 1
    }
  }
  return count
}

fn __render_submit_human(harness: Harness, report: dict) {
  if !safe_bool(report["ok"], false) {
    harness.stdio.eprintln("Batch submit failed")
  } else if safe_bool(report["dry_run"], false) {
    harness.stdio.println("Batch submit dry run written: " + safe_string(report["out"], ""))
  } else {
    harness.stdio.println("Batch submission receipt written: " + safe_string(report["out"], ""))
  }
  harness.stdio.println("  jobs: " + safe_int_string(report["job_count"], "0"))
  harness.stdio.println("  submitted: " + safe_int_string(report["submitted_count"], "0"))
  harness.stdio.println("  ready: " + safe_int_string(report["ready_count"], "0"))
  harness.stdio.println("  failed: " + safe_int_string(report["failed_count"], "0"))
  if safe_string(report["receipt_sha256"], "") != "" {
    harness.stdio.println("  receipt sha256: " + safe_string(report["receipt_sha256"], ""))
  }
  print_list(harness, "errors", safe_list(report["errors"]))
  print_list(harness, "warnings", safe_list(report["warnings"]))
}

fn __render_submit_json(report: dict) -> string {
  if safe_bool(report["ok"], false) {
    return json_stringify_pretty(cli_json_envelope({schema_version: 1, ok: true, data: report}))
  }
  return json_stringify_pretty(
    cli_json_envelope(
      {
        schema_version: 1,
        ok: false,
        error: {
          code: "batch_submit_failed",
          message: "Prepared batch jobs could not be submitted.",
          details: report,
        },
        warnings: safe_list(report["warnings"]),
      },
    ),
  )
}

fn __run_submit(harness: Harness) -> int {
  let receipt_path = trim(harness.env.get_or("HARN_MODELS_BATCH_RECEIPT", ""))
  let out_path = trim(harness.env.get_or("HARN_MODELS_BATCH_SUBMIT_OUT", ""))
  let dry_run = __truthy_env(harness.env.get_or("HARN_MODELS_BATCH_DRY_RUN", "0"))
  let loaded = __read_prepare_receipt(harness, receipt_path)
  var errors = safe_list(loaded["errors"])
  var warnings = []
  if out_path == "" {
    errors = errors.push("--out is required")
  }
  let prepare = safe_dict(loaded["receipt"])
  var jobs = []
  if len(errors) == 0 {
    for raw_job in safe_list(prepare["jobs"]) {
      let submitted = __submit_one_job(harness, safe_dict(raw_job), dry_run)
      jobs = jobs.push(safe_dict(submitted["job"]))
      errors = errors + safe_list(submitted["errors"])
    }
  }
  let status = __submission_status(jobs, errors, dry_run)
  var submission = {
    schemaVersion: 1,
    kind: "harn.model_batch_submission_receipt",
    producer: "harn models batch submit",
    dryRun: dry_run,
    source: {
      path: receipt_path,
      sha256: safe_string(loaded["source_sha256"], ""),
      prepare_status: safe_string(prepare["status"], ""),
      manifest: safe_dict(prepare["manifest"]),
    },
    out: out_path,
    jobCount: len(jobs),
    submittedCount: __count_jobs_with_status(jobs, "submitted"),
    readyCount: __count_jobs_with_status(jobs, "ready"),
    failedCount: __count_jobs_with_status(jobs, "failed"),
    jobs: jobs,
    status: status,
    warnings: warnings,
    errors: errors,
  }
  var report = {
    schema_version: 1,
    ok: len(errors) == 0,
    dry_run: dry_run,
    out: out_path,
    source: submission["source"],
    status: status,
    job_count: len(jobs),
    submitted_count: __count_jobs_with_status(jobs, "submitted"),
    ready_count: __count_jobs_with_status(jobs, "ready"),
    failed_count: __count_jobs_with_status(jobs, "failed"),
    jobs: jobs,
    errors: errors,
    warnings: warnings,
  }
  if out_path != "" {
    __mkdirp(harness, dirname(out_path) ?? ".")
    let receipt_text = json_stringify_pretty(submission) + "\n"
    harness.fs.write_text(out_path, receipt_text)
    submission = submission + {receipt_sha256: sha256(receipt_text)}
    report = report + {receipt_sha256: sha256(receipt_text)}
  }
  if harness.env.get_or("HARN_OUTPUT_JSON", "0") == "1" {
    harness.stdio.println(__render_submit_json(report))
  } else {
    __render_submit_human(harness, report)
  }
  if len(errors) == 0 {
    return 0
  }
  return 1
}

fn __provider_batch_status(raw_status) -> string {
  let status = lowercase(trim(to_string(raw_status ?? "")))
  if status == "" {
    return "unknown"
  }
  if status == "ready" {
    return "ready"
  }
  if status == "completed"
    || status == "succeeded"
    || status == "success"
    || status == "ended"
    || status == "job_state_succeeded"
    || status == "batch_state_succeeded" {
    return "completed"
  }
  if status == "failed"
    || status == "error"
    || status == "errored"
    || status == "expired"
    || status == "cancelled"
    || status == "canceled"
    || status == "job_state_failed"
    || status == "job_state_cancelled"
    || status == "job_state_canceled"
    || status == "job_state_expired"
    || status == "batch_state_failed"
    || status == "batch_state_cancelled"
    || status == "batch_state_canceled"
    || status == "batch_state_expired" {
    return "failed"
  }
  return "running"
}

fn __status_job_base(job: dict, dry_run: bool) -> dict {
  return {
    id: safe_string(job["id"], ""),
    provider: safe_string(job["provider"], ""),
    model: safe_string(job["model"], ""),
    workload: safe_string(job["workload"], ""),
    endpoint: safe_string(job["endpoint"], ""),
    tool_format: safe_string(job["tool_format"], ""),
    batch: safe_dict(job["batch"]),
    request_count: __safe_int(job["request_count"]) ?? 0,
    provider_batch_id: safe_string(job["provider_batch_id"], ""),
    dry_run: dry_run,
  }
}

fn __cached_status_job(job: dict, dry_run: bool) -> dict {
  let provider_status = job["provider_status"] ?? job["status"] ?? nil
  return __status_job_base(job, dry_run)
    + {
    status: __provider_batch_status(provider_status),
    provider_status: provider_status,
    status_checked: false,
    output_file_id: job["output_file_id"] ?? nil,
    error_file_id: job["error_file_id"] ?? nil,
    output_file: job["output_file"] ?? nil,
    error_file: job["error_file"] ?? nil,
    results_url: job["results_url"] ?? nil,
    responses_file: job["responses_file"] ?? nil,
    input_dataset_id: job["input_dataset_id"] ?? nil,
    input_dataset_resource: job["input_dataset_resource"] ?? nil,
    output_dataset_id: job["output_dataset_id"] ?? nil,
    output_dataset_resource: job["output_dataset_resource"] ?? nil,
    job_progress: job["job_progress"] ?? nil,
    response: job["response"] ?? nil,
  }
}

fn __poll_openai_compatible_status(harness: Harness, job: dict, provider: string) -> dict {
  let auth = __api_key_for_job(harness, provider)
  let base_job = __status_job_base(job, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let batch_id = safe_string(job["provider_batch_id"], "")
  let response = __http_result(
    harness,
    "GET",
    __join_url(__provider_base_url(harness, provider), "/batches/" + batch_id),
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], ""), "content-type": "application/json"},
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", response_status: response["status"], response: body},
      errors: [provider + " batch status failed for job " + safe_string(job["id"], "")],
    }
  }
  let provider_status = body["status"] ?? nil
  return {
    ok: true,
    job: base_job
      + {
      status: __provider_batch_status(provider_status),
      provider_status: provider_status,
      status_checked: true,
      output_file_id: body["output_file_id"] ?? nil,
      error_file_id: body["error_file_id"] ?? nil,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __poll_anthropic_status(harness: Harness, job: dict) -> dict {
  let auth = __api_key_for_job(harness, "anthropic")
  let base_job = __status_job_base(job, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let batch_id = safe_string(job["provider_batch_id"], "")
  let response = __http_result(
    harness,
    "GET",
    __join_url(__provider_base_url(harness, "anthropic"), "/v1/messages/batches/" + batch_id),
    {
      headers: {"x-api-key": safe_string(auth["value"], ""), "anthropic-version": "2023-06-01"},
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", response_status: response["status"], response: body},
      errors: ["anthropic batch status failed for job " + safe_string(job["id"], "")],
    }
  }
  let provider_status = body["processing_status"] ?? nil
  return {
    ok: true,
    job: base_job
      + {
      status: __provider_batch_status(provider_status),
      provider_status: provider_status,
      status_checked: true,
      results_url: body["results_url"] ?? nil,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __poll_mistral_status(harness: Harness, job: dict) -> dict {
  let auth = __api_key_for_job(harness, "mistral")
  let base_job = __status_job_base(job, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let batch_id = safe_string(job["provider_batch_id"], "")
  let response = __http_result(
    harness,
    "GET",
    __join_url(__provider_base_url(harness, "mistral"), "/batch/jobs/" + batch_id),
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], "")},
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", response_status: response["status"], response: body},
      errors: ["mistral batch status failed for job " + safe_string(job["id"], "")],
    }
  }
  let provider_status = body["status"] ?? nil
  return {
    ok: true,
    job: base_job
      + {
      status: __provider_batch_status(provider_status),
      provider_status: provider_status,
      status_checked: true,
      output_file: body["output_file"] ?? nil,
      error_file: body["error_file"] ?? nil,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __fireworks_batch_id_path(provider_batch_id: string) -> string {
  let trimmed = trim(provider_batch_id)
  if starts_with(trimmed, "accounts/") {
    let parts = split(trimmed, "/batchInferenceJobs/")
    if len(parts) > 1 {
      return safe_string(parts[1], trimmed)
    }
  }
  return trimmed
}

fn __poll_fireworks_status(harness: Harness, job: dict) -> dict {
  let auth = __api_key_for_job(harness, "fireworks")
  let account = __fireworks_account_id(harness)
  let base_job = __status_job_base(job, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  if !safe_bool(account["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(account["error"], "")]}
  }
  let account_id = safe_string(account["value"], "")
  let batch_id = __fireworks_batch_id_path(safe_string(job["provider_batch_id"], ""))
  let response = __http_result(
    harness,
    "GET",
    __join_url(
      __provider_base_url(harness, "fireworks"),
      "/accounts/" + account_id + "/batchInferenceJobs/" + batch_id,
    ),
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], ""), "content-type": "application/json"},
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", response_status: response["status"], response: body},
      errors: ["fireworks batch status failed for job " + safe_string(job["id"], "")],
    }
  }
  let provider_status = body["state"] ?? nil
  let output_dataset = safe_string(
    body["outputDatasetId"],
    safe_string(job["output_dataset_resource"], safe_string(job["output_dataset_id"], "")),
  )
  return {
    ok: true,
    job: base_job
      + {
      status: __provider_batch_status(provider_status),
      provider_status: provider_status,
      status_checked: true,
      input_dataset_id: job["input_dataset_id"] ?? nil,
      input_dataset_resource: job["input_dataset_resource"] ?? nil,
      output_dataset_id: if starts_with(output_dataset, "accounts/") {
        safe_string(job["output_dataset_id"], output_dataset)
      } else {
        output_dataset
      },
      output_dataset_resource: if starts_with(output_dataset, "accounts/") {
        output_dataset
      } else {
        __fireworks_dataset_resource(account_id, output_dataset)
      },
      job_progress: body["jobProgress"] ?? nil,
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __xai_provider_status(body: dict) -> string {
  let state = safe_dict(body["state"])
  let pending = __safe_int(state["num_pending"])
  let requests = __safe_int(state["num_requests"])
  let errors = __safe_int(state["num_error"]) ?? 0
  let cancelled = __safe_int(state["num_cancelled"]) ?? 0
  if pending != nil {
    if pending > 0 {
      return "running"
    }
    if errors > 0 || cancelled > 0 {
      return "failed"
    }
    if requests == nil || requests > 0 {
      return "completed"
    }
    return "ready"
  }
  return safe_string(body["status"], "")
}

fn __xai_results_url(harness: Harness, batch_id: string, pagination_token: string) -> string {
  var url = __join_url(__provider_base_url(harness, "xai"), "/batches/" + batch_id + "/results?limit=100")
  if trim(pagination_token) != "" {
    url = url + "&pagination_token=" + trim(pagination_token)
  }
  return url
}

fn __poll_xai_status(harness: Harness, job: dict) -> dict {
  let auth = __api_key_for_job(harness, "xai")
  let base_job = __status_job_base(job, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let batch_id = safe_string(job["provider_batch_id"], "")
  let response = __http_result(
    harness,
    "GET",
    __join_url(__provider_base_url(harness, "xai"), "/batches/" + batch_id),
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], ""), "content-type": "application/json"},
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", response_status: response["status"], response: body},
      errors: ["xai batch status failed for job " + safe_string(job["id"], "")],
    }
  }
  let provider_status = __xai_provider_status(body)
  return {
    ok: true,
    job: base_job
      + {
      status: __provider_batch_status(provider_status),
      provider_status: provider_status,
      status_checked: true,
      results_url: __xai_results_url(harness, batch_id, ""),
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __poll_gemini_status(harness: Harness, job: dict) -> dict {
  let auth = __api_key_for_job(harness, "gemini")
  let base_job = __status_job_base(job, false)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, job: base_job + {status: "failed"}, errors: [safe_string(auth["error"], "")]}
  }
  let batch_id = safe_string(job["provider_batch_id"], "")
  let response = __http_result(
    harness,
    "GET",
    __join_url(__provider_base_url(harness, "gemini"), "/v1beta/" + batch_id),
    {
      headers: {"x-goog-api-key": safe_string(auth["value"], ""), "content-type": "application/json"},
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      job: base_job + {status: "failed", response_status: response["status"], response: body},
      errors: ["gemini batch status failed for job " + safe_string(job["id"], "")],
    }
  }
  let provider_status = __gemini_provider_status(body)
  return {
    ok: true,
    job: base_job
      + {
      status: __provider_batch_status(provider_status),
      provider_status: provider_status,
      status_checked: true,
      input_file_name: job["input_file_name"] ?? nil,
      responses_file: __gemini_responses_file(body),
      response_status: response["status"],
      response: body,
    },
    errors: [],
  }
}

fn __status_one_job(harness: Harness, job: dict, dry_run: bool) -> dict {
  if dry_run {
    return {ok: true, job: __cached_status_job(job, true), errors: []}
  }
  let provider = safe_string(job["provider"], "")
  let wire = safe_string(safe_dict(job["batch"])["wire_format"], "")
  let provider_batch_id = safe_string(job["provider_batch_id"], "")
  if provider_batch_id == "" {
    return {
      ok: false,
      job: __cached_status_job(job, false) + {status: "failed"},
      errors: [
        "job "
          + safe_string(job["id"], "")
          + " has no provider_batch_id; submit live first or use --dry-run",
      ],
    }
  }
  let adapter = __live_batch_adapter_key(provider, wire)
  if __is_openai_compatible_live_adapter(adapter) {
    return __poll_openai_compatible_status(harness, job, adapter)
  }
  if adapter == "gemini" {
    return __poll_gemini_status(harness, job)
  }
  if adapter == "fireworks" {
    return __poll_fireworks_status(harness, job)
  }
  if adapter == "anthropic" {
    return __poll_anthropic_status(harness, job)
  }
  if adapter == "mistral" {
    return __poll_mistral_status(harness, job)
  }
  if adapter == "xai" {
    return __poll_xai_status(harness, job)
  }
  return {
    ok: false,
    job: __cached_status_job(job, false) + {status: "failed"},
    errors: [__live_batch_operation_error("status", provider, wire)],
  }
}

fn __aggregate_status(jobs: list, errors: list, dry_run: bool) -> string {
  if len(errors) > 0 {
    for job in jobs {
      let status = safe_string(safe_dict(job)["status"], "")
      if status == "completed" || status == "running" || status == "ready" || status == "unknown" {
        return "partial"
      }
    }
    return "failed"
  }
  if dry_run {
    return "dry_run"
  }
  let failed = __count_jobs_with_status(jobs, "failed")
  let running = __count_jobs_with_status(jobs, "running")
  let ready = __count_jobs_with_status(jobs, "ready")
  let unknown = __count_jobs_with_status(jobs, "unknown")
  if failed > 0 {
    return "partial"
  }
  if running > 0 || ready > 0 || unknown > 0 {
    return "running"
  }
  return "completed"
}

fn __render_status_human(harness: Harness, report: dict) {
  if !safe_bool(report["ok"], false) {
    harness.stdio.eprintln("Batch status failed")
  } else if safe_bool(report["dry_run"], false) {
    harness.stdio.println("Batch status dry run written: " + safe_string(report["out"], ""))
  } else {
    harness.stdio.println("Batch status receipt written: " + safe_string(report["out"], ""))
  }
  harness.stdio.println("  jobs: " + safe_int_string(report["job_count"], "0"))
  harness.stdio.println("  completed: " + safe_int_string(report["completed_count"], "0"))
  harness.stdio.println("  running: " + safe_int_string(report["running_count"], "0"))
  harness.stdio.println("  ready: " + safe_int_string(report["ready_count"], "0"))
  harness.stdio.println("  failed: " + safe_int_string(report["failed_count"], "0"))
  harness.stdio.println("  unknown: " + safe_int_string(report["unknown_count"], "0"))
  if safe_string(report["receipt_sha256"], "") != "" {
    harness.stdio.println("  receipt sha256: " + safe_string(report["receipt_sha256"], ""))
  }
  print_list(harness, "errors", safe_list(report["errors"]))
  print_list(harness, "warnings", safe_list(report["warnings"]))
}

fn __render_status_json(report: dict) -> string {
  if safe_bool(report["ok"], false) {
    return json_stringify_pretty(cli_json_envelope({schema_version: 1, ok: true, data: report}))
  }
  return json_stringify_pretty(
    cli_json_envelope(
      {
        schema_version: 1,
        ok: false,
        error: {code: "batch_status_failed", message: "Submitted batch jobs could not be polled.", details: report},
        warnings: safe_list(report["warnings"]),
      },
    ),
  )
}

fn __run_status(harness: Harness) -> int {
  let submission_path = trim(harness.env.get_or("HARN_MODELS_BATCH_SUBMISSION", ""))
  let out_path = trim(harness.env.get_or("HARN_MODELS_BATCH_STATUS_OUT", ""))
  let dry_run = __truthy_env(harness.env.get_or("HARN_MODELS_BATCH_DRY_RUN", "0"))
  let loaded = __read_submission_receipt(harness, submission_path)
  var errors = safe_list(loaded["errors"])
  var warnings = []
  if out_path == "" {
    errors = errors.push("--out is required")
  }
  let submission = safe_dict(loaded["receipt"])
  var jobs = []
  if len(errors) == 0 {
    for raw_job in safe_list(submission["jobs"]) {
      let checked = __status_one_job(harness, safe_dict(raw_job), dry_run)
      jobs = jobs.push(safe_dict(checked["job"]))
      errors = errors + safe_list(checked["errors"])
    }
  }
  let status = __aggregate_status(jobs, errors, dry_run)
  var receipt = {
    schemaVersion: 1,
    kind: "harn.model_batch_status_receipt",
    producer: "harn models batch status",
    dryRun: dry_run,
    source: {
      path: submission_path,
      sha256: safe_string(loaded["source_sha256"], ""),
      submission_status: safe_string(submission["status"], ""),
      prepare_source: safe_dict(safe_dict(submission["source"])["manifest"]),
    },
    out: out_path,
    jobCount: len(jobs),
    completedCount: __count_jobs_with_status(jobs, "completed"),
    runningCount: __count_jobs_with_status(jobs, "running"),
    readyCount: __count_jobs_with_status(jobs, "ready"),
    failedCount: __count_jobs_with_status(jobs, "failed"),
    unknownCount: __count_jobs_with_status(jobs, "unknown"),
    jobs: jobs,
    status: status,
    warnings: warnings,
    errors: errors,
  }
  var report = {
    schema_version: 1,
    ok: len(errors) == 0,
    dry_run: dry_run,
    out: out_path,
    source: receipt["source"],
    status: status,
    job_count: len(jobs),
    completed_count: __count_jobs_with_status(jobs, "completed"),
    running_count: __count_jobs_with_status(jobs, "running"),
    ready_count: __count_jobs_with_status(jobs, "ready"),
    failed_count: __count_jobs_with_status(jobs, "failed"),
    unknown_count: __count_jobs_with_status(jobs, "unknown"),
    jobs: jobs,
    errors: errors,
    warnings: warnings,
  }
  if out_path != "" {
    __mkdirp(harness, dirname(out_path) ?? ".")
    let receipt_text = json_stringify_pretty(receipt) + "\n"
    harness.fs.write_text(out_path, receipt_text)
    receipt = receipt + {receipt_sha256: sha256(receipt_text)}
    report = report + {receipt_sha256: sha256(receipt_text)}
  }
  if harness.env.get_or("HARN_OUTPUT_JSON", "0") == "1" {
    harness.stdio.println(__render_status_json(report))
  } else {
    __render_status_human(harness, report)
  }
  if len(errors) == 0 {
    return 0
  }
  return 1
}

fn __download_status(job: dict, dry_run: bool) -> dict {
  return {
    id: safe_string(job["id"], ""),
    provider: safe_string(job["provider"], ""),
    model: safe_string(job["model"], ""),
    workload: safe_string(job["workload"], ""),
    endpoint: safe_string(job["endpoint"], ""),
    tool_format: safe_string(job["tool_format"], ""),
    batch: safe_dict(job["batch"]),
    provider_batch_id: safe_string(job["provider_batch_id"], ""),
    provider_status: job["provider_status"] ?? nil,
    responses_file: job["responses_file"] ?? nil,
    input_dataset_id: job["input_dataset_id"] ?? nil,
    input_dataset_resource: job["input_dataset_resource"] ?? nil,
    output_dataset_id: job["output_dataset_id"] ?? nil,
    output_dataset_resource: job["output_dataset_resource"] ?? nil,
    job_progress: job["job_progress"] ?? nil,
    source_status: safe_string(job["status"], ""),
    dry_run: dry_run,
  }
}

fn __download_slug(value: string) -> string {
  var slug = lowercase(trim(value))
  slug = regex_replace("[^A-Za-z0-9_.-]+", "-", slug) ?? slug
  slug = regex_replace("^-+|-+$", "", slug) ?? slug
  if slug == "" {
    return "job"
  }
  return slug
}

fn __download_artifact_path(out_dir: string, job: dict, label: string) -> string {
  let job_id = __download_slug(safe_string(job["id"], safe_string(job["provider_batch_id"], "job")))
  return path_join(out_dir, job_id + "." + label + ".jsonl")
}

fn __redact_download_url(url: string) -> string {
  return regex_replace("[?#].*$", "", url) ?? url
}

fn __download_operation(
  provider: string,
  method: string,
  url: string,
  credential_env: string,
  out_path: string,
) -> dict {
  return {
    provider: provider,
    operation: method + " " + __redact_download_url(url),
    credential_env: credential_env,
    auth: if credential_env == "" {
      "none"
    } else {
      credential_env + "=<redacted>"
    },
    out: out_path,
  }
}

fn __download_artifact_record(
  label: string,
  handle: string,
  path: string,
  operation: dict,
  body_text: string,
  dry_run: bool,
) -> dict {
  var artifact = {label: label, handle: handle, path: path, operation: operation, dry_run: dry_run}
  if !dry_run {
    artifact = artifact + {sha256: sha256(body_text), bytes: len(body_text)}
  }
  return artifact
}

fn __download_text_file(
  harness: Harness,
  provider: string,
  url: string,
  headers: dict,
  credential_env: string,
  label: string,
  handle: string,
  out_path: string,
  max_bytes: int,
) -> dict {
  let operation = __download_operation(provider, "GET", url, credential_env, out_path)
  let response = __http_result(
    harness,
    "GET",
    url,
    {headers: headers, timeout_ms: 120000, max_response_bytes: max_bytes},
  )
  let body_text = safe_string(response["body_text"], "")
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      artifact: __download_artifact_record(label, handle, out_path, operation, "", false)
        + {response_status: response["status"], response: response["body"] ?? nil},
      errors: [provider + " batch result download failed for " + label + " handle " + handle],
    }
  }
  __mkdirp(harness, dirname(out_path) ?? ".")
  harness.fs.write_text(out_path, body_text)
  return {
    ok: true,
    artifact: __download_artifact_record(label, handle, out_path, operation, body_text, false)
      + {response_status: response["status"]},
    errors: [],
  }
}

fn __dry_run_download_artifact(
  provider: string,
  url: string,
  credential_env: string,
  label: string,
  handle: string,
  out_path: string,
) -> dict {
  return {
    ok: true,
    artifact: __download_artifact_record(
      label,
      handle,
      out_path,
      __download_operation(provider, "GET", url, credential_env, out_path),
      "",
      true,
    ),
    errors: [],
  }
}

fn __fireworks_download_endpoint_url(
  harness: Harness,
  account_id: string,
  output_dataset_id: string,
) -> string {
  return __join_url(
    __provider_base_url(harness, "fireworks"),
    "/" + __fireworks_dataset_resource(account_id, output_dataset_id) + ":getDownloadEndpoint",
  )
}

fn __fireworks_artifact_label(filename: string) -> string {
  let slug = __download_slug(basename(filename))
  if contains(slug, "error") {
    return "error"
  }
  if contains(slug, "result") || contains(slug, "response") {
    return "results"
  }
  return slug
}

fn __download_fireworks_results(
  harness: Harness,
  job: dict,
  out_dir: string,
  dry_run: bool,
  max_bytes: int,
) -> dict {
  var output_dataset_id = safe_string(job["output_dataset_id"], "")
  if output_dataset_id == "" {
    output_dataset_id = safe_string(job["output_dataset_resource"], "")
  }
  if output_dataset_id == "" {
    return {ok: true, artifacts: [], errors: []}
  }
  let account_id = if dry_run {
    __fireworks_account_id_or_placeholder(harness)
  } else {
    let account = __fireworks_account_id(harness)
    if !safe_bool(account["ok"], false) {
      return {ok: false, artifacts: [], errors: [safe_string(account["error"], "")]}
    }
    safe_string(account["value"], "")
  }
  let endpoint_url = __fireworks_download_endpoint_url(harness, account_id, output_dataset_id)
  let auth_env = __provider_auth_env("fireworks")
  if dry_run {
    let out_path = __download_artifact_path(out_dir, job, "download-endpoint")
    let artifact = __download_artifact_record(
      "download-endpoint",
      output_dataset_id,
      out_path,
      __download_operation("fireworks", "GET", endpoint_url, auth_env, out_path),
      "",
      true,
    )
    return {ok: true, artifacts: [artifact], errors: []}
  }
  let auth = __api_key_for_job(harness, "fireworks")
  if !safe_bool(auth["ok"], false) {
    return {ok: false, artifacts: [], errors: [safe_string(auth["error"], "")]}
  }
  let response = __http_result(
    harness,
    "GET",
    endpoint_url,
    {
      headers: {Authorization: "Bearer " + safe_string(auth["value"], "")},
      body: "{}",
      timeout_ms: 120000,
      max_response_bytes: 1048576,
    },
  )
  let body = safe_dict(response["body"])
  if !safe_bool(response["ok"], false) {
    return {
      ok: false,
      artifacts: [],
      errors: ["fireworks batch dataset download endpoint failed for " + output_dataset_id],
      response_status: response["status"],
      response: body,
    }
  }
  var artifacts = []
  var errors = []
  let signed_urls = safe_dict(body["filenameToSignedUrls"])
  for filename in signed_urls.keys().sort() {
    let signed_url = safe_string(signed_urls[filename], "")
    if signed_url == "" {
      continue
    }
    let label = __fireworks_artifact_label(filename)
    let out_path = __download_artifact_path(out_dir, job, label)
    let downloaded = __download_text_file(harness, "fireworks", signed_url, {}, "", label, filename, out_path, max_bytes)
    errors = errors + safe_list(downloaded["errors"])
    let artifact = downloaded["artifact"] ?? nil
    if artifact != nil {
      artifacts = artifacts.push(artifact)
    }
  }
  if len(artifacts) == 0 && len(errors) == 0 {
    errors = errors.push("fireworks batch download endpoint returned no signed result files")
  }
  return {ok: len(errors) == 0, artifacts: artifacts, errors: errors, response: body}
}

fn __download_openai_file(
  harness: Harness,
  job: dict,
  provider: string,
  label: string,
  file_id,
  out_dir: string,
  dry_run: bool,
  max_bytes: int,
) -> dict {
  let handle = safe_string(file_id, "")
  if handle == "" {
    return {ok: true, artifact: nil, errors: []}
  }
  let auth_env = __provider_auth_env(provider)
  let out_path = __download_artifact_path(out_dir, job, label)
  let url = __join_url(__provider_base_url(harness, provider), "/files/" + handle + "/content")
  if dry_run {
    return __dry_run_download_artifact(provider, url, auth_env, label, handle, out_path)
  }
  let auth = __api_key_for_job(harness, provider)
  if !safe_bool(auth["ok"], false) {
    return {ok: false, artifact: nil, errors: [safe_string(auth["error"], "")]}
  }
  return __download_text_file(
    harness,
    provider,
    url,
    {Authorization: "Bearer " + safe_string(auth["value"], "")},
    auth_env,
    label,
    handle,
    out_path,
    max_bytes,
  )
}

fn __download_mistral_file(
  harness: Harness,
  job: dict,
  label: string,
  file_id,
  out_dir: string,
  dry_run: bool,
  max_bytes: int,
) -> dict {
  let handle = safe_string(file_id, "")
  if handle == "" {
    return {ok: true, artifact: nil, errors: []}
  }
  let auth_env = __provider_auth_env("mistral")
  let out_path = __download_artifact_path(out_dir, job, label)
  let url = __join_url(__provider_base_url(harness, "mistral"), "/files/" + handle + "/content")
  if dry_run {
    return __dry_run_download_artifact("mistral", url, auth_env, label, handle, out_path)
  }
  let auth = __api_key_for_job(harness, "mistral")
  if !safe_bool(auth["ok"], false) {
    return {ok: false, artifact: nil, errors: [safe_string(auth["error"], "")]}
  }
  return __download_text_file(
    harness,
    "mistral",
    url,
    {Authorization: "Bearer " + safe_string(auth["value"], "")},
    auth_env,
    label,
    handle,
    out_path,
    max_bytes,
  )
}

fn __download_anthropic_results(
  harness: Harness,
  job: dict,
  out_dir: string,
  dry_run: bool,
  max_bytes: int,
) -> dict {
  let url = safe_string(job["results_url"], "")
  if url == "" {
    return {ok: true, artifact: nil, errors: []}
  }
  let auth_env = __provider_auth_env("anthropic")
  let handle = __redact_download_url(url)
  let out_path = __download_artifact_path(out_dir, job, "results")
  if dry_run {
    return __dry_run_download_artifact("anthropic", url, auth_env, "results", handle, out_path)
  }
  let auth = __api_key_for_job(harness, "anthropic")
  if !safe_bool(auth["ok"], false) {
    return {ok: false, artifact: nil, errors: [safe_string(auth["error"], "")]}
  }
  return __download_text_file(
    harness,
    "anthropic",
    url,
    {"x-api-key": safe_string(auth["value"], ""), "anthropic-version": "2023-06-01"},
    auth_env,
    "results",
    handle,
    out_path,
    max_bytes,
  )
}

fn __download_xai_results(
  harness: Harness,
  job: dict,
  out_dir: string,
  dry_run: bool,
  max_bytes: int,
) -> dict {
  let batch_id = safe_string(job["provider_batch_id"], "")
  if batch_id == "" {
    return {ok: true, artifact: nil, errors: []}
  }
  let auth_env = __provider_auth_env("xai")
  let out_path = __download_artifact_path(out_dir, job, "results")
  let first_url = __xai_results_url(harness, batch_id, "")
  if dry_run {
    return __dry_run_download_artifact("xai", first_url, auth_env, "results", batch_id, out_path)
  }
  let auth = __api_key_for_job(harness, "xai")
  if !safe_bool(auth["ok"], false) {
    return {ok: false, artifact: nil, errors: [safe_string(auth["error"], "")]}
  }
  var token = ""
  var lines = []
  var byte_count = 0
  var page_count = 0
  while true {
    let url = __xai_results_url(harness, batch_id, token)
    let response = __http_result(
      harness,
      "GET",
      url,
      {
        headers: {Authorization: "Bearer " + safe_string(auth["value"], ""), "content-type": "application/json"},
        timeout_ms: 120000,
        max_response_bytes: max_bytes,
      },
    )
    let body = safe_dict(response["body"])
    if !safe_bool(response["ok"], false) {
      return {
        ok: false,
        artifact: __download_artifact_record(
          "results",
          batch_id,
          out_path,
          __download_operation("xai", "GET", url, auth_env, out_path),
          "",
          false,
        )
          + {response_status: response["status"], response: body},
        errors: ["xai batch result download failed for batch " + batch_id],
      }
    }
    page_count = page_count + 1
    for result in safe_list(body["results"]) {
      let line = json_stringify(result)
      byte_count = byte_count + len(line) + 1
      if byte_count > max_bytes {
        return {
          ok: false,
          artifact: nil,
          errors: ["xai batch result download exceeded --max-bytes for batch " + batch_id],
        }
      }
      lines = lines.push(line)
    }
    token = safe_string(body["pagination_token"], "")
    if token == "" {
      break
    }
  }
  let body_text = if len(lines) == 0 {
    ""
  } else {
    join(lines, "\n") + "\n"
  }
  __mkdirp(harness, dirname(out_path) ?? ".")
  harness.fs.write_text(out_path, body_text)
  return {
    ok: true,
    artifact: __download_artifact_record(
      "results",
      batch_id,
      out_path,
      __download_operation("xai", "GET", first_url, auth_env, out_path),
      body_text,
      false,
    )
      + {page_count: page_count, result_count: len(lines)},
    errors: [],
  }
}

fn __download_gemini_responses(
  harness: Harness,
  job: dict,
  out_dir: string,
  dry_run: bool,
  max_bytes: int,
) -> dict {
  let handle = safe_string(job["responses_file"], "")
  if handle == "" {
    return {ok: true, artifact: nil, errors: []}
  }
  let auth_env = __provider_auth_env("gemini")
  let out_path = __download_artifact_path(out_dir, job, "responses")
  let url = __join_url(
    __provider_base_url(harness, "gemini"),
    "/download/v1beta/" + handle + ":download?alt=media",
  )
  if dry_run {
    return __dry_run_download_artifact("gemini", url, auth_env, "responses", handle, out_path)
  }
  let auth = __api_key_for_job(harness, "gemini")
  if !safe_bool(auth["ok"], false) {
    return {ok: false, artifact: nil, errors: [safe_string(auth["error"], "")]}
  }
  return __download_text_file(
    harness,
    "gemini",
    url,
    {"x-goog-api-key": safe_string(auth["value"], "")},
    auth_env,
    "responses",
    handle,
    out_path,
    max_bytes,
  )
}

fn __append_download_result(state: dict, result: dict) -> dict {
  var artifacts = safe_list(state["artifacts"])
  var errors = safe_list(state["errors"]) + safe_list(result["errors"])
  let artifact = result["artifact"] ?? nil
  if artifact != nil {
    artifacts = artifacts.push(artifact)
  }
  return {artifacts: artifacts, errors: errors}
}

fn __download_openai_compatible_files(
  harness: Harness,
  job: dict,
  provider: string,
  out_dir: string,
  dry_run: bool,
  max_bytes: int,
) -> dict {
  var state = {artifacts: [], errors: []}
  state = __append_download_result(
    state,
    __download_openai_file(
      harness,
      job,
      provider,
      "output",
      job["output_file_id"],
      out_dir,
      dry_run,
      max_bytes,
    ),
  )
  state = __append_download_result(
    state,
    __download_openai_file(
      harness,
      job,
      provider,
      "error",
      job["error_file_id"],
      out_dir,
      dry_run,
      max_bytes,
    ),
  )
  return state
}

fn __download_one_job(harness: Harness, job: dict, out_dir: string, dry_run: bool, max_bytes: int) -> dict {
  let provider = safe_string(job["provider"], "")
  let wire = safe_string(safe_dict(job["batch"])["wire_format"], "")
  let source_status = safe_string(job["status"], "")
  let job_report = __download_status(job, dry_run)
  if source_status != "completed" {
    return {
      ok: false,
      job: job_report + {status: "not_ready", artifacts: []},
      errors: [
        "job " + safe_string(job["id"], "") + " is " + source_status
          + "; run status after completion",
      ],
    }
  }
  var state = {artifacts: [], errors: []}
  let adapter = __live_batch_adapter_key(provider, wire)
  if __is_openai_compatible_live_adapter(adapter) {
    state = __download_openai_compatible_files(harness, job, adapter, out_dir, dry_run, max_bytes)
  } else if adapter == "gemini" {
    state = __append_download_result(
      state,
      __download_gemini_responses(harness, job, out_dir, dry_run, max_bytes),
    )
  } else if adapter == "fireworks" {
    let fireworks = __download_fireworks_results(harness, job, out_dir, dry_run, max_bytes)
    state = {artifacts: safe_list(fireworks["artifacts"]), errors: safe_list(fireworks["errors"])}
  } else if adapter == "anthropic" {
    state = __append_download_result(
      state,
      __download_anthropic_results(harness, job, out_dir, dry_run, max_bytes),
    )
  } else if adapter == "mistral" {
    state = __append_download_result(
      state,
      __download_mistral_file(harness, job, "output", job["output_file"], out_dir, dry_run, max_bytes),
    )
    state = __append_download_result(
      state,
      __download_mistral_file(harness, job, "error", job["error_file"], out_dir, dry_run, max_bytes),
    )
  } else if adapter == "xai" {
    state = __append_download_result(state, __download_xai_results(harness, job, out_dir, dry_run, max_bytes))
  } else {
    state = {artifacts: [], errors: [__live_batch_operation_error("download", provider, wire)]}
  }
  if len(safe_list(state["artifacts"])) == 0 && len(safe_list(state["errors"])) == 0 {
    state = {artifacts: [], errors: ["job " + safe_string(job["id"], "") + " has no provider result handles"]}
  }
  return {
    ok: len(safe_list(state["errors"])) == 0,
    job: job_report
      + {
      status: if len(safe_list(state["errors"])) == 0 {
        if dry_run {
          "ready"
        } else {
          "downloaded"
        }
      } else {
        "failed"
      },
      artifacts: safe_list(state["artifacts"]),
    },
    errors: safe_list(state["errors"]),
  }
}

fn __artifact_count(jobs: list) -> int {
  var count = 0
  for job in jobs {
    count = count + len(safe_list(safe_dict(job)["artifacts"]))
  }
  return count
}

fn __download_status_summary(jobs: list, errors: list, dry_run: bool) -> string {
  if len(errors) > 0 {
    if __artifact_count(jobs) > 0 {
      return "partial"
    }
    return "failed"
  }
  if dry_run {
    return "dry_run"
  }
  return "downloaded"
}

fn __render_download_human(harness: Harness, report: dict) {
  if !safe_bool(report["ok"], false) {
    harness.stdio.eprintln("Batch download failed")
  } else if safe_bool(report["dry_run"], false) {
    harness.stdio.println("Batch download dry run written: " + safe_string(report["receipt"], ""))
  } else {
    harness.stdio.println("Batch results receipt written: " + safe_string(report["receipt"], ""))
  }
  harness.stdio.println("  jobs: " + safe_int_string(report["job_count"], "0"))
  harness.stdio.println("  downloaded: " + safe_int_string(report["downloaded_count"], "0"))
  harness.stdio.println("  ready: " + safe_int_string(report["ready_count"], "0"))
  harness.stdio.println("  failed: " + safe_int_string(report["failed_count"], "0"))
  harness.stdio.println("  artifacts: " + safe_int_string(report["artifact_count"], "0"))
  if safe_string(report["receipt_sha256"], "") != "" {
    harness.stdio.println("  receipt sha256: " + safe_string(report["receipt_sha256"], ""))
  }
  print_list(harness, "errors", safe_list(report["errors"]))
  print_list(harness, "warnings", safe_list(report["warnings"]))
}

fn __render_download_json(report: dict) -> string {
  if safe_bool(report["ok"], false) {
    return json_stringify_pretty(cli_json_envelope({schema_version: 1, ok: true, data: report}))
  }
  return json_stringify_pretty(
    cli_json_envelope(
      {
        schema_version: 1,
        ok: false,
        error: {
          code: "batch_download_failed",
          message: "Completed batch result files could not be downloaded.",
          details: report,
        },
        warnings: safe_list(report["warnings"]),
      },
    ),
  )
}

fn __run_download(harness: Harness) -> int {
  let status_path = trim(harness.env.get_or("HARN_MODELS_BATCH_STATUS", ""))
  let out_dir = trim(harness.env.get_or("HARN_MODELS_BATCH_RESULTS_OUT_DIR", ""))
  let dry_run = __truthy_env(harness.env.get_or("HARN_MODELS_BATCH_DRY_RUN", "0"))
  let max_bytes = __filter_int(harness.env.get_or("HARN_MODELS_BATCH_MAX_DOWNLOAD_BYTES", "268435456")) ?? 268435456
  let loaded = __read_status_receipt(harness, status_path)
  var errors = safe_list(loaded["errors"])
  var warnings = []
  if out_dir == "" {
    errors = errors.push("--out-dir is required")
  }
  let status_receipt = safe_dict(loaded["receipt"])
  var jobs = []
  if len(errors) == 0 {
    for raw_job in safe_list(status_receipt["jobs"]) {
      let downloaded = __download_one_job(harness, safe_dict(raw_job), out_dir, dry_run, max_bytes)
      jobs = jobs.push(safe_dict(downloaded["job"]))
      errors = errors + safe_list(downloaded["errors"])
    }
  }
  let receipt_path = if out_dir == "" {
    ""
  } else {
    path_join(out_dir, "receipt.json")
  }
  let status = __download_status_summary(jobs, errors, dry_run)
  var receipt = {
    schemaVersion: 1,
    kind: "harn.model_batch_results_receipt",
    producer: "harn models batch download",
    dryRun: dry_run,
    source: {
      path: status_path,
      sha256: safe_string(loaded["source_sha256"], ""),
      status: safe_string(status_receipt["status"], ""),
      submission_source: safe_dict(status_receipt["source"]),
    },
    outDir: out_dir,
    receipt: receipt_path,
    maxBytes: max_bytes,
    jobCount: len(jobs),
    downloadedCount: __count_jobs_with_status(jobs, "downloaded"),
    readyCount: __count_jobs_with_status(jobs, "ready"),
    failedCount: __count_jobs_with_status(jobs, "failed") + __count_jobs_with_status(jobs, "not_ready"),
    artifactCount: __artifact_count(jobs),
    jobs: jobs,
    status: status,
    warnings: warnings,
    errors: errors,
  }
  var report = {
    schema_version: 1,
    ok: len(errors) == 0,
    dry_run: dry_run,
    receipt: receipt_path,
    out_dir: out_dir,
    source: receipt["source"],
    status: status,
    max_bytes: max_bytes,
    job_count: len(jobs),
    downloaded_count: __count_jobs_with_status(jobs, "downloaded"),
    ready_count: __count_jobs_with_status(jobs, "ready"),
    failed_count: __count_jobs_with_status(jobs, "failed") + __count_jobs_with_status(jobs, "not_ready"),
    artifact_count: __artifact_count(jobs),
    jobs: jobs,
    errors: errors,
    warnings: warnings,
  }
  if receipt_path != "" {
    __mkdirp(harness, out_dir)
    let receipt_text = json_stringify_pretty(receipt) + "\n"
    harness.fs.write_text(receipt_path, receipt_text)
    receipt = receipt + {receipt_sha256: sha256(receipt_text)}
    report = report + {receipt_sha256: sha256(receipt_text)}
  }
  if harness.env.get_or("HARN_OUTPUT_JSON", "0") == "1" {
    harness.stdio.println(__render_download_json(report))
  } else {
    __render_download_human(harness, report)
  }
  if len(errors) == 0 {
    return 0
  }
  return 1
}

fn main(harness: Harness) -> int {
  let provider_filter = trim(harness.env.get_or("HARN_MODELS_BATCH_PROVIDER", ""))
  let model_filter = trim(harness.env.get_or("HARN_MODELS_BATCH_MODEL", ""))
  let workload_raw = trim(harness.env.get_or("HARN_MODELS_BATCH_WORKLOAD", "eval"))
  let workload = __normalize_workload(workload_raw)
  let min_discount = __filter_int(harness.env.get_or("HARN_MODELS_BATCH_MIN_DISCOUNT_PERCENT", ""))
  let max_turnaround = __filter_int(harness.env.get_or("HARN_MODELS_BATCH_MAX_TURNAROUND_HOURS", ""))
  let catalog = harness.llm.catalog()
  let models = __collect(catalog, provider_filter, model_filter, workload, min_discount, max_turnaround)
  var warnings = []
  if workload != workload_raw && workload_raw != "" {
    warnings = warnings.push("unknown workload '" + workload_raw + "'; using generic policy")
  }
  if min_discount == nil && harness.env.get_or("HARN_MODELS_BATCH_MIN_DISCOUNT_PERCENT", "") != "" {
    warnings = warnings.push("ignored non-integer --min-discount-percent")
  }
  if max_turnaround == nil && harness.env.get_or("HARN_MODELS_BATCH_MAX_TURNAROUND_HOURS", "") != "" {
    warnings = warnings.push("ignored non-integer --max-turnaround-hours")
  }
  let mode = harness.env.get_or("HARN_MODELS_BATCH_MODE", "plan")
  if mode == "prepare" {
    return __run_prepare(harness)
  }
  if mode == "submit" {
    return __run_submit(harness)
  }
  if mode == "status" {
    return __run_status(harness)
  }
  if mode == "download" {
    return __run_download(harness)
  }
  if mode == "manifest" {
    return __run_manifest(
      harness,
      catalog,
      provider_filter,
      model_filter,
      workload,
      min_discount,
      max_turnaround,
      warnings,
    )
  }
  let report = {
    schema_version: 1,
    workload: workload,
    filters: {
      provider: provider_filter,
      model: model_filter,
      min_discount_percent: min_discount,
      max_turnaround_hours: max_turnaround,
    },
    models: models,
    constraints: __constraints(workload),
    warnings: warnings,
  }
  if harness.env.get_or("HARN_OUTPUT_JSON", "0") == "1" {
    harness.stdio.println(__render_json(report))
  } else {
    __render_human(harness, report)
  }
  return 0
}