Document Celery subinterpreter task seams (#66)

docs/adr/adr-003-celery-worker-scaffold.md

@@ -86,6 +86,11 @@ story across developer machines and CI runners.
   process fan-out and the repository's opt-in interpreter-pool path for
   selected pure-Python workloads, with the enabling knobs documented in the
   same place.
+- CPU tasks running in `prefork` workers may optionally delegate inner fan-out
+  to `InterpreterPoolCpuTaskExecutor`, enabled with
+  `EPISODIC_USE_INTERPRETER_POOL`, dispatched with
+  `EPISODIC_INTERPRETER_POOL_MIN_ITEMS`, and capped with
+  `EPISODIC_INTERPRETER_POOL_MAX_WORKERS`.
 - Future roadmap items can add Celery tasks by extending typed dependency seams
   rather than introducing ad hoc globals.
 
@@ -103,7 +108,9 @@ story across developer machines and CI runners.
 ## References
 
 - [docs/execplans/1-5-2-scaffold-celery-workers-with-rabbit-mq-integration.md](../execplans/1-5-2-scaffold-celery-workers-with-rabbit-mq-integration.md)
+- [docs/execplans/upgrade-python-to-3-14-adopt-concurrent-interpreters.md](../execplans/upgrade-python-to-3-14-adopt-concurrent-interpreters.md)
 - [docs/episodic-podcast-generation-system-design.md](../episodic-podcast-generation-system-design.md)
+- [episodic/concurrent_interpreters.py](../../episodic/concurrent_interpreters.py)
 - [episodic/worker/topology.py](../../episodic/worker/topology.py)
 - [episodic/worker/runtime.py](../../episodic/worker/runtime.py)
 - [episodic/worker/tasks.py](../../episodic/worker/tasks.py)

docs/developers-guide.md

@@ -49,7 +49,7 @@ The target runs the Hecate architecture import-boundary checker, Ruff, and a
 focused Pylint 4 pass. The Pylint pass is invoked through
 `uv tool run --python pypy` with the pinned `pylint-pypy-shim` wrapper from
 [github.com/leynos/pylint-pypy-shim](https://github.com/leynos/pylint-pypy-shim).
- That wrapper installs the PyPy-specific Astroid compatibility patch before
+That wrapper installs the PyPy-specific Astroid compatibility patch before
 delegating to Pylint.
 
 Pylint's message selection is allow-listed in `pyproject.toml` with
@@ -210,6 +210,41 @@ When adding new worker tasks:
   adapters directly into task code.
 - Extend `SCAFFOLD_TASK_WORKLOADS` and the topology-backed routing metadata, so
   the new task's queue assignment remains explicit.
+- For CPU-bound tasks on `episodic.cpu` that can split pure-Python inner work
+  into independent items, build the task-level executor from the environment:
+
+  ```python
+  import os
+
+  from episodic.concurrent_interpreters import (
+      build_cpu_task_executor_from_environment,
+  )
+
+  executor = build_cpu_task_executor_from_environment(os.environ)
+  try:
+      results = await executor.map_ordered(pure_python_fn, items)
+  finally:
+      shutdown = getattr(executor, "shutdown", None)
+      if shutdown is not None:
+          shutdown()
+  ```
+
+  `EPISODIC_USE_INTERPRETER_POOL=1` enables `InterpreterPoolCpuTaskExecutor`
+  when the runtime supports interpreter pools, and
+  `EPISODIC_INTERPRETER_POOL_MAX_WORKERS` caps its worker count. Keep
+  `EPISODIC_INTERPRETER_POOL_MIN_ITEMS` as task-level fan-out policy, not
+  Celery pool configuration. The task-level owner is responsible for executor
+  lifetime and cleanup; inline executors do not need shutdown, while
+  interpreter-pool executors must be shut down when their fan-out operation or
+  explicit worker-scoped owner is finished.
+- Export CPU-task executor metrics through `CpuTaskExecutorMetricsPort`; it
+  extends the shared `BoundedValueMetricsPort` in `episodic/metrics_ports.py`
+  for executor selection, interpreter-pool lifecycle, map item count, and
+  shutdown-latency collection. Keep labels bounded to low-cardinality outcome
+  and reason values. Wire production backends through
+  `DefaultWeightingStrategy(metrics=...)` or by calling
+  `build_cpu_task_executor_from_environment(..., metrics=...)` directly at the
+  composition root.
 
 ## Database migrations
 
@@ -238,7 +273,7 @@ The `make check-migrations` target detects drift between the ORM models and the
 applied migration history. It starts an ephemeral Postgres via py-pglite,
 applies all Alembic migrations, and uses
 `alembic.autogenerate.compare_metadata()` to compare the migrated schema against
- `Base.metadata`. If they differ, the check exits non-zero and reports the
+`Base.metadata`. If they differ, the check exits non-zero and reports the
 discrepancies.
 
 Run it locally before committing model changes:
@@ -308,7 +343,7 @@ Key expectations:
   fixtures raise a clear error instead of silently skipping tests.
 - `make check-migrations` uses the same database technology, but a separate
   bootstrap path. `episodic/canonical/storage/migration_check.py` starts a plain
-   `PGliteManager`, creates an async SQLAlchemy engine from
+  `PGliteManager`, creates an async SQLAlchemy engine from
   `config.get_connection_string()`, applies Alembic migrations, and compares
   the migrated schema against `Base.metadata`.
 
@@ -356,7 +391,7 @@ The `SqlAlchemyUnitOfWork` manages transaction boundaries:
 
 Database-level constraints (unique slugs, foreign keys, and CHECK constraints
 such as the weight bound on source documents) are enforced by Postgres and raise
- `sqlalchemy.exc.IntegrityError` on violation.
+`sqlalchemy.exc.IntegrityError` on violation.
 
 ### Architecture enforcement
 
@@ -575,10 +610,11 @@ Pedante and Chrono are implemented in the `episodic/qa/` package.
 
 - `episodic/qa/chrono.py` contains `ChronoRuntimeEstimator`, typed
   request/result objects, estimator metadata, the deterministic local
-  spoken-runtime heuristic, `ChronoMetricsPort`, and `ChronoClockPort`. The
-  module delegates TEI P5 parsing and spoken-text extraction to
-  `tei-rapporteur`; it must not add a separate XML parser or local TEI
-  traversal path.
+  spoken-runtime heuristic, `ChronoMetricsPort`, and `ChronoClockPort`.
+  `ChronoMetricsPort` extends the shared `BoundedMetricsPort` in
+  `episodic/metrics_ports.py`. The module delegates TEI P5 parsing and
+  spoken-text extraction to `tei-rapporteur`; it must not add a separate XML
+  parser or local TEI traversal path.
 - `episodic/qa/chrono_langgraph.py` contains the in-process LangGraph seam for
   running Chrono as a QA graph node without attaching Large Language Model
   (LLM) usage metadata.
@@ -755,7 +791,7 @@ async def enrich(llm_port, script_tei_xml: str) -> str:
   `<div type="chapters">` element into the TEI body using the representation
   defined by
   [`adr-008-chapter-marker-tei-representation.md`](adr/adr-008-chapter-marker-tei-representation.md).
-   The `<list>` contains one `<item>` per chapter, `<label>` carries the title,
+  The `<list>` contains one `<item>` per chapter, `<label>` carries the title,
   `@n` stores the required start time, and `@corresp` stores an optional source
   locator. Optional DTO `end` and `duration` values are validated but not
   emitted into TEI until the TEI tooling exposes supported attributes.
@@ -932,13 +968,12 @@ stored planner-result payloads.
 ### Testing the orchestration slice
 
 - Unit coverage for DTO validation, planner behaviour, orchestration dispatch,
-  show-notes execution, guest-bio execution, and properties lives in the
-  focused `tests/test_orchestration_*.py`, `tests/test_show_notes_executor.py`,
-  and `tests/test_guest_bios_executor.py` modules.
+  show-notes execution, guest-bio execution, and properties lives in the focused
+  `tests/test_orchestration_*.py`, `tests/test_show_notes_executor.py`, and
+  `tests/test_guest_bios_executor.py` modules.
 - Issue `#72` property coverage lives in
   `tests/test_orchestration_properties.py` and the focused sibling modules for
-  config/model-tier boundaries, planner format errors, and LangGraph
-  invariants.
+  config/model-tier boundaries, planner format errors, and LangGraph invariants.
 - `tests/test_generation_orchestration_snapshots.py` pins planner format-error
   messages and orchestration artefacts with Syrupy snapshots.
 - LangGraph seam coverage lives in
@@ -1006,7 +1041,9 @@ testing and initial deployments:
   coefficients from the series configuration or defaults. The configuration
   dictionary may contain a `"weighting"` key with `"quality_coefficient"`
   (default 0.5), `"freshness_coefficient"` (default 0.3), and
-  `"reliability_coefficient"` (default 0.2).
+  `"reliability_coefficient"` (default 0.2). Pass optional `metrics=` when
+  constructing the strategy so production deployments can wire
+  `CpuTaskExecutorMetricsPort` through to the environment-built executor.
 - `HighestWeightConflictResolver` — selects the highest-weighted source as
   canonical; all others are rejected with provenance preserved.
 
@@ -1109,9 +1146,9 @@ def configure_logging(
 ```
 
 `level` is matched case-insensitively against `LogLevel` members. Returns a
-`tuple[LogLevel, bool]` — the normalized effective level and a flag that is `True`
-when the default (`INFO`) was substituted because the input was absent or
-unrecognized. The first element is always a `LogLevel` member; because
+`tuple[LogLevel, bool]` — the normalized effective level and a flag that is
+`True` when the default (`INFO`) was substituted because the input was absent
+or unrecognized. The first element is always a `LogLevel` member; because
 `LogLevel` is a `StrEnum`, those values are also `str` instances. Passing
 `"WARN"` (any case) normalizes to `WARNING` and emits a `DeprecationWarning`.
 The `force` parameter is forwarded directly to `femtologging.basicConfig`.

docs/episodic-podcast-generation-system-design.md

@@ -417,13 +417,13 @@ The following rules are normative for LangGraph nodes and Celery tasks:
 
 #### Speech synthesis contracts
 
-The speech synthesis boundary separates authorial data from provider
-parameters. `TTSPort` remains the single-speaker segment renderer used for
-ordinary narration and partial regeneration. `DialogueSpeechPort` is a separate
-optional port for providers that render multiple speaker turns as one
-conversation-aware artefact. This split preserves editability for per-segment
-stems whilst allowing providers such as Inworld Realtime or ElevenLabs
-text-to-dialogue to use conversational context when the series profile opts in.
+The speech synthesis boundary separates authorial data from provider parameters.
+`TTSPort` remains the single-speaker segment renderer used for ordinary
+narration and partial regeneration. `DialogueSpeechPort` is a separate optional
+port for providers that render multiple speaker turns as one conversation-aware
+artefact. This split preserves editability for per-segment stems whilst
+allowing providers such as Inworld Realtime or ElevenLabs text-to-dialogue to
+use conversational context when the series profile opts in.
 
 Every speech request is derived from canonical TEI P5:
 
@@ -491,11 +491,11 @@ per-segment stems for workflows that require surgical partial regeneration.
   provider calls so renders remain correlated end to end, with spans around
   pronunciation lookup and provider dispatch.
 - Classify failures as transport, provider, capability, pronunciation
-  resolution, validation, or post-render so dashboards and alerts remain
-  stable across vendors.
+  resolution, validation, or post-render so dashboards and alerts remain stable
+  across vendors.
 - Alert on synthesis failures and capability mismatches using aggregate
-  metrics: page when the failure rate exceeds 5 percent over 15 minutes or
-  five consecutive renders fail, warn when repeated unsupported-capability
+  metrics: page when the failure rate exceeds 5 percent over 15 minutes or five
+  consecutive renders fail, warn when repeated unsupported-capability
   diagnostics reach three in 30 minutes, and page when an approved
   pronunciation cannot be resolved.
 
@@ -549,10 +549,10 @@ erDiagram
   SPEECH_RENDER_REQUESTS ||--o{ SPEECH_RENDER_ARTIFACTS : produces
 ```
 
-_Caption: Speech synthesis entity relationships. Pronunciation entries have
-one or more realizations; voice personas and provider capabilities are selected
-for speech render requests; each speech render request produces one or more
-speech render artefacts._
+_Caption: Speech synthesis entity relationships. Pronunciation entries have one
+or more realizations; voice personas and provider capabilities are selected for
+speech render requests; each speech render request produces one or more speech
+render artefacts._
 
 #### Pronunciation repository
 
@@ -717,8 +717,8 @@ sequenceDiagram
     participant InlineCpuTaskExecutor
     participant InterpreterPoolCpuTaskExecutor
 
-    Caller->>DefaultWeightingStrategy: __init__(cpu_executor=None, min_parallel_items=None)
-    DefaultWeightingStrategy->>DefaultWeightingStrategy: build_cpu_task_executor_from_environment()
+    Caller->>DefaultWeightingStrategy: __init__(cpu_executor=None, metrics=None, min_parallel_items=None)
+    DefaultWeightingStrategy->>DefaultWeightingStrategy: build_cpu_task_executor_from_environment(os.environ, metrics=metrics)
     DefaultWeightingStrategy-->>CpuTaskExecutor: selected executor instance
 
     Caller->>DefaultWeightingStrategy: compute_weights(sources, series_configuration)
@@ -874,8 +874,8 @@ ports, including optional
 adapters that expose tool catalogues. Skill-based tool loading narrows the
 available tool set per workflow, reducing context size and guiding model choice.
 
-Roadmap item `2.4.1` now ships the first narrow implementation of that design
-in `episodic/orchestration/`. `StructuredGenerationPlanner` calls `LLMPort` once
+Roadmap item `2.4.1` now ships the first narrow implementation of that design in
+`episodic/orchestration/`. `StructuredGenerationPlanner` calls `LLMPort` once
 to obtain strict JSON, parses it into a typed `ExecutionPlan`, and records the
 planning model separately from the execution model selected in
 `GenerationOrchestrationConfig`. `StructuredPlanningOrchestrator` then executes
@@ -1456,7 +1456,7 @@ Agentic workflow behaviour is configurable per series profile:
   reusable reference document, including hashes and author metadata.
 - `reference_document_bindings` links pinned reference revisions to a target
   context (series profile, episode template, or ingestion run), with an optional
-   `effective_from_episode_id` anchor for forward-only applicability.
+  `effective_from_episode_id` anchor for forward-only applicability.
 - `uploads` records file upload metadata, content hashes, content type, size
   limits, storage location, and idempotency keys before uploaded material is
   attached to ingestion jobs.
@@ -1542,7 +1542,7 @@ profile guidance changes mid-series, editors will add a new revision and will
 create a binding with `effective_from_episode_id`; that revision will apply
 from the anchor episode onwards until superseded. Ingestion workflows will
 resolve these bindings and will snapshot selected revisions into ingestion-bound
- `source_documents`, preserving reproducible TEI provenance while allowing
+`source_documents`, preserving reproducible TEI provenance while allowing
 independent document reuse across jobs.
 
 TEI header payloads include an `episodic_provenance` extension with

docs/execplans/upgrade-python-to-3-14-adopt-concurrent-interpreters.md

@@ -123,6 +123,10 @@ Delivered artefacts:
   `EPISODIC_INTERPRETER_POOL_MIN_ITEMS`.
 - Benchmark CLI added at `episodic/benchmarks/interpreters.py`.
 - User and architecture docs updated with feature-flag guidance.
+- PR `#99` resolved the Celery integration gap called out in `Surprises &
+  discoveries` for issue `#66`: CPU-bound Celery tasks now have documented
+  task-level guidance and tests showing how to call
+  `build_cpu_task_executor_from_environment()` from within an eager task body.
 
 Validation evidence:
 

docs/users-guide.md

@@ -175,7 +175,7 @@ Reusable reference-document workflows currently support:
   Stale updates return `409 Conflict`.
 - Creating and listing immutable document revisions at
   `POST /v1/series-profiles/{profile_id}/reference-documents/{document_id}/revisions`
-   and
+  and
   `GET /v1/series-profiles/{profile_id}/reference-documents/{document_id}/revisions`.
 - Creating, listing, and fetching target bindings at
   `POST /v1/reference-bindings`, `GET /v1/reference-bindings`, and
@@ -282,6 +282,15 @@ Optional interpreter-pool flags:
 - `EPISODIC_INTERPRETER_POOL_MAX_WORKERS` caps the interpreter-pool worker
   count when that path is enabled.
 
+CPU-task executor metrics are exported through the shared
+`CpuTaskExecutorMetricsPort`, which extends `BoundedValueMetricsPort` in
+`episodic/metrics_ports.py`. Deployments that wire a metrics backend through
+that port can collect executor selection, interpreter-pool lifecycle, map item
+count, and shutdown-latency signals with bounded labels. In ingestion
+pipelines, pass the same metrics sink to
+`DefaultWeightingStrategy(metrics=...)` so weighting fan-out records executor
+observability in production.
+
 Current queue model:
 
 - `episodic.tasks` topic exchange

episodic/canonical/adapters/weighting.py

@@ -22,6 +22,7 @@
 from episodic.canonical.ingestion import NormalizedSource, WeightingResult
 from episodic.concurrent_interpreters import (
     CpuTaskExecutor,
+    CpuTaskExecutorMetricsPort,
     build_cpu_task_executor_from_environment,
 )
 
@@ -118,18 +119,41 @@ class DefaultWeightingStrategy:
     Coefficients are read from the series configuration under the
     ``"weighting"`` key, falling back to defaults (quality=0.5, freshness=0.3,
     reliability=0.2) when absent. Results are clamped to [0, 1].
+
+    When ``cpu_executor`` is omitted, the strategy builds one from the process
+    environment and forwards optional ``metrics`` to
+    ``build_cpu_task_executor_from_environment`` so production deployments can
+    collect interpreter-pool observability at the composition root.
     """
 
     def __init__(
         self,
         *,
         cpu_executor: CpuTaskExecutor | None = None,
         min_parallel_items: int | None = None,
+        metrics: CpuTaskExecutorMetricsPort | None = None,
     ) -> None:
+        """Initialise the weighting strategy.
+
+        Parameters
+        ----------
+        cpu_executor : CpuTaskExecutor | None
+            Optional executor override. When omitted, the strategy selects one
+            from the process environment.
+        min_parallel_items : int | None
+            Minimum source count before interpreter-backed dispatch is attempted.
+            ``None`` reads ``EPISODIC_INTERPRETER_POOL_MIN_ITEMS``.
+        metrics : CpuTaskExecutorMetricsPort | None
+            Metrics sink forwarded to the environment-built executor. Ignored
+            when ``cpu_executor`` is supplied explicitly.
+        """
         self._cpu_executor = (
             cpu_executor
             if cpu_executor is not None
-            else build_cpu_task_executor_from_environment()
+            else build_cpu_task_executor_from_environment(
+                os.environ,
+                metrics=metrics,
+            )
         )
         if min_parallel_items is None:
             self._min_parallel_items = _parse_min_parallel_items(