"""cheap_otlp_sink.py — 便宜档主生成路 · OTLP span sink(阶段四观测 波③ · 工单 T3-1)。

【这份解决什么问题】
  便宜档 MVP 主生成路(Service `cheap_service_app` + CLI `cheap_studio`)现在只经
  observability.trace.make_jsonl_sink 把每步 trace 写进产物目录的 trace.jsonl,**不产 OTLP span**,
  观测栈(Tempo)看不到便宜档生成。本模块新接一个 OTLP span sink:与 make_jsonl_sink 同签名
  (吃一条 TraceStep dict),把每步推成一个 OTLP span 发给 mini-infra 的 OTel Collector
  (100.64.0.8:4318 · OTLP/HTTP),让便宜档生成在 Tempo 里自成一条 trace(端到端挂到 Java trace 下
  归波④,本波先让它自产 span)。jsonl 落盘保留不动,OTLP 只是额外发一份(见 build_trace_sink 并接)。

【与 tier2 studio_sink 的关系(照它的形态写,三点不同 + 一处生命周期改进)】
  形态照 tier2 observability/studio_sink.py(全仓唯一 OTLP exporter):私有 TracerProvider +
  OTLPSpanExporter(HTTP) + BatchSpanProcessor + best-effort 降级 + 惰性 import otel + **私有 provider
  不抢 otel 全局单例**。四点差异:
    ① 端点指 **Collector**(http://100.64.0.8:4318)而非 AgentScope Studio 私有端点;
    ② service.name = **cheap-gen-worker**(Studio 那条是 tier2-gen-worker),Tempo 按它归类便宜档这条线;
    ③ 端点来源默认关(env CHEAP_OTLP_ENDPOINT / infra_config,取不到即 no-op),开关默认态零影响;
    ④ **provider 做进程级单例、不做 per-run**:便宜档 Service 是长驻进程,中间件工厂 per-turn 建 sink;
       若照 studio_sink 每 run 建一个私有 TracerProvider + BatchSpanProcessor 后台线程,长驻进程会
       逐局泄漏 provider/线程。故这里把 provider 建成进程级共享单例(trace_id 只作 per-span 属性),
       BatchSpanProcessor 一条后台线程跨局复用,进程退出时经单一 atexit 兜底 flush。
  标准 gen-ai 语义键(gen_ai.conversation.id / gen_ai.usage.* / gen_ai.tool.name)与 studio_sink **保持一致**,
  便于将来在 Tempo 里按 conversation.id 跨线归组;service.name 有意不同(按线区分)。

【best-effort 铁律(设计 §8 安全红线:观测栈挂掉绝不能咬生成主链)】
  - 取不到 endpoint → make_cheap_otlp_sink 返回 no-op(现有便宜档生成一字节不变);
  - 建 provider / 发 span 失败(缺 otel wheel / Collector 不可达 / 网络错)→ best-effort 记一次告警、
    永久降级 no-op、绝不抛、绝不阻断生成;
  - 本 sink 只「读」TraceStep 往外发,从不回写任何生成 / 门判定字段(observe-only)。

【惰性 import(6c6g / cheap_service_app 顶层惰性红线)】
  opentelemetry SDK / exporter 只在函数体内 import,本模块顶层零重依赖 —— 对齐 cheap_service_app
  「顶层绝不 import agentscope / 重依赖」。真正建 provider / 发 span 只在装了 otel wheel 的 mini-desktop / Mac 发生。

【代理旁路(工单 T3-4)】
  OTLP 打 100.64.0.8:4318 走 Tailscale,必须绕本机 clash fake-ip 代理(198.18.x),否则 exporter 被拦。
  建 exporter 前复用 worker.client.install_proxy_bypass 把 Collector host 并入 NO_PROXY —— 与 new-api /
  本地 Service 旁路同源(便宜档 new-api 网关恰好也在 100.64.0.8,host 级 NO_PROXY 已覆盖,这里再显式装一次幂等兜底)。

【开关 / 地址来源】(默认关闭 = 取不到即 no-op)
  ① env CHEAP_OTLP_ENDPOINT(单点覆盖,临时开一次最省事;内网恢复后设为 http://100.64.0.8:4318);
  ② service/infra_config.py 的 [observability].otlp_endpoint(若该分区/项存在;infra.yaml 内网入仓口径);
  ③ 显式传给 make_cheap_otlp_sink(endpoint=...) 的参数。
  三者都没有 → resolve_otlp_endpoint() 返回 None → make_cheap_otlp_sink 返回 no-op sink(现有真跑不受影响)。
"""

from __future__ import annotations

import contextlib
import os
import threading
from typing import Any, Callable, Optional


# ── env 开关键(与 genconfig / infra_config 的 env 风格一致:全大写、CHEAP_ 前缀)──
#   取到非空即视为「开便宜档 OTLP 观测」并用作 Collector OTLP/HTTP base。
ENV_OTLP_ENDPOINT = "CHEAP_OTLP_ENDPOINT"

# ── 便宜档这条 trace 在 Collector/Tempo 里的服务名(OTLP resource service.name;Tempo 按它聚合归类)──
SERVICE_NAME = "cheap-gen-worker"

# ── Collector OTLP/HTTP 规范端点(内网恢复后把 env CHEAP_OTLP_ENDPOINT 设成它即开;默认不自动开)──
#   仅作文档常量:resolve_otlp_endpoint 不拿它当兜底(拿它兜底=默认自动开,违反「默认关」),开关必须显式给。
DEFAULT_COLLECTOR_ENDPOINT = "http://100.64.0.8:4318"

# ── 一次性告警去重(同一 tag 只 print 一次,避免长 run 刷屏;对齐 studio_sink._warn_once 风格)──
_WARNED: set[str] = set()

