排查 Agent 失败时，怎样留下不泄漏正文的追踪

一次库存查询没有完成。调试人员首先想知道：失败发生在生成回复还是工具调用，哪个子步骤向上报告了错误，耗时有多长。把整段提示词、工具参数和异常栈贴进工单，会顺手带上邮箱、URL 查询串，甚至凭据。本文用一份完全合成的 trace，练习把这些材料裁到够用。

完成后你会拿到一个本地 JSON 报告：关联编号保留，正文消失，工具超时仍能定位到预先登记的 inventory.fetch。所有数据都是虚构的；NOT-A-REAL-TOKEN-UNUSABLE-DEMO 明确是不可用的示例字符串。代码不读环境变量、不调用模型、不连接 collector，也不导入 Agents SDK。

这不是通用匿名化工具。本文采用自己定义的 fixture-trace-v1 格式，只处理一个完整批次。实际 SDK 数据需要另写有版本的适配层。本实验不评价模型质量、不统计 tokens，也不证明任何真实服务已完成脱敏。

截至 2026-10-07 核对的 OpenAI Agents SDK tracing 文档说明追踪默认开启，模型与函数 span 可能记录输入输出。RunConfig(trace_include_sensitive_data=False) 控制这些敏感载荷；如果本次运行不需要追踪，可用 RunConfig(tracing_disabled=True)。全局关闭方式包括 OPENAI_AGENTS_DISABLE_TRACING=1；应在创建 trace 或运行 Agent 之前完成配置。

这两个布尔配置只在这里说明，没有作为实测代码运行。关闭正文捕获不自动清理应用自己的异常对象、日志或自定义 metadata；关闭追踪也不撤回已经缓存或发送的数据。官方追踪生命周期说明特别区分载荷控制和异常链。用固定工作流名称，不要把客户邮箱塞到名称、关联 ID 或自定义标签里。

官方文档还提醒：add_trace_processor() 会保留默认 exporter，且各 processor 是独立观察者。把一个可能抛错的脱敏 processor 排在前面，不能保证后面的 exporter 收不到原文。需要强制先脱敏后送达时，应由同一个应用 exporter 负责两步，并在创建追踪前替换默认 processors。可对照 官方本地脱敏示例；它也不是后端 ingest 协议。

本篇实现是独立编写的离线文件投影。官方仓库采用 MIT 许可，本文未复制它的 exporter 实现，也未验证 SDK 队列、flush、关闭回调或远端接收行为。若原文已经被采集到文件中，下面的导出无法消除这份原始文件的敏感性。

默认不复制原始对象。程序新建输出字典，只填经过规则处理的字段。prompt、output、tool_args、metadata 作为明确登记的原始载荷，整个子树丢弃；不会因为里面藏着一个叫 status 的键，就把它提升为报告状态。error 的 message、stack、cause、context 同样整段删除。

顶层、span 和 error 的字段集合是封闭的。出现尚未登记的 headers、details 或新的顶层属性，整批停止，等待人工决定是删除还是增加受控表示。被声明为整体删除的载荷子树允许任意 JSON 内容，但仍受到大小、深度、重复键和编码检查。这个区别让“新增可导出字段”不能悄悄搭便车。

operation、kind、error.code、error.site 采用固定枚举。site 是应用维护的错误位置标签，例如 inventory.fetch，不能直接取 traceback 文件名或异常消息。URL 只接受一个虚构 HTTPS 源和一个固定路径，输出 inventory.stock 标签；query、fragment、userinfo 不输出，带 userinfo 的输入直接拒绝。扩展规则时必须先审查新的枚举值本身是否泄露内部信息。

原始 trace_id 也可能被误填成邮箱。本例即使接受了带换行的原始 ID，也只在内存里当作键，用本批次的 t1、s1、s2 等替换。parent_id 通过同一张映射表重写，映射不落盘；同一个原始 span ID 在不同 trace 中按不同对象处理。

它是局部编号，不是加密或稳定匿名身份。调换输入顺序会改变编号，不适合跨批次追踪同一个人或请求。需要跨批关联时，要另行设计由可信执行器生成的无业务含义 ID，并审查可关联性风险；直接对邮箱做无密钥哈希仍可能被字典枚举。

