a batch reads completed while some of its rows failed

"""Report OpenAI batches that read completed while rows inside them failed.

Read only. GET requests and nothing else: give this a project key set to Read
Only. The repair is printed, never performed, because re-submitting the failed
rows means spending money on inference and that is your decision to make.
"""
import argparse
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("openai_batch_partial_failure_audit")

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

# Still moving. None of these is a verdict about the rows, because the counts
# are not final until the batch stops.
IN_FLIGHT = ("validating", "in_progress", "finalizing", "cancelling")

# Terminal, and owned by the sibling notes rather than by this script: a failed
# batch never ran a single row, an expired one ran out of window, a cancelled
# one was stopped on purpose.
OTHER_TERMINAL = ("failed", "expired", "cancelled")

FINDINGS = ("partial", "unaccounted")


def counts_of(batch):
    """Read request_counts into three ints, or None when it cannot be read.

    Pure. Missing members are read as zero because the API omits nothing here,
    but a request_counts that is not an object at all returns None rather than
    three zeros: three zeros would classify as an empty batch, which is a
    completely different and much calmer finding than an unreadable one.
    """
    counts = batch.get("request_counts")
    if not isinstance(counts, dict):
        return None
    try:
        total = int(counts.get("total") or 0)
        done = int(counts.get("completed") or 0)
        failed = int(counts.get("failed") or 0)
    except (TypeError, ValueError):
        return None
    return (total, done, failed)


def verdict(batch):
    """Classify one object from GET /v1/batches. Pure.

    Returns (state, detail). The two findings are kept apart on purpose:
    "partial" is rows that ran and failed, which are in the error file, and
    "unaccounted" is rows that are in neither column, which are not.
    """
    status = str(batch.get("status") or "").strip().lower()

    if status in IN_FLIGHT:
        return ("running",
                "status is %s, so the counts are not final and there is nothing "
                "to reconcile yet" % status)
    if status in OTHER_TERMINAL:
        return ("other-terminal",
                "status is %s. The batch did not finish running, which is a "
                "different problem from finishing with failures inside it."
                % status)
    if status != "completed":
        return ("unreadable",
                "status is %r, which is not a lifecycle state this script "
                "recognises. Read the object by hand." % (status or None,))

    numbers = counts_of(batch)
    if numbers is None:
        return ("unreadable",
                "the batch says completed and carries no readable "
                "request_counts, so nothing here can be reconciled. That is not "
                "the same as a clean batch and is not reported as one.")

    total, done, failed = numbers
    if total <= 0:
        return ("empty",
                "completed with a total of 0 request(s). The input file was "
                "empty or never parsed into rows.")
    if failed > 0:
        return ("partial",
                "%d of %d row(s) failed and the batch still reads completed. "
                "The output file is %d line(s) shorter than the input file."
                % (failed, total, total - done))
    if done < total:
        return ("unaccounted",
                "%d of %d row(s) are neither completed nor failed. Rows in "
                "neither column were abandoned rather than attempted, which is "
                "what a closed completion window looks like in the counts."
                % (total - done, total))
    return ("clean", "all %d row(s) completed" % total)


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")
    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.get("data") or []
        for batch in data:
            yield batch
        if not page.get("has_more") or not data:
            return
        params = {"limit": page_size, "after": data[-1].get("id")}


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    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 that reconcile")
    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

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

    checked = 0
    bad = 0
    for batch in batches(session, args.limit, args.pages):
        state, detail = verdict(batch)
        batch_id = str(batch.get("id") or "?")
        line = "%-15s %s  %s" % (state, batch_id, detail)

        if state in FINDINGS:
            checked += 1
            bad += 1
            log.warning(line)
            error_file = batch.get("error_file_id")
            if error_file:
                log.warning("  repair: read the failures with GET "
                            "/v1/files/%s/content, bucket the lines by "
                            "error.code, and re-submit the failed custom_ids as "
                            "a new batch", error_file)
            else:
                log.warning("  repair: no error_file_id on this batch, so the "
                            "missing rows were never attempted. Re-submit them "
                            "and reconcile output lines against input lines.")
            log.warning("  repair: treat request_counts.failed > 0 as a job "
                        "failure in your orchestrator instead of trusting "
                        "status == completed")
        elif state == "clean":
            checked += 1
            if args.show_all:
                log.info(line)
        elif state in ("unreadable", "empty"):
            checked += 1
            log.warning(line)
        elif args.show_all:
            log.info(line)

    log.info("%d completed batch(es) checked, %d with rows missing", checked, bad)
    return 1 if bad else 0


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

/**
 * Report OpenAI batches that read completed while rows inside them failed.
 *
 * Read only. GET requests and nothing else: give this a project key set to Read
 * Only. The repair is printed, never performed.
 */
const API = 'https://api.openai.com/v1';