# ── 进程级共享 TracerProvider 单例(差异④:长驻 Service 逐局复用一条后台上报线程,不 per-run 泄漏)──
#   _shared 记 provider/tracer/endpoint/atexit 标记;_shared_lock 保并发建单例线程安全(Service 并发多局)。
_shared: dict = {"provider": None, "tracer": None, "endpoint": None, "atexit": False, "failed": False}
_shared_lock = threading.Lock()

# 工具结果状态 → span ERROR 的判据(observe 段 verdict 决定 span 颜色;只如实反映、不改任何生成判定)。
_ERROR_VERDICTS = {"error", "interrupted", "denied"}


def _warn_once(tag: str, msg: str) -> None:
    """同一 tag 的告警只落一次(best-effort 告警去重,绝不抛)。"""
    if tag in _WARNED:
        return
    _WARNED.add(tag)
    print(f"[cheap-otlp-sink] {msg}", flush=True)


def resolve_otlp_endpoint(explicit: Optional[str] = None) -> Optional[str]:
    """解析 Collector OTLP/HTTP base:env > infra_config[observability].otlp_endpoint > 显式参数;都没有 → None(=关闭)。

    :param explicit: 调用方显式传入的 endpoint(最低优先级,作兜底)。
    :return: 归一化后的 OTLP base(去尾斜杠);取不到任何来源 → None(make_cheap_otlp_sink 据此降级 no-op)。
    """
    # ① env 单点覆盖(最高优先级)。
    env_val = os.environ.get(ENV_OTLP_ENDPOINT)
    if env_val and env_val.strip():
        return env_val.strip().rstrip("/")

    # ② infra_config.py 的 [observability].otlp_endpoint(best-effort:读配置任何异常都不阻断)。
    try:
        # 包内/直跑兼容导入(直跑时 service 包仍可解析;_bootstrap 已把 tier2/gen-worker 加进 sys.path)。
        try:
            from service import infra_config  # type: ignore  # noqa: PLC0415
        except Exception:  # pragma: no cover —— 直跑兜底:把 gen-worker/ 加进 sys.path 再取
            import sys
            from pathlib import Path

            sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tier2" / "gen-worker"))
            from service import infra_config  # type: ignore  # noqa: PLC0415

        cfg_val = infra_config.get("observability", "otlp_endpoint", None)
        if cfg_val and str(cfg_val).strip():
            return str(cfg_val).strip().rstrip("/")
    except Exception as exc:  # best-effort:读配置任何异常都不阻断,落一次告警后继续
        _warn_once("infra_config", f"读 infra_config[observability].otlp_endpoint 失败(忽略):{exc}")

    # ③ 显式参数兜底。
    if explicit and explicit.strip():
        return explicit.strip().rstrip("/")

    return None


# ══════════════════════════════════════════════════════════════════════════════
#  跨进程 W3C 传播(阶段四观测 · 面四三跳)—— 把一次便宜档生成串成单条 trace 树
# ──────────────────────────────────────────────────────────────────────────────
#  一次生成的真实链路是三跳:Java 派发(:9501)→ worker_service 收 job → worker 打本机
#  cheap-service(:8300)真生成。三个断点原本让 trace 断成散落 root:worker 收到的
#  traceparent 没人提取、worker 出站 :8300 的 header 只有 X-User-Id、cheap-service 的生成
#  span 是私有 root 不读入站 context。下面三件把 W3C traceparent 接起来(全 best-effort,
#  otel/传播任何环节挂掉都退化成「少串一段 trace」,绝不咬生成——设计 §8 旁路铁律)。
# ══════════════════════════════════════════════════════════════════════════════

def _extract_remote_context(carrier: Optional[dict]) -> Any:
    """把入站 W3C carrier(traceparent[/tracestate])extract 成一个 OTel parent context。

    :param carrier: {'traceparent': ..., 'tracestate'?: ...};None / 无 traceparent → 返回 None。
    :return: 供 start_span(context=...) 当父的 OTel Context;取不到 / otel 不可用 → None(退回私有 root)。
    """
    try:
        if not carrier or not carrier.get("traceparent"):
            return None
        from opentelemetry import propagate  # noqa: PLC0415 —— 惰性 import(顶层零重依赖)

        return propagate.extract(carrier)
    except Exception:  # noqa: BLE001 —— 提取是旁路,失败退回无父,绝不咬主链
        return None


def current_traceparent_carrier() -> dict:
    """把「当前 OTel context」注入成一份 W3C carrier(dict:traceparent[/tracestate]),供出站带上(断点②)。

    便宜档 worker 打本机 cheap-service 的 httpx 请求、以及写给工厂的会话 sidecar,都用这份 carrier 把
    「worker 这跳的 span context」透传下去,让 cheap-service 端的生成 span 挂到同一条 trace 树下。
    纯增量、best-effort:otel 不可用或当前无 context → 返回 {}(出站少带一个 header,不影响生成)。
    """
    try:
        from opentelemetry import propagate  # noqa: PLC0415 —— 惰性 import(顶层零重依赖)

        carrier: dict = {}
        propagate.inject(carrier)  # 读当前 context → 写 traceparent(/tracestate)进 carrier
        return carrier
    except Exception:  # noqa: BLE001 —— 注入是旁路,失败返回空 carrier,绝不咬主链
        return {}