每条 trace 必须有且只有一个 run 根节点，span ID 在各自 trace 内唯一，父节点必须存在，父链不能循环，子步骤的时间区间必须落在父区间内。这样的严格要求适合已封存批次；流式、截断或尚未结束的材料需要先由采集端补齐，不能默默当作完整报告。

需要 Python 3.10 或更新版本，只用标准库。把下面三个完整文件放到一个新目录。页面配套包还包括 fixture.json、safe-report.json 与三幅 SVG；safe-report.json 是示例结果，运行时请换一个不存在的输出文件名。

python3 make_fixture.py
python3 -m unittest -v
python3 redact.py fixture.json my-safe-report.json

本次在 Python 3.12.14 上实际运行了 40 项离线测试，全部通过。生成器输出 WROTE: synthetic fixture.json，导出成功输出 EXPORTED: diagnostic-v1，退出码为 0。再次使用同一路径会拒绝覆盖并退出 2。失败只打印固定 REVIEW_REQUIRED: export did not complete，不打印原文、路径、异常消息或堆栈。

策略与解析检查发生在打开输出文件之前，因此这两类拒绝不会产生半批报告。文件写入本身不是事务：磁盘故障可能留下未完成的文件。必须以成功退出和有效 JSON 为交付条件，不能仅凭文件存在判断成功；失败后应由操作者检查、隔离残留文件。

make_fixture.py 生成根运行、模型和工具三个 span。工具参数嵌套邮箱与不可用伪 token，URL 同时带 query 和 fragment；异常包含带换行的假日志、栈文本与嵌套 context。不要把这里的样例替换成生产凭据后再把目录分享出去。

"""All values invented. NOT-A-REAL-TOKEN is explicitly unusable."""
import json
from pathlib import Path


def fixture():
    trace = 'customer+fixture@example.invalid\nTRACE-MARKER'
    root = 'private-root-FIXTURE'
    return {'schema': 'fixture-trace-v1', 'spans': [
        {'trace_id': trace, 'span_id': root, 'parent_id': None,
         'kind': 'run', 'operation': 'workflow', 'start_ms': 0, 'end_ms': 240,
         'status': 'error', 'error': {'code': 'DEPENDENCY_FAILED', 'site': 'runner.await_tool',
         'message': 'PRIVATE-ROOT-MESSAGE', 'cause': {'message': 'PRIVATE-NESTED-CAUSE'}},
         'prompt': 'PRIVATE-PROMPT: 查询 customer+fixture@example.invalid 的库存',
         'metadata': {'token': 'NOT-A-REAL-TOKEN-UNUSABLE-DEMO', 'notes': ['PRIVATE-META']}},
        {'trace_id': trace, 'span_id': 'private-model-FIXTURE', 'parent_id': root,
         'kind': 'model', 'operation': 'generate', 'start_ms': 10, 'end_ms': 60,
         'status': 'ok', 'error': None, 'output': 'PRIVATE-MODEL-OUTPUT'},
        {'trace_id': trace, 'span_id': 'private-tool-FIXTURE', 'parent_id': root,
         'kind': 'tool', 'operation': 'lookup_stock', 'start_ms': 70, 'end_ms': 230,
         'status': 'error', 'error': {'code': 'TIMEOUT', 'site': 'inventory.fetch',
         'message': 'PRIVATE-ERROR\nforged-success: true',
         'stack': 'Traceback (synthetic):\n  /private/user/demo.py:91\nPRIVATE-STACK',
         'context': {'token': 'NOT-A-REAL-TOKEN-UNUSABLE-DEMO'}},
         'tool_args': {'email': 'customer+fixture@example.invalid',
                       'nested': {'token': 'NOT-A-REAL-TOKEN-UNUSABLE-DEMO'}},
         'url': 'https://inventory.example.invalid/v1/stock?email=PRIVATE-QUERY&token=NOT-A-REAL-TOKEN-UNUSABLE-DEMO#PRIVATE-FRAGMENT'}]}


