/**
* `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, or prepare.
* 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_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 __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 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.",
)
}
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,
},
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")
return " "
+ safe_string(model["id"], "")
+ " "
+ wire
+ "/"
+ input
+ " discount "
+ discount
+ " turnaround "
+ turnaround
}
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 __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)}
}
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 == "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>"]},
}
}
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"
}
return uppercase(provider) + "_API_KEY"
}
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"),
),
)
}
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 env_name = __provider_auth_env(provider)
let value = trim(harness.env.get_or(env_name, ""))
if value == "" {
return {ok: false, env: env_name, error: "missing provider API key env " + env_name}
}
return {ok: true, env: env_name, value: value}
}
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 __upload_jsonl_file(
harness: Harness,
provider: string,
url: string,
credential_env: string,
api_key: string,
file_path: string,
) -> dict {
let response = __http_result(
harness,
"POST",
url,
{
headers: {Authorization: "Bearer " + api_key},
multipart: [
{name: "purpose", value: "batch"},
{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 __submit_openai(harness: Harness, job: dict, payload: dict) -> dict {
let auth = __api_key_for_job(harness, "openai")
let base = __provider_base_url(harness, "openai")
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,
"openai",
__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_file_id: safe_string(upload["file_id"], ""),
endpoint: safe_string(job["endpoint"], "/v1/chat/completions"),
completion_window: "24h",
}
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"])
if !safe_bool(response["ok"], false) {
return {
ok: false,
job: base_job
+ {status: "failed", upload: upload, response_status: response["status"], response: body},
errors: ["openai batch 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_file_id: safe_string(upload["file_id"], ""),
output_file_id: body["output_file_id"] ?? nil,
error_file_id: 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 __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"], "")
if provider == "anthropic" && wire == "anthropic_messages" {
return __submit_anthropic(harness, job, payload)
}
if provider == "openai" && wire == "openai" {
return __submit_openai(harness, job, payload)
}
if provider == "mistral" && wire == "mistral" {
return __submit_mistral(harness, job, payload)
}
let base_job = __submission_job_base(harness, job, payload, false)
return {
ok: false,
job: base_job + {status: "failed"},
errors: [
"live batch submit is not implemented for provider="
+ provider
+ " wire="
+ wire
+ "; use --dry-run or add a provider adapter",
],
}
}
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" {
return "completed"
}
if status == "failed"
|| status == "error"
|| status == "errored"
|| status == "expired"
|| status == "cancelled"
|| status == "canceled" {
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,
response: job["response"] ?? nil,
}
}
fn __poll_openai_status(harness: Harness, job: dict) -> dict {
let auth = __api_key_for_job(harness, "openai")
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, "openai"), "/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: ["openai 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 __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",
],
}
}
if provider == "openai" && wire == "openai" {
return __poll_openai_status(harness, job)
}
if provider == "anthropic" && wire == "anthropic_messages" {
return __poll_anthropic_status(harness, job)
}
if provider == "mistral" && wire == "mistral" {
return __poll_mistral_status(harness, job)
}
return {
ok: false,
job: __cached_status_job(job, false) + {status: "failed"},
errors: [
"live batch status is not implemented for provider="
+ provider
+ " wire="
+ wire
+ "; use --dry-run or add a provider adapter",
],
}
}
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 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 == "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
}