Agent 执行多步计划前，先检查依赖图

"""One fixed graph, synthetic completion events, no workers or external effects."""
from graphlib import TopologicalSorter
import json
import platform


GRAPH = {
    "read_a": (),
    "read_b": (),
    "extract_a": ("read_a",),
    "extract_b": ("read_b",),
    "compose": ("extract_a", "extract_b"),
}


def fresh():
    sorter = TopologicalSorter(GRAPH)
    sorter.prepare()
    assert sorted(sorter.get_ready()) == ["read_a", "read_b"]
    return sorter


def native_case(nodes):
    sorter = fresh()
    error = None
    try:
        sorter.done(*nodes)
    except ValueError as exc:
        error = type(exc).__name__
    # get_ready is a mutating issuance operation, not a passive observation.
    released = sorted(sorter.get_ready())
    return {
        "batch": list(nodes),
        "error": error,
        "released": released,
        "active": sorter.is_active(),
    }


class BatchRejected(ValueError):
    """Expected input rejection, before the first call to sorter.done."""


class CoordinatorStopped(RuntimeError):
    """Unexpected sorter failure; no rollback or automatic replay promised."""


class Coordinator:
    """Single-owner teaching ledger for this fixed graph; not a scheduler."""

    def __init__(self):
        self._sorter = TopologicalSorter(GRAPH)
        self._sorter.prepare()
        self.issued = set()
        self.completed = set()
        self.done_calls = 0
        self.stopped = False

    def take_ready(self):
        if self.stopped:
            raise CoordinatorStopped("coordinator stopped")
        ready = tuple(sorted(self._sorter.get_ready()))
        self.issued.update(ready)
        return ready

    def complete_batch(self, events):
        if self.stopped:
            raise CoordinatorStopped("coordinator stopped")
        if type(events) is not list or not 1 <= len(events) <= len(GRAPH):
            raise BatchRejected("expected a nonempty bounded event list")
        # Copy validated node strings into an immutable tuple before mutation.
        # No await, callbacks, worker threads, or independent sorter owners here.
        nodes = []
        seen = set()
        for event in events:
            if type(event) is not dict or set(event) != {"node", "success", "error"}:
                raise BatchRejected("event fields")
            node = event["node"]
            if type(node) is not str:
                raise BatchRejected("node type")
            if node in seen:
                raise BatchRejected("duplicate within batch")
            if node not in self.issued or node in self.completed:
                raise BatchRejected("not an outstanding issued node")
            if event["success"] is not True or event["error"] is not None:
                raise BatchRejected("not a successful completion")
            seen.add(node)
            nodes.append(node)
        batch = tuple(nodes)
        self.done_calls += 1
        try:
            self._sorter.done(*batch)
        except Exception as exc:
            # A bug, unexpected runtime fault, or illicit concurrent mutation
            # might leave partial state. Stop rather than claiming rollback.
            self.stopped = True
            raise CoordinatorStopped("unexpected done failure; inspect state") from exc
        self.completed.update(batch)
        return batch


def event(node, success=True, error=None):
    return {"node": node, "success": success, "error": error}


def guarded_case(events):
    coordinator = Coordinator()
    first = list(coordinator.take_ready())
    error = None
    try:
        coordinator.complete_batch(events)
    except BatchRejected as exc:
        error = type(exc).__name__
    rejected_state = {
        "error": error,
        "done_calls": coordinator.done_calls,
        "completed": sorted(coordinator.completed),
        "released": list(coordinator.take_ready()),
    }
    # These are corrected synthetic events for already-issued roots, not a
    # request to rerun completed workers or blindly replay an uncertain batch.
    if error:
        coordinator.complete_batch([event("read_a"), event("read_b")])
    return {
        "initial_ready": first,
        "after_attempt": rejected_state,
        "after_corrected_batch": {
            "done_calls": coordinator.done_calls,
            "completed": sorted(coordinator.completed),
            "released": list(coordinator.take_ready()),
        },
    }


def replay_case():
    sorter = fresh()
    batch = ("read_a", "read_a", "read_b")
    errors = []
    for _ in range(2):
        try:
            sorter.done(*batch)
        except ValueError as exc:
            errors.append(type(exc).__name__)
    first_release = sorted(sorter.get_ready())
    # Only read_b remains outstanding among the two roots.
    sorter.done("read_b")
    second_release = sorted(sorter.get_ready())
    return {"errors": errors, "after_replay": first_release,
            "after_explicit_read_b_completion": second_release}