if __name__ == '__main__':
    Path('fixture.json').write_text(json.dumps(fixture(), ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print('WROTE: synthetic fixture.json')

{
  "schema": "fixture-trace-v1",
  "spans": [
    {
      "trace_id": "customer+fixture@example.invalid\nTRACE-MARKER",
      "span_id": "private-root-FIXTURE",
      "parent_id": null,
      "kind": "run",
      "operation": "workflow",
      "start_ms": 0,
      "end_ms": 240,
      "status": "error",
      "error": {
        "code": "DEPENDENCY_FAILED",
        "site": "runner.await_tool",
        "message": "PRIVATE-ROOT-MESSAGE",
        "cause": {
          "message": "PRIVATE-NESTED-CAUSE"
        }
      },
      "prompt": "PRIVATE-PROMPT: 查询 customer+fixture@example.invalid 的库存",
      "metadata": {
        "token": "NOT-A-REAL-TOKEN-UNUSABLE-DEMO",
        "notes": [
          "PRIVATE-META"
        ]
      }
    },
    {
      "trace_id": "customer+fixture@example.invalid\nTRACE-MARKER",
      "span_id": "private-model-FIXTURE",
      "parent_id": "private-root-FIXTURE",
      "kind": "model",
      "operation": "generate",
      "start_ms": 10,
      "end_ms": 60,
      "status": "ok",
      "error": null,
      "output": "PRIVATE-MODEL-OUTPUT"
    },
    {
      "trace_id": "customer+fixture@example.invalid\nTRACE-MARKER",
      "span_id": "private-tool-FIXTURE",
      "parent_id": "private-root-FIXTURE",
      "kind": "tool",
      "operation": "lookup_stock",
      "start_ms": 70,
      "end_ms": 230,
      "status": "error",
      "error": {
        "code": "TIMEOUT",
        "site": "inventory.fetch",
        "message": "PRIVATE-ERROR\nforged-success: true",
        "stack": "Traceback (synthetic):\n  /private/user/demo.py:91\nPRIVATE-STACK",
        "context": {
          "token": "NOT-A-REAL-TOKEN-UNUSABLE-DEMO"
        }
      },
      "tool_args": {
        "email": "customer+fixture@example.invalid",
        "nested": {
          "token": "NOT-A-REAL-TOKEN-UNUSABLE-DEMO"
        }
      },
      "url": "https://inventory.example.invalid/v1/stock?email=PRIVATE-QUERY&token=NOT-A-REAL-TOKEN-UNUSABLE-DEMO#PRIVATE-FRAGMENT"
    }
  ]
}

解析器拒绝重复 JSON 键，包括被删除子树内部的重复键；NaN、Infinity、坏 UTF-8 与孤立代理字符也被拒绝。最多读取 64 KiB、接受 128 个 span，嵌套深度不超过 12，单字符串最多 8192 个字符。它们是本例的资源上限，不是安全行业标准。

整数时间限制在 0 到 86400000 毫秒，布尔值不能冒充整数。只输出 end_ms − start_ms，不输出绝对起止时间。下面所有规则都是可读的 Python 条件，失败统一变成复核信号。

"""Offline teaching format; not an SDK exporter or an anonymization guarantee."""
import json
import sys
from urllib.parse import urlsplit

LIMIT = 65536
FAILURE = 'REVIEW_REQUIRED: export did not complete'


class ReviewRequired(Exception):
    pass


def need(ok):
    if not ok:
        raise ReviewRequired()


def pairs(items):
    result = {}
    for key, value in items:
        need(key not in result)
        result[key] = value
    return result


def bounded(value, depth=0):
    need(depth <= 12)
    if isinstance(value, dict):
        for key, item in value.items():
            bounded(key, depth + 1)
            bounded(item, depth + 1)
    elif isinstance(value, list):
        for item in value:
            bounded(item, depth + 1)
    elif isinstance(value, str):
        need(len(value) <= 8192)
        # Reject lone surrogates before serializing or displaying anything.
        value.encode('utf-8')


def parse(raw):
    need(type(raw) is bytes and len(raw) <= LIMIT)
    def bad_constant(_):
        raise ReviewRequired()
    try:
        data = json.loads(raw.decode('utf-8'), object_pairs_hook=pairs,
                          parse_constant=bad_constant)
        bounded(data)
        return data
    except (ValueError, UnicodeError, RecursionError):
        raise ReviewRequired() from None


def keys(obj, required, optional=()):
    need(type(obj) is dict)
    need(set(required) <= set(obj) <= set(required) | set(optional))


def identity(value):
    need(type(value) is str and 0 < len(value) <= 256)
    return value


def integer(value):
    need(type(value) is int and 0 <= value <= 86400000)
    return value


def choice(value, allowed):
    need(type(value) is str and value in allowed)
    return value


def endpoint(value):
    need(type(value) is str and len(value) <= 8192)
    # urlsplit silently strips some controls; reject them first.
    need(not any(ord(c) < 33 or ord(c) == 127 for c in value))
    try:
        u = urlsplit(value)
        need(u.scheme == 'https' and u.netloc == 'inventory.example.invalid')
        need(u.path == '/v1/stock')
    except ValueError:
        raise ReviewRequired() from None
    # Never return origin, path, userinfo, query, fragment, or their hashes.
    return 'inventory.stock'


def project(raw):
    data = parse(raw)
    keys(data, ['schema', 'spans'])
    need(data['schema'] == 'fixture-trace-v1')
    rows = data['spans']
    need(type(rows) is list and 0 < len(rows) <= 128)
    required = ['trace_id', 'span_id', 'parent_id', 'kind', 'operation',
                'start_ms', 'end_ms', 'status', 'error']
    dropped = ['prompt', 'output', 'tool_args', 'metadata', 'url']
    traces, spans, by_id = {}, {}, {}
    for row in rows:
        keys(row, required, dropped)
        tid, sid = identity(row['trace_id']), identity(row['span_id'])
        pair = (tid, sid)
        need(pair not in by_id)
        traces.setdefault(tid, 't' + str(len(traces) + 1))
        spans[pair] = 's' + str(len(spans) + 1)
        by_id[pair] = row
        if row['parent_id'] is not None:
            identity(row['parent_id'])
        choice(row['kind'], ('run', 'model', 'tool'))
        operation = choice(row['operation'], ('workflow', 'generate', 'lookup_stock'))
        need(row['kind'] == {'workflow': 'run', 'generate': 'model',
                             'lookup_stock': 'tool'}[operation])
        start, end = integer(row['start_ms']), integer(row['end_ms'])
        need(start <= end)
        choice(row['status'], ('ok', 'error'))
        error = row['error']
        if row['status'] == 'ok':
            need(error is None)
        else:
            keys(error, ['code', 'site'], ['message', 'stack', 'cause', 'context'])
            choice(error['code'], ('TIMEOUT', 'DEPENDENCY_FAILED', 'PARSE_FAILED'))
            choice(error['site'], ('runner.await_tool', 'inventory.fetch', 'model.decode'))
        if 'url' in row:
            need(operation == 'lookup_stock')
            endpoint(row['url'])
    for tid in traces:
        roots = [r for r in rows if r['trace_id'] == tid and r['parent_id'] is None]
        need(len(roots) == 1 and roots[0]['kind'] == 'run')
    # Validate the whole graph before building a shareable object.
    for pair, row in by_id.items():
        seen, current = set(), pair
        while True:
            need(current not in seen)
            seen.add(current)
            child = by_id[current]
            pid = child['parent_id']
            if pid is None:
                break
            parent = (current[0], pid)
            need(parent in by_id)
            p = by_id[parent]
            need(p['start_ms'] <= child['start_ms'] <= child['end_ms'] <= p['end_ms'])
            current = parent
    out = []
    for row in rows:
        tid = row['trace_id']
        error = row['error']
        out.append(dict(trace=traces[tid], span=spans[(tid, row['span_id'])],
                        parent=None if row['parent_id'] is None else spans[(tid, row['parent_id'])],
                        kind=row['kind'], operation=row['operation'],
                        duration_ms=row['end_ms'] - row['start_ms'], status=row['status'],
                        error_code=None if error is None else error['code'],
                        error_site=None if error is None else error['site'],
                        endpoint=endpoint(row['url']) if 'url' in row else None))
    return dict(policy='diagnostic-v1', spans=out)


def cli(argv):
    # No argparse input echo, payload repr, exception message, or traceback.
    try:
        need(len(argv) == 2)
        with open(argv[0], 'rb') as src:
            raw = src.read(LIMIT + 1)
        report = project(raw)
        encoded = (json.dumps(report, ensure_ascii=True, indent=2) + '\n').encode('ascii')
        # New path only. Old reports are never mistaken for this run's success.
        with open(argv[1], 'xb') as dst:
            dst.write(encoded)
    except Exception:
        pass
    else:
        print('EXPORTED: diagnostic-v1')
        return 0
    # Outside exception context: no original exception chaining in this log.
    print(FAILURE, file=sys.stderr)
    return 2


if __name__ == '__main__':
    sys.exit(cli(sys.argv[1:]))

独立手写的预期是：t1 的根 s1 耗时 240 ms，状态 DEPENDENCY_FAILED；s2 是成功的 generate，耗时 50 ms；s3 的 lookup_stock 耗时 160 ms，TIMEOUT，位置 inventory.fetch。两个子 span 的 parent 都是 s1。下方是本次离线运行的完整报告。

{
  "policy": "diagnostic-v1",
  "spans": [
    {
      "trace": "t1",
      "span": "s1",
      "parent": null,
      "kind": "run",
      "operation": "workflow",
      "duration_ms": 240,
      "status": "error",
      "error_code": "DEPENDENCY_FAILED",
      "error_site": "runner.await_tool",
      "endpoint": null
    },
    {
      "trace": "t1",
      "span": "s2",
      "parent": "s1",
      "kind": "model",
      "operation": "generate",
      "duration_ms": 50,
      "status": "ok",
      "error_code": null,
      "error_site": null,
      "endpoint": null
    },
    {
      "trace": "t1",
      "span": "s3",
      "parent": "s1",
      "kind": "tool",
      "operation": "lookup_stock",
      "duration_ms": 160,
      "status": "error",
      "error_code": "TIMEOUT",
      "error_site": "inventory.fetch",
      "endpoint": "inventory.stock"
    }
  ]
}

你能据此决定下一步检查超时阈值、库存依赖或本地等待路径，却不能知道用户问了哪一件商品。缺少正文是刻意保留的限制。240、50、160 来自合成输入，不能当作真实服务性能；父子时长和并发子步骤也不能简单相加。需要重现内容相关错误时，在受控环境另建最小合成复现，不要把原始 prompt 补回共享报告。

test_redact.py 的 EXPECTED 是手写对象，未调用 project 来生成。精确比较同时检查字段没有多出、父链没有断、时长没有算错。canary 断言检查这一批已知敏感标记未出现，不能推导未知秘密也一定被发现。真正的主要边界是“不复制未允许的值”。

40 项测试包括嵌套重复键、转义与未转义换行、未知字段、循环父链、跨 trace 引用、重复 span、子时间窗越界、非法数字、URL 相似域与 userinfo、超深载荷，以及 CLI 的失败无原文、末条出错不留前半批、拒绝覆盖现有报告。异常 stack 与 cause/context 在正常夹具中就已经带标记，不只测试顶层 prompt。

import copy
import io
import json
from pathlib import Path
import tempfile
import unittest
from contextlib import redirect_stdout, redirect_stderr
from make_fixture import fixture
from redact import project, ReviewRequired, cli, FAILURE, LIMIT


def wire(obj):
    return json.dumps(obj, ensure_ascii=True).encode()


# Hand-authored expectation, not generated by the projection under test.
EXPECTED = {'policy': 'diagnostic-v1', 'spans': [
    {'trace': 't1', 'span': 's1', 'parent': None, 'kind': 'run', 'operation': 'workflow',
     'duration_ms': 240, 'status': 'error', 'error_code': 'DEPENDENCY_FAILED',
     'error_site': 'runner.await_tool', 'endpoint': None},
    {'trace': 't1', 'span': 's2', 'parent': 's1', 'kind': 'model', 'operation': 'generate',
     'duration_ms': 50, 'status': 'ok', 'error_code': None, 'error_site': None, 'endpoint': None},
    {'trace': 't1', 'span': 's3', 'parent': 's1', 'kind': 'tool', 'operation': 'lookup_stock',
     'duration_ms': 160, 'status': 'error', 'error_code': 'TIMEOUT',
     'error_site': 'inventory.fetch', 'endpoint': 'inventory.stock'}]}


class RedactionTests(unittest.TestCase):
    def reject(self, change):
        data = fixture()
        change(data)
        with self.assertRaises(ReviewRequired):
            project(wire(data))

    def test_exact_independent_output(self):
        self.assertEqual(project(wire(fixture())), EXPECTED)

    def test_canaries_absent(self):
        result = json.dumps(project(wire(fixture())))
        for marker in ['PRIVATE-', 'NOT-A-REAL-TOKEN', '@', 'TRACE-MARKER',
                       'private-', 'forged-success', 'example.invalid']:
            self.assertNotIn(marker, result)

    def test_dropped_nested_payload(self):
        d = fixture(); d['spans'][0]['prompt'] = {'a': [{'password': 'PRIVATE-NEW-KIND'}]}
        self.assertEqual(project(wire(d)), EXPECTED)

    def test_unknown_top_key(self):
        self.reject(lambda d: d.update(secret='PRIVATE-UNKNOWN'))

    def test_unknown_span_key(self):
        self.reject(lambda d: d['spans'][2].update(headers={'Authorization': 'NOT-A-REAL-TOKEN-UNUSABLE-DEMO'}))

    def test_unknown_error_key(self):
        self.reject(lambda d: d['spans'][2]['error'].update(details='PRIVATE-UNKNOWN'))

    def test_nested_duplicate_key(self):
        raw = wire(fixture()).replace(b'"PRIVATE-NESTED-CAUSE"', b'"x", "message": "PRIVATE-DUP"')
        with self.assertRaises(ReviewRequired): project(raw)

    def test_duplicate_root_key(self):
        with self.assertRaises(ReviewRequired): project(b'{"schema":1,"schema":2}')

    def test_literal_newline_invalid_json(self):
        with self.assertRaises(ReviewRequired): project(b'{"schema":"x\nPRIVATE"}')

    def test_escaped_newline_in_allowed_name_rejected(self):
        self.reject(lambda d: d['spans'][1].update(operation='generate\nPRIVATE'))

    def test_duplicate_span_id(self):
        self.reject(lambda d: d['spans'].append(copy.deepcopy(d['spans'][1])))

    def test_missing_parent(self):
        self.reject(lambda d: d['spans'][1].update(parent_id='missing'))

    def test_self_cycle(self):
        self.reject(lambda d: d['spans'][1].update(parent_id=d['spans'][1]['span_id']))

    def test_two_node_cycle(self):
        def change(d):
            a,b=d['spans'][1:]; a.update(parent_id=b['span_id'],start_ms=70,end_ms=230); b['parent_id']=a['span_id']
        self.reject(change)

    def test_second_root(self):
        self.reject(lambda d: d['spans'][1].update(parent_id=None))

    def test_cross_trace_parent(self):
        self.reject(lambda d: d['spans'][1].update(trace_id='another-trace'))

    def test_two_traces_preserve_local_links(self):
        d=fixture(); other=copy.deepcopy(d['spans'])
        for r in other: r['trace_id']='another-trace'
        d['spans'] += other
        out=project(wire(d))['spans']
        self.assertEqual((out[3]['trace'],out[3]['span'],out[4]['parent']),('t2','s4','s4'))

    def test_bool_time(self):
        self.reject(lambda d: d['spans'][0].update(start_ms=True))

    def test_negative_time(self):
        self.reject(lambda d: d['spans'][0].update(start_ms=-1))

    def test_reversed_time(self):
        self.reject(lambda d: d['spans'][1].update(start_ms=61))

    def test_outside_parent(self):
        self.reject(lambda d: d['spans'][1].update(end_ms=241))

    def test_nan(self):
        with self.assertRaises(ReviewRequired): project(b'{"schema":NaN}')

    def test_infinity(self):
        with self.assertRaises(ReviewRequired): project(b'{"schema":Infinity}')

    def test_unknown_code(self):
        self.reject(lambda d: d['spans'][2]['error'].update(code='PRIVATE-ERROR-CODE'))

    def test_unknown_site(self):
        self.reject(lambda d: d['spans'][2]['error'].update(site='/private/person.py:13'))

    def test_kind_operation_mismatch(self):
        self.reject(lambda d: d['spans'][2].update(kind='model'))

    def test_error_ok_mismatch(self):
        self.reject(lambda d: d['spans'][2].update(status='ok'))

    def test_url_controls(self):
        self.reject(lambda d: d['spans'][2].update(url='\nhttps://inventory.example.invalid/v1/stock'))

    def test_url_userinfo(self):
        self.reject(lambda d: d['spans'][2].update(url='https://person@inventory.example.invalid/v1/stock'))

    def test_url_unknown_path(self):
        self.reject(lambda d: d['spans'][2].update(url='https://inventory.example.invalid/private-customer'))

    def test_url_lookalike_host(self):
        self.reject(lambda d: d['spans'][2].update(url='https://inventory.example.invalid.evil.invalid/v1/stock'))

    def test_missing_field(self):
        self.reject(lambda d: d['spans'][0].pop('status'))

    def test_invalid_utf8(self):
        with self.assertRaises(ReviewRequired): project(b'\xff')

    def test_lone_surrogate(self):
        self.reject(lambda d: d['spans'][0].update(prompt='\ud800'))

    def test_oversize(self):
        with self.assertRaises(ReviewRequired): project(b' ' * (LIMIT+1))

    def test_deep_payload(self):
        d=fixture(); x='PRIVATE-DEEP'
        for _ in range(15): x=[x]
        d['spans'][0]['metadata']=x
        with self.assertRaises(ReviewRequired): project(wire(d))

    def test_empty_batch(self):
        self.reject(lambda d: d.update(spans=[]))

    def test_cli_no_payload_on_failure(self):
        with tempfile.TemporaryDirectory() as tmp:
            src, dst=Path(tmp)/'input', Path(tmp)/'new'
            src.write_bytes(b'{"PRIVATE-PARSER-FAIL":')
            out, err=io.StringIO(),io.StringIO()
            with redirect_stdout(out),redirect_stderr(err): status=cli([str(src),str(dst)])
            self.assertEqual((status,out.getvalue(),err.getvalue()),(2,'',FAILURE+'\n'))
            self.assertFalse(dst.exists())

    def test_cli_unknown_field_no_partial_report(self):
        with tempfile.TemporaryDirectory() as tmp:
            d=fixture(); d['spans'][-1]['late-secret']='PRIVATE-LATE'
            src,dst=Path(tmp)/'input',Path(tmp)/'new';src.write_bytes(wire(d))
            with redirect_stdout(io.StringIO()),redirect_stderr(io.StringIO()): status=cli([str(src),str(dst)])
            self.assertEqual(status,2); self.assertFalse(dst.exists())

    def test_cli_success_and_no_overwrite(self):
        with tempfile.TemporaryDirectory() as tmp:
            src,dst=Path(tmp)/'input',Path(tmp)/'new';src.write_bytes(wire(fixture()))
            with redirect_stdout(io.StringIO()),redirect_stderr(io.StringIO()):
                self.assertEqual(cli([str(src),str(dst)]),0)
                self.assertEqual(json.loads(dst.read_text()),EXPECTED)
                self.assertEqual(cli([str(src),str(dst)]),2)
            self.assertEqual(json.loads(dst.read_text()),EXPECTED)


if __name__ == '__main__':
    unittest.main()

遇到 REVIEW_REQUIRED，先不要把报错的整个 trace 附到工单。固定信号没有指出哪个原始字段，是为了避免字段名或异常内容成为另一个泄漏出口。在有权访问原文件的受控环境中复现，查明是格式升级、批次不完整、规则缺项还是输入损坏。外发材料只保留复核状态与应用生成的工单编号。

新增一个字段时，写清排障问题、保留的最小表示、允许值来源与测试。若只需知道错误类别，增加受控错误码；若字段没有明确用途，登记为整段删除。不要为了解除阻塞而加入“所有字符串通过正则替换后保留”的兜底。编码变化、分片、拼接、自然语言转述和未知凭据形状都可能逃过这种做法。

更新策略版本、手写预期和恶意输入测试后重新生成报告。共享前审查最终导出文件，而不是只看测试绿灯；关联结构、受控工具标签、错误位置和耗时仍可能泄露业务信息。本例不提供差分隐私、访问控制、加密存储或防恶意 covert channel 的保证，也无法证明输入采集完整、可信。

先给自己的 SDK 版本写适配测试，明确哪些 span 类型、异常链和自定义字段会进入 fixture-trace-v1。源数据已有默认 exporter、第三方 telemetry、日志 handler 或异常上报时，每条出口都需要检查；离线文件投影无法阻止上游先发出原文。

测试真实导出入口的序列化失败、批处理异常、flush、关闭过程与目的地不可用，确认任何失败都没有原始载荷回退路径。本文没有安装或运行这些组件，也没有向任何 collector 发送测试数据。原始 trace 的存储权限、保留期限、删除流程和共享对象，需要按实际组织政策单独处理。

最终交付给协作者的通常只应是 my-safe-report.json 和必要的复现步骤。本文配套包为了教学保留了合成原始样例；真实工作中不要把原始 trace、ID 对照表、环境文件或调试缓存一起打包。