// Still moving. None of these is a verdict about the rows, because the counts
// are not final until the batch stops.
const IN_FLIGHT = ['validating', 'in_progress', 'finalizing', 'cancelling'];

// Terminal, and owned by the sibling notes rather than by this script.
const OTHER_TERMINAL = ['failed', 'expired', 'cancelled'];

const FINDINGS = ['partial', 'unaccounted'];

/**
 * Read request_counts into three numbers, or null when it cannot be read. Pure.
 * A request_counts that is not an object returns null rather than three zeros,
 * because three zeros classify as an empty batch and that is a much calmer
 * finding than an unreadable one.
 */
export function countsOf(batch) {
  const counts = batch.request_counts;
  if (counts === null || typeof counts !== 'object' || Array.isArray(counts)) return null;
  const total = Number(counts.total ?? 0);
  const done = Number(counts.completed ?? 0);
  const failed = Number(counts.failed ?? 0);
  if (!Number.isFinite(total) || !Number.isFinite(done) || !Number.isFinite(failed)) {
    return null;
  }
  return [Math.trunc(total), Math.trunc(done), Math.trunc(failed)];
}

/**
 * Classify one object from GET /v1/batches. Pure. Returns [state, detail].
 * The two findings are kept apart on purpose: "partial" is rows that ran and
 * failed, which are in the error file, and "unaccounted" is rows that are in
 * neither column, which are not.
 */
export function verdict(batch) {
  const status = String(batch.status ?? '').trim().toLowerCase();

  if (IN_FLIGHT.includes(status)) {
    return ['running',
      `status is ${status}, so the counts are not final and there is nothing ` +
      'to reconcile yet'];
  }
  if (OTHER_TERMINAL.includes(status)) {
    return ['other-terminal',
      `status is ${status}. The batch did not finish running, which is a ` +
      'different problem from finishing with failures inside it.'];
  }
  if (status !== 'completed') {
    return ['unreadable',
      `status is ${JSON.stringify(status || null)}, which is not a lifecycle ` +
      'state this script recognises. Read the object by hand.'];
  }

  const numbers = countsOf(batch);
  if (numbers === null) {
    return ['unreadable',
      'the batch says completed and carries no readable request_counts, so ' +
      'nothing here can be reconciled. That is not the same as a clean batch ' +
      'and is not reported as one.'];
  }

  const [total, done, failed] = numbers;
  if (total <= 0) {
    return ['empty',
      'completed with a total of 0 request(s). The input file was empty or ' +
      'never parsed into rows.'];
  }
  if (failed > 0) {
    return ['partial',
      `${failed} of ${total} row(s) failed and the batch still reads ` +
      `completed. The output file is ${total - done} line(s) shorter than the ` +
      'input file.'];
  }
  if (done < total) {
    return ['unaccounted',
      `${total - done} of ${total} row(s) are neither completed nor failed. ` +
      'Rows in neither column were abandoned rather than attempted, which is ' +
      'what a closed completion window looks like in the counts.'];
  }
  return ['clean', `all ${total} row(s) completed`];
}

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.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 };
  }
}

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 pageSize = Number(process.env.LIMIT ?? 100);
  const maxPages = Number(process.env.PAGES ?? 20);
  const showAll = process.argv.includes('--show-all');

  let checked = 0;
  let bad = 0;
  for await (const batch of walk(key, pageSize, maxPages)) {
    const [state, detail] = verdict(batch);
    const batchId = String(batch.id ?? '?');
    const line = `${state.padEnd(15)} ${batchId}  ${detail}`;

    if (FINDINGS.includes(state)) {
      checked += 1;
      bad += 1;
      console.warn(line);
      console.warn(batch.error_file_id
        ? `  repair: read the failures with GET /v1/files/${batch.error_file_id}` +
          '/content, bucket the lines by error.code, and re-submit the failed ' +
          'custom_ids as a new batch'
        : '  repair: no error_file_id on this batch, so the missing rows were ' +
          'never attempted. Re-submit them and reconcile output lines against ' +
          'input lines.');
      console.warn('  repair: treat request_counts.failed > 0 as a job failure ' +
                   'in your orchestrator instead of trusting status == completed');
    } else if (state === 'clean') {
      checked += 1;
      if (showAll) console.log(line);
    } else if (state === 'unreadable' || state === 'empty') {
      checked += 1;
      console.warn(line);
    } else if (showAll) {
      console.log(line);
    }
  }

  console.log(`${checked} completed batch(es) checked, ${bad} with rows missing`);
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing key, and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from openai_batch_partial_failure_audit import counts_of, verdict


def batch(status="completed", total=100, completed=100, failed=0, **extra):
    """A batch object shaped like GET /v1/batches returns them."""
    body = {"id": "batch_test", "status": status,
            "request_counts": {"total": total, "completed": completed,
                               "failed": failed}}
    body.update(extra)
    return body


