1/2 Flat topology — core seam: batch-branch router hook + frozen fan-out bound

.test_durations

@@ -5856,10 +5856,6 @@
     "tests/unit/pipelex/observer/test_local_observer.py::TestLocalObserver::test_observe_writes_one_jsonl_line_with_event_type[observe_before_run-before_run]": 0.0015739160007797182,
     "tests/unit/pipelex/observer/test_local_observer.py::TestLocalObserver::test_two_observe_calls_append_two_lines_in_order": 0.0020540419936878607,
     "tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_blueprint.py::TestPipeBatchBlueprint::test_pipe_dependencies_correct": 0.8473865840060171,
-    "tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_concurrency.py::TestResolveBatchMaxConcurrency::test_setting_translates_to_gather_bounded_argument[1-1]": 0.000396165982238017,
-    "tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_concurrency.py::TestResolveBatchMaxConcurrency::test_setting_translates_to_gather_bounded_argument[100-100]": 0.07548241699987557,
-    "tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_concurrency.py::TestResolveBatchMaxConcurrency::test_setting_translates_to_gather_bounded_argument[8-8]": 0.0007538750069215894,
-    "tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_concurrency.py::TestResolveBatchMaxConcurrency::test_setting_translates_to_gather_bounded_argument[unbounded-None]": 0.7548824160039658,
     "tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_input.py::TestPipeBatchValidateInputs::test_validate_inputs_valid_cases[valid_multiple_inputs-blueprint1]": 0.21890241600340232,
     "tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_input.py::TestPipeBatchValidateInputs::test_validate_inputs_valid_cases[valid_simple_batch-blueprint0]": 0.8536362920131069,
     "tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_validation.py::TestPipeBatchValidation::test_accepts_valid_batch_config": 0.7366558319918113,

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog
 
+## [Unreleased]
+
+### Added
+
+- **`run_batch_branch` router hook**: `PipeRouterProtocol` gained a second dispatch entry point, which `PipeBatch` uses for each per-item fan-out branch. It is the only signal a router gets that a dispatch is a batch branch rather than an ordinary step — a branch's `PipeJob` is otherwise indistinguishable from any other. The default body delegates to `run`, so in-process execution is unchanged and every existing router implementation keeps working untouched; a distributed backend can override it to give each branch its own isolation. Documented on the "Pipe Routing and Execution" page and listed in the Orchestrator SPI.
+
+### Fixed
+
+- **A dry run now identifies the pipe it is running**: `dry_run_pipe` stamps the running pipe onto the `JobMetadata` it hands down, exactly as `live_run_pipe` already did. Previously `job_metadata.pipe_code` kept whatever the caller passed — usually nothing — for the whole of a dry run, so anything that identifies a step by it (log correlation, and the per-step labelling a distributed backend derives) saw an anonymous step in DRY and a named one in LIVE. Telemetry stays live-only: `pipe_run_id` and `otel_context` still belong to a real run, and the dry copy clears `otel_context` rather than inheriting a live span.
+- **`PipeBatch` fan-out bound is frozen onto the run (Breaking)**: the `[pipelex.pipeline_execution_config] max_concurrency` setting is now read once, when the run's parameters are built, and carried on the run as `PipeRunParams.batch_max_concurrency` instead of being re-read inside every `PipeBatch`. Editing the config while a run is in flight no longer reshapes it. This closes a durable-execution hazard: the bound is also the chunk size that decides where a backend's task boundaries fall between branch dispatches, so a worker redeploy mid-run could make a replay group its dispatches differently from the recorded history. Breaking only for code that mutated `max_concurrency` mid-run and expected the change to take effect.
+
 ## [v0.42.0] - 2026-08-01
 
 ### Added

docs/building-methods/pipes/pipe-controllers/PipeBatch.md

@@ -24,6 +24,8 @@ This is the ideal controller for processing collections of documents, images, or
 
 To restore unbounded fan-out (every branch started at once), set `max_concurrency = "unbounded"`.
 
+The setting is read **once, when the run's parameters are built**, and then frozen onto the run (`PipeRunParams.batch_max_concurrency`) — every `PipeBatch` in that run, at any depth, uses the value that was in effect when the run started. Editing the config while a run is in flight does not reshape it. This matters most on a durable-execution backend, where the bound is also the chunk size that decides where the backend's task boundaries fall between branch dispatches: a value re-read mid-run could make a replay group its dispatches differently from the recorded history.
+
 Results always preserve input order regardless of the concurrency bound. If a branch fails, the failure propagates and the first error by input index wins.
 
 For durable, rate-limited execution of very large batches, run the pipeline on the Temporal track.

docs/under-the-hood/orchestrator-plugins.md

