feat: update wide event log

gusye1234 · gusye1234 · commit 698a8aa032e9 · 2026-03-18T22:17:17.000+08:00
diff --git a/src/server/core/acontext_core/infra/async_mq.py b/src/server/core/acontext_core/infra/async_mq.py
@@ -2,7 +2,6 @@
 # The publish() method includes retry logic to handle reconnection automatically.
 import asyncio
 import json
-import traceback
 from enum import StrEnum
 from functools import partial
 from pydantic import ValidationError, BaseModel
@@ -17,6 +16,7 @@
 from ..env import LOG, DEFAULT_CORE_CONFIG
 from ..telemetry.log import (
     bound_logging_vars,
+    get_logging_contextvars,
     set_wide_event,
     clear_wide_event,
 )
@@ -31,6 +31,22 @@
     OTEL_AVAILABLE = False
 
 
+def _inject_otel_trace(target: dict) -> None:
+    """Snapshot current OTel span's trace_id/span_id into *target* dict."""
+    if not OTEL_AVAILABLE:
+        return
+    try:
+        span = trace.get_current_span()
+        if span is None:
+            return
+        ctx = span.get_span_context()
+        if ctx is not None and ctx.trace_id != 0:
+            target["trace_id"] = format(ctx.trace_id, "032x")
+            target["span_id"] = format(ctx.span_id, "016x")
+    except Exception:
+        pass
+
+
 def _extract_trace_context_from_headers(message: Message) -> Optional[Any]:
     """Extract trace context from message headers for distributed tracing."""
     if not OTEL_AVAILABLE or not message.headers:
@@ -77,7 +93,10 @@ class SpecialHandler(StrEnum):
         "timeout_seconds",
         "duration_ms",
         "_log_level",
+        "trace_id",
+        "span_id",
     }
+    | LOGGING_FIELDS
 )
 
 