def run():
    native_inputs = {
        "valid_roots": ("read_a", "read_b"),
        "duplicate_later": ("read_a", "read_a", "read_b"),
        "unknown_later": ("read_a", "missing", "read_b"),
        "unknown_first": ("missing", "read_a", "read_b"),
        "newly_ready_but_unissued": ("read_a", "extract_a"),
        "empty_native_batch": (),
    }
    guarded_inputs = {
        "duplicate_later": [event("read_a"), event("read_a"), event("read_b")],
        "unknown_later": [event("read_a"), event("missing"), event("read_b")],
        "unknown_first": [event("missing"), event("read_a"), event("read_b")],
        "unissued_successor": [event("read_a"), event("extract_a")],
        "failed_later": [event("read_a"), event("read_b", False, "synthetic failure")],
        "integer_success_later": [event("read_a"), event("read_b", 1)],
    }
    return {
        "environment": {"python": platform.python_version(), "implementation": platform.python_implementation()},
        "graph": GRAPH,
        "guarded_inputs": guarded_inputs,
        "native": {name: native_case(nodes) for name, nodes in native_inputs.items()},
        "replay": replay_case(),
        "guarded": {name: guarded_case(events) for name, events in guarded_inputs.items()},
        "scope": "fixed graph, single owner, synthetic events; no real tools, model, threads, network, persistence, or crash-atomicity test",
    }


if __name__ == "__main__":
    print(json.dumps(run(), ensure_ascii=False, indent=2, sort_keys=True))

import unittest
from unittest.mock import patch

from batch_completion_lab import (BatchRejected, Coordinator, CoordinatorStopped,
                   event, guarded_case, native_case, replay_case, run)


class BatchCompletionTests(unittest.TestCase):
    def test_valid_native_batch_releases_both_branches(self):
        self.assertEqual(native_case(("read_a", "read_b")), {
            "batch": ["read_a", "read_b"], "error": None,
            "released": ["extract_a", "extract_b"], "active": True})

    def test_invalid_later_mutates_valid_prefix(self):
        for nodes in [("read_a", "read_a", "read_b"),
                      ("read_a", "missing", "read_b"),
                      ("read_a", "extract_a")]:
            with self.subTest(nodes=nodes):
                result = native_case(nodes)
                self.assertEqual(result["error"], "ValueError")
                self.assertEqual(result["released"], ["extract_a"])
                self.assertTrue(result["active"])

    def test_invalid_first_does_not_release_any_successor(self):
        result = native_case(("missing", "read_a", "read_b"))
        self.assertEqual(result["error"], "ValueError")
        self.assertEqual(result["released"], [])

    def test_empty_native_batch_does_not_complete_roots(self):
        result = native_case(())
        self.assertIsNone(result["error"])
        self.assertEqual(result["released"], [])
        self.assertTrue(result["active"])

    def test_replaying_native_failed_batch_does_not_finish_read_b(self):
        self.assertEqual(replay_case(), {
            "errors": ["ValueError", "ValueError"],
            "after_replay": ["extract_a"],
            "after_explicit_read_b_completion": ["extract_b"]})

    def test_all_guarded_rejections_make_zero_done_calls(self):
        for name, result in run()["guarded"].items():
            with self.subTest(name=name):
                self.assertEqual(result["initial_ready"], ["read_a", "read_b"])
                self.assertEqual(result["after_attempt"], {
                    "error": "BatchRejected", "done_calls": 0,
                    "completed": [], "released": []})
                self.assertEqual(result["after_corrected_batch"], {
                    "done_calls": 1, "completed": ["read_a", "read_b"],
                    "released": ["extract_a", "extract_b"]})

    def test_success_marker_alone_with_error_is_rejected(self):
        result = guarded_case([event("read_a"), event("read_b", True, "failure")])
        self.assertEqual(result["after_attempt"]["done_calls"], 0)

    def test_batch_shape_and_types_reject_before_mutation(self):
        invalid = [(), [], [event("read_a")] * 6, [None],
                   [{"node": "read_a", "success": True}],
                   [event(1)], [event("read_a"), event("read_b", "true")]]
        for events in invalid:
            with self.subTest(events=events):
                c = Coordinator()
                c.take_ready()
                with self.assertRaises(BatchRejected):
                    c.complete_batch(events)
                self.assertEqual(c.done_calls, 0)
                self.assertEqual(c.completed, set())

    def test_complete_before_issuance_is_rejected(self):
        c = Coordinator()
        with self.assertRaises(BatchRejected):
            c.complete_batch([event("read_a")])
        self.assertEqual(c.done_calls, 0)
        self.assertEqual(c.take_ready(), ("read_a", "read_b"))

    def test_already_completed_event_does_not_complete_new_root(self):
        c = Coordinator()
        c.take_ready()
        c.complete_batch([event("read_a")])
        with self.assertRaises(BatchRejected):
            c.complete_batch([event("read_b"), event("read_a")])
        self.assertEqual(c.done_calls, 1)
        self.assertEqual(c.completed, {"read_a"})
        self.assertEqual(c.take_ready(), ("extract_a",))

    def test_valid_guarded_progress_finishes_fixed_graph(self):
        c = Coordinator()
        self.assertEqual(c.take_ready(), ("read_a", "read_b"))
        c.complete_batch([event("read_a"), event("read_b")])
        self.assertEqual(c.take_ready(), ("extract_a", "extract_b"))
        c.complete_batch([event("extract_a"), event("extract_b")])
        self.assertEqual(c.take_ready(), ("compose",))
        c.complete_batch([event("compose")])
        self.assertEqual(c.take_ready(), ())
        self.assertFalse(c._sorter.is_active())
        self.assertEqual(len(c.completed), 5)

    def test_unexpected_mutating_failure_stops_without_rollback_claim(self):
        c = Coordinator()
        c.take_ready()
        original = c._sorter.done

        def partial_then_fail(*nodes):
            original(nodes[0])
            raise RuntimeError("injected after first mutation")

        with patch.object(c._sorter, "done", side_effect=partial_then_fail):
            with self.assertRaises(CoordinatorStopped):
                c.complete_batch([event("read_a"), event("read_b")])
        self.assertTrue(c.stopped)
        self.assertEqual(c.completed, set())
        with self.assertRaises(CoordinatorStopped):
            c.take_ready()
        with self.assertRaises(CoordinatorStopped):
            c.complete_batch([event("read_b")])
        # Test-only direct inspection proves graph mutation was NOT rolled back.
        self.assertEqual(c._sorter.get_ready(), ("extract_a",))


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