@@ -186,7 +186,7 @@ What an out-of-tree orchestrator imports *is* a contract. The SPI is a documente
 | Mode + delivery + errors | `pipelex.runtime_bridge.orchestration_mode` (`OrchestrationMode`, `DIRECT_ORCHESTRATION_MODE`), `pipelex.runtime_bridge.delivery_mode` (`DeliveryMode`), `pipelex.runtime_bridge.exceptions` (`MissingOrchestratorError`, `PipelexBridgeDispatchError`) |
 | Working-memory hydration | `pipelex.runtime_bridge.primitives.hydration` (re-hydrate `working_memory_raw` → typed `WorkingMemory`; stayed open because it is host-agnostic — used by core delivery and the open `pipelex-api` runner, and re-used across the boundary by `pipelex-transport`) |
 | Plugin contract | `pipelex.plugins.contract` (`PipelexPlugin`, `PLUGIN_API_VERSION`), `pipelex.plugins.registrar` (`PluginRegistrar` menu: `add_orchestrator`, `add_bundle_validator`, `add_http_error_mapper`, `claim_*`, `add_teardown`; read accessor: `get_http_error_mappers`), `pipelex.plugins.orchestrator_registry` (`OrchestratorProtocol`), `pipelex.plugins.bundle_validator_registry` (`BundleValidatorProtocol`, `BundleValidationVerdict`) |
