Fix/langgraph manual instrumentation context (#261)

fix: use run_inline=True to fix ContextVar propagation in LangGraph async nodes

Author: zhirafovod

AGENTS.md

@@ -223,8 +223,16 @@ VS Code launch configurations are in `.vscode/launch.json` for debugging example
 - Always keep backward compatibility in mind when refactoring existing
 - Follow DRY and SOLID software engineering principles when readability and maintainability is not compromised.
 
+## Git Commit Rules
+
+- **Never add `Co-Authored-By` trailers** (or any similar attribution trailers such as `Co-authored-by`, `Signed-off-by`, etc.) that reference AI assistants, bots, or automated tools in commit messages. This includes but is not limited to Claude, Copilot, ChatGPT, or any `noreply@` addresses from AI vendors.
+- Commit messages should only attribute human contributors.
+
 ## Common Pitfalls to Avoid
 
 - Do not try to mock libraries if import in the current env fail. If in doubt - clearly communicate the problem to user
 - Always refer to `README.md` and `README.packages.architecture.md` for context.
 - avoid creating multiple copies of example apps, when can introduce parameters and reuse the same demo app
+- **Do not use `try/except` guards around test imports.** Test dependencies must be declared in `pyproject.toml` `[project.optional-dependencies] test` and are expected to be present at test time. If they are missing, the test should fail with an `ImportError`, not silently skip.
+- **Do not use `sys.path` hacks** (e.g., inserting `src/` into `sys.path`) in test files. Packages should be installed in editable mode (`pip install -e .`) and imports should work without path manipulation.
+- **Do not use `pytest.mark.skipif` to guard against missing test dependencies.** If a dependency is required for a test, add it to the test requirements and import it directly.

instrumentation-genai/opentelemetry-instrumentation-langchain/examples/manual/langgraph_manual_tool_calls.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""LangGraph + manual handler instrumentation example.
+
+Demonstrates combining auto-instrumented LangGraph agent spans with manual
+handler.start_tool_call() / handler.start_step() calls inside async node
+bodies.  All spans end up on the same trace with correct parent-child
+relationships.
+
+Usage:
+    # Start an OTel collector on localhost:4317, then:
+    export OTEL_SERVICE_NAME=langgraph-manual-demo
+    export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
+    export OTEL_EXPORTER_OTLP_PROTOCOL=grpc
+    export OTEL_INSTRUMENTATION_GENAI_EMITTERS=span_metric_event
+    export OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true
+    python langgraph_manual_tool_calls.py
+"""
+
+import asyncio
+
+from opentelemetry import _events, _logs, metrics, trace
+from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk._events import EventLoggerProvider
+from opentelemetry.sdk._logs import LoggerProvider
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+
+def configure_instrumentation():
+    """Set up OTel providers and SDOT instrumentors."""
+    trace.set_tracer_provider(TracerProvider())
+    trace.get_tracer_provider().add_span_processor(
+        BatchSpanProcessor(OTLPSpanExporter())
+    )
+    reader = PeriodicExportingMetricReader(OTLPMetricExporter())
+    metrics.set_meter_provider(MeterProvider(metric_readers=[reader]))
+    _logs.set_logger_provider(LoggerProvider())
+    _logs.get_logger_provider().add_log_record_processor(
+        BatchLogRecordProcessor(OTLPLogExporter())
+    )
+    _events.set_event_logger_provider(EventLoggerProvider())
+
+    from opentelemetry.instrumentation.langchain import LangchainInstrumentor
+
+    LangchainInstrumentor().instrument()
+
+
+# Must be called before importing LangGraph / LangChain
+configure_instrumentation()
+
+from typing import TypedDict  # noqa: E402
+
+from langgraph.graph import StateGraph  # noqa: E402
+from opentelemetry.util.genai.handler import get_telemetry_handler  # noqa: E402
+from opentelemetry.util.genai.types import ToolCall, Step  # noqa: E402
+
+
+class State(TypedDict):
+    query: str
+    result: str
+
+
+# -- Async node functions with manual tool/step instrumentation --
+
+
+async def lookup_node(state: State) -> State:
+    """Async node: looks up data using a manual ToolCall span."""
+    handler = get_telemetry_handler()
+
+    # Create a manual ToolCall — the handler automatically parents it
+    # to the current agent span (even inside LangGraph async nodes).
+    tool = ToolCall(
+        name="database_lookup",
+        tool_type="datastore",
+        tool_description="Look up customer record in database",
+        arguments={"query": state["query"]},
+        agent_name="lookup-agent",
+    )
+    handler.start_tool_call(tool)
+
+    # Simulate database query
+    await asyncio.sleep(0.05)
+    result = f"Found record for: {state['query']}"
+
+    tool.tool_result = result
+    handler.stop_tool_call(tool)
+
+    return {"query": state["query"], "result": result}
+
+
+async def analysis_node(state: State) -> State:
+    """Async node: analyses data using a manual Step span."""
+    handler = get_telemetry_handler()
+
+    step = Step(
+        name="sentiment_analysis",
+        objective="Analyse sentiment of the query",
+        step_type="processing",
+        source="agent",
+        assigned_agent="analysis-agent",
+    )
+    handler.start_step(step)
+
+    await asyncio.sleep(0.03)
+    analysis = f"Analysis complete for: {state['result']}"
+
+    handler.stop_step(step)
+
+    return {"query": state["query"], "result": analysis}
+
+
+async def main():
+    builder = StateGraph(State)
+
+    # metadata={"agent_name": ...} makes the auto-instrumented span carry
+    # gen_ai.agent.name, and enables the handler's agent context stack.
+    builder.add_node("lookup", lookup_node, metadata={"agent_name": "lookup-agent"})
+    builder.add_node(
+        "analysis", analysis_node, metadata={"agent_name": "analysis-agent"}
+    )
+    builder.add_edge("__start__", "lookup")
+    builder.add_edge("lookup", "analysis")
+    builder.add_edge("analysis", "__end__")
+
+    graph = builder.compile()
+
+    result = await graph.ainvoke({"query": "customer 42 order status", "result": ""})
+    print(f"Result: {result['result']}")
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    finally:
+        # Flush all providers before exit
+        try:
+            trace.get_tracer_provider().force_flush(timeout_millis=5000)
+        except Exception:
+            pass
+        try:
+            metrics.get_meter_provider().force_flush(timeout_millis=5000)
+        except Exception:
+            pass
+        try:
+            _logs.get_logger_provider().force_flush(timeout_millis=5000)
+        except Exception:
+            pass

instrumentation-genai/opentelemetry-instrumentation-langchain/pyproject.toml

@@ -35,6 +35,9 @@ instruments = ["langchain >= 0.3.21"]
 test = [
   "langchain-core >= 1.0.0",
   "langchain-openai >= 1.0.0",
+  "langgraph >= 0.4.0",
+  "opentelemetry-sdk >= 1.20.0",
+  "pytest-asyncio >= 0.23.0",
   "pytest-recording >= 0.13.0",
   "vcrpy >= 7.0.0",
   "pyyaml >= 6.0.0",

instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/callback_handler.py

@@ -131,23 +131,40 @@ def _make_command_input_message(command: Any) -> list[InputMessage]:
     return [InputMessage(role="user", parts=[Text(_safe_str(command))])]
 
 
-def _make_input_message(data: dict[str, Any]) -> list[InputMessage]:
+def _make_input_message(data: Any) -> list[InputMessage]:
     """Create structured input message with full data as JSON."""
+    if not isinstance(data, dict):
+        return []
     input_messages: list[InputMessage] = []
     messages = data.get("messages")
-    if messages is None:
-        return []
-    for msg in messages:
-        content = getattr(msg, "content", "")
-        if content:
-            # TODO: for invoke_agent type invocation, when system_messages is added, can filter SystemMessage separately if needed and only add here HumanMessage, currently all messages are added
-            input_message = InputMessage(role="user", parts=[Text(_safe_str(content))])
-            input_messages.append(input_message)
+    if messages is not None:
+        for msg in messages:
+            content = getattr(msg, "content", "")
+            if content:
+                # TODO: for invoke_agent type invocation, when system_messages is added, can filter SystemMessage separately if needed and only add here HumanMessage, currently all messages are added
+                input_message = InputMessage(
+                    role="user", parts=[Text(_safe_str(content))]
+                )
+                input_messages.append(input_message)
+        return input_messages
+    # Fallback: serialize non-message state fields as input.
+    # Common in LangGraph where nodes use structured state fields
+    # (e.g., user_query) rather than a message list.
+    exclude_keys = {"messages", "intermediate_steps"}
+    input_data = {
+        k: v for k, v in data.items() if k not in exclude_keys and v is not None
+    }
+    if input_data:
+        serialized = _serialize(input_data)
+        if serialized:
+            return [InputMessage(role="user", parts=[Text(serialized)])]
     return input_messages
 
 
-def _make_output_message(data: dict[str, Any]) -> list[OutputMessage]:
+def _make_output_message(data: Any) -> list[OutputMessage]:
     """Create structured output message with full data as JSON."""
+    if not isinstance(data, dict):
+        return []
     output_messages: list[OutputMessage] = []
     messages = data.get("messages")
     if messages is None:
@@ -176,14 +193,14 @@ def _make_last_output_message(data: dict[str, Any]) -> list[OutputMessage]:
     return []
 
 
-def _make_workflow_output_fallback(data: dict[str, Any]) -> Optional[str]:
+def _make_workflow_output_fallback(data: Any) -> Optional[str]:
     """Create output summary from non-message state fields.
 
     Fallback for when workflow output doesn't contain AI messages.
     This is common in LangGraph where agent nodes update structured
     state fields rather than the message list.
     """
-    if not data:
+    if not isinstance(data, dict):
         return None
     # Exclude messages and internal fields that don't represent output
     exclude_keys = {"messages", "intermediate_steps"}
@@ -356,6 +373,14 @@ def remove(self, run_id: UUID) -> None:
 
 
 class LangchainCallbackHandler(BaseCallbackHandler):
+    # Run callback methods directly in the caller's context instead of
+    # dispatching to a thread pool via copy_context().run().  Without this,
+    # LangChain Core's callback manager isolates each handler invocation in
+    # a copied context, making ContextVar modifications (e.g. the
+    # _current_genai_span set by _push_current_span) invisible to the node
+    # body where user code may call handler.start_tool_call() manually.
+    run_inline = True
+
     def __init__(
         self,
         telemetry_handler: Optional[TelemetryHandler] = None,

instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_langgraph_interrupt_resume.py

@@ -7,38 +7,20 @@
 from __future__ import annotations
 
 import os
-import sys
-from pathlib import Path
 from typing import TypedDict
 
 import pytest
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import StateGraph
+from langgraph.types import Command, interrupt
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import (
+    SimpleSpanProcessor,
+    SpanExporter,
+    SpanExportResult,
+)
 
-_PACKAGE_SRC = Path(__file__).resolve().parents[1] / "src"
-if _PACKAGE_SRC.exists():
-    sys.path.insert(0, str(_PACKAGE_SRC))
-
-try:
-    from langgraph.graph import StateGraph
-    from langgraph.types import Command, interrupt
-    from langgraph.checkpoint.memory import MemorySaver
-except ModuleNotFoundError:
-    StateGraph = None  # type: ignore[assignment]
-
-try:
-    from opentelemetry.sdk.trace import TracerProvider
-    from opentelemetry.sdk.trace.export import (
-        SimpleSpanProcessor,
-        SpanExporter,
-        SpanExportResult,
-    )
-except ModuleNotFoundError:
-    TracerProvider = None  # type: ignore[assignment]
-
-from opentelemetry.instrumentation.langchain import LangchainInstrumentor  # noqa: E402
-
-LANGGRAPH_AVAILABLE = StateGraph is not None
-OTEL_SDK_AVAILABLE = TracerProvider is not None
-DEPS_AVAILABLE = LANGGRAPH_AVAILABLE and OTEL_SDK_AVAILABLE
+from opentelemetry.instrumentation.langchain import LangchainInstrumentor
 
 
 class _CollectingExporter(SpanExporter):
@@ -126,7 +108,6 @@ def instrumented_graph():
     _handler_mod.TelemetryHandler._reset_for_testing()
 
 
-@pytest.mark.skipif(not DEPS_AVAILABLE, reason="langgraph or otel sdk not available")
 class TestLangGraphInterruptResume:
     """End-to-end tests for interrupt/resume with real LangGraph execution."""