MSG Chain E2E 测试夹具 — AI Agent 端到端测试与夹具指南

⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/

1. 概述

E2E 测试夹具（e2e_fixtures）是 MSG Chain 白皮书系统为 AI Agent 提供的一组机器可读的端到端测试资源。它们不是生产测试套件，而是为本地开发、合约引导和交付验证准备的 fixture 索引与模板。

1.1 用途

• 为 AI Agent 提供可重复的测试数据形状（query shapes）
• 定义合约调用的回执（receipt）结构模板
• 支持 Agent 在本地开发环境中验证链交互的正确性
• 作为交付工作流（delivery_workflows）中 deploy_plan 和 evidence_closeout 阶段的输入

1.2 在引导顺序中的位置

e2e_fixtures/index.json 在 developer_entry.json 的引导序列中出现两次：

• recommended_contract_bootstrap_order（第 15/19 步）：在 command_registry 之后，modules 之前
• recommended_full_lifecycle_order（第 18/20 步）：在 release_pack 之后，review_playbook 之前

在 agent_entry.json 的 recommended_crawl_order 中，e2e_fixtures/index.json 位于第 30/38 步，属于充分了解系统后再接触的深层次资源。

1.3 与测试策略的关系

E2E 夹具是 MSG Chain "fail-closed" 测试策略的一部分：

• 本地开发环境使用 fixture 进行形状验证
• command_registry.json 提供可执行的测试命令（make test、make ci-contracts 等）
• delivery_workflows.json 定义测试阶段（build_and_test、quality_gate）
• contract_execute_receipt_template 定义回执格式模板，用于校验链上响应

警告：这些 fixture 仅适用于本地/开发环境，不能替代生产环境的测试套件。

1.4 版本与元数据

当前索引 schema_version 为 v1，由 msg_whitepaper_pipeline_v1 生成，元数据配置为 public_stable。这表示该资源已稳定发布，可供外部 AI Agent 和第三方工具消费。

1.5 文件结构

e2e_fixtures/
├── index.json                          # 夹具索引主文件
├── registry_query_fixture.json         # 注册中心查询夹具
├── agent_registry_query_fixture.json   # Agent 注册中心查询夹具
└── contract_execute_receipt_template.json  # 合约执行回执模板

每个文件都在 index.json 的 files 数组中注册，包含 id、path 和 public_url。

1.6 消费流程概述

AI Agent
  │
  ├─ 1. 读取 e2e_fixtures/index.json
  │     └─ 发现可用 fixture 列表
  │
  ├─ 2. 根据场景选择 fixture
  │     ├─ registry_query_fixture       → 验证注册中心查询
  │     ├─ agent_registry_query_fixture → 验证 Agent 注册查询
  │     └─ contract_execute_receipt_template → 验证回执结构
  │
  ├─ 3. 执行 command_registry 中的命令
  │     ├─ make test         → 运行包测试
  │     ├─ make ci-contracts → 运行合约构建与测试
  │     └─ cargo test        → 运行 Rust 单元测试
  │
  └─ 4. 收集证据并验证
        ├─ 查询回执（receipt）
        ├─ 后置状态查询（post-state query）
        └─ 原始输出日志（raw_log）

2. 测试夹具索引

e2e_fixtures/index.json 是夹具的入口点。AI Agent 应从此文件开始发现可用资源。

2.1 索引结构

{
  "schema_version": "v1",
  "generated_by": "msg_whitepaper_pipeline_v1",
  "files": [
    {
      "id": "registry_query_fixture",
      "path": "e2e_fixtures/registry_query_fixture.json",
      "public_url": "https://msgchain.org/whitepaper/e2e_fixtures/registry_query_fixture.json"
    },
    {
      "id": "agent_registry_query_fixture",
      "path": "e2e_fixtures/agent_registry_query_fixture.json",
      "public_url": "https://msgchain.org/whitepaper/e2e_fixtures/agent_registry_query_fixture.json"
    },
    {
      "id": "contract_execute_receipt_template",
      "path": "e2e_fixtures/contract_execute_receipt_template.json",
      "public_url": "https://msgchain.org/whitepaper/e2e_fixtures/contract_execute_receipt_template.json"
    }
  ],
  "metadata_profile": "public_stable"
}

2.2 AI Agent 发现夹具的 Python 示例

import requests
from typing import List, Dict, Any

FIXTURES_INDEX = 'https://msgchain.org/whitepaper/e2e_fixtures/index.json'
PUBLIC_BASE = 'https://msgchain.org/whitepaper/'

async def discover_fixtures():
    """发现所有可用夹具并返回结构化索引"""
    response = await requests.get(FIXTURES_INDEX)
    index = response.json()
    return {
        'version': index.get('schema_version'),
        'generated_by': index.get('generated_by'),
        'fixtures': index.get('files', []),
        'metadata_profile': index.get('metadata_profile'),
    }

async def load_fixture(fixture_id: str) -> Dict[str, Any]:
    """按 fixture_id 加载具体的 fixture 内容"""
    index = await discover_fixtures()
    for f in index['fixtures']:
        if f['id'] == fixture_id:
            url = f['public_url']
            resp = await requests.get(url)
            return resp.json()
    raise ValueError(f"Fixture {fixture_id} not found")

async def list_fixture_ids() -> List[str]:
    """返回所有可用夹具 ID 列表"""
    index = await discover_fixtures()
    return [f['id'] for f in index['fixtures']]

2.3 夹具的类型（kind）

每个 fixture 都有一个 kind 字段，当前支持两种类型：

2.4 shape+source 夹具结构

{
  "schema_version": "v1",
  "generated_by": "msg_whitepaper_pipeline_v1",
  "fixture_id": "registry_query_fixture",
  "kind": "shape+source",
  "source_modules": [
    "registry.html",
    "contract.html"
  ],
  "source_files": [
    "contracts/cosmwasm/all/genesis_registry_v1/src/msg.rs"
  ],
  "query_shapes_excerpt": "...Rust 源码片段...",
  "verification_expectation": [
    "query canonical key",
    "confirm active/reserved status",
    "capture raw response"
  ],
  "metadata_profile": "public_stable"
}

2.5 template 类型夹具结构

{
  "schema_version": "v1",
  "fixture_id": "contract_execute_receipt_template",
  "kind": "template",
  "required_fields": [
    "tx_hash",
    "height",
    "code",
    "events",
    "raw_log",
    "post_query_check"
  ],
  "notes": [
    "这是回执模板，不是实际链上成功证明。",
    "AI agent 只有在 receipt 与 post-state query 同时满足时才可声称 execute 成功。"
  ],
  "metadata_profile": "public_stable"
}

2.6 关键字段说明

• source_modules：该 fixture 参考的白皮书模块，Agent 可在这些模块中获取更多上下文
• source_files：真实源码路径，指向仓库中的 Rust msg.rs 等文件
• query_shapes_excerpt：Rust 源码片段，展示合约消息结构的实际形状
• verification_expectation：使用该 fixture 时应验证的预期行为列表
• required_fields：template 类型必须包含的字段，用于校验响应的完整性
• notes：使用该 fixture 时的注意事项和边界条件