"""Execute intentionally wrong copies in memory; source files stay unchanged."""
from pathlib import Path
from hashlib import sha256
from types import ModuleType
from unittest.mock import patch
import json

SOURCE = Path(__file__).with_name('batch_completion_lab.py').read_text()
SUCCESS = 'if event["success"] is not True or event["error"] is not None:'
OUTSTANDING = 'if node not in self.issued or node in self.completed:'


def assert_equal(actual, expected, label):
    if actual != expected:
        raise AssertionError(f'{label}: expected {expected!r}, got {actual!r}')


def rejected_without_mutation(m, kind):
    c = m.Coordinator()
    c.take_ready()
    inputs = {
        'duplicate': [m.event('read_a'), m.event('read_a'), m.event('read_b')],
        'unknown': [m.event('read_a'), m.event('missing'), m.event('read_b')],
        'integer_success': [m.event('read_a'), m.event('read_b', 1)],
        'error_with_success': [m.event('read_a'), m.event('read_b', True, 'failure')],
    }
    if kind == 'already_completed':
        c.complete_batch([m.event('read_a')])
        assert_equal(c.take_ready(), ('extract_a',), 'first branch')
        events = [m.event('read_b'), m.event('read_a')]
    else:
        events = inputs[kind]
    before_calls = c.done_calls
    before_completed = set(c.completed)
    error_type = None
    try:
        c.complete_batch(events)
    except (m.BatchRejected, m.CoordinatorStopped) as exc:
        error_type = type(exc).__name__
    assert_equal(error_type, 'BatchRejected', 'input error class')
    assert_equal(c.done_calls, before_calls, 'native call count')
    assert_equal(c.completed, before_completed, 'completion ledger')
    assert_equal(c.stopped, False, 'preflight rejection remains usable')
    assert_equal(c.take_ready(), (), 'no successor released')