@contextlib.contextmanager
def worker_span_scope(carrier: Optional[dict], *, conversation_id: Any = None,
                      business_trace_id: Any = None):
    """便宜档 worker 这跳(:9501 收 Java 派发的 job)的 server span 作用域(断点①)。

    背景:worker_service 是裸标准库 http.server,OTel 不给它自动埋点;job 又经内部有界队列跨线程处理,
      故 do_POST 只把入站 traceparent 抓进 job(self.headers 仅那里可得),真正的 worker span 在 worker
      线程处理该 job 时(本作用域)才开——span 恰好覆盖「调 cheap-service 生成」这一跳、并成为下游的父。
    行为:
      · OTLP 已开(进程级共享 tracer 可用):在入站 context 下开一个 cheap.worker.generate 的 SERVER span、
        attach 为当前 context;下游(driver→:8300 的 httpx inject、写工厂 sidecar 的 traceparent)据此继承。
      · OTLP 未开 / 建 span 失败:只把入站 context attach 为当前(无 span),让下游 inject 仍能把 Java 的
        traceparent 原样透传出去(纯增量、不依赖 OTLP 开关);无入站 context 时退化成什么都不做。
    §10 决策5「业务 traceId 并存」:业务 traceId(承担产物归位/回调关联)作 span 属性保留、不改其格式;
      W3C traceparent 只负责跨进程贯穿。
    best-effort 铁律(设计 §8):提取 / 建 span / attach 全程吞异常,绝不让传播处理影响生成成败或阻塞。
    """
    parent_ctx = _extract_remote_context(carrier)

    # 尝试拿进程级共享 tracer(仅 OTLP 开时可用;复用波③ 同一 provider,不新造 root)。
    span = None
    try:
        endpoint = resolve_otlp_endpoint()
        if endpoint:
            tracer = _get_shared_tracer(endpoint)
            if tracer is not None:
                from opentelemetry.trace import SpanKind  # noqa: PLC0415

                span = tracer.start_span(
                    "cheap.worker.generate", context=parent_ctx, kind=SpanKind.SERVER)
                # 业务 traceId 并存(§10 决策5):gameId 作 conversation.id(与 cheap-service 生成 span 同键、
                #   Tempo 按它跨线归组),业务 traceId 另存一键;都只作属性,不改其格式、不参与 W3C 贯穿。
                if conversation_id is not None:
                    span.set_attribute("gen_ai.conversation.id", str(conversation_id))
                if business_trace_id is not None:
                    span.set_attribute("agentscope.cheap.business_trace_id", str(business_trace_id))
    except Exception:  # noqa: BLE001 —— 建 span 失败退回「只 attach 入站 context」路径,绝不咬生成
        span = None

    if span is not None:
        from opentelemetry.trace import use_span  # noqa: PLC0415

        # use_span:把 worker span attach 为当前 + 退出时 end(异常也 end 并记录);下游 inject 据此继承。
        with use_span(span, end_on_exit=True):
            yield span
        return

    # OTLP 未开 / 建 span 失败:只 attach 入站 context(让下游 inject 透传 Java traceparent);无则什么都不做。
    token = None
    try:
        if parent_ctx is not None:
            from opentelemetry import context as otel_context  # noqa: PLC0415

            token = otel_context.attach(parent_ctx)
        yield None
    finally:
        if token is not None:
            try:
                from opentelemetry import context as otel_context  # noqa: PLC0415

                otel_context.detach(token)
            except Exception:  # noqa: BLE001 —— detach 失败只影响 contextvar 清理,不咬主链
                pass


def _build_provider(endpoint: str, span_exporter: Any = None) -> Any:
    """惰性建一个私有 TracerProvider(装 exporter);失败抛给上层 best-effort 兜。仅在函数体内 import otel(惰性红线)。

    :param endpoint: Collector OTLP/HTTP base(如 http://100.64.0.8:4318;真正发时拼 /v1/traces)。
    :param span_exporter: 仅测试注入用——给一个内存 exporter(走 SimpleSpanProcessor 同步导出,断言 span 属性用);
                          生产恒 None → 建真 OTLPSpanExporter(HTTP)+ BatchSpanProcessor 异步后台发。
    :return: 配好 processor 的私有 TracerProvider(不碰 otel 全局单例)。
    """
    # 惰性 import:6c6g 无 opentelemetry SDK / exporter wheel,本模块顶层仍可裸 import(此处才触发依赖)。
    from opentelemetry.sdk.resources import Resource  # noqa: PLC0415
    from opentelemetry.sdk.trace import TracerProvider  # noqa: PLC0415
    from opentelemetry.sdk.trace.export import (  # noqa: PLC0415
        BatchSpanProcessor,
        SimpleSpanProcessor,
    )

    # resource.service.name = cheap-gen-worker:Tempo 按服务名把便宜档这条线的 span 归一类。
    resource = Resource.create({"service.name": SERVICE_NAME})
    provider = TracerProvider(resource=resource)

    if span_exporter is not None:
        # 测试路:同步 processor + 内存 exporter,__call__ 后立即可断言 span 属性(无需网络 / flush 时序)。
        provider.add_span_processor(SimpleSpanProcessor(span_exporter))
        return provider

    # 生产路:代理旁路(T3-4)必须在建 OTLPSpanExporter(内部起 httpx / requests session)之前装 —— 把
    #   Collector host 并入 NO_PROXY,否则 trust_env 的 http 客户端把发往 100.64.0.8 的上报经 clash fake-ip 转走被拦。
    #   复用 worker.client.install_proxy_bypass(与 new-api / 本地 Service 旁路同源),幂等、绝不抛。
    try:
        from worker import client  # noqa: PLC0415

        client.install_proxy_bypass(endpoint)
    except Exception as exc:  # best-effort:装旁路失败不阻断建 provider(host 级 NO_PROXY 可能已由启动装过)
        _warn_once("proxy-bypass", f"装 OTLP 代理旁路失败(忽略,可能已由启动装过):{exc}")

    from opentelemetry.exporter.otlp.proto.http.trace_exporter import (  # noqa: PLC0415
        OTLPSpanExporter,
    )

    # OTLP/HTTP exporter:显式给 traces 全路径 base + /v1/traces(与 studio_sink 同策略,避免不同 SDK 版本对
    #   base/自动拼路径行为不一致;Collector 的 OTLP/HTTP receiver 收 /v1/traces)。
    exporter = OTLPSpanExporter(endpoint=endpoint.rstrip("/") + "/v1/traces")
    # 批处理 processor:异步攒批后台发,推送不卡生成主链(observe-only 不阻塞)。
    provider.add_span_processor(BatchSpanProcessor(exporter))
    return provider