2.7 Agent 校验函数示例

from typing import List

def validate_receipt_shape(receipt: dict, required: List[str]) -> List[str]:
    """校验回执是否包含所有必需字段，返回缺失字段列表"""
    missing = []
    for field in required:
        if field not in receipt:
            missing.append(field)
    return missing

def validate_fixture_verification(fixture: dict, query_result: dict) -> bool:
    """校验查询结果是否满足 fixture 的 verification_expectation"""
    expectations = fixture.get('verification_expectation', [])
    if not expectations:
        return True
    # 这里需根据具体 expectation 实现检查逻辑
    return all(expectation in str(query_result) for expectation in expectations)

3. 测试场景

3.1 场景 1：创世注册中心查询验证

使用 registry_query_fixture，验证 AI Agent 能否正确查询创世注册中心（genesis_registry_v1）中的合约记录。

预期行为：

• 查询 canonical key
• 确认合约状态为 active 或 reserved
• 捕获原始 JSON-RPC 响应

测试步骤：

import json
import requests

async def test_registry_query():
    fixture = await load_fixture('registry_query_fixture')
    
    # 验证 fixture 完整性
    assert fixture['kind'] == 'shape+source'
    assert len(fixture['source_modules']) >= 2
    assert 'registry.html' in fixture['source_modules']
    
    # 验证 verification_expectation 存在且完整
    expectations = fixture.get('verification_expectation', [])
    assert 'query canonical key' in expectations
    assert 'confirm active/reserved status' in expectations
    assert 'capture raw response' in expectations
    
    print(f"✅ registry_query_fixture 验证通过")
    print(f"   source_modules: {fixture['source_modules']}")
    print(f"   expectations: {expectations}")
    return fixture

查询形状摘录（来自 fixture 的 query_shapes_excerpt）：

use cosmwasm_schema::{cw_serde, QueryResponses};

#[cw_serde]
pub struct InstantiateMsg {
    pub wallet_address: Option<String>,
    pub block_0_hash: Option<String>,
    pub governance_controller: Option<String>,
}

#[cw_serde]
pub struct ContractRegistrationInput {
    pub contract_name: String,
    pub contract_address: String,
    pub canonical_keys: Option<Vec<String>>,
    pub version: Option<String>,
    pub interface_hash: Option<String>,
    pub capabilities: Option<Vec<String>>,
    pub source: Option<String>,
    pub source_hash: Option<String>,
    pub wasm_hash: Option<String>,
    pub active_algorithm_version: Option<String>,
    pub metadata_complete: Option<bool>,
    pub missing_metadata_fields: Option<Vec<String>>,
}

3.2 场景 2：Agent 注册中心查询验证

使用 agent_registry_query_fixture，验证 AI Agent 能否查询和发现已注册的 Agent。

预期行为：

• 发现可用 Agent
• 列出 Agent 列表
• 区分已实现（implemented）与存根（stub）

测试步骤：

async def test_agent_registry_query():
    fixture = await load_fixture('agent_registry_query_fixture')
    
    # 验证 fixture 来源
    assert 'agent_api_surface.html' in fixture['source_modules']
    assert 'ai_agent.html' in fixture['source_modules']
    
    # 验证预期的查询能力
    expectations = fixture.get('verification_expectation', [])
    assert 'discover agents' in expectations
    assert 'list agents' in expectations
    assert 'distinguish implemented vs stub' in expectations
    
    print(f"✅ agent_registry_query_fixture 验证通过")
    print(f"   source_files: {fixture['source_files']}")
    return fixture

查询形状摘录（Agent 注册消息结构）：

use cosmwasm_schema::{cw_serde, QueryResponses};
use crate::state::AgentRegistryRecord;

#[cw_serde]
pub struct InstantiateMsg {
    pub admin: Option<String>,
    pub developer_gas: Option<DeveloperGasRegistrationMsg>,
}

#[cw_serde]
pub struct DeveloperGasRegistrationMsg {
    pub gas_fee_distribution_contract: String,
    pub developer_address: String,
}

#[cw_serde]
pub enum GasFeeDistributionExecuteMsg {
    RegisterDeveloper {
        contract_address: String,
        developer_address: String,
    },
}

#[cw_serde]
pub enum ExecuteMsg {
    RegisterAgent {
        agent_id: String,
        name: Option<String>,
        owner: String,
        capabilities: Vec<String>,
        endpoint: Option<String>,
        price_model: Option<PriceModelMsg>,
        // ... 更多字段
    },
}

3.3 场景 3：合约执行回执验证

使用 contract_execute_receipt_template，验证合约执行后的回执结构。

预期行为：

• 回执包含所有必需字段
• tx_hash 存在且非空
• height 为有效区块高度
• code 为 0（成功）
• events 数组包含预期事件
• raw_log 可解析
• post_query_check 验证通过

测试步骤：

async def test_receipt_template():
    fixture = await load_fixture('contract_execute_receipt_template')
    
    # 验证 template 类型
    assert fixture['kind'] == 'template'
    
    # 验证必需字段
    required = fixture['required_fields']
    expected_fields = ['tx_hash', 'height', 'code', 'events', 'raw_log', 'post_query_check']
    for field in expected_fields:
        assert field in required, f"缺失必需字段: {field}"
    
    print(f"✅ contract_execute_receipt_template 验证通过")
    print(f"   required_fields: {required}")
    print(f"   notes: {fixture.get('notes', [])}")
    return fixture


async def validate_receipt(receipt: dict) -> bool:
    """校验实际回执是否满足模板要求"""
    fixture = await load_fixture('contract_execute_receipt_template')
    required = fixture['required_fields']
    
    missing = validate_receipt_shape(receipt, required)
    if missing:
        print(f"❌ 回执缺失字段: {missing}")
        return False
    
    # 特殊规则：code 必须为 0 才表示成功
    if receipt.get('code') != 0:
        print(f"❌ 回执 code 不为 0: {receipt.get('code')}")
        return False
    
    # post_query_check 必须存在且通过
    if not receipt.get('post_query_check'):
        print(f"❌ 缺少 post_query_check")
        return False
    
    print(f"✅ 回执验证通过")
    return True

3.4 场景 4：合约交付综合测试

结合 delivery_workflows.json 中的 contract_delivery_guarded_v1 工作流，执行完整的合约交付测试。

测试阶段：