-| Execution protocols | `PipeRouterProtocol`, `PipeRunProtocol`, `ContentGeneratorProtocol`, the task-manager protocol |
+| Execution protocols | `PipeRouterProtocol` (incl. the [`run_batch_branch`](./pipe-routing-and-execution.md#the-batch-branch-hook) fan-out hook — concrete default, override it only to isolate batch branches), `PipeRunProtocol`, `ContentGeneratorProtocol`, the task-manager protocol |
 | Payload / core types | `PipeJob`, `PipeOutput`, `DeliveryAssignment`, `WorkingMemory` (+ factory, `dump_for_transport`), `JobMetadata`, `LibraryCrate` |
 | Library + hub scoping | `set_current_library` / `get_current_library`, `scoped_pipe_router`, `get_class_registry` (per-call library hydration via `library_crate_dump`) |
 | Tracing / graph hooks | `trace_events`, `graph_tracer_manager`, `tracing_assembly` (per-step trace/usage events across the boundary) |

docs/under-the-hood/pipe-routing-and-execution.md

@@ -137,6 +137,19 @@ async def _run_pipe_job(self, pipe_job: PipeJob) -> PipeOutput:
 
 The router does not route by pipe type — it delegates to the pipe itself. Controllers handle their own orchestration internally.
 
+### The batch-branch hook
+
+`PipeRouterProtocol` carries one more entry point beside `run`:
+
+```python
+async def run_batch_branch(self, pipe_job: PipeJob) -> PipeOutput:
+    return await self.run(pipe_job)
+```
+
+`PipeBatch` calls it instead of `run` for each per-item fan-out dispatch. That is the only place in the pipe tree where the *dispatch* carries semantics its `PipeJob` cannot express: a branch job holds the branch pipe and the item's memory, which is byte-for-byte the shape of any other dispatch. The hook is how a router learns "this one is a fan-out branch".
+
+The default body IS the behavior for in-process routers — a branch is just a run, so `PipeRouter` deliberately does not override it and nothing about direct execution changed. A **distributed** router may override it to give each branch its own isolation (own retry, own history partition) while every other dispatch it receives runs inline. Because the default delegates to `run` and not to `_run_pipe_job`, batch branches still pass through the observer hooks.
+
 ---
 
 ## Pipe Controllers
@@ -153,17 +166,17 @@ Controllers are pipes that orchestrate the execution of other pipes. They resolv
 All controllers follow the same pattern:
 
 1. Call `get_required_pipe(child_pipe_code)` to resolve the child pipe from the library
-2. Route through `get_pipe_router().run(PipeJob(...))` — the hub auto-selects the right router
+2. Route through `get_pipe_router().run(PipeJob(...))` — the hub auto-selects the right router. `PipeBatch` is the one exception: its per-item branch dispatches go through [`run_batch_branch`](#the-batch-branch-hook) instead, so a distributed router can isolate them.
 3. Aggregate results into working memory or output
 
 ### Auto-Switching Router
 
 The hub (`get_pipe_router()`) returns the router for whichever orchestrator the process is booted under:
 
 - **Direct execution**: the in-process `PipeRouter` — child pipes run in the same process.
-- **Distributed execution**: the booted host-runtime orchestrator's router, which has claimed the hub's `PIPE_ROUTER` slot. That router auto-detects whether it is dispatching from the **submitter** (start a top-level workflow) or from **inside a running workflow** (start a child workflow). Pipelex's Temporal backend realizes this as `TemporalPipeRouter`, which picks `execute_workflow` vs `execute_child_workflow` accordingly.
+- **Distributed execution**: the booted host-runtime orchestrator's router, which has claimed the hub's `PIPE_ROUTER` slot. That router auto-detects whether it is dispatching from the **submitter** (start a unit of durable work) or from **inside a running one** (continue within it, or spawn a nested unit).
 
-This means each child pipe in a controller gets its own workflow boundary in distributed mode — enabling independent retries, separate worker assignment, and per-pipe visibility in the host runtime's UI.
+How much of a controller tree a distributed backend spreads across separate durable units is that backend's call, not core's — core's contract is only that every dispatch reaches the router, and that batch branches are *labelled* as such via `run_batch_branch`. See the backend's own documentation for the topology it chooses.
 
 !!! note "Library Dependency"
     Controllers depend on the library being loaded in the current process. `get_required_pipe()` queries the library scoped to the current run via `ContextVar`, which must have been populated by loading a `LibraryCrate`. In distributed execution, each worker-side job loads the crate from the `PipeJob` into a per-run `Library` instance before resolving child pipes. (Temporal-specific detail: the backend disables the Temporal sandbox via `--no-sandbox` because library loading is a side effect incompatible with replay semantics.) See [Runtime Bridge & Transport](./runtime-bridge-and-transport.md) for how the crate and working memory cross the boundary.
@@ -189,18 +202,18 @@ sequenceDiagram
 
     Note over S: TemporalPipeRouter.run() (submitter side)
     S->>S: transport prep (closed)<br/>(WM → working_memory_raw)
-    S->>T: Submit WfPipeRouter(PipeJob)
+    S->>T: Submit run workflow(PipeJob)
     T->>W: Dispatch workflow
 
-    Note over W: WfPipeRouter.run()
+    Note over W: run workflow
     W->>W: Create per-workflow ClassRegistry
     W->>W: Load LibraryCrate (register classes)
     W->>W: Hydrate working_memory_raw → WM
     W->>W: pipe.run_pipe()
 
     alt Concrete pipe
         W->>W: Execute via Activity
-    else Controller pipe
+    else Dispatch the backend chooses to isolate
         W->>T: Child workflow (crate propagates)
         T->>W: Child result
     end

pipelex/pipe_controllers/batch/pipe_batch.py

@@ -5,7 +5,6 @@
 
 from pipelex import log
 from pipelex.cogt.content_generation.dry_mock import stamp_mock_main_coordination
-from pipelex.config import get_config
 from pipelex.core.memory.absence import AbsenceRecord
 from pipelex.core.memory.working_memory import WorkingMemory
 from pipelex.core.pipes.exceptions import PipeRunError
@@ -34,17 +33,6 @@
 LARGE_BATCH_ADVISORY_THRESHOLD = 100
 
 
-def resolve_batch_max_concurrency(max_concurrency_setting: int | str) -> int | None:
-    """Translate the ``pipeline_execution_config.max_concurrency`` setting into a ``gather_bounded`` bound.
-
-    The config exposes the explicit literal ``"unbounded"``; ``gather_bounded`` takes ``None`` for no
-    bound. Any int value is passed through unchanged. Centralizing this guards against passing the
-    raw ``"unbounded"`` string into ``gather_bounded``, which would raise ``TypeError`` on its
-    ``max_concurrency < 1`` check.
-    """
-    return None if isinstance(max_concurrency_setting, str) else max_concurrency_setting
-
-
 class PipeBatch(PipeController):
     type: Literal["PipeBatch"] = "PipeBatch"
 
@@ -142,13 +130,16 @@ async def _live_run_controller_pipe(
         batch_output_stuff_code = StuffFactory.make_stuff_code()
 
         item_count = len(input_content.items)
-        max_concurrency_setting = get_config().pipelex.pipeline_execution_config.max_concurrency
-        max_concurrency = resolve_batch_max_concurrency(max_concurrency_setting)
+        # Read off the payload, never off live config: the bound was frozen into the run params at
+        # submit time precisely so the fan-out shape stays a pure function of the run. See
+        # `PipeRunParams.batch_max_concurrency`.
+        max_concurrency = pipe_run_params.batch_max_concurrency
         if item_count > LARGE_BATCH_ADVISORY_THRESHOLD:
             log.warning(
                 f"PipeBatch '{self.code}' is fanning out over {item_count} items. Bounded fan-out "
-                f"(max_concurrency={max_concurrency_setting}) is a basic backpressure effort, not durable execution — "
-                f"for a workload this size, consider a durable execution backend for rate-limited, resumable runs: {URLs.durable_execution}"
+                f"(max_concurrency={max_concurrency if max_concurrency is not None else 'unbounded'}) is a basic backpressure "
+                f"effort, not durable execution — for a workload this size, consider a durable execution backend for "
+                f"rate-limited, resumable runs: {URLs.durable_execution}"
             )
 
         async def _run_branch(item_input_stuff: "Stuff", *, branch_output_item_code: str) -> PipeOutput:
@@ -167,7 +158,10 @@ async def _run_branch(item_input_stuff: "Stuff", *, branch_output_item_code: str
                     "output_multiplicity": None,
                 },
             )
-            return await get_pipe_router().run(
+            # `run_batch_branch`, not `run`: this is the one dispatch in the pipe tree whose
+            # semantics are "a per-item fan-out branch". In-process routers treat it as a plain
+            # run (the protocol's default body); a distributed router may isolate it.
+            return await get_pipe_router().run_batch_branch(
                 pipe_job=PipeJobFactory.make_pipe_job(
                     pipe=sub_pipe,
                     job_metadata=job_metadata,

pipelex/pipe_machinery/pipe_abstract.py

@@ -1000,8 +1000,17 @@ async def dry_run_pipe(
     ) -> PipeOutput:
         log.verbose(f"Dry run of {self.type}: '{self.code}'")
         assert pipe_run_params.run_mode.is_dry, f"Dry run of {self.type} '{self.code}' called with run_mode = {pipe_run_params.run_mode}"
+        # Stamp the running pipe onto the metadata handed down, exactly as `live_run_pipe` does.
+        # Without it a dry run's `job_metadata.pipe_code` stays whatever the caller passed — usually
+        # unset — so everything downstream that identifies a step by it (leaf-activity labelling in a
+        # distributed backend, log correlation) sees an anonymous step in DRY and a named one in LIVE.
+        # Telemetry stays live-only on purpose: `pipe_run_id` and `otel_context` belong to a real run.
+        # `otel_context=None` matches what `live_run_pipe` itself computes in dry mode, and clearing
+        # it explicitly is the point of that parameter being required — inheriting the parent's would
+        # attach a dry step to a live span.
+        child_metadata = job_metadata.copy_with_update(otel_context=None, pipe_code=self.code)
         return await self._dry_run_pipe(
-            job_metadata=job_metadata,
+            job_metadata=child_metadata,
             working_memory=working_memory,
             pipe_run_params=pipe_run_params,
             output_name=output_name,

pipelex/pipe_run/pipe_router_protocol.py

@@ -78,6 +78,23 @@ async def run(
 
         return pipe_output
 
+    async def run_batch_branch(
+        self,
+        pipe_job: PipeJob,
+    ) -> PipeOutput:
+        """Run ``pipe_job`` as one fan-out branch of a ``PipeBatch``.
+
+        This is the ONE dispatch site in the pipe tree that carries "this dispatch is a
+        per-item fan-out branch" as semantics rather than as a data shape: the branch job
+        carries the branch pipe and the per-item memory, which is indistinguishable from any
+        other dispatch. A distributed router MAY use that signal to isolate the branch (own
+        retry, own history partition); every other dispatch it receives runs inline.
+
+        The default body IS the behavior for in-process routers: a branch is just a run.
+        Implementations only override this when isolation is something they can offer.
+        """
+        return await self.run(pipe_job)
+
     @abstractmethod
     async def _run_pipe_job(
         self,

pipelex/pipe_run/pipe_run_params.py

@@ -148,6 +148,17 @@ class PipeRunParams(BaseModel):
     # frozen for the same reason as `run_mode`.
     is_mock_usage: bool = Field(default=False, frozen=True)
 
+    # Bounded fan-out for PipeBatch, resolved from `pipeline_execution_config.max_concurrency`
+    # once at construction (`PipeRunParamsFactory.make_run_params`). `None` means unbounded —
+    # `gather_bounded`'s own no-bound sentinel, so the field is passed straight through.
+    #
+    # Frozen and payload-borne rather than read live at fan-out time: the bound is PipeBatch's
+    # chunk size, and chunk size determines where a distributed backend's task boundaries fall
+    # between branch dispatches. Read from live worker config, a redeploy that changed the
+    # setting mid-run would make a replay emit a different command grouping than the recorded
+    # history. Carried in the payload, the fan-out shape is a pure function of the run.
+    batch_max_concurrency: int | None = Field(default=None, frozen=True)
+
     final_stuff_code: str | None = None
     output_multiplicity: VariableMultiplicity | None = None
     dynamic_output_concept_ref: str | None = None