feat(observability): trace skill execution

tests/test_tracing_content.py

@@ -21,6 +21,7 @@
 from veadk.tracing.telemetry import telemetry
 from veadk.tracing.telemetry.content_tracing import should_trace_content
 from veadk.tracing.telemetry.exporters.apmplus_exporter import MeterUploader
+from veadk.tracing.telemetry.skill_observability import ActiveSkill, set_active_skill
 
 
 @dataclass
@@ -130,6 +131,9 @@ def __init__(self):
     def record(self, value, attributes=None):
         self.records.append((value, attributes))
 
+    def add(self, value, attributes=None):
+        self.records.append((value, attributes))
+
 
 def _start_test_span(name: str):
     provider = trace_sdk.TracerProvider()
@@ -230,6 +234,56 @@ def test_apmplus_tool_metrics_skip_token_usage_when_tool_content_missing():
     assert meter_uploader.apmplus_tool_token_usage.records == []
 
 
+def test_apmplus_llm_metrics_attribute_actual_tokens_to_active_skill():
+    meter_uploader = object.__new__(MeterUploader)
+    meter_uploader.llm_invoke_counter = _FakeMetricRecorder()
+    meter_uploader.token_usage = _FakeMetricRecorder()
+    meter_uploader.skill_token_usage = _FakeMetricRecorder()
+    meter_uploader.duration_histogram = _FakeMetricRecorder()
+    meter_uploader.chat_exception_counter = _FakeMetricRecorder()
+    meter_uploader.apmplus_span_latency = _FakeMetricRecorder()
+    set_active_skill(ActiveSkill(name="pdf", invocation_id="invocation"))
+
+    with _start_test_span("call_llm"):
+        meter_uploader.record_call_llm(
+            _FakeInvocationContext(),
+            "event-id",
+            _FakeLlmRequest(),
+            _FakeLlmResponse(),
+        )
+
+    assert [value for value, _ in meter_uploader.skill_token_usage.records] == [
+        11,
+        7,
+    ]
+    assert all(
+        attributes["skill_name"] == "pdf"
+        for _, attributes in meter_uploader.skill_token_usage.records
+    )
+
+
+def test_skill_operation_metrics_record_count_error_and_both_durations():
+    meter_uploader = object.__new__(MeterUploader)
+    meter_uploader.skill_invoke_counter = _FakeMetricRecorder()
+    meter_uploader.skill_error_counter = _FakeMetricRecorder()
+    meter_uploader.skill_invoke_latency = _FakeMetricRecorder()
+    meter_uploader.skill_duration_histogram = _FakeMetricRecorder()
+
+    with _start_test_span("skill.run_script") as span:
+        meter_uploader.record_skill_operation(
+            span=span,
+            operation="run_script",
+            attributes={"skill_name": "pdf"},
+            success=False,
+            error_type="skill_execution_error",
+        )
+
+    assert len(meter_uploader.skill_invoke_counter.records) == 1
+    assert len(meter_uploader.skill_error_counter.records) == 1
+    assert len(meter_uploader.skill_invoke_latency.records) == 1
+    assert len(meter_uploader.skill_duration_histogram.records) == 1
+
+
 def test_agent_root_span_skips_content_when_env_false(monkeypatch):
     monkeypatch.setenv("OBSERVABILITY_OPENTELEMETRY_TRACE_CONTENT", "false")
 

tests/tools/builtin_tools/test_run_sandbox_agent.py

@@ -170,7 +170,12 @@ def test_custom_values_override_defaults_and_preserve_empty_values(self):
         )
 
     def test_rejects_framework_managed_values(self):
-        for key in ["TOOL_USER_SESSION_ID", "USER_SESSION_ID"]:
+        for key in [
+            "TOOL_USER_SESSION_ID",
+            "USER_SESSION_ID",
+            "TRACEPARENT",
+            "TRACESTATE",
+        ]:
             with self.subTest(key=key):
                 with self.assertRaisesRegex(ValueError, "managed by VeADK"):
                     self.module._merge_execution_env_vars({}, {key: "spoofed"})