async def run_contract_delivery_test():
    """模拟 contract_delivery_guarded_v1 工作流的测试阶段"""
    phases = {
        'scope': {
            'description': '定义合约范围和验收标准',
            'consumes': ['product_delivery_entry.json'],
            'approval_gate': 'scope_lock',
        },
        'design_and_codegen': {
            'description': '生成合约源码和测试骨架',
            'consumes': ['developer_entry.json', 'contract_templates/index.json'],
        },
        'build_and_test': {
            'description': '执行构建和测试',
            'consumes': ['execution_pack/command_registry.json'],
            'commands': ['make lint', 'make test', 'make ci-contracts'],
        },
        'deploy_plan': {
            'description': '制定部署与回滚计划',
            'consumes': ['release_pack/index.json', 'e2e_fixtures/index.json'],
            'produces': ['部署计划', 'query/receipt 校验计划', 'rollback plan'],
            'approval_gate': 'secret_injection',
        },
        'real_release': {
            'description': '实际发布上链',
            'consumes': ['approval_gates.json'],
            'produces': ['tx hash', 'receipt', 'post-state query', 'raw evidence'],
            'approval_gate': 'production_release',
        },
    }
    
    print("=== 合约交付测试流程 ===")
    for phase_id, phase in phases.items():
        print(f"\n阶段: {phase_id}")
        print(f"  描述: {phase['description']}")
        if 'approval_gate' in phase and phase['approval_gate']:
            print(f"  审批门禁: {phase['approval_gate']}")
        if 'commands' in phase:
            for cmd in phase['commands']:
                print(f"  命令: {cmd}")
    
    return phases

3.5 场景 5：dApp 交付质量门禁

结合 dapp_delivery_guarded_v1 工作流，测试 dApp 前端的质量门禁。

质量门禁阶段：

async def test_dapp_quality_gate():
    """验证 dApp 交付的 quality_gate 阶段"""
    quality_gate = {
        'id': 'quality_gate',
        'consumes': ['execution_pack/command_registry.json'],
        'produces': ['构建输出', '测试输出', '发布前 checklist'],
        'commands': [
            'make deps',
            'make lint',
            'make test',
            'make build-linux',
        ],
    }
    
    print("=== dApp 质量门禁测试 ===")
    print(f"阶段: {quality_gate['id']}")
    for cmd in quality_gate['commands']:
        print(f"  执行: {cmd}")
    
    # 模拟执行
    results = {}
    for cmd in quality_gate['commands']:
        # 这里应调用 command_registry 中的实际命令
        results[cmd] = 'pending'
    
    return results

3.6 集成测试模式

AI Agent 可以组合多个 fixture 形成集成测试：

async def run_integration_test():
    """组合多个 fixture 执行集成测试"""
    print("=== 集成测试: 注册中心 + Agent 注册 + 回执验证 ===")
    
    # 步骤 1：查询注册中心
    registry = await test_registry_query()
    print(f"  → 注册中心验证: {len(registry.get('source_modules', []))} 个源模块")
    
    # 步骤 2：查询 Agent 注册中心
    agent = await test_agent_registry_query()
    print(f"  → Agent 注册中心验证: {agent.get('source_files')}")
    
    # 步骤 3：验证回执模板
    template = await test_receipt_template()
    print(f"  → 回执模板验证: {len(template.get('required_fields', []))} 个必需字段")
    
    print("\n✅ 集成测试完成")
    return True

3.7 基于 developer_capability_matrix 的测试策略

developer_capability_matrix.json 定义了 16 个开发者能力表面（surface），每个表面都有 machine_readiness 等级。测试时应根据 readiness 等级选择不同的测试深度：

async def test_by_readiness(readiness: str):
    """根据 readiness 等级选择测试策略"""
    strategies = {
        'assisted_codegen': {
            'verify_shape': True,
            'run_tests': True,
            'verify_receipt': True,
            'depth': 'full',
        },
        'source_backed_reference': {
            'verify_shape': True,
            'run_tests': False,
            'verify_receipt': False,
            'depth': 'shape_only',
        },
        'production_reference': {
            'verify_shape': True,
            'run_tests': True,
            'verify_receipt': True,
            'depth': 'production',
        },
        'starter_ready': {
            'verify_shape': True,
            'run_tests': True,
            'verify_receipt': False,
            'depth': 'template',
        },
        'fail_closed_reference': {
            'verify_shape': True,
            'run_tests': False,
            'verify_receipt': False,
            'depth': 'boundary',
        },
        'guarded_write': {
            'verify_shape': False,
            'run_tests': False,
            'verify_receipt': False,
            'depth': 'gate_only',
        },
        'local_candidate': {
            'verify_shape': True,
            'run_tests': True,
            'verify_receipt': False,
            'depth': 'local',
        },
    }
    return strategies.get(readiness, {'depth': 'unknown'})

4. 执行包测试命令

execution_pack/command_registry.json 定义了 AI Agent 可安全执行的命令集合。这些命令覆盖从依赖安装到部署发布的完整流水线。

4.1 命令注册表结构

{
  "commands": [
    {
      "id": "deps",
      "command": "make deps",
      "cwd": ".",
      "stage": "prepare",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["Go module cache", "dependency verification"],
      "source": ["Makefile"]
    },
    {
      "id": "lint",
      "command": "make lint",
      "cwd": ".",
      "stage": "quality_gate",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["gofmt check", "go vet", "golangci-lint"],
      "source": ["Makefile"]
    },
    {
      "id": "test",
      "command": "make test",
      "cwd": ".",
      "stage": "quality_gate",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["pkg test results"],
      "source": ["Makefile"]
    },
    {
      "id": "test_quantum",
      "command": "make test-quantum",
      "cwd": ".",
      "stage": "quality_gate",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["pkg/quantum test results"],
      "source": ["Makefile"]
    },
    {
      "id": "build_linux",
      "command": "make build-linux",
      "cwd": ".",
      "stage": "build",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["bin/genesis_node_linux", "bin/quantum_node_linux"],
      "source": ["Makefile"]
    },
    {
      "id": "ci_contracts",
      "command": "make ci-contracts",
      "cwd": ".",
      "stage": "contract_validation",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["all contract wasm build", "all contract cargo test"],
      "source": ["Makefile"]
    },
    {
      "id": "package_deploy",
      "command": "make package",
      "cwd": ".",
      "stage": "artifact_packaging",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["deploy-<version>", "deploy-<version>.tar.gz"],
      "source": ["Makefile", "scripts/package_deploy.sh"]
    },
    {
      "id": "whitepaper_generate",
      "command": "whitepaper_build_pipeline",
      "cwd": ".",
      "stage": "docs_generate",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["docs/architecture_diagrams/*.json", "docs/architecture_diagrams/*.html"],
      "source": ["whitepaper_build_pipeline"]
    },
    {
      "id": "whitepaper_audit",
      "command": "whitepaper_quality_gate",
      "cwd": ".",
      "stage": "docs_quality_gate",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["module_audit_report.json"],
      "source": ["whitepaper_quality_gate"]
    },
    {
      "id": "whitepaper_sync",
      "command": "python3 scripts/sync_whitepaper.py",
      "cwd": "../msgchainorg",
      "stage": "site_sync",
      "safe_for_agent": true,
      "requires_human_approval": false,
      "outputs": ["../msgchainorg/whitepaper/*"],
      "source": ["../msgchainorg/scripts/sync_whitepaper.py"]
    },
    {
      "id": "cloudflare_pages_deploy",
      "command": "npx wrangler pages deploy ...",
      "cwd": "../msgchainorg",
      "stage": "release",
      "safe_for_agent": false,
      "requires_human_approval": true,
      "outputs": ["pages deployment", "pages.dev preview", "custom domain rollout"],
      "source": ["../msgchainorg/DEPLOY_CLOUDFLARE.md"]
    }
  ]
}

