fix(runtime): harden durable run lifecycle

README.md

@@ -79,8 +79,8 @@ Durability is split on purpose:
 - the artifact store keeps replayable request JSONL and downloaded raw batch payloads
 
 That split is what allows retries and fresh-process resume without keeping every request inline in the control-plane store.
-On resume, existing active provider batches are polled before `batchor` materializes or submits new local work, so completed batches and retry backoff are reconciled first. During long ingestion, active batches are also reconciled at durable item-chunk boundaries when the configured provider polling interval is due. Polling happens before that chunk's submission attempt, so terminal remote work releases locally accounted enqueue tokens without waiting for the entire source to be materialized. Deterministic sources also use stored checkpoint completion metadata to avoid opening a fully materialized Parquet or composite source just to discover there are no rows left.
-Incomplete checkpointed ingestion keeps the run non-terminal. In the process that
+On resume, existing active provider batches are polled before `batchor` materializes or submits new local work, so completed batches and retry backoff are reconciled first. During long ingestion, active batches are also reconciled at durable work-slice boundaries when the configured provider polling interval is due. Submission backoff never suppresses reconciliation. Fast sources retain the normal 1,000-item persistence path, while slow sources flush partial atomic slices on a time budget so pause, cancellation, polling, and deadlines can be observed. Provider-terminal batches remain locally consumable until every linked item transition and capacity release succeeds, making result consumption replayable after a crash.
+Every newly created run has an incomplete ingestion marker, so even an empty run cannot appear terminal while its first source slice is still being rendered. Incomplete checkpointed ingestion keeps the run non-terminal. In the process that
 started the job, `Run.resume()` can continue from the attached source. After a
 fresh-process rehydration, call `start(job, run_id=...)`; `refresh()` otherwise
 raises `RunIngestionSourceRequiredError` instead of silently completing only the
@@ -329,6 +329,7 @@ run = runner.start(
                 target_ratio=0.7,
                 headroom=50_000,
                 max_batch_enqueued_tokens=500_000,
+                quota_scope="research-account-a",
             ),
         ),
     )