def stops_after_partial_failure(m):
    c = m.Coordinator()
    c.take_ready()
    native_done = c._sorter.done

    def partial_then_fail(*nodes):
        native_done(nodes[0])
        raise RuntimeError('injected after first mutation')

    error_type = None
    with patch.object(c._sorter, 'done', side_effect=partial_then_fail):
        try:
            c.complete_batch([m.event('read_a'), m.event('read_b')])
        except m.CoordinatorStopped as exc:
            error_type = type(exc).__name__
    assert_equal(error_type, 'CoordinatorStopped', 'fault class')
    assert_equal(c.stopped, True, 'stopped flag')
    assert_equal(c.completed, set(), 'ledger not proof of rollback')
    for name, operation in [
        ('take_ready', c.take_ready),
        ('complete_batch', lambda: c.complete_batch([m.event('read_b')]))]:
        stopped = False
        try:
            operation()
        except m.CoordinatorStopped:
            stopped = True
        assert_equal(stopped, True, name + ' blocked')
    # Deliberately bypass the coordinator only to expose the injected mutation.
    assert_equal(c._sorter.get_ready(), ('extract_a',), 'no rollback')


MUTATIONS = [
    ('drop_duplicate_check', 'if node in seen:', 'if False:', 'duplicate'),
    ('drop_outstanding_check', OUTSTANDING, 'if False:', 'unknown'),
    ('forget_completed_check', OUTSTANDING, 'if node not in self.issued:', 'already_completed'),
    ('accept_truthy_success', SUCCESS, 'if not event["success"] or event["error"] is not None:', 'integer_success'),
    ('ignore_error_field', SUCCESS, 'if event["success"] is not True:', 'error_with_success'),
    ('mutate_during_preflight', '            nodes.append(node)', '            nodes.append(node)\n            self._sorter.done(node)', 'unknown'),
    ('omit_stopped_flag', '            self.stopped = True', '            self.stopped = False', 'fault'),
    ('allow_readiness_after_fault', '        if self.stopped:\n            raise CoordinatorStopped("coordinator stopped")\n        ready =', '        ready =', 'fault'),
]


def module_from(source, name):
    module = ModuleType(name)
    exec(compile(source, '<' + name + '>', 'exec'), module.__dict__)
    return module


def check(module, kind):
    if kind == 'fault':
        stops_after_partial_failure(module)
    else:
        rejected_without_mutation(module, kind)


def run():
    baseline = module_from(SOURCE, 'baseline')
    for kind in ('duplicate', 'unknown', 'already_completed', 'integer_success', 'error_with_success', 'fault'):
        check(baseline, kind)
    results = []
    for name, old, new, kind in MUTATIONS:
        if SOURCE.count(old) != 1:
            raise RuntimeError('mutation anchor not unique: ' + name)
        mutated = SOURCE.replace(old, new)
        module = module_from(mutated, name)
        # Only an oracle AssertionError counts. Syntax/import/runtime errors fail
        # this script rather than being mislabeled a detected behavior change.
        try:
            check(module, kind)
        except AssertionError as exc:
            results.append({'name': name, 'oracle': kind, 'detected': True,
                            'assertion': str(exc),
                            'source_sha256': sha256(mutated.encode()).hexdigest()})
        else:
            raise AssertionError('mutant survived: ' + name)
    return {'baseline_oracles_passed': 6, 'mutants_executed': len(results),
            'mutants_detected_by_assertion': len(results), 'results': results}


if __name__ == '__main__':
    print(json.dumps(run(), ensure_ascii=False, indent=2, sort_keys=True))

