otel vs analytics：两条管线对照

pub(crate) const OTEL_TARGET_PREFIX: &str = "codex_otel"; // 所有遥测目标共享这个前缀，导出器靠它判断事件归属pub(crate) const OTEL_LOG_ONLY_TARGET: &str = "codex_otel.log_only"; // log-only 通道：可含 PII（账号 ID、邮箱），只进日志导出器pub(crate) const OTEL_TRACE_SAFE_TARGET: &str = "codex_otel.trace_safe"; // trace-safe 通道：禁止 PII，可随追踪一起导出pub(crate) fn is_log_export_target(target: &str) -> bool { // 判断某事件目标是否属于日志导出器    target.starts_with(OTEL_TARGET_PREFIX) && !is_trace_safe_target(target) // 前缀匹配且不是 trace_safe——trace_safe 是前缀的子集，必须显式排除}pub(crate) fn is_trace_safe_target(target: &str) -> bool { // 判断某事件目标是否允许进追踪通道    target.starts_with(OTEL_TRACE_SAFE_TARGET) // 只有这个前缀的事件允许进追踪（trace_safe.summary 等子目标也算）}

macro_rules! log_event { // log-only 事件：允许携带 account_id / email 等 PII 字段    ($self:expr, $($fields:tt)*) => {{ // 宏参数：telemetry 实例 + 调用方字段列表，展开成 tracing::event!        tracing::event!(            target: $crate::targets::OTEL_LOG_ONLY_TARGET, // 钉死在 log_only 目标——这是与 trace_event! 的关键差异            tracing::Level::INFO, // 级别固定 INFO：遥测事件不参与日志级别过滤            $($fields)* // 展开调用方传入的自定义字段（如 event.name、prompt）            event.timestamp = %$crate::events::shared::timestamp(), // RFC3339 毫秒时间戳，下游解析格式统一            conversation.id = %$self.metadata.conversation_id, // 线程 ID：串联本会话所有事件的主键            app.version = %$self.metadata.app_version, // 客户端版本（编译期 CARGO_PKG_VERSION）            auth_mode = $self.metadata.auth_mode, // 认证模式已粗化为 ApiKey/Chatgpt——注意这里没有原始凭据            originator = %$self.metadata.originator, // 调用方标识（CLI/IDE/插件），构造时已做 sanitize            user.account_id = $self.metadata.account_id, // ⚠️ PII：账号 ID 只出现在日志通道            user.email = $self.metadata.account_email, // ⚠️ PII：邮箱也只出现在日志通道            terminal.type = %$self.metadata.terminal_type, // 终端类型（user-agent）            model = %$self.metadata.model, // 模型名            slug = %$self.metadata.slug, // 模型别名        );    }};}

pub fn user_prompt(&self, items: &[UserInput]) { // 用户输入上报入口，turn 开始时调用    let prompt = items // 把输入项里的文本拼成完整 prompt，用于长度统计        .iter() // 遍历所有输入项        .flat_map(|item| match item { // 逐项取出文本（其他类型跳过）            UserInput::Text { text, .. } => Some(text.as_str()), // 只收集文本片段            _ => None, // 图片/文件等其他类型不进 prompt 正文        })        .collect::<String>(); // 拼成完整字符串，用于长度统计    let text_input_count = items.iter().filter(|item| matches!(item, UserInput::Text { .. })).count(); // 文本输入条数——给 trace-safe 通道用    let image_input_count = items.iter().filter(|item| matches!(item, UserInput::Image { .. })).count(); // 远程图片条数    let local_image_input_count = items.iter().filter(|item| matches!(item, UserInput::LocalImage { .. })).count(); // 本地图片条数    let prompt_to_log = if self.metadata.log_user_prompts { // 配置开关 otel.log_user_prompt 决定是否记录正文（默认 false）        prompt.as_str() // 开关打开：记录完整正文    } else { // 开关关闭（默认）：只留占位符        "[REDACTED]" // 默认：只留占位符，不泄露用户内容    };    log_event!( // 日志通道：opt-in 时可带完整 prompt        self, // telemetry 实例，宏内部取 metadata        event.name = "codex.user_prompt", // 事件名两通道一致，靠 target 区分归属        prompt_length = %prompt.chars().count(), // 长度永远上报——即使正文被脱敏也能做分布统计        prompt = %prompt_to_log, // 正文（或 [REDACTED]）只进日志通道    );    trace_event!( // 追踪通道：只有计数，绝无内容        self, // telemetry 实例，宏内部取 metadata        event.name = "codex.user_prompt", // 事件名两通道一致，靠 target 区分归属        prompt_length = %prompt.chars().count(), // 长度两边都报——脱敏也不丢分布信息        text_input_count = text_input_count as i64, // 文本输入条数        image_input_count = image_input_count as i64, // 远程图片条数        local_image_input_count = local_image_input_count as i64, // 本地图片条数    );}

otel.* 内容类配置项（config/src/types.rs OtelConfig::default）

pub(crate) fn emit_tool_result( // 发射诊断日志与 trace-safe 元数据，不改动工具调用指标    telemetry: &SessionTelemetry, // 会话遥测上下文（metadata + 开关）    limits: ToolResultLogConfig, // 预览字节上限（默认 2048）    event: ToolResultEvent<'_>, // 工具结果事件载荷（借用字段，零拷贝）) {    let preview = telemetry_preview(output, limits); // 先把 output 截成预览——日志通道只看前 N 字节    let tool_result_seq = next_tool_result_seq(); // 进程级单调递增序号（AtomicU64），跨会话也能排顺序    log_and_trace_event!( // 一次宏调用发两个事件：common 字段两边都有，各自还有私有字段        telemetry, // 传入遥测上下文，宏内部拆两路        common: { // 两通道共有：安全元数据            event.name = "codex.tool_result", // 事件名：工具结果（两通道共有）            tool_result_seq = tool_result_seq, // 进程级序号，跨会话可排序            tool_name = %tool_name.name, // 工具名（exec / apply_patch / mcp__...）            call_id = %call_id, // 调用 ID：与 tool_call 事件配对            duration_ms = %duration.as_millis(), // 执行耗时（毫秒）            success = %success, // 是否成功——错误率指标的基础            output_truncated = preview.truncated, // 是否被遥测上限截断（不是工具结果本身是否有损）        },        log: { // 仅日志通道：可能含内容            arguments = %arguments, // 工具入参（仅日志通道，可能含路径等敏感信息）            output = %preview.text, // 截断后的预览文本            mcp_server = %mcp_server, // MCP server 名（内置工具为空串）        },        trace: { // 仅追踪通道：只有长度和计数，绝无内容            arguments_length = arguments.len() as i64, // 入参字节长度——只有大小没有内容            output_length = output.len() as i64, // 输出字节长度            output_line_count = output.lines().count() as i64, // 输出行数：粗粒度规模信号            tool_origin = if mcp_server.is_empty() { "builtin" } else { "mcp" }, // 区分内置工具与 MCP 工具        },    );}

pub(crate) fn telemetry_preview( // 把工具输出截成日志通道用的预览    content: &str, // 原始工具输出（可能很长）    limits: ToolResultLogConfig, // 截断上限配置) -> ToolResultPreview<'_> { // 返回预览文本 + 是否被截断    let prefix = take_bytes_at_char_boundary(content, limits.max_bytes); // 按 max_bytes 切，但回退到 UTF-8 字符边界——绝不把多字节字符劈成两半    if prefix.len() == content.len() { // 没超限 → 零拷贝借用原字符串，不分配内存        return ToolResultPreview { text: Cow::Borrowed(content), truncated: false }; // 直接借用原串，零分配返回    }    let separator = if prefix.is_empty() || prefix.ends_with('\n') { "" } else { "\n" }; // 若截在半行中间，先补换行，避免提示语粘在半个 token 后面    ToolResultPreview { // 超限分支：构造新字符串        text: Cow::Owned(format!("{prefix}{separator}[... telemetry preview truncated ...]")), // 追加截断标记：下游能区分"被遥测截断"和"输出本来就短"        truncated: true, // 标记"被遥测截断"，供下游查询    }}

const MILLISECOND_DURATION_UNIT: &str = "ms"; // 时长指标的单位声明（OTLP 语义约定）const MILLISECOND_DURATION_BOUNDARIES: &[f64] = &[ // 直方图桶边界：预定义，控制基数（cardinality）    0.0, 5.0, 10.0, 25.0, 75.0, 100.0, 250.0, 500.0, 750.0, 1_000.0, // 毫秒级：覆盖绝大多数 API 调用 / 工具执行    1_250.0, 1_500.0, 1_750.0, 2_000.0, 3_000.0, 4_000.0, 5_000.0, // 秒级：慢请求    6_000.0, 7_500.0, 9_000.0, 12_000.0, 15_000.0, 20_000.0, // 十几秒：长工具调用    30_000.0, 60_000.0, 120_000.0, // 上限 2 分钟——再长的就该看 turn 级指标了];// 整轮（turn）可能累积数分钟的推理 + 工具执行 + 用户等待static TURN_CRITICAL_PATH_DURATION_BOUNDARIES: LazyLock<Vec<f64>> = LazyLock::new(|| { // turn 关键路径需要更长的桶，在基础边界上扩展    let mut bounds = MILLISECOND_DURATION_BOUNDARIES.to_vec(); // 复制基础边界作为起点    bounds.extend([180_000.0, 300_000.0, 600_000.0, 900_000.0, // 3/5/10/15 分钟：长 turn        1_800_000.0, 3_600_000.0, 7_200_000.0]); // 最长到 2 小时——turn 可能因用户审批而长时间挂起    bounds // 返回扩展后的完整边界列表});

pub(crate) fn build_http_client( // 为 OTLP HTTP 导出器构建阻塞式客户端    tls: &OtelTlsConfig, // mTLS 配置（CA / 客户端证书）    timeout_var: &str, // 超时环境变量名) -> Result<reqwest::blocking::Client, Box<dyn Error>> { // 返回阻塞式 HTTP 客户端，失败带配置错误信息    if current_tokio_runtime_is_multi_thread() { // 情况1：在多线程 tokio runtime 里 → block_in_place 让出 worker 线程跑阻塞代码        tokio::task::block_in_place(|| build_http_client_inner(tls, timeout_var)) // 让出 worker 线程执行阻塞构建    } else if tokio::runtime::Handle::try_current().is_ok() { // 情况2：在 current-thread runtime 里（不能 block_in_place）→ 另起 OS 线程跑，join 取结果        let tls = tls.clone(); // 克隆配置移入新线程（Clone 廉价）        let timeout_var = timeout_var.to_string(); // 字符串化以便跨线程移动所有权        std::thread::spawn(move || build_http_client_inner(&tls, &timeout_var).map_err(|err| err.to_string())) // 新 OS 线程里构建，错误转 String 便于跨线程传递            .join() // 等子线程构建完成——客户端只建一次，这个等待发生在启动期            .map_err(|_| config_error("failed to join OTLP blocking HTTP client builder thread"))? // join 失败（线程 panic）→ 配置错误            .map_err(config_error) // 内层构建错误统一包装成配置错误    } else { // 情况3：完全不在 runtime 里 → 当前线程直接跑        build_http_client_inner(tls, timeout_var) // 无 runtime：当前线程直接构建    }}

pub(crate) fn resolve_exporter(exporter: &OtelExporter) -> OtelExporter { // 把逻辑导出器解析成具体端点配置    match exporter { // 按逻辑导出器分支处理        OtelExporter::Statsig => { // Statsig = OpenAI 内置摄取端点            if cfg!(debug_assertions) { // debug 构建（本地开发 / 测试运行）：强制关闭                return OtelExporter::None; // 本地跑不发 best-effort 遥测流量，避免污染生产数据            }            OtelExporter::OtlpHttp { // Statsig 本质就是一个内置的 OTLP HTTP 端点 + client API key                endpoint: STATSIG_OTLP_HTTP_ENDPOINT.to_string(), // 内置端点：https://ab.chatgpt.com/otlp/v1/metrics                headers: HashMap::from([(STATSIG_API_KEY_HEADER, STATSIG_API_KEY)]), // 客户端公开密钥（源码常量），用于 Statsig 摄取鉴权                protocol: OtelHttpProtocol::Json, // HTTP + JSON 传输                tls: None, // 走默认系统 CA，无需自定义 mTLS            }        }        _ => exporter.clone(), // 自定义 OTLP gRPC/HTTP 配置原样透传    }}

fn new(auth_manager: Arc<AuthManager>, destination: AnalyticsEventsDestination) -> Self { // 创建队列 + 单一消费任务    let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE); // 有界通道，容量 256——内存的第一道防线    tokio::spawn(async move { // 单 worker 任务：reducer 的独占所有者        let mut reducer = AnalyticsReducer::default(); // reducer 状态机独占在这个任务里，全程无需加锁        while let Some(input) = receiver.recv().await { // 逐条消费：Fact / ThreadProduct / Flush 三种消息            let input = match input { // 按消息类型分流处理                AnalyticsEventsQueueMessage::Fact(input) => *input, // 普通事实 → 交给下面的 reducer                AnalyticsEventsQueueMessage::ThreadProduct { .. } => continue, // 产品归属更新分支：登记线程 SKU，变化时先 flush 该线程挂起事件（此处省略）                AnalyticsEventsQueueMessage::Flush(done_tx) => { let mut e = Vec::new(); reducer.flush(&mut e); send_track_events(&auth_manager, &destination, e, &thread_products).await; let _ = done_tx.send(()); continue; } // Flush：排空所有未完成 turn 并上报，再通知调用方            };            let mut events = Vec::new(); // 收集本轮 reducer 产出的事件            reducer.ingest(input, &mut events).await; // fact → event：把零散协议消息聚合成完整 turn 级事件            send_track_events(&auth_manager, &destination, events, &thread_products).await; // 批量 POST 到 ChatGPT 后端 /codex/analytics-events/events        }    });}

fn try_send(&self, input: AnalyticsFact) { // 所有 track_*() 调用都从这里入队    if self.sender.try_send(AnalyticsEventsQueueMessage::Fact(Box::new(input))).is_err() { // 非阻塞发送：队列满立即失败，绝不等待        tracing::warn!("dropping analytics events: queue is full"); // 丢弃并告警——遥测永远不阻塞主路径等空位    }}

OTEL 通道（日志 / 追踪 / 指标）

▼

▼

Analytics 通道（产品事件）

▼

▼