def test_completed_does_not_mean_every_row_succeeded():
    # The whole note: these two facts are compatible and the status hides it.
    state, detail = verdict(batch(total=50000, completed=49131, failed=869))
    assert state == "partial"
    assert "869 of 50000" in detail
    assert "869 line(s) shorter" in detail


def test_a_clean_batch_needs_both_halves_of_the_arithmetic():
    assert verdict(batch(total=100, completed=100, failed=0))[0] == "clean"
    assert verdict(batch(total=100, completed=99, failed=1))[0] == "partial"


def test_rows_in_neither_column_are_their_own_finding():
    # Not failures. Abandoned rows: attempted by nobody, absent from the error
    # file, and the shape a closed completion window leaves behind.
    state, detail = verdict(batch(total=100, completed=60, failed=0))
    assert state == "unaccounted"
    assert "40 of 100" in detail
    assert "abandoned" in detail


def test_an_in_flight_batch_is_not_reconciled_yet():
    for status in ("validating", "in_progress", "finalizing", "cancelling"):
        state, detail = verdict(batch(status=status, total=100, completed=3))
        assert state == "running"
        assert "not final" in detail


def test_the_other_terminal_states_belong_to_the_sibling_notes():
    for status in ("failed", "expired", "cancelled"):
        assert verdict(batch(status=status, total=100, completed=4,
                             failed=0))[0] == "other-terminal"


def test_missing_counts_are_never_reported_as_clean():
    assert verdict({"id": "b", "status": "completed"})[0] == "unreadable"
    assert verdict({"id": "b", "status": "completed",
                    "request_counts": []})[0] == "unreadable"
    assert verdict({"id": "b"})[0] == "unreadable"
    assert verdict(batch(total=0, completed=0))[0] == "empty"


def test_counts_are_read_leniently_but_not_invented():
    assert counts_of({"request_counts": {"total": 10}}) == (10, 0, 0)
    assert counts_of({"request_counts": {"total": "10", "completed": "9",
                                         "failed": "1"}}) == (10, 9, 1)
    assert counts_of({"request_counts": {"total": "many"}}) is None
    assert counts_of({}) is None

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { countsOf, verdict } from './openai-batch-partial-failure-audit.mjs';

/** A batch object shaped like GET /v1/batches returns them. */
function batch({ status = 'completed', total = 100, completed = 100,
                 failed = 0, ...extra } = {}) {
  return {
    id: 'batch_test',
    status,
    request_counts: { total, completed, failed },
    ...extra,
  };
}

test('completed does not mean every row succeeded', () => {
  const [state, detail] = verdict(batch({ total: 50000, completed: 49131, failed: 869 }));
  assert.equal(state, 'partial');
  assert.match(detail, /869 of 50000/);
  assert.match(detail, /869 line\(s\) shorter/);
});

test('a clean batch needs both halves of the arithmetic', () => {
  assert.equal(verdict(batch({ total: 100, completed: 100, failed: 0 }))[0], 'clean');
  assert.equal(verdict(batch({ total: 100, completed: 99, failed: 1 }))[0], 'partial');
});

test('rows in neither column are their own finding', () => {
  const [state, detail] = verdict(batch({ total: 100, completed: 60, failed: 0 }));
  assert.equal(state, 'unaccounted');
  assert.match(detail, /40 of 100/);
  assert.match(detail, /abandoned/);
});

test('an in flight batch is not reconciled yet', () => {
  for (const status of ['validating', 'in_progress', 'finalizing', 'cancelling']) {
    const [state, detail] = verdict(batch({ status, total: 100, completed: 3 }));
    assert.equal(state, 'running');
    assert.match(detail, /not final/);
  }
});

test('the other terminal states belong to the sibling notes', () => {
  for (const status of ['failed', 'expired', 'cancelled']) {
    assert.equal(verdict(batch({ status, total: 100, completed: 4 }))[0],
                 'other-terminal');
  }
});

test('missing counts are never reported as clean', () => {
  assert.equal(verdict({ id: 'b', status: 'completed' })[0], 'unreadable');
  assert.equal(verdict({ id: 'b', status: 'completed', request_counts: [] })[0],
               'unreadable');
  assert.equal(verdict({ id: 'b' })[0], 'unreadable');
  assert.equal(verdict(batch({ total: 0, completed: 0 }))[0], 'empty');
});

test('counts are read leniently but not invented', () => {
  assert.deepEqual(countsOf({ request_counts: { total: 10 } }), [10, 0, 0]);
  assert.deepEqual(countsOf({ request_counts: { total: '10', completed: '9', failed: '1' } }),
                   [10, 9, 1]);
  assert.equal(countsOf({ request_counts: { total: 'many' } }), null);
  assert.equal(countsOf({}), null);
});