{
  "environment": {
    "implementation": "CPython",
    "python": "3.12.14"
  },
  "graph": {
    "compose": [
      "extract_a",
      "extract_b"
    ],
    "extract_a": [
      "read_a"
    ],
    "extract_b": [
      "read_b"
    ],
    "read_a": [],
    "read_b": []
  },
  "guarded": {
    "duplicate_later": {
      "after_attempt": {
        "completed": [],
        "done_calls": 0,
        "error": "BatchRejected",
        "released": []
      },
      "after_corrected_batch": {
        "completed": [
          "read_a",
          "read_b"
        ],
        "done_calls": 1,
        "released": [
          "extract_a",
          "extract_b"
        ]
      },
      "initial_ready": [
        "read_a",
        "read_b"
      ]
    },
    "failed_later": {
      "after_attempt": {
        "completed": [],
        "done_calls": 0,
        "error": "BatchRejected",
        "released": []
      },
      "after_corrected_batch": {
        "completed": [
          "read_a",
          "read_b"
        ],
        "done_calls": 1,
        "released": [
          "extract_a",
          "extract_b"
        ]
      },
      "initial_ready": [
        "read_a",
        "read_b"
      ]
    },
    "integer_success_later": {
      "after_attempt": {
        "completed": [],
        "done_calls": 0,
        "error": "BatchRejected",
        "released": []
      },
      "after_corrected_batch": {
        "completed": [
          "read_a",
          "read_b"
        ],
        "done_calls": 1,
        "released": [
          "extract_a",
          "extract_b"
        ]
      },
      "initial_ready": [
        "read_a",
        "read_b"
      ]
    },
    "unissued_successor": {
      "after_attempt": {
        "completed": [],
        "done_calls": 0,
        "error": "BatchRejected",
        "released": []
      },
      "after_corrected_batch": {
        "completed": [
          "read_a",
          "read_b"
        ],
        "done_calls": 1,
        "released": [
          "extract_a",
          "extract_b"
        ]
      },
      "initial_ready": [
        "read_a",
        "read_b"
      ]
    },
    "unknown_first": {
      "after_attempt": {
        "completed": [],
        "done_calls": 0,
        "error": "BatchRejected",
        "released": []
      },
      "after_corrected_batch": {
        "completed": [
          "read_a",
          "read_b"
        ],
        "done_calls": 1,
        "released": [
          "extract_a",
          "extract_b"
        ]
      },
      "initial_ready": [
        "read_a",
        "read_b"
      ]
    },
    "unknown_later": {
      "after_attempt": {
        "completed": [],
        "done_calls": 0,
        "error": "BatchRejected",
        "released": []
      },
      "after_corrected_batch": {
        "completed": [
          "read_a",
          "read_b"
        ],
        "done_calls": 1,
        "released": [
          "extract_a",
          "extract_b"
        ]
      },
      "initial_ready": [
        "read_a",
        "read_b"
      ]
    }
  },
  "guarded_inputs": {
    "duplicate_later": [
      {
        "error": null,
        "node": "read_a",
        "success": true
      },
      {
        "error": null,
        "node": "read_a",
        "success": true
      },
      {
        "error": null,
        "node": "read_b",
        "success": true
      }
    ],
    "failed_later": [
      {
        "error": null,
        "node": "read_a",
        "success": true
      },
      {
        "error": "synthetic failure",
        "node": "read_b",
        "success": false
      }
    ],
    "integer_success_later": [
      {
        "error": null,
        "node": "read_a",
        "success": true
      },
      {
        "error": null,
        "node": "read_b",
        "success": 1
      }
    ],
    "unissued_successor": [
      {
        "error": null,
        "node": "read_a",
        "success": true
      },
      {
        "error": null,
        "node": "extract_a",
        "success": true
      }
    ],
    "unknown_first": [
      {
        "error": null,
        "node": "missing",
        "success": true
      },
      {
        "error": null,
        "node": "read_a",
        "success": true
      },
      {
        "error": null,
        "node": "read_b",
        "success": true
      }
    ],
    "unknown_later": [
      {
        "error": null,
        "node": "read_a",
        "success": true
      },
      {
        "error": null,
        "node": "missing",
        "success": true
      },
      {
        "error": null,
        "node": "read_b",
        "success": true
      }
    ]
  },
  "native": {
    "duplicate_later": {
      "active": true,
      "batch": [
        "read_a",
        "read_a",
        "read_b"
      ],
      "error": "ValueError",
      "released": [
        "extract_a"
      ]
    },
    "empty_native_batch": {
      "active": true,
      "batch": [],
      "error": null,
      "released": []
    },
    "newly_ready_but_unissued": {
      "active": true,
      "batch": [
        "read_a",
        "extract_a"
      ],
      "error": "ValueError",
      "released": [
        "extract_a"
      ]
    },
    "unknown_first": {
      "active": true,
      "batch": [
        "missing",
        "read_a",
        "read_b"
      ],
      "error": "ValueError",
      "released": []
    },
    "unknown_later": {
      "active": true,
      "batch": [
        "read_a",
        "missing",
        "read_b"
      ],
      "error": "ValueError",
      "released": [
        "extract_a"
      ]
    },
    "valid_roots": {
      "active": true,
      "batch": [
        "read_a",
        "read_b"
      ],
      "error": null,
      "released": [
        "extract_a",
        "extract_b"
      ]
    }
  },
  "replay": {
    "after_explicit_read_b_completion": [
      "extract_b"
    ],
    "after_replay": [
      "extract_a"
    ],
    "errors": [
      "ValueError",
      "ValueError"
    ]
  },
  "scope": "fixed graph, single owner, synthetic events; no real tools, model, threads, network, persistence, or crash-atomicity test"
}