@@ -199,6 +204,27 @@ def test_runner_code_overrides_the_sandbox_process_environment(self):
         self.assertNotIn("if key not in env", code)
         self.assertIn('srv_pythonpath = env.get("SRV_PYTHONPATH")', code)
 
+    def test_current_trace_context_is_mapped_to_protected_environment(self):
+        def inject(carrier):
+            carrier.update(
+                {
+                    "traceparent": "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01",
+                    "tracestate": "vendor=value",
+                    "baggage": "secret=must-not-cross-the-boundary",
+                }
+            )
+
+        with patch.object(self.module.propagate, "inject", side_effect=inject):
+            result = self.module._current_trace_env_vars()
+
+        self.assertEqual(
+            result,
+            {
+                "TRACEPARENT": "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01",
+                "TRACESTATE": "vendor=value",
+            },
+        )
+
 
 class TestExecuteSkillsSkillApi(unittest.TestCase):
     def _tool_context(self):

tests/tracing/test_skill_observability.py

@@ -0,0 +1,89 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from veadk.tracing.telemetry import skill_observability
+
+
+class FakeSpan:
+    def __init__(self, invocation_id: str = "invocation-1") -> None:
+        self.attributes = {"invocation.id": invocation_id}
+        self.events = []
+
+    def set_attribute(self, name, value) -> None:
+        self.attributes[name] = value
+
+    def add_event(self, name, attributes=None) -> None:
+        self.events.append((name, attributes or {}))
+
+
+def test_google_skill_load_activates_skill_and_records_semantics():
+    span = FakeSpan()
+    tool = SimpleNamespace(name="load_skill")
+    response = SimpleNamespace(response={"status": "success"})
+
+    with (
+        patch.object(
+            skill_observability,
+            "get_event_function_responses",
+            return_value=[response],
+        ),
+        patch.object(skill_observability, "_record_skill_metrics") as record,
+    ):
+        skill_observability.observe_skill_tool_call(
+            span, tool, {"skill_name": "pdf"}, object()
+        )
+
+    assert span.attributes["skill.name"] == "pdf"
+    assert span.attributes["skill.operation"] == "load"
+    assert span.attributes["skill.phase"] == "completed"
+    assert span.events[0][0] == "skill.selected"
+    record.assert_called_once()
+
+
+def test_active_skill_does_not_leak_into_another_invocation():
+    skill_observability.set_active_skill(
+        skill_observability.ActiveSkill(name="pdf", invocation_id="invocation-previous")
+    )
+
+    assert skill_observability.active_skill_span_attributes("invocation-next") == {}
+    assert skill_observability.active_skill_metric_attributes("invocation-next") == {}
+
+
+def test_skill_script_failure_is_annotated_without_replacing_active_skill():
+    skill_observability.set_active_skill(
+        skill_observability.ActiveSkill(name="pdf", invocation_id="invocation-1")
+    )
+    span = FakeSpan()
+    tool = SimpleNamespace(name="run_skill_script")
+    response = SimpleNamespace(response={"error": "script failed"})
+
+    with (
+        patch.object(
+            skill_observability,
+            "get_event_function_responses",
+            return_value=[response],
+        ),
+        patch.object(skill_observability, "_record_skill_metrics") as record,
+    ):
+        skill_observability.observe_skill_tool_call(
+            span, tool, {"skill_name": "pdf"}, object()
+        )
+
+    assert span.attributes["skill.phase"] == "failed"
+    assert span.attributes["error.type"] == "skill_execution_error"
+    assert skill_observability.get_active_skill().name == "pdf"
+    assert record.call_args.args[2] is False

veadk/tools/builtin_tools/run_sandbox_agent.py

@@ -18,14 +18,36 @@
 from typing import Optional
 
 from google.adk.tools import ToolContext
+from opentelemetry import propagate, trace
+from opentelemetry.trace import SpanKind
 
 from veadk.tools.builtin_tools._agentkit import invoke_agentkit_run_code
 from veadk.utils.logger import get_logger
 
 logger = get_logger(__name__)
 
 _ENV_VAR_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
