diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bbdc42f..7c5eb3f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,14 +6,29 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The ## [Unreleased] +### Added + +- **`LangfuseObserver` Trace input/output sourcing** (proposal 0043, observability §8.4.1). New observer construction knobs populate `trace.input` and `trace.output` per the three-lever decision tree: + - **`disable_state_payload: bool = True`** — privacy knob symmetric to `disable_llm_payload`. When ON (default), Trace fields receive the minimal stub `{entry_node, correlation_id}` / `{final_node, status}`; when OFF, the raw state object is serialized. + - **`trace_input_from_state` / `trace_output_from_state`** — optional caller hooks returning the domain-shaped value to use for `trace.input` / `trace.output`. Returning `None` falls through to the next applicable lever. + - `status` is the closed `Literal["completed", "failed"]` enum from spec §8.4.1. +- **Two new observer event types** delivered through the existing `graph.observer.Observer` queue: + - **`InvocationStartedEvent(initial_state, invocation_id, correlation_id, entry_node)`** — emitted once at invocation entry before any node fires. + - **`InvocationCompletedEvent(final_state, status, final_node, invocation_id, correlation_id)`** — emitted once at invocation exit on both the success path (`status="completed"`) and failure path (`status="failed"`). + + The `Observer.__call__` signature widens to `NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent`. The new `ObserverEvent` type alias (re-exported from `openarmature.graph`) gives observer authors a one-name handle on the union; existing observers that ignore non-`NodeEvent` variants early-return after an `isinstance(event, NodeEvent)` check. +- **`LangfuseTrace.input` / `LangfuseTrace.output` dataclass fields** on the in-memory recorder, populated by the new observer paths. + ### Changed - **Reserved-key extension** (proposal 0042, observability §3.4). Three additional bare key names — `branch_name`, `detached`, `detached_from_invocation_id` — are reserved against caller-supplied `invocation_metadata` and `set_invocation_metadata` collision; the framework rejects them at the `invoke()` boundary and at the mid-invocation augmentation helper with `ValueError`. The reserved-name set grows from 21 to 24. These three are top-level Langfuse metadata keys the observer mapping already writes; without reservation a caller key matching one would silently shadow the OA-emitted field. - **`observation.metadata.detached: true` moves to the parent-side dispatching observation** (proposal 0042, observability §8.4.2). The Langfuse mapping previously emitted `detached: true` on the dispatch observation inside the detached child trace; the §8.4.2 row added by 0042 places it on the **parent-side** dispatching observation that fires the detached child (the link observation in the main trace for detached subgraphs; the parent fan-out node observation for detached fan-outs). The detached-side observation no longer carries the flag. +- **`LangfuseClient.update_trace` Protocol grows `input` / `output` keyword parameters** so observer-supplied values land on the Trace's headline fields. ### Notes -- **Pinned spec version bumped from v0.31.0 to v0.34.0.** Absorbs proposals 0042 (reserved-key extension; observation.metadata.detached + branch_name + trace.metadata.detached_from_invocation_id rows), 0038 (Google Gemini wire-format mapping — not yet implemented in python), and 0020 (sessions capability — not yet implemented in python). +- **Pinned spec version bumped from v0.31.0 to v0.35.0.** Absorbs proposals 0042 (reserved-key extension), 0043 (Langfuse trace.input/output sourcing), and the textual additions in v0.32.0 (Gemini wire-format mapping, 0038, not yet implemented) and v0.33.0 (sessions capability, 0020, not yet implemented). +- The SDK adapter caches `input` / `output` in its `_trace_info` map; landing the values on the live Langfuse Trace from outside an active span context requires SDK-version-specific calls (v4's `langfuse.update_current_trace` works inside a context; cross-context REST updates need `client.api.trace.update`). The `InMemoryLangfuseClient` used by tests applies the fields directly. SDK-adapter end-to-end emit lands in a follow-up. ## [0.10.0] — 2026-05-27 diff --git a/conformance.toml b/conformance.toml index 9d8f0eef..783844dd 100644 --- a/conformance.toml +++ b/conformance.toml @@ -32,7 +32,7 @@ [manifest] implementation = "openarmature-python" -spec_pin = "v0.34.0" +spec_pin = "v0.35.0" # Status values: # implemented — shipped behavior matches the proposal's contract @@ -205,3 +205,8 @@ status = "not-yet" [proposals."0042"] status = "implemented" since = "0.11.0" + +# Spec v0.35.0 (proposal 0043). +[proposals."0043"] +status = "implemented" +since = "0.11.0" diff --git a/examples/00-hello-world/main.py b/examples/00-hello-world/main.py index 1591b5ff..c34d2330 100644 --- a/examples/00-hello-world/main.py +++ b/examples/00-hello-world/main.py @@ -49,8 +49,8 @@ END, CompiledGraph, GraphBuilder, - MetadataAugmentationEvent, NodeEvent, + ObserverEvent, State, append, merge, @@ -195,7 +195,7 @@ def route(state: PipelineState) -> str: return state.classification.intent -async def trace(event: NodeEvent | MetadataAugmentationEvent) -> None: +async def trace(event: ObserverEvent) -> None: # OpenAIProvider emits NodeEvent-shaped events for LLM-span # tracking under a sentinel namespace; those have post_state=None. # ``set_invocation_metadata`` from within a node body emits a diff --git a/examples/03-observer-hooks/main.py b/examples/03-observer-hooks/main.py index 2326369b..9c8430cd 100644 --- a/examples/03-observer-hooks/main.py +++ b/examples/03-observer-hooks/main.py @@ -54,9 +54,9 @@ CompiledGraph, ExplicitMapping, GraphBuilder, - MetadataAugmentationEvent, NodeEvent, Observer, + ObserverEvent, State, append, ) @@ -187,7 +187,7 @@ def build_review_subgraph() -> CompiledGraph[ReviewState]: # fire on every invocation of the compiled graph until removed. -async def console_tracer(event: NodeEvent | MetadataAugmentationEvent) -> None: +async def console_tracer(event: ObserverEvent) -> None: """Print one structured line per node boundary to stderr. Format: `[step=N] namespace.path → fields_changed_in_this_step` @@ -197,7 +197,7 @@ async def console_tracer(event: NodeEvent | MetadataAugmentationEvent) -> None: reach observers as ``MetadataAugmentationEvent`` instances; this tracer ignores them. """ - if isinstance(event, MetadataAugmentationEvent): + if not isinstance(event, NodeEvent): return namespace = ".".join(event.namespace) if event.error is not None: @@ -239,8 +239,8 @@ def __init__(self) -> None: self.errors: int = 0 self.namespaces: set[tuple[str, ...]] = set() - async def __call__(self, event: NodeEvent | MetadataAugmentationEvent) -> None: - if isinstance(event, MetadataAugmentationEvent): + async def __call__(self, event: ObserverEvent) -> None: + if not isinstance(event, NodeEvent): return self.events += 1 if event.error is not None: diff --git a/examples/04-nested-subgraphs/main.py b/examples/04-nested-subgraphs/main.py index 6c34bc5a..ec2cf203 100644 --- a/examples/04-nested-subgraphs/main.py +++ b/examples/04-nested-subgraphs/main.py @@ -49,8 +49,8 @@ CompiledGraph, ExplicitMapping, GraphBuilder, - MetadataAugmentationEvent, NodeEvent, + ObserverEvent, State, append, ) @@ -350,8 +350,8 @@ def _fmt_state(state: Any) -> str: return " ".join(parts) if parts else "(empty)" -async def depth_observer(event: NodeEvent | MetadataAugmentationEvent) -> None: - if isinstance(event, MetadataAugmentationEvent): +async def depth_observer(event: ObserverEvent) -> None: + if not isinstance(event, NodeEvent): return depth = len(event.namespace) indent = " " * (depth - 1) diff --git a/examples/05-fan-out-with-retry/main.py b/examples/05-fan-out-with-retry/main.py index d2093466..ffe65638 100644 --- a/examples/05-fan-out-with-retry/main.py +++ b/examples/05-fan-out-with-retry/main.py @@ -78,8 +78,8 @@ END, CompiledGraph, GraphBuilder, - MetadataAugmentationEvent, NodeEvent, + ObserverEvent, State, append, ) @@ -297,7 +297,7 @@ def build_graph(error_policy: str = "fail_fast") -> CompiledGraph[BatchState]: ) -async def fan_out_config_observer(event: NodeEvent | MetadataAugmentationEvent) -> None: +async def fan_out_config_observer(event: ObserverEvent) -> None: """Print the fan-out node's resolved config when its dispatch event fires. diff --git a/examples/06-parallel-branches/main.py b/examples/06-parallel-branches/main.py index edce6500..2f407efc 100644 --- a/examples/06-parallel-branches/main.py +++ b/examples/06-parallel-branches/main.py @@ -70,8 +70,8 @@ BranchSpec, CompiledGraph, GraphBuilder, - MetadataAugmentationEvent, NodeEvent, + ObserverEvent, State, append, ) @@ -241,7 +241,7 @@ async def present(s: ArticleState) -> Mapping[str, Any]: return {"trace": ["present"]} -async def branch_attribution_observer(event: NodeEvent | MetadataAugmentationEvent) -> None: +async def branch_attribution_observer(event: ObserverEvent) -> None: """Print which branch each inner-node event came from. NodeEvent carries ``branch_name`` on events from nodes that diff --git a/openarmature-spec b/openarmature-spec index 37e519c5..97659ecd 160000 --- a/openarmature-spec +++ b/openarmature-spec @@ -1 +1 @@ -Subproject commit 37e519c5a04630db65aea2be0a88fba0314b8972 +Subproject commit 97659ecda387970cd92d3bb71e23719f06f24cb2 diff --git a/pyproject.toml b/pyproject.toml index 4d2c054e..c5b17082 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,7 +58,7 @@ Specification = "https://github.com/LunarCommand/openarmature-spec" openarmature = "openarmature.cli:main" [tool.openarmature] -spec_version = "0.34.0" +spec_version = "0.35.0" [dependency-groups] dev = [ diff --git a/src/openarmature/AGENTS.md b/src/openarmature/AGENTS.md index 9df72451..6753e621 100644 --- a/src/openarmature/AGENTS.md +++ b/src/openarmature/AGENTS.md @@ -1,6 +1,6 @@ # OpenArmature — Agent documentation -*This is the agent guide bundled with the openarmature Python package, version 0.10.0 (spec v0.34.0). For the full docs site see [openarmature.ai](https://openarmature.ai). For the canonical spec text see [openarmature.org/capabilities](https://openarmature.org/capabilities/). For project-specific conventions for the code you're editing, see the host project's `AGENTS.md` or `CLAUDE.md`.* +*This is the agent guide bundled with the openarmature Python package, version 0.10.0 (spec v0.35.0). For the full docs site see [openarmature.ai](https://openarmature.ai). For the canonical spec text see [openarmature.org/capabilities](https://openarmature.org/capabilities/). For project-specific conventions for the code you're editing, see the host project's `AGENTS.md` or `CLAUDE.md`.* ## TL;DR @@ -10,7 +10,7 @@ OpenArmature is a workflow framework for LLM pipelines and tool-calling agents ## Capability contracts -_Sourced from openarmature-spec v0.34.0. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md`. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._ +_Sourced from openarmature-spec v0.35.0. Each entry below reproduces §1 (Purpose) and §2 (Concepts) of the capability's `spec.md`. For the full spec text (execution model, error semantics, determinism, observer hooks, etc.) see the linked docs site._ ### Capability: `graph-engine` diff --git a/src/openarmature/__init__.py b/src/openarmature/__init__.py index 53b055ce..544603b7 100644 --- a/src/openarmature/__init__.py +++ b/src/openarmature/__init__.py @@ -25,4 +25,4 @@ """ __version__ = "0.10.0" -__spec_version__ = "0.34.0" +__spec_version__ = "0.35.0" diff --git a/src/openarmature/graph/__init__.py b/src/openarmature/graph/__init__.py index bd45a8b5..4cd13453 100644 --- a/src/openarmature/graph/__init__.py +++ b/src/openarmature/graph/__init__.py @@ -35,7 +35,12 @@ StateValidationError, UnreachableNode, ) -from .events import MetadataAugmentationEvent, NodeEvent +from .events import ( + InvocationCompletedEvent, + InvocationStartedEvent, + MetadataAugmentationEvent, + NodeEvent, +) from .fan_out import FanOutConfig, FanOutNode from .middleware import ( Middleware, @@ -48,7 +53,7 @@ exponential_jitter_backoff, ) from .nodes import FunctionNode, Node -from .observer import DrainSummary, Observer, RemoveHandle, SubscribedObserver +from .observer import DrainSummary, Observer, ObserverEvent, RemoveHandle, SubscribedObserver from .parallel_branches import BranchSpec, ParallelBranchesNode from .projection import ExplicitMapping, FieldNameMatching, ProjectionStrategy from .reducers import Reducer, append, concat_flatten, last_write_wins, merge, merge_all @@ -77,6 +82,8 @@ "FunctionNode", "GraphBuilder", "GraphError", + "InvocationCompletedEvent", + "InvocationStartedEvent", "MappingReferencesUndeclaredField", "MetadataAugmentationEvent", "Middleware", @@ -87,6 +94,7 @@ "NodeException", "NoDeclaredEntry", "Observer", + "ObserverEvent", "ParallelBranchesBranchFailed", "ParallelBranchesNoBranches", "ParallelBranchesNode", diff --git a/src/openarmature/graph/compiled.py b/src/openarmature/graph/compiled.py index 9c9667ec..8ca7ba5a 100644 --- a/src/openarmature/graph/compiled.py +++ b/src/openarmature/graph/compiled.py @@ -29,7 +29,7 @@ from collections.abc import Callable, Iterable, Mapping, Sequence from dataclasses import dataclass, field from dataclasses import replace as dataclass_replace -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, Literal, cast if TYPE_CHECKING: # ``FanOutNode`` lives in ``.fan_out`` which has a TYPE_CHECKING @@ -95,7 +95,12 @@ RuntimeGraphError, StateValidationError, ) -from .events import FanOutEventConfig, NodeEvent +from .events import ( + FanOutEventConfig, + InvocationCompletedEvent, + InvocationStartedEvent, + NodeEvent, +) from .middleware import ChainCall, Middleware, compose_chain from .nodes import Node from .observer import ( @@ -1023,9 +1028,57 @@ async def invoke( caller_invocation_metadata=current_invocation_metadata(), ), ) + # Proposal 0043: invocation-boundary event for trace.input + # sourcing. Carries the engine-constructed initial_state plus + # the §3 / §5.1 ids and the outermost-graph entry node name + # so Trace-level observers (Langfuse) can populate + # ``trace.input`` via the §8.4.1 three-lever decision tree. + # Dispatched AFTER the checkpoint-migrated event (when there + # is one) so the migration span is observable before the + # invocation-input event. + _dispatch( + context, + InvocationStartedEvent( + initial_state=starting_state, + invocation_id=invocation_id, + correlation_id=resolved_correlation_id, + entry_node=self.entry, + ), + ) + final_state: StateT | None = None + status: Literal["completed", "failed"] = "failed" try: - return await self._invoke(starting_state, context) + final_state = await self._invoke(starting_state, context) + status = "completed" + return final_state finally: + # Proposal 0043: invocation-boundary event for trace.output + # sourcing. Fires on both the success path + # (status="completed") and the failure path + # (status="failed"). ``final_node`` comes from the shared + # box the engine populates as nodes enter; on the failure + # path that's the inner-most node that raised, on the + # success path that's the last node before the END-routing + # edge. ``final_state`` is the engine's returned state on + # success and ``starting_state`` on the failure path (the + # engine doesn't expose intermediate state across raises). + if context.final_node_box: + final_node = context.final_node_box[0] + else: + # Defensive: invocation raised before any node fired + # (e.g., resume-path validation). Fall back to the + # declared entry node. + final_node = self.entry + _dispatch( + context, + InvocationCompletedEvent( + final_state=final_state if final_state is not None else starting_state, + status=status, + final_node=final_node, + invocation_id=invocation_id, + correlation_id=resolved_correlation_id, + ), + ) _reset_invocation_metadata(metadata_token) _reset_invocation_id(invocation_token) _reset_correlation_id(correlation_token) @@ -1098,6 +1151,17 @@ async def _invoke( from .fan_out import FanOutNode # noqa: PLC0415 from .parallel_branches import ParallelBranchesNode # noqa: PLC0415 + # Proposal 0043: track the most recent node about to run + # so the outermost ``invoke()`` can populate + # ``InvocationCompletedEvent.final_node`` on both the + # END-reached success path (last node before the + # END-routing edge) and the failure path (the node that + # raised). Subgraph descents reuse the same shared box + # via ``descend_into_subgraph``, so a failure deep in a + # subgraph leaves the innermost node's name in the box — + # the actual culprit, not the wrapper. + context.final_node_box[:] = [current] + if isinstance(node, FanOutNode): # Fan-out nodes are recognized as a distinct node type # per pipeline-utilities §9. Dispatched through @@ -1138,6 +1202,20 @@ async def _invoke( step_result = await self._step_function_node(node, current, state, context) state = step_result.state + # Proposal 0043 (post-PR-99 review): restore the outer + # ``current`` to the shared box after a successful step. + # Descended `_step_*` calls (subgraph, fan-out, parallel- + # branches) write inner-node names into the box; without + # this restore, the wrapper's name leaks out of the box + # when the wrapper is the last node before the END-routing + # edge — and for parallel-branches the box would end with + # whichever branch's inner finished last (nondeterministic). + # On the failure path, the raise above bypasses this line, + # so the inner-most node that raised stays in the box as + # the failure-path ``final_node`` (matching spec §4 + # attribution). + context.final_node_box[:] = [current] + # Per spec graph-engine §3 step 3 (revised in proposal # 0012 / v0.9.0): the engine MUST dispatch the # ``completed`` event AFTER edge evaluation completes. diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py index c2cb50bd..bc728ad3 100644 --- a/src/openarmature/graph/events.py +++ b/src/openarmature/graph/events.py @@ -273,4 +273,89 @@ class MetadataAugmentationEvent: branch_name: str | None = None -__all__ = ["FanOutEventConfig", "MetadataAugmentationEvent", "NodeEvent"] +# Spec: realizes observability §8.4.1 *Trace input/output sourcing* +# (proposal 0043). Emitted by the engine at invocation entry, BEFORE +# any node fires. Carries the initial state observers can use to +# resolve trace.input via the three-lever decision tree (caller hook +# → raw state when disable_state_payload is OFF → privacy-safe +# minimal stub). Distinct from NodeEvent because there is no node +# context — the event is invocation-scoped. +@dataclass(frozen=True) +class InvocationStartedEvent: + """An invocation-entry event delivered to observers. + + Emitted once per invocation, before any node fires. Observers that + populate Trace-level input fields (the Langfuse observer, today) + consume it to resolve ``trace.input`` per the three-lever decision + tree in observability §8.4.1. Observers without a Trace-level + input concept (the OTel observer) treat it as a no-op. + + Carries: + + - ``initial_state``: the raw state object the engine constructed + from ``invoke()``'s arguments (the typed-state instance). + - ``invocation_id``: the invocation id (caller-supplied or + framework-generated per proposal 0039). + - ``correlation_id``: the §3 correlation id when present. + - ``entry_node``: the outermost-graph entry node name. + + Per graph-engine §6 the event is NOT subject to the observer + ``phases`` filter (which only governs ``NodeEvent`` phases); the + delivery worker forwards it to every subscribed observer. + """ + + initial_state: Any + invocation_id: str + correlation_id: str | None + entry_node: str + + +# Spec: realizes observability §8.4.1 *Trace input/output sourcing* +# (proposal 0043). Emitted by the engine at invocation exit, on both +# the success path (status="completed") and the failure path +# (status="failed"). Carries the final state observers can use to +# resolve trace.output via the three-lever decision tree, plus the +# closed status enum for the privacy-safe minimal stub. +@dataclass(frozen=True) +class InvocationCompletedEvent: + """An invocation-exit event delivered to observers. + + Emitted once per invocation, after the last node has fired (and + after a failure boundary on the failure path). Observers that + populate Trace-level output fields (the Langfuse observer, today) + consume it to resolve ``trace.output`` per the three-lever + decision tree in observability §8.4.1. Observers without a + Trace-level output concept (the OTel observer) treat it as a no-op. + + Carries: + + - ``final_state``: the state at invocation exit (the engine's + returned state on the success path; the state at point-of- + failure on the failure path). + - ``status``: closed enum ``"completed"`` (END reached) or + ``"failed"`` (any node, edge, reducer, or boundary validator + raised before END). + - ``final_node``: the name of the node whose execution preceded + the END-reached transition on the success path, or the node + that raised on the failure path. + - ``invocation_id`` / ``correlation_id``: the §3 / §5.1 ids. + + Per graph-engine §6 the event is NOT subject to the observer + ``phases`` filter; the delivery worker forwards it to every + subscribed observer. + """ + + final_state: Any + status: Literal["completed", "failed"] + final_node: str + invocation_id: str + correlation_id: str | None + + +__all__ = [ + "FanOutEventConfig", + "InvocationCompletedEvent", + "InvocationStartedEvent", + "MetadataAugmentationEvent", + "NodeEvent", +] diff --git a/src/openarmature/graph/observer.py b/src/openarmature/graph/observer.py index 47ee619f..0487dafd 100644 --- a/src/openarmature/graph/observer.py +++ b/src/openarmature/graph/observer.py @@ -34,9 +34,21 @@ from dataclasses import dataclass, field from typing import Any, Literal, Protocol -from .events import MetadataAugmentationEvent, NodeEvent +from .events import ( + InvocationCompletedEvent, + InvocationStartedEvent, + MetadataAugmentationEvent, + NodeEvent, +) from .state import State +# Union of every event variant an Observer may receive. NodeEvent is +# the original §6 started/completed/checkpoint shape; the other three +# are side-channel events (proposal 0040 for augmentation; proposal +# 0043 for invocation-boundary trace.input/output sourcing) that +# bypass the phase filter and reach every subscribed observer. +ObserverEvent = NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent + class Observer(Protocol): """The shape of a callable that receives observer events. @@ -64,9 +76,9 @@ async def log_observer(event: NodeEvent | MetadataAugmentationEvent) -> None: conformance doesn't pin you to that name; any of `event`, `_event`, `e`, etc. matches. - Two event variants reach observers (graph-engine §6 + proposal - 0040). The signature is the union; observers ``isinstance``-narrow - on the first line and choose which variants they handle. + Four event variants reach observers (graph-engine §6 + proposals + 0040, 0043). The signature is the union; observers ``isinstance``- + narrow on the first line and choose which variants they handle. - :class:`NodeEvent` — the started/completed/checkpoint phase events. Subject to the ``phases`` filter on @@ -84,6 +96,19 @@ async def log_observer(event: NodeEvent | MetadataAugmentationEvent) -> None: subscribed observer sees it and isinstance-narrows to decide whether to act. Simple user observers typically early-return after ``isinstance(event, NodeEvent)`` checks. + - :class:`InvocationStartedEvent` — emitted once per invocation + before any node fires. Carries the engine-constructed + ``initial_state`` so Trace-level backends (Langfuse) can + populate ``trace.input`` via the proposal 0043 three-lever + decision tree. NOT subject to the ``phases`` filter; OTel-only + observers ignore it via the isinstance gate. + - :class:`InvocationCompletedEvent` — emitted once per invocation + after the last node fires (on both the success path and the + failure path). Carries ``final_state`` + a closed + ``status: {"completed", "failed"}`` enum so Trace-level + backends can populate ``trace.output``. NOT subject to the + ``phases`` filter; OTel-only observers ignore it via the + isinstance gate. Optional ``prepare_sync`` extension ----------------------------------- @@ -109,7 +134,7 @@ def prepare_sync(self, event: NodeEvent, /) -> None: ... carries). """ - async def __call__(self, event: NodeEvent | MetadataAugmentationEvent, /) -> None: ... + async def __call__(self, event: ObserverEvent, /) -> None: ... # Per spec v0.6.0 §6: the two valid phase strings. Used as the default @@ -228,20 +253,23 @@ class _QueuedItem: without the worker needing to know the graph topology. ``event`` is the union of ``NodeEvent`` (started / completed / - checkpoint phases) and ``MetadataAugmentationEvent`` (proposal - 0040, side-channel augmentation). The delivery worker branches by - type to apply the right delivery contract (phase-filter for - ``NodeEvent``, no filter for the augmentation event). + checkpoint phases), ``MetadataAugmentationEvent`` (proposal 0040, + side-channel augmentation), and the two invocation-boundary + events ``InvocationStartedEvent`` / ``InvocationCompletedEvent`` + (proposal 0043, Trace-level input/output sourcing). The delivery + worker branches by type to apply the right delivery contract + (phase-filter for ``NodeEvent``, no filter for the other three). """ - event: NodeEvent | MetadataAugmentationEvent + event: ObserverEvent observers: tuple[SubscribedObserver, ...] # A sentinel value the engine puts on the queue to signal the worker to # return after draining the events ahead of it. None is unambiguous — -# the queue carries `NodeEvent` and `MetadataAugmentationEvent` instances -# wrapped in `_QueuedItem`, never None. +# the queue carries ``NodeEvent``, ``MetadataAugmentationEvent``, and +# the two ``Invocation*Event`` variants wrapped in ``_QueuedItem``, +# never None. _DRAIN_SENTINEL = None @@ -459,6 +487,16 @@ class _InvocationContext: # ``Any`` rather than ``type[State]`` to avoid an import cycle # between graph and observer; 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 + # entered node's name. The outermost ``invoke()`` reads it on + # exit to populate ``InvocationCompletedEvent.final_node`` on + # both the success path (last node before END routing) and the + # failure path (the node whose execution raised). Shared by + # reference across subgraph / fan-out / parallel-branches + # 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]) def full_observers(self) -> tuple[SubscribedObserver, ...]: """Return the ordered observer list to deliver for events from @@ -506,6 +544,7 @@ def descend_into_subgraph( fan_out_progress_state=self.fan_out_progress_state, drain_counters=self.drain_counters, state_cls=self.state_cls, + final_node_box=self.final_node_box, ) def descend_into_fan_out_instance( @@ -556,6 +595,7 @@ def descend_into_fan_out_instance( fan_out_progress_state=self.fan_out_progress_state, drain_counters=self.drain_counters, state_cls=self.state_cls, + final_node_box=self.final_node_box, ) def descend_into_parallel_branch( @@ -609,6 +649,7 @@ def descend_into_parallel_branch( fan_out_progress_state=self.fan_out_progress_state, drain_counters=self.drain_counters, state_cls=self.state_cls, + final_node_box=self.final_node_box, ) def take_step(self) -> int: @@ -622,11 +663,11 @@ def take_step(self) -> int: def _dispatch( context: _InvocationContext, - event: NodeEvent | MetadataAugmentationEvent, + event: ObserverEvent, ) -> None: """Enqueue an event for the delivery worker. - Handles two event variants: + Handles four event variants: - :class:`NodeEvent`: the started/completed/checkpoint pair model. For ``"started"``-phase events, also calls any subscribed @@ -643,6 +684,13 @@ def _dispatch( anchored on ``"started"``, which only ``NodeEvent`` carries. Queued onto the same serial worker so observers see it in strict order with the surrounding node events. + - :class:`InvocationStartedEvent` / + :class:`InvocationCompletedEvent` (proposal 0043): invocation- + boundary events the engine enqueues at invocation entry / exit + so Trace-level backends can populate ``trace.input`` / + ``trace.output`` via the §8.4.1 three-lever decision tree. + Bypass ``prepare_sync`` (same rationale as + ``MetadataAugmentationEvent``: not a node-phase event). Phase-gated forwarding: ``prepare_sync`` only fires when ``"started"`` is in the subscribed observer's ``phases`` set, mirroring how the @@ -736,9 +784,11 @@ async def deliver_loop( the event's phase do NOT receive it. Phase filter applies at delivery, not dispatch; the engine still produces both events for every attempt. - - For :class:`MetadataAugmentationEvent` (proposal 0040), the - ``phases`` filter is bypassed entirely — the event isn't a - node-phase event, so every subscribed observer receives it + - For :class:`MetadataAugmentationEvent` (proposal 0040) and the + two invocation-boundary events :class:`InvocationStartedEvent` + / :class:`InvocationCompletedEvent` (proposal 0043), the + ``phases`` filter is bypassed entirely — none of those are + node-phase events, so every subscribed observer receives them regardless of ``phases``. Observers ``isinstance``-narrow on the first line and choose whether to act. - Observer exceptions don't propagate, don't break siblings, @@ -775,6 +825,7 @@ async def deliver_loop( "ALL_PHASES", "DrainSummary", "Observer", + "ObserverEvent", "RemoveHandle", "SubscribedObserver", # Engine-internal but listed so pyright sees them as exported (they're diff --git a/src/openarmature/observability/correlation.py b/src/openarmature/observability/correlation.py index 53f139b3..1ea67a2b 100644 --- a/src/openarmature/observability/correlation.py +++ b/src/openarmature/observability/correlation.py @@ -36,7 +36,12 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: - from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent + from openarmature.graph.events import ( + InvocationCompletedEvent, + InvocationStartedEvent, + MetadataAugmentationEvent, + NodeEvent, + ) from openarmature.graph.observer import SubscribedObserver @@ -208,12 +213,22 @@ def _reset_active_observers(token: Token[tuple[SubscribedObserver, ...]]) -> Non # --------------------------------------------------------------------------- -_active_dispatch_var: ContextVar[Callable[[NodeEvent | MetadataAugmentationEvent], None] | None] = ContextVar( - "openarmature.active_dispatch", default=None -) +_active_dispatch_var: ContextVar[ + Callable[ + [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + None, + ] + | None +] = ContextVar("openarmature.active_dispatch", default=None) -def current_dispatch() -> Callable[[NodeEvent | MetadataAugmentationEvent], None] | None: +def current_dispatch() -> ( + Callable[ + [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + None, + ] + | None +): """Return the engine's dispatch callable for the current invocation, or ``None`` outside any invocation. @@ -228,15 +243,30 @@ def current_dispatch() -> Callable[[NodeEvent | MetadataAugmentationEvent], None def _set_active_dispatch( - dispatch: Callable[[NodeEvent | MetadataAugmentationEvent], None], -) -> Token[Callable[[NodeEvent | MetadataAugmentationEvent], None] | None]: + dispatch: Callable[ + [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + None, + ], +) -> Token[ + Callable[ + [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + None, + ] + | None +]: """Set the engine's dispatch callable in scope. Internal — engine-only.""" return _active_dispatch_var.set(dispatch) def _reset_active_dispatch( - token: Token[Callable[[NodeEvent | MetadataAugmentationEvent], None] | None], + token: Token[ + Callable[ + [NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent], + None, + ] + | None + ], ) -> None: _active_dispatch_var.reset(token) diff --git a/src/openarmature/observability/langfuse/adapter.py b/src/openarmature/observability/langfuse/adapter.py index 5f74433d..dcce159f 100644 --- a/src/openarmature/observability/langfuse/adapter.py +++ b/src/openarmature/observability/langfuse/adapter.py @@ -221,6 +221,8 @@ def update_trace( id: str, name: str | None = None, metadata: dict[str, Any] | None = None, + input: Any | None = None, + output: Any | None = None, ) -> None: # Merge into the trace_info cache so subsequent observations # (and the first one if not yet created) pick up the updated @@ -228,17 +230,34 @@ def update_trace( # using cached info, update_trace takes effect on the NEXT # observation under this trace_id, not retroactively on prior # observations. + # + # Proposal 0043 ``input`` / ``output`` are cached but landing + # them on the live Langfuse Trace from outside an active + # span context is SDK-version-dependent (v4 exposes + # ``langfuse.update_current_trace(input=..., output=...)`` + # only inside a context; cross-context REST updates need + # ``client.api.trace.update``). The InMemoryLangfuseClient + # surface used by tests applies them directly. The SDK + # adapter's apply path is a follow-up — caching here so the + # Protocol contract is satisfied without breaking SDK-adapter + # users. entry = self._trace_info.get(id) if entry is None: self._trace_info[id] = { "name": name, "metadata": dict(metadata) if metadata is not None else {}, + "input": input, + "output": output, } return if name is not None: entry["name"] = name if metadata is not None: entry["metadata"].update(metadata) + if input is not None: + entry["input"] = input + if output is not None: + entry["output"] = output def span( self, diff --git a/src/openarmature/observability/langfuse/client.py b/src/openarmature/observability/langfuse/client.py index 9a7ba537..2132da3f 100644 --- a/src/openarmature/observability/langfuse/client.py +++ b/src/openarmature/observability/langfuse/client.py @@ -91,6 +91,11 @@ class LangfuseTrace: id: str name: str | None = None metadata: dict[str, Any] = field(default_factory=dict[str, Any]) + # Proposal 0043 §8.2: Trace gains explicit ``input`` / ``output`` + # payload fields. Populated by the Langfuse observer at the + # invocation-boundary events; absent when no observer wrote them. + input: Any | None = None + output: Any | None = None observations: list[LangfuseObservation] = field(default_factory=list[LangfuseObservation]) def find_observation(self, observation_id: str) -> LangfuseObservation | None: @@ -180,12 +185,16 @@ def update_trace( id: str, name: str | None = None, metadata: dict[str, Any] | None = None, + input: Any | None = None, + output: Any | None = None, ) -> None: """Update an existing Trace's mutable fields after creation. Used by the observer when the caller-supplied invocation - label (§8.6) lands later than the Trace's open call, or when - additional metadata becomes available mid-invocation. + label (§8.6) lands later than the Trace's open call, when + additional metadata becomes available mid-invocation, or + when the proposal 0043 invocation-boundary events populate + ``trace.input`` / ``trace.output``. """ ... @@ -358,6 +367,8 @@ def update_trace( id: str, name: str | None = None, metadata: dict[str, Any] | None = None, + input: Any | None = None, + output: Any | None = None, ) -> None: trace = self.traces.get(id) if trace is None: @@ -365,11 +376,19 @@ def update_trace( # happen under the observer's emission order but stays # defensive against re-ordered events. self.trace(id=id, name=name, metadata=metadata) - return + trace = self.traces[id] if name is not None: trace.name = name if metadata is not None: trace.metadata.update(metadata) + # Proposal 0043: input/output land on the Trace's headline + # fields, distinct from the metadata bag. None means "not + # supplied on this update call" — the existing value (if any) + # is preserved; explicit replacement requires a non-None. + if input is not None: + trace.input = input + if output is not None: + trace.output = output def span( self, diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index ea9ccedf..8dafe433 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -24,11 +24,16 @@ import json import uuid -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass, field from typing import Any, cast -from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent +from openarmature.graph.events import ( + InvocationCompletedEvent, + InvocationStartedEvent, + MetadataAugmentationEvent, + NodeEvent, +) from openarmature.observability.lineage import is_prefix_or_equal, is_strict_prefix from openarmature.observability.llm_event import LLM_NAMESPACE, LlmEventPayload @@ -224,6 +229,23 @@ class LangfuseObserver: each get their own Langfuse Trace. Same link mechanism on the fan-out node observation: each per-instance detached trace_id lands in the array. + - ``disable_state_payload``: default ``True`` per §8.4.1 *Trace + input/output sourcing* (proposal 0043). When ``True`` the + observer does NOT serialize ``initial_state`` / final state + directly onto ``trace.input`` / ``trace.output``; the minimal + stub applies unless ``trace_input_from_state`` / + ``trace_output_from_state`` overrides. When ``False`` the raw + state object is serialized to the Trace fields, subject to + ``payload_byte_cap`` truncation. Independent of + ``disable_llm_payload`` — the two payloads carry distinct + threat models (LLM-call transcript vs. application state). + - ``trace_input_from_state``: optional caller hook returning the + value to use as ``trace.input``. Called once per invocation at + the ``InvocationStartedEvent``. Returning ``None`` falls + through to the next lever (raw state when + ``disable_state_payload=False``, minimal stub otherwise). + - ``trace_output_from_state``: same shape for ``trace.output``, + called once per invocation at the ``InvocationCompletedEvent``. The observer reads the spec version from the package at construction time. Safe to share across concurrent invocations @@ -238,6 +260,10 @@ class LangfuseObserver: detached_subgraphs: frozenset[str] = field(default_factory=_empty_str_frozenset) detached_fan_outs: frozenset[str] = field(default_factory=_empty_str_frozenset) spec_version: str = field(default_factory=_read_spec_version) + # Proposal 0043 §8.4.1 *Trace input/output sourcing*. + disable_state_payload: bool = True + trace_input_from_state: Callable[[Any], Any] | None = None + trace_output_from_state: Callable[[Any], Any] | None = None # Internal state populated during invocation. _inv_states: dict[str, _InvState] = field(init=False, repr=False, default_factory=dict[str, _InvState]) @@ -252,7 +278,16 @@ def __post_init__(self) -> None: f"minimum of {_PAYLOAD_MIN_BYTES} bytes" ) - async def __call__(self, event: NodeEvent | MetadataAugmentationEvent) -> None: + async def __call__( + self, + event: (NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent), + ) -> None: + if isinstance(event, InvocationStartedEvent): + self._handle_invocation_started(event) + return + if isinstance(event, InvocationCompletedEvent): + self._handle_invocation_completed(event) + return if isinstance(event, MetadataAugmentationEvent): self._handle_metadata_augmentation(event) return @@ -460,6 +495,129 @@ def _handle_metadata_augmentation(self, event: MetadataAugmentationEvent) -> Non if is_prefix_or_equal(ns, aug_ns): observation.handle.update(metadata=metadata_delta) + # ------------------------------------------------------------------ + # Invocation-boundary events (proposal 0043 §8.4.1 sourcing) + # ------------------------------------------------------------------ + + 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. + 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) + self.client.update_trace(id=event.invocation_id, input=input_value) + + def _handle_invocation_completed(self, event: InvocationCompletedEvent) -> None: + # Spec proposal 0043 §8.4.1. Resolve ``trace.output`` via the + # same three-lever decision tree as input, with the minimal + # stub carrying {final_node, status}. + if event.invocation_id not in self._inv_states: + # Defensive: a fast-failure invocation may complete before + # any node event fired (e.g., resume-path validation + # rejected). Lazy-open the Trace so the stub still lands. + entry_node = event.final_node # best-effort fallback + self._open_trace_lazy(event.invocation_id, event.correlation_id, entry_node) + output_value = self._resolve_trace_output(event) + self.client.update_trace(id=event.invocation_id, output=output_value) + + def _resolve_trace_input(self, event: InvocationStartedEvent) -> Any: + # Lever 1: caller hook. + if self.trace_input_from_state is not None: + try: + hook_value = self.trace_input_from_state(event.initial_state) + except Exception: + # Hook raise: skip emission (defensive — caller code + # should not break observability). Fall through to the + # next lever rather than crash the observer. + hook_value = None + if hook_value is not None: + return self._maybe_truncate_for_extras(hook_value) + # Lever 2: raw state when knob is OFF. + if not self.disable_state_payload: + serialized = self._state_to_jsonable(event.initial_state) + return self._maybe_truncate_for_extras(serialized) + # Lever 3: minimal stub. + stub: dict[str, Any] = {"entry_node": event.entry_node} + if event.correlation_id is not None: + stub["correlation_id"] = event.correlation_id + return stub + + def _resolve_trace_output(self, event: InvocationCompletedEvent) -> Any: + # Lever 1: caller hook. + if self.trace_output_from_state is not None: + try: + hook_value = self.trace_output_from_state(event.final_state) + except Exception: + hook_value = None + if hook_value is not None: + return self._maybe_truncate_for_extras(hook_value) + # Lever 2: raw state when knob is OFF. + if not self.disable_state_payload: + serialized = self._state_to_jsonable(event.final_state) + return self._maybe_truncate_for_extras(serialized) + # Lever 3: minimal stub. + return {"final_node": event.final_node, "status": event.status} + + @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. + dumper = getattr(state, "model_dump", None) + if callable(dumper): + try: + return dumper(mode="json") + except Exception: + return str(state) + return str(state) + + def _open_trace_lazy( + self, + invocation_id: str, + correlation_id: str | None, + entry_node: str, + ) -> None: + # Open the Trace from a non-NodeEvent path (the proposal 0043 + # invocation-boundary events). The existing ``_open_trace`` + # entry point reads ``entry_node`` and caller metadata from a + # NodeEvent; this lazy path doesn't have one. Caller metadata + # is still readable via ``current_invocation_metadata`` — + # ``_apply_caller_metadata`` mirrors the existing path. + from openarmature.observability.metadata import current_invocation_metadata + + metadata: dict[str, Any] = { + "entry_node": entry_node, + "spec_version": self.spec_version, + } + if correlation_id is not None: + metadata["correlation_id"] = correlation_id + _apply_caller_metadata(metadata, current_invocation_metadata()) + self.client.trace(id=invocation_id, name=entry_node, metadata=metadata) + 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 diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index c598aa97..f44d98ea 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -96,7 +96,12 @@ ) from opentelemetry.trace.propagation import set_span_in_context -from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent +from openarmature.graph.events import ( + InvocationCompletedEvent, + InvocationStartedEvent, + MetadataAugmentationEvent, + NodeEvent, +) from openarmature.observability.lineage import is_prefix_or_equal, is_strict_prefix from openarmature.observability.llm_event import LLM_NAMESPACE, LlmEventPayload @@ -451,10 +456,23 @@ def _inv_state_for(self, invocation_id: str) -> _InvState: # ------------------------------------------------------------------ # Observer protocol — async callable accepting node events + the - # proposal-0040 metadata-augmentation event variant. + # proposal-0040 metadata-augmentation event variant + the + # proposal-0043 invocation-boundary events (no-op on the OTel + # mapping; OTel has no Trace-level input/output concept per the + # proposal's Out-of-Scope section). # ------------------------------------------------------------------ - async def __call__(self, event: NodeEvent | MetadataAugmentationEvent) -> None: + async def __call__( + self, + event: (NodeEvent | MetadataAugmentationEvent | InvocationStartedEvent | InvocationCompletedEvent), + ) -> None: + # Proposal 0043 invocation-boundary events: OTel has no + # Trace-level input/output payload concept (a trace is a + # collection of Spans sharing a trace_id; no Trace-level + # payload field). No-op gates here; isinstance early-return + # before any node-specific logic runs. + if isinstance(event, InvocationStartedEvent | InvocationCompletedEvent): + return if isinstance(event, MetadataAugmentationEvent): self._handle_metadata_augmentation(event) return diff --git a/tests/conformance/adapter.py b/tests/conformance/adapter.py index efc05c21..89a0d719 100644 --- a/tests/conformance/adapter.py +++ b/tests/conformance/adapter.py @@ -27,6 +27,8 @@ FanOutNode, FieldNameMatching, GraphBuilder, + NodeEvent, + ObserverEvent, ParallelBranchesNode, ProjectionStrategy, Reducer, @@ -38,7 +40,6 @@ merge, merge_all, ) -from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent from openarmature.graph.observer import Observer if TYPE_CHECKING: @@ -855,8 +856,8 @@ def make_observer_fn( the event unrecorded and the counter shows it as undelivered. """ - async def observer(event: NodeEvent | MetadataAugmentationEvent) -> None: - if isinstance(event, MetadataAugmentationEvent): + async def observer(event: ObserverEvent) -> None: + if not isinstance(event, NodeEvent): return sleep_ms = _resolve_sleep_ms(fixture) if sleep_ms > 0: diff --git a/tests/conformance/test_conformance.py b/tests/conformance/test_conformance.py index ef6962f5..ac9a4674 100644 --- a/tests/conformance/test_conformance.py +++ b/tests/conformance/test_conformance.py @@ -21,13 +21,14 @@ EdgeException, EndSentinel, GraphBuilder, + NodeEvent, NodeException, + ObserverEvent, RoutingError, RuntimeGraphError, State, SubscribedObserver, ) -from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent from openarmature.graph.observer import Observer from .adapter import ( @@ -605,8 +606,9 @@ class FixtureState(State): received: list[NodeEvent] = [] - async def observer(event: NodeEvent | MetadataAugmentationEvent) -> None: - assert isinstance(event, NodeEvent) + async def observer(event: ObserverEvent) -> None: + if not isinstance(event, NodeEvent): + return received.append(event) async def node_a(_state: Any) -> dict[str, Any]: diff --git a/tests/conformance/test_fixture_parsing.py b/tests/conformance/test_fixture_parsing.py index bbe3c3d8..b43ed1c1 100644 --- a/tests/conformance/test_fixture_parsing.py +++ b/tests/conformance/test_fixture_parsing.py @@ -215,6 +215,13 @@ def _id(case: tuple[str, Path]) -> str: "observability/036-caller-invocation-id-non-uuid": ( "Cross-capability parser doesn't model langfuse_trace; derivation pinned by unit tests" ), + # Proposal 0043 (trace.input/output) — fixture 037 uses langfuse_trace + # expected shape + hook-based directives the cross-capability parser + # doesn't model. Behavior pinned by unit tests at + # tests/unit/test_observability_langfuse.py::test_trace_input_output_*. + "observability/037-langfuse-trace-input-output": ( + "Cross-capability parser doesn't model langfuse_trace; behavior pinned by unit tests" + ), } diff --git a/tests/conformance/test_observability_langfuse.py b/tests/conformance/test_observability_langfuse.py index aed7b98b..3c29bd66 100644 --- a/tests/conformance/test_observability_langfuse.py +++ b/tests/conformance/test_observability_langfuse.py @@ -74,6 +74,16 @@ # before the LLM call, exercising the §3.4 MUST that open # spans in the augmenting context's lineage update in place. "034-caller-metadata-open-span-update-serial", + # 037 stays deferred in v0.11.0: the conformance fixture + # exercises hook-based cases (caller-supplied callables) that + # the YAML-only fixture format can't express directly without + # a harness extension. The five-case decision tree + # (default stub / disable_state_payload=False / hooks + # non-null / hooks null-fallthrough / resume) is verified + # end-to-end by the unit tests in + # ``tests/unit/test_observability_langfuse.py::test_trace_input_output_*``. + # Wiring fixture 037 lands when the harness grows directive + # support for caller-supplied hook returns. # 029 + 030 stay deferred in v0.11.0: # - 029 (fan-out per-instance): fixture omits ``collect_field`` # and ``target_field`` on the fan_out cfg, plus the inner diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 5ba2c5d6..a977c7a4 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -9,7 +9,7 @@ def test_package_versions() -> None: assert openarmature.__version__ == "0.10.0" - assert openarmature.__spec_version__ == "0.34.0" + assert openarmature.__spec_version__ == "0.35.0" def test_spec_version_matches_pyproject() -> None: diff --git a/tests/unit/test_drain.py b/tests/unit/test_drain.py index 49288d77..a41c172e 100644 --- a/tests/unit/test_drain.py +++ b/tests/unit/test_drain.py @@ -24,9 +24,10 @@ CompiledGraph, DrainSummary, GraphBuilder, + NodeEvent, + ObserverEvent, State, ) -from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent class _S(State): @@ -75,11 +76,15 @@ async def test_drain_without_timeout_waits_for_all_events() -> None: # DrainSummary with the consistent shape. received: list[str] = [] - async def slow_obs(event: NodeEvent | MetadataAugmentationEvent) -> None: - # ~50ms per event; the 3-node graph fires 6 events - # (3 nodes × started + completed) so ~300ms of work total. + async def slow_obs(event: ObserverEvent) -> None: + # ~50ms per event; the 3-node graph fires 8 events + # (3 nodes × started + completed = 6 NodeEvents, plus + # InvocationStarted + InvocationCompleted from proposal + # 0043). The observer only counts NodeEvents — boundary + # events early-return. await asyncio.sleep(0.05) - assert isinstance(event, NodeEvent) + if not isinstance(event, NodeEvent): + return received.append(event.node_name) compiled = _build_compiled() @@ -91,11 +96,13 @@ async def slow_obs(event: NodeEvent | MetadataAugmentationEvent) -> None: assert summary.timeout_reached is False assert summary.undelivered_count == 0 - # All 6 events delivered. + # 6 NodeEvents reach the receiver (the two boundary events early- + # return inside the observer, but they're still counted as + # delivered by the drain summary). assert len(received) == 6 - # Drain blocked for roughly the observer's total work (~300ms); - # allow generous slack for scheduler / CI variance. - assert elapsed >= 0.25 + # Drain blocked for the observer's total work — 8 events × 50ms. + # Allow generous slack for scheduler / CI variance. + assert elapsed >= 0.35 async def test_drain_with_timeout_not_reached_for_fast_observers() -> None: @@ -103,8 +110,9 @@ async def test_drain_with_timeout_not_reached_for_fast_observers() -> None: # fast. Summary reports clean delivery. received: list[str] = [] - async def fast_obs(event: NodeEvent | MetadataAugmentationEvent) -> None: - assert isinstance(event, NodeEvent) + async def fast_obs(event: ObserverEvent) -> None: + if not isinstance(event, NodeEvent): + return received.append(event.node_name) compiled = _build_compiled() @@ -123,11 +131,12 @@ async def test_drain_with_timeout_fires_reports_undelivered() -> None: # generous slack for cancellation settlement). received: list[str] = [] - async def slow_obs(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def slow_obs(event: ObserverEvent) -> None: # 200ms per event vs 100ms drain timeout; at most 0-1 events # complete before the deadline fires. await asyncio.sleep(0.2) - assert isinstance(event, NodeEvent) + if not isinstance(event, NodeEvent): + return received.append(event.node_name) compiled = _build_compiled() @@ -138,7 +147,9 @@ async def slow_obs(event: NodeEvent | MetadataAugmentationEvent) -> None: elapsed = time.monotonic() - started assert summary.timeout_reached is True - assert summary.undelivered_count >= 4 + # 6 NodeEvents + 2 boundary events = 8 enqueued; at most 0-1 + # deliver before the 100ms deadline. + assert summary.undelivered_count >= 6 # The hard deadline is non-negotiable. Allow ~250ms of slack for # cancellation settling + CI scheduler variance — the dispatched # event's await still resolves under cancellation, and the @@ -156,16 +167,17 @@ async def test_drain_after_timeout_leaves_graph_usable() -> None: call_count = [0] received_invocation_two: list[str] = [] - async def obs(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs(event: ObserverEvent) -> None: # First invocation: slow enough to force the timeout to # fire. Second invocation: fast, so drain completes cleanly. # The mode is controlled by `call_count[0]`: we bump it # between invocations. if call_count[0] == 0: await asyncio.sleep(0.1) - else: - assert isinstance(event, NodeEvent) - received_invocation_two.append(event.node_name) + return + if not isinstance(event, NodeEvent): + return + received_invocation_two.append(event.node_name) compiled = _build_compiled() compiled.attach_observer(obs) @@ -205,9 +217,10 @@ async def test_drain_with_zero_timeout_fires_immediately() -> None: # immediately with whatever the worker hasn't gotten to yet. received: list[str] = [] - async def obs(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs(event: ObserverEvent) -> None: await asyncio.sleep(0.05) - assert isinstance(event, NodeEvent) + if not isinstance(event, NodeEvent): + return received.append(event.node_name) compiled = _build_compiled() @@ -216,9 +229,10 @@ async def obs(event: NodeEvent | MetadataAugmentationEvent) -> None: summary = await compiled.drain(timeout=0.0) assert summary.timeout_reached is True - # All 6 events are still in flight or queued — none delivered - # before the zero-second deadline fired. - assert summary.undelivered_count == 6 + # All 8 events (6 NodeEvents + 2 boundary events) are still in + # flight or queued — none delivered before the zero-second + # deadline fired. + assert summary.undelivered_count == 8 assert len(received) == 0 diff --git a/tests/unit/test_observability_langfuse.py b/tests/unit/test_observability_langfuse.py index e20c885b..6b11fe20 100644 --- a/tests/unit/test_observability_langfuse.py +++ b/tests/unit/test_observability_langfuse.py @@ -579,3 +579,188 @@ async def test_metadata_augmentation_no_op_when_no_entries() -> None: observer._handle_metadata_augmentation(event) # noqa: SLF001 # No Trace was opened (no invocation in scope) and no exception. assert client.traces == {} + + +# --------------------------------------------------------------------------- +# Trace input/output sourcing (proposal 0043 §8.4.1) +# --------------------------------------------------------------------------- + + +class _S0043(State): + msg: str = "" + + +async def _emit_node(_s: _S0043) -> dict[str, Any]: + return {"msg": "ok"} + + +def _build_0043_graph() -> Any: + return GraphBuilder(_S0043).add_node("a", _emit_node).add_edge("a", END).set_entry("a").compile() + + +async def test_trace_input_output_default_emits_minimal_stub() -> None: + # Lever 3 (default). `disable_state_payload` defaults ON; no hooks + # supplied. trace.input = {entry_node, correlation_id}; + # trace.output = {final_node, status}. + client = InMemoryLangfuseClient() + observer = LangfuseObserver(client=client) + graph = _build_0043_graph() + graph.attach_observer(observer) + await graph.invoke(_S0043(), correlation_id="corr-1") + await graph.drain() + + trace = next(iter(client.traces.values())) + assert trace.input == {"entry_node": "a", "correlation_id": "corr-1"} + assert trace.output == {"final_node": "a", "status": "completed"} + + +async def test_trace_input_output_disable_state_payload_off_emits_raw_state() -> None: + # Lever 2. `disable_state_payload=False`; no hooks. trace.input + # and trace.output carry the serialized state. + client = InMemoryLangfuseClient() + observer = LangfuseObserver(client=client, disable_state_payload=False) + graph = _build_0043_graph() + graph.attach_observer(observer) + await graph.invoke(_S0043()) + await graph.drain() + + trace = next(iter(client.traces.values())) + # ``input`` reflects initial_state, ``output`` reflects final state. + assert trace.input == {"msg": ""} + assert trace.output == {"msg": "ok"} + + +async def test_trace_input_output_handles_non_json_native_state_fields() -> None: + # Regression for the PR #99 copilot finding: pydantic's + # ``model_dump()`` defaults to Python mode and leaves + # ``datetime`` / ``UUID`` / ``Decimal`` as Python objects. The + # downstream truncation path calls ``json.dumps`` without a + # ``default``, which raises ``TypeError`` on those types. The + # observer raise is swallowed by the engine's warnings-only + # observer-isolation contract, leaving trace.input / trace.output + # silently blank. + # + # ``_state_to_jsonable`` MUST call ``model_dump(mode="json")`` so + # these types serialize to their JSON-compatible string forms + # before the truncation step. + import uuid + from datetime import UTC, datetime + from decimal import Decimal + + class _DateState(State): + when: datetime = datetime(2026, 5, 29, 12, 0, 0, tzinfo=UTC) + request_id: uuid.UUID = uuid.UUID("12345678-1234-5678-1234-567812345678") + amount: Decimal = Decimal("99.99") + + async def _noop(_s: _DateState) -> dict[str, Any]: + return {} + + client = InMemoryLangfuseClient() + observer = LangfuseObserver(client=client, disable_state_payload=False) + graph = GraphBuilder(_DateState).add_node("a", _noop).add_edge("a", END).set_entry("a").compile() + graph.attach_observer(observer) + await graph.invoke(_DateState()) + await graph.drain() + + trace = next(iter(client.traces.values())) + # The non-JSON-native types serialize to JSON-compatible strings. + # Both trace.input and trace.output land successfully (the bug + # would leave them as ``None``). + assert trace.input is not None, "trace.input should not be blank on State with datetime/UUID/Decimal" + assert trace.output is not None + trace_input = cast("dict[str, Any]", trace.input) + assert trace_input["when"] == "2026-05-29T12:00:00Z" + assert trace_input["request_id"] == "12345678-1234-5678-1234-567812345678" + # Decimal serializes to its string form under ``mode="json"``. + assert trace_input["amount"] == "99.99" + + +async def test_trace_input_output_caller_hooks_replace_stub() -> None: + # Lever 1. Caller hooks supplied, returning non-None domain + # summaries. Hook return values appear on the trace fields verbatim; + # the stub does NOT appear; `disable_state_payload` is irrelevant. + client = InMemoryLangfuseClient() + + def input_hook(state: _S0043) -> dict[str, Any]: + return {"summary": f"received msg={state.msg!r}"} + + def output_hook(state: _S0043) -> dict[str, Any]: + return {"summary": f"final msg={state.msg!r}"} + + observer = LangfuseObserver( + client=client, + trace_input_from_state=input_hook, + trace_output_from_state=output_hook, + ) + graph = _build_0043_graph() + graph.attach_observer(observer) + await graph.invoke(_S0043()) + await graph.drain() + + trace = next(iter(client.traces.values())) + assert trace.input == {"summary": "received msg=''"} + assert trace.output == {"summary": "final msg='ok'"} + + +async def test_trace_input_output_caller_hooks_return_none_falls_through() -> None: + # Lever-1 null-fallthrough. Hooks supplied but return None; + # observer falls through to the next applicable lever — lever 3 + # (stub) when disable_state_payload defaults ON. + client = InMemoryLangfuseClient() + + def input_hook(_state: _S0043) -> None: + return None + + def output_hook(_state: _S0043) -> None: + return None + + observer = LangfuseObserver( + client=client, + trace_input_from_state=input_hook, + trace_output_from_state=output_hook, + ) + graph = _build_0043_graph() + graph.attach_observer(observer) + await graph.invoke(_S0043(), correlation_id="corr-2") + await graph.drain() + + trace = next(iter(client.traces.values())) + # Stub applies as if no hook had been supplied. + assert trace.input == {"entry_node": "a", "correlation_id": "corr-2"} + assert trace.output == {"final_node": "a", "status": "completed"} + + +class _FailState(State): + x: int = 0 + + +async def _raise_node(_s: _FailState) -> dict[str, Any]: + raise RuntimeError("boom") + + +async def test_trace_output_status_failed_on_node_raise() -> None: + # Failure path: `status` enum closed on {completed, failed}. A + # raise inside the node body fires the InvocationCompletedEvent + # with status="failed" and final_node set to the raising node. + client = InMemoryLangfuseClient() + observer = LangfuseObserver(client=client) + + graph = ( + GraphBuilder(_FailState) + .add_node("raises", _raise_node) + .add_edge("raises", END) + .set_entry("raises") + .compile() + ) + graph.attach_observer(observer) + + # Spec §4: node-raised exceptions surface as NodeException + # (the runtime category that wraps node body raises). + from openarmature.graph.errors import NodeException + + with pytest.raises(NodeException, match="raises"): + await graph.invoke(_FailState()) + await graph.drain() + + trace = next(iter(client.traces.values())) + assert trace.output == {"final_node": "raises", "status": "failed"} diff --git a/tests/unit/test_observer.py b/tests/unit/test_observer.py index d37946e1..99bdc056 100644 --- a/tests/unit/test_observer.py +++ b/tests/unit/test_observer.py @@ -12,8 +12,14 @@ from types import MappingProxyType from typing import Literal -from openarmature.graph import Observer, State, SubscribedObserver -from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent +from openarmature.graph import ( + MetadataAugmentationEvent, + NodeEvent, + Observer, + ObserverEvent, + State, + SubscribedObserver, +) from openarmature.graph.observer import ( _DRAIN_SENTINEL, RemoveHandle, @@ -64,7 +70,7 @@ async def _drain(queue: asyncio.Queue[_QueuedItem | None], worker: asyncio.Task[ async def test_events_delivered_in_queue_order() -> None: received: list[str] = [] - async def observer(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def observer(event: ObserverEvent) -> None: assert isinstance(event, NodeEvent) received.append(event.node_name) @@ -81,11 +87,11 @@ async def observer(event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_multiple_observers_fire_in_registration_order() -> None: received: list[str] = [] - async def obs1(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs1(event: ObserverEvent) -> None: assert isinstance(event, NodeEvent) received.append(f"obs1:{event.node_name}") - async def obs2(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs2(event: ObserverEvent) -> None: assert isinstance(event, NodeEvent) received.append(f"obs2:{event.node_name}") @@ -106,7 +112,7 @@ async def obs2(event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_observer_exception_does_not_propagate_to_caller() -> None: - async def boom(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def boom(_event: ObserverEvent) -> None: raise RuntimeError("nope") queue: asyncio.Queue[_QueuedItem | None] = asyncio.Queue() @@ -124,10 +130,10 @@ async def boom(_event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_raising_observer_does_not_block_siblings_on_same_event() -> None: received: list[str] = [] - async def obs1(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs1(_event: ObserverEvent) -> None: raise RuntimeError("obs1 boom") - async def obs2(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs2(event: ObserverEvent) -> None: assert isinstance(event, NodeEvent) received.append(event.node_name) @@ -145,10 +151,10 @@ async def obs2(event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_raising_observer_does_not_block_subsequent_events() -> None: received: list[str] = [] - async def always_raises(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def always_raises(_event: ObserverEvent) -> None: raise RuntimeError("always boom") - async def silent(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def silent(event: ObserverEvent) -> None: assert isinstance(event, NodeEvent) received.append(event.node_name) @@ -171,7 +177,7 @@ async def silent(event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_phase_filter_skips_unsubscribed_phase() -> None: received: list[tuple[str, str]] = [] - async def obs(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs(event: ObserverEvent) -> None: assert isinstance(event, NodeEvent) received.append((event.node_name, event.phase)) @@ -188,7 +194,7 @@ async def obs(event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_subscribed_observer_rejects_empty_phases() -> None: - async def obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs(_event: ObserverEvent) -> None: pass try: @@ -199,7 +205,7 @@ async def obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_subscribed_observer_rejects_unknown_phase() -> None: - async def obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs(_event: ObserverEvent) -> None: pass try: @@ -215,7 +221,7 @@ async def obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_sentinel_terminates_worker_after_processing_queued_events() -> None: received: list[str] = [] - async def observer(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def observer(event: ObserverEvent) -> None: assert isinstance(event, NodeEvent) received.append(event.node_name) @@ -244,10 +250,10 @@ async def test_dispatch_skips_when_no_observers_for_depth() -> None: async def test_dispatch_enqueues_with_full_observer_chain_in_order() -> None: - async def graph_obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def graph_obs(_event: ObserverEvent) -> None: pass - async def invocation_obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def invocation_obs(_event: ObserverEvent) -> None: pass graph_subscribed = _wrap(graph_obs) @@ -272,13 +278,13 @@ async def invocation_obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: async def test_descend_extends_chain_namespace_and_parent_states() -> None: - async def outer_obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def outer_obs(_event: ObserverEvent) -> None: pass - async def sub_obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def sub_obs(_event: ObserverEvent) -> None: pass - async def invocation_obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def invocation_obs(_event: ObserverEvent) -> None: pass outer_subscribed = _wrap(outer_obs) @@ -324,7 +330,7 @@ async def test_take_step_shares_counter_across_descended_contexts() -> None: def test_remove_handle_detaches_observer() -> None: - async def obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs(_event: ObserverEvent) -> None: pass subscribed = _wrap(obs) @@ -337,7 +343,7 @@ async def obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: def test_remove_handle_is_idempotent() -> None: - async def obs(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def obs(_event: ObserverEvent) -> None: pass subscribed = _wrap(obs) @@ -361,10 +367,10 @@ async def test_metadata_augmentation_event_bypasses_phase_filter() -> None: augment_received: list[MetadataAugmentationEvent] = [] node_received: list[NodeEvent] = [] - async def observer(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def observer(event: ObserverEvent) -> None: if isinstance(event, MetadataAugmentationEvent): augment_received.append(event) - else: + elif isinstance(event, NodeEvent): node_received.append(event) queue: asyncio.Queue[_QueuedItem | None] = asyncio.Queue() @@ -392,10 +398,10 @@ async def test_metadata_augmentation_observer_exception_is_isolated() -> None: still run, the worker keeps draining.""" sibling_received: list[MetadataAugmentationEvent] = [] - async def boom(_event: NodeEvent | MetadataAugmentationEvent) -> None: + async def boom(_event: ObserverEvent) -> None: raise RuntimeError("boom") - async def good(event: NodeEvent | MetadataAugmentationEvent) -> None: + async def good(event: ObserverEvent) -> None: if isinstance(event, MetadataAugmentationEvent): sibling_received.append(event) @@ -433,9 +439,9 @@ async def test_set_invocation_metadata_emits_augmentation_event_via_dispatch() - _set_namespace_prefix, ) - captured: list[NodeEvent | MetadataAugmentationEvent] = [] + captured: list[ObserverEvent] = [] - def dispatch(event: NodeEvent | MetadataAugmentationEvent) -> None: + def dispatch(event: ObserverEvent) -> None: captured.append(event) dispatch_token = _set_active_dispatch(dispatch) diff --git a/tests/unit/test_runtime_errors.py b/tests/unit/test_runtime_errors.py index e08c14ec..51d7629e 100644 --- a/tests/unit/test_runtime_errors.py +++ b/tests/unit/test_runtime_errors.py @@ -162,13 +162,17 @@ async def test_routing_error_lands_on_preceding_node_completed_event() -> None: conditional edge that returns an undeclared target lands on the preceding node's `completed` event with `error` populated, NOT in a separate event pair. The downstream node never fires events.""" - from openarmature.graph import RoutingError - from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent + from openarmature.graph import ( + NodeEvent, + ObserverEvent, + RoutingError, + ) received: list[NodeEvent] = [] - async def observer(event: NodeEvent | MetadataAugmentationEvent) -> None: - assert isinstance(event, NodeEvent) + async def observer(event: ObserverEvent) -> None: + if not isinstance(event, NodeEvent): + return received.append(event) async def node_a(_state: Any) -> dict[str, Any]: @@ -218,12 +222,16 @@ async def test_edge_exception_lands_on_preceding_node_completed_event() -> None: conditional edge function raising lands on the preceding node's `completed` event with `error` populated, NOT in a separate event pair. The downstream node never fires events.""" - from openarmature.graph.events import MetadataAugmentationEvent, NodeEvent + from openarmature.graph import ( + NodeEvent, + ObserverEvent, + ) received: list[NodeEvent] = [] - async def observer(event: NodeEvent | MetadataAugmentationEvent) -> None: - assert isinstance(event, NodeEvent) + async def observer(event: ObserverEvent) -> None: + if not isinstance(event, NodeEvent): + return received.append(event) async def node_a(_state: Any) -> dict[str, Any]: