Diagnostic LLM APIs
The batch finished and nobody ever collected the output
The batch ran, the model answered ninety thousand times, the invoice includes every one of those answers, and the file holding them was deleted a month later without ever being opened. There is no incident for this. The batch object still sits in the list saying completed, which is true, and pointing at an id that no longer resolves, which is also true. The only party that ever knew the results were wanted was a process that stopped running in March.
Join the batch list against your own record of what was consumed. On OpenAI, GET /v1/batches?limit=100 for every completed batch's output_file_id, then GET /v1/files?purpose=batch_output to see which of those ids still exist. On Anthropic, GET /v1/messages/batches?limit=1000 with a workspace key and read ended_at, results_url and archived_at.
Neither API records whether you downloaded anything. There is no read receipt on a file object and no flag on a batch. So the second half of the join has to come from your side: a list of batch ids your consumer has actually processed. If you cannot produce that list, that absence is the finding, and it is a bigger one than anything in the output.
Both clocks are short and they are anchored differently. An OpenAI batch output file is deleted thirty days after the batch is complete. Claude batch results are available for twenty-nine days after the batch was created — created, not ended — after which you can still see the batch but archived_at is set and the results cannot be downloaded by anyone.
Three states people usually name as separate problems are verdicts inside this one reading. A batch created and never polled is a non-terminal object older than its own 24 hour window. A batch whose results were never fetched is a terminal one with an artifact and no ledger entry. A batch that ended and was never claimed is the same thing said a third way. One script, four verdicts, one repair.
The problem in plain words
The Batch API is fire and poll. Creation returns an object, processing happens somewhere else, and the results appear as a file id on the batch when it is done. Nothing pushes. In the default flow there is no callback and no notification, so the entire mechanism by which results reach your database is a loop somebody wrote, running in a process somebody deployed, which is exactly the kind of thing that gets switched off during an incident and not switched back on.
When that loop stops, everything keeps working. Batches are still submitted, because submission is usually a different job. They still run and are still billed. The output files are still written. The only thing that stops is collection, and collection is the only step with no error path, because not reading a file is not an event.
Then the retention clock runs out and the loss becomes permanent. Up to that point this is a morning's work: list the batches, download what you missed, insert the rows. After it, the only way to get those answers back is to submit the same rows again and pay for them again, which on a bulk enrichment job is the difference between an inconvenience and a line item.
The variant that catches people who are otherwise careful is the batch that never reached a terminal state at all. A batch that has been validating or in_progress for three days is not slow; both providers cap processing at 24 hours, so an object still open past that window is a stale record rather than a running job. It is the same failure — nothing polled — wearing a different status.
Why it happens
This note and the error-file note read the same batch, join the same ledger, and mean opposite things. Say plainly which is which. error_file_id is the list of rows that failed: reading it tells you which requests to retry and why, and losing it costs you the knowledge of what is missing, not the work. output_file_id is the work itself: reading it is the reason the batch was run at all, and losing it costs you every token you paid for. If your downstream table is short and error_file_id is null, this note owns it. If error_file_id is non-null and unread, the other one does. Both live in the same completion handler and the handler needs both assertions, so run the two scripts together and fix them in one change.
The retention anchors differ and getting them wrong sorts the queue backwards. OpenAI's guide is explicit that the output file is deleted thirty days after the batch is complete, so the countdown starts at completed_at; where the file object carries its own expires_at, that value is authoritative and the script prefers it. Anthropic's clock is twenty-nine days from created_at, and the documentation says so twice, including a note that it is creation and not the end of processing. A batch that took twenty hours to run has twenty hours less runway than you would guess.
An output_file_id that is absent from the file list is already lost, and that is a different finding from one that is expiring. The batch object outlives the file, so a completed batch pointing at an id that no longer appears under purpose=batch_output is a permanent hole: no read call recovers it and the repair is to re-run and re-pay. Something with four days left is a task for this afternoon. Printing them in one list, sorted by urgency, is the only presentation that leads to the right order of work.
There is no read receipt anywhere in either API, so the ledger is an input rather than a derivation. File objects have no last_accessed_at, batches have no consumed flag, and Anthropic's archived_at tells you the results are gone rather than that they were taken. A script that guessed which batches had been consumed would be wrong in the direction that produces false calm, so this one takes the list from you and treats an empty one as its own verdict.
A non-terminal batch older than 24 hours is a stale object, not a slow one. Anthropic publishes expires_at as exactly 24 hours after creation, and OpenAI's completion window can only be set to 24h. So a batch still open past that plus a little slack has not been polled since the process that made it went away, and it is counted here rather than in a note of its own, because the repair is the same reconciler.
The fix, as a flow
This is the mirror of the published error file note. That one reads the id holding the rows that failed; this one reads the id holding the work itself, which is why losing it costs money rather than knowledge. Neither API records whether you downloaded anything, so the second half of the join is your own ingest ledger, and the clock runs from completion on one provider and creation on the other.
How to fix it
Write down what your consumer has actually processed
One batch id per line, from your own ingest table. This is the half of the join the API cannot supply, and producing it is often the moment the problem becomes obvious. No ledger at all is a valid input and produces its own verdict: everything terminal is reported as unclaimed, which is the honest reading when nothing records consumption.
List the batches on whichever providers you use
OpenAI: page /v1/batches on after. Anthropic: page /v1/messages/batches on after_id, up to 1000 per page. The script takes either key or both and names the providers it looked at, so a one-provider run is never mistaken for a clean two-provider one.
Resolve the OpenAI output files
GET /v1/files?purpose=batch_output and index by id. An output_file_id missing from that index is a file that has already been deleted. A present one carries bytes, created_at and, where the platform sets it, expires_at, which the script prefers over its own arithmetic.
Sort by the clock, not by size
Thirty days from completion on OpenAI, twenty-nine days from creation on Anthropic. The output leads with what is expiring soonest, because a small batch with two days left outranks a large one with three weeks. --warn-days sets the threshold; the default is five.
Download, then make the assertion
The repair is printed. Sweep the unclaimed batches, persist their output into your own store keyed by batch id, and then add the assertion that closes this for good: a batch is not done until its output has been archived. On future OpenAI submissions, output_expires_after lets you set a shorter retention deliberately rather than discovering the default one.
How to check it worked
Re-run with the ledger regenerated. Everything terminal should be either claimed or newly downloaded, and the count of unclaimed batches should be zero rather than small. Then run it weekly: the value is not the first run, which finds the backlog, but the second and third, which catch the reconciler stopping again.
python3 batch_output_unclaimed_audit.py --ledger ingested_batches.txt --warn-days 5
# anthropic msgbatch_01Rf expiring 41,880 succeeded, 2 days left
# openai batch_68d114 expiring 88,300 completed, 4 days left
# openai batch_68a002 lost output_file_id file_2b7c no longer exists
# anthropic msgbatch_01Kb lost archived_at set, 12,400 succeeded, gone
# openai batch_68e551 unclaimed 90,000 completed, 21 days left
# openai batch_68f9c1 stalled in_progress for 62 h, past the 24 h window
# batch-output-expiring 2 batch(es) hold results that expire within 5 days,
# 2 are already unrecoverable, 1 was never claimed and
# 1 never reached a terminal state
# measured: status, the result artifact and the retention clock from the batch
# lists, and file existence from GET /v1/files?purpose=batch_output
# inferred: that an id absent from your ledger was never consumed, since
# neither API records whether a result was downloaded
# repair: download the expiring outputs today and persist them keyed by batch
# id. After the clock runs out no read call recovers them.
# repair: the lost ones must be re-run and re-paid. Nothing in either API can
# return results after archival.
# 6 finding(s)The full code
Three paged GETs and eight pure functions. read_ledger, which turns your ingest record into a set and tolerates comments and duplicates; parse_time, which absorbs OpenAI's unix integers and Anthropic's RFC 3339 strings once; file_index, which makes the output-file lookup an index rather than a call per batch; openai_deadline, which prefers the file object's own expires_at over thirty days from completed_at and says which it used; openai_rows and anthropic_rows, which turn two very different object shapes into one row with a state and a number of days on it; verdict, which ranks the categories by whether you can still act on them rather than by count; and repair_lines, which separates the batches you can still save from the ones that have to be paid for twice.
"""Find batch results that were paid for and never collected, on both providers.
Read only. Three GET endpoints: /v1/batches and /v1/files on OpenAI,
/v1/messages/batches on Anthropic. Nothing is downloaded, deleted or re-run.
Neither API records whether you read a result. File objects have no
last_accessed_at, batches have no consumed flag, and Anthropic's archived_at
means the results are gone rather than that they were taken. So the ledger of
what your consumer has processed is an input, and an empty one is a verdict.
The clocks are anchored differently and it matters. An OpenAI batch output file
is deleted 30 days after the batch is complete; where the file object carries
its own expires_at, that is authoritative and is used instead. Claude batch
results are available for 29 days after the batch was created, not after it
ended.
This is the mirror of the error-file note. That one reads error_file_id, the
list of rows that failed. This one reads output_file_id and results_url, which
are the work itself.
"""
import argparse
import calendar
import datetime
import logging
import os
import sys
import time
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("batch_output_unclaimed_audit")
OPENAI_BATCHES_URL = "https://api.openai.com/v1/batches"
OPENAI_FILES_URL = "https://api.openai.com/v1/files"
ANTHROPIC_BATCHES_URL = "https://api.anthropic.com/v1/messages/batches"
# 30 days from completion on OpenAI, 29 days from creation on Anthropic. The
# anchors are different and sorting by the wrong one puts the queue backwards.
OPENAI_RETENTION = 30 * 86400
ANTHROPIC_RETENTION = 29 * 86400
# Both providers cap batch processing at 24 hours. Past that plus a little
# slack, a non-terminal batch is a stale object rather than a slow job.
OPEN_WINDOW = 24 * 3600
GRACE = 2 * 3600
OPENAI_TERMINAL = ("completed", "failed", "expired", "cancelled")
FINDINGS = ("batch-output-expiring", "batch-output-lost",
"batch-output-unclaimed", "batch-never-polled")
def read_ledger(text):
"""Set of batch ids your consumer has processed. Pure. Comments dropped."""
out = set()
for raw in (text or "").replace(",", "\n").splitlines():
line = raw.strip()
if line and not line.startswith("#"):
out.add(line)
return out
def parse_time(value):
"""Epoch seconds from a unix number or an RFC 3339 string. Pure.
A string with no offset is read as UTC rather than as whatever the machine
running the audit happens to be set to.
"""
if value is None or value == "" or isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return int(value)
text = str(value).strip()
if text[-1:] in ("Z", "z"):
text = text[:-1] + "+00:00"
try:
stamp = datetime.datetime.fromisoformat(text)
except (ValueError, TypeError):
return None
if stamp.tzinfo is None:
return calendar.timegm(stamp.timetuple())
return int(stamp.timestamp())
def file_index(files):
"""{file_id: file object}. Pure. One index instead of a call per batch."""
out = {}
for f in files or []:
if isinstance(f, dict) and f.get("id"):
out[str(f["id"])] = f
return out
def openai_deadline(batch, file_obj):
"""(epoch, source) for when this output disappears. Pure.
The file's own expires_at is the platform speaking. Everything else is
arithmetic on the documented 30 days from completion, and the source is
reported so nobody argues with a number whose provenance is invisible.
"""
stamp = parse_time((file_obj or {}).get("expires_at"))
if stamp:
return (stamp, "expires_at")
completed = parse_time((batch or {}).get("completed_at"))
if completed:
return (completed + OPENAI_RETENTION, "completed_at + 30d")
created = parse_time((batch or {}).get("created_at"))
if created:
return (created + OPENAI_RETENTION, "created_at + 30d")
return (None, "unknown")
def days_left(deadline, now):
"""Whole days until the deadline. Pure. None when there is no deadline."""
if deadline is None:
return None
return int((deadline - now) // 86400)
def openai_rows(batches, index, ledger, now, warn_days):
"""One row per OpenAI batch worth reporting. Pure."""
rows = []
for b in batches or []:
status = str((b or {}).get("status") or "")
ident = str(b.get("id"))
created = parse_time(b.get("created_at"))
if status not in OPENAI_TERMINAL:
if created is not None and now - created > OPEN_WINDOW + GRACE:
rows.append({"provider": "openai", "id": ident, "state": "stalled",
"done": 0, "days": None,
"detail": "%s for %d h, past the 24 h window"
% (status, (now - created) // 3600)})
continue
if status != "completed":
continue
counts = b.get("request_counts") or {}
done = int(counts.get("completed") or 0)
artifact = b.get("output_file_id")
if not artifact:
continue
if str(artifact) not in (index or {}):
rows.append({"provider": "openai", "id": ident, "state": "lost",
"done": done, "days": None,
"detail": "output_file_id %s no longer exists" % artifact})
continue
deadline, source = openai_deadline(b, index[str(artifact)])
left = days_left(deadline, now)
if left is not None and left <= warn_days:
state = "expiring"
detail = "%d completed, %d days left (%s)" % (done, max(0, left), source)
elif ident in (ledger or set()):
state = "claimed"
detail = "%d completed, in the ingest ledger" % done
else:
state = "unclaimed"
detail = "%d completed, %s days left" % (
done, "unknown" if left is None else max(0, left))
rows.append({"provider": "openai", "id": ident, "state": state,
"done": done, "days": left, "detail": detail})
return rows
def anthropic_rows(batches, ledger, now, warn_days):
"""One row per Claude batch worth reporting. Pure."""
rows = []
for b in batches or []:
ident = str((b or {}).get("id"))
status = str(b.get("processing_status") or "")
created = parse_time(b.get("created_at"))
counts = b.get("request_counts") or {}
done = int(counts.get("succeeded") or 0)
if status != "ended":
if created is not None and now - created > OPEN_WINDOW + GRACE:
rows.append({"provider": "anthropic", "id": ident,
"state": "stalled", "done": done, "days": None,
"detail": "%s for %d h, past the 24 h window"
% (status or "unknown",
(now - created) // 3600)})
continue
if done <= 0:
continue
if b.get("archived_at"):
rows.append({"provider": "anthropic", "id": ident, "state": "lost",
"done": done, "days": None,
"detail": "archived_at set, %d succeeded, gone" % done})
continue
left = days_left(created + ANTHROPIC_RETENTION, now) \
if created is not None else None
if left is not None and left <= warn_days:
state = "expiring"
detail = "%d succeeded, %d days left (created_at + 29d)" % (
done, max(0, left))
elif ident in (ledger or set()):
state = "claimed"
detail = "%d succeeded, in the ingest ledger" % done
else:
state = "unclaimed"
detail = "%d succeeded, %s days left" % (
done, "unknown" if left is None else max(0, left))
rows.append({"provider": "anthropic", "id": ident, "state": state,
"done": done, "days": left, "detail": detail})
return rows
def by_urgency(rows):
"""Rows ordered by what you can still act on. Pure. Stable within a state."""
rank = {"expiring": 0, "lost": 1, "unclaimed": 2, "stalled": 3, "claimed": 4}
return sorted(rows or [], key=lambda r: (rank.get(r.get("state"), 9),
99999 if r.get("days") is None
else r["days"], r.get("id") or ""))
def counts_by_state(rows):
"""{state: n}. Pure."""
out = {}
for row in rows or []:
out[row.get("state")] = out.get(row.get("state"), 0) + 1
return out
def verdict(rows, ledger, warn_days):
"""Grade the run. Pure. Returns (state, detail).
Ranked by whether you can still do something about it. An expiring result
is the only category with a deadline you can beat, so it wins even when the
unclaimed pile is larger.
"""
rows = list(rows or [])
if not rows:
return ("batch-output-clean",
"every batch on the providers checked is either open inside its "
"window or terminal with its output accounted for")
c = counts_by_state(rows)
parts = []
if c.get("lost"):
parts.append("%d are already unrecoverable" % c["lost"])
if c.get("unclaimed"):
parts.append("%d were never claimed" % c["unclaimed"])
if c.get("stalled"):
parts.append("%d never reached a terminal state" % c["stalled"])
tail = (", " + ", ".join(parts)) if parts else ""
if c.get("expiring"):
return ("batch-output-expiring",
"%d batch(es) hold results that expire within %d days%s"
% (c["expiring"], warn_days, tail))
if c.get("lost"):
return ("batch-output-lost",
"%d batch(es) hold results that are already gone and can only "
"be recovered by re-running them%s"
% (c["lost"], (", " + ", ".join(parts[1:])) if parts[1:] else ""))
if c.get("unclaimed"):
detail = ("%d batch(es) ended with results nothing has collected"
% c["unclaimed"])
if not ledger:
detail += (", and no ingest ledger was supplied, so every terminal "
"batch counts as unclaimed")
return ("batch-output-unclaimed", detail)
if c.get("stalled"):
return ("batch-never-polled",
"%d batch(es) have been open longer than the 24 hour window, "
"which means nothing has polled them" % c["stalled"])
return ("batch-output-clean",
"all %d terminal batch(es) are in the ingest ledger with runway "
"left on the clock" % len(rows))
def repair_lines(state, rows, ledger):
"""The repair for one verdict. Pure. Printed, never performed."""
rows = list(rows or [])
c = counts_by_state(rows)
if state == "batch-output-clean":
return ["nothing outstanding. Keep the assertion that a batch is not "
"done until its output has been archived into your own store."]
lines = []
if c.get("expiring"):
lines.append("download the expiring outputs today and persist them "
"keyed by batch id. After the clock runs out no read call "
"recovers them.")
if c.get("lost"):
lines.append("the lost ones must be re-run and re-paid. Nothing in "
"either API can return results after the retention window "
"closes.")
if c.get("unclaimed"):
lines.append("sweep the unclaimed batches: list, diff against your "
"ledger, download, and key the rows by custom_id, which "
"is the only join available since results are not "
"returned in request order.")
if c.get("stalled"):
lines.append("a batch open past 24 hours is a stale object rather than "
"a slow job. Poll every id you create to a terminal state, "
"and record the id at creation time so orphans are "
"identifiable.")
if not ledger:
lines.append("no ingest ledger was supplied, so nothing could be "
"confirmed as consumed. Record every batch id your "
"consumer processes: neither API offers a read receipt.")
lines.append("run the error-file audit alongside this one. That note reads "
"error_file_id, the list of rows that failed; this one reads "
"the work itself. Both assertions belong in the same batch "
"completion handler.")
return lines
def get_json(url, headers, params=None, timeout=30):
"""One GET. Returns (payload, error). Read only, always."""
try:
r = requests.get(url, headers=headers, params=params or {}, timeout=timeout)
except requests.RequestException as exc:
return (None, "request failed: %s" % exc)
if r.status_code != 200:
return (None, "HTTP %d %s" % (r.status_code, (r.text or "")[:160]))
try:
return (r.json(), None)
except ValueError:
return (None, "response was not JSON")
def page(url, headers, params, max_pages, cursor="after"):
"""Follow a cursor. Returns (rows, error). GETs only."""
rows, token = [], None
for _ in range(max(1, max_pages)):
query = dict(params or {})
if token:
query[cursor] = token
payload, err = get_json(url, headers, query)
if err:
return (rows, err)
data = payload.get("data") or []
rows.extend(data)
if not payload.get("has_more") or not data:
break
token = payload.get("last_id") or data[-1].get("id")
if not token:
break
return (rows, None)
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--ledger", help="file of batch ids your consumer processed")
ap.add_argument("--warn-days", type=int, default=5,
help="days of runway below which a result counts as expiring")
ap.add_argument("--max-pages", type=int, default=20)
args = ap.parse_args()
openai_key = os.environ.get("OPENAI_API_KEY")
anthropic_key = os.environ.get("ANTHROPIC_API_KEY")
if not openai_key and not anthropic_key:
log.error("set OPENAI_API_KEY (project key, Read Only) or "
"ANTHROPIC_API_KEY (workspace key), or both")
return 2
raw = os.environ.get("BATCH_INGEST_LEDGER", "")
if args.ledger:
try:
with open(args.ledger, "r", encoding="utf-8") as fh:
raw = fh.read()
except OSError as exc:
log.error("could not read %s: %s", args.ledger, exc)
return 2
ledger = read_ledger(raw)
now = int(time.time())
rows, checked = [], []
if openai_key:
checked.append("openai")
headers = {"Authorization": "Bearer %s" % openai_key}
batches, err = page(OPENAI_BATCHES_URL, headers, {"limit": 100},
args.max_pages)
if err:
log.warning("openai batch list stopped early: %s", err)
files, ferr = page(OPENAI_FILES_URL, headers,
{"limit": 10000, "purpose": "batch_output"},
args.max_pages)
if ferr:
log.warning("openai file list stopped early: %s", ferr)
rows.extend(openai_rows(batches, file_index(files), ledger, now,
args.warn_days))
if anthropic_key:
checked.append("anthropic")
headers = {"x-api-key": anthropic_key, "anthropic-version": "2023-06-01"}
batches, err = page(ANTHROPIC_BATCHES_URL, headers, {"limit": 1000},
args.max_pages, cursor="after_id")
if err:
log.warning("anthropic batch list stopped early: %s", err)
rows.extend(anthropic_rows(batches, ledger, now, args.warn_days))
reportable = [r for r in rows if r["state"] != "claimed"]
for row in by_urgency(reportable):
log.warning("%-10s %-14s %-12s %s", row["provider"], row["id"][:14],
row["state"], row["detail"])
state, detail = verdict(reportable, ledger, args.warn_days)
emit = log.warning if state in FINDINGS else log.info
emit("%-22s %s", state, detail)
emit(" checked: %s, %d batch id(s) in the ledger",
", ".join(checked) or "nothing", len(ledger))
emit(" measured: status, the result artifact and the retention clock from "
"the batch lists, and file existence from the file list")
emit(" inferred: that an id absent from your ledger was never consumed, "
"since neither API records whether a result was downloaded")
for line in repair_lines(state, reportable, ledger):
emit(" repair: %s", line)
log.info("%d finding(s)", len(reportable))
return 1 if reportable else 0
if __name__ == "__main__":
sys.exit(main())
/**
* Find batch results that were paid for and never collected, on both providers.
*
* Read only: /v1/batches and /v1/files on OpenAI, /v1/messages/batches on
* Anthropic. Nothing is downloaded, deleted or re-run.
*
* Neither API records whether you read a result, so the ledger of what your
* consumer processed is an input and an empty one is a verdict.
*
* An OpenAI batch output file is deleted 30 days after the batch is complete;
* the file object's own expires_at wins where it is set. Claude batch results
* are available 29 days after the batch was created, not after it ended.
*/
import { readFile } from 'node:fs/promises';
const OPENAI_BATCHES_URL = 'https://api.openai.com/v1/batches';
const OPENAI_FILES_URL = 'https://api.openai.com/v1/files';
const ANTHROPIC_BATCHES_URL = 'https://api.anthropic.com/v1/messages/batches';
export const OPENAI_RETENTION = 30 * 86400;
export const ANTHROPIC_RETENTION = 29 * 86400;
export const OPEN_WINDOW = 24 * 3600;
export const GRACE = 2 * 3600;
const OPENAI_TERMINAL = new Set(['completed', 'failed', 'expired', 'cancelled']);
const FINDINGS = new Set(['batch-output-expiring', 'batch-output-lost',
'batch-output-unclaimed', 'batch-never-polled']);
/** Set of batch ids your consumer has processed. Pure. */
export function readLedger(text) {
const out = new Set();
for (const raw of String(text ?? '').split(',').join('\n').split('\n')) {
const line = raw.trim();
if (line && !line.startsWith('#')) out.add(line);
}
return out;
}
/** Epoch seconds from a unix number or an RFC 3339 string. Pure. */
export function parseTime(value) {
if (value === null || value === undefined || value === '' || typeof value === 'boolean') {
return null;
}
if (typeof value === 'number') return Math.trunc(value);
const ms = Date.parse(String(value));
return Number.isFinite(ms) ? Math.floor(ms / 1000) : null;
}
/** {file_id: file object}. Pure. */
export function fileIndex(files) {
const out = {};
for (const f of files ?? []) {
if (f && typeof f === 'object' && f.id) out[String(f.id)] = f;
}
return out;
}
/** [epoch, source] for when this output disappears. Pure. */
export function openaiDeadline(batch, fileObj) {
const stamp = parseTime((fileObj ?? {}).expires_at);
if (stamp) return [stamp, 'expires_at'];
const completed = parseTime((batch ?? {}).completed_at);
if (completed) return [completed + OPENAI_RETENTION, 'completed_at + 30d'];
const created = parseTime((batch ?? {}).created_at);
if (created) return [created + OPENAI_RETENTION, 'created_at + 30d'];
return [null, 'unknown'];
}
/** Whole days until the deadline. Pure. Null when there is no deadline. */
export function daysLeft(deadline, now) {
if (deadline === null || deadline === undefined) return null;
return Math.floor((deadline - now) / 86400);
}
/** One row per OpenAI batch worth reporting. Pure. */
export function openaiRows(batches, index, ledger, now, warnDays) {
const rows = [];
for (const b of batches ?? []) {
const status = String((b ?? {}).status ?? '');
const id = String(b.id);
const created = parseTime(b.created_at);
if (!OPENAI_TERMINAL.has(status)) {
if (created !== null && now - created > OPEN_WINDOW + GRACE) {
rows.push({ provider: 'openai', id, state: 'stalled', done: 0, days: null,
detail: `${status} for ${Math.floor((now - created) / 3600)} h, past the 24 h window` });
}
continue;
}
if (status !== 'completed') continue;
const done = Number((b.request_counts ?? {}).completed) || 0;
const artifact = b.output_file_id;
if (!artifact) continue;
if (!(String(artifact) in (index ?? {}))) {
rows.push({ provider: 'openai', id, state: 'lost', done, days: null,
detail: `output_file_id ${artifact} no longer exists` });
continue;
}
const [deadline, source] = openaiDeadline(b, index[String(artifact)]);
const left = daysLeft(deadline, now);
let state;
let detail;
if (left !== null && left <= warnDays) {
state = 'expiring';
detail = `${done} completed, ${Math.max(0, left)} days left (${source})`;
} else if ((ledger ?? new Set()).has(id)) {
state = 'claimed';
detail = `${done} completed, in the ingest ledger`;
} else {
state = 'unclaimed';
detail = `${done} completed, ${left === null ? 'unknown' : Math.max(0, left)} days left`;
}
rows.push({ provider: 'openai', id, state, done, days: left, detail });
}
return rows;
}
/** One row per Claude batch worth reporting. Pure. */
export function anthropicRows(batches, ledger, now, warnDays) {
const rows = [];
for (const b of batches ?? []) {
const id = String((b ?? {}).id);
const status = String(b.processing_status ?? '');
const created = parseTime(b.created_at);
const done = Number((b.request_counts ?? {}).succeeded) || 0;
if (status !== 'ended') {
if (created !== null && now - created > OPEN_WINDOW + GRACE) {
rows.push({ provider: 'anthropic', id, state: 'stalled', done, days: null,
detail: `${status || 'unknown'} for ${Math.floor((now - created) / 3600)} h, `
+ 'past the 24 h window' });
}
continue;
}
if (done <= 0) continue;
if (b.archived_at) {
rows.push({ provider: 'anthropic', id, state: 'lost', done, days: null,
detail: `archived_at set, ${done} succeeded, gone` });
continue;
}
const left = created === null ? null : daysLeft(created + ANTHROPIC_RETENTION, now);
let state;
let detail;
if (left !== null && left <= warnDays) {
state = 'expiring';
detail = `${done} succeeded, ${Math.max(0, left)} days left (created_at + 29d)`;
} else if ((ledger ?? new Set()).has(id)) {
state = 'claimed';
detail = `${done} succeeded, in the ingest ledger`;
} else {
state = 'unclaimed';
detail = `${done} succeeded, ${left === null ? 'unknown' : Math.max(0, left)} days left`;
}
rows.push({ provider: 'anthropic', id, state, done, days: left, detail });
}
return rows;
}
/** Rows ordered by what you can still act on. Pure. */
export function byUrgency(rows) {
const rank = { expiring: 0, lost: 1, unclaimed: 2, stalled: 3, claimed: 4 };
return [...(rows ?? [])].sort((a, b) => {
const ra = rank[a.state] ?? 9;
const rb = rank[b.state] ?? 9;
if (ra !== rb) return ra - rb;
const da = a.days === null || a.days === undefined ? 99999 : a.days;
const db = b.days === null || b.days === undefined ? 99999 : b.days;
if (da !== db) return da - db;
return String(a.id ?? '').localeCompare(String(b.id ?? ''));
});
}
/** {state: n}. Pure. */
export function countsByState(rows) {
const out = {};
for (const row of rows ?? []) out[row.state] = (out[row.state] ?? 0) + 1;
return out;
}
/** Grade the run. Pure. Returns [state, detail]. */
export function verdict(rows, ledger, warnDays) {
const list = rows ?? [];
if (!list.length) {
return ['batch-output-clean',
'every batch on the providers checked is either open inside its window or '
+ 'terminal with its output accounted for'];
}
const c = countsByState(list);
const parts = [];
if (c.lost) parts.push(`${c.lost} are already unrecoverable`);
if (c.unclaimed) parts.push(`${c.unclaimed} were never claimed`);
if (c.stalled) parts.push(`${c.stalled} never reached a terminal state`);
const tail = parts.length ? `, ${parts.join(', ')}` : '';
if (c.expiring) {
return ['batch-output-expiring',
`${c.expiring} batch(es) hold results that expire within ${warnDays} days${tail}`];
}
if (c.lost) {
const rest = parts.slice(1);
return ['batch-output-lost',
`${c.lost} batch(es) hold results that are already gone and can only be `
+ `recovered by re-running them${rest.length ? `, ${rest.join(', ')}` : ''}`];
}
if (c.unclaimed) {
let detail = `${c.unclaimed} batch(es) ended with results nothing has collected`;
if (!ledger || !ledger.size) {
detail += ', and no ingest ledger was supplied, so every terminal batch '
+ 'counts as unclaimed';
}
return ['batch-output-unclaimed', detail];
}
if (c.stalled) {
return ['batch-never-polled',
`${c.stalled} batch(es) have been open longer than the 24 hour window, `
+ 'which means nothing has polled them'];
}
return ['batch-output-clean',
`all ${list.length} terminal batch(es) are in the ingest ledger with runway `
+ 'left on the clock'];
}
/** The repair for one verdict. Pure. Printed, never performed. */
export function repairLines(state, rows, ledger) {
const c = countsByState(rows ?? []);
if (state === 'batch-output-clean') {
return ['nothing outstanding. Keep the assertion that a batch is not done '
+ 'until its output has been archived into your own store.'];
}
const lines = [];
if (c.expiring) {
lines.push('download the expiring outputs today and persist them keyed by '
+ 'batch id. After the clock runs out no read call recovers them.');
}
if (c.lost) {
lines.push('the lost ones must be re-run and re-paid. Nothing in either API '
+ 'can return results after the retention window closes.');
}
if (c.unclaimed) {
lines.push('sweep the unclaimed batches: list, diff against your ledger, '
+ 'download, and key the rows by custom_id, which is the only join '
+ 'available since results are not returned in request order.');
}
if (c.stalled) {
lines.push('a batch open past 24 hours is a stale object rather than a slow '
+ 'job. Poll every id you create to a terminal state, and record the id at '
+ 'creation time so orphans are identifiable.');
}
if (!ledger || !ledger.size) {
lines.push('no ingest ledger was supplied, so nothing could be confirmed as '
+ 'consumed. Record every batch id your consumer processes: neither API '
+ 'offers a read receipt.');
}
lines.push('run the error-file audit alongside this one. That note reads '
+ 'error_file_id, the list of rows that failed; this one reads the work '
+ 'itself. Both assertions belong in the same batch completion handler.');
return lines;
}
async function getJson(url, headers, params) {
const target = new URL(url);
for (const [k, v] of Object.entries(params ?? {})) target.searchParams.set(k, String(v));
let res;
try {
res = await fetch(target, { headers });
} catch (err) {
return [null, `request failed: ${err.message}`];
}
if (res.status !== 200) return [null, `HTTP ${res.status}`];
try {
return [await res.json(), null];
} catch {
return [null, 'response was not JSON'];
}
}
async function page(url, headers, params, maxPages, cursor = 'after') {
const rows = [];
let token = null;
for (let i = 0; i < Math.max(1, maxPages); i += 1) {
const query = { ...(params ?? {}) };
if (token) query[cursor] = token;
const [payload, err] = await getJson(url, headers, query);
if (err) return [rows, err];
const data = payload.data ?? [];
rows.push(...data);
if (!payload.has_more || !data.length) break;
token = payload.last_id ?? data[data.length - 1]?.id;
if (!token) break;
}
return [rows, null];
}
function args(argv) {
const out = { warnDays: 5, maxPages: 20 };
for (let i = 0; i < argv.length; i += 1) {
if (argv[i] === '--ledger') out.ledger = argv[i += 1];
else if (argv[i] === '--warn-days') out.warnDays = Number.parseInt(argv[i += 1], 10);
else if (argv[i] === '--max-pages') out.maxPages = Number.parseInt(argv[i += 1], 10);
}
return out;
}
async function main() {
const opts = args(process.argv.slice(2));
const openaiKey = process.env.OPENAI_API_KEY;
const anthropicKey = process.env.ANTHROPIC_API_KEY;
if (!openaiKey && !anthropicKey) {
console.error('set OPENAI_API_KEY (project key, Read Only) or '
+ 'ANTHROPIC_API_KEY (workspace key), or both');
process.exitCode = 2;
return;
}
let raw = process.env.BATCH_INGEST_LEDGER ?? '';
if (opts.ledger) {
try {
raw = await readFile(opts.ledger, 'utf8');
} catch (err) {
console.error(`could not read ${opts.ledger}: ${err.message}`);
process.exitCode = 2;
return;
}
}
const ledger = readLedger(raw);
const now = Math.floor(Date.now() / 1000);
const rows = [];
const checked = [];
if (openaiKey) {
checked.push('openai');
const headers = { Authorization: `Bearer ${openaiKey}` };
const [batches, err] = await page(OPENAI_BATCHES_URL, headers, { limit: 100 },
opts.maxPages);
if (err) console.log(`openai batch list stopped early: ${err}`);
const [files, ferr] = await page(OPENAI_FILES_URL, headers,
{ limit: 10000, purpose: 'batch_output' }, opts.maxPages);
if (ferr) console.log(`openai file list stopped early: ${ferr}`);
rows.push(...openaiRows(batches, fileIndex(files), ledger, now, opts.warnDays));
}
if (anthropicKey) {
checked.push('anthropic');
const headers = { 'x-api-key': anthropicKey, 'anthropic-version': '2023-06-01' };
const [batches, err] = await page(ANTHROPIC_BATCHES_URL, headers, { limit: 1000 },
opts.maxPages, 'after_id');
if (err) console.log(`anthropic batch list stopped early: ${err}`);
rows.push(...anthropicRows(batches, ledger, now, opts.warnDays));
}
const reportable = rows.filter((r) => r.state !== 'claimed');
for (const row of byUrgency(reportable)) {
console.log(`${row.provider.padEnd(10)} ${row.id.slice(0, 14).padEnd(14)} `
+ `${row.state.padEnd(12)} ${row.detail}`);
}
const [state, detail] = verdict(reportable, ledger, opts.warnDays);
console.log(`${state.padEnd(22)} ${detail}`);
console.log(` checked: ${checked.join(', ') || 'nothing'}, ${ledger.size} `
+ 'batch id(s) in the ledger');
console.log(' measured: status, the result artifact and the retention clock '
+ 'from the batch lists, and file existence from the file list');
console.log(' inferred: that an id absent from your ledger was never consumed, '
+ 'since neither API records whether a result was downloaded');
for (const line of repairLines(state, reportable, ledger)) {
console.log(` repair: ${line}`);
}
console.log(`${reportable.length} finding(s)`);
process.exitCode = reportable.length ? 1 : 0;
void FINDINGS;
}
if (import.meta.url === `file://${process.argv[1]}`) await main();
Add a test
The first test is the retention arithmetic, and it is the one most likely to be wrong in a way nobody notices: OpenAI counts thirty days from completed_at, Anthropic counts twenty-nine from created_at, and the file object's own expires_at beats both when the platform sets it. Get the anchor wrong and the queue sorts backwards. The second asserts that a missing output file is lost rather than unclaimed, because the two have different repairs — one is a download and one is a re-run — and that Anthropic's archived_at means the same thing. The third covers the three states the backlog wanted as separate notes, proving they come out of one pass: a stalled non-terminal batch past 24 hours, an ended batch with no ledger entry, and an ended batch that is in the ledger and therefore silent. The fourth is the verdict ordering, where an expiring result outranks a larger unclaimed pile because it is the only one with a deadline you can still beat. The fifth is the empty ledger, which must produce its own sentence rather than a clean run. And the last checks that the repair text names the error-file note, since the pair is the whole point of the boundary.
from batch_output_unclaimed_audit import (anthropic_rows, by_urgency,
counts_by_state, days_left,
file_index, openai_deadline,
openai_rows, parse_time, read_ledger,
repair_lines, verdict)
NOW = 1_800_000_000
DAY = 86400
OPENAI_BATCHES = [
{"id": "batch_fresh", "status": "completed", "created_at": NOW - 26 * DAY,
"completed_at": NOW - 26 * DAY, "output_file_id": "file_soon",
"request_counts": {"total": 88300, "completed": 88300, "failed": 0}},
{"id": "batch_gone", "status": "completed", "created_at": NOW - 60 * DAY,
"completed_at": NOW - 60 * DAY, "output_file_id": "file_2b7c",
"request_counts": {"total": 40000, "completed": 40000, "failed": 0}},
{"id": "batch_open", "status": "completed", "created_at": NOW - 3 * DAY,
"completed_at": NOW - 3 * DAY, "output_file_id": "file_room",
"request_counts": {"total": 90000, "completed": 90000, "failed": 0}},
{"id": "batch_stuck", "status": "in_progress", "created_at": NOW - 62 * 3600},
]
OPENAI_FILES = [
{"id": "file_soon", "purpose": "batch_output", "bytes": 10,
"created_at": NOW - 26 * DAY},
{"id": "file_room", "purpose": "batch_output", "bytes": 10,
"created_at": NOW - 3 * DAY},
]
ANTHROPIC_BATCHES = [
{"id": "msgbatch_arch", "processing_status": "ended",
"created_at": "2026-01-02T00:00:00Z", "ended_at": "2026-01-02T04:00:00Z",
"archived_at": "2026-01-31T00:00:00Z", "results_url": None,
"request_counts": {"processing": 0, "succeeded": 12400, "errored": 0,
"canceled": 0, "expired": 0}},
{"id": "msgbatch_open", "processing_status": "in_progress",
"created_at": "2026-01-02T00:00:00Z",
"request_counts": {"processing": 500, "succeeded": 0, "errored": 0,
"canceled": 0, "expired": 0}},
]
def test_the_retention_anchors_are_different_on_each_provider():
index = file_index(OPENAI_FILES)
# No expires_at on the file, so 30 days from completion.
deadline, source = openai_deadline(OPENAI_BATCHES[0], index["file_soon"])
assert source == "completed_at + 30d"
assert days_left(deadline, NOW) == 4
# The platform's own expires_at wins whenever it is set.
stamped = dict(index["file_soon"], expires_at=NOW + 2 * DAY)
deadline, source = openai_deadline(OPENAI_BATCHES[0], stamped)
assert source == "expires_at" and days_left(deadline, NOW) == 2
assert openai_deadline({}, {}) == (None, "unknown")
assert days_left(None, NOW) is None
# Anthropic counts 29 days from created_at, not from ended_at.
created = parse_time("2026-01-02T00:00:00Z")
rows = anthropic_rows([dict(ANTHROPIC_BATCHES[0], archived_at=None)], set(),
created + 27 * DAY, 5)
assert rows[0]["state"] == "expiring"
assert "created_at + 29d" in rows[0]["detail"]
def test_a_missing_output_file_is_lost_and_not_merely_unclaimed():
rows = openai_rows(OPENAI_BATCHES, file_index(OPENAI_FILES), set(), NOW, 5)
states = {r["id"]: r["state"] for r in rows}
assert states["batch_gone"] == "lost"
assert states["batch_fresh"] == "expiring"
assert states["batch_open"] == "unclaimed"
lost = [r for r in rows if r["state"] == "lost"][0]
assert "no longer exists" in lost["detail"]
# Anthropic says the same thing with archived_at.
arch = anthropic_rows(ANTHROPIC_BATCHES, set(), NOW, 5)
assert [r["state"] for r in arch if r["id"] == "msgbatch_arch"] == ["lost"]
def test_never_polled_never_fetched_and_never_claimed_are_one_pass():
rows = (openai_rows(OPENAI_BATCHES, file_index(OPENAI_FILES), set(), NOW, 5)
+ anthropic_rows(ANTHROPIC_BATCHES, set(), NOW, 5))
counts = counts_by_state(rows)
# Created and never polled.
assert counts["stalled"] == 2
stalled = [r for r in rows if r["state"] == "stalled"]
assert any("past the 24 h window" in r["detail"] for r in stalled)
# Ended and never claimed.
assert counts["unclaimed"] == 1
# A batch in the ledger goes quiet, which is the whole point of the join.
claimed = openai_rows([OPENAI_BATCHES[2]], file_index(OPENAI_FILES),
{"batch_open"}, NOW, 5)
assert claimed[0]["state"] == "claimed"
assert "in the ingest ledger" in claimed[0]["detail"]
def test_the_verdict_leads_with_what_you_can_still_act_on():
rows = (openai_rows(OPENAI_BATCHES, file_index(OPENAI_FILES), set(), NOW, 5)
+ anthropic_rows(ANTHROPIC_BATCHES, set(), NOW, 5))
state, detail = verdict(rows, {"x"}, 5)
assert state == "batch-output-expiring"
assert "expire within 5 days" in detail
assert "already unrecoverable" in detail and "never claimed" in detail
# Order on the page follows the same rule, soonest deadline first.
ordered = [r["state"] for r in by_urgency(rows)]
assert ordered[0] == "expiring"
assert ordered.index("lost") < ordered.index("unclaimed")
assert ordered.index("unclaimed") < ordered.index("stalled")
# Without anything expiring, the lost pile leads.
no_expiring = [r for r in rows if r["state"] != "expiring"]
assert verdict(no_expiring, {"x"}, 5)[0] == "batch-output-lost"
assert verdict([], set(), 5)[0] == "batch-output-clean"
def test_an_absent_ledger_is_reported_rather_than_assumed_away():
rows = openai_rows([OPENAI_BATCHES[2]], file_index(OPENAI_FILES), set(), NOW, 5)
state, detail = verdict(rows, set(), 5)
assert state == "batch-output-unclaimed"
assert "no ingest ledger was supplied" in detail
lines = repair_lines(state, rows, set())
assert any("neither API offers a read receipt" in line for line in lines)
assert read_ledger("# note\nbatch_a\nbatch_b,batch_a\n") == {"batch_a",
"batch_b"}
assert read_ledger("") == set()
def test_the_repair_hands_the_other_half_to_the_error_file_note():
rows = (openai_rows(OPENAI_BATCHES, file_index(OPENAI_FILES), set(), NOW, 5)
+ anthropic_rows(ANTHROPIC_BATCHES, set(), NOW, 5))
lines = repair_lines("batch-output-expiring", rows, {"x"})
assert any("error_file_id, the list of rows that failed" in line
for line in lines)
assert any("download the expiring outputs today" in line for line in lines)
assert any("re-run and re-paid" in line for line in lines)
assert any("stale object rather than" in line for line in lines)
clean = repair_lines("batch-output-clean", [], {"x"})
assert clean[0].startswith("nothing outstanding")
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { anthropicRows, byUrgency, countsByState, daysLeft, fileIndex,
openaiDeadline, openaiRows, parseTime, readLedger, repairLines,
verdict } from './batch-output-unclaimed-audit.mjs';
const NOW = 1800000000;
const DAY = 86400;
const OPENAI_BATCHES = [
{ id: 'batch_fresh', status: 'completed', created_at: NOW - 26 * DAY,
completed_at: NOW - 26 * DAY, output_file_id: 'file_soon',
request_counts: { total: 88300, completed: 88300, failed: 0 } },
{ id: 'batch_gone', status: 'completed', created_at: NOW - 60 * DAY,
completed_at: NOW - 60 * DAY, output_file_id: 'file_2b7c',
request_counts: { total: 40000, completed: 40000, failed: 0 } },
{ id: 'batch_open', status: 'completed', created_at: NOW - 3 * DAY,
completed_at: NOW - 3 * DAY, output_file_id: 'file_room',
request_counts: { total: 90000, completed: 90000, failed: 0 } },
{ id: 'batch_stuck', status: 'in_progress', created_at: NOW - 62 * 3600 },
];
const OPENAI_FILES = [
{ id: 'file_soon', purpose: 'batch_output', bytes: 10, created_at: NOW - 26 * DAY },
{ id: 'file_room', purpose: 'batch_output', bytes: 10, created_at: NOW - 3 * DAY },
];
const ANTHROPIC_BATCHES = [
{ id: 'msgbatch_arch', processing_status: 'ended',
created_at: '2026-01-02T00:00:00Z', ended_at: '2026-01-02T04:00:00Z',
archived_at: '2026-01-31T00:00:00Z', results_url: null,
request_counts: { processing: 0, succeeded: 12400, errored: 0, canceled: 0,
expired: 0 } },
{ id: 'msgbatch_open', processing_status: 'in_progress',
created_at: '2026-01-02T00:00:00Z',
request_counts: { processing: 500, succeeded: 0, errored: 0, canceled: 0,
expired: 0 } },
];
test('the retention anchors are different on each provider', () => {
const index = fileIndex(OPENAI_FILES);
const [deadline, source] = openaiDeadline(OPENAI_BATCHES[0], index.file_soon);
assert.equal(source, 'completed_at + 30d');
assert.equal(daysLeft(deadline, NOW), 4);
const stamped = { ...index.file_soon, expires_at: NOW + 2 * DAY };
const [d2, s2] = openaiDeadline(OPENAI_BATCHES[0], stamped);
assert.equal(s2, 'expires_at');
assert.equal(daysLeft(d2, NOW), 2);
assert.deepEqual(openaiDeadline({}, {}), [null, 'unknown']);
assert.equal(daysLeft(null, NOW), null);
const created = parseTime('2026-01-02T00:00:00Z');
const rows = anthropicRows([{ ...ANTHROPIC_BATCHES[0], archived_at: null }],
new Set(), created + 27 * DAY, 5);
assert.equal(rows[0].state, 'expiring');
assert.ok(rows[0].detail.includes('created_at + 29d'));
});
test('a missing output file is lost and not merely unclaimed', () => {
const rows = openaiRows(OPENAI_BATCHES, fileIndex(OPENAI_FILES), new Set(), NOW, 5);
const states = Object.fromEntries(rows.map((r) => [r.id, r.state]));
assert.equal(states.batch_gone, 'lost');
assert.equal(states.batch_fresh, 'expiring');
assert.equal(states.batch_open, 'unclaimed');
assert.ok(rows.find((r) => r.state === 'lost').detail.includes('no longer exists'));
const arch = anthropicRows(ANTHROPIC_BATCHES, new Set(), NOW, 5);
assert.deepEqual(arch.filter((r) => r.id === 'msgbatch_arch').map((r) => r.state),
['lost']);
});
test('never polled never fetched and never claimed are one pass', () => {
const rows = [...openaiRows(OPENAI_BATCHES, fileIndex(OPENAI_FILES), new Set(), NOW, 5),
...anthropicRows(ANTHROPIC_BATCHES, new Set(), NOW, 5)];
const counts = countsByState(rows);
assert.equal(counts.stalled, 2);
assert.ok(rows.filter((r) => r.state === 'stalled')
.some((r) => r.detail.includes('past the 24 h window')));
assert.equal(counts.unclaimed, 1);
const claimed = openaiRows([OPENAI_BATCHES[2]], fileIndex(OPENAI_FILES),
new Set(['batch_open']), NOW, 5);
assert.equal(claimed[0].state, 'claimed');
assert.ok(claimed[0].detail.includes('in the ingest ledger'));
});
test('the verdict leads with what you can still act on', () => {
const rows = [...openaiRows(OPENAI_BATCHES, fileIndex(OPENAI_FILES), new Set(), NOW, 5),
...anthropicRows(ANTHROPIC_BATCHES, new Set(), NOW, 5)];
const [state, detail] = verdict(rows, new Set(['x']), 5);
assert.equal(state, 'batch-output-expiring');
assert.ok(detail.includes('expire within 5 days'));
assert.ok(detail.includes('already unrecoverable') && detail.includes('never claimed'));
const ordered = byUrgency(rows).map((r) => r.state);
assert.equal(ordered[0], 'expiring');
assert.ok(ordered.indexOf('lost') < ordered.indexOf('unclaimed'));
assert.ok(ordered.indexOf('unclaimed') < ordered.indexOf('stalled'));
const noExpiring = rows.filter((r) => r.state !== 'expiring');
assert.equal(verdict(noExpiring, new Set(['x']), 5)[0], 'batch-output-lost');
assert.equal(verdict([], new Set(), 5)[0], 'batch-output-clean');
});
test('an absent ledger is reported rather than assumed away', () => {
const rows = openaiRows([OPENAI_BATCHES[2]], fileIndex(OPENAI_FILES), new Set(), NOW, 5);
const [state, detail] = verdict(rows, new Set(), 5);
assert.equal(state, 'batch-output-unclaimed');
assert.ok(detail.includes('no ingest ledger was supplied'));
assert.ok(repairLines(state, rows, new Set())
.some((l) => l.includes('neither API offers a read receipt')));
assert.deepEqual([...readLedger('# note\nbatch_a\nbatch_b,batch_a\n')].sort(),
['batch_a', 'batch_b']);
assert.equal(readLedger('').size, 0);
});
test('the repair hands the other half to the error file note', () => {
const rows = [...openaiRows(OPENAI_BATCHES, fileIndex(OPENAI_FILES), new Set(), NOW, 5),
...anthropicRows(ANTHROPIC_BATCHES, new Set(), NOW, 5)];
const lines = repairLines('batch-output-expiring', rows, new Set(['x']));
assert.ok(lines.some((l) => l.includes('error_file_id, the list of rows that failed')));
assert.ok(lines.some((l) => l.includes('download the expiring outputs today')));
assert.ok(lines.some((l) => l.includes('re-run and re-paid')));
assert.ok(lines.some((l) => l.includes('stale object rather than')));
assert.ok(repairLines('batch-output-clean', [], new Set(['x']))[0]
.startsWith('nothing outstanding'));
});
FAQ
How is this different from the error-file note?
By which file id it reads, and therefore by what a finding costs you. <code>error_file_id</code> points at the rows that failed: reading it tells you what to retry, and losing it costs you the knowledge of what is missing rather than the work. <code>output_file_id</code> and <code>results_url</code> point at the work itself: losing those means paying for every one of those answers again. The join is identical — batch list against your own ingest ledger — and the repairs land in the same completion handler, so run both scripts and fix them together. If your table is short and <code>error_file_id</code> is null, this note owns it.
How long do batch results actually last?
OpenAI's batch guide is explicit that the output file is automatically deleted thirty days after the batch is complete, so the clock starts at completion; where the file object carries its own <code>expires_at</code>, that value is what the platform will act on and the script uses it in preference. Anthropic's clock is twenty-nine days from <code>created_at</code>, and the documentation stresses that it is creation and not the end of processing. A Claude batch that took twenty hours to run therefore has twenty hours less runway than a naive count from <code>ended_at</code> would suggest.
Can I tell from the API whether a result was downloaded?
No, and this is the reason the script asks you for a ledger. File objects carry <code>bytes</code>, <code>created_at</code>, <code>filename</code>, <code>purpose</code> and sometimes <code>expires_at</code>, but nothing resembling a last-accessed timestamp, and there is no access log to query. Anthropic's <code>archived_at</code> is not a read receipt either — it records that the results became unavailable, whether or not anybody took them first. So the only honest test is a diff against your own record of consumption.
What about batches that never reached a terminal state?
They are counted here as <code>stalled</code>, and they are the same failure wearing a different status. Both providers cap batch processing at 24 hours: Anthropic publishes <code>expires_at</code> as exactly 24 hours after creation, and OpenAI's <code>completion_window</code> can only be set to <code>24h</code>. So a batch still showing <code>validating</code>, <code>in_progress</code> or <code>finalizing</code> three days later is a stale object rather than a slow job, and the repair is the same reconciler that would have collected the output.
Should I set output_expires_after when I create a batch?
It is worth knowing about. OpenAI's batch create endpoint accepts an <code>output_expires_after</code> object with an <code>anchor</code> of <code>created_at</code> and a <code>seconds</code> value between 3,600 and 2,592,000, which lets you choose a shorter retention deliberately instead of inheriting the default. That is a storage-hygiene decision rather than a fix for this problem: it makes unread output disappear sooner. The actual fix is the assertion that a batch is not done until its output is in your own store.
Related field notes
- The other half of the same completion handler: the failures, written down and unread
- The clock that closes on a batch before it finishes
- A batch that ran, reads completed, and has failed rows inside it
Sources
Every figure in this note is traced to one of these. Prices are list rates and change — check them for your own region before acting.
- Batch API guide — OpenAI developer docs
- Batch — OpenAI API reference
- Batch processing — Claude platform docs
- List Message Batches — Claude API reference
If your setup is misbehaving in a way this note does not cover, message me on LinkedIn with what you are seeing.