Add callable and conditional parallel branches (0075) (#175)

* Add callable and conditional parallel branches (0075)

A BranchSpec's work may now be given as `call` (an inline async function
over the parent state returning a parent-shaped partial) instead of a
compiled `subgraph`, and any branch may carry a `when` predicate that
skips it at dispatch. `subgraph` becomes optional; exactly one of
subgraph/call per branch and no inputs/outputs on a callable branch,
else the new compile-time ParallelBranchesInvalidBranchSpec. A node may
mix both forms. All-branches-skipped is a valid no-op.

A callable branch is the unit of work: it emits one started/completed
observer pair keyed by branch_name at the parallel-branches node's own
namespace, which both bundled observers render as a single per-branch
dispatch span (OTel) / observation (Langfuse) with no inner-node span; a
when-skipped branch emits nothing. The NODE event's branch_count is the
number of branches that dispatch (when-skipped excluded), while
branch_names stays the full declared set.

Also fix a pre-existing duplicate Langfuse observation for
parallel-branches and fan-out nodes: those emit their own NODE
observation, so synthesizing a subgraph-wrapper observation at the same
namespace duplicated it. Mirror the OTel observer's existing guard so
inner work parents under the single NODE observation.

* Pin spec to v0.66.1 and wire 0075 conformance

Advance the spec submodule, __spec_version__, pyproject spec_version,
and conformance.toml spec_pin to v0.66.1. This absorbs 0075 (v0.66.0)
and the v0.66.1 patch (a call-level-retry Langfuse-mapping clarification
the observer already satisfies, no code change). Mark proposal 0075
implemented and regenerate the bundled AGENTS.md.

Wire conformance fixtures 073-075: the adapter parses the `call` and
`when` branch directives, the fixture schema makes `subgraph` optional,
and the pipeline-utilities driver registers the three fixtures.

Document callable branches and conditional `when` in the parallel-
branches concept page, and extend the example with a pure inline
reading-time callable branch and a `when`-gated translation branch so it
covers both branch forms.

* Carry node attempt index on callable-branch events

The callable-branch started/completed pair hard-coded attempt_index=0.
Under node-level retry that mis-reports the graph-engine §6 keying tuple
and the Langfuse observation's metadata.attempt_index on the second and
later attempts. Read current_attempt_index() once at dispatch entry,
before the branch chain (which may run its own retries) so the pair
shares one value, and thread it through all three dispatches.

Author: chris-colinsky

CHANGELOG.md

@@ -13,10 +13,11 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 - **Langfuse `trace.userId` / `trace.sessionId` population** (proposal 0064, observability §8.4.1, spec v0.62.0). The Langfuse observer now promotes a recognized `userId` key in the caller-supplied invocation metadata to Langfuse's first-class `trace.userId` field (the Users dashboard), additively: the key also remains at `trace.metadata.userId`. Promotion is automatic and unconditional; an absent key leaves `trace.userId` unset. The `LangfuseClient.trace()` surface (the Protocol, the in-memory client, and the SDK adapter) gains `session_id` / `user_id`. `trace.sessionId` is sourced from `openarmature.session_id`, which the sessions capability (proposal 0020) establishes; that capability is not yet implemented in python, so the `sessionId` plumbing is in place but dormant (no source) and unset in the interim. `conformance.toml` records proposal 0064 `partial` on that basis: fixture 084 cases 2/3/4 (not session-bound, `userId` present additively, `userId` absent) run, and the session-bound cases 1/5 defer until 0020. Langfuse-only: the OTel side already carries `openarmature.session_id` and `openarmature.user.*` as span attributes, and OTel has no trace-level session/user field.
 - **Per-fetch prompt cache control: `cache_ttl_seconds`** (proposal 0072, prompt-management §5 / §6, spec v0.63.0). `PromptBackend.fetch`, `PromptManager.fetch`, and `PromptManager.get` gain an optional `cache_ttl_seconds` read-side control: `None` preserves current behavior, `0` forces a fresh read past any client-side cache, and `N > 0` bounds a served entry's staleness to N seconds; a negative value is rejected at the manager. It governs only which cached entry may be served, not whether or how results are cached. The bundled filesystem backend is cacheless and ignores it; the bundled Langfuse backend forwards it to the Langfuse SDK's `get_prompt` cache. Conformance fixtures 033/034 run through a caching harness backend (conformance-adapter §6.8: `source_read_count` plus a controllable `advance_clock`).
 - **Failure-isolation `catch` gate + cause-chain classification primitive** (proposal 0074, pipeline-utilities §6.3 / §6.4, spec v0.65.0). `FailureIsolationMiddleware` gains an optional `catch`: a set of error categories. An exception is caught only if the *derived category* of its cause chain (the outermost non-carrier link's category, resolved through the engine's `node_exception` carriers, the same value reported as `caught_exception.category`) is in the set. This closes a degrade-into-crash footgun: at a wrapping placement (subgraph, fan-out instance, branch) the engine wraps the originating failure in a carrier, so a `predicate` inspecting the surface exception sees only the carrier and misses it, whereas `catch` classifies through the carrier. `catch` composes with `predicate` as a conjunction; both default permissive (both unset stays catch-all), and a null derived category never matches a non-empty set. The carrier-skipping walk behind `catch` and `caught_exception` is promoted to a public primitive, `classify_cause_chain(exc) -> CaughtException` (the ordered `chain`, the derived `category`, and its `message` — the same record the event carries), exported from `openarmature.graph` for use in a custom `predicate`, a router, a metric, or a full-chain retry classifier. The default retry classifier stays deliberately single-level (it classifies at re-attempt granularity); this is now documented, with no behavior change. Conformance fixture 072 (catch matches through an instance-placement carrier and degrades; a non-matching catch propagates with no event). The optional native-exception-type `catch` form (spec MAY) is not shipped.
