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

[lchoquel]

First of two PRs for the flat-workflow-topology reshape. This one is core-only and stands on its own; the second lives in the closed pipelex-temporal repo and depends on this landing (it pins this branch's rev today, and will move to the released core version).

What this is for

The Temporal backend currently starts one child workflow per sub-pipe dispatch — a sequence of five steps is five nested workflows, each paying a full crate load, memory dehydrate/rehydrate, and history round trip. The reshape runs the whole controller tree as inline workflow code and keeps child workflows only where they earn their keep: PipeBatch fan-out branches, which want per-item isolation and history partitioning.

That needs exactly two things from core, both backend-neutral.

1. PipeRouterProtocol.run_batch_branch

A second dispatch entry point, called by PipeBatch for each per-item branch.

It exists because the batch branch is the one dispatch in the pipe tree whose semantics its PipeJob cannot express: a branch job carries the branch pipe and the item's memory, which is byte-for-byte the shape of any other dispatch. Without a distinct call site, a distributed router cannot tell a batch branch from a sequence step.

async def run_batch_branch(self, pipe_job: PipeJob) -> PipeOutput:
    return await self.run(pipe_job)

The default body is the behavior for in-process routers — a branch is just a run. Core's PipeRouter deliberately does not override it, every existing router implementation is untouched, and nothing about direct execution changes. Note it delegates to run, not _run_pipe_job: that keeps batch branches inside the observer stream, and there is a test pinning it.

Alternatives considered and rejected: a dispatch_hint field on PipeJob (a wire-visible DTO change for a hint that never crosses a process boundary), and a ContextVar scope around the branch dispatches (must apply to exactly one dispatch depth, so a sequence inside a branch must not inherit it — implicit and fragile where the hook is explicit and local).

2. Freeze the batch fan-out bound onto the run

PipeBatch read pipeline_execution_config.max_concurrency from live config at fan-out time. That is a real durable-execution hazard, and a pre-existing one — PipeBatch already runs inline inside a workflow today:

the bound is gather_bounded's chunk size, and chunk size determines where workflow-task boundaries fall between branch-dispatch commands. A worker redeploy that changes the setting while a batch is in flight makes replay emit a different command grouping than history recorded — a genuine [TMPRL1100].

The setting is now resolved once, at run-params construction, and carried as frozen PipeRunParams.batch_max_concurrency (None = unbounded, gather_bounded's own sentinel). PipeBatch no longer imports get_config at all; resolve_batch_max_concurrency moves next to the factory that owns the read. Nested batches inherit it through the existing model_copy.

Breaking only for code that mutated max_concurrency mid-run and expected the change to take effect.

Tests

• Hook default reaches _run_pipe_job with the job it was handed, and goes through run so observers still fire.
• resolve_batch_max_concurrency translation table (moved with the helper).
• The factory freezes the live value; a later config change does not reach existing params.
• Integration: batch branches dispatch through the hook (one call per item, and the batch itself never enters through it), and the fan-out bound comes off the payload — built under max_concurrency = 2, still 2 after the config flips to 5.

Docs

The router-SPI page gains a "batch-branch hook" section; the Orchestrator SPI table row names it. The PipeBatch concurrency page documents the freeze.

Two spots in pipe-routing-and-execution.md asserted a Temporal-specific topology — the controller arm of the distributed sequence diagram, and a paragraph promising "each child pipe in a controller gets its own workflow boundary", plus WfPipeRouter by name. How much of a controller tree a backend spreads across durable units is that backend's call, not core's, and the reshape makes those claims false. Rewritten to say the backend decides. That is why the docs diff is larger than "document the hook" implies.

Housekeeping

Moving test_pipe_batch_concurrency.py left four dead entries in the committed .test_durations, which its own guard test catches. Dropped surgically (four lines, original formatting) rather than regenerating the whole file.

🤖 Generated with Claude Code

https://claude.ai/code/session_01JenThPdUXSpL4SzY5zyqnM

---

Summary by cubic

Adds a run_batch_branch hook to PipeRouterProtocol, freezes PipeBatch fan‑out concurrency onto the run, and fixes dry‑run metadata so steps are correctly identified. This enables flat-topology backends and removes a durable-execution replay hazard without changing in-process behavior.

• New Features

  • PipeRouterProtocol.run_batch_branch: called by PipeBatch for each per‑item branch; default delegates to run (observers still fire) so existing routers stay unchanged, while distributed routers can override to isolate branches.

• Bug Fixes

  • Fan‑out bound is now frozen at submit time as PipeRunParams.batch_max_concurrency (None = unbounded). PipeBatch stops reading live pipeline_execution_config.max_concurrency, preventing replay divergence on durable backends. Breaking: mid‑run changes to max_concurrency no longer affect in‑flight batches.
  • Dry runs now stamp the running pipe onto JobMetadata (pipe_code set, otel_context=None) so step identification matches live runs without attaching to a live span.

^{Written for commit 7d3ad12. Summary will update on new commits.}

[greptile-apps]

Confidence Score: 4/5

The PR is not yet safe to merge because previously compatible duck-typed routers can still fail on the first batch branch dispatch.

PipeBatch unconditionally invokes run_batch_branch, while router registration and retrieval neither require protocol inheritance nor adapt structural implementations that only expose the previous router API, leaving the previously reported AttributeError reachable.

Files Needing Attention: pipelex/pipe_controllers/batch/pipe_batch.py and pipelex/interpreter_hub.py

_{Reviews (2): Last reviewed commit: "fix(pipe-run): stamp the running pipe on..." | Re-trigger Greptile}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 3d8a697206

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[cubic-dev-ai]

All reported issues were addressed across 14 files

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[lchoquel]

Thanks — worked through all of these. They dedupe to four distinct items; one is fixed, one is deferred with a written note, two are dropped. Reasoning below so the drops are auditable.

1. run_batch_branch and duck-typed routers — dropped (refuted)

Raised by all three reviewers, with a suggested getattr(router, "run_batch_branch", router.run) fallback.

The Python fact is right in the abstract, but the scenario cannot occur and the type checker already blocks it. PipeRouterProtocol is used ABC-style — _run_pipe_job is @abstractmethod and run is a concrete body carrying the observer hooks and the PipeRunError → PipeRouterError wrapping — so every router in the workspace inherits it explicitly (core PipeRouter, TemporalPipeRouter, MistralWorkflowsPipeRouter, the test routers). There are no structural-only implementations.

More decisively: structural protocol compatibility requires all members, concrete ones included. A duck-typed router implementing every old member is rejected by mypy at its registration call site:

error: Argument "pipe_router" to "set_pipe_router" has incompatible type "DuckTypedRouter";
       expected "PipeRouterProtocol"
note:  "DuckTypedRouter" is missing following "PipeRouterProtocol" protocol member: run_batch_branch

So this is a compile-time-enforced breaking change on a repo that gates on strict pyright + mypy — which is exactly what the no-backward-compatibility principle sanctions, and the CHANGELOG documents it.

The suggested fallback would be actively harmful: it erases that type error, so a distributed router whose override was renamed or typo'd would silently degrade to inline run — the precise failure the hook exists to prevent.

2. batch_max_concurrency=None means unbounded — deferred, with a note

Confirmed as a fact and worth fixing, but unreachable from any production path: nothing outside tests constructs PipeRunParams(...) directly, and every derivation goes through model_copy, which preserves the field.

The reviewers are pointing at something real, and the best argument for it is eighteen lines up in the same file — run_mode has no default precisely so a payload missing it fails loud rather than defaulting toward the spending direction. Same shape here. The clean fix is to drop the default and make the field required, which forces edits to ~19 direct constructions across the test suite. That deserves its own change rather than riding inside a topology reshape, so it is recorded rather than rushed.

3. Per-run PipelineExecutionConfig.max_concurrency ignored — dropped

Correct reading of the code, but nothing in the workspace constructs a PipelineExecutionConfig, and with_execution_overrides — the only way a per-run copy is made — structurally cannot alter max_concurrency (its parameters are the graph/usage/data/mock flags). Per-run and global always agree. Pre-PR, PipeBatch read the same process-global live, so this is not a regression either. Threading an argument through make_run_params to serve a divergence no code can create is defensive plumbing; folded into the deferred note instead, to be revisited if max_concurrency ever becomes per-run overridable.

4. Two Test classes in one module — fixed (a9d031b)

Merged into a single TestBatchMaxConcurrency. Repo rule, brand-new file, four-line edit — cheaper to comply than to justify.

[lchoquel]

@greptileai @cubic-dev-ai please review the newly pushed commit 7d3ad12c2 — it makes dry_run_pipe stamp pipe_code onto the job metadata it hands down, matching live_run_pipe.

[cubic-dev-ai]

@greptileai @cubic-dev-ai please review the newly pushed commit 7d3ad12c2 — it makes dry_run_pipe stamp pipe_code onto the job metadata it hands down, matching live_run_pipe.

@lchoquel I have started the AI code review. It will take a few minutes to complete.

[cubic-dev-ai]

No issues found across 15 files

_{Re-trigger cubic}