diff --git a/.test_durations b/.test_durations
index f028b03266..1a18c4638e 100644
--- a/.test_durations
+++ b/.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,
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7c28fd31bc..a96bbe643e 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/docs/building-methods/pipes/pipe-controllers/PipeBatch.md b/docs/building-methods/pipes/pipe-controllers/PipeBatch.md
index c773ecf783..0cf3b83f98 100644
--- a/docs/building-methods/pipes/pipe-controllers/PipeBatch.md
+++ b/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.
diff --git a/docs/under-the-hood/orchestrator-plugins.md b/docs/under-the-hood/orchestrator-plugins.md
index 503e8e82ee..9150077567 100644
--- a/docs/under-the-hood/orchestrator-plugins.md
+++ b/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) |
diff --git a/docs/under-the-hood/pipe-routing-and-execution.md b/docs/under-the-hood/pipe-routing-and-execution.md
index 4b2343a7f2..c0afb13adf 100644
--- a/docs/under-the-hood/pipe-routing-and-execution.md
+++ b/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,7 +166,7 @@ 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
@@ -161,9 +174,9 @@ All controllers follow the same pattern:
 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,10 +202,10 @@ 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
@@ -200,7 +213,7 @@ sequenceDiagram
 
     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
diff --git a/pipelex/pipe_controllers/batch/pipe_batch.py b/pipelex/pipe_controllers/batch/pipe_batch.py
index 2e9d9773fe..df9ace0e30 100644
--- a/pipelex/pipe_controllers/batch/pipe_batch.py
+++ b/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,
diff --git a/pipelex/pipe_machinery/pipe_abstract.py b/pipelex/pipe_machinery/pipe_abstract.py
index f63ade5b12..4daa9d7964 100644
--- a/pipelex/pipe_machinery/pipe_abstract.py
+++ b/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,
diff --git a/pipelex/pipe_run/pipe_router_protocol.py b/pipelex/pipe_run/pipe_router_protocol.py
index 91512feee8..33bbe890d8 100644
--- a/pipelex/pipe_run/pipe_router_protocol.py
+++ b/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,
diff --git a/pipelex/pipe_run/pipe_run_params.py b/pipelex/pipe_run/pipe_run_params.py
index f29c2fc191..a9c227e689 100644
--- a/pipelex/pipe_run/pipe_run_params.py
+++ b/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
diff --git a/pipelex/pipe_run/pipe_run_params_factory.py b/pipelex/pipe_run/pipe_run_params_factory.py
index cee98a65f3..596b4cd5b3 100644
--- a/pipelex/pipe_run/pipe_run_params_factory.py
+++ b/pipelex/pipe_run/pipe_run_params_factory.py
@@ -9,6 +9,20 @@
 from pipelex.system.pipe_run_mode import PipeRunMode
 
 
+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.
+
+    Lives next to the factory because the factory is where the setting is read: the resolved bound is
+    frozen into ``PipeRunParams.batch_max_concurrency`` at construction, never re-read at fan-out time.
+    """
+    return None if isinstance(max_concurrency_setting, str) else max_concurrency_setting
+
+
 class PipeRunParamsFactory:
     @classmethod
     def make_run_params(
@@ -22,7 +36,7 @@ def make_run_params(
         batch_params: BatchParams | None = None,
         params: dict[str, Any] | None = None,
     ) -> PipeRunParams:
-        """Single writer of ``run_mode`` and ``is_mock_usage`` — direct fields on ``PipeRunParams``.
+        """Single writer of ``run_mode``, ``is_mock_usage`` and ``batch_max_concurrency`` — direct fields on ``PipeRunParams``.
 
         The keyless-boot forced-DRY flag (eng review D4) is resolved HERE — at the single writer —
         so every execution entry point is covered (``prepare_pipe_job``, the runtime bridge,
@@ -31,6 +45,9 @@ def make_run_params(
         The REQUESTED mode is validated before the forced-DRY coercion, so a contract violation
         (``is_mock_usage`` on a LIVE request) fails loud on every boot — the keyless coercion must
         not silently turn an illegal request into a legal one.
+
+        ``batch_max_concurrency`` is resolved here for the same single-writer reason and because the
+        read must happen ONCE, at submit time: see the field's docstring on ``PipeRunParams``.
         """
         check_mock_usage_requires_dry(run_mode=pipe_run_mode, is_mock_usage=is_mock_usage)
         if is_dry_run_forced() and pipe_run_mode.is_live:
