交给专门 Agent 前，先把上下文裁到刚好够用

客服协调 Agent 已经读过一整段会话，现在想请设备诊断 Agent 判断下一步该问什么。工单里有型号、错误码、重启结果，也混着旧收货地址、家属备注和工具返回的内部字段。把完整历史原封不动传过去，接收方当然能看懂问题，但同时也拿到了这次诊断用不上的资料。

我会先把目标收窄到一句具体的话：根据现有设备观察，给出下一条诊断问题。这个任务不需要联系客户、办理退款，也不需要读取家庭资料。接收方需要知道是哪张工单、已经确认了哪些观察、观察来自哪里，以及还缺什么。后面构造的包就围绕这些内容展开。

本篇使用 CASE-1042 这一张原创合成工单。所有私有信息都用测试标记表示，没有真人资料。运行部分只调用 Python 标准库，专门 Agent 由确定性函数替代，不调用真实模型、外部 Agent、通知或网络。示例验证的是转交内容的选择与验收。任务队列、重启恢复和幂等提交可以继续参看 多 Agent 消息协作与 Agent Bus 任务契约，这里不另造一套状态机。

本节核对 Agents Python v0.23.1 的 handoffs 文档，查阅日期为 2026 年 10 月 7 日。默认 handoff 会让接收 Agent 继续看到先前会话。input_type 描述转交工具的参数，它不会自动替换接收方的主输入。需要修改历史时，input_filter 接收并返回 HandoffInputData；单个 handoff 上的过滤器优先于运行配置中的过滤器。

HandoffInputData 区分启动前历史 input_history、此前生成的 pre_handoff_items，以及本轮 new_items；可选 input_items 可替换转交给模型的本轮项目。新版本的 nest_handoff_history 是默认关闭的 beta 功能，它会调整历史的组织形式。历史中有敏感内容时，换成摘要或嵌套形式仍可能保留那些内容。

对照 v0.23.1 过滤器实现，remove_all_tools 按工具项目类型筛选。已复制到普通消息里的工具参数和结果，还需要额外处理。下面的 drop_tool_rows 和 flatten_as_summary 是独立的教学反例，没有复刻或执行 SDK 算法。

官方还限定了服务端管理历史的情形：conversation_id、previous_response_id 或 auto_previous_response_id 不支持 handoff 输入过滤。如果隔离要求必须排除旧历史，应另开使用明确输入的运行，且不能复用原会话或响应标识。本文不提供未经测试的 SDK 接线代码，也不把 input_filter 当作完整的安全边界。

原始记录保存在 source_record()，同时附有便于查看的 source-record.json。MSG-02 包含型号和地址；TOOL-03 包含错误码与内部标记；MSG-04 同时有重启观察和从工具复制过来的标记。MSG-01 是无关旧会话，MSG-05 是旧摘要。这个安排是为了让同一份敏感内容出现在多个载体中，避免只测试工具返回值。

CATALOG 是另行维护的合成审核目录，选定三条允许转交的观察：PX-2、E17、已经重启。每条观察保留 evidence_ref、source_id、字段和值。投影时先核对原记录中的结构化观察确实与目录一致，遇到新型号或新结论就停止，等待目录重新审核。source_id、role、type 都只是本地标签，不认证原始来源。校验只核对独立审核目录中的结构化观察，不证明原文支持这些观察，也不证明抽取保留了完整语义。即使把原文改成相互矛盾的说法，只要结构化观察不变，当前校验仍会通过；原文与观察的一致性需要另行复核。

这个目录刻意做得很窄，便于看清边界。实际产品可以按任务类别维护字段与证据授权，并由可信业务流程生成每次任务的审核快照。不能让请求方连同候选包一起上传一个“可信目录”，再据此自证内容合规。

project() 从空对象开始构造结果，只写入七个顶层字段：schema_version、task_id、target、objective、facts、evidence、unresolved。原始 text、private、private_profile 和旧摘要没有复制路径。facts 中的每个值都必须与审核目录严格匹配；例如布尔值 true 不能由整数 1 冒充。新增一个原文属性不会悄悄扩大输出范围。

这个示例把指示灯状态留在 unresolved 中。尚未确认的事项不应该在压缩时消失，也不能被补成一个看似合理的状态。接收函数只返回 needs_information 和一条询问指示灯状态的问题，保留任务号及证据引用，不给出设备原因或保修结论。

证据引用也要控制范围。EV-01 只对应审核过的型号观察，接收方通过 evidence() 能读到这条投影及源记录编号，拿不到整段聊天的访问地址。若真实系统的引用可以直接打开完整工单，包本身再短也会留下越界读取的通道。示例没有联网取证，生产接入需要另行校验证据访问权限。

parse_packet() 在读取 JSON 时拒绝重复键，限制输入大小，然后交给 validate()。接收检查包括顶层与嵌套对象的准确字段集合、固定任务与目标、事实和证据数量、引用唯一性、来源和值的对应关系，以及指示灯状态这一未决项仍然存在。未知字段不会被默默忽略。

每条证据附有 sha256，用于检查这条小型观察是否改变。哈希本身不证明来源可信，所以测试专门构造了“改值后重新计算哈希”的候选，接收端仍然按独立目录拒绝。实际身份认证、授权与传输通道不在本地实验范围内。

packet.schema.json 将本工单允许的事实与证据完整枚举出来，便于阅读接口契约。它是固定夹具的封闭模式，不是覆盖所有客服场景的通用模式。运行时的拒绝码与业务一致性检查由 Python 实现；无需额外安装 JSON Schema 库。

先试一个看起来省事的办法：把 type 为 tool_result 的行删除，留下普通消息。测试仍找到三种标记，因为工具标记早已被 MSG-04 引用，地址和家属备注也在普通消息与摘要中。只检查结构化工具项，会漏掉这些文本副本。

另一条路径是把记录正文合并成一条摘要消息。对象数量减少了，里面的内容仍然是原文。本例故意使用最直白的拼接来展示问题，不评价某个模型的摘要质量。即使用生成式摘要，也需要独立验收内容范围，不能仅凭字数变短认定资料已经最小化。

leaks() 只寻找三种已知合成标记，是回归测试探针。测试也包含一个未见过的敏感字符串，探针无法识别它；严格的字段值校验仍会拒绝把它塞进本工单事实。这两种检查的证据不同。零标记检出只能说明这些测试标记没出现，无法推出真实场景中的个人信息已经全部脱敏。

把后面的投影实现保存为 context_packet.py，演示入口保存为 demo.py，测试保存为 test_context_packet.py，三份文件放在同一目录。另两份完整 JSON 可分别保存为 source-record.json 与 packet.schema.json，供人工对照；运行程序直接使用源码中的合成记录和 Python 校验，不读取这两份 JSON。使用 Python 3.10 或更新版本。先跑测试，再看演示。没有依赖安装、凭据配置或外部发送步骤。37 项测试在 Python 3.12 中通过；SDK 接线、真实模型输入和真实跨 Agent 传输均未测试。

python3 -m unittest -v
python3 demo.py

演示前四行应分别得到 3、3、3、0。随后打印完整最小包和接收结果。这里没有估算 token 数，也不声称能据此节省多少调用费用。字段最小化是否足够，还要看下游任务能否保持必要的证据与未知项。

