Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,16 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
- **`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.

### Fixed

- **`InvocationCompletedEvent.final_state` on the failure path now surfaces the partial state at failure point.** Spec §8.4.1 *Resume semantics* requires the failure-path `trace.output` hook to receive "the partial final state captured at the failure point"; the original PR #99 implementation defaulted to `starting_state`, so the hook saw pre-execution state when it should have seen post-execution-up-to-failure state. The engine now tracks the latest post-merge state via a `latest_state_box` on `_InvocationContext`, updated after every successful step and read on the failure path. Success-path behavior unchanged.
- **`latest_state_box` is per-context, not shared across subgraph descents.** Unlike the sibling `final_node_box` (which shares by reference because the spec wants the innermost failing node's name — the real culprit), `latest_state_box` must isolate per level so the outermost Langfuse trace receives outer-state-typed values. Without the isolation, a subgraph-internal step's inner-typed state would leak up to the outer trace.output hook, breaking the hook's typed contract. Each subgraph / fan-out instance / parallel-branches branch gets its own fresh box. Pinned by four regression tests covering flat, subgraph, fan-out, and parallel-branches failure paths.

### Notes

- **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).
- `LangfuseSDKAdapter` now applies `trace.input` / `trace.output` to the live Langfuse Trace. Input lands on the first real observation under the trace via `set_trace_io`; output uses a synthetic short-lived `openarmature.trace_io` observation as the carrier. The InMemoryLangfuseClient used by tests applies the fields directly.
- Conformance fixture `observability/conformance/037-langfuse-trace-input-output` activated for the four decision-tree cases (default stub / `disable_state_payload=False` / hooks non-null / hooks null-fallthrough). Case 5 (resume re-fire) is deferred to a follow-up — needs the langfuse harness to grow checkpointer wiring + flaky-node test seam + two-phase multi-trace assertion.
- Conformance fixture `observability/conformance/037-langfuse-trace-input-output` activated for all five cases (default stub / `disable_state_payload=False` / hooks non-null / hooks null-fallthrough / resume re-fire). The langfuse harness grew per-case `checkpointer: in_memory` wiring, a compact `flaky:` test seam, and a two-phase resume-flow assertion path.
- The Langfuse v4 SDK marks `set_current_trace_io` / `Span.set_trace_io` deprecated ("removal in a future major version"). Empirical verification against Langfuse Cloud (v4.7.1, 2026-05-29) confirms it remains the **only** path that populates the Traces list view's headline `Input` / `Output` columns; `propagate_attributes(metadata=...)` does not substitute for it in the current UI. We will revisit when Langfuse publishes a concrete migration guide for v5.

## [0.10.0] — 2026-05-27
Expand Down
35 changes: 31 additions & 4 deletions src/openarmature/graph/compiled.py
Original file line number Diff line number Diff line change
Expand Up @@ -1059,20 +1059,38 @@ async def invoke(
# 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).
# edge. ``final_state`` precedence: the engine's returned
# state on success → the most recent successful step's
# post-merge state on a mid-graph raise (per §8.4.1
# *Resume semantics* "partial final state captured at the
# failure point") → ``starting_state`` only when no step
# ever completed.
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
# ``latest_state_box`` is typed ``list[Any]`` on
# _InvocationContext (the context isn't parameterized on
# StateT), but at the outermost level (where this code
# runs) it always holds an outer ``StateT`` from a
# successful step's post-merge state. Cast for type
# narrowing; the per-context box-isolation pinned by
# ``test_failure_path_final_state_is_outer_type_*`` keeps
# this invariant honest.
event_final_state: StateT
if final_state is not None:
event_final_state = final_state
elif context.latest_state_box:
event_final_state = cast("StateT", context.latest_state_box[0])
else:
event_final_state = starting_state
_dispatch(
context,
InvocationCompletedEvent(
final_state=final_state if final_state is not None else starting_state,
final_state=event_final_state,
status=status,
final_node=final_node,
invocation_id=invocation_id,
Expand Down Expand Up @@ -1201,6 +1219,15 @@ async def _invoke(
else:
step_result = await self._step_function_node(node, current, state, context)
state = step_result.state
# Proposal 0043 (post-PR-99 review): surface the most
# recent successful step's post-merge state so the
# outermost ``invoke()`` can populate
# ``InvocationCompletedEvent.final_state`` on the failure
# path with the partial state, not the bare initial state.
# Updated AFTER ``state = step_result.state`` so an
# exception inside the step bypasses this assignment and
# the previous value (or the empty box) survives.
context.latest_state_box[:] = [state]

# Proposal 0043 (post-PR-99 review): restore the outer
# ``current`` to the shared box after a successful step.
Expand Down
26 changes: 26 additions & 0 deletions src/openarmature/graph/observer.py
Original file line number Diff line number Diff line change
Expand Up @@ -497,6 +497,23 @@ 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.
latest_state_box: list[Any] = field(default_factory=list[Any])

def full_observers(self) -> tuple[SubscribedObserver, ...]:
"""Return the ordered observer list to deliver for events from
Expand Down Expand Up @@ -545,6 +562,9 @@ def descend_into_subgraph(
drain_counters=self.drain_counters,
state_cls=self.state_cls,
final_node_box=self.final_node_box,
# latest_state_box is INTENTIONALLY NOT propagated — each
# context level tracks its own outer-state-typed latest
# successful step. See the field docstring above.
)

def descend_into_fan_out_instance(
Expand Down Expand Up @@ -596,6 +616,9 @@ def descend_into_fan_out_instance(
drain_counters=self.drain_counters,
state_cls=self.state_cls,
final_node_box=self.final_node_box,
# latest_state_box is INTENTIONALLY NOT propagated — each
# context level tracks its own outer-state-typed latest
# successful step. See the field docstring above.
)

def descend_into_parallel_branch(
Expand Down Expand Up @@ -650,6 +673,9 @@ def descend_into_parallel_branch(
drain_counters=self.drain_counters,
state_cls=self.state_cls,
final_node_box=self.final_node_box,
# latest_state_box is INTENTIONALLY NOT propagated — each
# context level tracks its own outer-state-typed latest
# successful step. See the field docstring above.
)

def take_step(self) -> int:
Expand Down
Loading