diff --git a/packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py b/packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py index d6e499a..b0ffadb 100644 --- a/packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py +++ b/packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py @@ -274,7 +274,17 @@ async def _subagent_execute( "$ld:ai:graph:handoff_success", ld_context, td, 1 ) - path.append(_node.key) + if _node.key not in path: + index = len(path) + path.append(_node.key) + if ld_context: + node_td = make_track_data(_node, def_obj.key, run_id) + get_client().track( + "$ld:ai:graph:node", + ld_context, + {**node_td, "nodeKey": _node.key, "index": index}, + 1, + ) node_start = time.monotonic() result = await _run_query( _node, @@ -328,7 +338,17 @@ async def _subagent_execute( if e.target_key in subagent_tool_ctx ] - path.append(root.key) + if root.key not in path: + root_index = len(path) + path.append(root.key) + if ld_context: + root_node_td = make_track_data(root, def_obj.key, run_id) + get_client().track( + "$ld:ai:graph:node", + ld_context, + {**root_node_td, "nodeKey": root.key, "index": root_index}, + 1, + ) root_start = time.monotonic() try: @@ -402,7 +422,6 @@ async def _subagent_execute( root_td, total_usage["total"], ) - client.track("$ld:ai:graph:path", ld_context, root_td, len(path)) client.track("$ld:ai:graph:invocation_success", ld_context, root_td, 1) return {"response": final_output, "usage": total_usage} diff --git a/packages/claude-agents/tests/test_native_graph.py b/packages/claude-agents/tests/test_native_graph.py index c8099ff..e01900f 100644 --- a/packages/claude-agents/tests/test_native_graph.py +++ b/packages/claude-agents/tests/test_native_graph.py @@ -546,13 +546,19 @@ async def _query(**kwargs: Any) -> AsyncIterator[Any]: evt for evt, _ in track_calls if evt == "$ld:ai:graph:handoff_success" ] assert len(handoff_events) >= 1 + child_nodes = [ + data + for evt, data in track_calls + if evt == "$ld:ai:graph:node" and data.get("nodeKey") == "child" + ] + assert len(child_nodes) == 1 @pytest.mark.asyncio async def test_emits_invocation_success_on_completion(self) -> None: - track_calls: list[str] = [] + track_calls: list[tuple[str, Any]] = [] mock_ld_client = MagicMock() mock_ld_client.track = MagicMock( - side_effect=lambda evt, ctx, data, val: track_calls.append(evt) + side_effect=lambda evt, ctx, data, val: track_calls.append((evt, data)) ) mock_sdk = _make_sdk_mock("done") @@ -571,7 +577,15 @@ async def test_emits_invocation_success_on_completion(self) -> None: opts={"context": ctx}, ).invoke("hi") - assert "$ld:ai:graph:invocation_success" in track_calls + assert "$ld:ai:graph:invocation_success" in [evt for evt, _ in track_calls] + root_nodes = [ + data + for evt, data in track_calls + if evt == "$ld:ai:graph:node" and data.get("nodeKey") == "root" + ] + assert len(root_nodes) == 1 + assert root_nodes[0]["index"] == 0 + assert all(evt != "$ld:ai:graph:path" for evt, _ in track_calls) @pytest.mark.asyncio async def test_emits_invocation_failure_on_error(self) -> None: diff --git a/packages/client/src/launchdarkly_ai_server/graph.py b/packages/client/src/launchdarkly_ai_server/graph.py index 0ef93b0..de33bd6 100644 --- a/packages/client/src/launchdarkly_ai_server/graph.py +++ b/packages/client/src/launchdarkly_ai_server/graph.py @@ -909,7 +909,6 @@ async def invoke( span.set_attribute("ld.ai.graph.key", self._key) start_time = time.monotonic() - path: list[str] = [] total_usage = {"input": 0, "output": 0, "total": 0} resolved_input = user_input or "" @@ -920,9 +919,23 @@ async def invoke( last: dict[str, Any] | None = None visited: set[str] = set() steps = 0 + entered = 0 while current and steps < MAX_TRAVERSAL_DEPTH: steps += 1 + # Record the node as it is entered, so a later failure still + # has this step. The path is the ordered series of these events. + get_client().track( + "$ld:ai:graph:node", + ld_ctx, + { + **graph_track_data, + "nodeKey": current.key, + "index": entered, + }, + 1, + ) + entered += 1 opts: dict[str, Any] = {"variables": variables} if previous_node: opts["from"] = previous_node @@ -933,7 +946,6 @@ async def invoke( opts["history"] = history res = await graph_def.route(current, current_input, opts) - path.append(current.key) total_usage["input"] += ( res["usage"].get("input", 0) if isinstance(res["usage"], dict) @@ -978,12 +990,6 @@ async def invoke( graph_track_data, total_usage["total"], ) - client.track( - "$ld:ai:graph:path", - ld_ctx, - {**graph_track_data, "path": path}, - len(path), - ) client.track( "$ld:ai:graph:invocation_success", ld_ctx, graph_track_data, 1 ) @@ -1121,7 +1127,6 @@ async def _stream_events( ended: set[int] = set() start_time = time.monotonic() - path: list[str] = [] total_usage = {"input": 0, "output": 0, "total": 0} # A consumer that stops reading unwinds as GeneratorExit, and that is abandonment. @@ -1137,9 +1142,17 @@ async def _stream_events( last: dict[str, Any] | None = None visited: set[str] = set() steps = 0 + entered = 0 while current and steps < MAX_TRAVERSAL_DEPTH: steps += 1 + get_client().track( + "$ld:ai:graph:node", + ld_ctx, + {**graph_track_data, "nodeKey": current.key, "index": entered}, + 1, + ) + entered += 1 route_opts: dict[str, Any] = {"variables": variables} if previous_node: route_opts["from"] = previous_node @@ -1156,7 +1169,6 @@ async def _stream_events( ): yield event - path.append(current.key) usage = outcome.get("usage") or {} total_usage["input"] += ( usage.get("input", 0) if isinstance(usage, dict) else 0 @@ -1203,12 +1215,6 @@ async def _stream_events( graph_track_data, total_usage["total"], ) - client.track( - "$ld:ai:graph:path", - ld_ctx, - {**graph_track_data, "path": path}, - len(path), - ) client.track("$ld:ai:graph:invocation_success", ld_ctx, graph_track_data, 1) judge_results: dict[str, JudgeResult] | None = None diff --git a/packages/client/tests/test_graph.py b/packages/client/tests/test_graph.py index 3edffc1..a172e02 100644 --- a/packages/client/tests/test_graph.py +++ b/packages/client/tests/test_graph.py @@ -273,12 +273,41 @@ async def bad_fn( await g.invoke("hi", CONTEXT) events = [c[0][0] for c in mock_ld_client.track.call_args_list] assert "$ld:ai:graph:invocation_failure" in events - - async def test_graph_path_tracked(self, mock_ld_client: MagicMock) -> None: + assert "$ld:ai:graph:path" not in events + node_events = [ + c[0] + for c in mock_ld_client.track.call_args_list + if c[0][0] == "$ld:ai:graph:node" + ] + assert len(node_events) == 1 + assert node_events[0][2]["nodeKey"] == "root-node" + assert node_events[0][2]["index"] == 0 + assert node_events[0][2]["graphKey"] == "graph-key" + assert node_events[0][3] == 1 + assert "path" not in node_events[0][2] + + async def test_graph_node_tracked_per_visited_node( + self, mock_ld_client: MagicMock + ) -> None: g = graph("graph-key", handlers=[_make_handler()]) await g.invoke("hi", CONTEXT) events = [c[0][0] for c in mock_ld_client.track.call_args_list] - assert "$ld:ai:graph:path" in events + assert "$ld:ai:graph:path" not in events + node_events = [ + c[0] + for c in mock_ld_client.track.call_args_list + if c[0][0] == "$ld:ai:graph:node" + ] + assert len(node_events) == 2 + assert node_events[0][2]["nodeKey"] == "root-node" + assert node_events[0][2]["index"] == 0 + assert node_events[0][3] == 1 + assert node_events[1][2]["nodeKey"] == "leaf-node" + assert node_events[1][2]["index"] == 1 + assert node_events[1][3] == 1 + assert node_events[0][2]["runId"] == node_events[1][2]["runId"] + assert node_events[0][2]["graphKey"] == "graph-key" + assert "path" not in node_events[0][2] async def test_node_variations_resolved_only_once( self, mock_ld_client: MagicMock diff --git a/packages/client/tests/test_graph_stream.py b/packages/client/tests/test_graph_stream.py index 5d7265b..950d229 100644 --- a/packages/client/tests/test_graph_stream.py +++ b/packages/client/tests/test_graph_stream.py @@ -334,13 +334,30 @@ async def test_tracks_duration_total(self, mock_ld_client: MagicMock) -> None: ) assert "$ld:ai:graph:duration:total" in _track_names(mock_ld_client) - async def test_tracks_path(self, mock_ld_client: MagicMock) -> None: + async def test_tracks_node_per_visited_node( + self, mock_ld_client: MagicMock + ) -> None: await _collect( graph("graph-key", handlers=[_make_streaming_handler(["ok"])]).stream( "hi", CONTEXT ) ) - assert "$ld:ai:graph:path" in _track_names(mock_ld_client) + assert "$ld:ai:graph:path" not in _track_names(mock_ld_client) + node_events = [ + c[0] + for c in mock_ld_client.track.call_args_list + if c[0][0] == "$ld:ai:graph:node" + ] + assert len(node_events) == 2 + assert node_events[0][2]["nodeKey"] == "root-node" + assert node_events[0][2]["index"] == 0 + assert node_events[0][3] == 1 + assert node_events[1][2]["nodeKey"] == "leaf-node" + assert node_events[1][2]["index"] == 1 + assert node_events[1][3] == 1 + assert node_events[0][2]["runId"] == node_events[1][2]["runId"] + assert node_events[0][2]["graphKey"] == "graph-key" + assert "path" not in node_events[0][2] async def test_tracks_handoff_success(self, mock_ld_client: MagicMock) -> None: await _collect( @@ -372,7 +389,18 @@ async def stream_fn( ) with pytest.raises(RuntimeError, match="stream boom"): await _collect(graph("graph-key", handlers=[h]).stream("hi", CONTEXT)) - assert "$ld:ai:graph:invocation_failure" in _track_names(mock_ld_client) + names = _track_names(mock_ld_client) + assert "$ld:ai:graph:invocation_failure" in names + assert "$ld:ai:graph:path" not in names + node_events = [ + c[0] + for c in mock_ld_client.track.call_args_list + if c[0][0] == "$ld:ai:graph:node" + ] + assert len(node_events) == 1 + assert node_events[0][2]["nodeKey"] == "root-node" + assert node_events[0][2]["index"] == 0 + assert node_events[0][3] == 1 async def test_generation_success_includes_graph_key( self, mock_ld_client: MagicMock @@ -674,7 +702,18 @@ async def test_abandoned_on_consumer_break(self, mock_ld_client: MagicMock) -> N assert len(graph_spans) >= 1 attrs = graph_spans[0].attributes or {} assert attrs.get("launchdarkly.stream.abandoned") is True - assert "$ld:ai:graph:invocation_success" not in _track_names(mock_ld_client) + names = _track_names(mock_ld_client) + assert "$ld:ai:graph:invocation_success" not in names + assert "$ld:ai:graph:path" not in names + node_events = [ + c[0] + for c in mock_ld_client.track.call_args_list + if c[0][0] == "$ld:ai:graph:node" + ] + assert len(node_events) == 1 + assert node_events[0][2]["nodeKey"] == "root-node" + assert node_events[0][2]["index"] == 0 + assert node_events[0][3] == 1 async def test_cancelled_stream_marks_run_cancelled_not_abandoned( self, mock_ld_client: MagicMock diff --git a/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/native_graph.py b/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/native_graph.py index 6a70914..b063586 100644 --- a/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/native_graph.py +++ b/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/native_graph.py @@ -242,7 +242,17 @@ async def _handoff_exec( async def _node_fn( state: WorkflowState, _node: GraphNode = node ) -> dict[str, Any]: - path.append(_node.key) + if _node.key not in path: + index = len(path) + path.append(_node.key) + if ld_context: + node_td = make_track_data(_node, def_obj.key, run_id) + get_client().track( + "$ld:ai:graph:node", + ld_context, + {**node_td, "nodeKey": _node.key, "index": index}, + 1, + ) node_start = time.monotonic() system_prompt = _build_system_prompt(_node, vs) @@ -408,7 +418,6 @@ def _content_str(msg: Any) -> str: client.track( "$ld:ai:graph:total_tokens", ld_context, root_td, total_usage["total"] ) - client.track("$ld:ai:graph:path", ld_context, root_td, len(path)) client.track("$ld:ai:graph:invocation_success", ld_context, root_td, 1) return {"response": final_output, "usage": total_usage} diff --git a/packages/langchain-agents/tests/test_native_graph.py b/packages/langchain-agents/tests/test_native_graph.py index 45b2ff5..53457dd 100644 --- a/packages/langchain-agents/tests/test_native_graph.py +++ b/packages/langchain-agents/tests/test_native_graph.py @@ -434,6 +434,38 @@ async def test_invocation_success_and_duration_tracked(self) -> None: ).invoke("hi") assert "$ld:ai:graph:invocation_success" in track_calls + assert "$ld:ai:graph:path" not in track_calls + + @pytest.mark.asyncio + async def test_node_function_emits_graph_node(self) -> None: + track_calls: list[tuple[str, Any, Any]] = [] + mock_ld_client = MagicMock() + mock_ld_client.track = MagicMock( + side_effect=lambda evt, ctx, data, val: track_calls.append((evt, data, val)) + ) + + ai_msg = _make_ai_msg("done") + mocks = _make_langgraph_mocks(ai_msg) + graph_def = _make_graph_def() + ctx = {"kind": "user", "key": "test"} + + with _patch_imports(mocks): + with patch( + "launchdarkly_ai_langchain_agents.native_graph.get_client", + return_value=mock_ld_client, + ): + await to_lang_graph( + _make_def_promise(graph_def), + opts={"context": ctx}, + ).invoke("hi") + node_fn = mocks["_node_fns"]["root"] + await node_fn({"messages": []}) + + node_events = [item for item in track_calls if item[0] == "$ld:ai:graph:node"] + assert len(node_events) == 1 + assert node_events[0][1]["nodeKey"] == "root" + assert node_events[0][1]["index"] == 0 + assert node_events[0][2] == 1 @pytest.mark.asyncio async def test_invocation_failure_tracked(self) -> None: diff --git a/packages/openai-agents/src/launchdarkly_ai_openai_agents/native_graph.py b/packages/openai-agents/src/launchdarkly_ai_openai_agents/native_graph.py index 4003c01..a8b8a4a 100644 --- a/packages/openai-agents/src/launchdarkly_ai_openai_agents/native_graph.py +++ b/packages/openai-agents/src/launchdarkly_ai_openai_agents/native_graph.py @@ -209,14 +209,24 @@ async def on_handoff( get_client().track( "$ld:ai:graph:handoff_success", ld_context, td, 1 ) - to_key = agent_name_to_key.get(to_agent.name) - if to_key and to_key not in path: - path.append(to_key) async def on_agent_start(self, context: Any, agent: Any) -> None: node_key = agent_name_to_key.get(agent.name) - if node_key and node_key not in path: - path.append(node_key) + if not node_key or node_key in path: + return + index = len(path) + path.append(node_key) + if not ld_context: + return + node = def_obj.get_node(node_key) + if node: + td = make_track_data(node, def_obj.key, run_id) + get_client().track( + "$ld:ai:graph:node", + ld_context, + {**td, "nodeKey": node_key, "index": index}, + 1, + ) hooks = _LDHooks() @@ -297,7 +307,6 @@ async def on_agent_start(self, context: Any, agent: Any) -> None: client = get_client() client.track("$ld:ai:graph:duration:total", ld_context, root_td, duration) client.track("$ld:ai:graph:total_tokens", ld_context, root_td, total_tokens) - client.track("$ld:ai:graph:path", ld_context, root_td, len(path)) client.track("$ld:ai:graph:invocation_success", ld_context, root_td, 1) return {"response": final_output, "usage": total_usage} diff --git a/packages/openai-agents/tests/test_native_graph.py b/packages/openai-agents/tests/test_native_graph.py index 9e22759..2df4696 100644 --- a/packages/openai-agents/tests/test_native_graph.py +++ b/packages/openai-agents/tests/test_native_graph.py @@ -401,6 +401,7 @@ async def test_success_path_emits_invocation_success_and_tokens(self) -> None: ).invoke("hi") assert "$ld:ai:graph:invocation_success" in track_calls + assert "$ld:ai:graph:path" not in track_calls @pytest.mark.asyncio async def test_error_path_emits_invocation_failure_and_rethrows(self) -> None: @@ -658,6 +659,12 @@ async def _run_and_fire_hooks(agent: Any, text: str, hooks: Any = None) -> Any: mock_span = MagicMock() mock_trace = MagicMock() mock_trace.get_tracer.return_value.start_span.return_value = mock_span + track_calls: list[tuple[str, Any]] = [] + mock_ld_client = MagicMock() + mock_ld_client.track = MagicMock( + side_effect=lambda evt, ctx, data, val: track_calls.append((evt, data)) + ) + ld_context = {"kind": "user", "key": "test"} with patch( "importlib.import_module", @@ -665,7 +672,13 @@ async def _run_and_fire_hooks(agent: Any, text: str, hooks: Any = None) -> Any: ): with patch.object(_openai_ng, "trace", mock_trace): with patch.object(_openai_ng, "_HAS_OTEL", True): - await to_openai_agents(_make_def_promise(graph_def)).invoke("hi") + with patch.object( + _openai_ng, "get_client", return_value=mock_ld_client + ): + await to_openai_agents( + _make_def_promise(graph_def), + opts={"context": ld_context}, + ).invoke("hi") # Extract the path from the span set_attribute call for "launchdarkly.graph.path" path_val: str | None = None @@ -681,6 +694,14 @@ async def _run_and_fire_hooks(agent: Any, text: str, hooks: Any = None) -> Any: f"'child' appeared {child_occurrences} times in path '{path_val}'. " "Each node key must appear at most once (on_agent_start must not re-add keys already in path)." ) + node_events = [ + data + for evt, data in track_calls + if evt == "$ld:ai:graph:node" and data.get("nodeKey") == "child" + ] + assert len(node_events) == 1 + assert node_events[0]["index"] == 0 + assert all(evt != "$ld:ai:graph:path" for evt, _ in track_calls) @pytest.mark.asyncio async def test_agent_handoff_hook_emits_handoff_success(self) -> None: