From 0520c1e6816eb573a88c97c48efd2faec602058f Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 3 Sep 2026 17:39:01 -0700 Subject: [PATCH 1/5] Compress narrative comments in the two observers Applies the comment rules: cut history, self-justification, restatements of what the code does, and prose duplicating a CHANGELOG entry. Spec citations stay, compressed to the citation and the obligation. langfuse/observer.py drops 227 comment lines, 31.4% to 25.2%. otel starts here and continues in the next commit. The ratio target of 10-15% is not reachable by cutting alone on files this size, which is the finding the ratio is meant to surface. Tracked separately as the file split. AGENTS.md gains the docstring convention this sweep works alongside: docstrings are published through mkdocstrings, so spec refs and normative language belong in comments instead. --- AGENTS.md | 37 ++ .../observability/langfuse/observer.py | 461 +++++------------- .../observability/otel/observer.py | 154 ++---- 3 files changed, 189 insertions(+), 463 deletions(-) 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/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 814bda1..2541c22 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, ...] ] @@ -251,15 +244,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 +300,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 +453,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 +542,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 +645,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 +771,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 +850,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 +920,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 +992,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 +1005,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 +1044,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 +1081,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 +1188,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 +1278,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 +1441,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 +1484,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 +1504,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 +1521,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 +1616,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 +1737,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 +1949,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 +1980,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 +2249,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 +2323,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 +2414,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 +2425,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/otel/observer.py b/src/openarmature/observability/otel/observer.py index c7789ef..7ea8193 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: `_publish_active_span` 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, ...] ] @@ -2387,29 +2370,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", @@ -2678,32 +2642,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),)`. - # - # 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. + # `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. # - # 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 +3048,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 +3075,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 From ba6750f4580db4ed252c9e4f76132fa6ea692fe5 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 3 Sep 2026 21:48:11 -0700 Subject: [PATCH 2/5] Cut change-justification from comments across src Each of these stated a real constraint and then justified the change that introduced it. The constraint stays, in present tense; the story of how it got there goes to git. Removed: what an earlier version did, what a defect used to be, which coord thread settled a question, and meta-commentary about a previous version of the comment itself. Spec citations and the obligations they carry are untouched, and so is the reasoning that describes what the code currently has to satisfy. --- src/openarmature/graph/compiled.py | 31 ++--- src/openarmature/llm/providers/openai.py | 12 +- .../observability/langfuse/observer.py | 5 +- src/openarmature/observability/lineage.py | 20 ++-- src/openarmature/observability/metadata.py | 12 +- .../observability/otel/observer.py | 107 ++++++------------ 6 files changed, 61 insertions(+), 126 deletions(-) diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py index 32d7b99..6c82b0e 100644 --- a/src/openarmature/graph/compiled.py +++ b/src/openarmature/graph/compiled.py @@ -331,10 +331,8 @@ def _find_innermost_fan_out_instance_state( # 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.) + # appends the branch name to the chain. The BRANCH name never enters + # the namespace; the node name does. # # Unsliced was reachable-wrong rather than harmless: with a branch # descent at or below the candidate fan-out's depth, the key carried @@ -1545,13 +1543,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) @@ -1854,15 +1849,11 @@ async def _step_fan_out_node( # load time. Fires once per fan-out step. 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: diff --git a/src/openarmature/llm/providers/openai.py b/src/openarmature/llm/providers/openai.py index cfa3bf3..4db6fec 100644 --- a/src/openarmature/llm/providers/openai.py +++ b/src/openarmature/llm/providers/openai.py @@ -1641,13 +1641,11 @@ 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) diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 2541c22..a3cf216 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -220,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] 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 7ea8193..dcc5548 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -263,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 @@ -365,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. @@ -938,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 @@ -1285,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 @@ -1334,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() @@ -2531,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 From 0b8972be6c11a9875e58d3daac2416e2afeddabf Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Thu, 3 Sep 2026 22:08:41 -0700 Subject: [PATCH 3/5] Cut over-explanation in compiled.py Bulleted walkthroughs of matching semantics, restatements of what the surrounding code does, and reasoning phrased as the story of a fix. Each is rewritten to state the constraint the code satisfies now. The one worth keeping in some form is the branch-axis slice: nothing reaches it today only because branch descent runs with checkpointer set to None, which is a policy in another module. That dependency is a fact about the code as it stands, so it stays. --- src/openarmature/graph/compiled.py | 204 +++++++++++------------------ 1 file changed, 73 insertions(+), 131 deletions(-) diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py index 6c82b0e..77574fb 100644 --- a/src/openarmature/graph/compiled.py +++ b/src/openarmature/graph/compiled.py @@ -320,29 +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. + # 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], @@ -390,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( @@ -463,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 @@ -1387,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: @@ -1832,21 +1806,13 @@ 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 # Resolution sits outside ``innermost``'s ``except Exception → @@ -1978,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, @@ -2065,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, @@ -2508,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, From b477d318077551810d4febfaac72bd971cfe8489 Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 4 Sep 2026 16:22:40 -0700 Subject: [PATCH 4/5] Cut over-explanation in the event and engine modules Same content test applied to events.py, openai.py, fan_out.py and graph/observer.py: keep what describes the code as it stands, drop what records how it got there. The recurring shape in these four was a real invariant followed by the story of the bug that revealed it. The fan-out tracking key is the clearest: the key must carry both concurrency axes or two live executions collide, which stays, while the account of the branch axis having been absent goes. Field docs on the event dataclasses lost their proposal-by-proposal change logs and kept the contract: when the event fires, what is mutually exclusive with what, and that payload fields are populated unconditionally with the privacy gates applied at rendering. --- src/openarmature/graph/events.py | 159 ++++++++--------------- src/openarmature/graph/fan_out.py | 83 +++++------- src/openarmature/graph/observer.py | 67 ++++------ src/openarmature/llm/providers/openai.py | 85 +++++------- 4 files changed, 136 insertions(+), 258 deletions(-) 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 4db6fec..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"] = { @@ -1647,24 +1634,12 @@ def _block_to_wire(block: ContentBlock) -> dict[str, Any]: # 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): From 64f2abc61229a5275bc63025d6f851f503fc955a Mon Sep 17 00:00:00 2001 From: chris-colinsky Date: Fri, 4 Sep 2026 16:53:44 -0700 Subject: [PATCH 5/5] Name the real symbol in the stack-key comment The comment pointed at `_publish_active_span`, which exists nowhere in the repo. The lookup it describes is in `prepare_sync`, whose `open_span is None` branch is the callable-parallel-branch path that depends on the key missing. The stale name predates this branch. --- src/openarmature/observability/otel/observer.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index dcc5548..0a1dcdf 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -149,9 +149,9 @@ # # 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: `_publish_active_span` 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 +# 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[