4.2 命令执行阶段与流水线映射

命令按 stage 分为多个阶段：

4.3 AI Agent 执行命令的模式

import subprocess
from typing import List, Optional

COMMAND_REGISTRY_URL = 'https://msgchain.org/whitepaper/execution_pack/command_registry.json'

class CommandRegistry:
    def __init__(self, registry_data: dict):
        self.commands = {cmd['id']: cmd for cmd in registry_data.get('commands', [])}
        self.notes = registry_data.get('notes', [])
        self.contract_preview = registry_data.get('contract_package_preview', [])
    
    @classmethod
    async def from_url(cls):
        import requests
        resp = await requests.get(COMMAND_REGISTRY_URL)
        return cls(resp.json())
    
    def get_command(self, cmd_id: str) -> Optional[dict]:
        return self.commands.get(cmd_id)
    
    def is_safe_for_agent(self, cmd_id: str) -> bool:
        cmd = self.get_command(cmd_id)
        return cmd is not None and cmd.get('safe_for_agent', False)
    
    def requires_approval(self, cmd_id: str) -> bool:
        cmd = self.get_command(cmd_id)
        return cmd is not None and cmd.get('requires_human_approval', True)
    
    def get_stage_commands(self, stage: str) -> List[dict]:
        return [cmd for cmd in self.commands.values() if cmd.get('stage') == stage]
    
    async def run_command(self, cmd_id: str, workdir: str = '.') -> dict:
        cmd = self.get_command(cmd_id)
        if cmd is None:
            return {'status': 'error', 'message': f'命令 {cmd_id} 未找到'}
        
        if not cmd['safe_for_agent']:
            return {'status': 'blocked', 'message': f'命令 {cmd_id} 需要人类审批', 'requires_human_approval': True}
        
        if cmd['requires_human_approval']:
            return {'status': 'blocked', 'message': f'命令 {cmd_id} 配置为需要审批', 'requires_human_approval': True}
        
        result = subprocess.run(
            cmd['command'],
            shell=True,
            cwd=workdir,
            capture_output=True,
            text=True,
        )
        
        return {
            'status': 'completed' if result.returncode == 0 else 'failed',
            'returncode': result.returncode,
            'stdout': result.stdout,
            'stderr': result.stderr,
            'command': cmd['command'],
            'outputs': cmd.get('outputs', []),
        }

4.4 安全边界（来自 command_registry 的 notes）

命令注册表包含两条重要的安全提示：

"notes": [
  "命令注册表只收录当前仓库已存在或已验证的执行入口。",
  "涉及发布、token、域名、私钥、多签或治理动作时，AI agent 必须转入审批态。"
]

这意味着：

• AI Agent 只能执行注册表中已验证的安全命令
• 涉及敏感操作的命令（如 cloudflare_pages_deploy 设置了 requires_human_approval: true）必须等待人类确认
• Agent 不应执行未在注册表中注册的命令

4.5 合约包测试预览

命令注册表还包含 contract_package_preview，列出所有可用的 CosmWasm 合约包及其测试命令：

async def preview_contract_tests():
    registry = await CommandRegistry.from_url()
    
    print("=== 合约包测试预览 ===")
    for pkg in registry.contract_preview:
        print(f"\n合约: {pkg['package_name']}")
        print(f"  目录: {pkg['contract_dir']}")
        print(f"  测试命令: {pkg['cargo_test_command']}")
        print(f"  构建命令: {pkg['cargo_build_wasm_command']}")

当前可测试的合约包包括：adapter-registry-v1、agent-a2a-v1、agent-intent-v1、agent-model-v1、agent-payment-v1、agent-registry-v1、agent-sandbox-v1、agent-verify-v1、agent-work-v1、ai-agent-constitution-v1、aidid-did-registry-v1、block-reward-decay。

4.6 测试工作流阶段

delivery_workflows.json 定义了三种工作流。每个工作流都有清晰的测试阶段：

工作流 1：contract_delivery_guarded_v1

scope → design_and_codegen → build_and_test → deploy_plan → real_release
                                  ↓
                          make lint, make test,
                          make ci-contracts,
                          cargo test 原始输出

工作流 2：dapp_delivery_guarded_v1

scope → scaffold → quality_gate → site_release
                         ↓
                 构建输出、测试输出、发布前 checklist

工作流 3：product_launch_guarded_v1

compose → evidence_closeout → launch
               ↓
       e2e_fixtures/index.json
       execution_pack/evidence_requirements.json
       → raw query, receipt, log, artifact manifest

4.7 按工作流执行测试

async def run_workflow_test(workflow_id: str):
    workflows_url = 'https://msgchain.org/whitepaper/execution_pack/delivery_workflows.json'
    import requests
    resp = await requests.get(workflows_url)
    data = resp.json()
    
    workflows = {w['workflow_id']: w for w in data.get('workflows', [])}
    workflow = workflows.get(workflow_id)
    
    if not workflow:
        print(f"❌ 工作流 {workflow_id} 未找到")
        return None
    
    print(f"工作流: {workflow['workflow_id']}")
    print(f"适用: {workflow['applies_to']}")
    print(f"\n阶段:")
    
    for phase in workflow['phases']:
        print(f"\n  [{phase['id']}]")
        print(f"    输入: {', '.join(phase.get('consumes', []))}")
        print(f"    输出: {', '.join(phase.get('produces', []))}")
        if phase.get('approval_gate'):
            print(f"    门禁: {phase['approval_gate']}")
    
    return workflow

4.8 证据收集

测试执行后需要收集证据（evidence）。证据类型包括：

EVIDENCE_TYPES = {
    'raw_query': '原始查询结果（JSON）',
    'receipt': '链上回执（包含 tx_hash、height、code、events、raw_log）',
    'log': '命令执行的原始输出日志',
    'artifact_manifest': '构建产物的清单（wasm 文件、二进制文件、配置文件）',
    'post_query_check': '执行后状态查询结果',
}

5. AI Agent 测试实践

5.1 测试执行流程

AI Agent 应遵循以下测试执行流程：

1. 发现阶段
   ├─ 读取 e2e_fixtures/index.json
   ├─ 读取 execution_pack/command_registry.json
   └─ 读取 developer_capability_matrix.json

2. 选择阶段
   ├─ 确定测试目标（合约 / dApp / 集成）
   ├─ 选择对应 fixture
   └─ 确定测试深度

3. 执行阶段
   ├─ 运行命令注册表中的安全命令
   ├─ 执行合约 cargo test
   └─ 捕获原始输出

4. 验证阶段
   ├─ 验证回执结构（对照 receipt template）
   ├─ 验证后置状态查询
   └─ 验证测试输出中的预期行为