def _get_shared_tracer(endpoint: str, span_exporter: Any = None) -> Any:
    """取进程级共享 tracer(惰性建一次、并发安全);建失败 best-effort 返回 None、永久降级(不每条重试)。

    差异④:便宜档 Service 长驻、中间件工厂 per-turn 建 sink,provider 若 per-run 会逐局泄漏后台线程;
    故 provider/tracer 做进程级单例,trace_id 只作 per-span 属性(见 CheapOtlpSink._emit_span)。
    """
    # 快路:已建好直接返回;已判失败直接 None(永久降级,不每条重试)。
    if _shared["tracer"] is not None:
        return _shared["tracer"]
    if _shared["failed"]:
        return None
    with _shared_lock:
        # 双检:抢锁期间别的线程可能已建好 / 已判失败。
        if _shared["tracer"] is not None:
            return _shared["tracer"]
        if _shared["failed"]:
            return None
        try:
            provider = _build_provider(endpoint, span_exporter=span_exporter)
            tracer = provider.get_tracer("cheap-gen-worker", "1.0.0")
            _shared["provider"] = provider
            _shared["tracer"] = tracer
            _shared["endpoint"] = endpoint
            # 进程退出兜底 flush:注册一次(避免 per-run 注册堆积)。BatchSpanProcessor 平时按调度发,
            #   进程退出时靠这条把攒批残留 span 尽力发出(best-effort、丢尾批可容忍,observe-only)。
            if not _shared["atexit"]:
                import atexit  # noqa: PLC0415

                atexit.register(flush_shared_tracer)
                _shared["atexit"] = True
            _warn_once("init-ok", f"便宜档 OTLP 观测已开(service.name={SERVICE_NAME} → {endpoint}/v1/traces)")
            return tracer
        except Exception as exc:  # best-effort:缺依赖 / 建 provider 失败 → 永久降级 no-op,不抛、不阻塞生成
            _shared["failed"] = True
            _warn_once("init-fail", f"建 OTLP→Collector 通道失败,便宜档降级为不发 span(只留 jsonl):{exc}")
            return None


def flush_shared_tracer() -> None:
    """进程退出 / 收口时把 BatchSpanProcessor 攒批里残留的 span 尽力发出(best-effort,绝不抛)。

    atexit 注册一次;CLI 收口也可显式调(波④ 若要精确 flush)。Service 长驻期间靠 BatchSpanProcessor 自身调度,
    无需 per-turn 调本函数(shutdown 单例 provider 会杀掉后台上报线程,长驻进程不该 per-turn 关)。
    """
    provider = _shared.get("provider")
    if provider is None:
        return
    try:
        provider.force_flush(timeout_millis=5000)  # 给上限超时,避免收口/退出卡住
    except Exception as exc:  # best-effort:flush 失败只告警
        _warn_once("flush-fail", f"force_flush 残留 span 失败(忽略):{exc}")


def reset_shared_tracer_for_test() -> None:
    """仅测试用:清进程级共享单例 + 告警去重集,让各用例互不串(生产路绝不调)。"""
    provider = _shared.get("provider")
    if provider is not None:
        try:
            provider.shutdown()
        except Exception:  # noqa: BLE001 —— 测试清理 best-effort
            pass
    _shared.update({"provider": None, "tracer": None, "endpoint": None, "failed": False})
    _WARNED.clear()


