给 MCP 做真实 stdio 契约回归

"""Fixed, synthetic read-only MCP server. No network, account or model."""
import asyncio
import json
import sys
from mcp import types
from mcp.server.lowlevel import Server
from mcp.server.stdio import stdio_server

MODE = sys.argv[1]
if MODE == 'text':
    print('FIXTURE READY', flush=True)
elif MODE == 'json':
    print('{"level":"info","message":"fixture ready"}', flush=True)
elif MODE == 'stderr':
    print('FIXTURE READY', file=sys.stderr, flush=True)
elif MODE != 'none':
    raise ValueError('unknown fixture mode')

server = Server('startup-stream-fixture', version='1')

@server.list_tools()
async def list_tools():
    return [types.Tool(name='read_answer', inputSchema={'type':'object', 'properties':{}})]

@server.call_tool()
async def call_tool(name, arguments):
    if name != 'read_answer':
        raise ValueError('unknown fixture tool')
    return [types.TextContent(type='text', text='7')]

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

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

"""Byte observer for this one finite fixture, NOT a general MCP proxy."""
import json
import os
from pathlib import Path
import subprocess
import sys
import threading
ROOT = Path(__file__).resolve().parent
mode, output = (sys.argv[1], Path(sys.argv[2]))
if not mode in ('none', 'text', 'json', 'stderr'):
    raise AssertionError("Fixture invariant: mode in ('none', 'text', 'json', 'stderr')")
if not output.is_dir():
    raise AssertionError('Fixture invariant: output.is_dir()')
child = subprocess.Popen([sys.executable, '-u', str(ROOT / 'server.py'), mode], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0)
errors = []

def write_all(fd, chunk):
    while chunk:
        count = os.write(fd, chunk)
        if count <= 0:
            raise OSError('zero-byte pipe write')
        chunk = chunk[count:]

def pump(source, destination, filename, close=None):
    try:
        with (output / filename).open('wb') as raw:
            while True:
                chunk = os.read(source, 4096)
                if not chunk:
                    break
                raw.write(chunk)
                raw.flush()
                write_all(destination, chunk)
    except BaseException as exc:
        errors.append({'stream': filename, 'error': type(exc).__name__})
    finally:
        if close is not None:
            close.close()
if not (child.stdin and child.stdout and child.stderr):
    raise AssertionError('Fixture invariant: child.stdin and child.stdout and child.stderr')
incoming = threading.Thread(target=pump, args=(0, child.stdin.fileno(), 'client-stdin.bin', child.stdin))
diagnostic = threading.Thread(target=pump, args=(child.stderr.fileno(), 2, 'server-stderr.bin'))
incoming.start()
diagnostic.start()
pump(child.stdout.fileno(), 1, 'server-stdout.bin')
returncode = child.wait()
incoming.join()
diagnostic.join()
child.stdout.close()
child.stderr.close()
(output / 'relay-result.json').write_text(json.dumps({'server_returncode': returncode, 'relay_errors': errors, 'threads_finished': not incoming.is_alive() and (not diagnostic.is_alive())}, indent=2) + '\n')
sys.exit(returncode if returncode else int(bool(errors)))

"""Official SDK stdio transport; message observer leaves all items unchanged."""
from datetime import timedelta
from importlib.metadata import version
import json
import logging
from pathlib import Path
import platform
import sys
import anyio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.shared.exceptions import McpError
ROOT = Path(__file__).resolve().parent

class DiagnosticFormatter(logging.Formatter):
    """Keep the entire SDK error text, without machine-specific stack paths."""

    def formatException(self, exc_info):
        return f'{exc_info[0].__name__}: {exc_info[1]}'