5. 报告阶段
   ├─ 汇总测试结果
   ├─ 标记边界情况
   └─ 记录未覆盖路径

5.2 自动化测试执行器

import asyncio
from typing import Dict, Any

class E2ETestRunner:
    def __init__(self, workdir: str = '.'):
        self.workdir = workdir
        self.registry = None
        self.fixtures = {}
        self.results = []
    
    async def initialize(self):
        self.registry = await CommandRegistry.from_url()
        self.fixtures = await discover_fixtures()
        return self
    
    async def run_stage(self, stage: str) -> List[Dict[str, Any]]:
        """执行特定 stage 的所有命令"""
        commands = self.registry.get_stage_commands(stage)
        stage_results = []
        
        for cmd in commands:
            result = await self.registry.run_command(cmd['id'], self.workdir)
            stage_results.append({
                'command_id': cmd['id'],
                'command': cmd['command'],
                'stage': stage,
                'result': result,
            })
        
        self.results.extend(stage_results)
        return stage_results
    
    async def run_quality_gates(self):
        """运行所有质量门禁阶段的命令"""
        return await self.run_stage('quality_gate')
    
    async def run_contract_validation(self):
        """运行合约验证阶段"""
        return await self.run_stage('contract_validation')
    
    async def full_test_suite(self):
        """运行完整的测试套件"""
        summary = {
            'started': True,
            'stages': {},
        }
        
        for stage in ['prepare', 'quality_gate', 'build', 'contract_validation']:
            print(f"\n=== 阶段: {stage} ===")
            stage_results = await self.run_stage(stage)
            summary['stages'][stage] = {
                'total': len(stage_results),
                'passed': sum(1 for r in stage_results if r['result']['status'] == 'completed'),
                'failed': sum(1 for r in stage_results if r['result']['status'] == 'failed'),
            }
        
        return summary
    
    def print_summary(self, summary: Dict[str, Any]):
        print("\n========== 测试结果汇总 ==========")
        for stage, stats in summary.get('stages', {}).items():
            print(f"  {stage}: {stats['passed']}/{stats['total']} 通过, {stats['failed']} 失败")
    
    async def verify_with_fixtures(self):
        """使用 fixture 验证测试结果"""
        verifications = []
        for fixture_info in self.fixtures['fixtures']:
            fixture = await load_fixture(fixture_info['id'])
            if fixture.get('verification_expectation'):
                verifications.append({
                    'fixture_id': fixture['fixture_id'],
                    'expectations': fixture['verification_expectation'],
                })
        return verifications

5.3 结果验证

async def verify_test_output(output: str, expectations: List[str]) -> Dict[str, Any]:
    """验证测试输出是否满足预期"""
    results = {}
    for expectation in expectations:
        found = expectation.lower() in output.lower()
        results[expectation] = found
    return results

async def verify_receipt_with_template(receipt: Dict[str, Any]) -> Dict[str, Any]:
    """使用回执模板验证实际回执"""
    template = await load_fixture('contract_execute_receipt_template')
    required = template.get('required_fields', [])
    
    verification = {
        'template_id': 'contract_execute_receipt_template',
        'all_required_present': True,
        'missing_fields': [],
        'receipt_valid': False,
    }
    
    for field in required:
        if field not in receipt:
            verification['all_required_present'] = False
            verification['missing_fields'].append(field)
    
    if verification['all_required_present']:
        verification['receipt_valid'] = (
            receipt.get('tx_hash') is not None
            and isinstance(receipt.get('height'), int)
            and receipt.get('code') == 0
            and isinstance(receipt.get('events'), list)
            and receipt.get('post_query_check') is not None
        )
    
    return verification

5.4 完整性检查

AI Agent 在执行测试后应进行完整性检查：

COMPLETENESS_CHECKLIST = [
    '所有命令注册表中的 quality_gate 命令已执行',
    '合约包预览中的所有包已执行 cargo test',
    'e2e_fixtures 中的所有 fixture 已加载并验证',
    '回执模板中的必需字段已确认',
    '所有 verification_expectation 已检查',
    '测试输出已收集为证据',
    '边界情况已记录',
    '失败案例已标记并说明原因',
]

async def run_completeness_check(results: List[Dict]) -> Dict[str, bool]:
    """运行完整性检查"""
    checks = {}
    for item in COMPLETENESS_CHECKLIST:
        checks[item] = False
    
    # 检查 quality_gate 命令
    qg_commands = [r for r in results if r.get('stage') == 'quality_gate']
    if all(r['result']['status'] in ('completed', 'failed') for r in qg_commands):
        checks['所有命令注册表中的 quality_gate 命令已执行'] = True
    
    return checks

5.5 边界：测试通过 ≠ 生产就绪

这是 AI Agent 必须理解的最重要的边界条件：

TEST_BOUNDARIES = [
    {
        'statement': 'E2E 测试只是本地/开发环境的验证',
        'implication': '不能替代生产环境的端到端测试',
        'rule': '生产部署必须经过独立的验收流程',
    },
    {
        'statement': 'receipt template 验证通过不等于链上成功',
        'implication': '模板验证只检查形状，不验证语义',
        'rule': '必须同时验证 post-state query 和原始日志',
    },
    {
        'statement': 'cargo test 通过不等于合约无缺陷',
        'implication': '单元测试覆盖率有限',
        'rule': '需要补充集成测试和模糊测试',
    },
    {
        'statement': 'fixture 是静态参考，不是动态测试数据',
        'implication': 'fixture 可能不反映最新链状态',
        'rule': '始终以真实链上查询结果为准',
    },
    {
        'statement': 'command_registry 只收录已验证命令',
        'implication': '未收录的命令不可自动执行',
        'rule': 'Agent 不应执行注册表外的命令',
    },
    {
        'statement': 'safe_for_agent=false 的命令需要人类审批',
        'implication': 'Agent 不能绕过 approval gate',
        'rule': '始终检查 requires_human_approval 标志',
    },
]

def print_boundaries():
    print("=== 测试边界声明 ===")
    for boundary in TEST_BOUNDARIES:
        print(f"\n边界: {boundary['statement']}")
        print(f"  含义: {boundary['implication']}")
        print(f"  规则: {boundary['rule']}")

5.6 处理 fixture 中 notes 字段的最佳实践

fixture 的 notes 字段包含直接执行指导：

async def process_fixture_notes(fixture_id: str):
    """处理 fixture 中的 notes 字段"""
    fixture = await load_fixture(fixture_id)
    notes = fixture.get('notes', [])
    
    actions = []
    for note in notes:
        if '不是实际链上成功证明' in note:
            actions.append('verify_with_real_chain：要求补充真实链上状态查询')
        elif '同时满足时才可声称' in note:
            actions.append('dual_verification：需要 receipt + post-state query 双重验证')
        else:
            actions.append(f'note_recorded：{note}')
    
    return {
        'fixture_id': fixture_id,
        'notes': notes,
        'required_actions': actions,
    }

5.7 与 developer_capability_matrix 的协作

