9494 "035-caller-invocation-id-uuid" ,
9595 "036-caller-invocation-id-non-uuid" ,
9696 "059-implementation-attribution-langfuse" ,
97- # 029 + 030 stay deferred in v0.11.0:
98- # - 029 (fan-out per-instance): fixture omits ``collect_field``
99- # and ``target_field`` on the fan_out cfg, plus the inner
100- # subgraph omits a ``state:`` block — both are required by
101- # the cross-cap adapter. The augmentation behavior IS
102- # verified end-to-end by the unit test
103- # ``test_observability_langfuse.py::test_metadata_augmentation_in_fan_out_isolates_per_instance``
104- # plus the OTel counterpart.
105- # - 030 (parallel-branches per-branch): the expected trace
106- # requires a per-branch dispatch span the Langfuse mapping
107- # doesn't synthesize today; the spec direction is in
108- # ``discuss-otel-parallel-branches-dispatch-span``.
109- # Sibling-skip behavior IS verified by the OTel unit test
110- # ``test_metadata_augmentation_in_parallel_branches_skips_sibling``.
111- # Both fixtures land once spec settles the dispatch-span
112- # shape AND the adapter learns to infer fan-out aggregation
113- # defaults from inner subgraphs.
114- # 039 (nested-lineage augmentation) — proposal 0045 v0.37.0.
115- # Stays deferred in PR 11. The three cases require harness
116- # extensions the existing primitives don't cover:
117- # - Case 1 + 3 (nested fan-out + fan-out-in-serial): the
118- # ``augment_metadata_from_field`` middleware captures the
119- # items list at fixture-build time from ``initial_state``;
120- # nested fan-outs read items from the OUTER instance's
121- # subgraph state (the ``inner_seed`` field threaded through
122- # per_product), which the current item-capture path can't
123- # resolve. Needs a runtime-state item-list lookup.
124- # - Case 2 (pb-inside-fan-out): introduces a new directive
125- # ``augment_metadata_from_outer_item: {key: field}`` — the
126- # pb branch's middleware sources the augmentation value
127- # from the SURROUNDING outer fan-out instance's item, not
128- # from a static value or the branch's own state. Needs a
129- # new middleware factory + a way to capture the outer-item
130- # reference at fixture-build time.
131- # 0045's behavioral contract IS exercised at unit level via the
132- # OTel observer's
133- # ``test_nested_fan_out_augmentation_reaches_outer_instance_dispatch_span``
134- # regression test, and the cross-impl coverage the spec relies
135- # on for nested cases is already in place. Activating 039 against the
136- # Langfuse harness is a follow-up PR — the harness extensions
137- # above are non-trivial and orthogonal to the engine + observer
138- # changes 0045 actually mandates.
97+ # 029 (fan-out per-instance): the fixture omits collect_field/
98+ # target_field on the fan_out cfg and the inner subgraph omits a
99+ # state: block, both required by the cross-cap adapter. The harness
100+ # synthesizes the inner state (_infer_state_fields_from_nodes) and a
101+ # throwaway aggregation sink (_synthesize_fan_out_aggregation) -- 029
102+ # asserts per-instance span metadata + sibling isolation, never the
103+ # collected results -- and augment_metadata_from_field drives the
104+ # per-instance set_invocation_metadata.
105+ "029-caller-metadata-fan-out-per-instance" ,
106+ # 030 (parallel-branches per-branch) stays deferred: a SRC gap, not a
107+ # harness one. The expected trace needs a per-branch dispatch-span
108+ # observation (inner branch spans parent under it), which §4.3 + §8.4.2
109+ # (proposal 0042 observation.metadata.branch_name) + proposal 0044
110+ # mandate. The OTel observer synthesizes it
111+ # (parallel_branches_branch_spans); the LangfuseObserver does not yet
112+ # (0044 shipped OTel-only in v0.11.0). Un-defers once that synthesis is
113+ # ported to the Langfuse observer (a src change). Sibling-skip behavior
114+ # IS verified by the OTel unit test
115+ # ``test_metadata_augmentation_in_parallel_branches_skips_sibling``.
116+ # 039 (nested-lineage augmentation, proposal 0045) stays deferred: the
117+ # three cases need harness extensions the existing primitives lack.
118+ # Cases 1 + 3 (nested fan-out / fan-out-in-serial) need the fan-out
119+ # augment middleware to read items_field from the executing subgraph's
120+ # RUNTIME state (the outer instance's threaded inner_seed), not the
121+ # build-time initial_state the current _make_augment_instance_middleware
122+ # captures. Case 2 (pb-inside-fan-out) needs a new
123+ # augment_metadata_from_outer_item factory AND depends on 030's
124+ # per-branch dispatch span landing first. 0045's contract IS exercised
125+ # at unit level via the OTel observer's
126+ # ``test_nested_fan_out_augmentation_reaches_outer_instance_dispatch_span``.
139127 }
140128)
141129
@@ -222,6 +210,64 @@ def _normalize_fan_out_subgraph_keys(spec: dict[str, Any]) -> None:
222210 fan_out_cfg ["subgraph" ] = fan_out_cfg .pop ("inner_subgraph" )
223211
224212
213+ def _synthesize_fan_out_aggregation (spec : dict [str , Any ]) -> None :
214+ """Synthesize a throwaway aggregation sink for a ``fan_out`` block that omits
215+ ``collect_field`` / ``target_field`` (fixture 029): collect the inner
216+ subgraph's ``stores_response_in`` value into a fresh outer list field, and
217+ declare the adapter's other required fan-out fields (``item_field``, the
218+ ``items_field`` source, the inner state).
219+
220+ Call AFTER ``_normalize_fan_out_subgraph_keys`` so the ``subgraph(s)`` keys
221+ are already resolved.
222+ """
223+ # 029 asserts per-instance span metadata + sibling isolation, never the
224+ # collected results, so the sink only satisfies the adapter's collect/target
225+ # requirement. The inner subgraph (spec["subgraphs"]) is shared across a
226+ # fixture's cases, so its state seed below uses setdefault to stay
227+ # idempotent; the outer state / fan_out writes are per-case.
228+ subgraphs = cast ("dict[str, Any]" , spec .get ("subgraphs" ) or {})
229+ state_block = cast ("dict[str, Any]" , spec .get ("state" ) or {})
230+ state_fields = cast ("dict[str, Any]" , state_block .get ("fields" ) or {})
231+ for node_name , node_spec in cast ("dict[str, Any]" , spec .get ("nodes" ) or {}).items ():
232+ if not isinstance (node_spec , dict ):
233+ continue
234+ fan_out_cfg = cast ("dict[str, Any] | None" , cast ("dict[str, Any]" , node_spec ).get ("fan_out" ))
235+ if fan_out_cfg is None or ("collect_field" in fan_out_cfg and "target_field" in fan_out_cfg ):
236+ continue
237+ sub_name = cast ("str | None" , fan_out_cfg .get ("subgraph" ))
238+ sub_spec = cast ("dict[str, Any]" , subgraphs .get (sub_name or "" , {}))
239+ inferred = _infer_state_fields_from_nodes (cast ("dict[str, Any]" , sub_spec .get ("nodes" ) or {}))
240+ # Any inferred field works as the collected value; the sink is never
241+ # asserted, so the first one is fine.
242+ collect_field = next (iter (inferred ), None )
243+ if collect_field is None :
244+ continue
245+ # The sink is node-scoped (one outer field per fan-out node). The item
246+ # slot is a fixed inner-state field name on purpose: distinct fan-outs
247+ # have distinct inner subgraphs, and a subgraph shared between two
248+ # fan-outs reuses the one slot, so scoping it per node would mismatch.
249+ sink = f"oa_fan_out_sink_{ node_name }
250+ item_field = "oa_fan_out_item"
251+ fan_out_cfg .setdefault ("collect_field" , collect_field )
252+ fan_out_cfg .setdefault ("target_field" , sink )
253+ # items_field mode also requires item_field (where the engine places
254+ # each item in the inner state). Seed an explicit inner state block --
255+ # the inferred response fields plus the item slot -- so
256+ # _build_inner_subgraph_with_llm uses it. The augment middleware reads
257+ # the item back out of this slot at runtime.
258+ fan_out_cfg .setdefault ("item_field" , item_field )
259+ sub_spec .setdefault ("state" , {"fields" : {** inferred , item_field : {"type" : "dict" , "default" : {}}}})
260+ state_fields [sink ] = {"type" : "list" , "reducer" : "append" , "default" : []}
261+ # The items_field source (e.g. products) must be declared on the outer
262+ # state; 029 ships it only via initial_state, so declare it here.
263+ items_field = cast ("str | None" , fan_out_cfg .get ("items_field" ))
264+ if items_field is not None :
265+ state_fields .setdefault (items_field , {"type" : "list<dict>" , "default" : []})
266+ if state_fields :
267+ state_block ["fields" ] = state_fields
268+ spec ["state" ] = state_block
269+
270+
225271def _build_augment_middlewares (
226272 case : Mapping [str , Any ],
227273) -> tuple [
@@ -244,7 +290,6 @@ def _build_augment_middlewares(
244290 """
245291 fan_out_mw : dict [str , list [Any ]] = {}
246292 branch_mw : dict [str , dict [str , list [Any ]]] = {}
247- initial_state = cast ("dict[str, Any]" , case .get ("initial_state" ) or {})
248293
249294 for node_name , node_spec_any in cast ("dict[str, Any]" , case .get ("nodes" ) or {}).items ():
250295 if not isinstance (node_spec_any , dict ):
@@ -254,11 +299,8 @@ def _build_augment_middlewares(
254299 if fan_out_cfg is not None :
255300 augment_field_map = cast ("dict[str, str] | None" , fan_out_cfg .get ("augment_metadata_from_field" ))
256301 if augment_field_map :
257- items_field = cast ("str | None" , fan_out_cfg .get ("items_field" ))
258- items_list = (
259- cast ("list[dict[str, Any]]" , initial_state .get (items_field , [])) if items_field else []
260- )
261- fan_out_mw [node_name ] = [_make_augment_instance_middleware (augment_field_map , items_list )]
302+ item_field = cast ("str | None" , fan_out_cfg .get ("item_field" ))
303+ fan_out_mw [node_name ] = [_make_augment_instance_middleware (augment_field_map , item_field )]
262304 pb_cfg = cast ("dict[str, Any] | None" , node_spec .get ("parallel_branches" ))
263305 if pb_cfg is not None :
264306 branches_cfg = cast ("dict[str, dict[str, Any]]" , pb_cfg .get ("branches" ) or {})
@@ -272,25 +314,28 @@ def _build_augment_middlewares(
272314 return fan_out_mw , branch_mw
273315
274316
275- def _make_augment_instance_middleware (field_map : dict [str , str ], items : list [dict [str , Any ]]) -> Any :
276- """Per-instance fan-out middleware that calls
277- ``set_invocation_metadata`` with per-item entries pulled from
278- ``items[current_fan_out_index()][field_path]``. Captures ``items``
279- at fixture-build time so each instance reads the same list."""
317+ def _make_augment_instance_middleware (field_map : dict [str , str ], item_field : str | None ) -> Any :
318+ """Per-instance fan-out middleware that reads the instance's own item from
319+ the ``item_field`` slot of its subgraph state and calls
320+ ``set_invocation_metadata`` with the mapped entries."""
280321
322+ # Reads runtime state, not a build-time list indexed by
323+ # current_fan_out_index(): instance middleware wraps the inner subgraph from
324+ # OUTSIDE, but the fan_out_index ContextVar is set deeper (inside the inner
325+ # node execution), so it is None here. The engine has already placed each
326+ # item in item_field by the time the chain runs, and set_invocation_metadata
327+ # lands before the inner spans open, so the instance's dispatch + inner spans
328+ # all carry the augmentation.
281329 class _AugmentInstanceMW :
282330 async def __call__ (self , state : Any , next_ : Any , / ) -> Any :
283- from openarmature .observability .correlation import ( # noqa: PLC0415
284- current_fan_out_index ,
285- )
286331 from openarmature .observability .metadata import ( # noqa: PLC0415
287332 set_invocation_metadata ,
288333 )
289334
290- idx = current_fan_out_index ()
291- if idx is not None and 0 <= idx < len ( items ):
292- item = items [ idx ]
293- entries = {key : item [field_path ] for key , field_path in field_map .items ()}
335+ item = getattr ( state , item_field , None ) if item_field else None
336+ if isinstance ( item , Mapping ):
337+ item_map = cast ( "Mapping[str, Any]" , item )
338+ entries = {key : item_map [field_path ] for key , field_path in field_map .items ()}
294339 set_invocation_metadata (** entries )
295340 return await next_ (state )
296341
@@ -657,6 +702,7 @@ async def _run_case(case: Mapping[str, Any]) -> None:
657702 # references. Pure key normalization; semantics unchanged.
658703 if isinstance (case , dict ):
659704 _normalize_fan_out_subgraph_keys (case )
705+ _synthesize_fan_out_aggregation (case )
660706 # Per proposal 0040 fixtures 029 / 030: synthesize the
661707 # augmentation middlewares that drive the per-instance /
662708 # per-branch ``set_invocation_metadata`` calls. Both flow into
0 commit comments