the batch left an error_file_id that nothing ever fetched

"""Report OpenAI batch error files that exist and were never fetched.

Read only. Two GET requests and nothing else: give this a project key set to
Read Only. The repair is printed, never performed.

The API cannot tell you whether you read a file: there is no last_accessed_at
on a file object and no access log to query. So the second half of this check
comes from you, as a list of error file ids your ingest has consumed. Absence
of that list is itself an answer.
"""
import argparse
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("openai_batch_error_file_audit")

API = "https://api.openai.com/v1"
DAY = 86400

# Batch input, output and error files are retained for 30 days from creation.
# After that the content is unrecoverable by any read call.
RETENTION_DAYS = 30

IN_FLIGHT = ("validating", "in_progress", "finalizing", "cancelling")

FINDINGS = ("unread", "expiring", "aged-out")


def days_left(created_at, now, retention_days=RETENTION_DAYS):
    """Whole days of retention left on a file, or None if unreadable. Pure.

    Floors the elapsed time, so a file created 29.9 days ago has 1 day left
    rather than 0.1: this number is printed to a human who will act on it
    tomorrow, and rounding it the other way promises time that is not there.
    """
    try:
        created = int(created_at)
    except (TypeError, ValueError):
        return None
    if created <= 0:
        return None
    return retention_days - int((int(now) - created) // DAY)


def verdict(batch, file_meta, fetched, now, retention_days=RETENTION_DAYS,
            urgent_days=3):
    """Classify one batch against its error file and your ingest record. Pure.

    file_meta is the object from GET /v1/files/{id}, or None when that call
    found nothing. fetched is the set of error file ids your pipeline has
    consumed. now is unix seconds, passed in so the retention boundary can be
    tested at a fixed instant. Returns (state, detail).
    """
    status = str(batch.get("status") or "").strip().lower()
    file_id = str(batch.get("error_file_id") or "").strip()

    if status in IN_FLIGHT:
        return ("running",
                "status is %s; an error file is not final until the batch stops"
                % status)
    if not file_id:
        return ("no-error-file",
                "no error_file_id on this batch, so nothing failed hard enough "
                "to be written to one")
    if file_id in set(fetched or ()):
        return ("fetched",
                "error file %s is in the ingest record, so the failures were "
                "read" % file_id)

    created = None
    if isinstance(file_meta, dict):
        created = file_meta.get("created_at")
    if not created:
        created = batch.get("created_at")
    left = days_left(created, now, retention_days)

    if not isinstance(file_meta, dict):
        if left is not None and left <= 0:
            return ("aged-out",
                    "error file %s is past the %d day retention window and "
                    "GET /v1/files no longer returns it. Which rows failed, and "
                    "why, cannot be recovered by any read call now."
                    % (file_id, retention_days))
        return ("unresolvable",
                "the batch names error file %s but GET /v1/files/%s returned "
                "nothing, and the file is still inside the retention window. "
                "Check that id by hand." % (file_id, file_id))

    try:
        size = int(file_meta.get("bytes") or 0)
    except (TypeError, ValueError):
        size = 0

    if size <= 0:
        return ("empty",
                "error file %s exists and holds 0 byte(s). The id was allocated "
                "and never written to, so there is nothing in it to read."
                % file_id)
    if left is not None and left <= 0:
        return ("aged-out",
                "error file %s holds %d byte(s) that are past the %d day "
                "retention window. The metadata is still listed; the content is "
                "not retrievable." % (file_id, size, retention_days))
    if left is not None and left <= urgent_days:
        return ("expiring",
                "error file %s holds %d byte(s), is not in the ingest record, "
                "and expires in %d day(s). Download it before the window closes."
                % (file_id, size, left))
    return ("unread",
            "error file %s holds %d byte(s) and is not in the ingest record. "
            "Every line in it is a row missing from the downstream table."
            % (file_id, size))


def get(session, path, params=None):
    r = session.get(API + path, params=params or {}, timeout=60)
    if r.status_code == 401:
        raise SystemExit("401 from OpenAI: the key is wrong, revoked, or belongs "
                         "to another project")
    if r.status_code == 404:
        return None
    r.raise_for_status()
    return r.json()


def batches(session, page_size, max_pages):
    """Walk GET /v1/batches, which paginates on the id of the last object."""
    params = {"limit": page_size}
    for _ in range(max_pages):
        page = get(session, "/batches", params)
        data = (page or {}).get("data") or []
        for batch in data:
            yield batch
        if not (page or {}).get("has_more") or not data:
            return
        params = {"limit": page_size, "after": data[-1].get("id")}


def read_fetched(args):
    """The error file ids your pipeline says it consumed. Local reads only."""
    ids = set(args.fetched)
    if args.fetched_file:
        with open(args.fetched_file, "r", encoding="utf-8") as fh:
            ids.update(line.strip() for line in fh if line.strip())
    return ids


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--fetched", action="append", default=[],
                    help="an error file id your pipeline has consumed; repeatable")
    ap.add_argument("--fetched-file",
                    help="a file of error file ids your pipeline has consumed, "
                         "one per line")
    ap.add_argument("--limit", type=int, default=100,
                    help="page size for GET /v1/batches (default 100)")
    ap.add_argument("--pages", type=int, default=20,
                    help="stop after this many pages (default 20)")
    ap.add_argument("--show-all", action="store_true",
                    help="also print batches with nothing to fetch")
    args = ap.parse_args()

    key = os.environ.get("OPENAI_API_KEY")
    if not key:
        log.error("set OPENAI_API_KEY (a project key set to Read Only)")
        return 2

    fetched = read_fetched(args)
    if not fetched:
        log.info("no ingest record passed, so every error file will be reported "
                 "as unread. Pass --fetched or --fetched-file once you have one.")

    session = requests.Session()
    session.headers.update({"Authorization": "Bearer " + key})

    now = int(time.time())
    with_file = 0
    bad = 0
    for batch in batches(session, args.limit, args.pages):
        file_id = str(batch.get("error_file_id") or "").strip()
        file_meta = get(session, "/files/" + file_id) if file_id else None

        state, detail = verdict(batch, file_meta, fetched, now)
        batch_id = str(batch.get("id") or "?")
        line = "%-15s %s  %s" % (state, batch_id, detail)

        if file_id:
            with_file += 1
        if state in FINDINGS:
            bad += 1
            log.warning(line)
            if state == "aged-out":
                log.warning("  repair: the content is gone. Re-run the batch "
                            "from the original input file and diff the output "
                            "custom_ids against it to find the missing rows.")
            else:
                log.warning("  repair: GET /v1/files/%s/content, group the lines "
                            "by error.code, retry the transient ones "
                            "(rate_limit_exceeded, server_error) as a new batch, "
                            "and fix the rest", file_id)
            log.warning("  repair: assert error_file_id is null in the "
                        "batch-completion handler rather than checking it by hand "
                        "once a year")
        elif state == "unresolvable":
            log.warning(line)
        elif args.show_all or state == "empty":
            log.info(line)

    log.info("%d batch(es) with an error file, %d never fetched", with_file, bad)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report OpenAI batch error files that exist and were never fetched.
 *
 * Read only. Two GET requests and nothing else: give this a project key set to
 * Read Only. The repair is printed, never performed.
 *
 * The API cannot tell you whether you read a file, so the second half of this
 * check comes from you, as a list of error file ids your ingest has consumed.
 */