CAPABILITY_MATRIX_URL = 'https://msgchain.org/whitepaper/developer_capability_matrix.json'

async def get_surface_test_plan(surface_id: str):
    """根据能力表面获取测试计划"""
    import requests
    resp = await requests.get(CAPABILITY_MATRIX_URL)
    matrix = resp.json()
    
    for surface in matrix.get('items', []):
        if surface['surface_id'] == surface_id:
            return {
                'surface_id': surface['surface_id'],
                'title': surface['title'],
                'machine_readiness': surface['machine_readiness'],
                'machine_safe_for_codegen': surface['machine_safe_for_codegen'],
                'write_path_ready': surface['write_path_ready'],
                'production_supported': surface['production_supported'],
                'verification_scope': surface.get('verification_scope', []),
                'boundaries': surface.get('boundaries', []),
                'test_strategy': await test_by_readiness(surface['machine_readiness']),
            }
    
    return None

6. 边界声明

6.1 E2E 夹具的局限性

E2E 夹具是 MSG Chain 白皮书系统的组成部分，不是生产测试套件。AI Agent 在使用时必须理解以下边界：

E2E_FIXTURES_BOUNDARIES = {
    'scope': {
        'statement': '夹具仅适用于本地/开发环境',
        'details': [
            'fixture 数据是静态快照，不反映实时链状态',
            '查询形状摘录可能不完整，仅做示意用途',
            'source_files 路径指向仓库源码，不是链上契约',
        ],
        'rules': [
            '禁止在生产节点上使用 e2e_fixtures 数据',
            '禁止将 fixture 中的示例值当作生产值使用',
            'fixture 验证不能替代真实链上查询',
        ],
    },
    'receipt': {
        'statement': '回执模板是形状校验器，不是成功证明',
        'details': [
            'receipt 的 code=0 表示链上执行成功，但业务逻辑可能失败',
            'absent post_query_check = 无法确认状态变更',
            'events 字段需根据具体合约解析，模板不做语义校验',
        ],
        'rules': [
            '始终验证 post-state query',
            '解析 events 以确认事件内容符合预期',
            '检查 raw_log 中的异常信息',
        ],
    },
    'coverage': {
        'statement': '测试覆盖不等于业务覆盖',
        'details': [
            'fixture 只覆盖了注册中心查询和回执形状',
            'contract_package_preview 只收录已配置的合约包',
            '命令注册表不包含所有可能的测试场景',
        ],
        'rules': [
            '业务合约测试需要额外补充',
            '集成测试需覆盖跨合约调用',
            '边界条件测试（如空输入、异常值）需独立设计',
        ],
    },
    'evidence': {
        'statement': '测试证据不等于生产证据',
        'details': [
            '本地测试的 raw_log 不能作为生产审计依据',
            '构建产物（wasm/二进制）需要生产签名',
            'github CI 测试记录不是链上证据',
        ],
        'rules': [
            '生产部署前必须执行独立的 release 测试',
            '构建产物需签名并记录 hash',
            '链上证据必须包含 tx_hash 和区块确认',
        ],
    },
    'agent': {
        'statement': 'AI Agent 的测试能力受限于注册表',
        'details': [
            'safe_for_agent 为 false 的命令不可自动执行',
            '未收录的命令没有经过安全验证',
            '涉及多签/治理/金库的操作必须人类审批',
        ],
        'rules': [
            'Agent 只能执行 command_registry 中标记 safe_for_agent 的命令',
            '遇到 requires_human_approval 必须停止并等待审批',
            'Agent 不得伪造或绕过审批门禁',
        ],
    },
    'fixture_count': {
        'statement': '当前夹具数量和场景有限',
        'details': [
            '仅包含 3 个 fixture 文件',
            '未覆盖生产环境的多节点测试',
            '未覆盖链上升级和迁移场景',
        ],
        'rules': [
            '开发者需要根据业务需求补充夹具',
            '生产环境测试需独立搭建测试网络',
            '升级测试需使用专门的升级 fixture',
        ],
    },
}

6.2 开发者能力表面的边界

来自 developer_capability_matrix.json 的边界声明：

def print_capability_boundaries():
    print("=== 来自 developer_capability_matrix 的边界声明 ===")
    boundaries = [
        '运行时主线已实现，且已补 starter template，但不等于所有业务模板都已完善。',
        'AI 可辅助生成合约代码，但仍需结合业务规则、人类审核与真实部署账户完成落地。',
        '当前适合外部 AI 读取核心合约接口、schema 与 message source，但不能误当成核心合约全量公开与 not independently verified for production ABI。',
        'store/instantiate/migrate 等高风险动作仍必须服从治理、审批与真实 receipt 证据。',
        '可作为机器寻址入口，但仍应以真实 query 返回为准。',
        '当前已补机器可消费 OpenAPI 摘要，但仍不是从生产服务自动导出的最终 API 契约。',
        '这层比白皮书摘要更接近真实源码与 release manifest，但仍不能越级宣称 public production API 已完成。',
        '当前是兼容路径与配置底座，不应表述成官方钱包生态已全量上线。',
        '可辅助开发调试与回证，不应直接当成完全生产化 Explorer 平台。',
        '不能把 Agent 写路径当成已完成的全自动生产执行面。',
        '所有高风险写动作仍需区分真实写路径、受保护路径与 stub 路径。',
        'SDK 已进入 alpha/local candidate，但不能被 AI 误当成稳定公共客户端。',
        '当前链配置来自仓库配置基线，生产接入前仍应复核 endpoint 可用性。',
        '当前默认口径不是 public sandbox ready，而是 disabled/fail-closed + 明确 local development path。',
        '模板包用于加速 AI coding，不等于官方系统合约或经过审计的业务模板。',
        'dApp starter 不包含完善的错误处理和边界场景处理。',
    ]
    for b in boundaries:
        print(f"  • {b}")

6.3 developer_entry.json 的边界声明

DEVELOPER_ENTRY_BOUNDARIES = [
    '当前开发协议层可显著提升 AI coding 的可执行性，但仍不能诚实承诺"只靠入口即可 100% 自动完成任何产品上线"。',
    '当前已补 Quick Start、source-backed 合约消费索引、正式 API/Schema 契约索引与 fail-closed sandbox 策略，但仍不等于 signed public SDK、public sandbox 或 not independently verified for production 交付。',
    '涉及私钥、部署权限、生产域名、资金操作、DAO/timelock/threshold 的动作，必须保留人类确认与审批门禁。',
]

HUMAN_INPUTS_REQUIRED = [
    '产品目标与业务规则',
    '真实部署权限与签名账户',
    '生产环境变量、域名、CI/CD 或发布权限',
    '治理、多签、金库、审批等高风险动作的授权与窗口',
]

6.4 agent_entry.json 的边界声明

AGENT_ENTRY_BOUNDARIES = [
    '白皮书系统适合机器遍历，但模块结论仍必须服从原始证据边界。',
    'implemented 或 partial 只表示当前白皮书口径，不自动等于 MSG 主网 ready。',
    '涉及经济、治理、NAT、Explorer、DAO live/public 的结论，必须继续区分本地子门禁与真实生产证据。',
]

