Add LangGraph durable-execution agent lifecycle events

instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/536.added

@@ -0,0 +1 @@
+Emit agent lifecycle events (paused, checkpointed, resumed) for LangGraph durable executions.

instrumentation/opentelemetry-instrumentation-genai-langchain/README.rst

@@ -16,6 +16,8 @@ application:
 * **Agent spans** for agent invocations nested inside a workflow, including the
   agent name, id, description, and conversation/session id when available.
 * **Tool spans** for tool calls made during a run.
+* **Lifecycle events** for LangGraph durable executions that pause, checkpoint,
+  and resume, correlated with the workflow span.
 
 The spans nest to reflect the graph, so a single graph invocation produces a
 workflow span with the agent, tool, and model calls it triggered as children.
@@ -96,6 +98,21 @@ calls nested underneath.
         }
     )
 
+LangGraph durable executions
+----------------------------
+
+When a LangGraph graph pauses on ``interrupt()``, the instrumentation emits a
+``gen_ai.agent.paused`` event carrying the interrupt id LangGraph minted and the
+checkpoint it paused at. Resuming the graph with ``Command(resume=...)`` emits
+``gen_ai.agent.resumed`` naming that same checkpoint. If the graph was compiled
+with a checkpointer, every checkpoint the saver persists emits a
+``gen_ai.agent.checkpointed`` event; LangGraph writes one checkpoint per
+superstep, so these are per-step records.
+
+No interrupt payload or graph state is recorded: the events carry only ids
+LangGraph itself produced. These event and attribute names are candidate
+semantic conventions and may change.
+
 Configuration
 -------------
 

instrumentation/opentelemetry-instrumentation-genai-langchain/examples/langgraph-lifecycle/README.rst