@@ -257,13 +276,16 @@ async def _process_message(
                                 if extracted_context and OTEL_AVAILABLE:
                                     token = otel_context.attach(extracted_context)
                                     try:
+                                        wide_event.update(get_logging_contextvars())
+                                        _inject_otel_trace(wide_event)
                                         await asyncio.wait_for(
                                             config.handler(validated_body, message),
                                             timeout=config.timeout,
                                         )
                                     finally:
                                         otel_context.detach(token)
                                 else:
+                                    wide_event.update(get_logging_contextvars())
                                     await asyncio.wait_for(
                                         config.handler(validated_body, message),
                                         timeout=config.timeout,
@@ -650,7 +672,7 @@ async def start(self) -> None:
 
                 for task in done:
                     try:
-                        r = task.result()
+                        task.result()
                         if task in self._consumer_loop_tasks:
                             self._consumer_loop_tasks.remove(task)
                     except Exception as e:
diff --git a/src/server/core/acontext_core/llm/agent/task.py b/src/server/core/acontext_core/llm/agent/task.py
@@ -189,6 +189,7 @@ async def task_agent_curd(
                 for tool_call in use_tools:
                     try:
                         tool_name = tool_call.function.name
+                        tools_called.append(tool_name)
                         if tool_name == "finish":
                             just_finish = True
                             continue
@@ -208,8 +209,6 @@ async def task_agent_curd(
                                 raise RuntimeError(
                                     f"Tool {tool_name} rejected: {r.error}"
                                 )
-                        if tool_name != "report_thinking":
-                            tools_called.append(tool_name)
                         tool_response.append(
                             {
                                 "role": "tool",
diff --git a/src/server/core/acontext_core/llm/complete/anthropic_sdk.py b/src/server/core/acontext_core/llm/complete/anthropic_sdk.py
@@ -5,6 +5,7 @@
 from time import perf_counter
 from ...env import LOG
 from ...schema.llm import LLMResponse
+from ...telemetry.log import get_wide_event
 
 
 def convert_openai_tool_to_anthropic_tool(tools: list[dict]) -> list[dict]:
@@ -108,14 +109,23 @@ async def anthropic_complete(
         )
         _end_s = perf_counter()
 
+        _input = response.usage.input_tokens
+        _output = response.usage.output_tokens
+        _cached = response.usage.cache_read_input_tokens or 0
+
+        wide = get_wide_event()
+        wide["llm_input_tokens"] = wide.get("llm_input_tokens", 0) + _input
+        wide["llm_output_tokens"] = wide.get("llm_output_tokens", 0) + _output
+        wide["llm_cached_tokens"] = wide.get("llm_cached_tokens", 0) + _cached
+
         LOG.info(
             "llm.complete",
             prompt_id=prompt_id,
             model=model,
-            cached_tokens=response.usage.cache_read_input_tokens,
-            input_tokens=response.usage.input_tokens,
-            output_tokens=response.usage.output_tokens,
-            total_tokens=response.usage.input_tokens + response.usage.output_tokens,
+            cached_tokens=_cached,
+            input_tokens=_input,
+            output_tokens=_output,
+            total_tokens=_input + _output,
             duration_s=round(_end_s - _start_s, 4),
         )
 
diff --git a/src/server/core/acontext_core/llm/complete/openai_sdk.py b/src/server/core/acontext_core/llm/complete/openai_sdk.py
@@ -6,6 +6,7 @@
 from time import perf_counter
 from ...env import LOG, DEFAULT_CORE_CONFIG
 from ...schema.llm import LLMResponse
+from ...telemetry.log import get_wide_event
 
 
 def convert_openai_tool_to_llm_tool(tool_body: ChatCompletionMessageToolCall) -> dict:
@@ -59,15 +60,23 @@ async def openai_complete(
         **kwargs,
     )
     _end_s = perf_counter()
-    cached_tokens = getattr(response.usage.prompt_tokens_details, "cached_tokens", None)
+    _input = response.usage.prompt_tokens
+    _output = response.usage.completion_tokens
+    _cached = getattr(response.usage.prompt_tokens_details, "cached_tokens", None) or 0
+
+    wide = get_wide_event()
+    wide["llm_input_tokens"] = wide.get("llm_input_tokens", 0) + _input
+    wide["llm_output_tokens"] = wide.get("llm_output_tokens", 0) + _output
+    wide["llm_cached_tokens"] = wide.get("llm_cached_tokens", 0) + _cached
+
     LOG.info(
         "llm.complete",
         prompt_id=prompt_id,
         model=model,
-        cached_tokens=cached_tokens,
-        input_tokens=response.usage.prompt_tokens,
-        output_tokens=response.usage.completion_tokens,
-        total_tokens=response.usage.total_tokens,
+        cached_tokens=_cached,
+        input_tokens=_input,
+        output_tokens=_output,
+        total_tokens=_input + _output,
         duration_s=round(_end_s - _start_s, 4),
     )
 
diff --git a/src/server/core/acontext_core/schema/result.py b/src/server/core/acontext_core/schema/result.py
@@ -1,3 +1,4 @@
+import sys
 from pydantic import BaseModel, ConfigDict
 from typing import Generic, TypeVar, Optional, Union
 from .error_code import Code
@@ -6,6 +7,17 @@
 T = TypeVar("T")
 
 
+def _caller_name(depth: int = 2) -> str:
+    """Return the qualified name of the caller's caller.
+
+    ``depth=2`` skips this function and the immediate caller (resolve/reject).
+    """
+    try:
+        return sys._getframe(depth).f_code.co_qualname
+    except (ValueError, AttributeError):
+        return "unknown"
+
+
 class ResultError(Exception):
     pass
 
@@ -29,14 +41,19 @@ class Result(BaseModel, Generic[T]):
 
     @classmethod
     def resolve(cls, data: T) -> "Result[T]":
+        wide = get_wide_event()
+        caller = _caller_name()
+        stack = wide.setdefault("success_stack", [])
+        if caller not in stack:
+            stack.append(caller)
         return cls(data=data, error=Error())
 
     @classmethod
     def reject(cls, errmsg: str, status: Code = Code.INTERNAL_ERROR) -> "Result[T]":
         assert status != Code.SUCCESS, "status must not be SUCCESS"
         wide = get_wide_event()
-        wide.setdefault("errors", []).append(
-            {"status": str(status), "errmsg": errmsg}
+        wide.setdefault("error_stack", []).append(
+            {"caller": _caller_name(), "status": str(status), "errmsg": errmsg}
         )
         return cls(data=None, error=Error.init(status, errmsg))
 
diff --git a/src/server/core/acontext_core/service/session_message.py b/src/server/core/acontext_core/service/session_message.py
@@ -37,6 +37,7 @@ async def insert_new_message(body: InsertNewMessage, message: Message):
         msg_status, eil = r.unpack()
         if eil or msg_status != "pending":
             wide["action"] = "skip_not_pending"
+            wide["_log_level"] = "debug"
             return
 
         r = await PD.get_project_config(session, body.project_id)
diff --git a/src/server/core/tests/schema/test_result.py b/src/server/core/tests/schema/test_result.py
@@ -54,7 +54,44 @@ def test_get_wide_event_throwaway_dict_does_not_pollute_contextvar():
 
 
 # ---------------------------------------------------------------------------
-# Result.reject appends errors to wide event
+# Result.resolve records deduplicated caller in success_stack
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_records_caller_in_success_stack():
+    event: dict = {}
+    set_wide_event(event)
+    try:
+        Result.resolve("a")
+        Result.resolve("b")
+        Result.resolve("c")
+        assert event["success_stack"] == ["test_resolve_records_caller_in_success_stack"]
+    finally:
+        clear_wide_event()
+
+
+def _helper_resolve():
+    return Result.resolve("from helper")
+
+
+def test_resolve_deduplicates_same_caller():
+    event: dict = {}
+    set_wide_event(event)
+    try:
+        Result.resolve("first")
+        _helper_resolve()
+        Result.resolve("second")
+        _helper_resolve()
+        assert event["success_stack"] == [
+            "test_resolve_deduplicates_same_caller",
+            "_helper_resolve",
+        ]
+    finally:
+        clear_wide_event()
+
+
+# ---------------------------------------------------------------------------
+# Result.reject appends to error_stack
 # ---------------------------------------------------------------------------
 
 
@@ -63,10 +100,10 @@ def test_reject_appends_error_to_wide_event():
     set_wide_event(event)
     try:
         Result.reject("something broke", Code.INTERNAL_ERROR)
-        assert "errors" in event
-        assert len(event["errors"]) == 1
-        assert event["errors"][0]["errmsg"] == "something broke"
-        assert event["errors"][0]["status"] == str(Code.INTERNAL_ERROR)
+        assert len(event["error_stack"]) == 1
+        assert event["error_stack"][0]["errmsg"] == "something broke"
+        assert event["error_stack"][0]["status"] == str(Code.INTERNAL_ERROR)
+        assert event["error_stack"][0]["caller"] == "test_reject_appends_error_to_wide_event"
     finally:
         clear_wide_event()
 
@@ -77,9 +114,9 @@ def test_reject_accumulates_multiple_errors():
     try:
         Result.reject("first", Code.BAD_REQUEST)
         Result.reject("second", Code.INTERNAL_ERROR)
-        assert len(event["errors"]) == 2
-        assert event["errors"][0]["errmsg"] == "first"
-        assert event["errors"][1]["errmsg"] == "second"
+        assert len(event["error_stack"]) == 2
+        assert event["error_stack"][0]["errmsg"] == "first"
+        assert event["error_stack"][1]["errmsg"] == "second"
     finally:
         clear_wide_event()
 
diff --git a/src/server/core/uv.lock b/src/server/core/uv.lock