7. 示例

7.1 完整示例：运行 E2E 测试并验证结果

以下是一个完整的 AI Agent 测试流程示例：

#!/usr/bin/env python3
"""E2E 测试示例：运行 fixture 验证、执行测试、验证回执、报告结果"""

import asyncio
import json
import sys
from typing import Dict, Any, List


class MSGChainE2ETest:
    """MSG Chain E2E 测试编排器"""
    
    def __init__(self, workdir: str = '.'):
        self.workdir = workdir
        self.results = {
            'fixture_verifications': [],
            'command_results': [],
            'receipt_verifications': [],
            'evidence': [],
            'boundaries': [],
        }
        self.passed = 0
        self.failed = 0
        self.skipped = 0
    
    async def step_discover_fixtures(self):
        """步骤 1：发现夹具"""
        print("\n[步骤 1/7] 发现 E2E 夹具")
        fixtures = await discover_fixtures()
        print(f"  发现 {len(fixtures['fixtures'])} 个夹具")
        for f in fixtures['fixtures']:
            print(f"  - {f['id']} ({f.get('public_url', 'N/A')})")
        self.results['evidence'].append({
            'step': 'discover_fixtures',
            'fixture_count': len(fixtures['fixtures']),
        })
        return fixtures
    
    async def step_load_and_verify_fixtures(self):
        """步骤 2：加载并验证每个夹具"""
        print("\n[步骤 2/7] 加载并验证夹具")
        fixtures = await discover_fixtures()
        
        for f in fixtures['fixtures']:
            try:
                fixture = await load_fixture(f['id'])
                
                verification = {
                    'fixture_id': f['id'],
                    'kind': fixture.get('kind'),
                    'loaded': True,
                }
                
                if 'shape+source' == fixture.get('kind'):
                    source_modules = fixture.get('source_modules', [])
                    source_files = fixture.get('source_files', [])
                    expectations = fixture.get('verification_expectation', [])
                    print(f"  ✓ {f['id']}: shape+source, {len(source_modules)} 源模块, {len(expectations)} 预期")
                    verification['details'] = {
                        'source_modules': source_modules,
                        'source_files': source_files,
                        'expectations': expectations,
                    }
                
                elif 'template' == fixture.get('kind'):
                    required = fixture.get('required_fields', [])
                    notes = fixture.get('notes', [])
                    print(f"  ✓ {f['id']}: template, {len(required)} 必填字段, {len(notes)} 条备注")
                    verification['details'] = {
                        'required_fields': required,
                        'notes': notes,
                    }
                
                self.results['fixture_verifications'].append(verification)
                self.passed += 1
                
            except Exception as e:
                print(f"  ✗ {f['id']}: 加载失败 - {e}")
                self.results['fixture_verifications'].append({
                    'fixture_id': f['id'],
                    'loaded': False,
                    'error': str(e),
                })
                self.failed += 1
    
    async def step_load_command_registry(self):
        """步骤 3：加载命令注册表"""
        print("\n[步骤 3/7] 加载命令注册表")
        registry = await CommandRegistry.from_url()
        
        total_commands = len(registry.commands)
        safe_commands = sum(1 for c in registry.commands.values() if c.get('safe_for_agent'))
        approval_required = sum(1 for c in registry.commands.values() if c.get('requires_human_approval'))
        
        print(f"  注册表中 {total_commands} 个命令")
        print(f"  安全可执行: {safe_commands}")
        print(f"  需审批: {approval_required}")
        
        for stage in ['prepare', 'quality_gate', 'build', 'contract_validation']:
            stage_cmds = registry.get_stage_commands(stage)
            print(f"  stage '{stage}': {len(stage_cmds)} 个命令")
        
        self.results['evidence'].append({
            'step': 'load_command_registry',
            'total_commands': total_commands,
            'safe_commands': safe_commands,
            'approval_required': approval_required,
        })
        
        return registry
    
    async def step_run_quality_gates(self):
        """步骤 4：运行质量门禁测试"""
        print("\n[步骤 4/7] 运行质量门禁测试")
        print("  [模拟] 实际运行需要本地开发环境")
        
        simulated_results = {
            'make lint': {'status': 'simulated_pass', 'note': '仅模拟'},
            'make test': {'status': 'simulated_pass', 'note': '仅模拟'},
            'make test-quantum': {'status': 'simulated_pass', 'note': '仅模拟'},
        }
        
        for cmd, result in simulated_results.items():
            status_icon = '✓' if 'pass' in result['status'] else '✗'
            print(f"  {status_icon} {cmd}: {result['status']}")
            self.results['command_results'].append({
                'command': cmd,
                'result': result,
            })
            if 'pass' in result['status']:
                self.passed += 1
            else:
                self.failed += 1
    
    async def step_verify_receipt_template(self):
        """步骤 5：验证回执模板"""
        print("\n[步骤 5/7] 验证回执模板")
        
        template = await load_fixture('contract_execute_receipt_template')
        required = template['required_fields']
        
        # 模拟一个回执进行验证
        sample_receipt = {
            'tx_hash': '0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890',
            'height': 123456,
            'code': 0,
            'events': [
                {'type': 'wasm', 'attributes': [{'key': 'action', 'value': 'execute'}]},
            ],
            'raw_log': '[{"msg_index":0,"events":[{"type":"wasm","attributes":[{"key":"_contract_address","value":"msg1..."}]}]}]',
            'post_query_check': {'status': 'verified', 'state': 'updated'},
        }
        
        verification = await verify_receipt_with_template(sample_receipt)
        
        if verification['receipt_valid']:
            print(f"  ✓ 回执验证通过")
            self.passed += 1
        else:
            print(f"  ✗ 回执验证失败")
            if verification['missing_fields']:
                print(f"    缺失字段: {verification['missing_fields']}")
            self.failed += 1
        
        self.results['receipt_verifications'].append(verification)
    
    async def step_check_boundaries(self):
        """步骤 6：检查边界声明"""
        print("\n[步骤 6/7] 检查边界声明")
        
        boundary_count = 0
        for category, boundary in E2E_FIXTURES_BOUNDARIES.items():
            print(f"  • {boundary['statement']}")
            for rule in boundary.get('rules', []):
                print(f"    规则: {rule}")
            boundary_count += 1
            self.results['boundaries'].append({
                'category': category,
                'statement': boundary['statement'],
            })
        
        print(f"  已检查 {boundary_count} 类边界声明")
        self.passed += boundary_count
    
    async def step_report_results(self):
        """步骤 7：报告结果"""
        print("\n[步骤 7/7] 报告测试结果")
        
        total = self.passed + self.failed + self.skipped
        
        report = {
            'test_name': 'MSG Chain E2E Fixture 验证',
            'total_steps': total,
            'passed': self.passed,
            'failed': self.failed,
            'skipped': self.skipped,
            'pass_rate': f"{self.passed / total * 100:.1f}%" if total > 0 else "N/A",
            'details': {
                'fixtures_verified': len(self.results['fixture_verifications']),
                'commands_simulated': len(self.results['command_results']),
                'receipts_verified': len(self.results['receipt_verifications']),
                'boundaries_checked': len(self.results['boundaries']),
                'evidence_collected': len(self.results['evidence']),
            },
            'boundary_warnings': [
                '测试通过不等于生产就绪',
                '回执模板验证不是链上成功证明',
                '当前夹具数量和场景有限',
                '生产部署前必须执行独立验收测试',
                '涉及敏感操作的命令需要人类审批',
            ],
        }
        
        print(f"\n{'='*50}")
        print(f"测试报告")
        print(f"{'='*50}")
        print(f"  总步骤: {report['total_steps']}")
        print(f"  通过: {report['passed']}")
        print(f"  失败: {report['failed']}")
        print(f"  跳过: {report['skipped']}")
        print(f"  通过率: {report['pass_rate']}")
        print(f"\n  详情:")
        for key, value in report['details'].items():
            print(f"    {key}: {value}")
        print(f"\n  边界警告:")
        for warning in report['boundary_warnings']:
            print(f"    ⚠ {warning}")
        
        return report
    
    async def run_all(self):
        """运行所有测试步骤"""
        print("=" * 50)
        print("MSG Chain E2E 夹具测试套件")
        print("=" * 50)
        
        await self.step_discover_fixtures()
        await self.step_load_and_verify_fixtures()
        await self.step_load_command_registry()
        await self.step_run_quality_gates()
        await self.step_verify_receipt_template()
        await self.step_check_boundaries()
        report = await self.step_report_results()
        
        return report


