diff --git a/AGENTS.md b/AGENTS.md index 446660a..887ef62 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -150,6 +150,43 @@ User-facing docs live in `docs/` and build via MkDocs Material; the deployed site is at `openarmature.ai`. CI build + deploy is in `.github/workflows/docs.yml`. Local preview: `uv run mkdocs serve`. +## Docstrings vs `#` comments + +They have different audiences, and the split is not a style preference. + +**Docstrings are published.** `mkdocstrings` renders them into +`docs/reference/*.md` and they surface through `help()`, so a docstring is +shipped end-user documentation. Write for someone *calling* the API who +cannot see the implementation: what it does, what the arguments and return +mean, what it raises, and any constraint the caller has to honour. If a +sentence only makes sense to someone editing the body, it is not a +docstring. + +**`#` comments are for maintainers**, and carry everything a caller does +not need: spec section references, MUST / SHOULD / MAY rules, rationale, +rejected alternatives, and why a line is the way it is. Keep them short +per the comment rules in the global `CLAUDE.md`. + +So these move out of a docstring and into a `#` comment: + +- Spec citations (`§8.4.2`, `proposal 0119`, `spec v0.116.0`) and bare + prose like "the spec defines" or "the spec requires". +- Normative language about what an implementation MUST or MAY do. A + caller does not implement the spec; we do. +- Rationale for the implementation, and comparisons to how another + module or observer handles the same thing. + +Two things stay in docstrings even though they look like spec references: +a `spec/` **path** (it is a location, not a normative claim) and a +parameter genuinely named `spec`. + +Applies to `tests/` too. A test docstring is read by whoever is deciding +whether the test still earns its place. + +Health check: the median docstring here is 6 lines. Past about 20, ask +whether the extra is caller-facing documentation or maintainer notes that +drifted in. + ## Engine design notes that are easy to miss - `State` is `frozen=True` AND `extra="forbid"`. Nodes that return an diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py index 32d7b99..77574fb 100644 --- a/src/openarmature/graph/compiled.py +++ b/src/openarmature/graph/compiled.py @@ -320,31 +320,20 @@ def _find_innermost_fan_out_instance_state( # fan-out's full key is (namespace_before_fan_out, fan_out_name) # where namespace_before_fan_out + (fan_out_name,) == prefix. for split in range(len(prefix), 0, -1): - # The fan-out at prefix[:split] registered under its ENCLOSING lineage at - # its own level, prefix depth split-1. Reconstruct that depth from the - # current chains so a fan-out nested inside an outer instance routes to - # the right outer instance's entry. + # The fan-out at prefix[:split] registered under its enclosing lineage + # at depth split-1, so reconstruct that depth from the current chains + # and a nested fan-out routes to the right outer instance's entry. # - # Same builder the registration uses, and BOTH axes slice to - # the same depth. Every descent grows all three of `namespace_prefix`, - # `fan_out_index_chain` and `branch_name_chain` by exactly one entry, so - # index i of either chain corresponds to `namespace_prefix[i]`. A branch - # descent is no exception: `descend_into_parallel_branch` appends the - # parallel-branches NODE name to the namespace in the same call it - # appends the branch name to the chain. (The BRANCH name never enters - # the namespace; the node name does. Conflating those two is what an - # earlier version of this comment did, to justify leaving the branch - # chain unsliced.) + # BOTH axes slice to that depth, using the registration's own builder. + # Every descent grows `namespace_prefix` and both chains by one entry, + # so chain index i corresponds to `namespace_prefix[i]`; a branch + # descent puts the parallel-branches NODE name in the namespace and the + # branch name in the chain, in one call. # - # Unsliced was reachable-wrong rather than harmless: with a branch - # descent at or below the candidate fan-out's depth, the key carried - # branch entries the registration key did not, so the lookup missed and - # `completed_inner_positions` silently lost a position. It does not fire - # today only because branch descent runs with `checkpointer=None` - # (observer.py), which makes `_maybe_save_checkpoint` early-return, so - # nothing calls this from inside a branch. That is a policy in another - # module, not a property of this key -- slice correctly and the - # correctness stops depending on it. + # Nothing reaches this from inside a branch today, but only because + # branch descent runs with `checkpointer=None` (observer.py). An + # unsliced branch axis carries entries the registration key lacks, so + # the lookup misses and `completed_inner_positions` loses a position. key = fan_out_progress_key( prefix[: split - 1], prefix[split - 1], @@ -392,21 +381,16 @@ def _project_fan_out_progress( # still order deterministically (preserving the byte-identical-record guarantee). for (namespace, name, _fan_out_lineage, branch_lineage), exec_state in sorted(state_dict.items()): # A branch-nested fan-out is NOT projected. `FanOutProgress` has no - # branch field and the spec keys the record by - # `(namespace, fan_out_node_name, enclosing_fan_out_lineage)` - # (pipeline-utilities §10.11), so sibling branches -- which share all - # three -- would emit entries indistinguishable on the record's own key. - # `_restore_fan_out_progress_state` is `out[key] = ...`, so restoring - # such a record silently keeps whichever sorted last and discards the - # rest. + # branch field and §10.11 keys the record by `(namespace, + # fan_out_node_name, enclosing_fan_out_lineage)`, which sibling branches + # all share, so their entries would be indistinguishable and restore + # (`out[key] = ...`) would keep whichever sorted last. # - # Emitting them buys nothing: restore always rebuilds the branch - # component as empty, so a branch-nested re-entry never positively - # matches one of these entries and re-runs regardless. Skipping keeps - # the record's key invariant intact and prevents an entry no live - # execution can ever key to from being restored, never matched, never - # popped by the branch-bearing cleanup key, and then re-projected onto - # every later save for the rest of the invocation. + # Nothing is lost by skipping: restore rebuilds the branch component as + # empty, so a branch-nested re-entry never matches such an entry and + # re-runs either way. Emitting one would leave a record no live + # execution can key to, never popped by the branch-bearing cleanup key + # and re-projected onto every later save. if branch_lineage: continue instances = tuple( @@ -465,25 +449,18 @@ def _restore_fan_out_progress_state( completed_inner_positions=list(inst.completed_inner_positions), ) ) - # Key by the persisted enclosing fan-out instance lineage (proposal - # 0085 consume-side). The in-memory tracking key's third element is the - # non-None fan_out_index chain of the enclosing fan-out instances (see - # FanOutNode.run_with_context and _find_innermost_fan_out_instance_state), - # so project the record's enclosing_fan_out_lineage down to that same - # flat index tuple. This realizes §10.11's no-mis-skip invariant + the - # exactly-once extension for free via the existing keyed re-entry: - # - a record entry with a lineage matching the re-entering execution's - # lineage element-for-element is a POSITIVE match, so its completed - # instances are skipped and rolled forward correctly; - # - an empty saved lineage (a flat / top-level / subgraph-nested - # fan-out, OR a legacy pre-0085 record) keys to (), which a non-empty - # re-entering lineage never matches -- the re-entry misses and re-runs - # from scratch rather than applying a different enclosing instance's - # skips (correctness-preserving per §10.7). Empty positively matches - # empty, so flat records resume exactly as before this field existed. - # The crash-PRODUCED write side (projecting the rich lineage onto a real - # crash record) is a tracked follow-up; until then a real nested-fan-out - # crash resumes at the safe re-run floor -- see _project_fan_out_progress. + # Key by the persisted enclosing fan-out lineage (0085 consume-side), + # projected down to the same flat index tuple the in-memory tracking key + # uses. §10.11's no-mis-skip invariant falls out of the keyed re-entry: + # a lineage matching element-for-element skips its completed instances, + # while an empty saved lineage keys to () and never matches a non-empty + # one, so a mismatched record re-runs from scratch rather than applying + # another instance's skips (§10.7). Empty matches empty, so flat and + # pre-0085 records resume unchanged. + # + # The write side, projecting the lineage onto a real crash record, is a + # tracked follow-up; until then a nested-fan-out crash resumes at the + # safe re-run floor. See _project_fan_out_progress. lineage = tuple(e.fan_out_index for e in fp.enclosing_fan_out_lineage) # The branch axis has no field on the record to source it from, so it # restores empty. A branch-nested fan-out re-enters with a NON-empty @@ -1389,23 +1366,18 @@ async def _invoke( step_result = await self._step_parallel_branches_node(node, current, state, context) elif isinstance(node, SubgraphNode): # Subgraph wrappers are transparent to the observer protocol - # (per fixture 013): no event is dispatched for the wrapper - # itself, the step counter does not advance for it, and any - # `RuntimeGraphError` bubbling up from the subgraph's - # _invoke is already wrapped with the inner node's identity - # — pass it through. Other exceptions (projection errors, - # subgraph state-class init errors) escape the spec §4 - # categories, so we wrap them as NodeException tagged with - # the wrapper's name. + # (fixture 013): no event for the wrapper, no step advance. A + # `RuntimeGraphError` from the subgraph already carries the + # inner node's identity, so it passes through; anything else + # (projection, subgraph state-class init) escapes the §4 + # categories and is wrapped as a NodeException tagged with the + # wrapper's name. # - # Per pipeline-utilities §4: the parent's middleware wraps - # the subgraph dispatch as a single atomic call. Subgraph- - # internal nodes have their own middleware (from the - # subgraph's own CompiledGraph.middleware tuple) and do - # NOT see the parent's middleware. Cast erases ChildT - # because the dispatcher only needs to invoke `node.run` - # and pass the parent's chain — the inner state class - # lives on the subgraph's own CompiledGraph. + # pipeline-utilities §4: the parent's middleware wraps the + # dispatch as one atomic call, and subgraph-internal nodes see + # only the subgraph's own middleware. The cast erases ChildT + # because the dispatcher only invokes `node.run`; the inner + # state class lives on the subgraph's CompiledGraph. sub = cast("SubgraphNode[StateT, State]", node) step_result = await self._step_subgraph_node(sub, current, state, context) else: @@ -1545,13 +1517,10 @@ async def innermost(s: Any) -> Mapping[str, Any]: # any exception that escapes the chain, OUTSIDE this layer. attempt_counter[0] += 1 - # Per graph-engine §6 (clarified in v0.16.1): event - # emission reads ``attempt_index`` from the ContextVar set - # by any enclosing retry middleware — direct (per-node - # MW) or transitive (instance / branch MW on a subgraph - # the retry re-invokes). The engine itself no longer - # writes the var; innermost-wins precedence falls out of - # Python's ContextVar token-stack semantics. + # graph-engine §6: ``attempt_index`` comes from the ContextVar set + # by any enclosing retry middleware, direct or transitive through a + # subgraph the retry re-invokes. The engine does not write it, and + # innermost-wins falls out of ContextVar token-stack semantics. attempt_index = current_attempt_index() self._dispatch_started(context, current, namespace, step, s, attempt_index=attempt_index) @@ -1837,32 +1806,20 @@ async def _step_fan_out_node( # hardcoded 0. attempt_counter: list[int] = [0] - # Resolve the fan-out config eagerly so the resolved values - # ride on every fan-out node event (per spec proposal 0013, - # v0.10.0: ``fan_out_config`` is populated on fan-out node - # events including retried attempts). For ``items_field`` - # mode the count is ``len(parent_state.)``; for - # ``count`` mode it's ``_resolve_count``. ``_resolve_concurrency`` - # is pure regardless. Repeating these inside - # ``FanOutNode.run_with_context`` is cheap and matches the - # values surfaced here. - # Lazy import: function-scope to avoid a module-top - # textual cycle CodeQL flags. ``fan_out`` has a - # TYPE_CHECKING back-reference to this module, so the - # static-analyzer view of an importable cycle goes away - # when the engine doesn't reach into ``fan_out`` at module - # load time. Fires once per fan-out step. + # Resolved eagerly so `fan_out_config` rides every fan-out node event + # including retried attempts (0013). `FanOutNode.run_with_context` + # repeats the resolution, which is cheap and yields the same values. + # + # Function-scope import: `fan_out` has a TYPE_CHECKING back-reference to + # this module, and keeping the import out of module scope is what stops + # CodeQL seeing an importable cycle. from .fan_out import _resolve_concurrency, _resolve_count # noqa: PLC0415 - # Resolver failures (callable count/concurrency raising, - # ``getattr`` on a malformed state, etc.) used to land inside - # ``innermost``'s ``except Exception → NodeException`` block - # below and produce a started/completed event pair via the - # surrounding dispatches. Hoisting resolution out of - # ``run_with_context`` for the eager ``FanOutEventConfig`` - # build moved them past that scope, so re-establish the - # contract here: surface a started/completed pair with - # ``fan_out_config=None`` (we never built one) and raise as + # Resolution sits outside ``innermost``'s ``except Exception → + # NodeException`` scope, so a resolver failure (a callable count or + # concurrency raising, ``getattr`` on a malformed state) has to + # reproduce that contract here: a started/completed pair with + # ``fan_out_config=None``, since none was built, and a # ``NodeException``. try: if node.config.items_field is not None: @@ -1987,23 +1944,13 @@ async def innermost(s: Any) -> Mapping[str, Any]: ) raise except CheckpointError as e: - # Spec proposal 0012's pairing contract requires - # every started event have a paired completed - # event. CheckpointError categories (notably - # proposal 0029's count-drift raise) are sibling- - # typed to RuntimeGraphError and propagate to the - # invoke() caller unwrapped so callers can branch - # on ``e.category``. To preserve pairing while - # keeping ``NodeEvent.error`` typed as - # ``RuntimeGraphError | None`` per spec §6, the - # completed event carries a ``NodeException`` - # wrapper whose ``__cause__`` is the original - # CheckpointError. The bare ``raise`` re-raises - # the active exception (the CheckpointError, not - # the wrapper) so the caller still sees the - # checkpoint category. Mirrors the ``except - # Exception`` branch below structurally; the - # difference is what gets re-raised. + # 0012 pairs every started event with a completed one, but + # a CheckpointError is sibling-typed to RuntimeGraphError + # and must reach the caller unwrapped so they can branch on + # ``e.category``. So the completed event carries a + # ``NodeException`` wrapper, keeping ``NodeEvent.error`` + # typed per §6, while the bare ``raise`` re-raises the + # active CheckpointError rather than the wrapper. wrapped = NodeException(node_name=current, cause=e, recoverable_state=s) self._dispatch_completed( context, @@ -2074,21 +2021,14 @@ async def innermost(s: Any) -> Mapping[str, Any]: # Per proposal 0045: drive per-depth chain ContextVars. fan_out_chain_token = _set_fan_out_index_chain(context.fan_out_index_chain) branch_chain_token = _set_branch_name_chain(context.branch_name_chain) - # Per spec §10.11 the ``fan_out_progress`` entry is "in-flight - # only"; the fan-out's own completion save below is the last - # point where the entry is needed (proposal 0009: that save - # "also finalizes fan_out_progress to mark all instances - # complete"). Pop the entry after the save fires, regardless of - # whether the fan-out completed normally, short-circuited, or - # raised, so subsequent saves in this invocation do not carry - # stale fan-out progress and a retry middleware on the fan-out - # node sees a fresh tracked state on the second attempt. - # Match the lineage-aware key FanOutNode.run registers (namespace, node - # name, enclosing fan-out instance lineage, enclosing branch lineage) so a - # nested fan-out's cleanup pops its OWN entry, not a sibling's. Both axes - # are required: the pop below passes a default, so a key that does not - # match the registration silently no-ops and leaves the entry to be - # carried into later saves as stale progress. + # §10.11 makes the ``fan_out_progress`` entry in-flight only, and the + # completion save below is its last use. Pop it after that save on every + # exit path, normal, short-circuited or raised, so later saves carry no + # stale progress and a retry middleware sees fresh tracked state. + # + # Key on all four axes FanOutNode.run registers, so a nested fan-out + # pops its own entry. The pop passes a default, so a mismatched key + # silently no-ops and the entry survives as stale progress. progress_key = fan_out_progress_key( context.namespace_prefix, current, @@ -2517,21 +2457,14 @@ async def _maybe_save_checkpoint( checkpointer = context.checkpointer if checkpointer is None: return - # Per spec §10.2: NodePosition.namespace is the containing- - # graph chain (outermost first), NOT including the node's - # own name — distinct from NodeEvent.namespace which - # includes it. The two are related by - # NodeEvent.namespace == NodePosition.namespace + - # (NodePosition.node_name,). + # §10.2: NodePosition.namespace excludes the node's own name, unlike + # NodeEvent.namespace, so `NodeEvent.namespace == + # NodePosition.namespace + (NodePosition.node_name,)`. # - # Inner-position scoping (per §10.11.1, in-flight observability - # rules): a position from inside a fan-out instance is scoped - # to that instance's inner subgraph execution, NOT the outer - # graph. It accumulates on the per-instance state's - # ``completed_inner_positions`` list rather than the outer - # ``completed_positions`` list. The outer list keeps the outer - # graph's positions plus the fan-out node's own completion - # position (added by ``_step_fan_out_node`` after fan-in). + # §10.11.1: a position from inside a fan-out instance is scoped to that + # instance and accumulates on its ``completed_inner_positions``. The + # outer ``completed_positions`` holds the outer graph's positions plus + # the fan-out node's own, added by ``_step_fan_out_node`` after fan-in. position = NodePosition( namespace=context.namespace_prefix, node_name=node_name, diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py index 506aaf6..28fb12f 100644 --- a/src/openarmature/graph/events.py +++ b/src/openarmature/graph/events.py @@ -86,20 +86,17 @@ class FanOutEventConfig: concurrency: int | None error_policy: str parent_node_name: str - # OPTIONAL fifth key, beyond the four graph-engine §6 requires. §6 says - # implementations MUST present all four whenever the field is populated; - # it does not close the set, and nothing forbids an additional key. + # An OPTIONAL fifth key. graph-engine §6 requires all four whenever the + # field is populated but does not close the set. # - # It exists because the declared identity otherwise reaches an observer ONLY - # through an inner node event's `subgraph_identities`. When a fan-out's - # instance middleware short-circuits -- issues its call and returns without - # calling `next_call` -- no inner node event is ever emitted, so a synthesized - # per-instance span carried `openarmature.subgraph.name=''` despite a - # declared identity, with nothing able to repair it. Same reason - # `parent_node_name` rides here rather than being rederived. + # It is the only carrier of the declared identity when no inner node event + # is emitted, which happens when a fan-out's instance middleware issues its + # call and returns without calling `next_call`. Otherwise the identity + # reaches an observer only through `subgraph_identities`. `parent_node_name` + # rides here for the same reason. # - # `None` when the fan-out declares no identity, which is the existing - # "no identity tracked" case and stays the empty string on the span. + # `None` when the fan-out declares no identity, which stays the empty string + # on the span. subgraph_identity: str | None = None @@ -292,34 +289,20 @@ class NodeEvent: # without re-deriving it from successive events. fan_out_index_chain: tuple[int | None, ...] = () branch_name_chain: tuple[str | None, ...] = () - # Per observability §5.3 + the coord-thread - # ``clarify-subgraph-name-semantics`` resolution: chain of - # compiled-subgraph identities parallel to the wrapper-depth - # positions of ``namespace``. Index ``i`` is the identity for - # the wrapper at ``namespace[i]`` (or ``None`` when that - # wrapper has no tracked identity); chain length equals the - # depth of wrapper nesting (always ``< len(namespace)`` since - # the last element of ``namespace`` is the current node, not - # a wrapper). Observers read by depth and emit it as - # ``observation.metadata.subgraph_name`` (Langfuse) / - # ``openarmature.subgraph.name`` (OTel), falling back to the - # empty string when ``None`` per §5.3's "if the implementation - # tracks one" clause. + # §5.3: compiled-subgraph identities parallel to the wrapper positions of + # ``namespace``, so index ``i`` is the identity for the wrapper at + # ``namespace[i]``, or ``None`` where it tracks none. Always shorter than + # ``namespace``, whose last element is the current node rather than a + # wrapper. Observers read by depth and fall back to the empty string. subgraph_identities: tuple[str | None, ...] = () - # Per observability §3.4 + §5.6 (proposal 0034): snapshot of the - # caller-supplied invocation metadata at event-construction - # time. The engine reads ``current_invocation_metadata()`` when - # it constructs the event (in the engine task / node body's - # Context); the observer reads from the snapshot on the event - # rather than re-reading the ContextVar at observer time — - # critical because the observer runs on the engine's - # ``deliver_loop`` task whose Context is frozen at invoke time - # (asyncio.create_task copies the parent Context at task - # creation), so the live ContextVar value in the deliver_loop - # would NOT reflect mid-invocation augmentations made by node - # bodies running in the main engine task. Observers emit each - # entry as ``openarmature.user.`` (OTel, §5.6) / - # ``metadata.`` (Langfuse, §8.4.1+§8.4.2). + # §3.4 + §5.6: a snapshot taken where the event is constructed, in the + # engine task's Context. An observer MUST read this rather than + # re-reading the ContextVar: it runs on `deliver_loop`, whose Context was + # copied at task creation and so never sees a mid-invocation augmentation + # made by a node body. + # + # Rendered as `openarmature.user.` (OTel) / `metadata.` + # (Langfuse). caller_invocation_metadata: Mapping[str, AttributeValue] = field(default_factory=lambda: _EMPTY_METADATA) @@ -453,34 +436,17 @@ class InvocationCompletedEvent: correlation_id: str | None -# Spec: realizes proposal 0049's first spec-normatively-typed event -# variant on the observer event union (graph-engine §6 + -# observability §5.5.7). Dispatched on every LLM provider call that -# returns a structured response, alongside the calling node's -# NodeEvent pair. Failure cases (provider exceptions, malformed -# responses) flow through the existing exception path and do NOT -# emit this variant. Not subject to the §6 ``phases`` subscription -# filter (matches MetadataAugmentationEvent / InvocationStartedEvent -# / InvocationCompletedEvent treatment). -# -# Field naming matches the spec-canonical names verbatim per the spec -# Q5 ack — Python snake_case happens to match the spec table 1:1. +# graph-engine §6 + observability §5.5.7. Dispatched on every LLM call that +# returns a structured response, alongside the calling node's NodeEvent pair. +# Failures take the exception path and emit `LlmFailedEvent` instead. Not +# subject to the §6 ``phases`` filter. Field names match the spec table. # -# Spec proposal 0057 (v0.51.0) extension: adds 8 additive request-side -# fields (input_messages, output_content, request_params, -# request_extras, active_prompt, active_prompt_group, call_id, -# response_model) and renames request_id → response_id to match the -# response-side data the field carries. Inline image bytes in -# input_messages MUST be redacted per observability §5.5.5 before -# population — the provider reuses _serialize_messages_for_payload -# which already enforces the redaction. The three payload-bearing -# fields (input_messages, output_content, request_extras) are -# populated unconditionally on the typed event per §5.5.7; observer- -# side privacy gates (OTel disable_provider_payload, Langfuse equivalents) -# apply at rendering, symmetric with the §5.5.1 span attribute path. -# Custom queryable observers (per observability §9) own their own -# redaction posture — gating belongs at rendering with the consumer's -# awareness. +# The payload-bearing fields (input_messages, output_content, request_extras) +# are populated unconditionally per §5.5.7, with the observer-side privacy gates +# applying at RENDERING, symmetric with the §5.5.1 span attribute path. A custom +# §9 observer therefore owns its own redaction posture. Inline image bytes are +# redacted per §5.5.5 before population, by the provider's +# `_serialize_messages_for_payload`. @dataclass(frozen=True) class LlmCompletionEvent: """A typed LLM provider call event delivered to observers. @@ -629,27 +595,17 @@ class LlmCompletionEvent: branch_name_chain: tuple[str | None, ...] = () -# Spec: realizes proposal 0058's second spec-normatively-typed event -# variant on the observer event union (graph-engine §6 + -# observability §5.5.7), accepted at spec v0.53.0. Dispatched on the -# observer delivery queue whenever a provider.complete() call raises -# a §7 category exception — covers BOTH the adapter-caught provider- -# exception path AND the pre-send validation raise path -# (provider_invalid_request / provider_unsupported_content_block -# raise before any provider contact). The event is dispatched -# ALONGSIDE the exception, not in place of it; caller-side exception -# flow is unchanged. +# graph-engine §6 + observability §5.5.7. Dispatched whenever `complete()` +# raises a §7 category exception, covering both the adapter-caught provider +# path and the pre-send validation raises that never reach the provider. It +# rides ALONGSIDE the exception; caller-side flow is unchanged. # -# Mutual exclusion with LlmCompletionEvent on the same -# provider.complete() call — implementations MUST NOT emit both for -# the same call. Conformance fixture 072 locks this down. +# Mutually exclusive with LlmCompletionEvent for one `complete()` call, never +# both. Fixture 072 locks that down. # -# Privacy posture identical to LlmCompletionEvent: input_messages / -# request_params / request_extras are populated unconditionally per -# §5.5.7; observer-side privacy gates (OTel disable_provider_payload, -# Langfuse equivalents) apply at rendering. Inline image bytes are -# redacted per observability §5.5.5 before population. Custom -# queryable observers own their own redaction posture. +# Privacy posture as LlmCompletionEvent: payload fields populated +# unconditionally, gates applied at rendering, image bytes redacted per +# §5.5.5. @dataclass(frozen=True) class LlmFailedEvent: """A typed LLM provider call failure event delivered to observers. @@ -976,18 +932,11 @@ class EmbeddingFailedEvent: branch_name_chain: tuple[str | None, ...] = () -# Spec: realizes graph-engine §6 -- the typed RerankEvent / RerankFailedEvent -# pair (proposal 0060, retrieval-provider rerank capability), the rerank -# sibling to the EmbeddingEvent / EmbeddingFailedEvent pair. Dispatched on the -# observer delivery queue per RerankProvider.rerank() call: the success variant -# after the response is parsed + validated (retrieval-provider §6), the failure -# variant alongside a raised §7 category exception (mutually exclusive per -# call). Scalar fan_out_index / branch_name only, matching the embedding pair -# (the lineage chains arrive uniformly across the provider events with proposal -# 0084). query / documents / request_extras / output_results are payload- -# bearing, populated unconditionally; observer-side privacy gates (OTel -# disable_provider_payload, Langfuse equivalents) apply at rendering, symmetric -# with EmbeddingEvent. +# graph-engine §6: the rerank sibling to the Embedding pair. One per +# `rerank()` call, the success variant after the response is parsed and +# validated, the failure variant alongside a raised §7 exception, never both. +# query / documents / request_extras / output_results are payload-bearing and +# populated unconditionally, with the gates applied at rendering. @dataclass(frozen=True) class RerankEvent: """A typed rerank provider call event delivered to observers. @@ -1120,17 +1069,11 @@ class RerankFailedEvent: branch_name_chain: tuple[str | None, ...] = () -# Spec: realizes pipeline-utilities §6.3 failure-isolation middleware -# (proposal 0050). Emitted by FailureIsolationMiddleware when it -# catches an exception escaping the inner chain and substitutes a -# degraded partial update. A distinct framework-emitted event kind -# (NOT a NodeEvent — does not reuse node_name / namespace / error), -# mirroring the proposal 0040 MetadataAugmentationEvent mechanism: -# enqueued on the engine's serial observer-delivery queue via -# ``current_dispatch()`` and NOT subject to the observer ``phases`` -# filter (matches MetadataAugmentationEvent / InvocationStartedEvent / -# InvocationCompletedEvent / LlmCompletionEvent / LlmFailedEvent -# treatment). +# pipeline-utilities §6.3. Emitted by FailureIsolationMiddleware when it catches +# an exception escaping the inner chain and substitutes a degraded update. A +# framework event kind rather than a NodeEvent, so it reuses none of node_name / +# namespace / error, and like the other framework events it is enqueued via +# ``current_dispatch()`` and exempt from the ``phases`` filter. @dataclass(frozen=True) class FailureIsolatedEvent: """A failure-isolation event delivered to observers. diff --git a/src/openarmature/graph/fan_out.py b/src/openarmature/graph/fan_out.py index 9087bdb..59a3bc0 100644 --- a/src/openarmature/graph/fan_out.py +++ b/src/openarmature/graph/fan_out.py @@ -209,29 +209,20 @@ async def run_with_context( # ``context.fan_out_progress_state``; first-run constructs a # fresh one with all instances ``not_started``. # - # The key carries the ENCLOSING CONCURRENCY lineage, not just the - # namespace + node name, on BOTH axes that can put two live executions of - # the same fan-out node at the same namespace. + # The key carries the ENCLOSING CONCURRENCY lineage on BOTH axes, since + # either can put two live executions of the same fan-out node at the + # same namespace: nesting inside an outer instance repeats the namespace + # per instance, and nesting inside a parallel branch repeats it per + # branch, because branch names never enter the namespace. # - # A fan-out nested inside an outer fan-out instance has the same namespace - # for every outer instance. A fan-out nested inside a parallel BRANCH has - # the same namespace for every branch, because branch names do not enter - # the namespace at all. Either way, without the lineage the shared dict - # collides across concurrent enclosing contexts and the second execution - # finds the first's instances already ``completed`` and rolls its results - # forward -- returning results it never computed. + # Without both axes the shared dict collides and the second execution + # finds the first's instances already ``completed``, rolling forward + # results it never computed. Nothing surfaces it: a branch emits no + # inner node events, and a single ``await`` in the leaf body hides the + # interleaving, so a regression test here must not yield in the leaf. # - # The branch axis was missing, and its absence was not visible: a branch - # descent contributes only None entries to fan_out_index_chain, which the - # fan-out comprehension filters out, so both branches built an identical - # key. The branch also emitted no inner node events, so no observer could - # see the work had not happened. It is schedule-dependent -- a single - # ``await`` in the leaf body hides it -- which is why a regression test - # here must not yield in the leaf. - # - # Top-level fan-outs contribute neither axis, so both lineages are empty - # there, matching the resume restore (which defaults both to empty) and - # leaving top-level resume unaffected. + # A top-level fan-out contributes neither axis, so both lineages are + # empty and match what the resume restore defaults to. key = fan_out_progress_key( context.namespace_prefix, self.name, @@ -248,41 +239,27 @@ async def run_with_context( ) context.fan_out_progress_state[key] = exec_state elif exec_state.instance_count != instance_count: - # Per spec §10.11 + §10.10 (proposal 0029): a saved - # ``instance_count`` that differs from the resumed run's - # resolved count MUST raise ``checkpoint_record_invalid`` - # before any fan-out instance work runs on this path. The - # pre-0029 pad/truncate behavior would silently drop - # ``completed`` contributions on shrink (breaking §10.11.1's - # exactly-once guarantee) and dispatch unsaved work on grow - # (violating §10.5's idempotency framing). The strict raise - # surfaces the divergence to the user; they cohere inputs - # or restart cleanly. - # Local import to avoid an engine ↔ checkpoint package cycle - # at module load (mirrors the existing - # ``CheckpointSaveFailed`` imports below in this file). + # §10.11 + §10.10: a saved ``instance_count`` differing from the + # resumed run's resolved count MUST raise + # ``checkpoint_record_invalid`` before any instance work runs. + # Padding or truncating instead would drop ``completed`` + # contributions on shrink, breaking §10.11.1 exactly-once, and + # dispatch unsaved work on grow. + # + # Local import to avoid an engine/checkpoint cycle at module load. from openarmature.checkpoint.errors import CheckpointRecordInvalid # noqa: PLC0415 - # ``context.resume_invocation`` identifies the SAVED record - # being validated (per spec §10.4 step 3); ``context.invocation_id`` - # is freshly minted for the resumed run (step 4). The fresh- - # run fallback is defensive only — the count-drift path can - # only fire on resume since fan_out_progress_state is empty - # on a fresh first run. - # - # That claim was FALSE for sibling parallel branches until the - # branch axis joined the key: two branches whose subgraphs held a - # same-named fan-out over different item counts collided on a FRESH - # run, and the second branch raised here with a message about a - # checkpoint record that did not exist. Adding the branch axis is - # what makes the sentence above true. + # ``resume_invocation`` names the SAVED record being validated + # (§10.4 step 3); ``invocation_id`` is minted fresh for the resumed + # run (step 4). The fresh-run fallback is defensive: the count-drift + # path fires only on resume, since fan_out_progress_state is empty + # on a first run. # - # The other side of that: a branch-nested fan-out's restored entry - # now carries an empty branch lineage and can never be found by a - # re-entering execution, so this MUST raise is unreachable for that - # shape and a genuine count drift there resumes by re-running rather - # than by raising. That is the §10.11 no-mis-skip floor, and closing - # it needs a branch lineage on the record itself. + # Unreachable for a branch-nested fan-out: its restored entry + # carries an empty branch lineage that no re-entering execution can + # match, so a genuine count drift there resumes by re-running + # instead of raising. That is the §10.11 no-mis-skip floor, and + # closing it needs a branch lineage on the record itself. raise CheckpointRecordInvalid( context.resume_invocation or context.invocation_id, f"fan_out {self.name!r} at namespace {context.namespace_prefix!r}: " diff --git a/src/openarmature/graph/observer.py b/src/openarmature/graph/observer.py index 15914a4..0eca06f 100644 --- a/src/openarmature/graph/observer.py +++ b/src/openarmature/graph/observer.py @@ -528,18 +528,13 @@ class _InvocationContext: # ---------------------------------------------------------------- # Checkpointing fields (spec pipeline-utilities §10) # - # ``invocation_id`` and ``correlation_id`` are minted once at the - # outermost ``invoke`` call (or restored from a saved record on - # resume) and propagated unchanged through every descent. The - # checkpointer reference is set when a backend is registered; it - # is intentionally **None inside fan-out instances** so per-instance - # internal saves are gated off (§10.7 atomic-restart). The mutable - # ``completed_positions`` list is shared across descents so the - # save call sites can append the just-completed position before - # the engine's next step. ``resume_skip_set`` is a frozen set of - # namespace tuples whose corresponding nodes have already - # completed in a prior run and MUST be skipped on this resumed - # invocation. + # ``invocation_id`` and ``correlation_id`` are minted once at the outermost + # ``invoke`` (or restored on resume) and propagate unchanged. The + # checkpointer is deliberately None inside fan-out instances, which gates + # off per-instance internal saves (§10.7 atomic restart). + # ``completed_positions`` is shared across descents so save sites can append + # before the next step. ``resume_skip_set`` holds namespaces already + # completed in a prior run, which MUST be skipped. # ---------------------------------------------------------------- invocation_id: str = "" correlation_id: str = "" @@ -585,20 +580,15 @@ class _InvocationContext: # counters because subgraphs share the parent's queue + worker, so # the parent context's counts naturally cover subgraph events. drain_counters: _DrainCounters = field(default_factory=_DrainCounters) - # Per spec §10.2 (proposal 0028): the canonical source for - # ``CheckpointRecord.schema_version``. Set once at the outermost - # ``invoke`` to the compiled graph's declared state class - # (``CompiledGraph.state_cls``); propagated unchanged through every - # descent (subgraphs, fan-out instances, parallel branches). All - # save sites within an invocation MUST read ``schema_version`` from - # this class — NOT from ``type(state)`` at save time — so the - # value is consistent across the outer dispatch save, fan-out - # instance internal saves, and the fan-out node's own completion - # save. The distinction matters only when a user passes a State - # subclass that shadows ``schema_version``; the declared class is - # the only consistent choice for §10.12 migration lookups. - # ``Any`` rather than ``type[State]`` to avoid an import cycle - # between graph and observer; callers narrow at the read site. + # §10.2: the canonical source for ``CheckpointRecord.schema_version``, set + # once at the outermost ``invoke`` and propagated unchanged. Every save site + # MUST read it from here rather than from ``type(state)``, or the value + # diverges across the outer, per-instance and completion saves when a caller + # passes a State subclass that shadows it. The declared class is the only + # consistent choice for §10.12 migration lookups. + # + # ``Any`` rather than ``type[State]`` to avoid a graph/observer import + # cycle; callers narrow at the read site. state_cls: Any = None # Per proposal 0043 (observability §8.4.1 trace.output sourcing): # shared mutable single-element box tracking the most recently @@ -610,22 +600,15 @@ class _InvocationContext: # descents so the inner-most node's name wins on failure (the # real culprit, not the wrapper). final_node_box: list[str] = field(default_factory=list[str]) - # Per proposal 0043 (observability §8.4.1 *Resume semantics* + - # "partial final state captured at the failure point" clause). - # Tracks the most recent successful step's post-merge state at THIS - # context level so the outermost ``invoke()`` can populate - # ``InvocationCompletedEvent.final_state`` on the failure path with - # the partial outer state, not the bare ``starting_state``. On the - # success path the box is unused — the engine's return value is the - # canonical ``final_state``. **Distinct from ``final_node_box``**: - # the latest-state box is per-level (each subgraph / fan-out - # instance / parallel-branches branch gets its own fresh box), - # because the OUTER Langfuse trace cares about the outer-graph's - # state type, and an inner state has a different type. The - # ``final_node_box`` shares by reference because the spec wants the - # innermost failing node's name (the real culprit); state has the - # opposite contract — the outermost level's state is what the - # outer trace.output hook receives. + # §8.4.1: the most recent successful step's post-merge state at THIS level, + # so a failing ``invoke()`` reports partial state rather than the bare + # ``starting_state``. Unused on the success path, where the return value is + # canonical. + # + # Per-level, unlike ``final_node_box`` which shares by reference. The two + # have opposite contracts: the spec wants the INNERMOST failing node's name, + # but the OUTERMOST level's state, since that is what the outer trace.output + # hook receives and an inner state has a different type. latest_state_box: list[Any] = field(default_factory=list[Any]) def full_observers(self) -> tuple[SubscribedObserver, ...]: diff --git a/src/openarmature/llm/providers/openai.py b/src/openarmature/llm/providers/openai.py index cfa3bf3..0fee562 100644 --- a/src/openarmature/llm/providers/openai.py +++ b/src/openarmature/llm/providers/openai.py @@ -136,21 +136,20 @@ # signal. Validate in ``__init__`` against this set instead. _VALID_READINESS_PROBES = frozenset({"models", "chat_completions", "both"}) -# §8.1 managed wire fields and their collision arms (proposals 0105 + 0108). The -# structural keys (model / messages / tools / tool_choice) are managed-internal -# (0105 §3.5): 0105 §3.5 gives them no "while producing it" qualifier (that is -# reserved for the conditionally-managed response_format / stream_options), so -# they are ALWAYS managed -- an extras key of a structural name is rejected even -# on a call that produced no such field. This closes the hole where an extras -# `tools` / `tool_choice` on a no-tools call would otherwise ride untouched and -# smuggle a raw, unvalidated tool array past validate_tools. The sampling scalars -# are declared-field realizations (0108); stop realizes the list-shaped -# stop_sequences so it MERGES; response_format is managed only on the -# structured-output path (0105 conditionally-managed). Those non-structural keys -# are managed only WHEN the mapping actually produced them, which the call site -# enforces by filtering to keys present in the built body -- so a sampling field -# the caller left None, or response_format on the free-form / §8.1.5.1 fallback -# path, is unmanaged and an extra of that name rides untouched. +# §8.1 managed wire fields and their collision arms (0105 + 0108). +# +# The structural keys (model / messages / tools / tool_choice) are ALWAYS +# managed: 0105 §3.5 gives them no "while producing it" qualifier, unlike the +# conditionally-managed response_format / stream_options. So an extras key of a +# structural name is rejected even where the call produced no such field, which +# is what stops an extras `tools` on a no-tools call smuggling an unvalidated +# tool array past validate_tools. +# +# The rest are managed only WHEN the mapping produced them, which the call site +# enforces by filtering to keys present in the built body. A sampling field left +# None, or response_format off the structured-output path, is unmanaged and an +# extra of that name rides untouched. `stop` realizes the list-shaped +# stop_sequences, so it MERGES rather than collides. _OPENAI_MANAGED_ARMS: dict[str, ManagedArm] = { "model": "reject", "messages": "reject", @@ -436,22 +435,14 @@ async def complete( ``RetryMiddleware``); an exception raised by it propagates out of the call. """ - # Spec observability §5.5 LLM provider span: when an - # observability backend is active in the current invocation, - # emit a typed LlmCompletionEvent (success) or LlmFailedEvent - # (failure) around the wire call so the backend can build a - # span / Generation observation. Queue-mediated dispatch - # preserves spec §6 serial event ordering across all event - # sources within an invocation. ``current_dispatch()`` returns - # ``None`` outside an openarmature invocation (direct provider - # use in scripts/tests), in which case the call proceeds - # without typed-event emission. + # §5.5: emit the typed completion / failure event around the wire call + # so a backend can build the span. Queue-mediated so §6 serial ordering + # holds across event sources. ``current_dispatch()`` is ``None`` outside + # an invocation (direct provider use), and the call then emits nothing. # - # ``call_id`` is minted once per ``complete()`` call. Per - # proposal 0058: a failed call gets its own ``call_id`` - # distinct from any retry-attempt sibling — the retry - # middleware re-enters ``complete()`` for each attempt, so a - # fresh mint per call automatically satisfies that contract. + # ``call_id`` is minted per ``complete()`` call, which gives a failed + # call an id distinct from any retry-attempt sibling (0058), since the + # retry middleware re-enters ``complete()`` per attempt. dispatch = current_dispatch() call_id = str(uuid.uuid4()) # Capture prompt context AT DISPATCH TIME (in the node task's @@ -1167,14 +1158,10 @@ def _build_request_body( # path the mapping produces no response_format, so it is UNMANAGED there # (0105 §3.5): a caller's extras response_format rides untouched via the # reconciliation below rather than being stripped. - # Per §8.1.1 (proposal 0025): map the spec-level `tool_choice` - # shape onto the OpenAI wire shape. ``None`` omits the field - # entirely so the OpenAI provider's own default applies — - # load-bearing for backward compat with pre-0025 callers. The - # string-literal modes pass through verbatim; the ``ForceTool`` - # record renames ``type: "tool"`` → ``type: "function"`` and - # nests the name under a ``function`` sub-object per OpenAI's - # request shape. + # §8.1.1: map the spec `tool_choice` onto the OpenAI wire shape. + # ``None`` omits the field so OpenAI's own default applies. String + # modes pass through; ``ForceTool`` renames ``type: "tool"`` to + # ``"function"`` and nests the name under a ``function`` object. if tool_choice is not None: if isinstance(tool_choice, ForceTool): body["tool_choice"] = { @@ -1641,32 +1628,18 @@ def _block_to_wire(block: ContentBlock) -> dict[str, Any]: return {"type": "image_url", "image_url": image_url} -# Spec 0047 §8 *Intra-impl wire-byte stability* canonicalizer. -# Recursively sorts dict keys at every nesting level; preserves list -# ordering (per Q5 ack on the proposal-0047 coord thread — array -# ORDER is caller-supplied and stays as-is; object KEYS inside -# arrays get sorted via the dict-recursion branch). Applied at every -# user-supplied-dict boundary in the wire body so equivalent OA -# inputs produce byte-identical wire output for APC hit reliability. +# 0047 §8 wire-byte stability canonicalizer. Sorts dict keys at every nesting +# level and preserves list ORDER, which is caller-supplied; object keys inside +# arrays still sort via the dict branch. Applied at every user-supplied-dict +# boundary so equivalent inputs produce byte-identical wire output, which is +# what makes prompt-cache hits reliable. # -# Recursion depth: bounded by the depth of the input dict, not by -# any internal accumulator. Python's default recursion limit (1000) -# is two orders of magnitude above realistic JSON Schema depths -# (typical schemas top out at 5-10 nesting levels — OpenAI's API -# rejects deeper ones at the wire layer before the cache prefix -# matters). We don't impose our own cap; if a caller hands us a -# 1000-deep nested dict, RecursionError surfaces immediately at -# canonicalization time rather than producing silently-broken wire -# bytes downstream. +# No depth cap of its own: a pathologically nested dict raises RecursionError +# here rather than producing broken wire bytes downstream. # -# Byte-stability requires Python's dict insertion-order preservation -# guarantee (PEP 468, 3.7+) AND httpx serializing the body via the -# stdlib ``json.dumps`` default (which respects dict iteration -# order). Both are stable contracts on the supported Python versions -# + httpx 0.27+. If a future httpx release internalizes ordering -# (e.g., switches to alphabetical key emission), the canonicalizer -# becomes redundant but tests would continue to pass; if it -# randomizes ordering, the wire-byte tests in +# Byte-stability rests on dict insertion order (PEP 468) and on httpx +# serializing through the stdlib ``json.dumps``, which respects it. If httpx +# ever randomized key order the wire-byte tests in # ``tests/unit/test_llm_provider.py`` would fail loudly. def _canonicalize_dict_keys(value: Any) -> Any: if isinstance(value, dict): diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 814bda1..a3cf216 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -116,18 +116,11 @@ def _read_implementation_version() -> str: return __version__ -# In-flight Span observation handle, keyed by the scalars -# (namespace, attempt_index, fan_out_index, branch_name) AND the enclosing -# fan-out / branch lineage chains (proposal 0084). ``branch_name`` discriminates -# concurrent same-named inner nodes across sibling parallel-branches branches; -# the chains discriminate an inner node under two concurrent OUTER fan-out -# instances (which share the innermost scalar fan_out_index), so the nested -# exact-match Generation-parent lookup finds its own calling node's observation -# rather than a sibling's. The scalars are retained (as in the OTel observer): -# a callable parallel-branch carries branch_name on the event but does not -# extend branch_name_chain, so key[3] keeps it distinct from its own -# parallel-branches node. Mirrors the OTel observer's ``_StackKey`` shape but -# holds a Langfuse handle instead of an OTel Span. +# Keyed by the scalars AND the enclosing lineage chains (0084). Both are +# needed: the chains separate an inner node under two concurrent outer +# instances, which share the innermost scalar; the scalars keep a callable +# parallel-branch distinct from its own node, since it never extends the chain. +# Mirrors the OTel observer's `_StackKey`. _StackKey = tuple[ tuple[str, ...], int, int | None, str | None, tuple[int | None, ...], tuple[str | None, ...] ] @@ -227,9 +220,8 @@ def _subgraph_identity_at(event: NodeEvent, depth: int) -> str: value; the empty-string path keeps direct callers conformant but failing those fixtures. """ - # Spec observability §5.3 (coord thread - # clarify-subgraph-name-semantics): empty-string fallback is - # conformant for callers that don't track a subgraph identity. + # §5.3: the empty-string fallback is conformant for callers that do not + # track a subgraph identity. idx = depth - 1 if 0 <= idx < len(event.subgraph_identities): identity = event.subgraph_identities[idx] @@ -251,15 +243,10 @@ class _InvState: open_observations: dict[_StackKey, _OpenObservation] = field( default_factory=dict[_StackKey, _OpenObservation] ) - # Synthetic subgraph dispatch Span observations, keyed by namespace - # prefix. Per spec §8.3 each subgraph wrapper produces a Span - # observation in its parent's Trace; descendant node observations - # parent under it. For a detached subgraph, this dictionary holds - # the dispatch Span observation that lives in the DETACHED Trace - # (so descendants in that subtree parent under it via the detached - # Trace's observation tree); the main Trace carries a separate - # link observation surfacing metadata.detached_child_trace_ids - # that's opened and closed in one shot, not tracked here. + # Synthetic subgraph dispatch observations (§8.3), keyed by namespace + # prefix. For a detached subgraph this holds the one in the DETACHED Trace; + # the main Trace's link observation opens and closes in one shot and is not + # tracked here. subgraph_observations: dict[tuple[str, ...], _OpenObservation] = field( default_factory=dict[tuple[str, ...], _OpenObservation] ) @@ -312,13 +299,9 @@ class _InvState: parallel_branches_branch_names: dict[tuple[str, ...], frozenset[str]] = field( default_factory=dict[tuple[str, ...], frozenset[str]] ) - # Side-cache: accumulator for `metadata.detached_child_trace_ids` - # on dispatch observations that spawn detached children. Keyed by - # the dispatch observation's prefix (the fan-out node's namespace, - # or the detached-subgraph parent's prefix). Each new detached - # child append-then-snapshot lets us preserve §8.5's string-array - # shape across multiple instances without re-reading metadata - # from the client (the Protocol doesn't expose a read accessor). + # Accumulator for `detached_child_trace_ids`, keyed by the dispatch + # observation's prefix. Held here because the client Protocol exposes no + # read accessor, so §8.5's array cannot be rebuilt from the client. detached_child_trace_ids: dict[tuple[str, ...], list[str]] = field( default_factory=dict[tuple[str, ...], list[str]] ) @@ -469,14 +452,10 @@ def _apply_isolation_policy(self) -> None: self.trace_input_from_state = None self.trace_output_from_state = None elif status == ISOLATION_LEAKED: - # Reachable only when no construction-time channel is live, so nothing - # is refused and nothing is currently leaking. Reported because the - # binding is a latent problem, and the two ways of enabling a channel - # fail closed differently: re-opening a knob on THIS observer is caught - # at emission by _isolation_blocks_payload() and the payload is - # withheld, while constructing a NEW observer over the same client with - # a channel live raises in __post_init__. Deliberately silent about the - # error message, which disable_provider_payload governs, not this status. + # Reached only when no channel is live, so nothing leaks yet. The + # binding is still latent: re-opening a knob here is caught at + # emission, while a new observer over the same client raises at + # construction. _logger.info( "OA's Langfuse client is bound to a TracerProvider it did not isolate; " "no payload channel is enabled, so nothing is being exported to it; " @@ -562,32 +541,16 @@ def from_credentials( # observer built by handing a from_credentials adapter to the constructor. return cls(client=client, **observer_kwargs) - # An emitted error_message IS capped, through `_capped_error_message` below. - # - # It was written verbatim until proposal 0119 (spec v0.116.0). 0118 had - # classified the field for GATING without saying it was subject to §5.5.5 - # truncation, so a provider returning a very large exception string rendered - # it in full. That reading was raised for the batched spec review rather than - # changed unilaterally, and 0119 answered it: §5.5.5 now governs every - # payload-classified VALUE, not only values written as span attributes, and - # §8.7 gives this one a direct-application arm because it has no span - # attribute to inherit a cap from. - # - # Do not re-derive the old behaviour from 0118 alone; 0119 is the governing - # text and it is later. + # An emitted error_message is capped, via `_capped_error_message` below: + # §5.5.5 governs every payload-classified value, not only span attributes + # (0119). def _emits_harvested_error_message(self) -> bool: - # A failed observation's error_message is harvested exception text, so the - # provider-payload flag governs it (0118) for every failure category and on - # every provider observation -- the category does not tell you what the - # string contains, and a provider 4xx routinely quotes the request or the - # flagged prompt. + # error_message is harvested exception text, so the provider-payload + # flag governs it (0118): a provider 4xx routinely quotes the request or + # the flagged prompt, and the error category does not tell you which. # - # There is no separate isolation check: once the flag covers the field the - # §6 arms already decide every configuration. Flag on, nothing is emitted - # to suppress; flag off on an isolated provider, it emits legitimately; - # flag off on a detected shared provider, construction raised; flag off - # where isolation could not be established, suppress-all set the flag; flag - # off with the caller opted in, it emits as an acknowledged leak. + # No separate isolation check is needed; once the flag covers the field + # the §6 arms already decide every configuration. return self._emits_provider_payload() async def __call__( @@ -681,15 +644,10 @@ def _open_started_observation(self, event: NodeEvent) -> None: self._open_trace(invocation_id, correlation_id, event) inv_state = self._inv_states[invocation_id] - # Cache the fan-out node's parent_node_name from its own - # started event so synthetic per-instance dispatch observations - # can attach metadata.fan_out_parent_node_name (the inner - # events from inside the fan-out don't carry fan_out_config - # themselves; this cache bridges). fan_out_config is set only on - # the NODE's own events, so it alone identifies them -- NOT - # ``fan_out_index is None``, which would miss a fan-out node nested - # inside an outer fan-out instance (its own event carries the OUTER - # instance index), leaving the inner dispatch unsynthesized. + # Inner events carry no fan_out_config, so cache it here for the + # synthetic per-instance dispatches. Identify the node's own events by + # fan_out_config, NOT by `fan_out_index is None`: a nested fan-out node + # carries the OUTER instance index on its own event. if event.fan_out_config is not None: inv_state.fan_out_parent_node_name[event.namespace] = event.fan_out_config.parent_node_name @@ -812,29 +770,12 @@ def _handle_completed(self, event: NodeEvent) -> None: # ------------------------------------------------------------------ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> None: - # Spec proposal 0040 §3.4 MUST: open observations whose lineage - # ancestor-or-equals the augmenting context get the entries - # applied in place via the Langfuse handle's - # ``update(metadata=...)`` method. Sibling instances / branches - # and ancestors above the containment are skipped (same scoping - # rule as the OTel mapping — see - # ``OTelObserver._handle_metadata_augmentation`` for the algebra). - # - # For an outermost-serial augmenter (FI=None, BN=None), the - # invocation's Trace itself is updated via - # ``client.update_trace`` so the augmented keys land on - # ``trace.metadata.`` for §8.4-style top-level filtering. - # Inside a fan-out instance / parallel-branches branch the - # Trace is OUT of scope (it's shared with siblings); only the - # innermost containment + the augmenter's own subtree update. - # - # Per-instance / per-branch isolation: - # ``set_invocation_metadata`` runs in the calling node's task - # whose Context already carries the per-async-context COW - # mapping (proposal 0034 §3.4). The augmentation event's - # ``entries`` are that delta only — applying them to matching - # open observations preserves the per-async-context isolation - # 029 / 030 encode. + # 0040 §3.4: update observations whose lineage ancestor-or-equals the + # augmenting context; skip siblings and ancestors above it. An + # outermost-serial augmenter also updates the Trace, which is otherwise + # shared with siblings and out of scope. `entries` is the + # per-async-context delta only (0034 §3.4), which is what isolates + # siblings. from openarmature.observability.correlation import current_invocation_id invocation_id = current_invocation_id() @@ -908,16 +849,11 @@ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> Non # ------------------------------------------------------------------ def _handle_failure_isolated(self, event: FailureIsolatedEvent) -> None: - # Render the FailureIsolationMiddleware catch as a marker observation. - # The wrapped node's observation is typically already closed by delivery - # time (the node-body raise fires the node's completed event before the - # middleware recovers), so this usually takes the orphan fallback: the - # marker parents under the nearest enclosing wrapper on its lineage - # (proposal 0084 §5.5), matching the OTel observer -- which routes - # _handle_failure_isolated through _resolve_llm_parent -- for - # cross-observer parity (the enclosing fan-out instance / branch / - # subgraph observation, else None -> the Trace itself). The wrapped - # node's name rides on ``metadata.failure_isolation_node`` regardless. + # The wrapped node's observation is usually closed by delivery time + # (its completed event fires before the middleware recovers), so this + # normally takes the §5.5 orphan fallback and parents under the nearest + # enclosing wrapper. The OTel observer routes the same event through + # `_resolve_llm_parent` to keep the two in agreement. from openarmature.observability.correlation import ( current_correlation_id, current_invocation_id, @@ -983,19 +919,9 @@ def _handle_failure_isolated(self, event: FailureIsolatedEvent) -> None: # ------------------------------------------------------------------ def _handle_invocation_started(self, event: InvocationStartedEvent) -> None: - # Spec proposal 0043 §8.4.1 *Trace input/output sourcing*. - # Lazy-open the Trace if this is the first signal for the - # invocation_id (no node event has fired yet), then resolve - # ``trace.input`` via the three-lever decision tree: - # 1. Hook supplied AND returns non-None → hook value. - # 2. ``disable_state_payload`` is False → raw initial_state - # serialized (subject to payload_byte_cap truncation). - # 3. Otherwise → minimal stub: - # {entry_node, correlation_id}. - # The stub carries no application payload — both fields are - # already in ``trace.metadata``; surfacing them on - # ``trace.input`` makes the Langfuse Traces list view - # scannable without revealing state shape. + # 0043 §8.4.1 trace input/output sourcing. The stub arm duplicates two + # fields already in trace.metadata so the Traces list view is scannable + # without exposing state shape. if event.invocation_id not in self._inv_states: self._open_trace_lazy(event.invocation_id, event.correlation_id, event.entry_node) input_value = self._resolve_trace_input(event) @@ -1065,21 +991,10 @@ def _resolve_trace_output(self, event: InvocationCompletedEvent) -> Any: @staticmethod def _state_to_jsonable(state: Any) -> Any: - # Best-effort conversion of a State instance to a JSON-able - # shape. Pydantic models expose ``model_dump`` directly; other - # objects fall through to a str representation. The serialized - # form is what ends up on the Langfuse Trace's - # ``input`` / ``output`` field. - # - # ``mode="json"`` (rather than the default Python mode) coerces - # non-JSON-native types — ``datetime``, ``UUID``, ``Decimal``, - # etc. — into JSON-compatible strings BEFORE the dict reaches - # the downstream ``json.dumps`` truncation path. Without it the - # truncation path raises ``TypeError`` and the observer's - # ``__call__`` raise is swallowed by the engine's warnings-only - # observer-isolation contract, leaving ``trace.input`` / - # ``trace.output`` silently blank on states containing those - # types. + # `mode="json"` coerces datetime / UUID / Decimal before the value + # reaches `json.dumps`. Without it that raises TypeError, the engine + # swallows an observer raise as a warning, and trace.input / output go + # silently blank for any state carrying those types. dumper = getattr(state, "model_dump", None) if callable(dumper): try: @@ -1089,15 +1004,9 @@ def _state_to_jsonable(state: Any) -> Any: return str(state) def _client_trace(self, *, id: str, name: str | None, metadata: dict[str, Any]) -> None: - # Proposal 0064 §8.4.1: every Trace open routes through here so the - # sessionId / userId promotions apply uniformly across the main, - # lazy, and detached trace-open sites. - # - trace.userId: promoted from the recognized ``userId`` caller - # key (already merged into ``metadata`` by _apply_caller_metadata). - # - trace.sessionId: sourced from openarmature.session_id (sessions - # capability, observability §5.6 / proposal 0020). python has no - # session_id source until 0020 lands, so it is unset (None) today; - # this is the single hook 0020 wires the source into. + # Every Trace open routes through here so the §8.4.1 userId / sessionId + # promotions apply at the main, lazy and detached sites alike. + # sessionId has no source until 0020 lands; this is the hook it wires. self.client.trace( id=id, name=name, @@ -1134,15 +1043,9 @@ def _open_trace_lazy( self._inv_states[invocation_id] = _InvState(trace_id=invocation_id) def _open_trace(self, invocation_id: str, correlation_id: str | None, event: NodeEvent) -> None: - # ``entry_node`` and the trace name MUST identify the outer-graph - # entry, not whichever node fired first. Subgraph wrappers do not - # emit their own events — when the outer entry is a SubgraphNode - # the first event the observer sees comes from inside the - # subgraph (with ``event.namespace = (wrapper, inner)`` and - # ``event.node_name = inner``). Using ``event.namespace[0]`` - # walks back to the outermost prefix component, which IS the - # outer entry by construction (the graph engine fires inner - # events under the wrapper's namespace). + # `namespace[0]`, not `node_name`: a subgraph wrapper emits no event of + # its own, so when the outer entry is one the first event seen comes + # from inside it and would name the inner node. entry_node = event.namespace[0] if event.namespace else event.node_name metadata: dict[str, Any] = { "entry_node": entry_node, @@ -1177,15 +1080,9 @@ def _key_for(self, event: NodeEvent) -> _StackKey: ) def _resolve_parent_observation_id(self, inv_state: _InvState, event: NodeEvent) -> str | None: - # Parent precedence (innermost wins): - # 1. Per-instance fan-out / per-branch dispatch observation on the - # event's lineage (the enclosing wrapper) — resolved by the shared - # helper below. - # 2. Subgraph dispatch observation at any matching ancestor prefix - # (also the shared helper). - # 3. Leaf node observation at any matching ancestor prefix, walked - # longest-first. - # 4. None — the Trace itself becomes the implicit parent. + # Innermost wins: per-instance dispatch on the event's lineage, then + # subgraph dispatch at an ancestor prefix, then a leaf node observation + # walked longest-first, else the Trace. wrapper_id = self._resolve_enclosing_wrapper_observation_id( inv_state, namespace=event.namespace, @@ -1290,13 +1187,8 @@ def _sync_subgraph_observations( correlation_id: str | None, event: NodeEvent, ) -> None: - # Open synthetic subgraph dispatch / fan-out per-instance - # dispatch observations for any ancestor prefix of this - # event's namespace that doesn't have one yet. Also closes - # subgraph dispatch observations whose subtree we've left. - # - # Called BEFORE opening the leaf observation, so descendants - # find the right parent via _resolve_parent_observation_id. + # Must run BEFORE the leaf observation opens, so descendants resolve + # the right parent. namespace = event.namespace # 1. Close subgraph dispatch observations whose prefix is no # longer an ancestor of the current namespace. @@ -1385,14 +1277,10 @@ def _sync_subgraph_observations( inv_state, correlation_id, prefix, event ) continue - # A parallel-branches or fan-out NODE prefix already has its own - # leaf observation (from the NODE's own started event), unlike a - # transparent subgraph wrapper. Don't synthesize a duplicate - # subgraph wrapper observation over it; inner branch / instance - # events parent under the NODE observation via the - # _resolve_parent_observation_id leaf fallback. Mirrors the OTel - # observer's same guard (it skips the synthetic subgraph span at a - # pb / fan-out NODE depth for the same reason). + # A fan-out or parallel-branches NODE already has its own leaf + # observation, unlike a transparent subgraph wrapper, so + # synthesizing one here would duplicate it. The OTel observer + # carries the same guard. if ( prefix in inv_state.parallel_branches_parent_node_name or prefix in inv_state.fan_out_parent_node_name @@ -1552,45 +1440,16 @@ def _open_detached_subgraph_trace( prefix: tuple[str, ...], event: NodeEvent, ) -> None: - # Mint a fresh Trace for the detached subtree. The main Trace's - # dispatch observation surfaces the link via - # metadata.detached_child_trace_ids; the detached Trace gets - # its own dispatch observation that descendants parent under. - # - # Asymmetry note vs. _open_detached_fan_out_instance_trace: - # subgraphs are namespace-prefix-only constructs with no - # per-subgraph node event of their own. The observer never - # opens a leaf Span observation for the subgraph itself, only - # synthesized dispatch observations. To carry the cross-Trace - # link in the main Trace's shape, this helper opens an extra - # "link" Span observation in the main Trace — a small - # observation whose subtree is empty but whose - # detached_child_trace_ids metadata points at the new Trace. - # Dashboard users see two observations named ``prefix[-1]``: - # one in the main Trace (link with link metadata, no subtree) - # and one in the detached Trace (the real dispatch with the - # subgraph subtree under it). - # - # Detached fan-out instances, by contrast, already have a - # parent observation in the main Trace (the fan-out node's - # leaf observation opened on its own started event). The - # link metadata accumulates on that pre-existing observation - # instead of synthesizing a separate link observation. + # A subgraph has no node event of its own, so the main Trace holds no + # observation to hang the cross-Trace link on and this opens an empty + # one for it. A detached fan-out instance needs no equivalent: the + # fan-out node's own observation already exists to carry the link. detached_trace_id = str(uuid.uuid4()) - # Open the link observation in the main Trace and update its - # metadata immediately — the array-form preserves §8.5's - # "string array, one entry per detached child" shape so - # later detached siblings under the same parent can append. + # §8.5 array form so later detached siblings can append; §8.4.2 + # `detached: True` marks the dispatching side. # - # `detached: True` per §8.4.2 (proposal 0042) — the - # parent-side dispatching observation marks itself when it - # fires a detached child. - # - # Note: `subgraph_name` is intentionally NOT on this link - # observation. Per §5.3 + §8.5, in detached mode the wrapper - # role migrates to the detached trace's dispatch observation; - # the main trace's link observation IS the SubgraphNode span - # (no wrapper role) and so does not carry `subgraph_name`. + # No `subgraph_name` here: in detached mode the wrapper role migrates + # to the detached trace's dispatch observation (§5.3, §8.5). link_metadata: dict[str, Any] = { "detached_child_trace_ids": [detached_trace_id], "detached": True, @@ -1624,31 +1483,14 @@ def _open_detached_subgraph_trace( detached_metadata["correlation_id"] = correlation_id _apply_caller_metadata(detached_metadata, event.caller_invocation_metadata) identity = _subgraph_identity_at(event, len(prefix)) - # The detached trace's wrapper observation IS the migrated - # SubgraphNode wrapper. Per the resolution in coord thread - # ``clarify-subgraph-name-semantics`` and fixture 033's - # expected shape, the observation name uses the compiled- - # subgraph identity (e.g., ``"long_running_workflow"``); its - # ``metadata.subgraph_name`` carries the same identity. - # - # When the identity is empty (BC path — ``SubgraphNode`` - # constructed without ``subgraph_identity``), the two - # diverge intentionally: the observation NAME falls back to - # the wrapper node name (an empty observation name is worse - # UX than a wrapper-named one), but ``metadata.subgraph_name`` - # stays empty per §5.3's "empty string when no identity is - # tracked" contract. Filtering on - # ``metadata.subgraph_name == "X"`` then matches only - # wrappers explicitly registered with - # ``subgraph_identity = "X"``, not every wrapper that - # happens to be named ``X``. + # Name and `subgraph_name` both take the compiled-subgraph identity + # (fixture 033). With no identity the name falls back to the wrapper + # node while `subgraph_name` stays empty per §5.3, so filtering on it + # matches only wrappers actually registered with that identity. wrapper_obs_name = identity or prefix[-1] self._client_trace(id=detached_trace_id, name=wrapper_obs_name, metadata=detached_metadata) - # §8.4.2 (proposal 0042): `detached: true` lives on the - # PARENT-side dispatching observation (the link observation - # above), not on the dispatch observation IN the detached - # trace. The detached-side observation is the migrated - # SubgraphNode wrapper and carries `subgraph_name` only. + # §8.4.2: `detached: true` belongs on the parent-side link observation + # above, not here. This side carries `subgraph_name` only. dispatch_metadata: dict[str, Any] = { "subgraph_name": identity, } @@ -1661,9 +1503,8 @@ def _open_detached_subgraph_trace( metadata=dispatch_metadata, parent_observation_id=None, ) - # Per proposal 0045: detached subgraph wrapper sits in its own - # trace; chain still mirrors the parent-trace path so the - # augmentation lookup is consistent with non-detached. + # 0045: the chain mirrors the parent-trace path even though this sits + # in its own trace, so the augmentation lookup stays uniform. chain_len = len(prefix) inv_state.subgraph_observations[prefix] = _OpenObservation( handle=handle, @@ -1679,18 +1520,9 @@ def _open_detached_fan_out_instance_trace( prefix: tuple[str, ...], event: NodeEvent, ) -> None: - # Mint a fresh Trace per instance. The fan-out node's own - # Span observation in the parent Trace accumulates the - # detached_child_trace_ids array (one entry per instance); - # each detached Trace gets its own per-instance dispatch - # observation that inner-node observations parent under. - # - # See _open_detached_subgraph_trace's docstring for why the - # detached-fan-out path doesn't synthesize a separate "link" - # observation in the main Trace: the fan-out node already - # has a leaf observation there (opened on its started event), - # so the link metadata accumulates on that existing - # observation rather than on a parallel link observation. + # One Trace per instance. No separate link observation is synthesized + # here, unlike the subgraph path: the fan-out node already has a leaf + # observation in the main Trace to accumulate the link metadata on. detached_trace_id = str(uuid.uuid4()) # Accumulate the per-fan-out link-ids list via the side cache # so each new instance appends to the array on the fan-out @@ -1783,18 +1615,12 @@ def _close_parallel_branches_branch_dispatch_observation( def _find_node_observation( self, inv_state: _InvState, prefix: tuple[str, ...], event: NodeEvent ) -> _OpenObservation | None: - # Find a NODE's own open leaf observation at ``prefix`` (the fan-out or - # parallel-branches NODE, whose per-instance / per-branch dispatches - # parent under it). Match the ENCLOSING lineage, not just the namespace: - # when the NODE is itself nested inside an outer fan-out instance / - # branch, several instances of the same NODE namespace are open at once - # under concurrency, so a namespace-only scan would bind the wrong one. - # Disambiguate by the full enclosing chain (proposal 0084): the NODE - # observation sits on the event's ancestor path iff its stored lineage - # chain is a prefix of the event's -- the same lineage-boundary rule the - # augmentation scoping uses (``_observation_chain_on_path``). Matching - # the innermost scalar alone is ambiguous at >=3 levels, where an - # intermediate instance index repeats across concurrent outer instances. + # Matched on the enclosing lineage, not the namespace alone: when this + # node is itself nested, several instances of the same namespace are + # open concurrently and a namespace-only scan binds the wrong one. The + # node is on the event's ancestor path iff its chain is a prefix of the + # event's (0084). The innermost scalar alone is ambiguous at >=3 levels, + # where an intermediate index repeats across outer instances. for key, observation in inv_state.open_observations.items(): if key[0] == prefix and _observation_chain_on_path( observation, event.fan_out_index_chain, event.branch_name_chain @@ -1910,14 +1736,9 @@ def _observation_metadata(self, event: NodeEvent, correlation_id: str | None) -> # Generation observation lifecycle (LLM provider events) # ------------------------------------------------------------------ - # v0.13.0 (proposals 0049 + 0057 + 0058): both Generation - # observation lifecycles are driven by typed events — success path - # from LlmCompletionEvent, failure path from LlmFailedEvent. Both - # handlers open + close in one shot at typed-event arrival, with - # start_time back-dated by latency_ms so duration reflects the - # adapter-boundary measurement rather than dispatcher queue delay. - # The provider dropped sentinel-namespace NodeEvent emission for - # LLM events entirely in this release. + # Both Generation lifecycles open and close in one shot on the typed event, + # with start_time back-dated by latency_ms so the duration reflects the + # adapter boundary rather than dispatcher queue delay. def _handle_typed_llm_completion(self, event: LlmCompletionEvent) -> None: """Open + close the Generation observation from the typed LlmCompletionEvent (success path).""" @@ -2127,19 +1948,8 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: calling_fan_out_index_chain=event.fan_out_index_chain, calling_branch_name_chain=event.branch_name_chain, ) - # §8.4.6 metadata: tool name always, tool_call_id when present. - # Caller metadata FIRST, same ordering as every other handler, so an - # OA-emitted key wins a collision. It goes on at all because §8.4.2 maps - # the caller set to `observation.metadata.` on EVERY Observation: - # the table scopes its other rows explicitly ("fan-out node Span - # observation only"), so the unscoped wording is deliberate. - # - # This handler carried NO caller metadata until now, which made the Tool - # observation the only one of the four provider observations to drop it, - # and left the two observers disagreeing about the same event: the OTel - # `_handle_tool_call` has applied it all along. The LLM handlers get it - # via `_typed_event_metadata`, and embedding / rerank apply it directly, - # so this was the one path with neither. + # §8.4.6 metadata. Caller set first, as in every handler, so an + # OA-emitted key wins a collision; §8.4.2 puts it on every observation. metadata: dict[str, Any] = {} _apply_caller_metadata(metadata, event.caller_invocation_metadata) metadata["openarmature_tool_name"] = event.tool_name @@ -2169,16 +1979,10 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: # is withheld statusMessage stays null rather than taking the message # instead, which would smuggle the harvested string out. if self._emits_harvested_error_message(): - # BOTH surfaces are capped. §5.5.5 as 0119 restates it governs - # every payload-classified VALUE, not every payload-classified - # FIELD, and a Tool failure renders the harvested string twice: - # once in metadata and once as the status message. Capping one - # and not the other would leave the uncapped copy carrying the - # whole exception, which is the outcome the cap exists to stop. - # - # The other `status_message` writes in this file take - # `error_category`, a classification token rather than harvested - # content, and are correctly uncapped. + # Both surfaces capped: §5.5.5 governs the payload-classified + # VALUE, and a Tool failure renders it twice. The other + # `status_message` writes take `error_category`, a + # classification token, and stay uncapped. capped = self._capped_error_message(event.error_message) metadata["error_message"] = capped status_message = capped @@ -2444,15 +2248,10 @@ def _resolve_llm_parent_observation_id( calling_fan_out_index_chain: tuple[int | None, ...], calling_branch_name_chain: tuple[str | None, ...], ) -> str | None: - # Calling-node identity precedence: - # 1. Exact-match leaf node observation at the calling key (the - # lineage-disambiguated calling node). - # 2. Orphan fallback (proposal 0084 §5.5 "Lineage-resolved parent"): - # the calling node's observation is not open (a middleware / wrapper - # call), so the Generation parents under the nearest enclosing - # wrapper observation per §4.3, resolved via the lineage chain to - # the correct inner instance -- the same ancestor walk the node - # parent uses, not the old top-level-scalar shortcut. None -> Trace. + # Exact-match the calling node's observation first. If it is not open + # (a middleware or wrapper call) take the §5.5 orphan fallback: parent + # under the nearest enclosing wrapper on the lineage chain, else the + # Trace. key: _StackKey = ( calling_namespace_prefix, calling_attempt_index, @@ -2523,34 +2322,20 @@ def _typed_event_metadata( budget["total_max_tokens"] = total_max if budget: metadata["token_budget"] = budget - # §8.4.3 (proposal 0109): a flat sibling token_budget_exceeded boolean - # gives the Langfuse FAILURE path parity with the OTel - # openarmature.llm.token_budget.exceeded attribute. Because this - # metadata is shared with the failed Generation, the flag SURVIVES the - # ERROR-precedence rule (the ERROR level / statusMessage still win as - # the primary signal). Same evaluation as the OTel span + §11 counter: - # true if any evaluated bound was crossed, false if all held, ABSENT - # when no bound is evaluable (a not-reported counter is not evaluated, - # per 0101), mirroring the attribute's suppression from a null counter. - # Two known limitations, both pending a spec follow-up: (a) the flag - # reflects the TERMINAL attempt's usage, so on a RETRIED call it can - # differ from the OTel per-attempt attribute -- the parity is of the - # FORMULA, not of per-attempt values, an intentional consequence of the - # terminal-only Generation; (b) the caller-metadata collision - # this key once shared with the token_budget bounds is CLOSED: 0119 - # reserved `token_budget` / `token_budget_exceeded`, so a caller - # invocation-metadata key of either name is now rejected at the - # invoke() boundary rather than shadowing the emitted value. + # §8.4.3 (0109): parity with the OTel token_budget.exceeded + # attribute. Absent, not false, when no bound is evaluable (0101). + # + # A Generation is terminal-only, so on a retried call this reflects + # the last attempt and can differ from the OTel per-attempt value. + # The parity is of the formula, not of per-attempt numbers. evaluations = _token_budget_evaluations(token_budget, event.usage) if evaluations: metadata["token_budget_exceeded"] = any(ev["actual"] > ev["max"] for ev in evaluations) if event.caller_invocation_metadata is not None: _apply_caller_metadata(metadata, event.caller_invocation_metadata) - # Response-side metadata. A completion always carries it; a - # structured_output_invalid failure also carries it (proposal 0082), - # since its wire response was intact, so finish_reason (the truncation - # signal) and the response identity render on the failed Generation too. - # Every other failure category received no response and renders none. + # A `structured_output_invalid` failure carries response-side metadata + # too (0082): its wire response was intact. Other failure categories + # received no response. if isinstance(event, LlmCompletionEvent): renders_response_side = True else: @@ -2628,14 +2413,9 @@ def _open_trace_for_typed_event( self._inv_states[invocation_id] = _InvState(trace_id=invocation_id) def _maybe_truncate_for_input(self, value: Any) -> Any: - # Returns the native value (list of message dicts) when it - # fits the cap, or the truncated marker-bearing string when - # it doesn't. The list-or-str union return is intentional per - # spec §8.7: the unparseable JSON IS the truncation signal — - # surfacing the marker preserves the diagnostic without - # faking a parse, and the Langfuse UI renders the string view - # rather than the structured-input view. Callers MUST NOT - # assume the return value is JSON-parseable. + # The list-or-str union is deliberate (§8.7): over the cap this returns + # the marker string, and the unparseable JSON IS the truncation signal. + # Callers cannot assume the result parses. serialized = self._serialize_payload_value(value) truncated = _truncate(serialized, self.payload_byte_cap) if truncated is None: @@ -2644,21 +2424,13 @@ def _maybe_truncate_for_input(self, value: Any) -> Any: def _capped_error_message(self, message: str) -> str: """A failed observation's ``error_message``, capped for emission.""" - # The cap is §8.7's. Applied DIRECTLY, under THIS observer's - # `payload_byte_cap`: the value has no span attribute to inherit a cap - # from, which is why §8.7 gives it - # a direct-application arm, and an observer MUST NOT take the OTel - # observer's cap for it. The two are configured independently and a - # deployment can set one and leave the other at its default. + # §8.7 direct-application arm: no span attribute carries this value, so + # it inherits no cap and takes this observer's own `payload_byte_cap`, + # never the OTel observer's. # - # Encoding is guarded because this runs on the FAILURE path, where the - # value is harvested exception text and is the likeliest string in the - # observer to carry a lone surrogate: a `FileNotFoundError` naming a - # surrogateescape-decoded path, or a provider body decoded the same way. - # `"\udcff".encode("utf-8")` raises, and an observer that raises is only - # `warnings.warn`-ed by the engine, so the whole failed observation would - # vanish and the failure path would take out its own reporting. Degrade - # the one field instead. + # The encode is guarded because harvested exception text can carry a + # lone surrogate, which raises. The engine only warns on an observer + # raise, so that would drop the observation reporting the failure. try: truncated = _truncate(message, self.payload_byte_cap) except UnicodeEncodeError: diff --git a/src/openarmature/observability/lineage.py b/src/openarmature/observability/lineage.py index 9054354..edbfc99 100644 --- a/src/openarmature/observability/lineage.py +++ b/src/openarmature/observability/lineage.py @@ -101,16 +101,13 @@ def branch_dispatch_key( """Lineage-aware identity key for a per-branch dispatch span at namespace ``prefix``. """ - # Shared by both observers. They each held an identical copy, and the same - # defect had to be fixed twice by hand three times running. + # Shared by both observers so the keying cannot drift between them. n = len(prefix) # Chains are normalized to the prefix DEPTH in both directions: truncated - # when longer, padded with None when shorter. Truncating alone was a - # defect: an orphan provider call issued from branch middleware carries - # empty chains and built `(prefix, (), (), branch)` where the span had been - # registered under `(prefix, (None,), (), branch)`. Those denote the same - # lineage, "no enclosing fan-out at that depth", and differed only as tuple - # keys, so the lookup missed and the orphan fell through to the root. + # when longer, padded with None when shorter. Padding is what makes the + # key canonical: `(prefix, (), (), branch)` and `(prefix, (None,), (), + # branch)` denote the same lineage, "no enclosing fan-out at that depth", + # and would otherwise differ as tuple keys. fan_out = tuple(fan_out_index_chain[:n]) + (None,) * max(0, n - len(fan_out_index_chain)) branches = tuple(branch_name_chain[: max(0, n - 1)]) + (None,) * max(0, (n - 1) - len(branch_name_chain)) # The branch's identity at THIS key's position, which is not always the @@ -162,9 +159,8 @@ def dispatch_key( # because every lookup is gated on the fan-out axis at the lookup depth: # each call site computes `fi_axis` as the chain entry for that depth and # skips the lookup when it is None, which implies the chain already reaches - # `n`. Grep `fi_axis` for the sites -- line numbers were tried here and - # went stale within the same commit that wrote them. An ungated lookup with - # a short chain would build a short tuple and miss the padded registration - # key, which is the orphan-lookup miss fixture 152 exists for. + # `n`. Grep `fi_axis` for the sites. An ungated lookup with a short chain + # builds a short tuple and misses the padded registration key, which is the + # orphan-lookup miss fixture 152 covers. n = len(prefix) return (prefix, tuple(fan_out_index_chain[:n]), tuple(branch_name_chain[:n])) diff --git a/src/openarmature/observability/metadata.py b/src/openarmature/observability/metadata.py index b18f4f7..21695f3 100644 --- a/src/openarmature/observability/metadata.py +++ b/src/openarmature/observability/metadata.py @@ -61,10 +61,6 @@ "openarmature.invocation_metadata", default=_EMPTY_METADATA ) -# The module docstring names these two sets rather than spelling them out. A -# hand-copied list there went stale the moment 0119 extended the tuple, and the -# copy is what a reader trusts. -# # Reserved key prefixes per §3.4. Keys with these prefixes are # off-limits to caller-supplied metadata; the engine rejects at the # boundary so observers never see a colliding key. @@ -280,11 +276,9 @@ def _validate_metadata_key(key: Any) -> None: raise ValueError(f"invocation metadata key must be a string; got {type(key).__name__}") for reserved in _RESERVED_PREFIXES: if key.startswith(reserved): - # The list is rendered FROM the tuple, never hand-written. A - # hardcoded copy went stale the moment 0119 added a third prefix, - # leaving the message naming a prefix its own guidance excluded. - # "namespaces" rather than "attributes": these cover OTel span - # attributes AND the Langfuse top-level metadata keys. + # Rendered from the tuple so the message cannot name a prefix its + # own guidance omits. "Namespaces" rather than "attributes": these + # cover OTel span attributes and Langfuse metadata keys alike. known = ", ".join(f"{p}*" for p in _RESERVED_PREFIXES) raise ValueError( f"invocation metadata key {key!r} uses reserved namespace prefix {reserved!r}; " diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index c7789ef..0a1dcdf 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -145,32 +145,15 @@ logger = logging.getLogger("openarmature.observability") -# Span-stack key shape: -# ``(namespace, attempt_index, fan_out_index, branch_name)`` — these -# four fields jointly identify any node attempt within an invocation. -# ``branch_name`` discriminates concurrent same-named inner nodes -# across sibling parallel-branches branches (pipeline-utilities §11); -# without it the two inner ``ask`` nodes of two branches with the -# same namespace + fan_out_index would collide on the same key. -# Proposal 0084 (spec v0.81.0): a node span is keyed by the innermost scalars -# AND the full enclosing fan-out / branch lineage chains. The chains are the -# addition: an inner node under two concurrent outer fan-out instances shares -# the same innermost scalar across the outer instances, so a scalar-only key -# would collide (last-writer-wins drops / mis-closes the second) and the -# LLM-parent exact-match would resolve to a sibling's span. The scalars are -# RETAINED (not replaced by the chains) because ``_key_for`` is also used to -# LOOK UP a callable-parallel-branch event in ``open_spans`` on the publish -# path (``_publish_active_span``): a callable branch carries its branch_name on -# the event but never extends branch_name_chain (no subgraph descent), so on -# the chains alone its key would EQUAL its own parallel-branches NODE's key -- -# the lookup would return that NODE instead of missing, and the branch's -# per-branch dispatch span would never be synthesized (verified: blanking the -# scalars drops the dispatch span). The scalar branch_name (key[3]) keeps the -# two distinct. Past that lookup the scalars are redundant with the chains for -# storage uniqueness (each equals its chain's innermost non-None entry); the -# merged spec keys the driving span chains-only, so the retained scalar is a -# harmless superset kept for the callable-branch lookup. The common -# single-level case keys as before plus the (empty / length-1) chains. +# Keyed by the innermost scalars AND the enclosing lineage chains (0084). +# +# The chains stop an inner node under two concurrent outer instances from +# colliding, since it shares the innermost scalar across them. The scalars are +# retained for one reason: `prepare_sync` looks up a callable parallel-branch +# here, and such a branch carries branch_name on the event without ever +# extending the chain. On the chains alone its key would EQUAL its +# own node's key, the lookup would hit that node instead of missing, and the +# per-branch dispatch span would never be synthesized. _StackKey = tuple[ tuple[str, ...], int, int | None, str | None, tuple[int | None, ...], tuple[str | None, ...] ] @@ -280,21 +263,14 @@ def _read_implementation_version() -> str: class _LineageEvent(Protocol): """The lineage an event must carry to place a span in the trace tree.""" - # Typed as a Protocol rather than `Any` because `Any` is exactly what let a - # `FailureIsolatedEvent` reach an opener annotated `event: NodeEvent`, where - # it raised on a field it does not declare. Structural, not a union of the - # eight concrete kinds: what synthesis needs is these six fields, and a new - # event kind that carries them should work without editing a list here. + # A Protocol rather than `Any`, which is what let a `FailureIsolatedEvent` + # reach an opener annotated `event: NodeEvent` and raise on a field it does + # not declare. Structural rather than a union, so a new event kind carrying + # these fields works without editing a list here. # - # The fields that vary -- `correlation_id` and `subgraph_identities` -- are - # deliberately ABSENT from this Protocol and read defensively at each use, - # because `FailureIsolatedEvent` declares neither and a Protocol cannot - # express "may be absent". - # - # `caller_invocation_metadata` belonged to that list until the change that - # added it to `FailureIsolatedEvent`, so no shipped kind now lacks it. It is - # still absent from this Protocol and still read defensively: the Protocol is - # the contract, and a conforming event is free not to carry the field. + # `correlation_id`, `subgraph_identities` and `caller_invocation_metadata` + # are deliberately absent and read defensively: a Protocol cannot express + # "may be absent", and a conforming event may not carry them. @property def namespace(self) -> tuple[str, ...]: ... @property @@ -382,10 +358,9 @@ def _subgraph_identity_at(event: object, depth: int) -> str: callers using ``SubgraphNode(name=..., compiled=...)`` without supplying ``subgraph_identity``. """ - # Spec observability §5.3 (coord thread - # clarify-subgraph-name-semantics). - # `getattr`, because a dispatch span may now be synthesized from a provider - # or tool event, which carries the lineage fields but not the identities. + # §5.3. `getattr` because a dispatch span may be synthesized from a + # provider or tool event, which carries the lineage fields but not the + # identities. # The empty-string fallback below is that case; `_backfill_subgraph_identity` # fills it in from the first node event, so the attribute does not depend on # which event happened to trigger synthesis. @@ -955,10 +930,9 @@ async def __call__( if not self.disable_llm_spans: self._handle_typed_llm_retry_attempt(event) return - # The terminal LlmCompletionEvent / LlmFailedEvent no longer - # drive the OTel span (the per-attempt event does); they stay on - # the queue for the Langfuse mapping and payload/latency - # consumers, so the OTel observer ignores them here. + # The per-attempt event drives the OTel span, so the terminal + # completion / failure events are ignored here. They stay on the queue + # for the Langfuse mapping and the payload and latency consumers. if isinstance(event, LlmCompletionEvent | LlmFailedEvent): return # Proposal 0059 embedding observability (observability §5.5.8): emit @@ -1302,24 +1276,14 @@ def _handle_completed(self, event: NodeEvent) -> None: inv_state.detached_roots.pop(event.namespace, None) def _propagate_error_to_detached_spans(self, inv_state: _InvState, event: NodeEvent) -> None: - # Proposal 0061 §4.2 (Detached invocation span status): a node - # raising inside a detached subtree surfaces ERROR on that - # trace's OWN carriers, not just the parent trace's. For each - # enclosing detached prefix: - # - the detached invocation span (the detached trace's root / - # authoritative carrier) and the parent-trace dispatch span - # (the §4.4 Link carrier) each get the FULL treatment — - # ERROR status + an OTel exception event + the §4 category - # attribute, mirroring the parent invocation span; - # - the detached subgraph / instance span between them gets - # ERROR status only (the invocation span above it carries - # the exception event for that trace). - # Set while the spans are still open; the synthetic close paths - # SKIP their default ``set_status(OK)`` for keys recorded in - # ``errored_detached_keys`` (OTel treats OK as final and lets it - # override a prior ERROR), so the ERROR survives to export. Keys - # cover both the detached-subgraph (prefix) and detached-fan-out- - # instance (prefix + index) schemes. + # 0061 §4.2: a raise inside a detached subtree surfaces ERROR on that + # trace's own carriers, not only the parent's. The detached invocation + # span and the parent-trace dispatch span get the full treatment; the + # span between them gets status only. + # + # `errored_detached_keys` makes the synthetic close paths skip their + # default `set_status(OK)`, which OTel treats as final and would + # otherwise override the ERROR before export. if event.error is None: return err = event.error @@ -1351,26 +1315,13 @@ def _propagate_error_to_detached_spans(self, inv_state: _InvState, event: NodeEv # ------------------------------------------------------------------ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> None: - # Spec proposal 0040: spans whose lineage ancestor-or-equals the - # augmenting context (within the same fan-out instance / - # parallel-branch boundary) get ``openarmature.user.`` - # applied in place. Sibling instances / branches and ancestors - # ABOVE the boundary are skipped. + # 0040: apply `openarmature.user.` to spans whose lineage + # ancestor-or-equals the augmenting context, skipping sibling instances + # and everything above the boundary. # - # Match rule (using the augmentation event's lineage tuple - # ``(NS, AI, FI, BN)``): - # - Invocation span: included iff ``FI is None and BN is None`` - # (outermost-serial context). The shared fan-out node span and - # the invocation span are explicitly out of scope when - # augmenting from inside a fan-out instance or branch. - # - Subgraph wrapper spans: included on the outermost-serial - # path when their namespace is a strict prefix of NS. - # - Fan-out instance dispatch spans: included iff the dispatch - # span's FI suffix matches ``str(FI)`` and the anchor namespace - # is a strict prefix of NS. - # - Per-attempt node spans (``open_spans``): included iff the - # span's FI equals the augmenter's FI and its namespace is a - # prefix of (or equal to) NS. + # The invocation span and the shared fan-out node span are in scope only + # for an outermost-serial augmenter; from inside an instance or branch + # they belong to siblings too. from openarmature.observability.correlation import current_invocation_id invocation_id = current_invocation_id() @@ -2387,29 +2338,10 @@ def _handle_failure_isolated(self, event: FailureIsolatedEvent) -> None: cid = current_correlation_id() if cid is not None: attrs["openarmature.correlation_id"] = cid - # The cross-cutting caller set. It carried none until the event gained - # the field, because there was nothing to read. - # - # NOT a §5.6 obligation, though two earlier versions of this comment got - # the reason wrong in opposite directions. - # - # The reason is that `openarmature.failure_isolated` appears nowhere in - # the observability spec: the failure-isolation EVENT is mandated by - # pipeline-utilities, but the span we emit from it is ours and no §8.4.x - # table maps it. §5.6 cannot mandate attributes on a span the spec never - # defines. - # - # It is NOT that §5.6's list of span kinds excludes this one. A previous - # version argued exactly that, and spec has since ruled the list is - # illustrative rather than exhaustive: the rule reaches every span - # emitted during the invocation, because a framework span missing - # `openarmature.correlation_id` is broken in the way that attribute - # exists to prevent. So do not resurrect the enumeration argument. - # - # The practical consequence: once the span IS mapped, §5.6 reaches it and - # this stops being a consistency choice and becomes required. Spec has - # committed to that mapping (coord release-v0.17.0/52, /54); until it - # lands this is ours. + # Not a §5.6 obligation: `openarmature.failure_isolated` appears + # nowhere in the observability spec, so §5.6 cannot reach it. The span + # is ours and no §8.4.x table maps it. Once it is mapped this becomes + # required rather than a consistency choice. _apply_caller_metadata(attrs, _event_caller_metadata(event)) span = self._tracer.start_span( name="openarmature.failure_isolated", @@ -2567,22 +2499,11 @@ def _resolve_enclosing_wrapper_context( node event's lineage) and the §5.5 orphan LLM-span fallback (called with the calling node's lineage when its span is not open), so both resolve to the same parent.""" - # 1. Walk prefix lengths longest-to-shortest. The INNERMOST - # matching synthetic dispatch span wins. Three keying - # schemes live alongside each other at each prefix: - # - per-branch dispatch (proposal 0044, v0.36.0): keyed by - # ``prefix + (branch_name,)`` in - # ``parallel_branches_branch_spans`` - # - detached fan-out instance root: keyed by - # ``prefix + (str(fan_out_index),)`` in - # ``detached_roots`` - # - non-detached fan-out instance dispatch (proposal 0013, - # v0.10.0): keyed by ``prefix + (str(fan_out_index),)`` - # in ``fan_out_instance_spans`` - # Walking longest-to-shortest gives the right answer for - # arbitrary composition (parallel-branches inside fan-out - # instance and vice versa) — the dispatch span at the - # deepest matching depth is the most-immediate parent. + # Longest-to-shortest so the innermost dispatch span wins, which is + # what makes arbitrary composition work (branches inside instances and + # the reverse). Three keying schemes coexist at each prefix: per-branch + # under `prefix + (branch_name,)`, and both detached and non-detached + # instances under `prefix + (fan_out_index,)`. for prefix_len in range(len(namespace), 0, -1): prefix = namespace[:prefix_len] # Lineage-aware keys (proposal 0045): carry the enclosing fan-out @@ -2678,32 +2599,13 @@ def _sync_subgraph_spans( prefix = namespace[:depth] if prefix in inv_state.subgraph_spans: continue - # `detached_roots` holds TWO key shapes. A detached SUBGRAPH root is - # stored under the bare prefix; a detached FAN-OUT INSTANCE root is - # stored under `prefix + (str(fan_out_index),)`. + # `detached_roots` holds two key shapes: a subgraph root under the + # bare prefix, an instance root under `prefix + (fan_out_index,)`. + # Each arm must test its own. # - # The bare prefix therefore never matched an instance root, so the - # instance arm below re-fired once per inner node event instead of - # once per instance. The second open replaced both dict entries, - # leaving the first root and its detached invocation span unended - # and unexported, and splitting one instance's inner nodes across - # traces with the first pointing at a parent nothing emitted. - # Visible only with two or more nodes in the instance subgraph: with - # one there is a single event and a single open. The instance arm - # now tests its OWN key. - # - # This test is REDUNDANT today, and is kept as depth rather than - # because anything reaches it. `_open_detached_subgraph_root` writes - # `subgraph_spans[prefix]` as well as `detached_roots[prefix]`, and - # the `subgraph_spans` guard immediately above runs first, so this - # one can never be the guard that fires. Established by mutation: - # deleting this leaves the whole suite green, while deleting the - # `subgraph_spans` guard is caught by conformance fixture 002. - # - # It is retained because a future path that populates - # `detached_roots` without `subgraph_spans` would need it, and the - # #279 rework may add exactly that. Delete it if that stops being - # plausible -- but do not keep it believing it is load-bearing. + # Redundant today: the `subgraph_spans` guard above always fires + # first, and mutation confirms deleting this leaves the suite green. + # Kept for a future path that populates `detached_roots` alone. if prefix in inv_state.detached_roots: continue # The fan-out instance axis at THIS depth -- the chain entry for the @@ -3103,27 +3005,17 @@ def _synthesize_call_site_wrapper_spans( ) -> None: """Open any dispatch span the CALLING lineage sits inside that has not been synthesized yet.""" - # Spec observability §5.5 (Lineage-resolved parent), as ruled in the - # release-v0.17.0 coord thread: the parent is resolved STRUCTURALLY. A - # call issued from branch or instance middleware is inside that branch or - # instance, so its nearest enclosing wrapper is that dispatch span, - # whether or not the observer has materialized it yet. - # - # Without this, the parent depended on drain scheduling. Dispatch spans - # are synthesized from inner NODE events, and a wrapper-issued provider - # call is enqueued BEFORE the wrapper's first inner node starts. Whether - # the span existed at resolution time came down to whether anything - # yielded to the event loop in between: one `await asyncio.sleep(0)` in - # user middleware moved the span from its branch to the invocation root. - # §10 covers parentage, so that was non-conforming rather than untidy. + # §5.5: the parent resolves STRUCTURALLY. A call from branch or + # instance middleware is inside that branch, so its enclosing wrapper is + # that dispatch span whether or not the observer has materialized it. + # Resolving it dynamically made the parent depend on drain scheduling: + # one `await asyncio.sleep(0)` in user middleware moved the span to the + # invocation root. # - # Two differences from `_sync_subgraph_spans`, which does this for node - # events. It walks PROPER ancestors (`range(1, len(namespace))`), but a - # call from branch middleware sits AT the parallel-branches namespace, so - # its branch's prefix IS the full namespace and that walk never reaches - # it. And this opens dispatch spans only: no subgraph wrappers, no - # detached roots, and nothing is closed, since a provider event is not a - # position change. + # Unlike `_sync_subgraph_spans` this walks the FULL namespace, not + # proper ancestors, because a branch-middleware call sits AT the + # parallel-branches namespace. It opens dispatch spans only and closes + # nothing: a provider event is not a position change. # Correlation comes from the invocation, not the event. Reading it off # the event silently omitted `openarmature.correlation_id` for any kind # that does not declare the field (`FailureIsolatedEvent`), and nothing @@ -3140,32 +3032,13 @@ def _synthesize_call_site_wrapper_spans( for depth in range(1, len(namespace) + 1): prefix = namespace[:depth] fi_axis = fan_out_index_chain[depth - 1] if depth - 1 < len(fan_out_index_chain) else None - # The two DETACHED arms of `_sync_subgraph_spans` are deliberately - # NOT mirrored here. Doing so was attempted and reverted: an - # adversarial review found four separate defects, three of which - # only exist once this path can reach the detached openers. - # - # The root cause is pre-existing and has to be fixed first. - # `_sync_subgraph_spans` guards its detached fan-out arm with - # `if prefix in inv_state.detached_roots`, the BARE prefix, while - # `_open_detached_fan_out_instance_root` stores under - # `prefix + (str(fan_out_index),)`. That guard therefore never - # matches. With one caller it is merely dead; with two it is a - # double-open that overwrites the first root, leaks two unended - # spans, and strands the orphan in an abandoned trace. - # - # The others: the openers assume an invocation span already exists, - # which is true from the node path and not from here, so a call - # arriving first mints a disconnected trace; and - # `_open_detached_subgraph_root` reads the identity off the - # triggering event, so an identity-less provider event permanently - # blanks `openarmature.subgraph.name` on a subgraph that declares - # one -- a regression against this file's own behaviour today. + # The detached arms of `_sync_subgraph_spans` are deliberately not + # mirrored here: the detached openers assume an invocation span + # already exists, which holds from the node path and not from this + # one. Mirroring them is the open half of issue #279. # - # Consequence of leaving it: for a wrapper-issued call inside a - # detached wrapper the enclosing span is still decided by drain - # scheduling, and the divergence is in trace id rather than only in - # parent. That is the open half of issue #279. + # Consequence: for a wrapper-issued call inside a detached wrapper + # the enclosing span still depends on drain scheduling. if ( fi_axis is not None and prefix[-1] not in self.detached_fan_outs