feat(batch): expose failure_policy/finalizer on SubJobEnqueuer.enqueue_batch (in-actor fan-out parity)

[rcbevans]

Problem

The new batch subsystem landed on the client path only. JobsClient.enqueue_batch / TaskQ.enqueue_batch accept failure_policy= and finalizer= (single-transaction batch row + children + finalizer), create a discoverable batches row, and give wait_for_batch its finalizer_job_id auto-exclusion and expect_at_least/on_empty safety. SubJobEnqueuer.enqueue_batch — the in-actor fan-out path, ctx.jobs.enqueue_batch(...) — accepts only (items, *, batch_id=None, connection=None) (src/taskq/client/_enqueuer.py:288-296). Consequences for apps whose fan-out happens inside actors (the documented pipeline idiom):

• AbortBatchAfter (Batch-level consecutive-failure abort policy #55) is undeclarable — the app-side failure-breaker counter has to be re-implemented after all.
• The load-bearing "finalizer enqueue is the last statement" invariant (Fan-out-then-finalize: an empty batch reports complete, and finalizer ordering is load-bearing #58) stays a comment-plus-regression-test instead of a same-transaction guarantee.
• No batches row is created, so list_batches (No way to enumerate or discover batches — batch_id is recoverable only if the caller stored it #59) cannot discover in-actor batches, and wait_for_batch on them falls back to the on_empty="error" no-row behavior.

This is the same surface-divergence shape as #57 (sub-job tags), which PR #65 fixed for single enqueue; the batch path diverged again when #62 landed.

Evidence (integration @ 63c25ed)

Client signature src/taskq/client/_jobs.py:392-399; batch-row creation gated on has_batch_extras (_jobs.py:536-537,560-565); sub-job signature _enqueuer.py:288-296. Related: the sub-job path also does not enforce MAX_BATCH_SIZE (client path caps at 1000, _jobs.py:470-473) — worth resolving or documenting while aligning the surfaces.

Why AACRTool needs it

Every fan-out in the AACRTool pipeline happens inside stage actors: s3-decompose enqueues the analyze batch via ctx.jobs.enqueue_batch(group, batch_id=b4) and then a finalizer via ctx.jobs.enqueue(...) — precisely the pattern #58 describes as hazardous. Declaring failure_policy=AbortBatchAfter(consecutive_failures=5) on that call would retire AACRTool's app-side consecutive-failure counter, and finalizer= would retire its ordering regression test. Neither is reachable without this parity.

Suggested API direction

Mirror the client signature on SubJobEnqueuer.enqueue_batch (failure_policy: BatchFailurePolicy | None = None, finalizer: EnqueueItem | None = None), reusing the same atomic path on the loop-scope connection; define the behavior under the transactional-simulation buffer (supports_transactional_simulation) so InMemoryBackend pipelines behave identically; add the MAX_BATCH_SIZE check (or document its deliberate absence).

Acceptance criteria

1. ctx.jobs.enqueue_batch(items, batch_id=…, failure_policy=…, finalizer=…) works from inside an actor with the same single-transaction semantics as the client path.
2. A batches row is created (discoverable via list_batches); wait_for_batch auto-excludes the finalizer.
3. Batch abort from an in-actor-declared policy cancels pending/scheduled children and surfaces BatchAbortedError to the finalizer.
4. InMemoryBackend parity + unit tests; e2e covering abort-from-sub-job and finalizer-in-transaction.