Runtime Bridge & Transport

This page is for contributors working on Pipelex internals. For the capability overview, see the user-facing Distributed Execution page instead.

Pipelex runs a pipe either direct (everything in one Python process) or distributed on a host runtime (a separate worker process, potentially on another machine). Both paths reach pipe execution through the same framework-agnostic runtime bridge. This page documents that bridge backend-neutrally: the boundary data types, how a library snapshot and working memory cross a process boundary, and how concurrent runs stay isolated. The concrete realizations — Temporal-backed durable execution and the Mistral Workflows integration — are commercial Pipelex platform capabilities; their backend-specific topology lives in their own plugin repos.

Direct vs distributed, one bridge

Both modes share the same pipe definitions, library loading, and controller logic. The only difference is where a pipe runs and how its PipeJob gets there. Distributing execution across separate processes introduces problems that don't exist in-process — those problems, and their solutions, are this page's subject. For the routing layer that sits above the bridge (how a controller dispatches its child pipes in each mode), see Pipe Routing & Execution.

The boundary data types

Three open-core Pydantic models are everything that crosses a process boundary.

PipeJob — the unit of execution

PipeJob (pipelex/pipe_run/pipe_job.py) carries everything needed to run a single pipe:

PipeOutput — the result that comes back

PipeOutput (pipelex/core/pipes/pipe_output.py) carries the run result. Like PipeJob, it has both a typed working_memory and a working_memory_raw dict — the raw field is how a result travels back across a boundary without requiring class resolution at every hop (see Deferred hydration).

LibraryCrate — a serializable library snapshot

A LibraryCrate (pipelex/libraries/library_crate.py) is a flat, serializable snapshot of the library — the pipes and concepts a worker needs to run the job:

class LibraryCrate(BaseModel):
    concepts: dict[str, ConceptBlueprint | str]  # concept ref -> blueprint or description
    pipes: dict[str, PipeBlueprintUnion]  # pipe ref -> blueprint
    domains: dict[str, DomainBlueprint]  # domain code -> domain metadata
    source_map: dict[str, str]  # ref -> source file (error traceability)
    fingerprint: str  # SHA256 of serialized content

Domain is encoded in the dictionary keys (e.g., scoring.WeightedScore, scoring.compute_score), not in a structural container. source_map lets error messages trace back to origin files. The fingerprint enables idempotent loading — a worker that already loaded a crate with the same fingerprint skips the reload. The crate is a plain Pydantic model, so a host runtime serializes it as structured JSON in the job input — fully inspectable, not opaque bytes.

The three cross-process challenges

Sending a pipe run to a separate worker process introduces three challenges that single-process execution never faces:

1. Pipe resolution — Controllers call get_required_pipe() to resolve child pipes. The worker's library must contain those pipes. → solved by LibraryCrate propagation.
2. Dynamic class deserialization — .mthds bundles generate Python classes at runtime (e.g., RawText inheriting from TextContent). A receiving process must have those classes registered before it can deserialize WorkingMemory. → solved by deferred hydration.
3. Concurrent isolation — One worker may run many jobs at once. Two jobs that each define a concept named Result with different structures must not share a class registry or library. → solved by per-call scoping.

LibraryCrate propagation