class CheapOtlpSink:
    """把统一 trace 事件(TraceStep dict)推成 OTLP span 发给 Collector 的旁路 sink(便宜档主路)。

    契约 = Callable[[dict], None](与 make_jsonl_sink 的 sink 一致):吃一条 TraceStep dict → 发一个 span。
    任何环节失败(缺 otel 依赖 / 建 provider 失败 / 发送失败)一律 best-effort:落一次告警、自动降级 no-op、绝不抛。
    span 属性映射与 tier2 studio_sink 的标准 gen-ai 键保持一致(便于将来 Tempo 跨线按 conversation.id 归组)。
    """

    enabled = True

    def __init__(self, trace_id: str, endpoint: str, *, span_exporter: Any = None,
                 parent_carrier: Optional[dict] = None) -> None:
        """
        Args:
            trace_id: 本次生成 traceId(贯穿;作 span 的 gen_ai.conversation.id,Tempo 据它把同一 run 的步归一条 trace)。
            endpoint: Collector OTLP/HTTP base(已归一化;发时拼 /v1/traces)。
            span_exporter: 仅测试注入(内存 exporter);生产恒 None → 走真 OTLP exporter。
            parent_carrier: 入站 W3C carrier(traceparent[/tracestate];面四断点③)。给定则每步生成 span 挂到
                入站 context 下当子 span(不再私有 root),使 Java→worker→cheap-service 串成单条 trace 树;
                缺省 None → 保持波③ 私有 root 行为(默认关时字节不变)。
        """
        self.trace_id = trace_id
        self.endpoint = endpoint
        self._span_exporter = span_exporter
        self._parent_carrier = parent_carrier or None
        self._parent_ctx: Any = None            # extract 出的 parent context(惰性缓存一次)
        self._parent_ctx_resolved = False

    def __call__(self, trace_step: dict) -> None:
        """吃一条 TraceStep dict → 发一个 OTLP span(全程 best-effort,绝不抛、绝不阻塞生成主链)。"""
        tracer = _get_shared_tracer(self.endpoint, span_exporter=self._span_exporter)
        if tracer is None:
            return  # 建通道失败已永久降级 no-op(告警已落),observe-only 不影响主链
        try:
            self._emit_span(tracer, trace_step)
        except Exception as exc:  # best-effort:单条 span 发送失败只告警一次,不连累后续、不抛
            _warn_once("emit-fail", f"推 OTLP span 失败(best-effort 计告警,traceId={self.trace_id}):{exc}")

    def _parent_context(self) -> Any:
        """把入站 W3C carrier extract 成 parent context(本 sink 实例缓存一次;面四断点③)。

        None(未给 carrier / extract 失败)→ start_as_current_span 用 None,退回私有 root(波③ 行为)。
        """
        if not self._parent_ctx_resolved:
            self._parent_ctx_resolved = True
            self._parent_ctx = _extract_remote_context(self._parent_carrier)
        return self._parent_ctx

    def _emit_span(self, tracer: Any, trace_step: dict) -> None:
        """把单条 TraceStep 组成一个 span 并即时收尾(同步 start→set→end,交 processor 发出)。

        映射(与 studio_sink._emit_span 结构对齐;标准 gen-ai 键一致,私有键用 agentscope.cheap.* 区分线):
          - span.name              = 取 ext.action.tool / 相位 kind / raw.event,组成可读步名(如 "step 3 · write_source")。
          - gen_ai.conversation.id = traceId(与 studio_sink 同键:Tempo 据它把同一 run 的步归一条 trace)。
          - gen_ai.usage.*         = cost.tokens 的 in/out(标准 token 用量键)。
          - gen_ai.tool.name       = 动作段工具名。
          - agentscope.cheap.*     = tier2 三段(推理/动作/观察)+ step/timestamp/verdict(JSON 序列化,便于反查)。
          - span 状态:observe 段 verdict ∈ {error/interrupted/denied} → 标 ERROR(如实反映,不改任何生成判定)。
        """
        import json  # noqa: PLC0415 —— 仅序列化属性用
        from opentelemetry.trace import StatusCode  # noqa: PLC0415

        step = trace_step.get("step")
        ext = trace_step.get("ext") or {}
        raw = ext.get("raw") or {}

        # ── span 名:优先工具名 / 相位 kind / 原始事件类型,拼上步序,便于在 Tempo 列表一眼认出这步在干嘛 ──
        label = (
            (ext.get("action") or {}).get("tool")
            or (ext.get("action") or {}).get("kind")
            or (ext.get("observation") or {}).get("kind")
            or (ext.get("reasoning") or {}).get("kind")
            or raw.get("event")
            or "step"
        )
        span_name = f"step {step} · {label}"

        # 即起即收(with 退出即 end):本步是「一条已成事实的 trace 记录」,无需跨步嵌套,也不会因异常漏 end。
        #   面四断点③:context=入站 parent → 每步生成 span 挂到 Java→worker 这条入站 trace 下当子 span
        #   (不再私有 root);parent 为 None(未开传播)时等价于原 start_as_current_span()(波③ 行为不变)。
        with tracer.start_as_current_span(span_name, context=self._parent_context()) as span:
            # traceId → gen_ai.conversation.id(与 studio_sink 同键,跨线归组用)。
            span.set_attribute("gen_ai.conversation.id", str(self.trace_id))
            if step is not None:
                span.set_attribute("agentscope.cheap.step", int(step))
            ts = trace_step.get("timestamp")
            if ts:
                span.set_attribute("agentscope.cheap.timestamp", str(ts))

            # cost.tokens → gen_ai.usage.*(标准 token 用量键)。
            cost = trace_step.get("cost") or {}
            tokens = (cost or {}).get("tokens") or {}
            if "in" in tokens:
                span.set_attribute("gen_ai.usage.input_tokens", int(tokens.get("in") or 0))
            if "out" in tokens:
                span.set_attribute("gen_ai.usage.output_tokens", int(tokens.get("out") or 0))

            # tier2 扩展三段 → agentscope.cheap.*(JSON 序列化,反查用)。
            for seg_key in ("reasoning", "action", "observation"):
                seg = ext.get(seg_key)
                if seg:
                    span.set_attribute(f"agentscope.cheap.{seg_key}", json.dumps(seg, ensure_ascii=False))
            tool = (ext.get("action") or {}).get("tool")
            if tool:
                span.set_attribute("gen_ai.tool.name", str(tool))
            if raw.get("event"):
                span.set_attribute("agentscope.cheap.raw_event", str(raw.get("event")))

            # verdict → span 状态(observe-only,不改任何生成判定)。
            verdict = trace_step.get("verdict")
            if verdict is not None:
                span.set_attribute("agentscope.cheap.verdict", str(verdict))
                if str(verdict).lower() in _ERROR_VERDICTS:
                    span.set_status(StatusCode.ERROR, str(verdict))
                else:
                    span.set_status(StatusCode.OK)

    def shutdown(self) -> None:
        """收口:flush 进程级共享 provider 的攒批残留 span(best-effort,绝不抛)。

        注意:只 flush 不 shutdown provider —— provider 是进程级单例(长驻 Service 逐局复用),不能 per-turn 关掉
        它的后台上报线程;真正 shutdown 交进程退出的 atexit(flush_shared_tracer)一次性做。
        """
        flush_shared_tracer()


class _NoopSink:
    """关闭态占位 sink:__call__ / shutdown 都不做。让调用方无需判 None、统一拿 sink。"""

    enabled = False

    def __call__(self, trace_step: dict) -> None:  # noqa: D401 —— 故意 no-op
        return

    def shutdown(self) -> None:  # noqa: D401 —— 故意 no-op
        return