+- **Inline-callable parallel branches and conditional `when`** (proposal 0075, pipeline-utilities §11, spec v0.66.0). `ParallelBranchesNode` gains two additive branch forms. A branch may now give its work as `call`, an inline async function over the parent state returning a parent-shaped partial update, instead of a compiled `subgraph` with its own state schema and `inputs` / `outputs` projection; the returned partial is the branch's contribution directly, merged via the parent reducer with no projection. This makes the primitive adoptable for the "M heterogeneous lightweight parallel calls over shared state, each independently failure-isolated" shape (hybrid recall, paired reads) that previously dropped to a hand-rolled gather, while reusing the existing concurrency, fail-fast cancellation, per-branch failure isolation, and reducer fan-in. A branch gives its work as exactly one of `subgraph` / `call`, and a callable branch declares no `inputs` / `outputs`, else a new compile-time `ParallelBranchesInvalidBranchSpec`; a node may mix the two forms freely. A branch (either form) may also carry an optional `when` predicate over the parent state, evaluated once at dispatch: a `False` result skips the branch entirely (no dispatch, contribution, observer events, or span), and an all-skipped node is a valid no-op distinct from the compile-time `ParallelBranchesNoBranches`. A callable branch is the unit of work, so it emits one `started` / `completed` observer pair keyed by `branch_name` (rendered as a single branch span); a skipped branch emits nothing. `ParallelBranchesInvalidBranchSpec` is exported from `openarmature.graph`. Conformance fixtures 073 (two callable branches merge to disjoint fields), 074 (conditional `when` skips / dispatches), and 075 (callable branch failure-isolation degrade) run in `test_pipeline_utilities`.
 
 ### Changed
 
-- **Pinned spec advances v0.60.0 → v0.65.0** across the v0.15.0 cycle: v0.61.0 (proposal 0061, the detached-trace invocation span above), v0.62.0 (proposal 0064, the Langfuse session/user population above), v0.63.0 (proposal 0072, the prompt cache control above), the v0.63.1 patch (pipeline-utilities coverage fixtures 070/071 for the already-implemented 0069 / 0070 behavior, no new proposal), and v0.64.0 (proposal 0073, GenAI semconv adoption reconciliation: OA retains `gen_ai.system` despite the upstream rename to `gen_ai.provider.name`; textual-only, with no emitted-attribute or fixture change, so the existing `gen_ai.*` fixtures stand as the retention regression), and v0.65.0 (proposal 0074, the failure-isolation `catch` gate above). `conformance.toml` records 0061 / 0072 / 0074 `implemented`, 0064 `partial` (its `sessionId` half is dormant pending the sessions capability), and 0073 `textual-only`. Proposal 0050 needed no pin bump of its own (it was already within the pin from its v0.42.0 acceptance); its v0.14.0 `partial` entry flips to `implemented` with the per-attempt span surface above.
+- **Pinned spec advances v0.60.0 → v0.66.1** across the v0.15.0 cycle: v0.61.0 (proposal 0061, the detached-trace invocation span above), v0.62.0 (proposal 0064, the Langfuse session/user population above), v0.63.0 (proposal 0072, the prompt cache control above), the v0.63.1 patch (pipeline-utilities coverage fixtures 070/071 for the already-implemented 0069 / 0070 behavior, no new proposal), and v0.64.0 (proposal 0073, GenAI semconv adoption reconciliation: OA retains `gen_ai.system` despite the upstream rename to `gen_ai.provider.name`; textual-only, with no emitted-attribute or fixture change, so the existing `gen_ai.*` fixtures stand as the retention regression), v0.65.0 (proposal 0074, the failure-isolation `catch` gate above), v0.66.0 (proposal 0075, the inline-callable parallel branches and conditional `when` above), and the v0.66.1 patch (an observability §8 call-level-retry Langfuse-mapping clarification reconciling §8 with the per-attempt §5.5 spans: one terminal Generation per `complete()` call, not one per attempt, which the Langfuse observer already renders by driving the Generation from the terminal `LlmCompletionEvent` / `LlmFailedEvent` and skipping the per-attempt `LlmRetryAttemptEvent`; no behavior or fixture change). `conformance.toml` records 0061 / 0072 / 0074 / 0075 `implemented`, 0064 `partial` (its `sessionId` half is dormant pending the sessions capability), and 0073 `textual-only`. Proposal 0050 needed no pin bump of its own (it was already within the pin from its v0.42.0 acceptance); its v0.14.0 `partial` entry flips to `implemented` with the per-attempt span surface above.
 
 ## [0.14.0] — 2026-06-17
 

