Implement proposal 0043 (trace input/output) (#99)

* Implement proposal 0043 (trace input/output)

Adds the observability §8.4.1 *Trace input/output sourcing* mechanism:
the Langfuse observer populates `trace.input` at invocation entry
and `trace.output` at invocation exit via a three-lever decision
tree (caller hook returning non-null → hook value; raw state when
`disable_state_payload=False` → serialized state; default → minimal
stub `{entry_node, correlation_id}` / `{final_node, status}` where
status is the closed `Literal["completed", "failed"]` enum).

Wires through two new observer event types delivered on the
existing serial-delivery queue: `InvocationStartedEvent` and
`InvocationCompletedEvent`. The engine enqueues both at the
invocation lifecycle's outermost boundaries (entry before any node
fires; exit on both success and failure paths). Mirrors the 0040
pattern used for `MetadataAugmentationEvent`. The `Observer.__call__`
signature widens to a four-variant union; the new `ObserverEvent`
type alias gives observer authors a one-name handle and is
re-exported from `openarmature.graph`.

The OTel observer no-ops on both new events (OTel has no Trace-
level input/output concept). The LangfuseSDKAdapter caches input
and output on `_trace_info`; live-Trace emission via the v4 SDK is
deferred to a follow-up (the InMemoryLangfuseClient used by tests
applies the fields directly so the contract is unit-test-pinned).

Bumps the spec pin from v0.34.0 to v0.35.0. `conformance.toml`
records 0043 as implemented since 0.11.0. Conformance fixture 037
is deferred because cases 3/4/5 need a caller-hook YAML directive
the cross-capability harness doesn't model yet; the five-case
decision tree is pinned by new unit tests at
`tests/unit/test_observability_langfuse.py::test_trace_input_output_*`.

* Fix final_node_box leak + JSON-mode state dump

Two issues surfaced by PR #99 review:

`final_node_box` is shared by reference across subgraph, fan-out,
and parallel-branches descents (`descend_into_*` propagates the
list). Inner-node writes leak into the outer box on the success
path, so the outermost `invoke()` reads the wrong `final_node`
when an outer wrapper is the last node before the END-routing
edge. For parallel-branches the leaked value depends on which
branch finishes last, making `InvocationCompletedEvent.final_node`
nondeterministic.

Restore the outer `current` to the box after each `_step_*` call
returns successfully. The restore is on the success path only —
the failure path's raise bypasses it, so the inner-most node that
raised stays in the box for the spec §4 attribution. A follow-up
race remains for parallel-branches and fan-out failure cases:
concurrent inner writes mean the box may end with a successful
sibling's inner rather than the failing sibling's. Addressing that
requires error-aware tracking the engine doesn't currently expose.

Pydantic's `model_dump()` defaults to Python mode and leaves
`datetime` / `UUID` / `Decimal` as Python objects. The downstream
`json.dumps` truncation path raises `TypeError` on those types,
and the observer raise is swallowed by the engine's warnings-only
observer-isolation contract, silently leaving `trace.input` /
`trace.output` blank under `disable_state_payload=False`.

`_state_to_jsonable` now calls `model_dump(mode="json")` so the
common non-JSON-native types serialize to their JSON-compatible
string forms before reaching the truncation step. Adds a
regression test using a State with `datetime`, `UUID`, and
`Decimal` fields.

Author: chris-colinsky

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
 

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"

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

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:

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)

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.
 

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

openarmature-spec

@@ -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 = [

pyproject.toml

@@ -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`