full_history: fixture markers = 3
without_tool_rows: fixture markers = 3
summary_of_all_rows: fixture markers = 3
minimal_packet: fixture markers = 0
packet:
{
  "schema_version": 1,
  "task_id": "CASE-1042",
  "target": "device-triage",
  "objective": "choose_next_diagnostic_question",
  "facts": [
    {
      "field": "device_model",
      "value": "PX-2",
      "evidence_ref": "EV-01"
    },
    {
      "field": "error_code",
      "value": "E17",
      "evidence_ref": "EV-02"
    },
    {
      "field": "restart_attempted",
      "value": true,
      "evidence_ref": "EV-03"
    }
  ],
  "evidence": [
    {
      "ref": "EV-01",
      "source_id": "MSG-02",
      "field": "device_model",
      "value": "PX-2",
      "sha256": "5bfeb538fa14360c7b9ebd0043cecca923a2b3c11090582a125337f2fc38e3d3"
    },
    {
      "ref": "EV-02",
      "source_id": "TOOL-03",
      "field": "error_code",
      "value": "E17",
      "sha256": "d6812b9590a9ee706f5cc1ceb8b07489d8d4192682cbb8f862c3134d80b0db65"
    },
    {
      "ref": "EV-03",
      "source_id": "MSG-04",
      "field": "restart_attempted",
      "value": true,
      "sha256": "493c65e95d60b903c4bef5faf1b5e38d97ea24feacaa4c2f7e21b83f7e466c80"
    }
  ],
  "unresolved": [
    "indicator_state"
  ]
}
receiver:
{
  "task_id": "CASE-1042",
  "status": "needs_information",
  "question": "重新启动后，指示灯是常亮、闪烁还是熄灭？",
  "evidence_refs": [
    "EV-01",
    "EV-02",
    "EV-03"
  ]
}

测试覆盖普通消息中的工具副本、摘要副本、私有字段变化、额外字段、未知与重复引用、来源替换、内容指纹变化、重新计算指纹后的伪造、布尔与整数混淆、重复 JSON 键、缺失事实、缺失来源、任务与目标错配，以及未决问题被删掉。错误码用于定位本地契约失败；它们没有包含原文。

这份 JSON 与 source_record() 返回值相同，保留完整的混合输入。测试标记不是任何真实人的地址、健康资料或凭据。

{
  "task_id": "CASE-1042",
  "records": [
    {
      "id": "MSG-01",
      "type": "message",
      "role": "user",
      "text": "旧会话：家属就诊备注 PRIVATE_HEALTH_4R，与设备报修无关。"
    },
    {
      "id": "MSG-02",
      "type": "message",
      "role": "user",
      "text": "PX-2 打不开。上次收货地址为 PRIVATE_ADDRESS_8Q。",
      "facts": {
        "device_model": "PX-2"
      },
      "private": {
        "address": "PRIVATE_ADDRESS_8Q"
      }
    },
    {
      "id": "TOOL-03",
      "type": "tool_result",
      "text": "模拟诊断：错误码 E17；内部凭据标记 PRIVATE_TOKEN_2Z。",
      "facts": {
        "error_code": "E17"
      },
      "private": {
        "token": "PRIVATE_TOKEN_2Z"
      }
    },
    {
      "id": "MSG-04",
      "type": "message",
      "role": "assistant",
      "text": "已经重新启动一次，仍报 E17。工具备注 PRIVATE_TOKEN_2Z。",
      "facts": {
        "restart_attempted": true
      }
    },
    {
      "id": "MSG-05",
      "type": "message",
      "role": "assistant",
      "text": "旧摘要：PX-2，收货地址 PRIVATE_ADDRESS_8Q；家属备注 PRIVATE_HEALTH_4R。"
    }
  ],
  "private_profile": {
    "contact": "PRIVATE_ADDRESS_8Q"
  },
  "unresolved": [
    "indicator_state"
  ]
}

数组内允许的值与证据都在此列明。修改任务范围时，审核目录、模式和验收测试需要一起更新。

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "CASE-1042 diagnostic packet (fixture-specific)",
  "type": "object",
  "additionalProperties": false,
  "required": [
    "schema_version",
    "task_id",
    "target",
    "objective",
    "facts",
    "evidence",
    "unresolved"
  ],
  "properties": {
    "schema_version": {
      "type": "integer",
      "const": 1
    },
    "task_id": {
      "const": "CASE-1042"
    },
    "target": {
      "const": "device-triage"
    },
    "objective": {
      "const": "choose_next_diagnostic_question"
    },
    "facts": {
      "type": "array",
      "minItems": 3,
      "maxItems": 3,
      "uniqueItems": true,
      "items": {
        "enum": [
          {
            "field": "device_model",
            "value": "PX-2",
            "evidence_ref": "EV-01"
          },
          {
            "field": "error_code",
            "value": "E17",
            "evidence_ref": "EV-02"
          },
          {
            "field": "restart_attempted",
            "value": true,
            "evidence_ref": "EV-03"
          }
        ]
      }
    },
    "evidence": {
      "type": "array",
      "minItems": 3,
      "maxItems": 3,
      "uniqueItems": true,
      "items": {
        "enum": [
          {
            "ref": "EV-01",
            "source_id": "MSG-02",
            "field": "device_model",
            "value": "PX-2",
            "sha256": "5bfeb538fa14360c7b9ebd0043cecca923a2b3c11090582a125337f2fc38e3d3"
          },
          {
            "ref": "EV-02",
            "source_id": "TOOL-03",
            "field": "error_code",
            "value": "E17",
            "sha256": "d6812b9590a9ee706f5cc1ceb8b07489d8d4192682cbb8f862c3134d80b0db65"
          },
          {
            "ref": "EV-03",
            "source_id": "MSG-04",
            "field": "restart_attempted",
            "value": true,
            "sha256": "493c65e95d60b903c4bef5faf1b5e38d97ea24feacaa4c2f7e21b83f7e466c80"
          }
        ]
      }
    },
    "unresolved": {
      "const": [
        "indicator_state"
      ]
    }
  }
}

将下面完整实现保存为 context_packet.py，与后面的 demo.py 和 test_context_packet.py 放在同一目录。后两个文件通过 from context_packet import ... 导入它，文件名不要改成章节标题。

"""Original offline projection lab. No Agents SDK, model, network or credentials."""
import copy
import hashlib
import json

TASK = 'CASE-1042'
TARGET = 'device-triage'
OBJECTIVE = 'choose_next_diagnostic_question'
SENTINELS = ('PRIVATE_ADDRESS_8Q', 'PRIVATE_HEALTH_4R', 'PRIVATE_TOKEN_2Z')
# Fixture policy maintained independently from the candidate packet.
# These synthetic observations are manually selected for this diagnostic task.
CATALOG = {
    'EV-01': {'source_id': 'MSG-02', 'field': 'device_model', 'value': 'PX-2'},
    'EV-02': {'source_id': 'TOOL-03', 'field': 'error_code', 'value': 'E17'},
    'EV-03': {'source_id': 'MSG-04', 'field': 'restart_attempted', 'value': True},
}
QUESTIONS = {'indicator_state': '重新启动后，指示灯是常亮、闪烁还是熄灭？'}