-_PROTECTED_ENV_VARS = frozenset({"TOOL_USER_SESSION_ID", "USER_SESSION_ID"})
+_TRACE_ENV_HEADERS = {
+    "traceparent": "TRACEPARENT",
+    "tracestate": "TRACESTATE",
+}
+_PROTECTED_ENV_VARS = frozenset(
+    {"TOOL_USER_SESSION_ID", "USER_SESSION_ID", *_TRACE_ENV_HEADERS.values()}
+)
+
+tracer = trace.get_tracer("veadk.sandbox_agent")
+
+
+def _current_trace_env_vars() -> dict[str, str]:
+    """Serialize the current W3C trace context for the sandbox subprocess."""
+
+    carrier: dict[str, str] = {}
+    propagate.inject(carrier)
+    return {
+        env_name: carrier[header]
+        for header, env_name in _TRACE_ENV_HEADERS.items()
+        if carrier.get(header)
+    }
 
 
 def _merge_execution_env_vars(
@@ -196,33 +218,41 @@ def run_sandbox_agent(
     tool_user_session_id = agent_name + "_" + user_id + "_" + session_id
     logger.debug(f"tool_user_session_id: {tool_user_session_id}")
 
-    base_env_vars = {
-        "TOOL_USER_SESSION_ID": tool_user_session_id,
-    }
-    skill_space_id = os.getenv("SKILL_SPACE_ID", "")
-    if skill_space_id:
-        base_env_vars["SKILL_SPACE_ID"] = skill_space_id
-    env_vars = _merge_execution_env_vars(base_env_vars, extra_env_vars)
-
-    logger.debug(
-        f"Run sandbox agent in session_id={session_id}, tool_id={tool_id}, timeout={timeout}, skills={skills}"
-    )
+    with tracer.start_as_current_span(
+        "skill.sandbox.invoke", kind=SpanKind.CLIENT
+    ) as span:
+        span.set_attribute("sandbox.tool.id", tool_id)
+        span.set_attribute("sandbox.session.id", tool_user_session_id)
+        span.set_attribute("sandbox.invoke.mode", "invoke_tool")
+
+        base_env_vars = {
+            "TOOL_USER_SESSION_ID": tool_user_session_id,
+            **_current_trace_env_vars(),
+        }
+        skill_space_id = os.getenv("SKILL_SPACE_ID", "")
+        if skill_space_id:
+            base_env_vars["SKILL_SPACE_ID"] = skill_space_id
+        env_vars = _merge_execution_env_vars(base_env_vars, extra_env_vars)
+
+        logger.debug(
+            f"Run sandbox agent in session_id={session_id}, tool_id={tool_id}, timeout={timeout}, skills={skills}"
+        )
 
-    cmd = _build_agent_command(workflow_prompt=workflow_prompt, skills=skills)
-    code = _build_agent_runner_code(
-        cmd=cmd,
-        timeout=timeout,
-        env_vars=env_vars,
-        working_dir=working_dir,
-    )
-    res = invoke_agentkit_run_code(
-        tool_id=tool_id,
-        tool_user_session_id=tool_user_session_id,
-        code=code,
-        timeout=timeout,
-        kernel_name="python3",
-        tool_state=tool_context.state if tool_context else None,
-    )
+        cmd = _build_agent_command(workflow_prompt=workflow_prompt, skills=skills)
+        code = _build_agent_runner_code(
+            cmd=cmd,
+            timeout=timeout,
+            env_vars=env_vars,
+            working_dir=working_dir,
+        )
+        res = invoke_agentkit_run_code(
+            tool_id=tool_id,
+            tool_user_session_id=tool_user_session_id,
+            code=code,
+            timeout=timeout,
+            kernel_name="python3",
+            tool_state=tool_context.state if tool_context else None,
+        )
     # The response can echo the submitted runner code, including custom env values.
     logger.debug("Invoke run sandbox agent completed")