def make_cheap_otlp_sink(
    trace_id: str,
    *,
    endpoint: Optional[str] = None,
    span_exporter: Any = None,
    parent_carrier: Optional[dict] = None,
) -> Callable[[dict], None]:
    """构造便宜档 OTLP 旁路 sink:解析到 endpoint → CheapOtlpSink;取不到 → no-op sink(默认关闭)。

    :param trace_id: 本次生成 traceId(贯穿)。
    :param endpoint: 显式 Collector base(最低优先级;一般留空,走 env CHEAP_OTLP_ENDPOINT / infra_config)。
    :param span_exporter: 仅测试注入(内存 exporter);生产恒 None。
    :param parent_carrier: 入站 W3C carrier(面四断点③);给定则生成 span 挂到入站 trace 下当子 span。
    :return: Callable[[dict], None] 且带 .shutdown() 的 sink(CheapOtlpSink 或 _NoopSink)。
    """
    url = resolve_otlp_endpoint(endpoint)
    if not url:
        # 取不到 endpoint = 默认关闭:返回 no-op(现有便宜档真跑不受任何影响)。
        return _NoopSink()
    return CheapOtlpSink(trace_id=trace_id, endpoint=url, span_exporter=span_exporter,
                         parent_carrier=parent_carrier)


class _FanoutSink:
    """把一条 trace_step 同时喂给多个 sink 的 fan-out 壳:每侧各 best-effort、互不连累(observe-only 不咬主链)。

    用于把既有 jsonl 落盘 sink 与新 OTLP sink 并接成「一个 sink」交给 Tier2TraceMiddleware(它只吃单 sink 契约)。
    任一侧抛异常只告警、不抛、不连累另一侧与生成主链(镜像 observability.trace.with_studio 的 _fanout 语义)。
    """

    def __init__(self, sinks: list) -> None:
        self._sinks = [s for s in sinks if s is not None]

    def __call__(self, trace_step: dict) -> None:
        for s in self._sinks:
            try:
                s(trace_step)
            except Exception as exc:  # noqa: BLE001 —— 一侧失败只告警,不连累其余 sink 与主链
                _warn_once("fanout", f"fan-out 某 sink 写失败(忽略,不连累其余):{type(exc).__name__}: {exc}")

    def shutdown(self) -> None:
        """把带 .shutdown() 的子 sink 依次收口(best-effort;jsonl sink 无 shutdown 自然跳过)。"""
        for s in self._sinks:
            fn = getattr(s, "shutdown", None)
            if callable(fn):
                try:
                    fn()
                except Exception:  # noqa: BLE001 —— 收口 best-effort
                    pass


def build_trace_sink(
    jsonl_sink: Optional[Callable[[dict], None]],
    *,
    trace_id: str,
    endpoint: Optional[str] = None,
    span_exporter: Any = None,
    parent_carrier: Optional[dict] = None,
) -> Optional[Callable[[dict], None]]:
    """把既有 jsonl sink 与(可选)OTLP sink 并接成一个交给 Tier2TraceMiddleware 的 sink。

    **默认关(取不到 endpoint):原样返回传入的 jsonl sink,零包装 —— 现有行为字节不变。**
    开(env / config / 显式给 endpoint):返回 fan-out sink(jsonl 先落盘照旧、OTLP 额外发一份,两侧各 best-effort)。
    jsonl 落盘永远保留不动;OTLP 只是旁路多发一份,挂掉绝不咬生成(设计 §8)。

    :param jsonl_sink: 既有 make_jsonl_sink 产的落盘 sink(可能为 None:上游 jsonl 接线已失败降级)。
    :param trace_id: 本次生成 traceId(贯穿)。
    :param endpoint: 显式 Collector base(一般留空,走 env / infra_config)。
    :param span_exporter: 仅测试注入。
    :param parent_carrier: 入站 W3C carrier(面四断点③);给定则 OTLP 生成 span 挂到入站 trace 下当子 span。
    :return: 交给 Tier2TraceMiddleware(sink=...) 的单 sink(jsonl 原件 / OTLP 单件 / fan-out / None)。
    """
    otlp = make_cheap_otlp_sink(trace_id, endpoint=endpoint, span_exporter=span_exporter,
                                parent_carrier=parent_carrier)
    otlp_on = not isinstance(otlp, _NoopSink)
    if not otlp_on:
        # 默认关:零包装,原样返回 jsonl(现有行为字节不变;jsonl 为 None 时也原样返回 None)。
        return jsonl_sink
    if jsonl_sink is None:
        # jsonl 上游已降级无 sink:只发 OTLP(仍不阻断)。
        return otlp
    # 两条都在:fan-out(jsonl 先落盘、OTLP 后发,各 best-effort)。
    return _FanoutSink([jsonl_sink, otlp])


# ══════════════════════════════════════════════════════════════════════════════
#  面二 · Python 侧业务 metric(OTLP metrics)—— 只补面五(game-cloud GenMetrics)拿不到的两个
# ──────────────────────────────────────────────────────────────────────────────
#  面五已从回调 trace 发 gen_task_total{status} / gen_queue_depth / gen_gate_fail_total{gate} /
#  gen_duration_seconds / llm_cost —— 成本、耗时、成功率、门失败、队列深度都归它;面二绝不重发这些(重发=双计)。
#  面二只发面五缺的两个纯 Python 侧 token 维度 metric:
#    · llm_tokens(Counter{kind=input|output|cached})—— 一次生成累计的输入/输出/缓存 token 计数;
#      Prometheus 侧规范化后呈现为 `llm_tokens_total{kind}`(counter 由 exporter 追加 `_total` 后缀,
#      是 OTel→Prometheus 的既定约定,故 instrument 本身不自带 `_total`)。
#    · llm_cache_hit_rate(Histogram)—— 一次生成的缓存命中率 cached/input([0,1]);Prometheus 侧
#      展开为 llm_cache_hit_rate_{bucket,sum,count},既能看单次命中率分布、也能按 run 聚合平均。
#  与上面 span sink 复用同一套:开关(env CHEAP_OTLP_ENDPOINT / infra_config,默认关)、端点解析、
#  代理旁路(install_proxy_bypass)、service.name(cheap-gen-worker)、best-effort 铁律;只是换 OTLP metrics 通道。
#  注:token 加权的整体命中率(Σcached/Σinput)由 Prometheus 侧用 llm_tokens_total{kind} 两条 counter 相除算,
#  这里的 histogram 记的是「每次 run 的命中率」——两者互补,一个看整体、一个看单次分布与告警。
# ══════════════════════════════════════════════════════════════════════════════