def source_record():
    """Entire original synthetic record; none of these people or values are real."""
    return {
        'task_id': TASK,
        'records': [
            {'id': 'MSG-01', 'type': 'message', 'role': 'user',
             'text': '旧会话：家属就诊备注 PRIVATE_HEALTH_4R，与设备报修无关。'},
            {'id': 'MSG-02', 'type': 'message', 'role': 'user',
             'text': 'PX-2 打不开。上次收货地址为 PRIVATE_ADDRESS_8Q。',
             'facts': {'device_model': 'PX-2'},
             'private': {'address': 'PRIVATE_ADDRESS_8Q'}},
            {'id': 'TOOL-03', 'type': 'tool_result',
             'text': '模拟诊断：错误码 E17；内部凭据标记 PRIVATE_TOKEN_2Z。',
             'facts': {'error_code': 'E17'},
             'private': {'token': 'PRIVATE_TOKEN_2Z'}},
            {'id': 'MSG-04', 'type': 'message', 'role': 'assistant',
             'text': '已经重新启动一次，仍报 E17。工具备注 PRIVATE_TOKEN_2Z。',
             'facts': {'restart_attempted': True}},
            {'id': 'MSG-05', 'type': 'message', 'role': 'assistant',
             'text': '旧摘要：PX-2，收货地址 PRIVATE_ADDRESS_8Q；家属备注 PRIVATE_HEALTH_4R。'},
        ],
        'private_profile': {'contact': 'PRIVATE_ADDRESS_8Q'},
        'unresolved': ['indicator_state'],
    }


class PacketError(ValueError):
    pass


def require(ok, code):
    if not ok:
        raise PacketError(code)


def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'))


def digest(value):
    return hashlib.sha256(canonical(value).encode('utf-8')).hexdigest()


def same(a, b):
    # Python's True == 1 is not acceptable for typed fact equality.
    return type(a) is type(b) and a == b


def evidence(ref):
    require(type(ref) is str and ref in CATALOG, 'unknown_evidence')
    body = dict(CATALOG[ref])
    return {'ref': ref, **body, 'sha256': digest(body)}


def project(source):
    require(type(source) is dict and source.get('task_id') == TASK, 'wrong_task')
    records = source.get('records')
    require(type(records) is list, 'bad_records')
    indexed = {}
    for row in records:
        require(type(row) is dict and type(row.get('id')) is str, 'bad_record')
        require(row['id'] not in indexed, 'duplicate_source')
        indexed[row['id']] = row
    facts, references = [], []
    for ref, approved in CATALOG.items():
        row = indexed.get(approved['source_id'])
        require(row is not None, 'missing_source')
        observed = row.get('facts')
        require(type(observed) is dict, 'missing_fact')
        require(approved['field'] in observed, 'missing_fact')
        require(same(observed[approved['field']], approved['value']), 'unreviewed_fact')
        facts.append({'field': approved['field'], 'value': approved['value'], 'evidence_ref': ref})
        references.append(evidence(ref))
    require(source.get('unresolved') == ['indicator_state'], 'unreviewed_question')
    packet = {
        'schema_version': 1, 'task_id': TASK, 'target': TARGET,
        'objective': OBJECTIVE, 'facts': facts,
        'evidence': references, 'unresolved': ['indicator_state'],
    }
    validate(packet)
    return packet


def exact_keys(obj, keys, code):
    require(type(obj) is dict and set(obj) == set(keys), code)


def validate(packet):
    exact_keys(packet, ('schema_version', 'task_id', 'target', 'objective',
                        'facts', 'evidence', 'unresolved'), 'bad_packet_keys')
    require(type(packet['schema_version']) is int and packet['schema_version'] == 1, 'bad_version')
    require(packet['task_id'] == TASK, 'wrong_task')
    require(packet['target'] == TARGET, 'wrong_target')
    require(packet['objective'] == OBJECTIVE, 'wrong_objective')
    require(type(packet['facts']) is list and len(packet['facts']) == len(CATALOG), 'bad_fact_count')
    require(type(packet['evidence']) is list and len(packet['evidence']) == len(CATALOG), 'bad_evidence_count')
    seen = set()
    for row in packet['evidence']:
        exact_keys(row, ('ref', 'source_id', 'field', 'value', 'sha256'), 'bad_evidence_keys')
        ref = row['ref']
        require(type(ref) is str and ref in CATALOG, 'unknown_evidence')
        require(ref not in seen, 'duplicate_evidence')
        seen.add(ref)
        expected = evidence(ref)
        require(all(same(row[k], expected[k]) for k in expected), 'evidence_mismatch')
    seen = set()
    for fact in packet['facts']:
        exact_keys(fact, ('field', 'value', 'evidence_ref'), 'bad_fact_keys')
        ref = fact['evidence_ref']
        require(type(ref) is str and ref in CATALOG, 'unknown_evidence')
        expected = CATALOG[ref]
        require(ref not in seen, 'duplicate_fact')
        seen.add(ref)
        require(same(fact['field'], expected['field']) and same(fact['value'], expected['value']), 'fact_mismatch')
    require(type(packet['unresolved']) is list and packet['unresolved'] == ['indicator_state'], 'bad_unresolved')
    require(not leaks(packet), 'fixture_leak')
    return True


def parse_packet(text):
    def unique(pairs):
        obj = {}
        for key, value in pairs:
            require(key not in obj, 'duplicate_json_key')
            obj[key] = value
        return obj
    require(type(text) is str and len(text.encode('utf-8')) <= 8192, 'bad_wire_size')
    try:
        obj = json.loads(text, object_pairs_hook=unique)
    except json.JSONDecodeError as exc:
        raise PacketError('bad_json') from exc
    validate(obj)
    return obj


def receive(text):
    packet = parse_packet(text)
    # Deterministic specialist stub. It cannot fetch a transcript or act externally.
    return {'task_id': packet['task_id'], 'status': 'needs_information',
            'question': QUESTIONS[packet['unresolved'][0]],
            'evidence_refs': [x['ref'] for x in packet['evidence']]}


def leaks(value):
    rendered = canonical(value)
    return [marker for marker in SENTINELS if marker in rendered]


def drop_tool_rows(source):
    """Counterexample only, NOT an implementation of SDK remove_all_tools."""
    return [copy.deepcopy(r) for r in source['records'] if r['type'] == 'message']


def flatten_as_summary(rows):
    """Counterexample only, NOT the SDK nested-history algorithm."""
    return {'type': 'message', 'role': 'assistant',
            'text': '\n'.join(r['text'] for r in rows)}

demo.py 只打印四种本地输入的标记检出数及接收结果。

import json
from context_packet import (source_record, project, canonical, receive, leaks,
                            drop_tool_rows, flatten_as_summary)

source = source_record()
packet = project(source)
paths = {
    'full_history': source,
    'without_tool_rows': drop_tool_rows(source),
    'summary_of_all_rows': flatten_as_summary(source['records']),
    'minimal_packet': packet,
}
for name, value in paths.items():
    print(f'{name}: fixture markers = {len(leaks(value))}')
print('packet:')
print(json.dumps(packet, ensure_ascii=False, indent=2))
print('receiver:')
print(json.dumps(receive(canonical(packet)), ensure_ascii=False, indent=2))

test_context_packet.py 可直接运行，所有故障都在合成数据的副本上构造。

import copy
import json
import unittest
from context_packet import (PacketError, source_record, project, validate, receive,
                            canonical, parse_packet, leaks, drop_tool_rows,
                            flatten_as_summary, digest, evidence)

class ContextTests(unittest.TestCase):
    def setUp(self):
        self.source = source_record()
        self.packet = project(self.source)

    def rejected(self, code, fn, value):
        with self.assertRaisesRegex(PacketError, '^' + code + '$'):
            fn(value)

    def test_round_trip(self):
        self.assertEqual(parse_packet(canonical(self.packet)), self.packet)

    def test_no_fixture_markers(self):
        self.assertEqual(leaks(self.packet), [])

    def test_raw_history_exposes_three(self):
        self.assertEqual(len(leaks(self.source)), 3)

    def test_removing_tools_retains_message_copies(self):
        rows = drop_tool_rows(self.source)
        self.assertTrue(all(r['type'] == 'message' for r in rows))
        self.assertEqual(len(leaks(rows)), 3)

    def test_summary_retains_tool_marker(self):
        summary = flatten_as_summary(self.source['records'])
        self.assertIn('PRIVATE_TOKEN_2Z', summary['text'])

    def test_no_raw_text_keys(self):
        self.assertNotIn('text', canonical(self.packet))
        self.assertNotIn('private', canonical(self.packet))

    def test_no_original_aliases(self):
        before = copy.deepcopy(self.source)
        packet = project(self.source)
        packet['facts'][0]['value'] = 'changed'
        self.assertEqual(self.source, before)

    def test_unrelated_history_does_not_change_projection(self):
        self.source['records'].append({'id':'MSG-99','type':'message','text':'PRIVATE_HEALTH_4R'})
        self.assertEqual(project(self.source), self.packet)

    def test_private_fields_do_not_change_projection(self):
        self.source['records'][2]['private']['token'] = 'unrecognized-secret'
        self.assertEqual(project(self.source), self.packet)

    def test_free_text_summary_is_rejected(self):
        self.packet['summary'] = 'PRIVATE_HEALTH_4R'
        self.rejected('bad_packet_keys', validate, self.packet)

    def test_nested_extra_field_is_rejected(self):
        self.packet['facts'][0]['private'] = 'x'
        self.rejected('bad_fact_keys', validate, self.packet)

    def test_evidence_extra_field_is_rejected(self):
        self.packet['evidence'][0]['url'] = 'https://example.invalid/all-history'
        self.rejected('bad_evidence_keys', validate, self.packet)

    def test_unknown_evidence(self):
        self.packet['evidence'][0]['ref'] = 'EV-99'
        self.rejected('unknown_evidence', validate, self.packet)

    def test_duplicate_evidence(self):
        self.packet['evidence'][1] = copy.deepcopy(self.packet['evidence'][0])
        self.rejected('duplicate_evidence', validate, self.packet)

    def test_duplicate_fact(self):
        self.packet['facts'][1] = copy.deepcopy(self.packet['facts'][0])
        self.rejected('duplicate_fact', validate, self.packet)

    def test_wrong_value_rejected_even_with_recomputed_hash(self):
        row = self.packet['evidence'][0]
        row['value'] = 'PRIVATE_ADDRESS_8Q'
        row['sha256'] = digest({k:row[k] for k in ('source_id','field','value')})
        self.rejected('evidence_mismatch', validate, self.packet)

    def test_source_reference_substitution(self):
        self.packet['evidence'][0]['source_id'] = 'MSG-01'
        self.rejected('evidence_mismatch', validate, self.packet)

    def test_hash_change(self):
        self.packet['evidence'][0]['sha256'] = '0' * 64
        self.rejected('evidence_mismatch', validate, self.packet)

    def test_fact_claim_disagrees_with_evidence(self):
        self.packet['facts'][0]['value'] = 'PX-9'
        self.rejected('fact_mismatch', validate, self.packet)

    def test_integer_cannot_replace_boolean(self):
        self.packet['facts'][2]['value'] = 1
        self.rejected('fact_mismatch', validate, self.packet)

    def test_boolean_cannot_replace_version(self):
        self.packet['schema_version'] = True
        self.rejected('bad_version', validate, self.packet)

    def test_missing_fact(self):
        self.packet['facts'].pop()
        self.rejected('bad_fact_count', validate, self.packet)

    def test_missing_evidence(self):
        self.packet['evidence'].pop()
        self.rejected('bad_evidence_count', validate, self.packet)

    def test_wrong_task(self):
        self.packet['task_id'] = 'CASE-9999'
        self.rejected('wrong_task', validate, self.packet)

    def test_wrong_target(self):
        self.packet['target'] = 'refund-executor'
        self.rejected('wrong_target', validate, self.packet)

    def test_wrong_objective(self):
        self.packet['objective'] = 'send_refund'
        self.rejected('wrong_objective', validate, self.packet)

    def test_missing_unresolved_question(self):
        self.packet['unresolved'] = []
        self.rejected('bad_unresolved', validate, self.packet)

    def test_question_cannot_be_free_text(self):
        self.packet['unresolved'] = ['PRIVATE_HEALTH_4R']
        self.rejected('bad_unresolved', validate, self.packet)

    def test_duplicate_source_id(self):
        self.source['records'].append(copy.deepcopy(self.source['records'][1]))
        self.rejected('duplicate_source', project, self.source)

    def test_missing_source(self):
        self.source['records'] = [r for r in self.source['records'] if r['id'] != 'MSG-02']
        self.rejected('missing_source', project, self.source)

    def test_source_fact_requires_new_review(self):
        self.source['records'][1]['facts']['device_model'] = 'PRIVATE_ADDRESS_8Q'
        self.rejected('unreviewed_fact', project, self.source)

    def test_duplicate_json_keys(self):
        wire = canonical(self.packet).replace('"schema_version":1', '"schema_version":1,"schema_version":1')
        self.rejected('duplicate_json_key', parse_packet, wire)

    def test_invalid_json(self):
        self.rejected('bad_json', parse_packet, '{')

    def test_oversize(self):
        self.rejected('bad_wire_size', parse_packet, 'x' * 8193)

    def test_receiver_keeps_uncertainty(self):
        result = receive(canonical(self.packet))
        self.assertEqual(result['status'], 'needs_information')
        self.assertEqual(result['task_id'], 'CASE-1042')
        self.assertEqual(result['evidence_refs'], ['EV-01', 'EV-02', 'EV-03'])
        self.assertIn('指示灯', result['question'])

    def test_evidence_lookup_returns_only_approved_projection(self):
        self.assertEqual(set(evidence('EV-01')), {'ref','source_id','field','value','sha256'})
        self.rejected('unknown_evidence', evidence, 'MSG-01')

    def test_sentinel_check_is_not_general_privacy_classifier(self):
        self.assertEqual(leaks({'text':'an unseen sensitive fact'}), [])
        self.packet['facts'][0]['value'] = 'an unseen sensitive fact'
        self.rejected('fact_mismatch', validate, self.packet)

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

本文的包只覆盖下一位 Agent 收到的任务上下文。实际接线时，还要检查动态 instructions、共享应用 context、会话恢复、工具可见范围，以及之后通过证据引用读取的数据；否则历史虽已裁剪，敏感内容仍可能从别处进入模型输入。过滤器只改变某条输入路径，不能代替应用中的权限控制。

如果自定义过滤器还要调用 SDK 的嵌套历史助手，应按该版本文档检查它实际读取哪些历史字段，不能只清空 input_items 就假定旧内容已经消失。生产验证要观察最终提交给模型的输入，而不仅是某个中间对象；观察和保存这个输入本身也需要妥善控制数据范围。

追踪导出是另一项工作。追踪资料最小化关注排障日志留下什么，本篇关注接收 Agent 获得什么。这里的 37 项测试不证明 tracing、session 存储或模型服务端都没有保存原内容。将这些通道分别列入接线验收，才能知道一次转交真正经过了哪些地方。

可以先用这张固定工单跑通字段投影，再换成自己业务中的一类窄任务。每增加一个自由文本字段、可解引用地址或新的接收方，都重新问一遍：下游具体需要它做什么，来源如何核对，范围扩大后由谁批准。

前面的固定目录实验有一个明确边界：它核对结构化 facts，不检查原文与 facts 的语义一致性。这个边界在真实交接中很容易出现：客服把“已经重启”更正成“尚未重启”，而抽取结果还没更新。接收方拿到很短、字段齐全、指纹一致的包，仍可能继续问“重启后指示灯怎样”。字段少，并不等于事实仍然有效。

这里沿用 CASE-1042，但单独做第二个离线实验。把前面 source_record() 的 MSG-04.text 改成“更正：尚未重新启动设备。”，其余字段不动，再运行 project()，结果仍是 restart_attempted=true、unresolved=["indicator_state"]。在 Python 3.12.14 中重新运行原来的 37 项测试全部通过，这条反例也确实通过投影。它复现的是原文已经披露的局限，不表示旧测试验证过这个场景。

新实验不让模型自行判定更正是否可信。发送侧保存预先编写的合成审核快照，把被审核的原文、原文版本和允许输出的事实绑在一起；任一项变化就暂停投影。重新核对后，第二份合成快照明确把 restart_attempted 改成 false，并把未决问题改成 restart_confirmation。接收方因此问“请先确认是否已经重新启动设备。”，不会继续沿用重启后的问题。

R1、R2 是应用维护的合成审核快照，不是候选包提供的 approved=true，也不是 SDK 内建协议。R1 对应原文版本 7，“已重启”；R2 对应原文版本 8，“尚未重启”。两份快照在代码中静态写好，代替真实的复核过程。实验没有证明任何真人审批已经发生，也没有实现审批界面、身份认证、数据库或跨进程隔离。

project() 比较来源编号、版本、原文 UTF-8 的 SHA-256 和精确类型的 facts。原文及其摘要指纹只留在发送侧审核目录，不进入最小包。这里选择保守的精确字节规则：哪怕只多一个空格，也要求重新复核。它能发现审核后原文发生变化，不能理解句子；若最初的审核快照本来就错，代码不会纠正它。

receive() 的 current 参数必须由接收应用从自己可信的当前快照取得，不能从候选包取出 review_revision 再回填。旧包与当前 R2 不符时返回 stale_review；只把标签改成 R2、只改布尔值、保留旧证据版本或旧未决问题，都不能通过完整包比对。本例采用封闭、固定工单契约，因此可以把候选包与应用当前快照构造的完整包作精确比较。多工单、多字段业务需要独立维护相应审核记录，不能直接把固定目录当通用抽取器。

接收端没有实时查询原文，也没有原子地锁住业务资料。如果源资料已改、接收应用仍传入旧的 current=R1，旧包依然会被接受，测试明确保留这个限制。生产系统必须定义当前快照何时作废、谁可以发布新快照，以及从接收检查到实际使用之间如何处理再次更正。本实验只确定一次函数调用所给快照下的结果，不声称解决并发更新或撤销已经发送的回答。

为了让一份离线脚本可复制，发送和接收函数共用同一进程中的 REVIEWS 常量；这不是实际隔离部署。只有 packet_for() 的白名单结果跨越本例的序列化边界，接收判断只用快照投影，不需要原文摘要。在实际部署中，接收侧应只持有允许转交的当前快照投影，不能把整个私有审核目录作为运行上下文一起交给模型。源正文到事实的判断仍由复核过程负责，代码不会从“尚未”自动推断 false。

继续使用前文 context_packet.py、demo.py 与 test_context_packet.py 所在的目录，将本节保存为 revision_packet.py，后两节分别保存为 test_revision_packet.py、demo_revision_packet.py 和 run_checks.py。如果另建空目录，先把前文三份 Python 文件复制进去；demo_revision_packet.py 会导入 context_packet.py，下面的完整检查也会运行 test_context_packet.py。代码使用 Python 标准库，无需安装 Agents SDK，也不会读取真实工单、调用模型或发送信息。它与前面的 context_packet.py 是两个独立示例，不要覆盖旧文件。未知顶层字段、重复 JSON 键、非有限数和超过 4096 UTF-8 字节的输入会被拒绝；接收错误不会附带原文。

2026-10-20 修正了一个接收错误出口：把合法包中的 schema_version 或嵌套事实替换成 1e400，原实现会在重新序列化时抛出原生 ValueError，绕过只捕获 Rejected 的调用方。parse_constant 只接收 NaN、Infinity、-Infinity 这类常量；1e400 走 parse_float，可能转换成无穷大。现在 finite_float 在解析时检查 math.isfinite，并统一抛出 Rejected("bad_json")。正常有限浮点数仍保留其类型，2e0 不会被当作整数版本 2 接受。这是错误分类修正，旧实现没有把这些包当作有效请求接受。

"""Offline application contract, not an Agents SDK implementation."""
import copy
import hashlib
import json
import math

class Rejected(ValueError):
    pass

def need(condition, code):
    if not condition:
        raise Rejected(code)

def wire(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True,
                      separators=(',', ':'), allow_nan=False)

def text_hash(text):
    return hashlib.sha256(text.encode('utf-8')).hexdigest()

# Application-owned reviewed snapshots. Never accept these from the packet.
# Text is synthetic and stays on the sender/review side.
TEXT1 = '设备 PX-2；已重启，错误码 E17。'
TEXT2 = '更正：设备 PX-2；尚未重启，错误码 E17。'

def snapshot(revision, source_revision, text, restarted):
    return {'revision': revision, 'task_id': 'CASE-1042',
            'target': 'device-triage', 'source_id': 'MSG-04',
            'source_revision': source_revision, 'text_hash': text_hash(text),
            'facts': {'device_model': 'PX-2', 'error_code': 'E17',
                      'restart_attempted': restarted},
            'unresolved': ['indicator_state' if restarted else 'restart_confirmation']}

REVIEWS = {'R1': snapshot('R1', 7, TEXT1, True),
           'R2': snapshot('R2', 8, TEXT2, False)}

def source(version=1):
    review = REVIEWS['R1' if version == 1 else 'R2']
    return {'task_id': 'CASE-1042', 'records': [{
        'id': 'MSG-04', 'revision': review['source_revision'],
        'text': TEXT1 if version == 1 else TEXT2,
        'facts': copy.deepcopy(review['facts']),
        'internal_note': '本地备注，不转交'}]}

def packet_for(review):
    return {'schema_version': 2, 'task_id': review['task_id'],
            'target': review['target'], 'review_revision': review['revision'],
            'facts': copy.deepcopy(review['facts']),
            'evidence': {'source_id': review['source_id'],
                         'source_revision': review['source_revision']},
            'unresolved': list(review['unresolved'])}