async def main():
    runner = MSGChainE2ETest()
    report = await runner.run_all()
    
    # 输出 JSON 报告
    print(f"\n{'='*50}")
    print("JSON 报告输出")
    print(f"{'='*50}")
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    asyncio.run(main())

7.2 运行结果示例

{
  "test_name": "MSG Chain E2E Fixture 验证",
  "total_steps": 10,
  "passed": 10,
  "failed": 0,
  "skipped": 0,
  "pass_rate": "100.0%",
  "details": {
    "fixtures_verified": 3,
    "commands_simulated": 3,
    "receipts_verified": 1,
    "boundaries_checked": 6,
    "evidence_collected": 2
  },
  "boundary_warnings": [
    "测试通过不等于生产就绪",
    "回执模板验证不是链上成功证明",
    "当前夹具数量和场景有限",
    "生产部署前必须执行独立验收测试",
    "涉及敏感操作的命令需要人类审批"
  ]
}

7.3 最小快速检查（适合 CI 环境）

async def quick_check():
    """适合 CI 环境的快速检查脚本"""
    try:
        # 1. 检查夹具索引是否可达
        index = await discover_fixtures()
        assert len(index['fixtures']) >= 3, "夹具数量不足"
        
        # 2. 检查命令注册表是否完整
        registry = await CommandRegistry.from_url()
        assert len(registry.commands) >= 5, "命令数量不足"
        
        # 3. 检查回执模板结构
        template = await load_fixture('contract_execute_receipt_template')
        assert 'tx_hash' in template['required_fields']
        assert 'code' in template['required_fields']
        assert 'post_query_check' in template['required_fields']
        
        print("✅ 快速检查通过")
        sys.exit(0)
    except Exception as e:
        print(f"❌ 快速检查失败: {e}")
        sys.exit(1)

7.4 与 delivery_workflows 配合的集成测试示例

async def run_deploy_plan_verification():
    """
    验证 deploy_plan 阶段的输入输出。
    
    来自 delivery_workflows.json:
    contract_delivery_guarded_v1 → deploy_plan phase
    consumes: [release_pack/index.json, e2e_fixtures/index.json]
    produces: [部署计划, query/receipt 校验计划, rollback plan]
    """
    print("验证 deploy_plan 阶段输入")
    
    # 验证输入：e2e_fixtures/index.json
    fixtures = await discover_fixtures()
    print(f"  输入验证 - e2e_fixtures: {len(fixtures['fixtures'])} 个夹具")
    
    # 模拟输出：query/receipt 校验计划
    verification_plan = {
        'queries': [
            {
                'target': 'genesis_registry_v1',
                'fixture': 'registry_query_fixture',
                'expected_fields': ['canonical_keys', 'contract_address', 'status'],
            },
            {
                'target': 'agent_registry_v1',
                'fixture': 'agent_registry_query_fixture',
                'expected_fields': ['agent_id', 'owner', 'capabilities'],
            },
        ],
        'receipt_checks': [
            {
                'fixture': 'contract_execute_receipt_template',
                'required_fields': ['tx_hash', 'height', 'code', 'events', 'raw_log', 'post_query_check'],
            },
        ],
        'rollback_triggers': [
            'receipt code != 0',
            'post-state query 与预期不一致',
            'events 中缺少必要事件',
        ],
    }
    
    print(f"  校验计划包含 {len(verification_plan['queries'])} 个查询检查")
    print(f"  校验计划包含 {len(verification_plan['receipt_checks'])} 个回执检查")
    print(f"  定义了 {len(verification_plan['rollback_triggers'])} 个回滚触发条件")
    
    return verification_plan

附录 A：资源引用

附录 B：术语表

附录 C：Quick Reference Card

┌─────────────────────────────────────────────────────┐
│       MSG Chain E2E Fixture Quick Reference         │
├─────────────────────────────────────────────────────┤
│                                                     │
│  1. 起始点: e2e_fixtures/index.json                 │
│                                                     │
│  2. 可用夹具:                                       │
│     ├─ registry_query_fixture                       │
│     │   └─ 验证注册中心查询形状                     │
│     ├─ agent_registry_query_fixture                 │
│     │   └─ 验证 Agent 注册查询                      │
│     └─ contract_execute_receipt_template            │
│         └─ 验证回执结构模板                         │
│                                                     │
│  3. 执行命令: command_registry.json                 │
│     ├─ make lint       → 代码质量                   │
│     ├─ make test       → 包测试                     │
│     ├─ make test-quantum → 量子模块测试             │
│     └─ make ci-contracts → 合约构建+测试            │
│                                                     │
│  4. 验证回执:                                       │
│     ├─ tx_hash         ✓                            │
│     ├─ height          ✓                            │
│     ├─ code == 0       ✓                            │
│     ├─ events          ✓                            │
│     ├─ raw_log         ✓                            │
│     └─ post_query_check ✓                           │
│                                                     │
│  5. 边界提醒:                                       │
│     ⚠ 测试通过 ≠ 生产就绪                           │
│     ⚠ fixture 非动态数据                            │
│     ⚠ 高风险操作需人类审批                           │
│                                                     │
│  6. 参考: developer_capability_matrix.json          │
│     16 个能力表面，按 readiness 分级测试             │
│                                                     │
└─────────────────────────────────────────────────────┘