feat(AIC-3211): record each graph node as its own tracking event

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}

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:

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

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

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

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}

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: