fix(deer-flow): address code review blockers

Blockers:
- task_tool.py: initialize `result = None` before try, so capture-on + wrapped-raise no longer raises `UnboundLocalError` swallowing the original exception.
- sandbox.py: `_ProviderAcquireAsyncWrapper` now awaits `wrapped` directly and manages the TASK span lifecycle inline, instead of routing through `asyncio.to_thread` (which returned a coroutine object).

Non-blockers:
- Attach all three manual TASK spans (task_tool / sandbox provider lifecycle / memory) to the OTel Context via `set_span_in_context` so the langchain TOOL span emitted inside `wrapped` becomes a child of the TASK span (execute.md §3.4.3).
- `memory.py::_MemoryLoadSaveWrapper`: differentiate `load` vs `save` positional args — `save(memory_data, agent_name, *, user_id)` puts `agent_name` at `args[1]`, not `args[0]`.
- entry.py: drop dead `_call_arg(args, kwargs, 5, "graph_input")` fallback — `graph_input` is kw-only.
- README: document subagent AGENT span usage-metrics limitation and `graph_input` kw-only contract.

Tests: three regression tests covering Blockers 1/2 and Non-Blocker 3.

Author: 123liuziming

instrumentation-loongsuite/loongsuite-instrumentation-deer-flow/README.md

@@ -69,3 +69,13 @@ opentelemetry-instrument --instrumentation_modules deer_flow
   span (langchain instrumentation, child). This is intentional — the TASK
   span describes "dispatch to subagent" semantics while the TOOL span covers
   the LangChain `@tool` invocation mechanics.
+* **Subagent AGENT span does not carry usage metrics.** DeerFlow constructs
+  `SubagentTokenCollector` locally inside `SubagentExecutor._aexecute`, so
+  the wrapper cannot reach it via an instance attribute. Token usage is still
+  emitted on the langchain LLM child span via `loongsuite-instrumentation-langchain`,
+  but `gen_ai_llm_usage_tokens` will not be set on the AGENT (subagent) span.
+  This is a known limitation of the v1 patch surface; a follow-up could
+  expose the collector on the instance.
+* **`run_agent` `graph_input` is keyword-only.** The ENTRY span builder reads
+  it from `kwargs` only; the previous positional fallback was dead code and
+  has been removed.

instrumentation-loongsuite/loongsuite-instrumentation-deer-flow/src/opentelemetry/instrumentation/deer_flow/patches/entry.py

@@ -94,7 +94,10 @@ async def __call__(
             return await wrapped(*args, **kwargs)
 
         record = _call_arg(args, kwargs, 2, "record")
-        graph_input = kwargs.get("graph_input") or _call_arg(args, kwargs, 5, "graph_input")
+        # ``graph_input`` is kw-only in ``run_agent``; the positional fallback
+        # was dead code (the positional index would never contain it). Reading
+        # it from kwargs only.
+        graph_input = kwargs.get("graph_input")
         invocation = _safe_call(
             "build_entry_invocation", _build_entry_invocation, record, graph_input
         )

instrumentation-loongsuite/loongsuite-instrumentation-deer-flow/src/opentelemetry/instrumentation/deer_flow/patches/memory.py

@@ -41,7 +41,8 @@
     _should_capture_content,
     _should_capture_memory_content,
 )
-from opentelemetry.trace import SpanKind, Status, StatusCode, get_tracer
+from opentelemetry import context as otel_context
+from opentelemetry.trace import SpanKind, Status, StatusCode, get_tracer, set_span_in_context
 from opentelemetry.util.genai.extended_handler import ExtendedTelemetryHandler
 from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import (
     GEN_AI_SPAN_KIND,
@@ -92,16 +93,27 @@ def __call__(
         span.set_attribute(DEER_FLOW_COMPONENT, "memory")
         span.set_attribute(DEER_FLOW_TASK_NAME, task_name)
 
+        # ``FileMemoryStorage.load`` signature: ``load(agent_name=None, *,
+        # user_id=None)`` — ``args[0]`` is ``agent_name``.
+        # ``FileMemoryStorage.save`` signature: ``save(memory_data,
+        # agent_name=None, *, user_id=None)`` — ``args[0]`` is ``memory_data``,
+        # ``args[1]`` is ``agent_name``. Using ``args[0]`` for both would write
+        # ``memory_data`` repr into ``gen_ai.agent.name``.
         agent_name = kwargs.get("agent_name") if kwargs else None
         if agent_name is None and args:
-            agent_name = args[0]
+            if method_name == "save":
+                agent_name = args[1] if len(args) > 1 else None
+            else:
+                agent_name = args[0]
         if agent_name is not None:
             span.set_attribute("gen_ai.agent.name", str(agent_name))
 
         user_id = kwargs.get("user_id") if kwargs else None
         if user_id is not None:
             span.set_attribute("gen_ai.user.id", str(user_id))
 
+        token = otel_context.attach(set_span_in_context(span))
+
         if _capture_memory():
             _safe_call(
                 "set_input_value",
@@ -112,6 +124,7 @@ def __call__(
                 ),
             )
 
+        result: Any = None
         try:
             result = wrapped(*args, **kwargs)
         except Exception as exc:
@@ -126,6 +139,7 @@ def __call__(
                         "output.value", _serialize(result) or ""
                     ),
                 )
+            otel_context.detach(token)
             span.end()
         return result
 
@@ -163,6 +177,8 @@ async def __call__(
         if agent_name is not None:
             span.set_attribute("gen_ai.agent.name", str(agent_name))
 
+        token = otel_context.attach(set_span_in_context(span))
+
         if _capture_memory():
             _safe_call(
                 "set_input_value",
@@ -173,6 +189,7 @@ async def __call__(
                 ),
             )
 
+        result: Any = None
         try:
             result = await wrapped(*args, **kwargs)
         except Exception as exc:
@@ -187,6 +204,7 @@ async def __call__(
                         "output.value", _serialize(result) or ""
                     ),
                 )
+            otel_context.detach(token)
             span.end()
         return result
 

instrumentation-loongsuite/loongsuite-instrumentation-deer-flow/src/opentelemetry/instrumentation/deer_flow/patches/sandbox.py

@@ -36,7 +36,8 @@
     _safe_call,
     _should_capture_content,
 )
