feat/N4: Batch mode (submit → fetch → finalize → record recovery)

CHANGELOG.md

@@ -50,6 +50,7 @@ Production releases follow semantic versioning. TestPyPI validation builds use P
 - **`DatasetHub` registration-ordering bug** — calling `DatasetHub.register(name, Loader)` before any built-in dataset was touched caused all built-in loaders and downloaders to be permanently suppressed. Built-ins are now seeded idempotently (via `setdefault`) at the start of every public mutating method, so external registrations can safely happen in any order.
 - JEEBench loader handling for numeric answer categories and retained metadata.
 - Workflow module behavior in domain assessment, theorem review, and workflow composition paths.
+- **Anthropic structured output 400** — the Anthropic client no longer sends an invalid `name` key inside `output_config.format` (a key Anthropic's API forbids), which previously made every native-schema call — `parse()`, `chat_structured()`, `response(response_format=...)`, and the batch structured path — fail with HTTP 400. The request is now built via the SDK's typed `OutputConfigParam`, so future schema drift surfaces as a type error rather than a runtime 400. OpenAI (which requires `name`) and Gemini are unaffected.
 
 ### Deprecated
 

src/prkit/CONTRACT.md

@@ -32,14 +32,13 @@ and returns the same canonical `Verdict`.
   `check_model_client`, `ConformanceTestMixin`) is a stable companion: use it to
   verify your own loader/scorer/client satisfies the contract.
 
-The contract pins four structural (`typing.Protocol`) nouns plus one result type:
+The contract pins three structural (`typing.Protocol`) nouns plus one result type:
 
 | Noun | Protocol | Reference implementation |
 |------|----------|--------------------------|
 | Dataset loader | `DatasetProvider` | `BaseDatasetLoader` subclasses |
 | Inference client | `ModelClient` | `BaseModelClient` subclasses |
 | Scorer | `Scorer` | `prkit.scoring.SemanticsScorer` (binary); `EedScorer`/`SeedScorer` (vendor edit-distance baselines); `SemanticsEedScorer`/`SemanticsSeedScorer` (our-semantics edit distance, graded); `LLMJudgeScorer` (model-graded) |
-| Runner | `Runner` | *(reserved; no implementation yet)* |
 | Result | `Verdict` | `prkit.core.verdict.Verdict` |
 
 > **Note on `@runtime_checkable`:** `isinstance(x, Scorer)` only verifies that the
@@ -135,3 +134,10 @@ changes to those are documented in the package release notes, not the contract v
   compatibility shim is deleted; import from `prkit.semantics.build` directly.
 - `prkit.evaluation.llm_judge` (model-graded scoring) is **retained** — it is a distinct
   capability, not a duplicate of the deterministic scoring path.
+- **`Runner` Protocol removed.** The reserved `Runner` structural Protocol (it had no
+  implementation; its docstring reserved it "for roadmap N4") is removed from
+  `prkit.api` / `prkit.api.__all__`. Batch-mode runs (N4) ship as a bounded *submitter*
+  (`prkit.batch.submit_batch_physics_reasoning` + the `BaseModelClient` facade), not as a
+  `Runner` noun. The contract now pins **three** Protocol nouns plus `Verdict`. Per the
+  `API_VERSION` policy this provisional-1.0 removal is tracked here, **not** signalled by a
+  major bump.

src/prkit/__init__.py

@@ -9,7 +9,7 @@
 
 Integration entry point:
     :mod:`prkit.api` is the frozen, semver-governed public contract (the
-    ``DatasetProvider``/``ModelClient``/``Scorer``/``Runner`` protocols and the
+    ``DatasetProvider``/``ModelClient``/``Scorer`` protocols and the
     ``Verdict`` result). Integrate against it rather than reaching into
     subpackages. See ``prkit/CONTRACT.md`` for the stability and version policy.
 

src/prkit/api.py