@@ -39,11 +56,13 @@ def make_run_params(
                 "outputs will be synthetic mocks, not real inference."
             )
             pipe_run_mode = PipeRunMode.DRY
-        pipe_stack_limit = pipe_stack_limit or get_config().pipelex.pipe_run_config.pipe_stack_limit
+        config = get_config().pipelex
+        pipe_stack_limit = pipe_stack_limit or config.pipe_run_config.pipe_stack_limit
         return PipeRunParams(
             run_mode=pipe_run_mode,
             is_mock_usage=is_mock_usage,
             pipe_stack_limit=pipe_stack_limit,
+            batch_max_concurrency=resolve_batch_max_concurrency(config.pipeline_execution_config.max_concurrency),
             output_multiplicity=output_multiplicity,
             dynamic_output_concept_ref=dynamic_output_concept_ref,
             batch_params=batch_params,
diff --git a/subject_grants.toml b/subject_grants.toml
index 0ee2ea4b63..b6b3720b7b 100644
--- a/subject_grants.toml
+++ b/subject_grants.toml
@@ -3980,10 +3980,6 @@ rationale = "Verb-object: scans the pipe for taint triggers; slot_taints is cont
 param = "item_input_stuff"
 rationale = "Verb-object: runs the branch for the item stuff; invoked positionally via functools.partial."
 
-["pipelex/pipe_controllers/batch/pipe_batch.py::resolve_batch_max_concurrency"]
-param = "max_concurrency_setting"
-rationale = "Verb-object: resolves the config setting value; single operand, callers pass a labelled config read."
-
 ["pipelex/pipe_controllers/condition/special_outcome.py::SpecialOutcome.is_continue"]
 param = "outcome"
 rationale = "Predicate on the outcome string; single operand."
@@ -4336,6 +4332,10 @@ rationale = "Protocol hook: runs the pipe job; single operand."
 param = "pipe_job"
 rationale = "Protocol method: runs the pipe job; single operand across implementations."
 
+["pipelex/pipe_run/pipe_router_protocol.py::PipeRouterProtocol.run_batch_branch"]
+param = "pipe_job"
+rationale = "Protocol hook: the pipe job being dispatched as a fan-out branch; single operand."
+
 ["pipelex/pipe_run/pipe_run_params.py::PipeRunParams.copy_by_injecting_multiplicity"]
 param = "pipe_run_params"
 rationale = "Verb-object: copies the given run params; multiplicity stays keyword."
@@ -4352,6 +4352,10 @@ rationale = "Verb-object: pushes the pipe named by the code; single operand."
 param = "pipe_code"
 rationale = "Verb-object: pushes the pipe named by the code; single operand."
 
+["pipelex/pipe_run/pipe_run_params_factory.py::resolve_batch_max_concurrency"]
+param = "max_concurrency_setting"
+rationale = "Verb-object: resolves the config setting value; single operand, callers pass a labelled config read."
+
 ["pipelex/pipe_run/pipe_run_protocol.py::PipeRunProtocol.run"]
 param = "pipe_job"
 rationale = "Protocol method: runs the pipe job; single operand across implementations."
diff --git a/tests/integration/pipelex/pipes/controller/pipe_batch/test_pipe_batch_branch_dispatch.py b/tests/integration/pipelex/pipes/controller/pipe_batch/test_pipe_batch_branch_dispatch.py
new file mode 100644
index 0000000000..9ba0ce58a2
--- /dev/null
+++ b/tests/integration/pipelex/pipes/controller/pipe_batch/test_pipe_batch_branch_dispatch.py
@@ -0,0 +1,172 @@
+"""PipeBatch's branch dispatch: it goes through the `run_batch_branch` hook, and its fan-out bound
+comes off the payload rather than live config.
+
+Both are the core half of the flat-topology seam. The hook is the ONLY signal a distributed router
+gets that a dispatch is a per-item fan-out branch (the branch job is otherwise indistinguishable
+from a sequence step); the frozen bound is what keeps the fan-out grouping a pure function of the
+run, so a config redeploy cannot reshape an in-flight batch.
+"""
+
+from collections.abc import Awaitable, Callable, Sequence
+
+import pytest
+from typing_extensions import override
+
+from pipelex.config import get_config
+from pipelex.core.memory.working_memory import WorkingMemory
+from pipelex.core.memory.working_memory_factory import WorkingMemoryFactory
+from pipelex.core.pipes.pipe_output import PipeOutput
+from pipelex.core.stuffs.list_content import ListContent
+from pipelex.core.stuffs.stuff_factory import StuffFactory
+from pipelex.core.stuffs.text_content import TextContent
+from pipelex.interpreter_hub import get_pipe_library, scoped_pipe_router
+from pipelex.pipe_controllers.batch import pipe_batch as pipe_batch_module
+from pipelex.pipe_controllers.batch.pipe_batch import PipeBatch
+from pipelex.pipe_controllers.batch.pipe_batch_blueprint import PipeBatchBlueprint
+from pipelex.pipe_machinery.pipe_factory import PipeFactory
+from pipelex.pipe_operators.func.pipe_func import PipeFunc
+from pipelex.pipe_operators.func.pipe_func_blueprint import PipeFuncBlueprint
+from pipelex.pipe_run.pipe_job import PipeJob
+from pipelex.pipe_run.pipe_router import PipeRouter
+from pipelex.pipe_run.pipe_run_params_factory import PipeRunParamsFactory
+from pipelex.system.job_metadata import JobMetadata
+from pipelex.system.pipe_run_mode import PipeRunMode
+from pipelex.system.registries.func_registry import func_registry
+
+_DOMAIN_CODE = "test_batch_branch_dispatch"
+
+
+def batch_branch_dispatch_shout_item(working_memory: WorkingMemory) -> TextContent:
+    return TextContent(text=working_memory.get_stuff_as_str(name="item").upper())
+
+
+class _HookCountingRouter(PipeRouter):
+    """Real in-process router that records which dispatch door each job came through."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.branch_dispatched_pipe_codes: list[str] = []
+        self.plain_run_pipe_codes: list[str] = []
+
+    @override
+    async def run(self, pipe_job: PipeJob) -> PipeOutput:
+        self.plain_run_pipe_codes.append(pipe_job.pipe.code)
+        return await super().run(pipe_job)
+
+    @override
+    async def run_batch_branch(self, pipe_job: PipeJob) -> PipeOutput:
+        self.branch_dispatched_pipe_codes.append(pipe_job.pipe.code)
+        return await super().run_batch_branch(pipe_job)
+
+
+def _build_batch() -> PipeBatch:
+    """A batch over Text items whose branch just uppercases each item."""
+    shout_pipe = PipeFactory[PipeFunc].make_from_blueprint(
+        domain_code=_DOMAIN_CODE,
+        pipe_code="branch_dispatch_shout",
+        blueprint=PipeFuncBlueprint(
+            description="Uppercase the batch item",
+            inputs={"item": "Text"},
+            output="Text",
+            function_name="batch_branch_dispatch_shout_item",
+        ),
+    )
+    pipe_library = get_pipe_library()
+    pipe_library.add_new_pipe(pipe=shout_pipe)
+
+    batch = PipeFactory[PipeBatch].make_from_blueprint(
+        domain_code=_DOMAIN_CODE,
+        pipe_code="branch_dispatch_batch",
+        blueprint=PipeBatchBlueprint(
+            description="Batch the shout over the items",
+            branch_pipe_code="branch_dispatch_shout",
+            output="Text",
+            input_list_name="items",
+            input_item_name="item",
+            inputs={"items": "Text"},
+        ),
+    )
+    pipe_library.add_new_pipe(pipe=batch)
+    return batch
+
+
+def _make_items_memory(batch: PipeBatch, texts: list[str]):
+    items_stuff = StuffFactory.make_stuff(
+        concept=batch.inputs.get_required_stuff_spec("items").concept,
+        content=ListContent[TextContent](items=[TextContent(text=text) for text in texts]),
+        name="items",
+    )
+    return WorkingMemoryFactory.make_from_single_stuff(items_stuff)
+
+
+@pytest.mark.asyncio(loop_scope="class")
+class TestPipeBatchBranchDispatch:
+    @classmethod
+    def setup_class(cls):
+        func_registry.register_function(batch_branch_dispatch_shout_item)
+
+    @classmethod
+    def teardown_class(cls):
+        if func_registry.has_function(batch_branch_dispatch_shout_item.__name__):
+            func_registry.unregister_function_by_name(batch_branch_dispatch_shout_item.__name__)
+
+    async def test_branches_dispatch_through_the_hook(self, job_metadata: JobMetadata, load_empty_library: Callable[[], str]):
+        """One `run_batch_branch` call per item, and the batch itself never enters through the hook."""
+        load_empty_library()
+        batch = _build_batch()
+        router = _HookCountingRouter()
+
+        with scoped_pipe_router(router):
+            await batch.run_pipe(
+                job_metadata=job_metadata,
+                working_memory=_make_items_memory(batch, ["alpha", "beta", "gamma"]),
+                pipe_run_params=PipeRunParamsFactory.make_run_params(pipe_run_mode=PipeRunMode.LIVE),
+            )
+
+        assert router.branch_dispatched_pipe_codes == ["branch_dispatch_shout"] * 3
+        assert "branch_dispatch_batch" not in router.branch_dispatched_pipe_codes
+
+    async def test_fan_out_bound_comes_off_the_payload_not_live_config(
+        self,
+        job_metadata: JobMetadata,
+        load_empty_library: Callable[[], str],
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        """Params built under `max_concurrency = 2` keep that bound after the config flips to 5.
+
+        This is the replay hazard the freeze exists for: the bound is `gather_bounded`'s chunk size,
+        so a live read here would let a mid-run worker redeploy regroup the branch dispatches.
+        """
+        load_empty_library()
+        batch = _build_batch()
+
+        captured_bounds: list[int | None] = []
+        real_gather_bounded = pipe_batch_module.gather_bounded
+
+        async def _spying_gather_bounded(
+            task_factories: "Sequence[Callable[[], Awaitable[PipeOutput]]]",
+            *,
+            max_concurrency: int | None,
+        ) -> list[PipeOutput]:
+            captured_bounds.append(max_concurrency)
+            return await real_gather_bounded(task_factories, max_concurrency=max_concurrency)
+
+        monkeypatch.setattr(pipe_batch_module, "gather_bounded", _spying_gather_bounded)
+
+        execution_config = get_config().pipelex.pipeline_execution_config
+        original_setting = execution_config.max_concurrency
+        try:
+            execution_config.max_concurrency = 2
+            run_params = PipeRunParamsFactory.make_run_params(pipe_run_mode=PipeRunMode.LIVE)
+            execution_config.max_concurrency = 5
+
+            with scoped_pipe_router(_HookCountingRouter()):
+                await batch.run_pipe(
+                    job_metadata=job_metadata,
+                    working_memory=_make_items_memory(batch, ["alpha", "beta", "gamma"]),
+                    pipe_run_params=run_params,
+                )
+        finally:
+            execution_config.max_concurrency = original_setting
+
+        assert captured_bounds == [2]
diff --git a/tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_concurrency.py b/tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_concurrency.py
deleted file mode 100644
index 069169fc7f..0000000000
--- a/tests/unit/pipelex/pipe_controllers/batch/test_pipe_batch_concurrency.py
+++ /dev/null
@@ -1,26 +0,0 @@
-import pytest
-
-from pipelex.pipe_controllers.batch.pipe_batch import resolve_batch_max_concurrency
-
-
-class TestResolveBatchMaxConcurrency:
-    @pytest.mark.parametrize(
-        ("max_concurrency_setting", "expected_bound"),
-        [
-            ("unbounded", None),
-            (1, 1),
-            (8, 8),
-            (100, 100),
-        ],
-    )
-    def test_setting_translates_to_gather_bounded_argument(
-        self,
-        max_concurrency_setting: int | str,
-        expected_bound: int | None,
-    ) -> None:
-        """The literal "unbounded" config maps to None (gather_bounded's no-bound sentinel); an int passes through.
-
-        Guards the PipeBatch fan-out wiring against regressing to passing the raw "unbounded"
-        string straight into gather_bounded, which would raise TypeError on its `max_concurrency < 1` check.
-        """
-        assert resolve_batch_max_concurrency(max_concurrency_setting) == expected_bound
diff --git a/tests/unit/pipelex/pipe_run/test_batch_max_concurrency.py b/tests/unit/pipelex/pipe_run/test_batch_max_concurrency.py
new file mode 100644
index 0000000000..f40ccb72fa
--- /dev/null
+++ b/tests/unit/pipelex/pipe_run/test_batch_max_concurrency.py
@@ -0,0 +1,58 @@
+"""`batch_max_concurrency`: resolved from config once, at run-params construction, then frozen.
+
+The bound is PipeBatch's fan-out chunk size. Read live at fan-out time it would let a config
+redeploy reshape an in-flight run's dispatch grouping; carried in the payload it cannot. These
+tests pin the resolution table and the write-once discipline. (The field is `frozen=True`, so a
+post-construction write is a *type* error — no runtime test needed for what the checkers block.)
+"""
+
+import pytest
+
+from pipelex.config import get_config
+from pipelex.pipe_run.pipe_run_params_factory import PipeRunParamsFactory, resolve_batch_max_concurrency
+from pipelex.system.pipe_run_mode import PipeRunMode
+
+
+class TestBatchMaxConcurrency:
+    @pytest.mark.parametrize(
+        ("max_concurrency_setting", "expected_bound"),
+        [
+            ("unbounded", None),
+            (1, 1),
+            (8, 8),
+            (100, 100),
+        ],
+    )
+    def test_setting_translates_to_gather_bounded_argument(
+        self,
+        max_concurrency_setting: int | str,
+        expected_bound: int | None,
+    ) -> None:
+        """The literal "unbounded" config maps to None (gather_bounded's no-bound sentinel); an int passes through.
+
+        Guards the PipeBatch fan-out wiring against regressing to passing the raw "unbounded"
+        string straight into gather_bounded, which would raise TypeError on its `max_concurrency < 1` check.
+        """
+        assert resolve_batch_max_concurrency(max_concurrency_setting) == expected_bound
+
+    def test_factory_freezes_the_live_config_value(self) -> None:
+        execution_config = get_config().pipelex.pipeline_execution_config
+        expected = resolve_batch_max_concurrency(execution_config.max_concurrency)
+
+        run_params = PipeRunParamsFactory.make_run_params(pipe_run_mode=PipeRunMode.DRY)
+
+        assert run_params.batch_max_concurrency == expected
+
+    def test_later_config_change_does_not_reach_existing_run_params(self) -> None:
+        """The whole point: params built before a config edit keep the bound they were born with."""
+        execution_config = get_config().pipelex.pipeline_execution_config
+        original_setting = execution_config.max_concurrency
+        try:
+            execution_config.max_concurrency = 3
+            run_params = PipeRunParamsFactory.make_run_params(pipe_run_mode=PipeRunMode.DRY)
+            assert run_params.batch_max_concurrency == 3
+
+            execution_config.max_concurrency = 7
+            assert run_params.batch_max_concurrency == 3
+        finally:
+            execution_config.max_concurrency = original_setting
diff --git a/tests/unit/pipelex/pipe_run/test_run_batch_branch_hook.py b/tests/unit/pipelex/pipe_run/test_run_batch_branch_hook.py
new file mode 100644
index 0000000000..2424a8e558
--- /dev/null
+++ b/tests/unit/pipelex/pipe_run/test_run_batch_branch_hook.py
@@ -0,0 +1,94 @@
+"""The `run_batch_branch` router hook: its default body IS the behavior for in-process routers.
+
+`PipeBatch` marks its per-item fan-out dispatches by calling this hook instead of `run`. Every
+router that has no isolation to offer inherits the concrete default, which delegates straight to
+`run` — so adding the hook changed no in-process behavior. These tests pin that delegation, because
+a distributed router's override is only correct if the un-overridden case stays a plain run.
+"""
+
+from typing import TYPE_CHECKING, cast
+
+import pytest
+from typing_extensions import override
+
+from pipelex.core.memory.working_memory_factory import WorkingMemoryFactory
+from pipelex.core.pipes.pipe_output import PipeOutput
+from pipelex.observer.observer_protocol import ObserverNoOp
+from pipelex.pipe_run.pipe_job import PipeJob
+from pipelex.pipe_run.pipe_router_protocol import PipeRouterProtocol
+from pipelex.pipe_run.pipe_run_params_factory import PipeRunParamsFactory
+from pipelex.system.job_metadata import JobMetadata
+from pipelex.system.pipe_run_mode import PipeRunMode
+
+if TYPE_CHECKING:
+    from pipelex.pipe_machinery.pipe_abstract import PipeAbstract
+
+
+class _StubPipe:
+    """Minimal pipe stand-in: the router only reads `.code` off it."""
+
+    code = "stub_pipe"
+
+
+class _RecordingRouter(PipeRouterProtocol):
+    """Router that records the jobs reaching `_run_pipe_job` and never touches a real pipe."""
+
+    def __init__(self) -> None:
+        self.observer = ObserverNoOp()
+        self.dispatched_jobs: list[PipeJob] = []
+
+    @override
+    async def _run_pipe_job(self, pipe_job: PipeJob) -> PipeOutput:
+        self.dispatched_jobs.append(pipe_job)
+        return PipeOutput(
+            working_memory=WorkingMemoryFactory.make_empty(),
+            pipeline_run_id=pipe_job.job_metadata.pipeline_run_id,
+        )
+
+
+def _make_pipe_job() -> PipeJob:
+    return PipeJob.model_construct(
+        pipe=cast("PipeAbstract", _StubPipe()),
+        working_memory=None,
+        working_memory_raw=None,
+        pipe_run_params=PipeRunParamsFactory.make_run_params(pipe_run_mode=PipeRunMode.LIVE),
+        job_metadata=JobMetadata(user_id="test-user", pipeline_run_id="test-run"),
+        output_name=None,
+        library_crate=None,
+    )
+
+
+@pytest.mark.asyncio(loop_scope="class")
+class TestRunBatchBranchHook:
+    async def test_default_delegates_to_run(self) -> None:
+        """The un-overridden hook reaches `_run_pipe_job` with the very job it was handed."""
+        router = _RecordingRouter()
+        pipe_job = _make_pipe_job()
+
+        await router.run_batch_branch(pipe_job)
+
+        assert router.dispatched_jobs == [pipe_job]
+
+    async def test_default_runs_the_observer_hooks_like_run(self) -> None:
+        """Delegation goes through `run`, not around it — so observers still see the branch.
+
+        Pins the "delegate to `run`" contract rather than "delegate to `_run_pipe_job`": a hook that
+        short-circuited to the private dispatch would silently drop every batch branch out of the
+        observer stream.
+        """
+        observed: list[str] = []
+
+        class _ObservingRouter(_RecordingRouter):
+            @override
+            async def _before_run(self, pipe_job: PipeJob) -> None:
+                observed.append("before")
+
+            @override
+            async def _after_successful_run(self, pipe_job: PipeJob, *, pipe_output: PipeOutput) -> None:
+                observed.append("after")
+
+        router = _ObservingRouter()
+
+        await router.run_batch_branch(_make_pipe_job())
+
+        assert observed == ["before", "after"]