The submitter builds a LibraryCrate from its loaded library (pipelex/libraries/library_crate_factory.py) and attaches it to the PipeJob. On the worker, the job's entry point loads the crate into a scoped library, registering the bundle's dynamic concept classes, so get_required_pipe() resolves every pipe at every level of nesting. Because a distributed run can fan out into nested jobs (a controller's child pipes), every nested job carries the same crate; loading is idempotent via the fingerprint, so re-receiving a crate already loaded on a worker is a no-op.

A load ends by validating the whole library (Library.validate_library), which checks every pipe against the concepts and pipes it depends on. That check is pure — it raises or returns, and leaves nothing behind that a run later reads — and the crate a worker receives was built from a library the submitter already loaded and validated. So a worker running the same pipelex version as its submitter can call load_from_crate(library_id=…, crate=…, is_crate_prevalidated=True), which skips that final validation and nothing else: fingerprint idempotency, domain and concept loading with the registration of the dynamic classes, the resolution of their references to one another, the concept-cycle check, pipe construction with each pipe's static validation, and source tracking all still run. The precondition belongs to the host: pass it only where submitter and worker ship on the same pipelex, as a deployment that builds them from one lockfile does. A crate from anywhere else keeps the default, validating load.

A crate carries no method package the library depends on, so a structure field typed by a dependency's concept cannot be resolved on the worker. A crate load therefore never refuses a structure class it could not complete, whichever the flag: the refusal of a field naming a concept that cannot be found belongs to the load from bundles, which resolves the dependencies first, and the submitter's library went through it. Such a class fails only if a run on the worker uses it, as would the dependency's own pipes, which the crate does not carry either.

Deferred hydration

The chicken-and-egg problem

A host runtime's data converter deserializes a payload (a job input, a return value) before any bundle code has run — but reconstructing a typed WorkingMemory needs the bundle's dynamic concept classes registered, and those only get registered when the LibraryCrate is loaded inside the job. Two deserialization gaps follow:

1. Input — the PipeJob input contains WorkingMemory whose dynamic classes don't exist yet on the worker.
2. Output — a nested job's PipeOutput return value contains WorkingMemory with dynamic classes, and the parent's data converter typically runs outside the job's scope, so it can't reach the per-job class registry.

The solution: carry working memory raw

Both PipeJob and PipeOutput carry a working_memory_raw field — a plain JSON dict — alongside the typed working_memory. The raw dict needs no class resolution to cross a boundary, which decouples transport from class registration.

A working memory therefore has two shapes. In memory, a Stuff holds a resolved Concept — the runtime's own subclass of the protocol's ConceptAbstract, carrying the description, the structure_class_name and the refines the standard keeps in the library a method loads. On the wire, a stuff names its concept: every dump — model_dump, smart_dump, dump_for_transport, the working_memory.json a run saves, the runner's PipeOutput response — emits {"stuff_code": …, "stuff_name": …, "concept": "<domain>.<Code>", "content": …}, the form the MTHDS CLI I/O contract requires, because StuffAbstract serializes concept as the ref string. The definition never leaves the runtime, so a dump is one-way: a reader resolves each ref through the concept library the method loaded, the same way the input side resolves an input envelope's concept.

This runtime emits and resolves <domain>.<Code> only. The standard also defines <package_address>::<domain>.<Code>, the spelling for a concept a dependency package contributes, and nothing here writes it or accepts it — that form is not implemented yet. The gap is closed on the reading side rather than on the writing side, by the one rule both readers share, ConceptProviderAbstract.resolve_wire_concept_ref: a bare ref is matched against every key the concept library holds for that spelling — the host/native entry keyed by the ref itself, plus each <alias>-><domain>.<Code> entry a dependency landed under (ConceptLibrary.add_dependency_concept, where the alias is the whole package address on the live route). One candidate resolves, none raises naming the ref, and more than one raises naming every key that matched, because a bare ref genuinely cannot say whether the host's scoring.WeightedScore or a dependency's was meant, and binding one package's definition to another package's data is a silent data fault rather than a failure. Enumeration is why list_concept_keys_for_ref sits on the provider abstraction at all: resolution alone cannot distinguish "one match" from "two".

• Input dehydration — before dispatch, the working memory is moved into working_memory_raw (a plain dict produced by WorkingMemory.dump_for_transport(), which stays in open core at pipelex/core/memory/working_memory.py). After the worker loads the crate, the job's entry point hydrates the raw dict back to a typed WorkingMemory, inside the job's scope.
• Output dehydration — before a nested job returns, its working memory is dehydrated to working_memory_raw. A parent job that consumes the nested output (a controller combining branch results — PipeParallel, PipeBatch) rehydrates it on receive, inside its own already-loaded scope: the parent's library carries the same crate (checked by fingerprint via is_crate_loaded), so hydration rebinds the nested output to the parent's own dynamic classes — the identity the parent's models reference. Delivery workers never load the crate — they render from the raw dict, only best-effort hydrating the main stuff via globally registered classes (falling back to a generic dict render when the class isn't registered), so they stay crate-free. The top-level submitter rehydrates too, opening a fresh scoped library when it hasn't loaded the crate itself.