def project(raw, current):
    need(type(current) is str and current in REVIEWS, 'unknown_review')
    review = REVIEWS[current]
    need(type(raw) is dict and raw.get('task_id') == review['task_id'], 'wrong_task')
    rows = raw.get('records')
    need(type(rows) is list, 'bad_records')
    indexed = {}
    for row in rows:
        need(type(row) is dict and type(row.get('id')) is str, 'bad_source')
        need(row['id'] not in indexed, 'duplicate_source')
        indexed[row['id']] = row
    row = indexed.get(review['source_id'])
    need(row is not None, 'missing_source')
    need(type(row.get('revision')) is int and
         row['revision'] == review['source_revision'], 'source_revision_changed')
    need(type(row.get('text')) is str and
         text_hash(row['text']) == review['text_hash'], 'source_text_changed')
    facts = row.get('facts')
    need(type(facts) is dict and set(facts) == set(review['facts']), 'unreviewed_facts')
    need(all(type(facts[k]) is type(v) and facts[k] == v
             for k, v in review['facts'].items()), 'unreviewed_facts')
    return packet_for(review)

def receive(text, current):
    need(type(current) is str and current in REVIEWS, 'unknown_review')
    need(type(text) is str, 'bad_wire')
    try:
        size = len(text.encode('utf-8'))
    except UnicodeEncodeError as exc:
        raise Rejected('bad_wire') from exc
    need(size <= 4096, 'bad_wire')
    def unique(pairs):
        value = {}
        for key, item in pairs:
            need(key not in value, 'duplicate_key')
            value[key] = item
        return value
    def constant(_):
        raise Rejected('bad_json')
    def finite_float(token):
        # parse_constant does not see numeric forms such as 1e400.
        value = float(token)
        need(math.isfinite(value), 'bad_json')
        return value
    try:
        candidate = json.loads(text, object_pairs_hook=unique,
                               parse_constant=constant, parse_float=finite_float)
    except (json.JSONDecodeError, RecursionError) as exc:
        raise Rejected('bad_json') from exc
    expected = packet_for(REVIEWS[current])
    need(type(candidate) is dict and set(candidate) == set(expected), 'bad_packet_fields')
    need(candidate['review_revision'] == current, 'stale_review')
    # Canonical JSON distinguishes true/1 and 8/8.0, unlike Python equality.
    need(wire(candidate) == wire(expected), 'packet_mismatch')
    question = ('重新启动后，指示灯是常亮、闪烁还是熄灭？'
                if candidate['facts']['restart_attempted']
                else '请先确认是否已经重新启动设备。')
    return {'review_revision': current, 'status': 'needs_information', 'question': question}

原有 27 项测试保留，新增 2 项测试方法覆盖 16 个正负溢出位置与 3 个有限浮点类型反例，共 29 项。测试输入直接改写 JSON 数字字面量，不先用 wire() 序列化无穷大，以免尚未调用 receive() 就被编码器拦截。

P1、P2、A1、A2 是单独写出的固定预期对象，测试没有调用 packet_for() 或复制 REVIEWS 来生成期望值。它们同时约束最终字段与下游问题，防止实现把事实改了、问题却仍来自旧版本。恢复需要用更新后的原文和新审核快照重新投影，改标签不能代替这一步。下面保存为 test_revision_packet.py。

import copy
import unittest
from revision_packet import Rejected, source, project, receive, wire, TEXT2

# Fixed expected objects are authored independently of packet_for()/REVIEWS.
P1 = {'schema_version': 2, 'task_id': 'CASE-1042', 'target': 'device-triage',
      'review_revision': 'R1', 'facts': {'device_model': 'PX-2',
      'error_code': 'E17', 'restart_attempted': True},
      'evidence': {'source_id': 'MSG-04', 'source_revision': 7},
      'unresolved': ['indicator_state']}
P2 = {'schema_version': 2, 'task_id': 'CASE-1042', 'target': 'device-triage',
      'review_revision': 'R2', 'facts': {'device_model': 'PX-2',
      'error_code': 'E17', 'restart_attempted': False},
      'evidence': {'source_id': 'MSG-04', 'source_revision': 8},
      'unresolved': ['restart_confirmation']}
A1 = {'review_revision': 'R1', 'status': 'needs_information',
      'question': '重新启动后，指示灯是常亮、闪烁还是熄灭？'}
A2 = {'review_revision': 'R2', 'status': 'needs_information',
      'question': '请先确认是否已经重新启动设备。'}

class Cases(unittest.TestCase):
    def reject(self, code, fn, *args):
        with self.assertRaisesRegex(Rejected, '^' + code + '$'):
            fn(*args)
    def test_fixed_old_packet(self):
        self.assertEqual(project(source(1), 'R1'), P1)
        self.assertEqual(receive(wire(P1), 'R1'), A1)
    def test_fixed_recovery_packet(self):
        self.assertEqual(project(source(2), 'R2'), P2)
        self.assertEqual(receive(wire(P2), 'R2'), A2)
    def test_text_only_correction_requires_review(self):
        raw = source(); raw['records'][0]['text'] = TEXT2
        self.reject('source_text_changed', project, raw, 'R1')
    def test_even_whitespace_requires_review(self):
        raw = source(); raw['records'][0]['text'] += ' '
        self.reject('source_text_changed', project, raw, 'R1')
    def test_revision_only_change(self):
        raw = source(); raw['records'][0]['revision'] = 8
        self.reject('source_revision_changed', project, raw, 'R1')
    def test_fact_only_change(self):
        raw = source(); raw['records'][0]['facts']['restart_attempted'] = False
        self.reject('unreviewed_facts', project, raw, 'R1')
    def test_boolean_source_revision(self):
        raw = source(); raw['records'][0]['revision'] = True
        self.reject('source_revision_changed', project, raw, 'R1')
    def test_integer_source_fact(self):
        raw = source(); raw['records'][0]['facts']['restart_attempted'] = 1
        self.reject('unreviewed_facts', project, raw, 'R1')
    def test_old_source_new_review(self):
        self.reject('source_revision_changed', project, source(), 'R2')
    def test_old_packet_after_current_changed(self):
        self.reject('stale_review', receive, wire(P1), 'R2')
    def test_relabel_only_cannot_repair(self):
        p = copy.deepcopy(P1); p['review_revision'] = 'R2'
        self.reject('packet_mismatch', receive, wire(p), 'R2')
    def test_change_fact_but_keep_old_question(self):
        p = copy.deepcopy(P2); p['unresolved'] = ['indicator_state']
        self.reject('packet_mismatch', receive, wire(p), 'R2')
    def test_change_fact_but_keep_old_evidence(self):
        p = copy.deepcopy(P2); p['evidence']['source_revision'] = 7
        self.reject('packet_mismatch', receive, wire(p), 'R2')
    def test_integer_cannot_replace_true(self):
        p = copy.deepcopy(P1); p['facts']['restart_attempted'] = 1
        self.reject('packet_mismatch', receive, wire(p), 'R1')
    def test_float_cannot_replace_revision(self):
        p = copy.deepcopy(P2); p['evidence']['source_revision'] = 8.0
        self.reject('packet_mismatch', receive, wire(p), 'R2')
    def test_candidate_review_cannot_authorize_itself(self):
        p = copy.deepcopy(P2); p['approved'] = True
        self.reject('bad_packet_fields', receive, wire(p), 'R2')
    def test_private_text_cannot_ride_packet(self):
        p = copy.deepcopy(P2); p['evidence']['text'] = TEXT2
        self.reject('packet_mismatch', receive, wire(p), 'R2')
    def test_unrelated_data_does_not_expand_packet(self):
        raw = source(); raw['records'][0]['internal_note'] = '另一条本地备注'
        raw['records'].append({'id': 'MSG-99', 'text': '无关信息'})
        self.assertEqual(project(raw, 'R1'), P1)
    def test_duplicate_source(self):
        raw = source(); raw['records'].append(copy.deepcopy(raw['records'][0]))
        self.reject('duplicate_source', project, raw, 'R1')
    def test_missing_source(self):
        raw = source(); raw['records'] = []
        self.reject('missing_source', project, raw, 'R1')
    def test_no_alias_mutation(self):
        raw = source(); before = copy.deepcopy(raw)
        p = project(raw, 'R1'); p['facts']['error_code'] = 'changed'
        self.assertEqual(raw, before)
        self.assertEqual(project(source(), 'R1'), P1)
    def test_duplicate_wire_key(self):
        self.reject('duplicate_key', receive,
                    wire(P1).replace('"schema_version":2', '"schema_version":2,"schema_version":2'), 'R1')
    def test_nonfinite_wire(self):
        self.reject('bad_json', receive, '{"value":NaN}', 'R1')
    def test_exponent_overflow_at_any_depth(self):
        base = wire(P1)
        for token in ('1e400', '-1e400', '9e9999', '-9e9999'):
            candidates = (
                base.replace('"schema_version":2', '"schema_version":' + token),
                base.replace('"restart_attempted":true', '"restart_attempted":' + token),
                base.replace('"source_revision":7', '"source_revision":' + token),
                base.replace('"unresolved":["indicator_state"]', '"unresolved":[' + token + ']'),
            )
            for location, candidate in enumerate(candidates):
                with self.subTest(token=token, location=location):
                    self.reject('bad_json', receive, candidate, 'R1')
    def test_finite_float_still_has_wrong_type(self):
        base = wire(P1)
        for candidate in (
            base.replace('"schema_version":2', '"schema_version":2e0'),
            base.replace('"restart_attempted":true', '"restart_attempted":1e0'),
            base.replace('"source_revision":7', '"source_revision":7e0'),
        ):
            with self.subTest(candidate=candidate):
                self.reject('packet_mismatch', receive, candidate, 'R1')
    def test_oversized_wire(self):
        self.reject('bad_wire', receive, 'x' * 4097, 'R1')
    def test_unknown_current(self):
        self.reject('unknown_review', receive, wire(P1), 'R99')
    def test_wrong_task(self):
        p = copy.deepcopy(P1); p['task_id'] = 'CASE-9999'
        self.reject('packet_mismatch', receive, wire(p), 'R1')
    def test_stale_trusted_current_is_not_detectable(self):
        # Deliberate limit: if application passes old current, old packet still passes.
        self.assertEqual(receive(wire(P1), 'R1'), A1)

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