import { readFileSync } from 'node:fs';

const API = 'https://api.openai.com/v1';
const DAY = 86400;

// Batch input, output and error files are retained for 30 days from creation.
const RETENTION_DAYS = 30;

const IN_FLIGHT = ['validating', 'in_progress', 'finalizing', 'cancelling'];

const FINDINGS = ['unread', 'expiring', 'aged-out'];

/**
 * Whole days of retention left on a file, or null if unreadable. Pure. Floors
 * the elapsed time, so a file created 29.9 days ago has 1 day left rather than
 * 0.1: the number is printed to a human who will act on it tomorrow.
 */
export function daysLeft(createdAt, now, retentionDays = RETENTION_DAYS) {
  const created = Number(createdAt);
  if (!Number.isFinite(created) || created <= 0) return null;
  return retentionDays - Math.floor((Number(now) - created) / DAY);
}

/**
 * Classify one batch against its error file and your ingest record. Pure.
 * fileMeta is the object from GET /v1/files/{id}, or null when that call found
 * nothing. now is unix seconds, passed in so the retention boundary can be
 * tested at a fixed instant. Returns [state, detail].
 */
export function verdict(batch, fileMeta, fetched, now,
                        retentionDays = RETENTION_DAYS, urgentDays = 3) {
  const status = String(batch.status ?? '').trim().toLowerCase();
  const fileId = String(batch.error_file_id ?? '').trim();

  if (IN_FLIGHT.includes(status)) {
    return ['running',
      `status is ${status}; an error file is not final until the batch stops`];
  }
  if (!fileId) {
    return ['no-error-file',
      'no error_file_id on this batch, so nothing failed hard enough to be ' +
      'written to one'];
  }
  const seen = fetched instanceof Set ? fetched : new Set(fetched ?? []);
  if (seen.has(fileId)) {
    return ['fetched',
      `error file ${fileId} is in the ingest record, so the failures were read`];
  }

  const isMeta = fileMeta !== null && typeof fileMeta === 'object';
  const created = (isMeta && fileMeta.created_at) || batch.created_at;
  const left = daysLeft(created, now, retentionDays);

  if (!isMeta) {
    if (left !== null && left <= 0) {
      return ['aged-out',
        `error file ${fileId} is past the ${retentionDays} day retention ` +
        'window and GET /v1/files no longer returns it. Which rows failed, ' +
        'and why, cannot be recovered by any read call now.'];
    }
    return ['unresolvable',
      `the batch names error file ${fileId} but GET /v1/files/${fileId} ` +
      'returned nothing, and the file is still inside the retention window. ' +
      'Check that id by hand.'];
  }

  const raw = Number(fileMeta.bytes ?? 0);
  const size = Number.isFinite(raw) ? Math.trunc(raw) : 0;

  if (size <= 0) {
    return ['empty',
      `error file ${fileId} exists and holds 0 byte(s). The id was allocated ` +
      'and never written to, so there is nothing in it to read.'];
  }
  if (left !== null && left <= 0) {
    return ['aged-out',
      `error file ${fileId} holds ${size} byte(s) that are past the ` +
      `${retentionDays} day retention window. The metadata is still listed; ` +
      'the content is not retrievable.'];
  }
  if (left !== null && left <= urgentDays) {
    return ['expiring',
      `error file ${fileId} holds ${size} byte(s), is not in the ingest ` +
      `record, and expires in ${left} day(s). Download it before the window ` +
      'closes.'];
  }
  return ['unread',
    `error file ${fileId} holds ${size} byte(s) and is not in the ingest ` +
    'record. Every line in it is a row missing from the downstream table.'];
}

