diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py index a2e35a5..0fdb371 100644 --- a/src/openarmature/graph/events.py +++ b/src/openarmature/graph/events.py @@ -709,7 +709,7 @@ class LlmRetryAttemptEvent: ``finish_reason`` / ``output_content`` / ``output_tool_calls``): populated on a successful attempt; ``None`` / empty list on a failed attempt. ``output_tool_calls`` is the source the OTel - observer renders the §5.5.1 / §5.5.10 output tool-call attributes + observer renders the output tool-call attributes from (this is the per-attempt event that drives the LLM span). - failure side (``error_category`` / ``error_message`` / ``error_type``): populated on a failed attempt; ``None`` on a @@ -745,8 +745,9 @@ class LlmRetryAttemptEvent: # Proposal 0076: the attempt's output tool calls (ToolCall records), # mirroring LlmCompletionEvent.output_tool_calls. Populated on a # successful attempt; empty list on a failed one (no response). The - # OTel observer renders the output tool-call span attributes from - # this field (the per-attempt event is the LLM-span source). + # OTel observer renders the observability §5.5.1 / §5.5.10 output + # tool-call span attributes from this field (the per-attempt event is + # the LLM-span source). output_tool_calls: list["ToolCall"] = field(default_factory=list["ToolCall"]) diff --git a/src/openarmature/graph/parallel_branches.py b/src/openarmature/graph/parallel_branches.py index 7f0a4c3..891f85b 100644 --- a/src/openarmature/graph/parallel_branches.py +++ b/src/openarmature/graph/parallel_branches.py @@ -139,10 +139,12 @@ async def run(self, state: ParentT) -> Mapping[str, Any]: "compiled.invoke)." ) + # Spec pipeline-utilities §11.10: a branch dispatches when its ``when`` + # predicate is absent or returns true. def dispatched_branches(self, state: Any) -> list[tuple[str, BranchSpec[Any]]]: """Return the branches that dispatch for ``state``, in insertion order: every declared branch whose ``when`` predicate is absent or - returns true (§11.10). + returns true. ``when`` MUST be a pure function of the dispatch-time parent state, so this is deterministic and safe to evaluate both here (the @@ -288,6 +290,11 @@ async def innermost(s: Any) -> Mapping[str, Any]: return await self._fail_fast(state, tasks, contributions) return await self._collect(state, tasks, contributions, errors) + # Spec pipeline-utilities §11.4 (the buffer-then-project subgraph path is + # bypassed for inline callables) + §11.7 (per-leg branch middleware). The + # callable emits its own graph-engine §6 started/completed pair keyed by + # ``branch_name``, rendered per observability §5.7 as a per-branch dispatch + # span with no inner-node span beneath it. async def _dispatch_callable_branch( self, branch_name: str, @@ -300,14 +307,14 @@ async def _dispatch_callable_branch( The callable reads the parent state and returns a parent-shaped partial update directly — no subgraph, no ``inputs`` / ``outputs`` - projection (§11.4). Branch ``middleware`` (e.g. a per-leg - ``FailureIsolationMiddleware``, §11.7) wraps the callable. + projection. Branch ``middleware`` (e.g. a per-leg + ``FailureIsolationMiddleware``) wraps the callable. A callable branch has no inner nodes, so it IS the unit of work: it emits one ``started`` / ``completed`` observer pair keyed by its - ``branch_name`` (graph-engine §6), which the observers render as the - branch's per-branch dispatch span with NO inner-node span beneath it - (observability §5.7). To render that way the pair is emitted at the + ``branch_name``, which the observers render as the + branch's per-branch dispatch span with NO inner-node span beneath it. + To render that way the pair is emitted at the parallel-branches NODE's own namespace (not a descended branch namespace), tagged with ``branch_name`` (set on the ContextVar in ``run_branch``): an event at a pb-node's own namespace carrying a diff --git a/src/openarmature/observability/langfuse/client.py b/src/openarmature/observability/langfuse/client.py index 523e135..d05c43d 100644 --- a/src/openarmature/observability/langfuse/client.py +++ b/src/openarmature/observability/langfuse/client.py @@ -185,11 +185,13 @@ def trace( Implementations track Traces internally; observation calls pass `trace_id` to associate. - `session_id` / `user_id` (proposal 0064 §8.4.1) populate + `session_id` / `user_id` populate Langfuse's cross-trace grouping fields (the Sessions / Users dashboards); each is unset when its source is absent. """ # Spec §8.4.1: the Trace id is the OA invocation_id verbatim. + # session_id / user_id (proposal 0064) populate Langfuse's + # cross-trace grouping (the Sessions / Users dashboards). ... # The current observer doesn't invoke this method — it sets the @@ -248,6 +250,8 @@ def generation( start_time: datetime | None = None, ) -> LangfuseGenerationHandle: ... + # Spec proposal 0063: dedicated Tool observation (Langfuse asType="tool"), + # not a Generation with a discriminator. def tool( self, *, @@ -261,7 +265,7 @@ def tool( output: Any = None, start_time: datetime | None = None, ) -> LangfuseSpanHandle: - """Open a dedicated Tool observation (proposal 0063, Langfuse + """Open a dedicated Tool observation (Langfuse ``asType="tool"``). Like :meth:`generation` minus the model / usage / prompt surface: a Tool carries ``input`` (arguments) / ``output`` (result) / metadata / level. Returns a minimal diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 06db3ab..cbfd063 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -1821,9 +1821,10 @@ def _handle_typed_llm_failed(self, event: LlmFailedEvent) -> None: status_message=event.error_category, ) + # Spec proposal 0063: dedicated Langfuse Tool observation (asType="tool"). def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: """Open + close a dedicated Tool observation (Langfuse - ``asType="tool"``, proposal 0063) under the calling node's Span + ``asType="tool"``) under the calling node's Span observation. DEFAULT level on a ToolCallEvent; ERROR (with ``error_type`` / ``error_message`` in metadata and as the status message) on a ToolCallFailedEvent. ``input`` (arguments) / diff --git a/src/openarmature/observability/llm_event.py b/src/openarmature/observability/llm_event.py index 7e29a38..ef64139 100644 --- a/src/openarmature/observability/llm_event.py +++ b/src/openarmature/observability/llm_event.py @@ -147,9 +147,9 @@ class LlmEventPayload(BaseModel): caller_invocation_metadata: dict[str, Any] = Field(default_factory=dict) +# Spec observability §5.5.5: tool-call serialization. def serialize_tool_calls(tool_calls: Sequence[ToolCall]) -> list[dict[str, Any]]: - """The observability §5.5.5 tool-call serialization, - ``[{id, name, arguments}, ...]``. + """Serialize tool calls to ``[{id, name, arguments}, ...]``. The single home for the encoding, shared by the input-message payload (the provider's ``input.messages`` serialization, where the diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index 658f1cd..fb148a4 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -359,7 +359,7 @@ def _serialize_for_attribute(value: Any) -> str: renders via its ``str()`` rather than raising inside the observer (which would lose the whole span). It is a no-op for the already-encodable payloads (messages, params); it matters for the - proposal 0063 tool ``result``, which is an opaque, language-idiomatic + tool ``result``, which is an opaque, language-idiomatic value the tool produced (a model, dataclass, datetime, ...).""" return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str) @@ -1389,8 +1389,10 @@ def _emit_checkpoint_save_span(self, event: NodeEvent) -> None: # active_prompt_group snapshots taken at dispatch time — NOT the # ContextVar. The dispatch worker's task-local Context doesn't see # node-body ContextVar writes. + # Spec observability §11 GenAI metrics. §11.2: the attempt index is + # deliberately not a metric dimension (cardinality control). def _record_llm_metrics(self, event: LlmRetryAttemptEvent) -> None: - """Record the §11 GenAI metric observations for one LLM-call + """Record the GenAI metric observations for one LLM-call attempt. Duration is recorded for every attempt (including a failed one, @@ -1398,8 +1400,8 @@ def _record_llm_metrics(self, event: LlmRetryAttemptEvent) -> None: returned a usage record (a failed attempt has none). Sourced from the per-attempt ``LlmRetryAttemptEvent`` — one duration sample per attempt under call-level retry, matching the per-attempt span - model. The attempt index is deliberately not a dimension (§11.2 - cardinality). + model. The attempt index is deliberately not a dimension + (cardinality control). """ if self._duration_histogram is None or self._token_histogram is None: return @@ -1615,16 +1617,18 @@ def _handle_typed_llm_retry_attempt(self, event: LlmRetryAttemptEvent) -> None: self._run_enrichers(span, event) span.end(end_time=end_time_ns) + # Spec proposal 0063 tool span. Payload attributes (arguments / result) + # follow observability §5.5.4 opt-out + §5.5.5 truncation. def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: - """Emit an ``openarmature.tool.call`` span for a tool execution - (proposal 0063), parented under the calling node. + """Emit an ``openarmature.tool.call`` span for a tool execution, + parented under the calling node. A ``ToolCallEvent`` renders OK with the result attribute; a ``ToolCallFailedEvent`` renders ERROR with the standard OTel ``error.type`` attribute + an exception event carrying the message. ``arguments`` / ``result`` are payload, gated by - ``disable_provider_payload`` (§5.5.4) + the §5.5.5 truncation - contract. The OA-namespace ``openarmature.tool.*`` attributes + ``disable_provider_payload`` and truncated. The OA-namespace + ``openarmature.tool.*`` attributes mirror the Development ``gen_ai.tool.*`` surface, which is NOT emitted in v1. ``disable_llm_spans`` does not gate this span. """ diff --git a/tests/conformance/adapter.py b/tests/conformance/adapter.py index c98c7c3..9b4942b 100644 --- a/tests/conformance/adapter.py +++ b/tests/conformance/adapter.py @@ -1063,11 +1063,12 @@ def _add_fan_out_node( ) +# Spec proposal 0075: callable branch ``call:`` directive. def _build_call_fn( branch_name: str, call_cfg: Mapping[str, Any], ) -> Callable[[Any], Awaitable[Mapping[str, Any]]]: - """Translate a callable branch's ``call:`` directive (proposal 0075) + """Translate a callable branch's ``call:`` directive into an async function over the parent state. Reuses the node-behavior factories (``update`` / ``flaky`` / @@ -1087,8 +1088,9 @@ def _build_call_fn( raise ValueError(f"callable branch {branch_name!r}: unsupported call directive {dict(call_cfg)!r}") +# Spec proposal 0075 §11.10: branch ``when:`` predicate. def _build_when_predicate(when_cfg: Mapping[str, Any]) -> Callable[[Any], bool]: - """Translate a branch ``when:`` directive (proposal 0075 §11.10) into + """Translate a branch ``when:`` directive into a parent-state predicate. Supports ``{field: }`` — a truthy check on a parent-state field at dispatch time.""" if "field" in when_cfg: @@ -1115,7 +1117,7 @@ def _add_parallel_branches_node( A branch is either a ``subgraph`` (name resolved against the shared ``subgraphs`` registry, with optional ``inputs`` / ``outputs``) or an - inline ``call`` (proposal 0075), and may carry a ``when`` predicate. + inline ``call``, and may carry a ``when`` predicate. ``branch_middleware`` maps branch-name to a pre-translated middleware list; the test driver populates it from each branch's ``middleware:`` block. diff --git a/tests/conformance/harness/directives.py b/tests/conformance/harness/directives.py index 0925008..91df347 100644 --- a/tests/conformance/harness/directives.py +++ b/tests/conformance/harness/directives.py @@ -305,13 +305,14 @@ def _normalize_instance_middleware(cls, data: Any) -> Any: return data_dict +# Spec proposal 0075: parallel-branches branch entry. class ParallelBranchSpec(_AllowExtras): """One entry inside a ``parallel_branches.branches`` mapping. A branch's work is given by exactly one of ``subgraph`` (a compiled subgraph referenced by name, with optional ``inputs`` / ``outputs``) or ``call`` (an inline node-behavior directive — ``update`` / ``flaky`` - / ``raises`` — run as a function over the parent state, proposal 0075). + / ``raises`` — run as a function over the parent state). An optional ``when`` directive (``{field: }``) skips the branch at dispatch. Permissive on extras because fixtures may carry extra knobs (e.g., per-branch annotations the harness ignores). @@ -385,8 +386,9 @@ class MockToolSpec(_AllowExtras): raises: dict[str, Any] | None = None +# Spec proposal 0063: tool-using node directive. class CallsToolSpec(_AllowExtras): - """Tool-using node (proposal 0063): the node body enters the + """Tool-using node: the node body enters the ``with_tool_call`` instrumentation scope, runs the ``mock_tool``, and OA emits the ToolCallEvent / ToolCallFailedEvent. ``tool_call_id`` is optional (null = a standalone instrumented function); @@ -400,11 +402,12 @@ class CallsToolSpec(_AllowExtras): stores_result_in: str | None = None +# Spec proposal 0059: embedding-using node directive. class CallsEmbedSpec(_AllowExtras): """Embedding-using node: sends ``input`` to a mock embedding provider and stores the result in ``stores_response_in``. Used by the GenAI-metrics embedding fixture (089). The embedding capability - (proposal 0059) is unimplemented in python until v0.16.0, so 089 is + is unimplemented in python until v0.16.0, so 089 is deferred at the runner; this directive is modelled here only so the fixture parses + round-trips through the harness schema.""" diff --git a/tests/integration/test_langfuse_sdk_adapter.py b/tests/integration/test_langfuse_sdk_adapter.py index 6be3c4e..c39af61 100644 --- a/tests/integration/test_langfuse_sdk_adapter.py +++ b/tests/integration/test_langfuse_sdk_adapter.py @@ -249,16 +249,19 @@ async def test_sdk_adapter_generation_timestamps_round_trip_through_langfuse() - client.shutdown() +# Spec proposal 0064 §8.4.1: trace(session_id=, user_id=) populates the live +# Trace's sessionId / userId grouping. proposal 0020 (sessions capability) will +# later supply openarmature.session_id along the same passthrough. @pytest.mark.integration async def test_sdk_adapter_populates_session_and_user_id_on_live_langfuse() -> None: - """End-to-end (proposal 0064 §8.4.1): trace(session_id=, user_id=) + """End-to-end: trace(session_id=, user_id=) populates the live Trace's sessionId / userId grouping fields. The observer leaves session_id dormant until the sessions capability - (proposal 0020) supplies openarmature.session_id, but the adapter + supplies openarmature.session_id, but the adapter passes whatever it is given, so this exercises BOTH passthroughs at - the SDK boundary: when 0020 lands, the session_id rides the same - propagate_attributes path the userId promotion uses today. + the SDK boundary: when the sessions capability lands, the session_id + rides the same propagate_attributes path the userId promotion uses today. """ from langfuse import Langfuse diff --git a/tests/unit/test_tool_call.py b/tests/unit/test_tool_call.py index a79c085..13ffead 100644 --- a/tests/unit/test_tool_call.py +++ b/tests/unit/test_tool_call.py @@ -1,4 +1,5 @@ -"""Unit tests for the tool-call instrumentation scope (proposal 0063). +# Spec proposal 0063: tool-call instrumentation scope. +"""Unit tests for the tool-call instrumentation scope. Exercise ``with_tool_call`` directly by installing a collecting dispatch callback + a node-scope identity into the correlation ContextVars, the