feat(core): Pipeline + Batcher primitives (Step 1.1d)

TRACKER.md

@@ -10,7 +10,7 @@
 
 **Last updated:** 2026-05-24
 **Current phase:** Phase 1 — Ingestion + Knowledge Store
-**Next action:** Phase 1 Step 1.1d — Pipeline + Batcher primitives in `rag-core` (async DAG with bounded queues; DataLoader-pattern Batcher middleware coalescing concurrent SPI calls)
+**Next action:** Phase 1 Step 1.1e — Cache SPI split (`EmbeddingCache` / `RetrievalCache` / `AnswerCache`) + hot-path discipline + async telemetry path with drop-on-overflow counter
 
 > **Refactor window (Steps 1.1a–1.1f):** Before resuming the connectors framework (1.2), we insert a six-step refactor that locks in architecture + optimization decisions which are very expensive to retrofit later (PolicyEngine PDP, RequestContext-threaded SPIs, split Retrieval/Index backends, bulk + streaming + ID-only methods, Pipeline + Batcher primitives, three-way cache split, hot-path discipline). See [docs/adr/ADR-0005…0009] and [docs/architecture/policy-engine.md], [request-context.md], [caching.md], [performance.md].
 
@@ -71,7 +71,7 @@
 | 1.1a | Core type & SPI refactor | ✅ | `build/phase-1/step-1.1a-core-type-spi-refactor` | [#44](https://github.com/officialCodeWork/AgentContextOS/pull/44) | `RequestContext` frozen model threaded through every SPI; `tenant_id` + `acl_labels` typed required on `Chunk`/`Embedding` (not metadata dict); `trust_level` on `Chunk` for prompt-injection defense; `dtype` on `Embedding` (float32/int8/binary); `BlobRef` for lazy chunk text; `QueryPlan` + `ChunkRef` + `Cost` + `PlanNode` types; typed `StageEvent`. `tests/contract/spi_signature.py` linter (RequestContext-first); rag-backends (`PgVectorStore`, `QdrantVectorStore`, `RedisCache`, `S3Storage`, `LocalFileStorage`) migrated; conformance + integration tests updated; `Budget.spend()` for agent-loop sub-turn budgets; schemas regenerated. ADR-0005 / ADR-0007 / ADR-0008 / ADR-0009 referenced. |
 | 1.1b | SPI split — Retrieval/Index, bulk + streaming + ID-only | ✅ | `build/phase-1/step-1.1b-spi-split-retrieval-index` | [#45](https://github.com/officialCodeWork/AgentContextOS/pull/45) | Split `VectorStore`/`KeywordStore`/`GraphStore` into `*RetrievalBackend` (read) + `*IndexBackend` (write) composite ABCs. `retrieve_ids` returns `list[ChunkRef]`; `hydrate` lives on the retrieval side (keyword full-Chunk; vector pass-through). Bulk: `bulk_index`/`bulk_delete` (+ graph bulk node/edge variants); streaming: `stream_index` async-iterator default that batches into `bulk_index`. `Embedder` split into single `embed` + canonical `bulk_embed`. New `IndexHint` + `WriteVolume` types passed to writes (per ADR-0009). Noop impls, `PgVectorStore`, `QdrantVectorStore` migrated; conformance + integration tests updated; `spi_signature.py` extended to enforce the split. Schemas regenerated (`IndexHint.json`). |
 | 1.1c | PolicyEngine package | ✅ | `build/phase-1/step-1.1c-policy-engine-package` | [#46](https://github.com/officialCodeWork/AgentContextOS/pull/46) | New `packages/policy/` (`rag-policy` v0.1.0): `PolicyEngine` SPI + `NoopPolicyEngine` (always-ALLOW with tenant-scoped `filter_pushdown`); `PolicyDecision` enum (read_chunk / ingest_doc / egress_text / quota_check / rate_limit / execute_plan); `PolicyResult` (allow/deny/transform) with predicate helpers; `QuotaSubject` / `RateLimitSubject`; `FilterExpr` mini-language (Eq / AnyIn / And / Or / Not / TrueExpr) returned by `filter_pushdown`; `PolicyWriter` facade mirroring `AuditWriter` and emitting `policy.decision` structured logs via `rag-observability`. Coverage linter `tests/policy/coverage.py` greps for governance-relevant SPI calls without adjacent `PolicyEngine`/`PolicyWriter` consultation, with file allowlist that consumers shrink as they wire the PDP in. Workspace + pytest pythonpath updated. 20 conformance tests added. ADR-0005 finalized. |
-| 1.1d | Pipeline + Batcher primitives | ⏳ | 1.1a | — | `Pipeline` primitive in `rag-core`: async DAG with bounded queues, per-stage worker counts, backpressure (used by Step 1.10 write path). `Batcher[Req, Resp]` middleware (DataLoader pattern) coalescing concurrent SPI calls into batched provider calls; sits under Embedder/Reranker SPIs. |
+| 1.1d | Pipeline + Batcher primitives | 🚧 | `build/phase-1/step-1.1d-pipeline-batcher-primitives` | — | `Pipeline` primitive in `rag-core` (`rag_core.pipeline`): async DAG with bounded queues, per-stage worker counts, backpressure, joiner-based shutdown handling per-stage worker-count differences, `on_error="fail"`/`"skip"` policies, optional `error_handler`. `Batcher[Req, Resp]` middleware (`rag_core.batcher`, DataLoader pattern) coalescing concurrent `load(req)` calls into batched `batch_fn(reqs)` invocations; size + time triggers (`max_batch_size`, `max_wait_ms`); per-request `asyncio.Future` for isolated error propagation; `flush()` for deterministic shutdown. PEP-695-typed generics. 22 unit tests (8 Batcher + 14 Pipeline). `docs/architecture/pipeline-batcher.md` + `rag-core` reference + `performance.md` backlinks. `rag-core` 0.5.0 → 0.6.0. |
 | 1.1e | Cache SPI split + perf discipline + async telemetry | ⏳ | 1.1a | — | Split `Cache` into `EmbeddingCache` (key model_id+text_hash), `RetrievalCache` (plan_hash+corpus_version), `AnswerCache` (plan_hash+corpus_version+policy_version). Each has distinct invalidation. Hot-path convention doc (Pydantic at SPI boundary, `model_construct`/msgspec inside). Async telemetry path with bounded buffer + drop-on-overflow counter. |
 | 1.1f | ADRs 0005–0009 + reviewer checklist | ⏳ | 1.1a–1.1e | — | ADR-0005 (PolicyEngine PDP), ADR-0006 (two-stage reranker default), ADR-0007 (tiered storage with BlobRef), ADR-0008 (cost-aware planner replacing reactive fallback), ADR-0009 (vector index strategy + quantization). Reviewer checklist in `docs/architecture/performance.md`. |
 | 1.2 | Connectors framework | ⏳ | 1.1a–1.1f | — | `Connector` SPI implementation (now receives `RequestContext` and `ConnectorState` watermark); built-in: filesystem, S3, GCS; crawler base class |
@@ -217,6 +217,7 @@
 | [#46](https://github.com/officialCodeWork/AgentContextOS/pull/46) | feat(policy): rag-policy package — PolicyEngine PDP + coverage linter (Step 1.1c) | `build/phase-1/step-1.1c-policy-engine-package` | ✅ Merged | 2026-05-24 |
 | [#47](https://github.com/officialCodeWork/AgentContextOS/pull/47) | chore(schemas): track generated JSON Schemas + add drift CI gate | `chore/track-generated-schemas-drift-gate` | ✅ Merged | 2026-05-24 |
 | [#48](https://github.com/officialCodeWork/AgentContextOS/pull/48) | chore(ci): lint scripts/ + ignore S603/S607 in scripts/** | `chore/lint-ignore-subprocess-in-scripts` | ✅ Merged | 2026-05-24 |
+| [#49](https://github.com/officialCodeWork/AgentContextOS/pull/49) | chore(tracker): sync PR links for steps 1.1a/1.1c + log #46–#48 | `chore/tracker-sync-pr44-48` | ✅ Merged | 2026-05-24 |
 
 ---
 

docs/README.md

@@ -10,6 +10,7 @@
 | [policy-engine.md](architecture/policy-engine.md) | `PolicyEngine` (PDP) — single decision point for ACL, PII, quotas, redaction; replaces scattered governance checks |
 | [caching.md](architecture/caching.md) | Three-cache split: `EmbeddingCache`, `RetrievalCache`, `AnswerCache` — distinct invalidation rules |
 | [performance.md](architecture/performance.md) | Hot-path discipline, per-SPI p99 budgets, async telemetry, reviewer checklist |
+| [pipeline-batcher.md](architecture/pipeline-batcher.md) | `Pipeline` (async DAG, bounded queues, per-stage workers) + `Batcher` (DataLoader-pattern coalescing) primitives — Step 1.1d |
 | [eval-skeleton.md](architecture/eval-skeleton.md) | Eval framework architecture: golden-set schema, metric functions, RAGAS spike, `ragctl eval` CLI, extension points |
 | [iac.md](architecture/iac.md) | IaC overview: Terraform module design, Helm chart structure, dev/prod environments, extension points |
 | [storage-backends.md](architecture/storage-backends.md) | Storage backend architecture: PgVector, Qdrant, Redis, S3/MinIO, tenant isolation, integration test strategy |
@@ -20,7 +21,7 @@
 |------|-------------|
 | [ragctl.md](reference/ragctl.md) | Full `ragctl` command reference — public usage, internals, extension points |
 | [backends.md](reference/backends.md) | `rag-backends` reference — PgVectorStore, QdrantVectorStore, RedisCache, S3Storage, LocalFileStorage |
-| [rag-core.md](reference/rag-core.md) | `rag-core` type surface — `RequestContext`, `Budget`, `BlobRef`, `QueryPlan`, `ChunkRef`, `StageEvent` |
+| [rag-core.md](reference/rag-core.md) | `rag-core` type surface — `RequestContext`, `Budget`, `BlobRef`, `QueryPlan`, `ChunkRef`, `StageEvent`, `Pipeline`, `Batcher` |
 | [rag-policy.md](reference/rag-policy.md) | `rag-policy` reference — `PolicyEngine`, `PolicyWriter`, `PolicyDecision`, `PolicyResult`, `FilterExpr` |
 
 ## guides/

docs/architecture/performance.md

@@ -107,5 +107,6 @@ Apply to every PR that touches an SPI, a hot path, or an SPI consumer:
 - [request-context.md](request-context.md) — the per-request envelope.
 - [policy-engine.md](policy-engine.md) — governance PDP.
 - [caching.md](caching.md) — three caches.
+- [pipeline-batcher.md](pipeline-batcher.md) — `Pipeline` and `Batcher` primitives (bounded queues + DataLoader-pattern coalescing).
 - [ADR-0006](../adr/ADR-0006-two-stage-rerank.md), [ADR-0008](../adr/ADR-0008-cost-aware-planner.md), [ADR-0009](../adr/ADR-0009-vector-index-strategy.md) — performance-relevant decisions.
 - TRACKER.md Steps 1.1a–1.1f (refactor window), 4.6 (latency tuning), 7.1 (load testing).

docs/architecture/pipeline-batcher.md

@@ -0,0 +1,158 @@
+# Pipeline + Batcher primitives
+
+## Overview
+
+`rag-core` ships two cross-cutting concurrency primitives used throughout
+the ingest and retrieval paths:
+
+- **`Pipeline`** — a linear async DAG with bounded queues and per-stage
+  worker counts. Lives in `rag_core.pipeline`. Drives the Step 1.10 write
+  path (connector → parse → chunk → enrich → PII → embed → store) and any
+  other multi-stage processing.
+- **`Batcher[Req, Resp]`** — a DataLoader-pattern middleware that coalesces
+  concurrent `load(req)` calls into a single `batch_fn(reqs)` invocation.
+  Lives in `rag_core.batcher`. Sits under every Embedder / Reranker / LLM
+  adapter that talks to a billed external API, so that N in-flight concurrent
+  calls collapse to one provider call.
+
+They are introduced in Step 1.1d and locked in ahead of Step 1.2 because
+both shapes show up everywhere — adding them once, correctly, with the
+right backpressure and error semantics is far cheaper than retrofitting
+ten consumers later.
+
+## Usage
+
+### `Pipeline`
+
+Each stage is an `async fn(item) -> result`. The result type is overloaded:
+
+| Return value      | Semantics                                          |
+|-------------------|----------------------------------------------------|
+| a single object   | forwarded to the next stage as one item            |
+| `list[...]`       | each element forwarded separately (fan-out)        |
+| `None`            | dropped (filter)                                   |
+
+Build with `add_stage`, then drive with `run(source)`:
+
+```python
+from rag_core import Pipeline
+
+async def parse(doc):     return await parse_async(doc)
+async def chunk(parsed):  return chunker.split(parsed)         # → list[Chunk]
+async def embed(chunk):   return await embedder.embed(ctx, chunk.text, chunk.id)
+
+pipe = (
+    Pipeline()
+    .add_stage("parse", parse, workers=2)
+    .add_stage("chunk", chunk, workers=2)
+    .add_stage("embed", embed, workers=8)        # downstream wider than upstream
+)
+
+async for embedding in pipe.run(source_documents()):
+    await vector_index.append(ctx, embedding)
+```
+
+Bounded-queue backpressure is automatic — if the consumer of `pipe.run`
+stops pulling, the embed workers fill their input queue, the chunk workers
+fill theirs, and so on back to the source iterator.
+
+#### Error policy
+
+- `on_error="fail"` (default) — the first stage exception aborts the whole
+  pipeline. The exception surfaces at the consumer as `PipelineError` with
+  the original error in `__cause__`. Matches the fail-fast convention.
+- `on_error="skip"` — the failing item is dropped; the pipeline keeps
+  draining. Useful for best-effort backfills.
+
+In either mode, an optional `error_handler(stage_name, item, exc)`
+callback fires for every failure — wire it up to structured logging or
+metrics. Error handlers must not raise; exceptions from the handler are
+swallowed to keep the pipeline draining.
+
+### `Batcher`
+
+```python
+from rag_core import Batcher
+
+async def call_provider(reqs: list[str]) -> list[list[float]]:
+    return await openai.embeddings.create(input=reqs, model="...")
+
+embed_batcher: Batcher[str, list[float]] = Batcher(
+    call_provider,
+    max_batch_size=64,
+    max_wait_ms=5,
+)
+
+# Inside an Embedder.embed implementation:
+vec = await embed_batcher.load(text)
+```
+
+A pending batch flushes when *either* `max_batch_size` is reached *or*
+`max_wait_ms` elapses since the first request landed. Default window is
+5 ms — long enough to coalesce a burst, short enough not to add visible
+latency to single requests.
+
+`flush()` forces an immediate flush of whatever is pending. Use at
+shutdown or in deterministic tests.
+
+## Internals
+
+### Pipeline shutdown — joiners and sentinels
+
+Each stage transition is governed by a *joiner* task that awaits all of
+the upstream stage's workers and then injects exactly `next_stage.workers`
+`_DONE` sentinels into the next queue. This lets per-stage worker counts
+differ without losing or leaking sentinels. The feeder injects sentinels
+for the first stage.
+
+When a stage worker hits an exception under `on_error="fail"`:
+
+1. The worker calls `error_handler` (if set).
+2. It pushes a `_Failure(PipelineError(...))` marker into its output queue
+   and sets the run's abort event.
+3. The marker propagates through downstream stages (workers forward it on
+   sight and exit). At the output queue, the consumer pulls the marker
+   and `raise`s.
+4. Other workers on the failing stage drain their queues without invoking
+   the stage function (the abort event short-circuits them) until they
+   see a `_DONE` sentinel and exit.
+5. Joiners complete normally; the consumer's `try/finally` in `run`
+   cancels any tasks that are still alive (e.g. the feeder waiting on a
+   full queue).
+
+This shape lets `aclose()` on the async iterator returned by `run` perform
+clean shutdown if the consumer breaks early — no stranded tasks, no
+leaked workers.
+
+### Batcher — single-flight flush task
+
+A pending batch holds at most one timer task. When a request arrives:
+
+- If the batch is empty, the request kicks off the timer.
+- If the batch reaches `max_batch_size`, the current pending slot is
+  rotated and the batch is flushed synchronously by the caller. The
+  outstanding timer is cancelled.
+
+`asyncio.Future` per request decouples error handling: an exception from
+`batch_fn` is set on every future in the batch, never on a single one.
+
+## Extension points
+
+- **Pipeline observation.** Wire an `error_handler` to emit structured
+  logs or `StageEvent` records; the gateway publishes `StageEvent`s over
+  SSE per `docs/architecture/RAG-Platform-HLD.md`.
+- **Pipeline shape.** Linear-only by design. A second primitive
+  (`Workflow`) can be added later if a real consumer needs a non-linear
+  DAG; do not generalize this one preemptively.
+- **Batcher under SPIs.** Every adapter for a billed external API should
+  wrap its provider call in a `Batcher`. The reviewer checklist in
+  [performance.md](performance.md) flags missing batchers in code review.
+
+## Related
+
+- [performance.md](performance.md) — hot-path discipline, p99 budgets,
+  reviewer checklist (where the Batcher rule is enforced).
+- [request-context.md](request-context.md) — the envelope every stage
+  receives via closure over `ctx`.
+- TRACKER.md Step 1.1d (this work) and Step 1.10 (Pipeline's first big
+  consumer — the ingest write path).

docs/reference/rag-core.md

@@ -221,6 +221,22 @@ Removing or renaming a field is a major change and requires an ADR.
 
 ---
 
+## Concurrency primitives
+
+`rag-core` also ships two cross-cutting concurrency primitives used by
+ingest and retrieval consumers (added in Step 1.1d):
+
+| Name | Module | Purpose |
+|---|---|---|
+| `Pipeline`, `Stage` | `rag_core.pipeline` | Async DAG with bounded queues + per-stage worker counts; drives the ingest write path |
+| `Batcher[Req, Resp]` | `rag_core.batcher` | DataLoader-pattern middleware; coalesces concurrent SPI calls into batched provider calls |
+
+Full design notes, usage, and error semantics live in
+[docs/architecture/pipeline-batcher.md](../architecture/pipeline-batcher.md).
+The reviewer checklist in
+[docs/architecture/performance.md](../architecture/performance.md) calls
+out where each must be used.
+
 ## Related
 
 - [docs/architecture/request-context.md](../architecture/request-context.md) — design rationale.

packages/core/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "rag-core"
-version = "0.3.0"
+version = "0.6.0"
 description = "AgentContextOS — core domain types, error hierarchy, and plugin SPIs"
 readme = "README.md"
 requires-python = ">=3.12"

packages/core/src/rag_core/__init__.py

@@ -5,6 +5,7 @@
 Eval domain types added in Step 0.8.
 """
 
+from rag_core.batcher import Batcher
 from rag_core.errors import (
     ACLDeniedError,
     AuthError,
@@ -46,6 +47,7 @@
 )
 from rag_core.logging import configure_logging, get_logger, reset_logging, set_log_context
 from rag_core.metrics import SpiMetrics
+from rag_core.pipeline import Pipeline, Stage
 from rag_core.spi import (
     LLM,
     OCR,
@@ -122,7 +124,7 @@
     WriteVolume,
 )
 
-__version__ = "0.5.0"
+__version__ = "0.6.0"
 
 __all__ = [
     # eval
@@ -208,6 +210,10 @@
     "SchemaValidationError",
     "TenantIsolationError",
     "TimeoutError",
+    # pipeline + batcher primitives
+    "Batcher",
+    "Pipeline",
+    "Stage",
     # telemetry + metrics
     "SpiMetrics",
     "configure_telemetry",