async function get(key, path, params = {}) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) {
    if (v !== undefined && v !== null) url.searchParams.set(k, String(v));
  }
  const res = await fetch(url, { headers: { Authorization: `Bearer ${key}` } });
  if (res.status === 401) {
    throw new Error('401 from OpenAI: the key is wrong, revoked, or belongs to ' +
                    'another project');
  }
  if (res.status === 404) return null;
  if (!res.ok) throw new Error(`${res.status} from ${path}`);
  return res.json();
}

async function* walk(key, pageSize, maxPages) {
  let params = { limit: pageSize };
  for (let i = 0; i < maxPages; i += 1) {
    const page = await get(key, '/batches', params);
    const data = page?.data ?? [];
    for (const batch of data) yield batch;
    if (!page?.has_more || data.length === 0) return;
    params = { limit: pageSize, after: data[data.length - 1].id };
  }
}

function readFetched() {
  const ids = new Set();
  process.argv.forEach((arg, i) => {
    if (arg === '--fetched' && process.argv[i + 1]) ids.add(process.argv[i + 1]);
    if (arg === '--fetched-file' && process.argv[i + 1]) {
      for (const line of readFileSync(process.argv[i + 1], 'utf8').split('\n')) {
        if (line.trim()) ids.add(line.trim());
      }
    }
  });
  return ids;
}