class ObserveRead:

    def __init__(self, stream):
        self.stream = stream
        self.errors = []
        self.events = []

    async def __aenter__(self):
        await self.stream.__aenter__()
        return self

    async def __aexit__(self, *args):
        return await self.stream.__aexit__(*args)

    def __aiter__(self):
        return self

    async def __anext__(self):
        item = await self.stream.__anext__()
        if isinstance(item, Exception):
            self.errors.append(type(item).__name__)
            self.events.append('parse_error:' + type(item).__name__)
        else:
            message = item.message.model_dump(mode='json', by_alias=True, exclude_none=True)
            self.events.append({'message': message})
        return item

    async def aclose(self):
        await self.stream.aclose()

async def run_case(name, folder):
    condition, policy = name.rsplit('_', 1)
    mode = 'none' if condition == 'rpc_error' else condition
    error_code, answer, outcome = (None, None, 'startup_stream_rejected')
    handler = logging.FileHandler(folder / 'client-sdk.log', encoding='utf-8')
    handler.setFormatter(DiagnosticFormatter())
    logging.getLogger().addHandler(handler)
    try:
        parameters = StdioServerParameters(command=sys.executable, args=['-u', str(ROOT / 'relay.py'), mode, str(folder)], cwd=str(ROOT))
        with (folder / 'relay-stderr.log').open('w') as diagnostics:
            async with stdio_client(parameters, errlog=diagnostics) as (read, write):
                observer = ObserveRead(read)
                async with ClientSession(observer, write, read_timeout_seconds=timedelta(seconds=5)) as session:
                    initialized = await session.initialize()
                    if not initialized.protocolVersion == '2025-11-25':
                        raise AssertionError("Fixture invariant: initialized.protocolVersion == '2025-11-25'")
                    if not initialized.serverInfo.name == 'startup-stream-fixture':
                        raise AssertionError("Fixture invariant: initialized.serverInfo.name == 'startup-stream-fixture'")
                    (folder / 'startup-observation.json').write_text(json.dumps({'initialize_completed': True, 'events': observer.events.copy(), 'errors': observer.errors.copy()}, indent=2) + '\n')
                    if policy != 'gate' or not observer.errors:
                        if condition == 'rpc_error':
                            try:
                                await session.list_resources()
                            except McpError as exc:
                                error_code = exc.error.code
                            else:
                                raise AssertionError('Expected Method not found')
                        result = await session.call_tool('read_answer', {})
                        if not (result.isError is False and len(result.content) == 1):
                            raise AssertionError('Fixture invariant: result.isError is False and len(result.content) == 1')
                        answer = result.content[0].text
                        outcome = 'tool_returned'
    finally:
        logging.getLogger().removeHandler(handler)
        handler.close()
    sent = [json.loads(row) for row in (folder / 'client-stdin.bin').read_bytes().split(b'\n') if row]
    actual = {'errors': len(observer.errors), 'tool_calls': sum((row.get('method') == 'tools/call' for row in sent)), 'answer': answer, 'outcome': outcome, 'protocol_error': error_code}
    cleanup = json.loads((folder / 'relay-result.json').read_text())
    if not cleanup == {'server_returncode': 0, 'relay_errors': [], 'threads_finished': True}:
        raise AssertionError("Fixture invariant: cleanup == {'server_returncode': 0, 'relay_errors': [], 'threads_finished': True}")
    (folder / 'observed.json').write_text(json.dumps(actual, indent=2) + '\n')
    return actual

async def main():
    versions = {'python': platform.python_version(), **{x: version(x) for x in ['mcp', 'pydantic', 'anyio']}}
    if not versions == {'python': '3.12.14', 'mcp': '1.29.0', 'pydantic': '2.13.4', 'anyio': '4.14.2'}:
        raise AssertionError(versions)
    expected = json.loads((ROOT.parent / 'expected/cases.json').read_text())
    output = Path(sys.argv[1]).resolve()
    output.mkdir(exist_ok=False)
    results = {}
    for name, wanted in expected.items():
        folder = output / name
        folder.mkdir()
        with anyio.fail_after(15):
            actual = await run_case(name, folder)
        if not actual == wanted:
            raise AssertionError((name, actual, wanted))
        results[name] = actual
    (output / 'results.json').write_text(json.dumps({'versions': versions, 'cases': results}, indent=2) + '\n')
    print(json.dumps({'passed': len(results), 'sdk': '1.29.0', 'protocol': '2025-11-25'}, sort_keys=True))