@@ -338,7 +339,7 @@ run.wait()
 print(run.results()[0].output)
 ```
 
-While ingesting or waiting, `batchor` polls active batches before submitting more work when the provider's monotonic polling cadence is due. This avoids one provider request per fast ingestion chunk while allowing long ingestion to reuse capacity released by completed batches. If a refresh consumes a completed batch, records terminal item state, or submits another provider batch, `Run.wait()` immediately continues draining instead of sleeping for the next poll interval.
+While ingesting or waiting, `batchor` polls active batches before submitting more work when the provider's monotonic polling cadence is due. This avoids one provider request per fast ingestion chunk while allowing long ingestion to reuse capacity released by completed batches. OpenAI capacity reservations are atomic and shared across runs by model and `quota_scope`; omit the scope for a conservative store-wide default, or assign stable labels when one store serves multiple accounts/projects. If a refresh consumes a completed batch, records terminal item state, or submits another provider batch, `Run.wait()` immediately continues draining instead of sleeping for the next poll interval.
 
 Structured-output models are validated up front against the OpenAI strict-schema subset used by `batchor`.
 
@@ -583,6 +584,8 @@ Current events include run creation/resume, automatic quota pause, item ingestio
 - Item `attempt_count` means consumed provider attempts: successful submitted completions increment it, counted item-level failures increment it, and local or batch-level retry/reset paths that did not consume an item attempt leave it unchanged.
 - Durable artifacts now go through an `ArtifactStore` seam. The built-in implementation is `LocalArtifactStore`, intended for local disk or a shared mounted volume.
 - Fresh-process resume requeues any locally claimed but not yet submitted items before continuing, so a process crash after request-artifact persistence does not strand work in `queued_local`.
+- Batch creation uses a durable submission intent. If the process or connection fails after the provider may have created remote work, advancement raises `RunSubmissionIndeterminateError` instead of silently duplicating execution. After inspecting the provider, resolve the intent with `resolve_indeterminate_submission_as_created(provider_batch_id=..., status=...)` or `resolve_indeterminate_submission_as_not_created()`.
+- `Run.wait(timeout=...)` propagates its deadline through library-controlled ingestion and polling boundaries and never starts another provider operation after expiry. An already-running arbitrary Python callback or provider SDK call remains subject to that operation's own timeout and cannot be forcibly interrupted safely.
 
 ## Durable artifacts
 

docs/design_docs/ARCHITECTURE.md

@@ -145,16 +145,16 @@ Internally that expands to:
 1. Resolve provider and storage implementations.
 2. Persist run config and ingest items into durable state.
 3. On resume, poll any already-active provider batches before ingesting or submitting new work.
-4. Persist source items in durable chunks; at each chunk boundary, poll active batches before submission when the monotonic provider cadence is due.
+4. Persist source items in durable work slices; keep the 1,000-item fast path, but flush a smaller atomic slice when the time/control budget requires it.
 5. Claim a bounded submission window from pending items.
 6. Build or replay request JSONL rows.
 7. Persist request artifacts before upload.
-8. Submit one or more provider batch files.
+8. Atomically reserve quota-scoped capacity and persist a submission intent before remote creation, then atomically register the returned batch with its item linkage.
 9. Poll active batches.
 10. Download output/error files.
 11. Parse terminal item results back into the state store.
 
-Ingestion polling stays inside the ingestion module behind its existing poll callback. Its local monotonic cadence state does not add storage schema or background-thread coordination. After each ingestion poll, control state and retry backoff are re-read before any submission or further checkpointed-source materialization.
+Ingestion polling stays inside the ingestion module behind its existing poll callback. Provider cadence and work-slice clocks are separate: fast sources retain normal batch granularity, while slow sources reach durable cooperative boundaries on a time budget. After each ingestion poll, control state and retry backoff are re-read before submission or further checkpointed-source materialization. Submission backoff blocks materialization/submission but never active-batch reconciliation.
 
 When a caller uses `Run.wait()`, the runtime repeats that poll-and-submit pass until the run is terminal. A pass that changes durable work state, such as consuming completed batches or submitting more items, immediately triggers the next pass rather than sleeping for the configured poll interval.
 
@@ -473,6 +473,10 @@ That split gives `batchor`:
     can advance, unless cancellation intentionally abandons the unmaterialized
     source tail and finalizes the checkpoint.
 21. `BatchItem.metadata["batchor_lineage"]` is reserved for lightweight source/join metadata when provided by built-in adapters or callers.
+22. Provider-terminal status and local result consumption are distinct: a terminal batch remains locally active while submitted items or an unreleased capacity reservation still require replay.
+23. Remote batch creation is preceded by a durable submission intent. An ambiguous create outcome requires explicit operator resolution as either the known remote batch or verified-not-created; the runtime never guesses and resubmits.
+24. OpenAI enqueue capacity is reserved atomically across runs by a stable provider/model/quota scope without persisting credential-derived identities.
+25. `wait(timeout=...)` is cooperative: the deadline bounds sleeps and prevents new library-controlled work, while an already-running user callback or provider SDK call is governed by its own timeout.
 
 ## Extension seams
 

docs/design_docs/OPENAI_BATCHING.md

@@ -125,13 +125,14 @@ OpenAI-specific enqueue settings live on `OpenAIProviderConfig.enqueue_limits`:
 - `target_ratio`
 - `headroom`
 - `max_batch_enqueued_tokens`
+- `quota_scope`
 
 From those settings, `batchor` derives:
 
 - an effective inflight token budget
 - an effective per-batch token limit
 
-Submission is constrained by both token budget and generic chunking rules such as max request count and max request file size.
+Submission is constrained by both token budget and generic chunking rules such as max request count and max request file size. Capacity is reserved atomically in durable storage before upload/create, so concurrent runs sharing the same model and quota scope cannot each claim the full account budget. The default scope is conservative and store-wide per model; set a stable `quota_scope` when one store serves multiple OpenAI accounts or projects.
 During `Run.wait()`, each drain cycle polls active batches before submitting new work, so capacity freed by completed batches is visible before the next submit attempt. If that cycle makes durable progress, the wait loop immediately runs another cycle instead of sleeping.
 
 ## Batch splitting
@@ -155,7 +156,8 @@ Not all failures are treated the same way.
 
 Control-plane failures:
 
-- transient upload/create/poll failures are treated as retryable batch-control-plane failures
+- transient upload/poll failures are treated as retryable batch-control-plane failures
+- a create-call outcome that may have reached the provider is recorded as an indeterminate durable intent and requires explicit operator resolution instead of automatic resubmission
 - retryable control-plane failures do not consume item attempts
 - batch submit failures can trigger batch-level backoff
 - transient poll failures do not stall unrelated submissions when other capacity remains
@@ -174,9 +176,10 @@ Item-level failures:
 
 Cleanup behavior:
 
-- if upload succeeds but batch creation fails, `batchor` makes a best-effort attempt to delete the uploaded OpenAI input file
+- if upload succeeds but a known provider rejection proves batch creation did not occur, `batchor` makes a best-effort attempt to delete the uploaded OpenAI input file
 - if upload or batch creation fails because OpenAI reports insufficient quota, local queued items are released back to pending before the run is paused
-- if a process dies after local artifact persistence but before durable batch registration, fresh-process resume requeues those items and resubmits from persisted request artifacts
+- if creation may have succeeded but local registration did not, `RunSubmissionIndeterminateError` blocks duplicate execution until the operator attaches the confirmed remote batch with `resolve_indeterminate_submission_as_created(...)` or verifies absence with `resolve_indeterminate_submission_as_not_created()`
+- terminal result download, parsing, item persistence, and capacity release are replayable; the batch remains locally reconcilable until all linked work is durably consumed
 
 ## Run control
 

docs/design_docs/RUNTIME_DURABILITY_0_0_7.md

@@ -0,0 +1,77 @@
+# Runtime Durability 0.0.7
+
+Status: released design and verification record for `0.0.7`.
+
+## Outcome
+
+`0.0.7` closes durability and liveness gaps in the run lifecycle without
+changing the provider request formats or introducing background execution. The
+release makes remote terminal consumption replayable, keeps reconciliation live
+during ingestion and submit backoff, records submission ambiguity durably, and
+enforces OpenAI enqueue capacity across runs sharing a quota scope.
+
+## Design boundaries
+
+- Batchor owns durable local reconciliation; providers remain the authority for
+  remote batch status and result payloads.
+- Result consumption is at-least-once/replayable locally. A terminal remote
+  batch remains active until submitted items are terminalized and its capacity
+  reservation is released.
+- Generic provider creation cannot promise exactly-once remote execution. A
+  timeout or crash after the create call becomes an explicit
+  `RunSubmissionIndeterminateError`, requiring operator resolution as created
+  or not created rather than an unsafe automatic retry.
+- Cooperative deadlines prevent starting another Batchor-controlled operation
+  after expiry and cap sleeps. They cannot interrupt an arbitrary prompt
+  callback or provider SDK call already running in the current thread; those
+  calls retain their own timeout contract.
+- Non-checkpointable iterators are retained only by their in-process execution
+  owner. Manual pause can stop and resume that iterator in the same process,
+  but fresh-process mid-ingestion recovery remains unsupported.
+- Capacity is shared only among runs using the same durable store and explicit
+  quota scope. The default scope is conservative for one account/model; users
+  sharing a store across accounts or projects must configure distinct scopes.
+
+## Change list and acceptance criteria
+
+| Finding | Design/fix boundary | Acceptance evidence |
+| --- | --- | --- |
+| Terminal batch lost after a local failure | Terminal batches with submitted items or unreleased intent capacity remain pollable. Consumption filters already-terminal rows and persists item changes conditionally so it can replay after download, parse, or partial-write failure. | Fake-provider terminal-replay and partial-mutation tests in polling/storage suites. |
+| Submit backoff suppresses polling | Submission backoff blocks materialization/submission only; active remote batches are reconciled at each due work boundary. | Ingestion regression with pre-existing submit backoff and a completed active batch. |
+| A slow 1,000-row slice blocks progress | Materialization uses a normal 1,000-item fast path plus a cooperative time budget that flushes partial atomic slices. | Slow-source clock tests proving polling/control observation before another full chunk. |
+| Empty non-checkpointed run appears terminal | Every new run receives an incomplete generic ingest marker atomically; v5 backfill treats unknown legacy state as incomplete. | Fresh SQLite rehydration and legacy migration tests. |
+| Remote create and local linkage can split | Persist a submission intent before create; atomically finalize intent, batch, and item linkage after confirmed success. | Crash-window/indeterminate-resolution storage and submission tests. |
+| Manual pause drains arbitrary ingestion | Manual pause stops at a cooperative boundary and retains the iterator for same-process resume; quota auto-pause still drains finite arbitrary input. | Non-checkpointed pause/resume and quota-pause tests. |
+| OpenAI capacity is per-run | Atomically reserve tokens in a durable provider/model/quota-scope counter and include active legacy submitted rows. | Multi-run scope and legacy-reservation tests. |
+| A success clears another failure's backoff | Polling aggregates its cycle before retry-state mutation; a completed batch does not clear unrelated retry delay. | Mixed failed/completed batch backoff tests. |
+| `wait(timeout)` overruns | Thread an absolute deadline through refresh, ingestion, polling, and sleep decisions. | Blocked-ingestion/provider and sleep deadline tests. |
+| Cancellation waits only on stale backoff | Cancellation clears retry backoff after its final active batch drains. | Cancelled terminal quota/error regression test. |
+
+## Storage migration contract
+
+The schema advances from 4 to 6. Version 5 establishes the generic
+`run_ingest_state` invariant and conservative marker backfill. Version 6 adds
+`submission_intents` and `capacity_scopes`. Both upgrades are additive,
+performed during storage startup, and retain all existing rows. See
+[`STORAGE_MIGRATIONS.md`](STORAGE_MIGRATIONS.md) for operator guidance.
+
+## Non-goals
+
+- Background polling, asynchronous ingestion, or forcibly cancelling arbitrary
+  user/provider calls.
+- Provider-side remote cancellation guarantees.
+- Fresh-process resume for arbitrary non-checkpointable iterables.
+- Universal provider idempotency/discovery for `create_batch`.
+- Cross-store or credential-derived capacity coordination.
+
+## Evidence contract
+
+The deterministic evidence for this release is the fake-provider, memory,
+SQLite, and storage-contract suite. It must retain at least 85% coverage and
+pass type checking, linting, strict documentation build, package build, and
+`git diff --check` on the release commit.
+
+No paid live-provider smoke is required before push: all release criteria are
+directly represented by deterministic fake-provider/storage regressions. A live
+provider smoke, when credentials are available, remains an operational
+post-merge check rather than proof of these local state-machine invariants.