feat/N4: Batch mode (submit → fetch → finalize → record recovery) - #7
Merged
Conversation
…400) The Anthropic client placed a `name` key inside output_config.format, which Anthropic's API forbids, so every native-schema call (parse / chat_structured / response(response_format=...) and the batch structured path) failed with HTTP 400. Build the request via the SDK's typed OutputConfigParam and drop `name`, which is still required by the OpenAI wire object and the neutral spec. Flip the unit test into a regression guard and add a batch-path regression test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Introduce the N4 Stage 1 (submit) surface: a new leaf module prkit.batch with submit_batch_physics_reasoning(client, dataset_or_problems, ...) -> list[BatchSubmission], the dumps_batch_jsonl/write_batch_jsonl/validate_batch_requests helpers, and the BatchInputError/BatchSubmitError exceptions. The orchestrator preprocesses each problem like the synchronous path, splits the dataset into provider batches, writes provider-correct JSONL under one run folder per call, submits each batch via the existing client.submit_batch, and writes a run-level metadata.json before returning typed receipts. Add two methods to BaseModelClient: build_problem_batch_request (free-text batch analogue of solve_physics_problem, reusing build_plain_question_prompt for batch == sync prompt parity) and a thin submit_batch_physics_reasoning facade that lazily imports prkit.batch. The leaf imports only prkit.core.domain + stdlib at module load; the client is duck-typed. Tests run fully offline (MagicMock+patch the provider SDKs) and cover splitting, JSONL artifacts, submission, id validation/correlation, surrogate ids, partial failure, the facade, the Anthropic inline path, run-folder layout, metadata.json content, id round-trips, defaults/overrides, and import isolation. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Delete the 100%-unused structured-output batch path: build_batch_structured_request and parse_batch_structured_response plus the _build_batch_structured_request default on BaseModelClient, and the four per-provider overrides (openai/anthropic/gemini/xai). These had zero callers anywhere — free-text batch submit mirrors the free-text-only solve_physics_problem and needs none of it; structured batch can return as a follow-up when structured output lands on the sync path too. xAI's only batch method was this override; it now has no batch surface (it is not a supported batch provider — submit/poll/retrieve already raise NotImplementedError). Prune the imports orphaned by the deletions and fix the build_batch_request docstring cross-reference. The structured-output engine (StructuredOutputPlan / .parse() / _resolve_structured_output_plan) and the retained batch lifecycle (build_batch_request / submit_batch / poll_batch / retrieve_batch_results / _parse_*_result_line) are untouched. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Delete the Runner structural Protocol from prkit.api (and prkit.api.__all__), the docstring mention in prkit/__init__.py, the reserved row in CONTRACT.md, and its references in tests/prkit/test_api.py. Runner had no implementation; its docstring reserved it "for roadmap N4", which is now batch-mode submit shipped as a bounded submitter (prkit.batch.submit_batch_physics_reasoning + the BaseModelClient facade), not a Runner noun. The contract now pins three Protocol nouns (DatasetProvider / ModelClient / Scorer) plus Verdict. The unrelated annotation Runner (a Callable alias) and the llm_judge OpenAIJudgeRunner are different symbols and untouched. Per the API_VERSION policy this provisional-1.0 removal is recorded in CONTRACT.md, not signalled by a bump. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Replace the frozen per-minibatch BatchSubmission receipt with one mutable whole-batch ledger per run, in preparation for a resumable fetch step that advances it. Batch-level facts (provider, model, created_at, dataset, ...) are stored once; each provider job becomes a dict in `minibatches` carrying a mutable `status` plus its batch_id / id_map / num_requests / counts / output_path. - BatchSubmission gains `from_dict`, `save`, `load`, and the pure status helpers `minibatches_to_fetch` / `set_status` / `is_complete` / `status_counts`; it is now a dataclass (was frozen). `from_dict` parses `created_at` with `datetime.fromisoformat` and compares tz by equality (not identity). - `submit_batch_physics_reasoning` now saves the ledger to `<run_dir>/metadata.json` and returns the run-folder path (str) instead of `list[BatchSubmission]`; disk is the source of truth across the ~24h provider window. The `BaseModelClient` facade return type follows. - Rename batch->minibatch throughout the unit sense: `batch_size`->`minibatch_size`, `num_batches`->`minibatch_count`, `batch_index`->minibatch `index`, `batch_*.jsonl`->`minibatch_*.jsonl`. Add the status-string vocabulary (submitted/running/.../fetch_error) the ledger records. - Remove `BatchSubmitError`: it carried per-minibatch BatchSubmission receipts for resume, a shape the single-ledger reshape obsoletes (a failed submit is now a `submit_error` minibatch in the ledger). It was never raised. Stage-1 submit tests rewritten to load the ledger and assert on its fields; new test_batch_submission_ledger.py covers the status helpers + save/load round-trip. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Drive the fetch side of the batch lane over the existing client primitives: poll
each non-final minibatch, download terminal-and-retrievable ones, correlate each
result back to its problem, and advance + persist the ledger. A bounded, resumable
helper — no scoring, no pricing, no end-to-end runner.
- `fetch_batch(client, run_dir|submission, *, wait=False, poll_interval=10.0,
timeout=None, outputs_dirname="outputs", progress=True) -> BatchSubmission`:
poll-once-and-persist by default (resumes across processes via the ledger;
skips fetched / terminal-failed / submit_error minibatches), `wait=True` loops
to completion with `time.sleep` backoff honouring `timeout`. Writes normalized
BatchResult JSONL to `<run_dir>/outputs/minibatch_XXXX.jsonl`; EXPIRED minibatches
are still retrieved (partial subset -> fetched), only FAILED/CANCELLED have nothing
to fetch; a raised retrieve marks `fetch_error` and is retried next pass. Logs a
one-line INFO progress summary (completion line once terminal) on `prkit.batch`.
- `iter_batch_results(run_dir|submission) -> Iterator[(problem_id, BatchResult)]`:
pure offline reader, correlates custom_id -> problem_id via each minibatch's id_map,
emits one result per submitted problem (synthetic ERRORED for ids the provider never
returned), drops + counts uncorrelated extras. No network.
- Capability gate: `batch_fetch_supported` + `_FETCH_CAPABLE_PROVIDERS`
{openai, anthropic, google} (Gemini is "google"; xAI has no batch surface).
`fetch_batch` raises `BatchFetchUnsupportedError` up front, never a raw
NotImplementedError mid-sweep.
- Thin `BaseModelClient.fetch_batch_physics_reasoning` facade (lazy import, mirrors
the submit facade). Reuses poll_batch / retrieve_batch_results / batch_types as-is;
the wait/backoff loop lives here, not in the client.
Leaf discipline: batch_types is imported lazily inside the fetch functions, so
`import prkit.batch` stays light at load. test_import_isolation extended with the
new symbols (load-time only); test_fetch_batch.py covers the gate, happy path,
resume/skip, non-terminal, EXPIRED partials, terminal failures, submit-error skip,
cross-provider correlation, completeness, the handle contract, progress reporting,
the facade, and the wait loop + timeout.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… helper
Stage-3 (finalize) structural groundwork on the prkit.batch leaf, ahead of the
two new finalize verbs:
- New minibatch status CONSOLIDATED (a FETCHED minibatch whose per-problem
results/ files are written); added to __all__, _SKIP_FETCH_STATUSES (fetch
never re-polls it), and _COMPLETE_STATUSES (terminal for is_complete()).
- New status sets _HAS_OUTPUT_STATUSES {FETCHED, CONSOLIDATED} and
_RESUBMIT_STATUSES {FAILED, SUBMIT_ERROR, EXPIRED} (excludes CANCELLED).
- iter_batch_results' per-minibatch correlation body extracted into a private
_iter_minibatch_results(mb) (reused by consolidation), and its status gate
widened from == FETCHED to in _HAS_OUTPUT_STATUSES so re-scoring still reads a
minibatch's outputs/ file after consolidation. The batch_types import stays
lazy inside the helper (leaf import discipline preserved).
- New BatchNotTerminalError(BatchInputError) for the resubmit precondition.
- New stdlib infra _atomic_write_text (tempfile + os.replace) and
_safe_results_filename; add import os / import tempfile.
- submit_batch_physics_reasoning(overwrite=True) cleanup also clears results/
and results_manifest.json (the only auto-clear of results/).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…verbs The two Stage-3 (finalize) verbs over a fetched ledger, plus the resubmit client facade and the owner's next-command guidance prompts. Bounded helpers — no dataset loading, no auto-chaining; the consumer still drives the outer loop. - consolidate_batch_results(submission, *, results_dirname="results"): offline, lenient, incremental. Streams each FETCHED-not-yet-CONSOLIDATED minibatch's results to per-problem results/<problem_id>.json (atomic write, one record in RAM at a time — never a whole-run aggregate), marks each minibatch CONSOLIDATED and saves after each (crash-safe/resumable), warns about minibatches not yet succeeded, and refreshes results_manifest.json (atomic, written last). Raises BatchInputError on an empty ledger or a filename collision (two problems sanitizing to one file — never a silent overwrite). - resubmit_failed_minibatches(client, submission): re-submits each FAILED / SUBMIT_ERROR / EXPIRED minibatch (not CANCELLED) by re-reading its persisted inputs/ file. Submit-first / mutate-second / save-third per minibatch; a per item failure becomes SUBMIT_ERROR and the loop continues. Requires a terminal ledger (BatchNotTerminalError otherwise) and a batch-capable provider (BatchFetchUnsupportedError up front). - BaseModelClient.resubmit_failed_minibatches facade (thin, lazy-imports prkit.batch), mirroring fetch_batch_physics_reasoning. - Next-command prompts at the end of submit / fetch (3-way) / resubmit / consolidate via the prkit.batch logger (guidance text, not orchestration). Folds in the one forced existing-test adjustment (the fetch progress test now isolates the per-pass summary from the new end-of-fetch next-command line). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… coverage New offline test modules and extensions for the finalize half: - test_consolidate_results.py: happy path (per-problem files + manifest at the run-dir root, results/ holds only per-problem files), lenient succeeded-subset (+WARNING), incremental resume (already-CONSOLIDATED not rewritten), streaming (one record per file, manifest written last — guards the no-aggregate rule), filename sanitization + collision (BatchInputError, no silent overwrite), crash-safety before the manifest and mid-minibatch (atomic, no partial file), iter_batch_results after consolidate (the _HAS_OUTPUT_STATUSES widening), empty-ledger guard, and the consolidate next-command prompt. - test_resubmit_minibatches.py: happy path (requests re-read from the persisted input file + merged metadata, entries reset to SUBMITTED with cleared fetch fields), CANCELLED exclusion + only-CANCELLED no-op, terminal + capability preconditions (no submit on failure), per-item failure + continue, submit-first ordering, the client facade, and the resubmit next-command prompt. - test_import_isolation.py: import consolidate_batch_results / resubmit_failed_minibatches in the subprocess load-time cleanliness check. - test_fetch_batch.py: the 3-way end-of-fetch next-command prompt. - test_submit_physics_reasoning.py: submit's next-command prompt and overwrite=True clearing a stale results/ + results_manifest.json. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Single source of truth for the prkit batch lane, reconciling the three
per-stage design notes (internal/N4_BATCH_DESIGN_STAGE_{1,2,3}.md, kept as
historical rationale). Documents the live submit/fetch/finalize API with all
three stages implemented, the ledger and run-folder layout, the status
lifecycle, the scoring and cost-meter seams, and a Design-evolution table
recording the terminology, return-type, and ledger changes across stages.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The first prkit-synthesized per-record status beyond the provider set: a record that has exhausted MAX_ATTEMPTS total submissions (whole-minibatch + record-level retries) is given up on as MAX_ATTEMPTED, distinct from a transient ERRORED so a downstream scorer can tell "we gave up" from "errored". Mirrors the synthetic-ERRORED precedent in the batch leaf's correlation reader. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
fetch_batch now partitions each retrieved minibatch and siphons recoverable per-record failures (real ERRORED/EXPIRED/CANCELED + synthetic-missing, under MAX_ATTEMPTS=3) onto the source minibatch's ledger entry — pruning its id_map and decrementing num_requests in lockstep — writing succeeded-only to outputs/ and refreshing a derived failed-records-batch-input.jsonl; a record that has exhausted its submissions is instead rewritten as MAX_ATTEMPTED and kept in id_map so it consolidates terminally. resubmit_failed_minibatches is renamed resubmit_failures (module + BaseModelClient facade): it now also resubmits CANCELLED minibatches (reversing the Stage-3 exclusion) and bumps each whole-minibatch attempt, and additionally drains the failed-records accumulator into fresh retry minibatches (monotonic index, is_retry/retry_sources/record_attempts, minibatch_count++). The results manifest gains pending_failed_records and gates fully_consolidated on it being zero; the next-command prompts drop the CANCELLED dead-end and point at resubmit_failures, and consolidate emits the record-recovery hint when records are still pending. The leaf stays import-light: batch_types (BatchResult/BatchItemStatus.MAX_ATTEMPTED) is referenced only inside the already-lazy helpers, and _SIPHON_RECORD_STATUSES holds string values so the partition needs no enum at module load. Amends the Stage-1/2/3 surface and updates the affected tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
test_fetch_siphon.py: the siphon happy path (succeeded-only outputs; one failed_records entry; id_map pruned + num_requests in lockstep; derived accumulator), synthetic-missing siphon, re-fetch idempotency, no-failure no-artifact, and MAX_ATTEMPTS exhaustion (rewritten max_attempted, consolidated). test_resubmit_records.py: drain mechanics (fresh retry minibatch shape; minibatch_count bump; consumed accumulator; submit_batch fed the failed input line by wire id), a drain submit error keeping the record recoverable, the headline end-to-end recovery loop to fully_consolidated, the field-parse correlation regression (custom_id and Gemini key, with reordered input lines), the MAX_ATTEMPTS off-by-one spanning whole-minibatch + record retries, the terminal precondition, the ledger round-trip of the new minibatch keys, and the consolidate record-recovery hint. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Updates the rollup doc with Stage-4 design: siphon-at-fetch, record-drain in resubmit_failures, MAX_ATTEMPTED terminal status, CANCELLED reversal, the new end-to-end loop, and all vocabulary/state-machine sections. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements the full N4 batch lane in 4 stages across 13 commits:
submit_batch_physics_reasoning,build_problem_batch_request, mutableBatchSubmissionledger; removed deadRunnerProtocol + structured-batch wrappers.fetch_batch,iter_batch_results,batch_fetch_supported; resumable poll-and-download with crash-safe ledger writes.consolidate_batch_results(per-problemresults/<id>.json+ manifest),resubmit_failures(whole-minibatch resubmit forFAILED/SUBMIT_ERROR/EXPIRED/CANCELLED).fetch_batch; drain them into fresh minibatches insideresubmit_failures;MAX_ATTEMPTS=3bound;BatchItemStatus.MAX_ATTEMPTEDterminal status;CANCELLEDreversal (now resubmitted, not a dead-end).Design rollup:
BATCH_MODE.mdat project root (single source of truth for the lane).Test plan
pytest tests/prkit/batch/— 2201 passed, 2 skipped, 83.19% coverage on branchbatch_fetch_supportedreturnsTrueforopenai/anthropic/gemini,Falseforxaifully_consolidated = TrueMAX_ATTEMPTS=3yields aMAX_ATTEMPTEDresult (notERRORED) initer_batch_results🤖 Generated with Claude Code