if __name__ == '__main__':
    anyio.run(main)

{
  "none_observe":{"errors":0,"tool_calls":1,"answer":"7","outcome":"tool_returned","protocol_error":null},
  "none_gate":{"errors":0,"tool_calls":1,"answer":"7","outcome":"tool_returned","protocol_error":null},
  "stderr_observe":{"errors":0,"tool_calls":1,"answer":"7","outcome":"tool_returned","protocol_error":null},
  "stderr_gate":{"errors":0,"tool_calls":1,"answer":"7","outcome":"tool_returned","protocol_error":null},
  "text_observe":{"errors":1,"tool_calls":1,"answer":"7","outcome":"tool_returned","protocol_error":null},
  "text_gate":{"errors":1,"tool_calls":0,"answer":null,"outcome":"startup_stream_rejected","protocol_error":null},
  "json_observe":{"errors":1,"tool_calls":1,"answer":"7","outcome":"tool_returned","protocol_error":null},
  "json_gate":{"errors":1,"tool_calls":0,"answer":null,"outcome":"startup_stream_rejected","protocol_error":null},
  "rpc_error_observe":{"errors":0,"tool_calls":1,"answer":"7","outcome":"tool_returned","protocol_error":-32601},
  "rpc_error_gate":{"errors":0,"tool_calls":1,"answer":"7","outcome":"tool_returned","protocol_error":-32601}
}

{
  "versions": {
    "python": "3.12.14",
    "mcp": "1.29.0",
    "pydantic": "2.13.4",
    "anyio": "4.14.2"
  },
  "cases": {
    "none_observe": {
      "errors": 0,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    },
    "none_gate": {
      "errors": 0,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    },
    "stderr_observe": {
      "errors": 0,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    },
    "stderr_gate": {
      "errors": 0,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    },
    "text_observe": {
      "errors": 1,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    },
    "text_gate": {
      "errors": 1,
      "tool_calls": 0,
      "answer": null,
      "outcome": "startup_stream_rejected",
      "protocol_error": null
    },
    "json_observe": {
      "errors": 1,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    },
    "json_gate": {
      "errors": 1,
      "tool_calls": 0,
      "answer": null,
      "outcome": "startup_stream_rejected",
      "protocol_error": null
    },
    "rpc_error_observe": {
      "errors": 0,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": -32601
    },
    "rpc_error_gate": {
      "errors": 0,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": -32601
    }
  }
}

"""Two unwrapped ClientSession controls, independent of ObserveRead and gate."""
from datetime import timedelta
import json
import logging
from pathlib import Path
import sys
import anyio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from probe import DiagnosticFormatter

ROOT = Path(__file__).resolve().parent

def require(condition, label):
    if not condition:
        raise AssertionError(label)