@@ -101,31 +101,12 @@ def score(
     def get_info(self) -> dict[str, Any]: ...  # MUST include "version"
 
 
-@runtime_checkable
-class Runner(Protocol):
-    """Orchestration noun: drive a :class:`ModelClient` over a
-    :class:`PhysicsDataset` and score with a :class:`Scorer`.
-
-    No implementation ships today; the contract is reserved for a later
-    orchestration item (roadmap N4).
-    """
-
-    def run(
-        self,
-        dataset: PhysicsDataset,
-        model: ModelClient,
-        scorer: Scorer,
-        **kwargs: Any,
-    ) -> Any: ...
-
-
 __all__ = [
     "API_VERSION",
     # the four structural nouns + the canonical result
     "DatasetProvider",
     "ModelClient",
     "Scorer",
-    "Runner",
     "Verdict",
     # canonical answer ontology
     "AnswerObjectKind",

src/prkit/core/model_clients/anthropic.py

@@ -5,7 +5,7 @@
 import os
 import re
 from collections.abc import Callable, Iterator, Sequence
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from pydantic import BaseModel
 
@@ -34,6 +34,9 @@
 except ImportError:  # pragma: no cover - tested via runtime error path
     pass
 
+if TYPE_CHECKING:
+    from anthropic.types import OutputConfigParam
+
 TOOL_NAME = "emit_structured_output"
 _TOOL_NAME_RE = re.compile(r"[^A-Za-z0-9_-]+")
 ANTHROPIC_OPTIONAL_PARAMETER_LIMIT = 24
@@ -329,13 +332,16 @@ def _build_messages_params(
             params["system"] = instructions
         if response_format is not None:
             normalized = normalize_response_format(response_format)
-            params["output_config"] = {
+            # Anthropic's output_config.format accepts only ``type`` and ``schema``;
+            # a ``name`` key (an OpenAI-ism) makes the API reject the request with a
+            # 400. The typed annotation makes any future stray key a mypy error.
+            output_config: OutputConfigParam = {
                 "format": {
                     "type": "json_schema",
-                    "name": normalized["name"],
                     "schema": normalized["schema"],
                 }
             }
+            params["output_config"] = output_config
         if extra:
             params.update(extra)
         return params
@@ -438,34 +444,6 @@ def _resolve_structured_output_plan(
             response_format=_anthropic_transformed_response_format(spec),
         )
 
-    def _build_batch_structured_request(
-        self,
-        *,
-        request_id: str,
-        user_prompt: str,
-        response_model: type[BaseModel],
-        image_paths: tuple[str, ...],
-        max_output_tokens: int | None,
-        plan: StructuredOutputPlan,
-        **kwargs: Any,
-    ) -> dict[str, Any]:
-        del response_model, kwargs
-        if plan.mode != "json_schema":
-            raise ValueError(
-                "Anthropic batch structured requests require json_schema mode. "
-                f"Got {plan.mode!r}."
-            )
-        params = self._build_messages_params(
-            input=user_prompt,
-            instructions=None,
-            image_paths=image_paths,
-            max_output_tokens=(
-                max_output_tokens if max_output_tokens is not None else 4096
-            ),
-            response_format=plan.response_format or {},
-        )
-        return {"custom_id": request_id, "params": params}
-
 
 def _parse_anthropic_result_entry(entry: Any) -> BatchResult:
     """Parse one streamed Message Batch result entry into a ``BatchResult``."""

src/prkit/core/model_clients/base.py

@@ -10,7 +10,7 @@
 
 from pydantic import BaseModel, ValidationError
 
-from ..domain import PhysicsProblem
+from ..domain import PhysicsDataset, PhysicsProblem
 from ..logging_config import PRKitLogger
 from ..project_env import load_project_dotenv
 from .batch_types import BatchResult, BatchStatus
@@ -28,6 +28,9 @@
 )
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
+    from prkit.batch import BatchSubmission
     from prkit.semantics.schema import PhysicsQuestionSemantics
 
 T = TypeVar("T", bound=BaseModel)
@@ -293,61 +296,6 @@ def solve_physics_problem(
             **kwargs,
         )
 
-    def build_batch_structured_request(
-        self,
-        *,
-        request_id: str,
-        user_prompt: str,
-        response_model: type[T],
-        image_paths: Sequence[str] | None = None,
-        max_output_tokens: int | None = None,
-        structured_policy: StructuredOutputPolicy = "native_required",
-        **kwargs: Any,
-    ) -> dict[str, Any]:
-        """Build a provider-specific batch request dict for structured output."""
-        plan = self.resolve_structured_output_plan(
-            response_model,
-            structured_policy=structured_policy,
-        )
-        return self._build_batch_structured_request(
-            request_id=request_id,
-            user_prompt=user_prompt + (plan.prompt_suffix or ""),
-            response_model=response_model,
-            image_paths=tuple(image_paths or ()),
-            max_output_tokens=max_output_tokens,
-            plan=plan,
-            **kwargs,
-        )
-
-    def parse_batch_structured_response(
-        self,
-        *,
-        response_model: type[T],
-        raw_response_text: str,
-        structured_policy: StructuredOutputPolicy = "native_required",
-        structured_output_strategy: str | None = None,
-    ) -> StructuredCallResult[T]:
-        """Parse a raw batch response text into a typed ``StructuredCallResult``."""
-        plan = self.resolve_structured_output_plan(
-            response_model,
-            structured_policy=structured_policy,
-        )
-        if structured_output_strategy is not None:
-            plan = StructuredOutputPlan(
-                mode=plan.mode,
-                strategy=structured_output_strategy,
-                native_schema_enforced=plan.native_schema_enforced,
-                accepted_artifact_modes=plan.accepted_artifact_modes,
-                accepted_artifact_strategies=plan.accepted_artifact_strategies,
-                response_format=plan.response_format,
-                prompt_suffix=plan.prompt_suffix,
-            )
-        return self._build_structured_call_result(
-            response_model=response_model,
-            raw_text=raw_response_text,
-            plan=plan,
-        )
-
     # ------------------------------------------------------------------
     # Free-text batch requests + the asynchronous job lifecycle
     # ------------------------------------------------------------------
@@ -368,8 +316,7 @@ def build_batch_request(
         no ``response_format``. Passing ``instructions=""`` suppresses the system
         prompt on every provider (see :meth:`_resolve_instructions`), so the
         resulting request matches a synchronous
-        ``response(input=..., instructions="")`` call. Use
-        :meth:`build_batch_structured_request` instead for schema-enforced output.
+        ``response(input=..., instructions="")`` call.
         """
         return self._build_batch_request(
             request_id=request_id,
@@ -381,6 +328,114 @@ def build_batch_request(
             **kwargs,
         )
 
+    def build_problem_batch_request(
+        self,
+        problem: PhysicsProblem | str,
+        *,
+        request_id: str | None = None,
+        instructions: str | None = None,
+        max_output_tokens: int | None = None,
+        temperature: float | None = None,
+        **kwargs: Any,
+    ) -> dict[str, Any]:
+        """Build a free-text batch request line for *problem*.
+
+        Batch analogue of :meth:`solve_physics_problem`, dispatching on the
+        runtime type of *problem* the same way: a ``str`` is treated as the
+        question text (no images, and *request_id* is required since there is no
+        ``problem_id``); a :class:`~prkit.core.domain.PhysicsProblem` is rendered
+        with :func:`build_plain_question_prompt` and its ``image_path`` images,
+        defaulting *request_id* to ``problem.problem_id``. Reusing the same prompt
+        builder and image path as the synchronous solver guarantees batch ≡ sync
+        prompts. Free-text only, matching the ANSWER_TEXT-only sync path; delegates
+        to :meth:`build_batch_request`.
+        """
+        if isinstance(problem, str):
+            if request_id is None:
+                raise ValueError(
+                    "request_id is required when problem is a raw question string."
+                )
+            prompt = problem.strip()
+            image_paths: list[str] | None = None
+        elif isinstance(problem, PhysicsProblem):
+            prompt = build_plain_question_prompt(problem)
+            image_paths = problem.image_path or None
+            if request_id is None:
+                request_id = problem.problem_id
+        else:
+            raise TypeError(
+                f"Unsupported problem input type: {type(problem)!r}. Expected a "
+                "PhysicsProblem or a question string."
+            )
+
+        return self.build_batch_request(
+            request_id=request_id,
+            input=prompt,
+            instructions=instructions,
+            image_paths=image_paths,
+            max_output_tokens=max_output_tokens,
+            temperature=temperature,
+            **kwargs,
+        )
+
+    def submit_batch_physics_reasoning(
+        self,
+        problems: PhysicsDataset | Sequence[PhysicsProblem],
+        **kwargs: Any,
+    ) -> str:
+        """Submit *problems* as provider batch jobs; return the run-folder path.
+
+        Thin one-line facade mirroring :meth:`solve_physics_problem`: lazily imports
+        :mod:`prkit.batch` (kept off the import-light path, like the lazy
+        ``prkit.semantics`` import in :meth:`solve_physics_problem`) and delegates to
+        :func:`prkit.batch.submit_batch_physics_reasoning`. See that function for the
+        keyword arguments (``output_dir`` / ``run_name`` / ``minibatch_size`` / …).
+        The ledger is saved to ``<run_dir>/metadata.json``; reconstruct it later via
+        :meth:`fetch_batch_physics_reasoning` (or ``BatchSubmission.load(run_dir)``).
+        """
+        from prkit.batch import submit_batch_physics_reasoning as _submit_batch
+
+        return _submit_batch(self, problems, **kwargs)
+
+    def fetch_batch_physics_reasoning(
+        self,
+        run_dir_or_submission: BatchSubmission | str | Path,
+        **kwargs: Any,
+    ) -> BatchSubmission:
+        """Poll + download a submitted batch run; return the updated ledger.
+
+        Thin one-line facade mirroring :meth:`submit_batch_physics_reasoning`: lazily
+        imports :mod:`prkit.batch` and delegates to :func:`prkit.batch.fetch_batch`.
+        Typically called with the ``run_dir`` string that submit returned;
+        reconstructs the ledger via :meth:`BatchSubmission.load` and returns the
+        freshly-updated :class:`~prkit.batch.BatchSubmission`. See
+        :func:`prkit.batch.fetch_batch` for the keyword arguments (``wait`` /
+        ``poll_interval`` / ``timeout`` / ``outputs_dirname`` / ``progress``).
+        """
+        from prkit.batch import fetch_batch
+
+        return fetch_batch(self, run_dir_or_submission, **kwargs)
+
+    def resubmit_failures(
+        self,
+        run_dir_or_submission: BatchSubmission | str | Path,
+        **kwargs: Any,
+    ) -> BatchSubmission:
+        """Re-drive a terminal batch run's failures (minibatches + records); return it.
+
+        Thin one-line facade mirroring :meth:`fetch_batch_physics_reasoning`: lazily
+        imports :mod:`prkit.batch` and delegates to
+        :func:`prkit.batch.resubmit_failures`. Re-submits each
+        FAILED / SUBMIT_ERROR / EXPIRED / CANCELLED minibatch in place (re-reading its
+        persisted ``inputs/`` file) *and* drains the run's siphoned failed-records
+        accumulator into fresh minibatches; the run must be terminal first (run
+        :meth:`fetch_batch_physics_reasoning` until it is). Returns the updated
+        :class:`~prkit.batch.BatchSubmission`; fetch the new jobs next.
+        """
+        from prkit.batch import resubmit_failures
+
+        return resubmit_failures(self, run_dir_or_submission, **kwargs)
+
     def _build_batch_request(
         self,
         *,
@@ -528,27 +583,3 @@ def _extract_structured_payload(
     ) -> dict[str, Any] | list[Any] | None:
         del plan
         return extract_json_payload(raw_text)
-
-    def _build_batch_structured_request(
-        self,
-        *,
-        request_id: str,
-        user_prompt: str,
-        response_model: type[T],
-        image_paths: tuple[str, ...],
-        max_output_tokens: int | None,
-        plan: StructuredOutputPlan,
-        **kwargs: Any,
-    ) -> dict[str, Any]:
-        del (
-            request_id,
-            user_prompt,
-            response_model,
-            image_paths,
-            max_output_tokens,
-            plan,
-            kwargs,
-        )
-        raise NotImplementedError(
-            f"Batch structured requests are not implemented for provider={self._provider_name()!r}."
-        )

src/prkit/core/model_clients/batch_types.py

@@ -45,6 +45,12 @@ class BatchItemStatus(_StrEnum):
     ERRORED = "errored"
     EXPIRED = "expired"
     CANCELED = "canceled"
+    # prkit-synthesized (not a provider status): a record that has exhausted
+    # ``prkit.batch.MAX_ATTEMPTS`` total submissions (whole-minibatch + record-level
+    # retries) and is given up on as a terminal failure. Distinct from a transient
+    # ERRORED so a downstream scorer can tell "we gave up" from "errored". Mirrors
+    # the synthetic-ERRORED precedent in ``prkit.batch._iter_minibatch_results``.
+    MAX_ATTEMPTED = "max_attempted"
 
 
 @dataclass(frozen=True)