# OTel instrument 名(InMemoryMetricReader / SDK 层按这个名读;Prometheus 呈现名见上注)。
METRIC_TOKENS = "llm_tokens"                   # Counter → Prometheus llm_tokens_total{kind}
METRIC_CACHE_HIT_RATE = "llm_cache_hit_rate"   # Histogram([0,1] 每 run 命中率)

# 命中率 histogram 的桶边界:cache_hit_rate 是 [0,1] 比率,默认延迟型桶(0/5/10/…)不适用,
#   显式给一组 [0,1] 细分,让「命中率落在哪一档」在 Grafana 里可分辨。
_CACHE_HIT_RATE_BUCKETS = (0.0, 0.1, 0.25, 0.5, 0.75, 0.9, 0.99, 1.0)

# 进程级共享 MeterProvider 单例(差异④同 span:长驻 Service 逐局复用一条 PeriodicExportingMetricReader
#   后台上报线程,不 per-run 建 provider 泄漏线程)。counter/histogram instrument 一并缓存(建一次)。
_shared_meter: dict = {
    "provider": None, "counter": None, "histogram": None,
    "endpoint": None, "atexit": False, "failed": False,
}
_shared_meter_lock = threading.Lock()


def _build_meter_provider(endpoint: str, metric_reader: Any = None) -> Any:
    """惰性建私有 MeterProvider(装 OTLP metric reader);失败抛给上层 best-effort 兜。仅函数体内 import otel(惰性红线)。

    :param endpoint: Collector OTLP/HTTP base(如 http://100.64.0.8:4318;真正发时拼 /v1/metrics)。
    :param metric_reader: 仅测试注入——给一个 InMemoryMetricReader(record 后 .get_metrics_data() 立即可断言,
                          无网络无导出周期);生产恒 None → 建真 OTLPMetricExporter + PeriodicExportingMetricReader 异步后台发。
    :return: 配好 reader + service.name resource 的私有 MeterProvider(不碰 otel 全局单例)。
    """
    from opentelemetry.sdk.metrics import MeterProvider  # noqa: PLC0415
    from opentelemetry.sdk.resources import Resource  # noqa: PLC0415

    # resource.service.name = cheap-gen-worker:与 span 同服务名,Tempo/Prometheus 按它把便宜档这条线归一类。
    resource = Resource.create({"service.name": SERVICE_NAME})

    if metric_reader is not None:
        # 测试路:注入 InMemoryMetricReader,同步聚合、无导出周期,record 后立即 get_metrics_data() 断言。
        return MeterProvider(metric_readers=[metric_reader], resource=resource)

    # 生产路:代理旁路(与 span 同源 T3-4)必须在建 OTLPMetricExporter(内部起 requests session)之前装——把
    #   Collector host 并入 NO_PROXY,否则 trust_env 的 http 客户端把发往 100.64.0.8 的上报经 clash fake-ip 转走被拦。
    try:
        from worker import client  # noqa: PLC0415

        client.install_proxy_bypass(endpoint)
    except Exception as exc:  # best-effort:装旁路失败不阻断建 provider(host 级 NO_PROXY 可能已由启动/span 侧装过)
        _warn_once("proxy-bypass-metric", f"装 OTLP metric 代理旁路失败(忽略,可能已装过):{exc}")

    from opentelemetry.exporter.otlp.proto.http.metric_exporter import (  # noqa: PLC0415
        OTLPMetricExporter,
    )
    from opentelemetry.sdk.metrics.export import (  # noqa: PLC0415
        PeriodicExportingMetricReader,
    )

    # OTLP/HTTP metric exporter:显式给 metrics 全路径 base + /v1/metrics(与 span 的 /v1/traces 同策略,
    #   避免不同 SDK 版本对 base/自动拼路径行为不一致;Collector 的 OTLP/HTTP receiver 收 /v1/metrics)。
    exporter = OTLPMetricExporter(endpoint=endpoint.rstrip("/") + "/v1/metrics")
    # 周期性导出 reader:后台攒批定期发,不卡生成主链(observe-only 不阻塞)。
    reader = PeriodicExportingMetricReader(exporter)
    return MeterProvider(metric_readers=[reader], resource=resource)


def _get_shared_meter_instruments(endpoint: str, metric_reader: Any = None):
    """取进程级共享 (counter, histogram);惰性建一次、并发安全;建失败 best-effort 返回 (None, None)、永久降级(不每条重试)。

    差异④:便宜档 Service 长驻,provider 若 per-run 会逐局泄漏后台上报线程;故 MeterProvider/instrument 做进程级单例。
    """
    # 快路:已建好直接返回;已判失败直接 (None, None)(永久降级)。
    if _shared_meter["counter"] is not None:
        return _shared_meter["counter"], _shared_meter["histogram"]
    if _shared_meter["failed"]:
        return None, None
    with _shared_meter_lock:
        # 双检:抢锁期间别的线程可能已建好 / 已判失败。
        if _shared_meter["counter"] is not None:
            return _shared_meter["counter"], _shared_meter["histogram"]
        if _shared_meter["failed"]:
            return None, None
        try:
            provider = _build_meter_provider(endpoint, metric_reader=metric_reader)
            meter = provider.get_meter("cheap-gen-worker", "1.0.0")
            counter = meter.create_counter(
                METRIC_TOKENS,
                unit="{token}",
                description="便宜档生成每次 run 的 LLM token 用量(kind=input/output/cached);"
                            "面五不发 token 计数,面二补(Prometheus 侧 llm_tokens_total{kind})",
            )
            histogram = meter.create_histogram(
                METRIC_CACHE_HIT_RATE,
                unit="1",
                description="便宜档生成每次 run 的缓存命中率 cached/input([0,1]);面五不发,面二补",
                explicit_bucket_boundaries_advisory=list(_CACHE_HIT_RATE_BUCKETS),
            )
            _shared_meter.update({
                "provider": provider, "counter": counter,
                "histogram": histogram, "endpoint": endpoint,
            })
            # 进程退出兜底 flush:注册一次(避免 per-run 注册堆积)。
            if not _shared_meter["atexit"]:
                import atexit  # noqa: PLC0415

                atexit.register(flush_shared_meter)
                _shared_meter["atexit"] = True
            _warn_once("meter-init-ok",
                       f"便宜档 OTLP metric 已开(service.name={SERVICE_NAME} → {endpoint}/v1/metrics)")
            return counter, histogram
        except Exception as exc:  # best-effort:缺依赖 / 建 provider 失败 → 永久降级不发,不抛、不阻塞生成
            _shared_meter["failed"] = True
            _warn_once("meter-init-fail",
                       f"建 OTLP metric→Collector 通道失败,便宜档降级为不发 metric:{exc}")
            return None, None