async def main():
    output = Path(sys.argv[1]).resolve()
    output.mkdir(exist_ok=False)
    results = {}
    for mode in ('text', 'json'):
        folder = output / mode
        folder.mkdir()
        handler = logging.FileHandler(folder / 'client-sdk.log', encoding='utf-8')
        handler.setFormatter(DiagnosticFormatter())
        logging.getLogger().addHandler(handler)
        try:
            parameters = StdioServerParameters(command=sys.executable,
                args=['-u', str(ROOT / 'relay.py'), mode, str(folder)], cwd=str(ROOT))
            with (folder / 'relay-stderr.log').open('w') as diagnostics:
                async with stdio_client(parameters, errlog=diagnostics) as (read, write):
                    # No stream wrapper, custom handler, or error gate here.
                    async with ClientSession(read, write,
                            read_timeout_seconds=timedelta(seconds=5)) as session:
                        initialized = await session.initialize()
                        require(initialized.protocolVersion == '2025-11-25', 'Protocol')
                        result = await session.call_tool('read_answer', {})
                        require(result.isError is False, 'Tool isError')
                        require([item.text for item in result.content] == ['7'], 'Answer')
        finally:
            logging.getLogger().removeHandler(handler)
            handler.close()
        messages = [json.loads(line) for line in
                    (folder / 'client-stdin.bin').read_bytes().splitlines()]
        require(sum(row.get('method') == 'tools/call' for row in messages) == 1, 'Call count')
        require((folder / 'client-sdk.log').read_text().count(
            'Failed to parse JSONRPC message from server') == 1, 'SDK diagnostic')
        require(json.loads((folder / 'relay-result.json').read_text()) == {
            'server_returncode': 0, 'relay_errors': [], 'threads_finished': True}, 'Cleanup')
        results[mode] = {'initialized': True, 'answer': '7', 'tool_calls': 1,
                         'sdk_parse_diagnostics': 1}
    (output / 'results.json').write_text(json.dumps(results, indent=2) + '\n')
    print('PASS: 2 unwrapped default-session controls returned 7 with parser diagnostics')

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

{
  "text": {
    "initialized": true,
    "answer": "7",
    "tool_calls": 1,
    "sdk_parse_diagnostics": 1
  },
  "json": {
    "initialized": true,
    "answer": "7",
    "tool_calls": 1,
    "sdk_parse_diagnostics": 1
  }
}

{
  "initialize_completed": true,
  "events": [
    "parse_error:ValidationError",
    {
      "message": {
        "jsonrpc": "2.0",
        "id": 0,
        "result": {
          "protocolVersion": "2025-11-25",
          "capabilities": {
            "experimental": {},
            "tools": {
              "listChanged": false
            }
          },
          "serverInfo": {
            "name": "startup-stream-fixture",
            "version": "1"
          }
        }
      }
    }
  ],
  "errors": [
    "ValidationError"
  ]
}