The worker-side hydration helper (pipelex/runtime_bridge/primitives/hydration.py) is open core: it iterates the raw dict, resolves each stuff's concept ref through the concept library of the current library — load_from_crate having run first, so the crate's concepts are declared and its dynamic classes registered — uses StuffContentFactory to rebuild typed content from the registered classes, and preserves aliases. A stuff whose concept is not a ref string, names a concept the library does not hold, or names a spelling the library holds more than once, fails the hydration with a PipeJobError. A current library is not a precondition, because a caller legitimately has none: Pipelex.make() sets no current library, and the transport boundary's scoped_library_for_crate(None, …) is a documented no-op that falls back to the active class registry. With no library, hydration resolves the native refs from the pinned native set and refuses anything a bundle declares, naming the stuff — the same answer the delivery executor's best-effort local hydration gives, which likewise resolves a native ref from the pinned native set with no library and falls back to the raw render for anything else, since a delivery worker never loads the crate. That reader treats an ambiguous ref as "not known here" too — a logged warning naming the colliding keys and a raw render — because rendering a result must never fail a delivery. It also guards against cross-exec class-identity mismatches (_validate_as_known_class round-trips items through model_dump()): the dynamic classes a worker re-execs from the crate are new Python identities sharing a name with whatever a data converter eagerly rebuilt, and a naive model_validate would reject the older instance.

Boundary-by-boundary (backend-neutral)

Because delivery renders from the raw dict (pipelex/pipe_run/delivery_executor.py branches on which field is populated), a delivery worker never needs the crate loaded — which is the precondition for true distributed execution. Dynamic concepts lose typed rendering on the delivery worker (a generic field-walking fallback still produces readable JSON/HTML), but the worker stays crate-free.

Per-call scoping

A single process may handle many concurrent jobs — and, in a server, many concurrent requests that are not jobs at all. If two of them define a concept named Result with different fields, they must not share a ClassRegistry or Library: structure classes are registered under a name key, so the second registration overwrites the first and the two silently swap schemas and output types. The isolation is therefore structural rather than something a caller opts into — every Library the manager opens carries its own registry:

1. ClassRegistry — library_manager.open_library() attaches a fresh ClassRegistry to every library it creates, pre-seeded from the global registry (which holds the base classes from PIPELEXPATH). A load's dynamic classes register there, never in the global registry, and no caller can forget to ask for it.
2. Library — a fresh Library (library_manager.open_fresh_library()), set as current via a ContextVar keyed by the run id. Run-id keying (not job/workflow id) is deliberate: a reused workflow id across retries/resets could otherwise let a closed predecessor's cleanup tear down a live successor's library on the same worker.
3. Lookup chain — hub.get_class_registry() reads the ContextVar, returns the scoped library's registry, and falls back to the global registry when no library is set (e.g. a data converter running outside the job scope — which is exactly why output uses deferred hydration).
4. Cleanup — library_manager.teardown(library_id) drops the library and its registry; the ContextVar is reset. No manual GC needed.

One consequence is worth stating for host code: a structure class registered while a library is current lands in that library's registry and is gone when it is torn down. A Python structure class that every load should see must be registered before any library is opened — which is what importing it at boot, or loading it from a library directory into the library that uses it, already does.

The class-registry accessor and its library scoping (pipelex/runtime_hub.py, pipelex/system/registries/class_registry_access.py, pipelex/interpreter_hub.py, pipelex/libraries/library.py, pipelex/libraries/library_manager.py) is open core — the same machinery serves direct execution, the open runner, and any host-runtime plugin.

Where open core ends

The boundary types and the host-agnostic helpers are open (source-available under the Elastic License 2.0, in the pipelex distribution):

The cross-process plumbing that drives a pipe run across the boundary — the host-side bridge entry point, the transport-prep step, and the dispatch primitives (submitter rehydration, delivery, trace flush, scoped-library, pipe classification, and the submit-arg envelope) — is not open core. Pipelex's own host-runtime embedding is a commercial capability.

A third-party orchestrator compiles against the open surface only (the Orchestrator SPI); Pipelex's own host-runtime plugins (Temporal, Mistral Workflows) additionally build on the closed side.

Next steps

• Orchestrator Plugins — the orchestration-mode seam, the Orchestrator SPI, and the open/closed boundary in contract terms.
• Pipe Routing & Execution — how PipeJobs are created and routed in direct and distributed modes.
• Content Generation Across Boundaries — how dynamic Pydantic models and large binary payloads cross a worker boundary.
• Distributed Execution — the user-facing capability overview, and the workflow-orchestration offer (Temporal-backed durable execution, the Mistral Workflows integration).