@@ -0,0 +1,122 @@
+OpenTelemetry LangGraph agent lifecycle example
+===============================================
+
+This is an example of the agent lifecycle telemetry the instrumentation
+captures from a LangGraph durable execution: a graph that pauses on
+``interrupt()``, checkpoints, and resumes on a second invoke.
+
+`main.py <main.py>`_ needs no API key and no collector. It runs the graph twice
+under the instrumentor with in-memory exporters and writes every span and log
+record it produced to `sample-output.json <sample-output.json>`_, which is the
+captured run committed here.
+
+Setup
+-----
+
+Set up a virtual environment like this:
+
+::
+
+    python3 -m venv .venv
+    source .venv/bin/activate
+    pip install -r requirements.txt
+
+Run
+---
+
+Run the example from the repository root like this:
+
+::
+
+    python instrumentation/opentelemetry-instrumentation-genai-langchain/examples/langgraph-lifecycle/main.py
+
+You should see ``gen_ai.agent.paused``, ``gen_ai.agent.checkpointed``, and
+``gen_ai.agent.resumed`` events correlated with the workflow span of the invoke
+that produced them.
+
+Design note
+-----------
+
+Every name below is a candidate semantic convention proposed in
+open-telemetry/semantic-conventions-genai#445, and none of them is stable.
+Everything stated here was measured against langgraph 1.2.9 and
+langchain-core 1.5.0.
+
+Interception points
+~~~~~~~~~~~~~~~~~~~
+
+1. ``langgraph.callbacks`` lifecycle dispatch. LangGraph 1.2 calls
+   ``on_interrupt`` and ``on_resume`` on the handlers in a run's callback
+   manager, passing ``GraphInterruptEvent`` and ``GraphResumeEvent`` with the
+   real ``Interrupt`` objects, the checkpoint id, and the top level run id. The
+   released instrumentor is already in that handler list: without these two
+   methods LangGraph logs ``AttributeError`` on every interrupt and resume, so
+   adding them also removes existing log noise.
+2. ``StateGraph.compile(checkpointer=...)``. The saver the compiled graph
+   retains is wrapped so every persisted checkpoint reports its id.
+   ``BaseCheckpointSaver.put`` is abstract and every saver overrides it, so the
+   instance is patched rather than the base class. ``aput`` may delegate to
+   ``put``, possibly on a worker thread, so a write is de-duplicated by holding
+   the identity of the ``Checkpoint`` object while the write is in flight, not
+   by remembering ids and not by relying on context propagation: a nested call
+   receiving the same object is the delegation and stays silent, and the
+   outermost call reports. This assumes one checkpoint object per logical
+   write, which LangGraph guarantees by handing the saver a freshly constructed
+   mapping per superstep (``_loop.py`` passes
+   ``copy_checkpoint(self.checkpoint)``), so two independent writes never share
+   an object. A saver that copies the checkpoint before delegating is reported
+   twice, which is the safe direction.
+
+The interrupt is not observable from ``on_chain_end``: LangGraph adds
+``__interrupt__`` to the invoke return value after the callback fires.
+
+Correlation
+~~~~~~~~~~~
+
+``paused`` and ``resumed`` use the run id on the lifecycle event, resolved to
+the nearest workflow invocation. A checkpointer call has no run id, so
+``checkpointed`` maps the ``configurable.thread_id`` and
+``configurable.checkpoint_ns`` in its config to a live run. Runs are tracked per
+thread id, each bound with the namespace its own checkpoints use ("" for a top
+level graph, otherwise the run's ``langgraph_checkpoint_ns``). A write resolves
+to the run bound with that exact namespace, or to the nearest enclosing run when
+none is. A nested run never displaces the run containing it, and when two
+equally specific live runs are unrelated the event is dropped rather than
+guessed.
+
+LangGraph classifies a subgraph's own graph run as a plain chain, not a
+workflow, so today the child namespace resolves to the parent workflow. The
+resolution is namespace aware regardless, so a nested workflow would own its own
+writes.
+
+``resumed_from.type`` is always ``checkpoint``. That is a constant in the
+instrumentation, but it is determined by LangGraph, not chosen: the resume event
+supplies a checkpoint id and nothing else.
+
+``GenAIInvocation.emit_event`` is byte-identical to the hunk in open PR #507,
+saved as ``util/opentelemetry-util-genai/tests/fixtures/pr507_emit_event.py.txt``
+and checked by a test, and is dropped when rebasing onto that PR. It is not a
+competing API.
+
+Checkpoint volume
+~~~~~~~~~~~~~~~~~
+
+``put`` fires once per superstep. The captured run emits 3 ``checkpointed``
+events for the invoke that pauses, 2 for the invoke that resumes. Nested
+subgraphs checkpoint independently, under their own namespace, and emit one
+``resumed`` per graph level, which the flat event model does not distinguish.
+
+Not demonstrable
+~~~~~~~~~~~~~~~~
+
+``gen_ai.agent.execution.id`` is omitted. LangGraph mints no id spanning suspend
+and resume: ``thread_id`` is a conversation reused across runs, and every other
+id changes on the resuming invoke. The captured run shows the two invokes as
+separate traces with nothing linking them, which is the gap the attribute
+describes.
+
+``gen_ai.agent.pause.reason`` is omitted. ``interrupt(value)`` carries an opaque
+application payload and nothing saying who resolves the pause, so neither
+``human_input`` nor ``external_system`` is derivable. The ``pause`` member of
+``resumed_from.type`` is likewise absent: LangGraph never reports a pause id at
+resume time.

instrumentation/opentelemetry-instrumentation-genai-langchain/examples/langgraph-lifecycle/main.py

@@ -0,0 +1,146 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""LangGraph agent lifecycle telemetry example.
+
+Runs a small human-in-the-loop LangGraph application twice, once until it
+interrupts and once to resume it, under the LangChain instrumentor, and writes
+every span and log record it produced to ``sample-output.json`` beside this
+file. Needs no API key and no collector.
+
+Usage, from this directory::
+
+    python main.py
+
+See `README.rst <README.rst>`_ for what the events mean and where they come
+from.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, TypedDict
+
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import Command, interrupt
+
+from opentelemetry.instrumentation.genai.langchain import LangChainInstrumentor
+from opentelemetry.sdk._logs import LoggerProvider
+from opentelemetry.sdk._logs.export import (
+    InMemoryLogRecordExporter,
+    SimpleLogRecordProcessor,
+)
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
+    InMemorySpanExporter,
+)
+
+OUTPUT = Path(__file__).with_name("sample-output.json")
+
+
+class ExpenseState(TypedDict, total=False):
+    amount: int
+    approval: str
+    status: str
+
+
+def build_graph():
+    def prepare(_state: ExpenseState) -> ExpenseState:
+        return {"status": "prepared"}
+
+    def await_approval(state: ExpenseState) -> ExpenseState:
+        decision = interrupt(
+            {"question": "approve expense?", "amount": state.get("amount")}
+        )
+        return {"approval": str(decision)}
+
+    def submit(_state: ExpenseState) -> ExpenseState:
+        return {"status": "submitted"}
+
+    builder = StateGraph(ExpenseState)
+    builder.add_node("prepare", prepare)
+    builder.add_node("await_approval", await_approval)
+    builder.add_node("submit", submit)
+    builder.add_edge(START, "prepare")
+    builder.add_edge("prepare", "await_approval")
+    builder.add_edge("await_approval", "submit")
+    builder.add_edge("submit", END)
+    return builder.compile(checkpointer=InMemorySaver())
+
+
+def _span_json(span: Any) -> dict[str, Any]:
+    return {
+        "name": span.name,
+        "trace_id": f"{span.context.trace_id:032x}",
+        "span_id": f"{span.context.span_id:016x}",
+        "parent_span_id": (
+            f"{span.parent.span_id:016x}" if span.parent else None
+        ),
+        "kind": str(span.kind),
+        "attributes": dict(span.attributes or {}),
+    }
+
+
+def _log_json(record: Any) -> dict[str, Any]:
+    return {
+        "event_name": record.event_name,
+        "body": record.body,
+        "trace_id": f"{record.trace_id:032x}" if record.trace_id else None,
+        "span_id": f"{record.span_id:016x}" if record.span_id else None,
+        "attributes": dict(record.attributes or {}),
+    }
+
+
+def main() -> None:
+    span_exporter = InMemorySpanExporter()
+    tracer_provider = TracerProvider()
+    tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
+    log_exporter = InMemoryLogRecordExporter()
+    logger_provider = LoggerProvider()
+    logger_provider.add_log_record_processor(
+        SimpleLogRecordProcessor(log_exporter)
+    )
+
+    instrumentor = LangChainInstrumentor()
+    instrumentor.instrument(
+        tracer_provider=tracer_provider, logger_provider=logger_provider
+    )
+    try:
+        graph = build_graph()
+        config = {"configurable": {"thread_id": "expense-4711"}}
+        paused = graph.invoke({"amount": 250}, config=config)
+        interrupts = [
+            {"id": item.id, "value": item.value}
+            for item in paused["__interrupt__"]
+        ]
+        resumed = graph.invoke(Command(resume="approved"), config=config)
+    finally:
+        instrumentor.uninstrument()
+
+    document = {
+        "description": (
+            "Telemetry captured by opentelemetry-instrumentation-genai-"
+            "langchain from an unmodified LangGraph human-in-the-loop "
+            "application: one invoke that interrupts, one that resumes."
+        ),
+        "langgraph_ground_truth": {
+            "interrupts_returned_by_invoke": interrupts,
+            "final_state": resumed,
+        },
+        "spans": [
+            _span_json(span) for span in span_exporter.get_finished_spans()
+        ],
+        "logs": [
+            _log_json(item.log_record)
+            for item in log_exporter.get_finished_logs()
+        ],
+    }
+    OUTPUT.write_text(json.dumps(document, indent=2) + "\n")
+    print(f"wrote {OUTPUT}")
+
+
+if __name__ == "__main__":
+    main()

instrumentation/opentelemetry-instrumentation-genai-langchain/examples/langgraph-lifecycle/requirements.txt

@@ -0,0 +1,3 @@
+langgraph>=1.2.11
+opentelemetry-sdk>=1.43.0
+opentelemetry-instrumentation-genai-langchain~=1.0b0