{"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"mcp","version":"0.1.0"}},"jsonrpc":"2.0","id":0}
{"method":"notifications/initialized","jsonrpc":"2.0"}

"""Independent oracle: stdlib, captured bytes, and handwritten constants only."""
import json
from pathlib import Path
import sys


def require(condition, label):
    if not condition:
        raise AssertionError(label)


root = Path(sys.argv[1])
expected_names = {f'{mode}_{policy}'
    for mode in ('none', 'stderr', 'text', 'json', 'rpc_error')
    for policy in ('observe', 'gate')}
require({p.name for p in root.iterdir() if p.is_dir()} == expected_names, 'Case set')
findings = {}
for name in sorted(expected_names):
    condition, policy = name.rsplit('_', 1)
    folder = root / name
    client_raw = (folder / 'client-stdin.bin').read_bytes()
    server_raw = (folder / 'server-stdout.bin').read_bytes()
    require(client_raw.endswith(b'\n') and server_raw.endswith(b'\n'), 'Final LF')
    client = [json.loads(row) for row in client_raw[:-1].split(b'\n')]
    raw_lines = server_raw[:-1].split(b'\n')
    bad_kind = None
    if condition == 'text':
        require(raw_lines.pop(0) == b'FIXTURE READY', 'Exact text banner')
        bad_kind = 'invalid_json'
    elif condition == 'json':
        banner = raw_lines.pop(0)
        require(banner == b'{"level":"info","message":"fixture ready"}', 'JSON banner')
        require(json.loads(banner) == {'level': 'info', 'message': 'fixture ready'},
                'Valid JSON, without a JSON-RPC envelope')
        bad_kind = 'json_without_jsonrpc_envelope'
    server = [json.loads(row) for row in raw_lines]
    require(all(type(row) is dict and row.get('jsonrpc') == '2.0'
                for row in client + server), 'Fixture envelopes')
    requests = [row for row in client if 'id' in row]
    require(len({row['id'] for row in requests}) == len(requests), 'Distinct request IDs')
    require(len(server) == len(requests), 'Response count')
    require([row['id'] for row in server] == [row['id'] for row in requests], 'Response IDs')
    require(all(('result' in row) != ('error' in row) for row in server), 'Result XOR error')
    require(server[0]['result']['protocolVersion'] == '2025-11-25', 'Protocol')
    require(server[0]['result']['serverInfo'] == {
        'name': 'startup-stream-fixture', 'version': '1'}, 'Server identity')
    wanted_tools = 0 if condition in ('text', 'json') and policy == 'gate' else 1
    methods = ['initialize', 'notifications/initialized']
    if wanted_tools:
        if condition == 'rpc_error':
            methods.append('resources/list')
        methods.extend(['tools/call', 'tools/list'])
    require([row['method'] for row in client] == methods, 'Exact fixture method order')
    calls = [row for row in requests if row['method'] == 'tools/call']
    require(len(calls) == wanted_tools, 'Tool call count')
    if wanted_tools:
        require(calls[0]['params'] == {'name': 'read_answer', 'arguments': {}}, 'Call input')
        reply = next(row for row in server if row['id'] == calls[0]['id'])
        require(reply['result']['isError'] is False, 'Tool result status')
        require(reply['result']['content'] == [{'type': 'text', 'text': '7'}], 'Exact answer')
        listing = next(row for row in requests if row['method'] == 'tools/list')
        listed = next(row for row in server if row['id'] == listing['id'])
        require(listed['result'] == {'tools': [{'name': 'read_answer',
            'inputSchema': {'type': 'object', 'properties': {}}}]}, 'Tool schema')
    else:
        require(len(requests) == 1, 'No business request after gate rejection')
    rpc_errors = [row for row in server if 'error' in row]
    if condition == 'rpc_error':
        require(len(rpc_errors) == 1, 'One protocol error')
        request = next(row for row in requests if row['id'] == rpc_errors[0]['id'])
        require(request['method'] == 'resources/list', 'Protocol error request')
        require(rpc_errors[0]['error'] == {
            'code': -32601, 'message': 'Method not found'}, 'Protocol error content')
    else:
        require(rpc_errors == [], 'No JSON-RPC error response')
    server_stderr = (folder / 'server-stderr.bin').read_bytes()
    require(server_stderr == (b'FIXTURE READY\n' if condition == 'stderr' else b''),
            'Exact server stderr control')
    require((folder / 'relay-stderr.log').read_bytes() == server_stderr,
            'Relay stderr equals forwarded server stderr')
    sdk_log = (folder / 'client-sdk.log').read_text()
    require(sdk_log.count('Failed to parse JSONRPC message from server')
            == int(condition in ('text', 'json')), 'Separate SDK diagnostic count')
    startup = json.loads((folder / 'startup-observation.json').read_text())
    require(startup['initialize_completed'] is True, 'Initialize completed')
    require(startup['errors'] == (['ValidationError']
            if condition in ('text', 'json') else []), 'Observed startup exception')
    kinds = ['parse_error' if isinstance(row, str) else 'message'
             for row in startup['events']]
    require(kinds == (['parse_error', 'message']
            if condition in ('text', 'json') else ['message']), 'Error before init response')
    require(startup['events'][-1]['message'] == server[0], 'Same initialization response')
    cleanup = json.loads((folder / 'relay-result.json').read_text())
    require(cleanup == {'server_returncode': 0, 'relay_errors': [],
            'threads_finished': True}, 'Finite fixture cleanup')
    findings[name] = {'raw_contamination': bad_kind, 'tool_calls': wanted_tools,
                     'rpc_error_responses': len(rpc_errors), 'server_returncode': 0}
(root / 'independent-audit.json').write_text(json.dumps(findings, indent=2) + '\n')
print(json.dumps({'independent_raw_cases': len(findings), 'passed': True}, sort_keys=True))

{"passed": 10, "protocol": "2025-11-25", "sdk": "1.29.0"}

{"independent_raw_cases": 10, "passed": true}

"""Run the fixed matrix and a separate raw-byte oracle; compare full output."""
import json
from pathlib import Path
import subprocess
import sys

ROOT = Path(__file__).resolve().parent
output = Path(sys.argv[1]).resolve()
if output.exists():
    raise FileExistsError('Use a new output directory: ' + str(output))
flags = ['-B'] + (['-O'] if sys.flags.optimize else [])
receipts = {}
for name in ('probe', 'audit'):
    process = subprocess.run(
        [sys.executable, *flags, str(ROOT / (name + '.py')), str(output)],
        capture_output=True, timeout=180,
    )
    if output.is_dir():
        (output / (name + '.stdout.txt')).write_bytes(process.stdout)
        (output / (name + '.stderr.txt')).write_bytes(process.stderr)
    wanted = (ROOT.parent / 'expected' / (name + '.stdout.txt')).read_bytes()
    receipts[name] = {'returncode': process.returncode,
                      'stdout_exact': process.stdout == wanted,
                      'stderr_empty': process.stderr == b''}
    if output.is_dir():
        (output / 'command-results.json').write_text(json.dumps(receipts, indent=2) + '\n')
    if process.returncode != 0 or process.stdout != wanted or process.stderr != b'':
        raise AssertionError((name, receipts[name]))
    sys.stdout.buffer.write(process.stdout)
print('PASS: fixed matrix and independent raw oracle')

"""A passing baseline, then three wrong policies and a nonzero child-exit fault."""
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile

ROOT = Path(__file__).resolve().parent
output = Path(sys.argv[1]).resolve()
output.mkdir(exist_ok=False)
flags = ['-B'] + (['-O'] if sys.flags.optimize else [])

mutations = {
    'no_gate': ('probe.py',
        "if policy != 'gate' or not observer.errors:", 'if True:',
        'text_gate', {'errors': 1, 'tool_calls': 1, 'answer': '7',
         'outcome': 'tool_returned', 'protocol_error': None}),
    'json_syntax_only': ('probe.py',
        'self.errors.append(type(item).__name__)',
        "if any(error.get('type') == 'json_invalid' for error in item.errors()):\n                self.errors.append(type(item).__name__)",
        'json_observe', {'errors': 0, 'tool_calls': 1, 'answer': '7',
         'outcome': 'tool_returned', 'protocol_error': None}),
    'stderr_is_failure': ('probe.py',
        "    sent = [json.loads(row)",
        "    if (folder / 'server-stderr.bin').read_bytes():\n        observer.errors.append('StderrOutput')\n    sent = [json.loads(row)",
        'stderr_observe', {'errors': 1, 'tool_calls': 1, 'answer': '7',
         'outcome': 'tool_returned', 'protocol_error': None}),
    'child_exit_7': ('server.py',
        'asyncio.run(main())', 'asyncio.run(main())\n    sys.exit(7)',
        'none_observe', None),
}
records = {}
for name, change in [('baseline', None), *mutations.items()]:
    with tempfile.TemporaryDirectory(prefix='stdio-noise-mutant-') as temporary:
        project = Path(temporary)
        shutil.copytree(ROOT, project / 'source', ignore=shutil.ignore_patterns('__pycache__'))
        shutil.copytree(ROOT.parent / 'expected', project / 'expected')
        if change:
            filename, old, new, case, wanted = change
            target = project / 'source' / filename
            code = target.read_text()
            if code.count(old) != 1:
                raise AssertionError('Mutation anchor: ' + name)
            target.write_text(code.replace(old, new, 1))
        run = output / name
        process = subprocess.run([sys.executable, *flags,
            str(project / 'source/probe.py'), str(run)], capture_output=True, timeout=180)
        (output / (name + '.stdout.txt')).write_bytes(process.stdout)
        # Tracebacks contain the temporary copy path; retain them for local diagnosis.
        (output / (name + '.stderr.txt')).write_bytes(process.stderr)
        if change is None:
            if process.returncode != 0 or process.stderr != b'' or process.stdout != (
                    ROOT.parent / 'expected/probe.stdout.txt').read_bytes():
                raise AssertionError('Unmodified baseline failed')
            evidence = {'baseline_accepted': True}
        else:
            if process.returncode != 1 or b'AssertionError' not in process.stderr:
                raise AssertionError((name, 'Did not fail at an assertion', process.returncode))
            if name == 'child_exit_7':
                evidence = json.loads((run / case / 'relay-result.json').read_text())
                if evidence != {'server_returncode': 7, 'relay_errors': [], 'threads_finished': True}:
                    raise AssertionError((name, evidence))
                if b'cleanup ==' not in process.stderr:
                    raise AssertionError('Wrong child-exit rejection')
            else:
                evidence = json.loads((run / case / 'observed.json').read_text())
                if evidence != wanted:
                    raise AssertionError((name, evidence, wanted))
                if case.encode() not in process.stderr:
                    raise AssertionError('Wrong mismatch case')
        records[name] = {'returncode': process.returncode, 'evidence': evidence}
(output / 'mutants.json').write_text(json.dumps(records, indent=2) + '\n')
print('PASS: baseline accepted; 3 policy mutants and child exit 7 rejected')

{
  "baseline": {
    "returncode": 0,
    "evidence": {
      "baseline_accepted": true
    }
  },
  "no_gate": {
    "returncode": 1,
    "evidence": {
      "errors": 1,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    }
  },
  "json_syntax_only": {
    "returncode": 1,
    "evidence": {
      "errors": 0,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    }
  },
  "stderr_is_failure": {
    "returncode": 1,
    "evidence": {
      "errors": 1,
      "tool_calls": 1,
      "answer": "7",
      "outcome": "tool_returned",
      "protocol_error": null
    }
  },
  "child_exit_7": {
    "returncode": 1,
    "evidence": {
      "server_returncode": 7,
      "relay_errors": [],
      "threads_finished": true
    }
  }
}

"""Keep the failed startup evidence; move the banner to stderr in a new process."""
import hashlib
import json
from pathlib import Path
import sys
import anyio
from probe import run_case

def hashes(folder):
    return {p.name: hashlib.sha256(p.read_bytes()).hexdigest()
            for p in sorted(folder.iterdir()) if p.is_file()}

async def main():
    output = Path(sys.argv[1]).resolve()
    output.mkdir(exist_ok=False)
    rejected = output / 'before'
    rejected.mkdir()
    before = await run_case('text_gate', rejected)
    if before != {'errors': 1, 'tool_calls': 0, 'answer': None,
                  'outcome': 'startup_stream_rejected', 'protocol_error': None}:
        raise AssertionError(before)
    preserved = hashes(rejected)
    repaired = output / 'after'
    repaired.mkdir()
    after = await run_case('stderr_gate', repaired)
    if after != {'errors': 0, 'tool_calls': 1, 'answer': '7',
                 'outcome': 'tool_returned', 'protocol_error': None}:
        raise AssertionError(after)
    if hashes(rejected) != preserved:
        raise AssertionError('Old evidence changed')
    (output / 'recovery.json').write_text(json.dumps({
        'before': before, 'after': after, 'old_evidence_unchanged': True,
        'old_sha256': preserved}, indent=2) + '\n')
    print('PASS: stdout gate rejected; fresh stderr control returned 7; old evidence unchanged')

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

{
  "before": {
    "errors": 1,
    "tool_calls": 0,
    "answer": null,
    "outcome": "startup_stream_rejected",
    "protocol_error": null
  },
  "after": {
    "errors": 0,
    "tool_calls": 1,
    "answer": "7",
    "outcome": "tool_returned",
    "protocol_error": null
  },
  "old_evidence_unchanged": true,
  "old_sha256": {
    "client-sdk.log": "543874857e68583defdf91017689ebf5a0b1776dc4bb828c24a5340986567efe",
    "client-stdin.bin": "744e5f41136d80c686e800077d5e37bd25ab5822607dd532a116f829a4c20dbd",
    "observed.json": "6045f4d4782648e9932b7e2ec5a1a3aec0b9c4c99ca98c7573416bc6fabee016",
    "relay-result.json": "3f5c338ac454f6dd1b9e2517ce9b1e3f5bc9c4b0d3aa92706262ad06b42b99d0",
    "relay-stderr.log": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
    "server-stderr.bin": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
    "server-stdout.bin": "f9b5ec09d4de5f4614b83cbc9b8ed9d62d6aa929e76517294105520c0fdd80e9",
    "startup-observation.json": "6e723d9cf3cb29247d67acce421c44ac05b70aa4473c94aa37d37debff856a7b"
  }
}

mcp==1.29.0
pydantic==2.13.4
anyio==4.14.2

"""Fresh-output, fixed-stdout verification of all supplement experiments."""
import json
from pathlib import Path
import subprocess
import sys

ROOT = Path(__file__).resolve().parent
output = Path(sys.argv[1]).resolve()
output.mkdir(exist_ok=False)
flags = ['-B'] + (['-O'] if sys.flags.optimize else [])
commands = (
    ('check_run', 'matrix'),
    ('direct_baseline', 'direct'),
    ('recovery', 'recovery'),
    ('run_mutants', 'mutants'),
)
records = {}
for script, folder in commands:
    process = subprocess.run([sys.executable, *flags, str(ROOT / (script + '.py')),
        str(output / folder)], capture_output=True, timeout=600)
    (output / (script + '.stdout.txt')).write_bytes(process.stdout)
    (output / (script + '.stderr.txt')).write_bytes(process.stderr)
    expected = (ROOT.parent / 'expected' / (script + '.stdout.txt')).read_bytes()
    record = {'returncode': process.returncode,
              'stdout_exact': process.stdout == expected,
              'stderr_empty': process.stderr == b''}
    records[script] = record
    (output / 'verification.json').write_text(json.dumps(records, indent=2) + '\n')
    if record != {'returncode': 0, 'stdout_exact': True, 'stderr_empty': True}:
        raise AssertionError((script, record))
    sys.stdout.buffer.write(process.stdout)
print('PASS: all fixed supplement checks')

{"passed": 10, "protocol": "2025-11-25", "sdk": "1.29.0"}
{"independent_raw_cases": 10, "passed": true}
PASS: fixed matrix and independent raw oracle

PASS: 2 unwrapped default-session controls returned 7 with parser diagnostics

PASS: stdout gate rejected; fresh stderr control returned 7; old evidence unchanged

PASS: baseline accepted; 3 policy mutants and child exit 7 rejected

{"passed": 10, "protocol": "2025-11-25", "sdk": "1.29.0"}
{"independent_raw_cases": 10, "passed": true}
PASS: fixed matrix and independent raw oracle
PASS: 2 unwrapped default-session controls returned 7 with parser diagnostics
PASS: stdout gate rejected; fresh stderr control returned 7; old evidence unchanged
PASS: baseline accepted; 3 policy mutants and child exit 7 rejected
PASS: all fixed supplement checks

{
  "check_run": {
    "returncode": 0,
    "stdout_exact": true,
    "stderr_empty": true
  },
  "direct_baseline": {
    "returncode": 0,
    "stdout_exact": true,
    "stderr_empty": true
  },
  "recovery": {
    "returncode": 0,
    "stdout_exact": true,
    "stderr_empty": true
  },
  "run_mutants": {
    "returncode": 0,
    "stdout_exact": true,
    "stderr_empty": true
  }
}