conformance.toml

@@ -32,7 +32,7 @@
 
 [manifest]
 implementation = "openarmature-python"
-spec_pin = "v0.65.0"
+spec_pin = "v0.66.1"
 
 # Status values:
 #   implemented   — shipped behavior matches the proposal's contract
@@ -726,3 +726,8 @@ note = "Governance + observability §5.5 rationale change: reconciles the gen_ai
 status = "implemented"
 since = "0.15.0"
 note = "FailureIsolationMiddleware gains an optional `catch` set of error categories (§6.3): an exception is caught only if the DERIVED category of its cause chain (the outermost non-carrier link, resolved THROUGH node_exception carriers -- the same value reported as caught_exception.category) is in the set, composing with `predicate` as a conjunction (both default permissive, both unset = catch-all; a null derived category never matches a non-empty set). This classifies a carrier-wrapped failure correctly at a wrapping placement where a surface check sees only the carrier. The §6.4 cause-chain classification walk is promoted to a public primitive classify_cause_chain(exc) -> CaughtException (the existing failure-isolation record: chain + derived category + message) in openarmature.graph, shared by the catch gate, the emitted event, and any consumer. §6.1: the default retry classifier's single-level depth is documented as deliberate (re-run granularity vs §6.3 full-chain degrade); no behavior change. Fixture 072 (catch matches through an instance-placement carrier and degrades; a non-matching catch propagates with no event). The optional native-exception-type catch sugar (spec MAY) is not shipped."
+
+[proposals."0075"]
+status = "implemented"
+since = "0.15.0"
+note = "ParallelBranchesNode gains two additive branch forms. (1) Inline-callable branches (§11.1.1): a BranchSpec may give its work as `call` (an async function over the parent state returning a parent-shaped partial update) instead of a compiled `subgraph` + inputs/outputs projection; the contribution is the returned partial directly, merged via the parent reducer with no projection (§11.4). Exactly one of subgraph/call per branch, and a callable branch declares no inputs/outputs, else parallel_branches_invalid_branch_spec (a new compile-time category); a node MAY mix subgraph and callable branches. Per-leg failure isolation on a callable branch is the existing §11.7 branch-middleware contract (wrap the callable in FailureIsolationMiddleware). (2) Conditional branches (§11.10): a BranchSpec may carry an optional `when` predicate (parent_state) -> bool, evaluated once at dispatch; false skips the branch entirely (no dispatch, contribution, observer events, or span). All-branches-skipped is a valid no-op, distinct from the compile-time parallel_branches_no_branches (empty declared mapping). graph-engine §6 / observability §5.7: a callable branch is the unit -- it emits one started/completed pair keyed by branch_name (rendered as a branch span via the existing §5.7 machinery), a skipped branch emits nothing. Fixtures 073 (two callable branches merge to disjoint fields), 074 (when false skips / true dispatches), 075 (callable branch + FailureIsolationMiddleware degrades, sibling completes, category resolves through the chain)."

docs/concepts/parallel-branches.md

@@ -76,6 +76,61 @@ order**: first the branch declared first in the `branches` dict,
 then the next, and so on. This is deterministic regardless of which
 branch's inner work finishes first.
 
+## Lightweight callable branches
+
+Not every branch needs a compiled subgraph. When a branch is really
+just *"call this one async function over the shared state"*, give its
+work as `call` instead of `subgraph`: an async function that reads the
+parent state and returns a parent-shaped partial update. No subgraph,
+no state schema, no `inputs` / `outputs` projection.
+
+```python
+async def vector_recall(state: SearchState) -> dict[str, object]:
+    hits = await vector_store.query(state.query)
+    return {"vector_hits": hits}
+
+async def keyword_recall(state: SearchState) -> dict[str, object]:
+    hits = await fts_index.search(state.query)
+    return {"keyword_hits": hits}
+
+builder.add_parallel_branches_node(
+    "recall",
+    branches={
+        "vector": BranchSpec(call=vector_recall),
+        "keyword": BranchSpec(call=keyword_recall),
+    },
+)
+```
+
+This is the right form for hybrid recall (vector plus full-text),
+paired reads, and other *"M heterogeneous lightweight calls over
+shared state, each independently failure-isolated"* shapes. Each
+branch returns parent fields directly, so the returned partial is the
+branch's contribution as-is: it merges via the parent's reducers in
+branch insertion order, exactly like a subgraph branch's projected
+outputs, with no projection step.
+
+A branch gives its work as **exactly one** of `subgraph` or `call`.
+Declaring both, neither, or `inputs` / `outputs` on a callable branch
+(which has no subgraph state to project against) is a compile-time
+`ParallelBranchesInvalidBranchSpec`. A single parallel-branches node
+may freely mix subgraph branches and callable branches.
+
+Per-leg failure isolation works the same on a callable branch: wrap it
+in a `FailureIsolationMiddleware` via the branch's `middleware` (see
+[Branch middleware](#branch-middleware) below). A failing callable
+that degrades to its configured update "succeeds" from the node's view,
+so it contributes the degraded update and does not trip `fail_fast`.
+
+Because a callable branch has no inner nodes, the branch itself is the
+unit of work: it emits one `started` / `completed` observer pair keyed
+by its `branch_name`, so per-branch observability comes from the branch
+as a whole. The bundled OTel observer renders it as a single per-branch
+dispatch span keyed by `branch_name`, with no inner-node spans beneath
+it (a subgraph branch, by contrast, spans its inner nodes under the
+dispatch span); the Langfuse observer renders it as a single observation
+under the node. A `when`-skipped branch produces no span at all.
+
 ## Error policy
 
 - **`"fail_fast"`** (default): the first branch failure cancels
@@ -113,6 +168,44 @@ Branch middleware is independent across branches: branch A may
 have `[retry, timing]`; branch B may have `[]`; branch C may have
 some custom breaker. Each branch's chain composes in isolation.
 
+## Conditional branches
+
+Any branch (subgraph or callable) may carry an optional `when`
+predicate over the parent state. It is evaluated **once at dispatch**,
+against the state the parallel-branches node received:
+
+```python
+builder.add_parallel_branches_node(
+    "recall",
+    branches={
+        # Only run the vector leg when the query carries an embedding.
+        "vector": BranchSpec(call=vector_recall, when=lambda s: s.embedding is not None),
+        "keyword": BranchSpec(call=keyword_recall),
+    },
+)
+```
+
+- `when` absent (default): the branch always dispatches.
+- `when` returns `True`: the branch dispatches normally.
+- `when` returns `False`: the branch is **skipped** entirely. It is
+  not dispatched, runs no work, contributes nothing to parent state,
+  and emits no observer events or span. It simply does not appear in
+  the run.
+
+This expresses *"skip this leg when it does not apply"* directly,
+without an always-run self-no-op branch cluttering the trace. The
+`branches` mapping and its insertion order are unchanged; skipping is
+a runtime decision over the declared set, and the branches that do
+dispatch keep their insertion-order determinism. If **every** branch
+is skipped, the node completes as a valid no-op (it contributes
+nothing), distinct from the compile-time `ParallelBranchesNoBranches`
+that an empty declared mapping raises.
+
+Keep `when` a deterministic function of the dispatch-time parent
+state so repeated runs skip the same set; a `when` that consults
+nondeterministic sources carries the same caveat as a conditional
+middleware.
+
 ## Composition with other constructs
 
 Parallel branches compose with the rest of the engine the way
@@ -144,8 +237,10 @@ Per-branch progress is not individually persisted in v1.
   this subgraph for each item in a list," reach for
   [fan-out](fan-out.md).
 - **Not a router.** A router is a conditional-edge pattern that
-  picks one branch based on state. Parallel branches runs *all*
-  branches concurrently.
+  picks one path based on state. Parallel branches runs all the
+  branches that dispatch concurrently; a branch's `when` can skip an
+  individual branch, but it gates each branch independently rather
+  than selecting exactly one.
 - **Not a coordinator.** Branches don't communicate with each other
   during execution; if branch B's work depends on branch A's
   output, you want a linear pipeline (A → B), not parallel branches.