async function main() {
  const key = process.env.OPENAI_API_KEY;
  if (!key) {
    console.error('set OPENAI_API_KEY (a project key set to Read Only)');
    process.exitCode = 2;
    return;
  }

  const fetched = readFetched();
  if (fetched.size === 0) {
    console.log('no ingest record passed, so every error file will be reported ' +
                'as unread. Pass --fetched or --fetched-file once you have one.');
  }

  const pageSize = Number(process.env.LIMIT ?? 100);
  const maxPages = Number(process.env.PAGES ?? 20);
  const showAll = process.argv.includes('--show-all');
  const now = Math.floor(Date.now() / 1000);

  let withFile = 0;
  let bad = 0;
  for await (const batch of walk(key, pageSize, maxPages)) {
    const fileId = String(batch.error_file_id ?? '').trim();
    const fileMeta = fileId ? await get(key, `/files/${fileId}`) : null;

    const [state, detail] = verdict(batch, fileMeta, fetched, now);
    const line = `${state.padEnd(15)} ${String(batch.id ?? '?')}  ${detail}`;

    if (fileId) withFile += 1;
    if (FINDINGS.includes(state)) {
      bad += 1;
      console.warn(line);
      console.warn(state === 'aged-out'
        ? '  repair: the content is gone. Re-run the batch from the original ' +
          'input file and diff the output custom_ids against it to find the ' +
          'missing rows.'
        : `  repair: GET /v1/files/${fileId}/content, group the lines by ` +
          'error.code, retry the transient ones (rate_limit_exceeded, ' +
          'server_error) as a new batch, and fix the rest');
      console.warn('  repair: assert error_file_id is null in the ' +
                   'batch-completion handler rather than checking it by hand ' +
                   'once a year');
    } else if (state === 'unresolvable') {
      console.warn(line);
    } else if (showAll || state === 'empty') {
      console.log(line);
    }
  }

  console.log(`${withFile} batch(es) with an error file, ${bad} never fetched`);
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly, so importing this module from the test file
// does not fire main() and fail on the missing key.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from openai_batch_error_file_audit import days_left, verdict

# 2026-08-30T00:00:00Z. Fixed, because the retention boundary is the point.
NOW = 1788048000
DAY = 86400


def batch(status="completed", error_file_id="file_err", age_days=10):
    return {"id": "batch_test", "status": status,
            "error_file_id": error_file_id,
            "created_at": NOW - age_days * DAY}


def meta(size=4096, age_days=10):
    return {"id": "file_err", "bytes": size, "purpose": "batch_output",
            "created_at": NOW - age_days * DAY}


def test_an_error_file_nobody_fetched_is_the_finding():
    state, detail = verdict(batch(), meta(size=4096), set(), NOW)
    assert state == "unread"
    assert "4096 byte(s)" in detail
    assert "missing from the downstream table" in detail


def test_the_ingest_record_is_what_clears_it():
    state, _ = verdict(batch(), meta(), {"file_err"}, NOW)
    assert state == "fetched"


def test_retention_turns_a_task_into_a_hole():
    # 29 days old: one day left, and urgent.
    state, detail = verdict(batch(age_days=29), meta(age_days=29), set(), NOW)
    assert state == "expiring"
    assert "1 day(s)" in detail

    # 31 days old: the content is unrecoverable by any read call.
    state, detail = verdict(batch(age_days=31), meta(age_days=31), set(), NOW)
    assert state == "aged-out"
    assert "not retrievable" in detail


def test_a_missing_file_object_reads_differently_inside_and_outside_the_window():
    assert verdict(batch(age_days=40), None, set(), NOW)[0] == "aged-out"
    assert verdict(batch(age_days=2), None, set(), NOW)[0] == "unresolvable"


def test_an_empty_error_file_is_not_a_pile_of_failures():
    state, detail = verdict(batch(), meta(size=0), set(), NOW)
    assert state == "empty"
    assert "never written to" in detail


def test_batches_with_nothing_to_read_are_left_alone():
    assert verdict(batch(error_file_id=None), None, set(), NOW)[0] == "no-error-file"
    assert verdict(batch(error_file_id=""), None, set(), NOW)[0] == "no-error-file"
    assert verdict(batch(status="in_progress"), meta(), set(), NOW)[0] == "running"


def test_days_left_floors_and_admits_ignorance():
    assert days_left(NOW - 10 * DAY, NOW) == 20
    assert days_left(NOW - int(29.9 * DAY), NOW) == 1
    assert days_left(NOW - 30 * DAY, NOW) == 0
    assert days_left(None, NOW) is None
    assert days_left("yesterday", NOW) is None

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { daysLeft, verdict } from './openai-batch-error-file-audit.mjs';

// 2026-08-30T00:00:00Z. Fixed, because the retention boundary is the point.
const NOW = 1788048000;
const DAY = 86400;

function batch({ status = 'completed', errorFileId = 'file_err',
                 ageDays = 10 } = {}) {
  return {
    id: 'batch_test',
    status,
    error_file_id: errorFileId,
    created_at: NOW - ageDays * DAY,
  };
}

function meta({ size = 4096, ageDays = 10 } = {}) {
  return {
    id: 'file_err',
    bytes: size,
    purpose: 'batch_output',
    created_at: NOW - ageDays * DAY,
  };
}

test('an error file nobody fetched is the finding', () => {
  const [state, detail] = verdict(batch(), meta({ size: 4096 }), new Set(), NOW);
  assert.equal(state, 'unread');
  assert.match(detail, /4096 byte\(s\)/);
  assert.match(detail, /missing from the downstream table/);
});

test('the ingest record is what clears it', () => {
  assert.equal(verdict(batch(), meta(), new Set(['file_err']), NOW)[0], 'fetched');
});

test('retention turns a task into a hole', () => {
  const [near, nearDetail] = verdict(batch({ ageDays: 29 }),
                                     meta({ ageDays: 29 }), new Set(), NOW);
  assert.equal(near, 'expiring');
  assert.match(nearDetail, /1 day\(s\)/);

  const [gone, goneDetail] = verdict(batch({ ageDays: 31 }),
                                     meta({ ageDays: 31 }), new Set(), NOW);
  assert.equal(gone, 'aged-out');
  assert.match(goneDetail, /not retrievable/);
});

test('a missing file object reads differently inside and outside the window', () => {
  assert.equal(verdict(batch({ ageDays: 40 }), null, new Set(), NOW)[0], 'aged-out');
  assert.equal(verdict(batch({ ageDays: 2 }), null, new Set(), NOW)[0],
               'unresolvable');
});

test('an empty error file is not a pile of failures', () => {
  const [state, detail] = verdict(batch(), meta({ size: 0 }), new Set(), NOW);
  assert.equal(state, 'empty');
  assert.match(detail, /never written to/);
});

test('batches with nothing to read are left alone', () => {
  assert.equal(verdict(batch({ errorFileId: null }), null, new Set(), NOW)[0],
               'no-error-file');
  assert.equal(verdict(batch({ errorFileId: '' }), null, new Set(), NOW)[0],
               'no-error-file');
  assert.equal(verdict(batch({ status: 'in_progress' }), meta(), new Set(), NOW)[0],
               'running');
});

test('daysLeft floors and admits ignorance', () => {
  assert.equal(daysLeft(NOW - 10 * DAY, NOW), 20);
  assert.equal(daysLeft(NOW - Math.trunc(29.9 * DAY), NOW), 1);
  assert.equal(daysLeft(NOW - 30 * DAY, NOW), 0);
  assert.equal(daysLeft(null, NOW), null);
  assert.equal(daysLeft('yesterday', NOW), null);
});