# src/pages/builder/templates/agent-python/context.py
"""AgentContext — the single argument passed to every agent handler.

也定义流式响应所需的公共类型 StreamResponse + SSE 便捷构造。

推荐用法（通过 ctx.utils，无需任何 import）：

    async def handler(ctx):
        async def gen():
            yield ctx.utils.sse({"event": "start"})
            yield ctx.utils.sse({"token": "hi"}, event="delta")
        return ctx.utils.stream_sse(gen())

向后兼容：旧的显式 import 方式仍然有效：

    from _platform.context import StreamResponse, sse

之所以放在 context.py 而非 runtime.py：业务 handler 只 import 这个模块就够了，
runtime.py 是 adapter 内部 dispatch 用的，业务无需感知。
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import os
import re
from dataclasses import dataclass, field
from typing import Any, AsyncIterator, Callable, Iterable, Mapping, Optional, TYPE_CHECKING, Union

if TYPE_CHECKING:
    from .store import InMemoryStore
    from .runtime import AgentsApi
    from .memory import ConversationMemory


PAGES_SANDBOX_SEALED_TOKEN = "{{PAGES_SANDBOX_SEALED_TOKEN}}"


class SandboxRuntimeError(RuntimeError):
    def __init__(self, message: str, code: str, operation: str = "init", retryable: bool = False):
        super().__init__(message)
        self.code = code
        self.operation = operation
        self.retryable = retryable


def _is_placeholder_token(value: str) -> bool:
    return value.startswith("{{") and value.endswith("}}")


def _is_valid_sandbox_sealed_token(value: str) -> bool:
    token = str(value or "").strip()
    return len(token) <= 4096 and re.fullmatch(r"sandbox\.v1\.[A-Za-z0-9._~-]+", token) is not None


def _resolve_local_login_token_from_persisted_storage() -> str:
    try:
        storage_file = os.path.join(
            os.path.expanduser("~"),
            ".edgeone",
            hashlib.sha256("eo_token".encode("utf-8")).hexdigest(),
        )
        with open(storage_file, "r", encoding="utf-8") as file:
            datum = json.load(file)
        return str(((datum or {}).get("value") or {}).get("Token") or "").strip()
    except Exception:
        return ""


def _resolve_sandbox_auth_token(env: Mapping[str, str], mode: str) -> str:
    sealed_token = str(PAGES_SANDBOX_SEALED_TOKEN or "").strip()
    if _is_valid_sandbox_sealed_token(sealed_token):
        return sealed_token
    if sealed_token and not _is_placeholder_token(sealed_token):
        raise SandboxRuntimeError(
            "Invalid sandbox sealed token. Expected sandbox.v1.* token generated by Pages deployment.",
            "SANDBOX_INVALID_AUTH_TOKEN",
        )
    if mode == "dev":
        dev_token = str(env.get("EDGEONE_PAGES_API_TOKEN", "") or "").strip()
        if dev_token:
            return dev_token
        local_login_token = _resolve_local_login_token_from_persisted_storage()
        if local_login_token:
            return local_login_token
        raise SandboxRuntimeError(
            "Sandbox auth token is missing. Please run `edgeone login` or set EDGEONE_PAGES_API_TOKEN in dev mode.",
            "SANDBOX_AUTH_TOKEN_MISSING",
        )
    raise SandboxRuntimeError(
        "Sandbox sealed token is missing. Ensure CI runs `edgeone pages replace` with sandbox sealed token before deployment.",
        "SANDBOX_AUTH_TOKEN_MISSING",
    )


def _resolve_sandbox_api_env(env: Mapping[str, str], configured_api_env: str = "") -> str:
    for value in (
        env.get("SANDBOX_API_ENV"),
        configured_api_env,
        env.get("API_ENV"),
    ):
        if isinstance(value, str) and value.strip():
            return value.strip()
    return ""


@dataclass
class EoContext:
    """EdgeOne 请求元数据，与 Node Agent 的 AgentEoContext 对齐。

    包含从 eo-connecting-geo / eo-connecting-ip 请求头解析的地理位置信息。
    在 adapter 层解析后传入 runtime，挂载到 ctx.eo 和 ctx.request.eo。
    """
    geo: dict = field(default_factory=dict)
    client_ip: str = ""


@dataclass
class RequestInfo:
    body: dict
    headers: dict
    query: dict  # URL query parameters, e.g. ?foo=bar → {"foo": "bar"}
    signal: asyncio.Event  # is_set() → cancelled
    eo: EoContext = field(default_factory=EoContext)

    @property
    def is_cancelled(self) -> bool:
        return self.signal.is_set()


@dataclass
class AbortActiveRunResult:
    """``ctx.utils.abortActiveRun`` 的返回值。

    与 Node ``AbortActiveRunResult`` 字段命名对齐（snake_case 版本）。

    Attributes:
        aborted: 是否真的取消了一个活跃 run。
        conversation_id: 被取消的 run 所属 conversation（仅 aborted=True 时有值）。
        run_id: 被取消的 run id（仅 aborted=True 时有值）。
    """

    aborted: bool
    conversation_id: Optional[str] = None
    run_id: Optional[str] = None


@dataclass
class AgentContext:
    conversation_id: str
    run_id: str
    parent_run_id: str | None
    request: RequestInfo
    env: dict
    kv: Any  # Per-route KV store (BlobBackedStore or InMemoryStore)
    agents: Any  # AgentsApi (avoid circular import)
    # EdgeOne 请求元数据，与 Node Agent ctx.eo 对齐
    # 包含 geo（地理位置）和 client_ip（客户端 IP），从 eo-connecting-geo/ip 请求头解析
    eo: EoContext = field(default_factory=EoContext)
    # 同 conversation 的活跃 run_id（如果有）。供业务自管并发场景使用：
    # 对齐 Node `AgentContext.active_run_id`，业务在自定义 stop / cancel handler 中
    # 可以根据这个字段判断是否需要主动 abort 旧 run；非 index 路由不会自动 409。
    active_run_id: str | None = None
    _store_blob: Any = None  # Raw blob store for ctx.store (set by runtime)
    _tracer: Any = None  # AgentTracer or NoOpTracer, injected by runtime
    _deadline: float = 0.0
    _depth: int = 0
    _call_chain: set[str] = field(default_factory=set)
    # 当前 handler 所属的 route_path（仅用于 invoke 时检测「自调用自己」），
    # 与 Node `executeRoute(input.currentRoutePath)` 的 self-recursion 检查对齐。
    _call_chain_self: str = ""
    # runtime 注入的 abortActiveRun 回调。已自动排除当前 run_id 自身，避免误取消自己。
    # 业务侧通过 ctx.utils.abortActiveRun(conversation_id) 调用，与 Node
    # `ctx.utils.abortActiveRun` 一致。runtime 没注入时（极少见，比如单元测试
    # 手搓 ctx）调用会得到 aborted=False。
    _abort_active_run_fn: Optional[Callable[[str], "AbortActiveRunResult"]] = field(
        default=None, repr=False
    )
    _loop: Any = field(default=None, repr=False)  # asyncio event loop for sync wrapper
    _sandbox: Any = field(default=None, init=False, repr=False)
    _tools: Any = field(default=None, init=False, repr=False)

    @property
    def sandbox(self):
        """沙箱原子 API（懒加载：首次访问时触发 acquire）"""
        if self._sandbox is None:
            from pages_agent_toolkit import build_sandbox
            try:
                from .runtime_config import SANDBOX_TIMEOUT, AGENT_MODE, SANDBOX_API_ENV
            except Exception:
                SANDBOX_TIMEOUT = ""
                AGENT_MODE = "prod"
                SANDBOX_API_ENV = ""
            import logging as _logging
            _logging.warning(
                "[sandbox-debug] build_sandbox env: "
                "SANDBOX_SITE=%r, EDGEONE_PAGES_API_REGION=%r, "
                "SANDBOX_API_BASE=%r, SANDBOX_REGION=%r, "
                "TENCENTCLOUD_REGION=%r, SANDBOX_API_ENV=%r, "
                "API_ENV=%r, resolved_site=%r, resolved_api_env=%r",
                self.env.get("SANDBOX_SITE"),
                self.env.get("EDGEONE_PAGES_API_REGION"),
                self.env.get("SANDBOX_API_BASE"),
                self.env.get("SANDBOX_REGION"),
                self.env.get("TENCENTCLOUD_REGION"),
                self.env.get("SANDBOX_API_ENV"),
                self.env.get("API_ENV"),
                self.env.get("SANDBOX_SITE") or self.env.get("EDGEONE_PAGES_API_REGION"),
                _resolve_sandbox_api_env(self.env, SANDBOX_API_ENV),
            )
            object.__setattr__(self, "_sandbox", build_sandbox(
                conversation_id=self.conversation_id,
                api_base=self.env.get("SANDBOX_API_BASE", ""),
                auth_token=_resolve_sandbox_auth_token(self.env, AGENT_MODE),
                project_id=(
                    self.env.get("PROJECT_ID")
                    or self.env.get("EDGEONE_PROJECT_ID")
                    or self.env.get("ProjectId")
                    or self.env.get("PAGES_PROJECT_ID")
                ),
                deployment_id=(
                    self.env.get("DEPLOYMENT_ID")
                    or self.env.get("EDGEONE_DEPLOYMENT_ID")
                    or self.env.get("PAGES_DEPLOYMENT_ID")
                ),
                region=self.env.get("SANDBOX_REGION") or self.env.get("TENCENTCLOUD_REGION"),
                site=self.env.get("SANDBOX_SITE") or self.env.get("EDGEONE_PAGES_API_REGION"),
                timeout=SANDBOX_TIMEOUT or None,
                sandbox_domain=self.env.get("SANDBOX_DOMAIN"),
                api_env=_resolve_sandbox_api_env(self.env, SANDBOX_API_ENV),
            ))
        return self._sandbox

    @property
    def tools(self):
        """预注册工具（按 framework 自动包装）"""
        if self._tools is None:
            from pages_agent_toolkit import build_tools
            try:
                from .runtime_config import AGENT_FRAMEWORK
            except Exception:
                AGENT_FRAMEWORK = ""
            # 不在此硬编码 framework 默认值：留空时由 pages_agent_toolkit 的
            # build_tools 负责应用其规范默认值，避免 CLI 与 toolkit 默认值漂移、
            # 也避免与已安装 toolkit 版本耦合。
            framework = AGENT_FRAMEWORK or self.env.get("AGENT_FRAMEWORK") or ""
            object.__setattr__(self, "_tools", build_tools(framework, self.sandbox))
        return self._tools

    @property
    def kv_sync(self) -> Any:
        """同步版 KV store —— 在 run_in_executor 等非 event-loop 线程中使用。

        首次访问时懒创建 SyncStoreWrapper 并缓存。

        典型用法::

            async def handler(ctx):
                import asyncio
                def blocking_work():
                    data = ctx.kv_sync.get("key")   # 同步调用
                    ctx.kv_sync.set("result", data)
                    return data
                result = await asyncio.get_running_loop().run_in_executor(None, blocking_work)
        """
        cached = getattr(self, '_kv_sync_cached', None)
        if cached is not None:
            return cached
        if self._loop is None:
            raise RuntimeError(
                "kv_sync 不可用：event loop 引用缺失。"
                "通常是因为 AgentContext 构造时未传入 _loop 参数。"
            )
        from .store import SyncStoreWrapper
        wrapper = SyncStoreWrapper(self.kv, self._loop)
        object.__setattr__(self, '_kv_sync_cached', wrapper)
        return wrapper

    @property
    def store(self) -> "ConversationMemory":
        """会话存储 — 消息历史 CRUD、LangGraph checkpointer/store 适配器。

        首次访问时惰性构造，使用独立的 blob store（与 per-route 的 ctx.kv 分开）。

        典型用法::

            async def handler(ctx):
                # 追加 user 消息
                msg_id = await ctx.store.append_message(
                    ctx.conversation_id, "user", "Hello!"
                )
                # 获取对话历史（默认时间升序，方便直接拼 prompt）
                messages = await ctx.store.get_messages(ctx.conversation_id)
                # 转换为 OpenAI 格式
                openai_msgs = ctx.store.to_openai_input(messages)
                # LangGraph 适配器
                checkpointer = ctx.store.langgraph_checkpointer
                store = ctx.store.langgraph_store
        """
        cached = getattr(self, '_store_cached', None)
        if cached is not None:
            return cached
        if self._store_blob is None:
            raise RuntimeError(
                "ctx.store 不可用：blob store 未配置。"
                "通常是因为运行时未设置 store。"
            )
        from .memory import ConversationMemory
        instance = ConversationMemory(self._store_blob, self.run_id)
        object.__setattr__(self, '_store_cached', instance)
        return instance

    @property
    def memory(self) -> "ConversationMemory":
        """.. deprecated:: Use ctx.store instead.

        会话存储的旧入口，请迁移至 ctx.store。首次访问时会打印一次 deprecation warning。
        """
        if not getattr(self, '_memory_deprecated_warned', False):
            import warnings
            warnings.warn(
                "ctx.memory is deprecated, use ctx.store instead. "
                "ctx.memory will be removed in a future version.",
                DeprecationWarning,
                stacklevel=2,
            )
            object.__setattr__(self, '_memory_deprecated_warned', True)
        return self.store

    @property
    def store_sync(self) -> Any:
        """.. deprecated:: Use ctx.kv_sync instead.

        旧版同步 KV store 入口，请迁移至 ctx.kv_sync。
        """
        if not getattr(self, '_store_sync_deprecated_warned', False):
            import warnings
            warnings.warn(
                "ctx.store_sync is deprecated, use ctx.kv_sync instead. "
                "ctx.store_sync will be removed in a future version.",
                DeprecationWarning,
                stacklevel=2,
            )
            object.__setattr__(self, '_store_sync_deprecated_warned', True)
        return self.kv_sync

    @property
    def utils(self) -> "ContextUtils":
        """平台工具函数命名空间，无需任何 import 即可使用。

        典型用法（SSE 流式响应）::

            async def handler(ctx):
                async def gen():
                    yield ctx.utils.sse({"prompt": prompt}, event="start")
                    for token in stream_llm():
                        yield ctx.utils.sse({"token": token}, event="delta")
                    yield ctx.utils.sse({"done": True}, event="end")
                return ctx.utils.stream_sse(gen())

        典型用法（中断目标 conversation 的活跃 run，与 Node
        ``ctx.utils.abortActiveRun`` 命名一致）::

            async def stop_handler(ctx):
                target = ctx.request.body.get("conversation_id") or ""
                result = ctx.utils.abortActiveRun(target)
                return {
                    "status": "aborted" if result.aborted else "idle",
                    "conversation_id": result.conversation_id,
                    "run_id": result.run_id,
                }

        可用方法：
          - ctx.utils.sse(data, *, event, id, retry) → bytes   构造一帧 SSE bytes
          - ctx.utils.stream_sse(gen, **kwargs) → StreamResponse  SSE 流式响应
          - ctx.utils.stream(gen, **kwargs) → StreamResponse      自定义流式响应
          - ctx.utils.abortActiveRun(conversation_id) → AbortActiveRunResult
            (alias: ctx.utils.abort_active_run)
        """
        cached = getattr(self, '_utils_cached', None)
        if cached is not None:
            return cached
        instance = ContextUtils(abort_callback=self._abort_active_run_fn)
        object.__setattr__(self, '_utils_cached', instance)
        return instance

    @property
    def tracer(self) -> Any:
        """手动埋点 API — 对齐 Node context.tracer。

        典型用法::

            async def handler(ctx):
                async def do_work(span):
                    span.event("start", {"input": user_input})
                    # ... LLM call (auto-instrumented spans become children) ...
                    return result
                return await ctx.tracer.span("my-task", do_work)

        可用方法：
          - ctx.tracer.start_span(name, attrs) -> TracerSpan
          - ctx.tracer.span(name, fn, attrs) -> result
          - ctx.tracer.event(name, attrs)
          - ctx.tracer.set_attributes(attrs)
          - ctx.tracer.record_exception(err, attrs)
        """
        cached = getattr(self, '_tracer_cached', None)
        if cached is not None:
            return cached
        tracer = self._tracer
        if tracer is None:
            try:
                from ._observability.tracer import get_agent_tracer
                tracer = get_agent_tracer()
            except (ImportError, Exception):
                tracer = _NoOpTracerFallback()
        object.__setattr__(self, '_tracer_cached', tracer)
        return tracer


# ─── Streaming response types ───
#
# 设计原则：
#   1. handler 既可以返回普通对象（一次性响应，旧行为），也可以返回
#      async generator / StreamResponse（流式响应，新行为）。runtime
#      用 isinstance + inspect.isasyncgen 区分。
#   2. StreamResponse 显式包装时允许设置 status / headers / content_type，
#      用于 SSE / NDJSON / 自定义二进制等场景。
#   3. async generator 是最简形式：handler 直接 `yield chunk`，runtime
#      用默认 200 + content-type=application/octet-stream（除非 chunk
#      本身是 str，那就 text/plain）。
#   4. chunk 类型：bytes 直接发，str 自动 utf-8 编码，dict/list 自动
#      `json.dumps + \n` 包成 NDJSON 帧（业务无需手动序列化）。
#   5. SSE 帮一把：sse(data, event=..., id=...) 返回符合 SSE 规范的
#      bytes 帧（`data: xxx\n\n`），业务在 generator 里 `yield sse(...)`
#      就行。

StreamChunk = Union[bytes, str, Mapping[str, Any], Iterable[Any]]


@dataclass
class StreamResponse:
    """显式流式响应 —— 用于需要自定义 status/headers/content-type 的场景。

    body 必须是 async iterator (`async def gen(): yield ...`)，runtime 会
    逐 chunk 发到 ASGI send。

    典型用法（SSE）：

        from _platform.context import StreamResponse, sse

        async def handler(ctx):
            async def gen():
                for token in stream_llm():
                    yield sse({"token": token})
                yield sse({"done": True})
            return StreamResponse(gen(), content_type="text/event-stream")

    若只需要默认 SSE，可以直接 `return StreamResponse.sse(gen())`。
    """

    body: AsyncIterator[StreamChunk]
    status: int = 200
    headers: Optional[dict[str, str]] = None
    content_type: Optional[str] = None

    @classmethod
    def sse(
        cls,
        body: AsyncIterator[StreamChunk],
        *,
        status: int = 200,
        extra_headers: Optional[dict[str, str]] = None,
    ) -> "StreamResponse":
        """构造 SSE 响应：自动设 content-type=text/event-stream + 反缓冲 headers。

        body 里 yield 的 chunk 应该是 sse(...) 输出的 bytes（已含 `data:` 前缀
        和 `\\n\\n` 分隔）。如果 yield 的是 dict / str，runtime 会按 SSE 规范
        自动包装一次（防呆）。
        """
        # 反缓冲：禁止网关 / 反代 / 浏览器中间层做 buffering，否则 SSE
        # 实时性会被中间层拖死。X-Accel-Buffering 是 nginx 的非标但事实
        # 标准；Cache-Control: no-cache 防 CDN 缓存；Connection: keep-alive
        # 大多数中间件已默认，写出来更明确。
        headers = {
            "cache-control": "no-cache, no-transform",
            "x-accel-buffering": "no",
            "connection": "keep-alive",
        }
        if extra_headers:
            headers.update(extra_headers)
        return cls(
            body=body,
            status=status,
            headers=headers,
            content_type="text/event-stream; charset=utf-8",
        )


def sse(
    data: Any,
    *,
    event: Optional[str] = None,
    id: Optional[str] = None,
    retry: Optional[int] = None,
) -> bytes:
    """构造一帧符合 SSE 规范的 bytes 帧。

    SSE 帧格式（每行一个字段，空行结束一帧）：
        event: <name>
        id: <id>
        retry: <ms>
        data: <line1>
        data: <line2>
        <empty line>

    data 自动序列化：
      - str → 直接作为 data 行（按 \\n 拆成多 data: 行）
      - dict / list / 其他 → json.dumps（ensure_ascii=False）

    多行 data：SSE 规范要求每行单独 `data: ` 前缀，浏览器收到后再用 \\n 拼。

    返回 bytes 而不是 str，便于直接 yield 进 generator。
    """
    parts: list[str] = []
    if event is not None:
        parts.append(f"event: {event}")
    if id is not None:
        parts.append(f"id: {id}")
    if retry is not None:
        parts.append(f"retry: {int(retry)}")

    # data 序列化
    if isinstance(data, str):
        text = data
    elif isinstance(data, (bytes, bytearray)):
        # bytes 形式：要求是 utf-8 文本数据；二进制内容 SSE 本身不支持，
        # 只能让用户自行 base64。
        text = bytes(data).decode("utf-8", errors="replace")
    else:
        try:
            text = json.dumps(data, ensure_ascii=False, separators=(",", ":"))
        except (TypeError, ValueError):
            text = str(data)

    # SSE 规范：data 字段按 \n 拆行，每行加 data: 前缀
    for line in text.split("\n"):
        parts.append(f"data: {line}")

    # 帧之间用空行分隔
    frame = "\n".join(parts) + "\n\n"
    return frame.encode("utf-8")


# ─── ContextUtils — ctx.utils 命名空间 ───

class ContextUtils:
    """挂在 AgentContext.utils 上的平台工具集合。

    业务侧通过 ``ctx.utils.<method>`` 调用，无需任何额外 import。
    所有方法都是对模块级 ``sse`` / ``StreamResponse`` 等的薄封装，零额外开销。

    与 Node agent 的 ``ctx.utils`` 入口一一对齐：
      - ``abortActiveRun(conversation_id)``：取消同实例内目标 conversation 的活跃 run。
        runtime 注入回调时已自动排除当前 run_id 自身，不会误取消自己。
        Python 这边额外保留 snake_case 别名 ``abort_active_run`` 以贴合 Python 风格。

    ``abort_callback`` 由 runtime 在构建 context 时注入；如果为 None（极少见，
    比如单测里手搓 AgentContext），调用 ``abortActiveRun`` 会得到
    ``AbortActiveRunResult(aborted=False)``，不会抛异常。
    """

    def __init__(
        self,
        abort_callback: Optional[Callable[[str], "AbortActiveRunResult"]] = None,
    ) -> None:
        self._abort_callback = abort_callback

    # ── SSE ──

    def sse(
        self,
        data: Any,
        *,
        event: Optional[str] = None,
        id: Optional[str] = None,
        retry: Optional[int] = None,
    ) -> bytes:
        """构造一帧符合 SSE 规范的 bytes，在 async generator 里 yield 使用。

        等价于模块级 ``sse(data, event=..., id=..., retry=...)``。

        示例::

            yield ctx.utils.sse({"token": "hi"}, event="delta")
            yield ctx.utils.sse("[DONE]", event="end")
        """
        return sse(data, event=event, id=id, retry=retry)

    # ── StreamResponse 工厂 ──

    def stream_sse(
        self,
        body: AsyncIterator[StreamChunk],
        *,
        status: int = 200,
        extra_headers: Optional[dict[str, str]] = None,
    ) -> StreamResponse:
        """构造 SSE 流式响应（自动设 content-type + 反缓冲 headers）。

        等价于 ``StreamResponse.sse(body, ...)``.

        示例::

            return ctx.utils.stream_sse(gen())
        """
        return StreamResponse.sse(body, status=status, extra_headers=extra_headers)

    def stream(
        self,
        body: AsyncIterator[StreamChunk],
        *,
        status: int = 200,
        content_type: Optional[str] = None,
        headers: Optional[dict[str, str]] = None,
    ) -> StreamResponse:
        """构造自定义流式响应（可指定 status / content_type / headers）。

        等价于 ``StreamResponse(body, ...)``.

        示例::

            return ctx.utils.stream(gen(), content_type="application/x-ndjson")
        """
        return StreamResponse(body=body, status=status, content_type=content_type, headers=headers)

    # ── Abort ──

    def abortActiveRun(self, conversation_id: str) -> "AbortActiveRunResult":  # noqa: N802 — 与 Node 命名对齐
        """主动取消同实例内 **目标 conversation** 的活跃 run。

        与 Node ``ctx.utils.abortActiveRun(conversation_id)`` 命名/语义一致：
          - 必须显式传 ``conversation_id``，避免同实例多会话场景下误取消其他会话；
          - 调用方自身的 run **不会被取消**（runtime 注入回调时排除了 current run_id）；
          - 找不到匹配的活跃 run 时返回 ``aborted=False``，不视为错误。

        Python 风格的别名见 :meth:`abort_active_run`。

        示例（业务自定义 ``/stop`` 路由）::

            async def stop(ctx):
                target = ctx.request.body.get("conversation_id") or ""
                r = ctx.utils.abortActiveRun(target)
                return {"status": "aborted" if r.aborted else "idle",
                        "conversation_id": r.conversation_id,
                        "run_id": r.run_id}
        """
        if not conversation_id or self._abort_callback is None:
            return AbortActiveRunResult(aborted=False)
        return self._abort_callback(conversation_id)

    def abort_active_run(self, conversation_id: str) -> "AbortActiveRunResult":
        """:meth:`abortActiveRun` 的 snake_case 别名（Python 风格）。

        两者完全等价；推荐 ``ctx.utils.abortActiveRun(...)`` 与 Node 端命名保持一致，
        在以 Python 为主的代码库里也可以用 ``ctx.utils.abort_active_run(...)``。
        """
        return self.abortActiveRun(conversation_id)


# ─── No-op tracer fallback (when _observability/ not present) ───

class _NoOpTracerFallback:
    """Minimal no-op tracer for when _observability/ is not deployed."""

    def start_span(self, name: str, attributes=None):
        return self

    async def span(self, name, fn, attributes=None):
        import asyncio
        result = fn(self)
        if asyncio.iscoroutine(result):
            return await result
        return result

    def event(self, name: str, attributes=None) -> None:
        pass

    def set_attribute(self, key: str, value) -> None:
        pass

    def set_attributes(self, attributes) -> None:
        pass

    def record_exception(self, error, attributes=None) -> None:
        pass

    def end(self) -> None:
        pass

    @property
    def span_id(self) -> str:
        return "0" * 16