-from opentelemetry.trace import SpanKind, Status, StatusCode, get_tracer
+from opentelemetry import context as otel_context
+from opentelemetry.trace import SpanKind, Status, StatusCode, get_tracer, set_span_in_context
 from opentelemetry.util.genai.extended_handler import ExtendedTelemetryHandler
 from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import (
     GEN_AI_SPAN_KIND,
@@ -176,7 +177,9 @@ class _ProviderLifecycleWrapper:
     def __init__(self, tracer: Any):
         self._tracer = tracer
 
-    def _run(self, wrapped: Any, instance: Any, args: Any, kwargs: Any, *, method_name: str) -> Any:
+    def _start_span(
+        self, wrapped: Any, args: Any, kwargs: Any, *, method_name: str
+    ) -> tuple[Any, Any]:
         task_name = f"sandbox.{method_name}"
         span_name = f"run_task {task_name}"
         span = self._tracer.start_span(name=span_name, kind=SpanKind.INTERNAL)
@@ -190,13 +193,24 @@ def _run(self, wrapped: Any, instance: Any, args: Any, kwargs: Any, *, method_na
             thread_id = args[0]
         if thread_id is not None:
             span.set_attribute("gen_ai.session.id", str(thread_id))
+        # Attach so downstream spans (e.g. Sandbox TOOL spans created inside
+        # ``acquire``) become children of this lifecycle TASK span.
+        token = otel_context.attach(set_span_in_context(span))
+        return span, token
+
+    def _run(self, wrapped: Any, instance: Any, args: Any, kwargs: Any, *, method_name: str) -> Any:
+        span, token = self._start_span(
+            wrapped, args, kwargs, method_name=method_name
+        )
+        result: Any = None
         try:
             result = wrapped(*args, **kwargs)
         except Exception as exc:
             span.record_exception(exc)
             span.set_status(Status(StatusCode.ERROR, str(exc)))
             raise
         finally:
+            otel_context.detach(token)
             span.end()
         return result
 
@@ -210,12 +224,24 @@ async def __call__(
         self, wrapped: Any, instance: Any, args: Any, kwargs: Any
     ) -> Any:
         method_name = getattr(wrapped, "__name__", "acquire_async")
-        # Run sync path in a thread so the span covers the actual work.
-        import asyncio
-
-        return await asyncio.to_thread(
-            self._run, wrapped, instance, args, kwargs, method_name=method_name
+        # ``wrapped`` is a coroutine function; awaiting it directly (rather
+        # than routing through ``asyncio.to_thread``) preserves the caller's
+        # event loop and returns the awaited ``str`` sandbox id instead of a
+        # coroutine object.
+        span, token = self._start_span(
+            wrapped, args, kwargs, method_name=method_name
         )
+        result: Any = None
+        try:
+            result = await wrapped(*args, **kwargs)
+        except Exception as exc:
+            span.record_exception(exc)
+            span.set_status(Status(StatusCode.ERROR, str(exc)))
+            raise
+        finally:
+            otel_context.detach(token)
+            span.end()
+        return result
 
 
 def instrument(handler: ExtendedTelemetryHandler) -> list[tuple[str, str]]:

instrumentation-loongsuite/loongsuite-instrumentation-deer-flow/src/opentelemetry/instrumentation/deer_flow/patches/task_tool.py

@@ -36,7 +36,8 @@
     _safe_call,
     _should_capture_content,
 )
-from opentelemetry.trace import SpanKind, Status, StatusCode, get_tracer
+from opentelemetry import context as otel_context
+from opentelemetry.trace import SpanKind, Status, StatusCode, get_tracer, set_span_in_context
 from opentelemetry.util.genai.extended_handler import ExtendedTelemetryHandler
 from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import (
     GEN_AI_SPAN_KIND,
@@ -92,6 +93,11 @@ async def __call__(
         span.set_attribute(DEER_FLOW_TASK_NAME, task_name)
         span.set_attribute(GenAI.GEN_AI_AGENT_NAME, f"subagent:{subagent_type}")
 
+        # Attach the TASK span to the OTel Context so the langchain TOOL span
+        # emitted inside ``wrapped`` becomes a child of this TASK span (see
+        # execute.md §3.4.3).
+        token = otel_context.attach(set_span_in_context(span))
+
         if _should_capture_content():
             try:
                 span.set_attribute(
@@ -109,6 +115,7 @@ async def __call__(
             except Exception:
                 pass
 
+        result: Any = None
         try:
             result = await wrapped(*args, **kwargs)
         except Exception as exc:
@@ -122,6 +129,7 @@ async def __call__(
                     lambda r: span.set_attribute("output.value", str(r)),
                     result,
                 )
+            otel_context.detach(token)
             span.end()
         return result