安全采集、回放与线上到离线闭环
可观测性本身可能变成数据泄漏面。prompt、工具参数、错误字符串与文档原文含有秘密，先发送到平台再脱敏已经太晚。采集端允许列表、访问控制、保留期和删除流程需要共同设计。
对字段和值同时约束
完整标准库实验只允许受控错误码和整数时长，原始message被丢弃。
def sanitize(event):
if event.get("code") not in {"ok","timeout","denied"}:
raise ValueError("unknown code")
ms=event.get("duration_ms")
if type(ms) is not int or ms<0:raise ValueError("invalid duration")
return {"code":event["code"],"duration_ms":ms}
raw={"code":"timeout","duration_ms":7,"message":"secret=123","prompt":"private document"}
assert sanitize(raw)=={"code":"timeout","duration_ms":7}
try:sanitize({"code":"secret=123","duration_ms":1})
except ValueError:print("secret cannot hide in code")
else:raise AssertionError("unbounded field")字段名允许不等于字段值安全。哈希可能仍可关联个人，脱敏后也要限制查询与导出；访问审计记录谁因何目的看了哪类轨迹。删除范围包含平台、缓存、向量、导出与备份，法律保留等规则要由明确的治理策略决定。
回放必须控制副作用
诊断优先用脱敏输入、固定prompt/模型版本与记录的工具结果。工具替身能够重现编排和解析，不能证明真实服务当前状态；真实只读集成另行运行。写工具回放默认禁止，若确需测试在独立沙箱与幂等空间执行，避免重复发信/付款。
失败回流数据集时保存期望行为、来源证据、失败分类与敏感级别，人工确认后固定版本。线上随机样本、投诉样本和攻击样本比例不同，评估报告分切片，不能把投诉集当全体流量代表。模型judge需要与人工校准，不能用观测平台的一个分数代替业务验收。
事故响应先限制影响范围，再保存证据与回滚版本，最后修复根因并新增回归。停止工具、轮换凭证、恢复旧prompt和删除泄漏数据各自有责任人；观察到恢复正常不表示过去的副作用已清理。平台更换时保留稳定的业务事件与版本语义，供应商字段映射在适配层。
从实际执行采集到受限回放
静态span只能练习读图，无法证明真实执行被采集。下面task实际调用一个会先超时后成功的工具，产生父子span、错误与重试事件；emit先按白名单投影，再写JSONL。工具响应是本例允许记录的结构化业务字段，生产中仍需对每种result制定字段策略，不能默认保存任意工具原文。
回放使用记录结果代替真实工具，复现同一错误/重试控制流。它不再次访问上游，也不允许写工具。保存为replay.py运行。
import json, time, uuid
from tempfile import TemporaryDirectory
from pathlib import Path
writes=[]
def task(tool, emit):
trace=uuid.uuid4().hex
emit({'trace':trace,'span':'task','parent':None,'event':'start'})
value=None
for attempt in range(2):
span=f'tool-{attempt}'
emit({'trace':trace,'span':span,'parent':'task','event':'start'})
try:
value=tool('lookup',{'id':'p1'})
emit({'trace':trace,'span':span,'parent':'task','event':'result','result':value})
break
except TimeoutError:
emit({'trace':trace,'span':span,'parent':'task','event':'error','error':'timeout'})
emit({'trace':trace,'span':'task','parent':None,'event':'end','status':'ok' if value else 'failed'})
return value
calls=0
def real_tool(name,args):
global calls
calls+=1
if calls==1:raise TimeoutError('upstream detail must not enter log')
return {'source':'p1','days':2}
def scrub(event):
# 白名单在持久化前执行；这里从未记录用户原文、密钥或工具异常全文。
allowed={'trace','span','parent','event','result','error','status'}
return {k:v for k,v in event.items() if k in allowed}
with TemporaryDirectory() as folder:
path=Path(folder)/'trace.jsonl'
def emit(event):
with path.open('a') as f:f.write(json.dumps(scrub(event))+'\n')
expected=task(real_tool,emit)
recorded=[json.loads(line) for line in path.read_text().splitlines()]
results=iter(e for e in recorded if e['parent']=='task' and e['event'] in ('error','result'))
def replay_tool(name,args):
if name!='lookup':raise PermissionError('replay forbids write tools')
item=next(results)
if item['event']=='error':raise TimeoutError(item['error'])
return item['result']
replayed=[]
assert task(replay_tool,replayed.append)==expected
assert calls==2 and writes==[]
assert [e['event'] for e in recorded]==[e['event'] for e in replayed]
try:replay_tool('charge',{})
except PermissionError:pass
else:raise AssertionError('write accepted')
print('timeout retry replayed without calling live tool')回放trace ID是新的，但事件序列与结果相同。回放能证明这个固定输入下的控制分支，不能还原远程模型随机性、系统时钟或当时外部数据。要对不同请求通用回放，应额外记录脱敏后的工具名、参数规范哈希和顺序，并逐次匹配；不能把本题结果套到另一个文档ID。
只记录最终答案将丢失第一次timeout，误以为正常完成；只统计总耗时无法定位是模型还是工具。诊断时先按trace连接阶段，再定位失败事件，再选择是否回放。
练习
回放变体：把真实工具改为两次都超时，保存状态应为failed，回放也必须failed且不能产生写操作；解释为何这不证明修复后的上游已经恢复。
将sanitize扩展为允许可选retry_count，必须为0到3的整数；其他字段继续丢弃。检查True不能作为合法次数。