下面的 demo_revision_packet.py 先调用前文原封不动的 context_packet.py，输出完整更正夹具、旧包和错误的下一问题；再输出新实验的完整输入、拒绝结果和 R2 恢复结果。所有源内容都是合成夹具。为了让读者检查反例，演示会在本地终端打印合成源记录；这份调试输出不是转交包，真实系统不要这样记录私有正文。

"""Fixed input/output replay, including the unchanged original projection."""
import context_packet as baseline
import revision_packet as lab

def show(label, value):
    print(label + ': ' + lab.wire(value))

def outcome(fn, *args):
    try:
        return fn(*args)
    except lab.Rejected as exc:
        return {'error': str(exc)}

raw = baseline.source_record()
next(r for r in raw['records'] if r['id'] == 'MSG-04')['text'] = '更正：尚未重新启动设备。'
show('baseline corrected input', raw)
old_baseline = baseline.project(raw)
show('baseline unchanged packet', old_baseline)
show('baseline wrong next question', baseline.receive(baseline.canonical(old_baseline)))
old = lab.project(lab.source(), 'R1')
corrected = lab.source()
corrected['records'][0]['text'] = lab.TEXT2
show('review R1 source', lab.source())
show('review R1 packet', old)
show('review R1 answer', lab.receive(lab.wire(old), 'R1'))
show('text-only correction input', corrected)
show('text-only correction rejected', outcome(lab.project, corrected, 'R1'))
show('old packet with current R2', outcome(lab.receive, lab.wire(old), 'R2'))
show('review R2 source', lab.source(2))
new = lab.project(lab.source(2), 'R2')
show('recovered packet', new)
show('recovered answer', lab.receive(lab.wire(new), 'R2'))
# Explicit runtime checks survive python -O. Fixed test oracles cover exact packets.
if new['facts']['restart_attempted'] is not False or new['unresolved'] != ['restart_confirmation']:
    raise RuntimeError('recovery oracle failed')
for packet in [old, new]:
    serialized = lab.wire(packet)
    if any(secret in serialized for secret in [lab.TEXT1, lab.TEXT2, 'text_hash', 'internal_note']):
        raise RuntimeError('private source leaked')

再保存以下 run_checks.py。它执行更新后的 29 项固定预期测试，两个 --mutate 模式真正替换实现路径，而不是打印一条“模拟失败”。text-binding 让摘要函数忽略传入文字；candidate-current 用候选包自报的版本取代应用当前版本。两者都应被固定断言捕获，退出 1。测试使用 unittest 的断言方法，优化模式不会移除它们。

"""Run the fixed 29-test suite, optionally introducing a real defective path."""
import argparse
import json
import sys
import unittest
import revision_packet as lab

parser = argparse.ArgumentParser()
parser.add_argument('--mutate', choices=['text-binding', 'candidate-current'])
args = parser.parse_args()
if args.mutate == 'text-binding':
    # Defect: ignore what text was actually reviewed for source revision 7.
    original_hash = lab.text_hash
    lab.text_hash = lambda text: lab.REVIEWS['R1']['text_hash']
elif args.mutate == 'candidate-current':
    # Defect: discard application current and authorize using candidate metadata.
    original_receive = lab.receive
    def trusting_receive(text, current):
        if type(text) is str:
            try:
                candidate = json.loads(text)
                if type(candidate) is dict and type(candidate.get('review_revision')) is str:
                    current = candidate['review_revision']
            except (ValueError, RecursionError):
                pass
        return original_receive(text, current)
    lab.receive = trusting_receive
# Import after installing the mutant so imported function references use it.
suite = unittest.defaultTestLoader.loadTestsFromName('test_revision_packet')
result = unittest.TextTestRunner(verbosity=2).run(suite)
if result.testsRun != 29:
    raise RuntimeError('Expected exactly 29 tests')
sys.exit(0 if result.wasSuccessful() else 1)

运行顺序如下。原例 37 项测试与修订后的 29 项分别执行；各个负对照之后再运行正常版本，正常版本应恢复退出 0。所有命令同时在正常模式和 -O 模式实测。没有真实 SDK、模型请求、网络传输或生产并发测试。

python -m unittest test_context_packet
python run_checks.py
python run_checks.py --mutate text-binding
python run_checks.py --mutate candidate-current
python run_checks.py
python demo_revision_packet.py
python -O -m unittest test_context_packet
python -O run_checks.py
python -O run_checks.py --mutate text-binding
python -O run_checks.py --mutate candidate-current
python -O run_checks.py
python -O demo_revision_packet.py