def flush_shared_meter() -> None:
    """进程退出 / 收口把 PeriodicExportingMetricReader 攒批里残留的 metric 尽力发出(best-effort,绝不抛)。"""
    provider = _shared_meter.get("provider")
    if provider is None:
        return
    try:
        provider.force_flush(timeout_millis=5000)  # 给上限超时,避免收口/退出卡住
    except Exception as exc:  # best-effort:flush 失败只告警
        _warn_once("meter-flush-fail", f"force_flush 残留 metric 失败(忽略):{exc}")


def reset_shared_meter_for_test() -> None:
    """仅测试用:清进程级共享 meter 单例 + 告警去重集,让各用例互不串(生产路绝不调)。"""
    provider = _shared_meter.get("provider")
    if provider is not None:
        try:
            provider.shutdown()
        except Exception:  # noqa: BLE001 —— 测试清理 best-effort
            pass
    _shared_meter.update({
        "provider": None, "counter": None, "histogram": None,
        "endpoint": None, "failed": False,
    })
    _WARNED.clear()


def record_llm_token_metrics(
    input_tokens: int,
    output_tokens: int,
    cached_tokens: Optional[int] = None,
    *,
    endpoint: Optional[str] = None,
    metric_reader: Any = None,
) -> None:
    """一次生成 run 收口时发面二两个 token 维度 metric(默认关 / best-effort / 绝不抛、绝不阻断生成)。

    调用契约:生成 run 收尾拿到本次累计 token 后调一次;取不到 endpoint(默认关)直接返回、字节不变。

    Args:
        input_tokens:  本次 run 累计输入(prompt)token —— CLI 路 Σ RecordingModel.records[0](=usage_sum()[0]);
                       Service 路 Σ ModelCallEndEvent.input_tokens。
        output_tokens: 本次 run 累计输出(completion)token(同上,取 records[1] / ModelCallEndEvent.output_tokens)。
        cached_tokens: 本次 run 命中缓存的 prompt token —— **仅 CLI 路可得(Σ RecordingOpenAIChatModel.records[2],
                       由 usage.cache_input_tokens 归一)**;生产 Service 路的框架默认 model 无 recording wrapper、
                       ModelCallEndEvent 不带 cached,故传 **None** 表示「该路拿不到」→ 不发 cached 计数、不发
                       cache_hit_rate(如实降级,绝不塞 0 伪造成「命中率 0」)。
        endpoint:      显式 Collector base(一般留空,走 env CHEAP_OTLP_ENDPOINT / infra_config)。
        metric_reader: 仅测试注入 InMemoryMetricReader;生产恒 None → 走真 OTLP metric exporter。
    """
    url = resolve_otlp_endpoint(endpoint)
    if not url:
        # 默认关:取不到 endpoint 即不发(与 span sink 默认关同源,现有便宜档真跑字节不变)。
        return
    try:
        counter, histogram = _get_shared_meter_instruments(url, metric_reader=metric_reader)
        if counter is None:
            return  # 建通道失败已永久降级(告警已落),observe-only 不影响主链
        in_tok = int(input_tokens or 0)
        out_tok = int(output_tokens or 0)
        # token 计数:input / output 恒发(kind 属性区分)。
        counter.add(in_tok, {"kind": "input"})
        counter.add(out_tok, {"kind": "output"})
        if cached_tokens is not None:
            # cached 可得(CLI 路):发 cached 计数 + 命中率;cached 截到 [0, input](防脏数据命中数超 prompt)。
            cached = max(0, min(int(cached_tokens or 0), in_tok)) if in_tok > 0 else max(0, int(cached_tokens or 0))
            counter.add(cached, {"kind": "cached"})
            # cache_hit_rate 仅 input>0 有意义(除零无值);命中率截到 [0,1]。
            if in_tok > 0:
                histogram.record(min(1.0, cached / in_tok))
    except Exception as exc:  # best-effort:发 metric 任何异常只告警一次,绝不抛、绝不连累生成主链
        _warn_once("metric-emit-fail", f"发 OTLP metric 失败(best-effort 计告警):{exc}")


if __name__ == "__main__":  # pragma: no cover —— 本地自检:无 endpoint 时应拿到 no-op,__call__/shutdown 不炸
    s = make_cheap_otlp_sink(trace_id="demo-trace-001")
    print("sink type:", type(s).__name__)  # 期望 _NoopSink(未设 CHEAP_OTLP_ENDPOINT)
    s({"traceId": "demo-trace-001", "step": 0, "ext": {"raw": {"event": "ReplyStartEvent"}}})
    s.shutdown()
    print("noop sink ok(observe-only 关闭态:吃事件不报错、shutdown 不报错)")