{
  "runtime": {
    "implementation": "CPython",
    "python": "3.12.14"
  },
  "first_sha256": "4d1d619f7932fcd6d8038360f1b8b9f329929a836fbca027270ce104198c24a6",
  "second_sha256": "4d1d619f7932fcd6d8038360f1b8b9f329929a836fbca027270ce104198c24a6",
  "byte_identical": true
}

{
  "baseline_oracles_passed": 6,
  "mutants_detected_by_assertion": 8,
  "mutants_executed": 8,
  "results": [
    {
      "assertion": "input error class: expected 'BatchRejected', got 'CoordinatorStopped'",
      "detected": true,
      "name": "drop_duplicate_check",
      "oracle": "duplicate",
      "source_sha256": "8c3384a8ba8ac1a8eb07805fc8a6204c5d93a37b76f0bc80583e9501138153a5"
    },
    {
      "assertion": "input error class: expected 'BatchRejected', got 'CoordinatorStopped'",
      "detected": true,
      "name": "drop_outstanding_check",
      "oracle": "unknown",
      "source_sha256": "618471de28c918107eae540feab9164d65cfc536a44d86397e485b535a2008d6"
    },
    {
      "assertion": "input error class: expected 'BatchRejected', got 'CoordinatorStopped'",
      "detected": true,
      "name": "forget_completed_check",
      "oracle": "already_completed",
      "source_sha256": "3c5f7d440def5ed5dcd0395380908c2bd4764c71997769c5e8f6223c7c0ed234"
    },
    {
      "assertion": "input error class: expected 'BatchRejected', got None",
      "detected": true,
      "name": "accept_truthy_success",
      "oracle": "integer_success",
      "source_sha256": "d417a24a50a02df46455047958875d0db8d241f6bd3df3cb77d2baecd3ec0558"
    },
    {
      "assertion": "input error class: expected 'BatchRejected', got None",
      "detected": true,
      "name": "ignore_error_field",
      "oracle": "error_with_success",
      "source_sha256": "99533b6329907943632439678a17526369ca1f070590ecf5aabf45e8a36199c7"
    },
    {
      "assertion": "no successor released: expected (), got ('extract_a',)",
      "detected": true,
      "name": "mutate_during_preflight",
      "oracle": "unknown",
      "source_sha256": "62025063512134fbe7df476a35dc409e86e9d63f647cc184d63ebf1ee0f2443b"
    },
    {
      "assertion": "stopped flag: expected True, got False",
      "detected": true,
      "name": "omit_stopped_flag",
      "oracle": "fault",
      "source_sha256": "18fd58129ad7925946ec639ba329d5aa370600bdad0dcb0310ad68ceee7cd7c0"
    },
    {
      "assertion": "take_ready blocked: expected True, got False",
      "detected": true,
      "name": "allow_readiness_after_fault",
      "oracle": "fault",
      "source_sha256": "ff015d4856765389ccb799e9391ad9e9699a1a51ceab382f422674994c61072b"
    }
  ]
}

graphlib 批量完成回执：可独立复现的补充实验

实测：CPython 3.12.14，仅标准库，无需安装依赖。
在解压后的同一目录运行：
  python3 -B batch_completion_lab.py
  python3 -B test_batch_completion_lab.py
  python3 -B mutation_check.py
原文代码也可单独检查：
  cd original-body
  python3 -B test_plan_lab.py

batch_completion_lab.py：固定五节点图、原生批量调用、失败批次重试与整批预检。
test_batch_completion_lab.py：12 个测试方法；包含多个子例，不是覆盖率或生产验证。
mutation_check.py：在内存中执行八份故意改错的源码；每份都必须触发指定行为断言。
observed/results.json 与 repeat-results.json：两次独立进程的完整实际输出，逐字节相同。
mutations.json：八份错误源码的哈希、检查名和实际断言结果。
repeat-verification.json：两份输出的字节哈希。
original-body/：从原文代码块提取的 plan_lab.py 和 test_plan_lab.py，25 项原测试通过。
official-sources.json：文档与版本固定源码链接。在线文档版本不等于执行版本。

get_ready() 会发出节点并改变状态。所有“released”都记录该次实际调用的返回值。
预检只是同一所有者、固定图、合成内存事件下的输入检查，不提供事务、并发安全或回滚。
异常保护仅处理原生 done 调用抛出的 Exception；测试中的强制检查底层图只用于证明没有回滚。
不运行图中的任务，不调用模型、网络或外部工具，不实现持久化、重启恢复或真实回执认证。