2026-10-14，Python 3.12.14 本地输出如下：

baseline corrected input: {"private_profile":{"contact":"PRIVATE_ADDRESS_8Q"},"records":[{"id":"MSG-01","role":"user","text":"旧会话：家属就诊备注 PRIVATE_HEALTH_4R，与设备报修无关。","type":"message"},{"facts":{"device_model":"PX-2"},"id":"MSG-02","private":{"address":"PRIVATE_ADDRESS_8Q"},"role":"user","text":"PX-2 打不开。上次收货地址为 PRIVATE_ADDRESS_8Q。","type":"message"},{"facts":{"error_code":"E17"},"id":"TOOL-03","private":{"token":"PRIVATE_TOKEN_2Z"},"text":"模拟诊断：错误码 E17；内部凭据标记 PRIVATE_TOKEN_2Z。","type":"tool_result"},{"facts":{"restart_attempted":true},"id":"MSG-04","role":"assistant","text":"更正：尚未重新启动设备。","type":"message"},{"id":"MSG-05","role":"assistant","text":"旧摘要：PX-2，收货地址 PRIVATE_ADDRESS_8Q；家属备注 PRIVATE_HEALTH_4R。","type":"message"}],"task_id":"CASE-1042","unresolved":["indicator_state"]}
baseline unchanged packet: {"evidence":[{"field":"device_model","ref":"EV-01","sha256":"5bfeb538fa14360c7b9ebd0043cecca923a2b3c11090582a125337f2fc38e3d3","source_id":"MSG-02","value":"PX-2"},{"field":"error_code","ref":"EV-02","sha256":"d6812b9590a9ee706f5cc1ceb8b07489d8d4192682cbb8f862c3134d80b0db65","source_id":"TOOL-03","value":"E17"},{"field":"restart_attempted","ref":"EV-03","sha256":"493c65e95d60b903c4bef5faf1b5e38d97ea24feacaa4c2f7e21b83f7e466c80","source_id":"MSG-04","value":true}],"facts":[{"evidence_ref":"EV-01","field":"device_model","value":"PX-2"},{"evidence_ref":"EV-02","field":"error_code","value":"E17"},{"evidence_ref":"EV-03","field":"restart_attempted","value":true}],"objective":"choose_next_diagnostic_question","schema_version":1,"target":"device-triage","task_id":"CASE-1042","unresolved":["indicator_state"]}
baseline wrong next question: {"evidence_refs":["EV-01","EV-02","EV-03"],"question":"重新启动后，指示灯是常亮、闪烁还是熄灭？","status":"needs_information","task_id":"CASE-1042"}
review R1 source: {"records":[{"facts":{"device_model":"PX-2","error_code":"E17","restart_attempted":true},"id":"MSG-04","internal_note":"本地备注，不转交","revision":7,"text":"设备 PX-2；已重启，错误码 E17。"}],"task_id":"CASE-1042"}
review R1 packet: {"evidence":{"source_id":"MSG-04","source_revision":7},"facts":{"device_model":"PX-2","error_code":"E17","restart_attempted":true},"review_revision":"R1","schema_version":2,"target":"device-triage","task_id":"CASE-1042","unresolved":["indicator_state"]}
review R1 answer: {"question":"重新启动后，指示灯是常亮、闪烁还是熄灭？","review_revision":"R1","status":"needs_information"}
text-only correction input: {"records":[{"facts":{"device_model":"PX-2","error_code":"E17","restart_attempted":true},"id":"MSG-04","internal_note":"本地备注，不转交","revision":7,"text":"更正：设备 PX-2；尚未重启，错误码 E17。"}],"task_id":"CASE-1042"}
text-only correction rejected: {"error":"source_text_changed"}
old packet with current R2: {"error":"stale_review"}
review R2 source: {"records":[{"facts":{"device_model":"PX-2","error_code":"E17","restart_attempted":false},"id":"MSG-04","internal_note":"本地备注，不转交","revision":8,"text":"更正：设备 PX-2；尚未重启，错误码 E17。"}],"task_id":"CASE-1042"}
recovered packet: {"evidence":{"source_id":"MSG-04","source_revision":8},"facts":{"device_model":"PX-2","error_code":"E17","restart_attempted":false},"review_revision":"R2","schema_version":2,"target":"device-triage","task_id":"CASE-1042","unresolved":["restart_confirmation"]}
recovered answer: {"question":"请先确认是否已经重新启动设备。","review_revision":"R2","status":"needs_information"}

text-binding 负对照使更正原文与空白变化未被拒绝，并破坏 R2 恢复投影；candidate-current 负对照使旧包和未知的应用当前版本检查失效。失败表现是测试输出 FAILED，进程退出 1。去掉 --mutate 后应再次得到 Ran 29 tests / OK。不要通过修改固定预期来消除这些失败。

实际处理中，source_text_changed 要返回审核侧查看更正，source_revision_changed 要重新读取对应来源，stale_review 要丢弃旧包并从当前审核状态重建，packet_mismatch 要查明事实、证据和未决项哪处未同步。不要统一自动重试原包。恢复后必须同时核对 false、restart_confirmation 和“请先确认是否已经重新启动设备。”，仅换 R2 标签不构成恢复。

2026-10-20 在 Python 3.12.14 中从页面代码重新建立干净目录：原例 37 项和修订后 29 项测试在正常及 -O 模式均通过，两份演示均运行成功；两种已有变异模式均退出 1，移除变异后恢复退出 0。新溢出反例对修订前实现抛出原生 ValueError，修订后统一抛出 Rejected: bad_json；19 个新增子用例均按固定错误码通过。此次未运行真实 SDK、模型请求、身份认证或生产并发场景。

查阅日期：2026 年 10 月 14 日。OpenAI Agents SDK handoffs 文档区分 handoff 参数与接收 Agent 的主输入：input_type 不会自动替换下游会话输入，改变接收内容需要相应输入处理。本文新增的来源版本、审核快照与恢复规则均为原创应用契约，不把它们描述成 SDK 自带能力。

Python hashlib 文档说明 SHA-256 的字节输入与摘要接口；本例只用它检测已审核文字是否变化，不把摘要当签名或来源认证。Python json 文档说明 object_pairs_hook、parse_constant 以及重复名称和非有限数的默认处理，本例在接收入口显式收紧这些行为。JSON 的排序与紧凑序列化只用于本地固定数据比较，不声称是跨语言标准化签名格式。

这份补充解决的具体问题是：原文变化之后，旧的最小事实与旧问题还能否继续向下游流动。它没有评估语言模型的更正理解能力，也不验证真实审核人员、长期会话存储或后续工具读取。接入实际系统时，应以同样的更正夹具观察真正送给下游模型的输入；目前这些接线均未测试。

上述 Python 当前文档入口在核对时解析为 Python 3.15 文档；实际运行解释器为 3.12.14。本文所用标准库接口已在该实测版本执行，未据文档页面版本推断本地运行版本。

本次数字边界核对使用 Python 3.12 json 文档中的 parse_float、parse_constant 与 allow_nan 说明，以及 math.isfinite 文档；查阅日期为 2026-10-20。解析时拒绝非有限结果，不改变这份固定工单的事实、证据或审核版本规则。