diff --git a/.github/workflows/capu-atman-http-recovery-v2.yml b/.github/workflows/capu-atman-http-recovery-v2.yml new file mode 100644 index 0000000..3fa009e --- /dev/null +++ b/.github/workflows/capu-atman-http-recovery-v2.yml @@ -0,0 +1,35 @@ +name: CaPU ATMAN loopback HTTP recovery +on: + pull_request: + paths: + - 'experiments/capu_atman_recovery_v1/**' + - 'experiments/capu_atman_http_recovery_v2/**' + - '.github/workflows/capu-atman-http-recovery-v2.yml' + workflow_dispatch: +permissions: + contents: read +jobs: + bounded-http: + runs-on: ubuntu-latest + timeout-minutes: 8 + strategy: + fail-fast: false + matrix: + python: ['3.11', '3.13'] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + - run: python -m pip install -r experiments/capu_atman_recovery_v1/requirements.txt + - run: python experiments/capu_atman_recovery_v1/bootstrap.py + - run: python experiments/capu_atman_recovery_v1/run.py --output evidence-v1 + - run: python experiments/capu_atman_http_recovery_v2/run.py --output evidence-v2 + - uses: actions/upload-artifact@v4 + if: always() + with: + name: capu-atman-http-python-${{ matrix.python }} + path: | + evidence-v1/ + evidence-v2/ + if-no-files-found: error diff --git a/.github/workflows/capu-atman-recovery-v1.yml b/.github/workflows/capu-atman-recovery-v1.yml new file mode 100644 index 0000000..15f0849 --- /dev/null +++ b/.github/workflows/capu-atman-recovery-v1.yml @@ -0,0 +1,34 @@ +name: CaPU ATMAN recovery laboratory +on: + pull_request: + paths: + - 'experiments/capu_atman_recovery_v1/**' + - '.github/workflows/capu-atman-recovery-v1.yml' + workflow_dispatch: +permissions: + contents: read +jobs: + bounded-composition: + runs-on: ubuntu-latest + timeout-minutes: 8 + strategy: + fail-fast: false + matrix: + python: ['3.11', '3.13'] + defaults: + run: + working-directory: experiments/capu_atman_recovery_v1 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + - run: python -m pip install -r requirements.txt + - run: python bootstrap.py + - run: python run.py --output evidence + - uses: actions/upload-artifact@v4 + if: always() + with: + name: capu-atman-recovery-python-${{ matrix.python }} + path: experiments/capu_atman_recovery_v1/evidence/ + if-no-files-found: error diff --git a/experiments/capu_atman_http_recovery_v2/.gitignore b/experiments/capu_atman_http_recovery_v2/.gitignore new file mode 100644 index 0000000..1b34339 --- /dev/null +++ b/experiments/capu_atman_http_recovery_v2/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +evidence/ diff --git a/experiments/capu_atman_http_recovery_v2/README.md b/experiments/capu_atman_http_recovery_v2/README.md new file mode 100644 index 0000000..af7d90c --- /dev/null +++ b/experiments/capu_atman_http_recovery_v2/README.md @@ -0,0 +1,92 @@ +# CaPU × ATMAN — bounded loopback HTTP recovery v2 + +This experiment extends the **process boundary**, not the production claim. +The v1 source and PR #103 head (`977864167c65f161e6db87b3d14257a11a67516f`) +remain unchanged. This candidate belongs on a separate branch and **draft PR**. +No merge, deployment, main publication, or ready-for-review transition. + +## What is actually external? + +1. A controller subprocess owns only its controller SQLite path and a loopback URL. +2. A separate HTTP device process executes a non-idempotent counter append in its + own SQLite ledger. The durable insertion **is** the effect in this lab. +3. A third, stdlib-only observer process opens that device ledger read-only. It + counts actual requests/effects and hashes the observed rows, without importing + controller code or reading controller state. + +This is a real TCP/HTTP and process boundary on **one trusted host**, not a real +payment, physical actuator, third-party deployment, WAN or independent attestor. +Process separation does not create OS permission separation. The device deliberately +accepts direct lab calls; bypass-resistant endpoint enforcement is out of scope. + +## Central scenario + +``` +ATMAN authority -> durable A6 UNKNOWN reservation -> one HTTP POST + -> device effect committed -> response held/lost + -> independent observer sees effect -> controller process killed + -> new controller process refuses retry while UNKNOWN + -> policy changes -> exact historical positive receipt arrives + -> A7 reconciliation commits outcome -> new attempts remain blocked +``` + +The device database path is never passed to a controller worker. The observer +triggers the kill only after it has read the effect, while the controller is +still awaiting the held response. The controller and device have no shared +transaction. A receipt query with no matching row returns UNKNOWN, never a +manufactured NOT_COMMITTED. HTTP 503, malformed JSON and socket timeout do not +release the reservation. The HTTP client makes one request with no automatic +retry, redirect or proxy behavior. + +## Tests and equal-guarantee comparison + +There are 14 named scenarios for each of two arms (28 tests): normal ACK, +connection loss after/before the effect, HTTP 503 after the effect, malformed ACK, +observer-triggered controller kill, timeout, device-process restart, receipt +tampering, full-token receipt lookup, competing dispatch workers, stale authority, +receipt replay, and unguarded direct-call duplication. + +The ordinary FSM arm shares HTTP/SQLite I/O, ATMAN authority verification and the +synthetic receipt primitive, but not native CaPU lifecycle transitions. The runner +compares the recorded controller states and **independently observed device rows** +for every named scenario, not merely test pass counts. + +The unguarded control intentionally sends two identical HTTP effect requests: +they must produce two effects. An ambiguous duplicate ledger yields CONFLICT, +not a fabricated clean COMMITTED receipt. This control also demonstrates the +explicit lack of bypass resistance; it is not an architectural superiority claim. + +## Reproduce + +From repository root (Python 3.11 or 3.13): + +```sh +python -m pip install -r experiments/capu_atman_recovery_v1/requirements.txt +python experiments/capu_atman_recovery_v1/bootstrap.py +python experiments/capu_atman_recovery_v1/run.py --output evidence-v1 +python experiments/capu_atman_http_recovery_v2/run.py --output evidence-v2 +``` + +The isolated workflow bootstraps the three pinned upstream modules, runs both +suites, and uploads results. `result.json` includes source hashes, environment, +all observations and a digest. Check the exit code and actual CI run: a local +PASS is not CI PASS or review approval. Source verification pins the reused v1 +bootstrap and adapter; upstream modules are verified by the unchanged bootstrap. + +## Non-claims / review boundary + +- Public deterministic fixture keys and A7 synthetic rotate/XOR receipt tags are + unchanged. No production authentication or cryptographic security claim. +- One trusted controller/device/lineage, one unresolved attempt. No general + exactly-once or liveness guarantee; UNKNOWN may remain blocked indefinitely. +- No physical power-loss/NVRAM, storage rollback, Byzantine device, malicious + host, distributed atomic commit or production transport proof. +- Authorization linearizes at durable admission, not at physical actuation. + A late positive receipt records history; it does not renew old authority. +- The observer is a separate observation path, not an independent organization, + timestamp, trust anchor, or proof that all external events were recorded. +- No CPU/FPGA speed, energy, full ATMAN, Bardo/COSMIC integration, or advantage + over the equal-guarantee ordinary FSM is established. +- Only native Codex review is the review gate. One request per final candidate + head; do not mutate that head after review. Keep draft and unmerged if quota + prevents review. Tests or manual inspection are not review approval. diff --git a/experiments/capu_atman_http_recovery_v2/http_boundary.py b/experiments/capu_atman_http_recovery_v2/http_boundary.py new file mode 100644 index 0000000..14706ef --- /dev/null +++ b/experiments/capu_atman_http_recovery_v2/http_boundary.py @@ -0,0 +1,273 @@ +"""Loopback HTTP recovery lab. NOT a deployed or bypass-resistant actuator. + +The device owns its database. Controller workers receive only a URL and their +own state directory. A separate stdlib-only observer reads the device ledger. +Public fixture keys and synthetic A7 receipts are unchanged from v1. +""" +from __future__ import annotations + +import argparse +from dataclasses import asdict +import hashlib +import http.client +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +from pathlib import Path +import socket +import sqlite3 +import sys +import threading +from urllib.parse import urlsplit + +ROOT = Path(__file__).resolve().parent +V1 = ROOT.parent / 'capu_atman_recovery_v1' +PINS = { + 'bootstrap.py': 'cf66f576ca5d6a67815805261c949a3a87dd890832c81ac386c0600d98d21fcb', + 'proof.py': 'b948f9903f9abec5b6cabbfb5131e5854eb79cfc8b6b9ef0833fe96968b89520', +} +for name, expected in PINS.items(): + if hashlib.sha256((V1 / name).read_bytes()).hexdigest() != expected: + raise RuntimeError('v1 source mismatch: ' + name) +sys.path.insert(0, str(V1)) +from proof import (A6, A7, ISSUER_PUBLIC, Lab, baseline_dispatch, decode_store, + encoded, fixture_receipt, token) # noqa: E402 + +MAX_BODY = 65536 +FAULTS = ('normal', 'drop_before_effect', 'drop_after_effect', + 'http503_after_effect', 'malformed_after_effect', 'hold_after_effect') + + +def endpoint(url: str) -> tuple[str, int]: + parts = urlsplit(url) + if (parts.scheme != 'http' or parts.hostname != '127.0.0.1' or + parts.username or parts.password or parts.path or parts.query or + parts.fragment or parts.port is None): + raise ValueError('only explicit http://127.0.0.1:PORT lab endpoints allowed') + return parts.hostname, parts.port + + +def exchange(url: str, route: str, body: dict, timeout: float = 2.0) -> dict: + """Exactly one HTTP request, no retry/redirect/proxy handling.""" + host, port = endpoint(url) + connection = http.client.HTTPConnection(host, port, timeout=timeout) + try: + connection.request('POST', route, encoded(body), {'Content-Type': 'application/json'}) + response = connection.getresponse() + payload = response.read(MAX_BODY + 1) + if response.status != 200: + return {'transport': 'HTTP_ERROR', 'status': response.status} + if len(payload) > MAX_BODY: + return {'transport': 'INVALID_RESPONSE'} + data = json.loads(payload) + if not isinstance(data, dict): + return {'transport': 'INVALID_RESPONSE'} + return {'transport': 'RESPONSE_RECEIVED', 'body': data} + except (OSError, http.client.HTTPException, ValueError, UnicodeError): + return {'transport': 'NO_USABLE_ACK'} + finally: + connection.close() + + +class HTTPController(Lab): + def __init__(self, root: str | Path, engine: str, url: str): + endpoint(url) # Reject unexpected destinations BEFORE reserving work. + super().__init__(root, engine) + self.url = url + + def initialize(self): + # Initialize controller state only; never create/open the device database. + self.root.mkdir(parents=True, exist_ok=True) + if self.control.exists(): + raise FileExistsError('refusing to overwrite controller state') + store = A6.PersistentOutcomeStore() + if not store.provision(token()): + raise RuntimeError('provision failed') + initial = dict(store=asdict(store), trust=asdict(A7.TrustedDeviceStore(60, 2, 0xBEEF)), + generation=1, state_version='state/1', engine=self.engine, + issuer_key=ISSUER_PUBLIC.hex()) + db = self.connect(self.control, create=True) + try: + db.execute('CREATE TABLE state(id INTEGER PRIMARY KEY CHECK(id=1), value TEXT NOT NULL)') + db.execute('CREATE TABLE audit(id INTEGER PRIMARY KEY, event TEXT NOT NULL, detail TEXT NOT NULL)') + db.execute('INSERT INTO state VALUES(1, ?)', (encoded(initial),)) + finally: + db.close() + + def dispatch(self, bundle: dict, *, now: int = 20, timeout: float = 2.0) -> dict: + # Same admission transaction as pinned v1; only post-commit I/O is changed. + with self.transaction(self.control) as db: + data = self._get(db) + try: + t = A6.AuthorityToken(**bundle['token']) + reasons = self.check_authority(bundle, t, data, now) + except (KeyError, TypeError, ValueError, AttributeError): + reasons = ('MALFORMED_AUTHORITY',) + if reasons: + result = dict(forwarded=False, reason='AUTHORITY', details=list(reasons)) + else: + if self.engine == 'native': + store = decode_store(data['store']) + controller = A6.A6Controller(store) + if not controller.load(t): + raise RuntimeError('fresh-controller load failed') + decision = controller.dispatch(t, commit_effect=False) + data['store'] = asdict(store) + reason = decision.reject_code.name + else: + reason = baseline_dispatch(data['store'], t) + result = dict(forwarded=reason == 'NONE', reason=reason) + event = 'AUTHORIZATION_COMMITTED' if result['forwarded'] else 'DISPATCH_REJECTED' + self._save(db, data, event, {'result': result, 'request': bundle}) + if result['forwarded']: + # The reservation survives all transport errors. Never infer a negative. + result['http'] = exchange(self.url, '/effect', {'token': bundle['token']}, timeout) + return result + + def recover(self, t: dict) -> dict: + received = exchange(self.url, '/receipt', {'token': t}) + if received['transport'] != 'RESPONSE_RECEIVED': + return {'applied': False, 'reason': 'NO_RECEIPT', 'http': received} + body = received['body'] + if body.get('status') != 'COMMITTED' or not isinstance(body.get('receipt'), dict): + return {'applied': False, 'reason': 'NO_POSITIVE_RECEIPT', 'http': received} + # A7 checks device/epoch/sequence/tag AND the stored unresolved identity. + return self.reconcile(body['receipt']) + + def observation(self) -> dict: + data = self.state() + return dict(outcome=data['store']['last_outcome'], + next_attempt=data['store']['next_attempt'], + next_receipt_seq=data['trust']['next_receipt_seq'], + terminal=data['store']['terminal_committed']) + + +class Device(ThreadingHTTPServer): + daemon_threads = True + + def __init__(self, root: Path, fault: str): + if fault not in FAULTS: + raise ValueError('unknown fault') + root.mkdir(parents=True, exist_ok=True) + self.path, self.fault = root / 'effects.sqlite', fault + self.release = threading.Event() + db = self.connect() + try: + db.execute('CREATE TABLE IF NOT EXISTS requests(id INTEGER PRIMARY KEY, token TEXT NOT NULL)') + # NO UNIQUE constraint or deduplication by token, operation or attempt. + db.execute('CREATE TABLE IF NOT EXISTS effects(id INTEGER PRIMARY KEY, token TEXT NOT NULL, receipt TEXT NOT NULL)') + finally: + db.close() + super().__init__(('127.0.0.1', 0), Handler) + + def connect(self): + db = sqlite3.connect(self.path, isolation_level=None, timeout=10) + db.execute('PRAGMA synchronous=FULL') + return db + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def answer(self, value: dict, status: int = 200): + payload = encoded(value).encode() + try: + self.send_response(status) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(payload))) + self.end_headers() + self.wfile.write(payload) + except (BrokenPipeError, ConnectionResetError): + pass + + def drop(self): + self.close_connection = True + try: + self.connection.shutdown(socket.SHUT_RDWR) + except OSError: + pass + self.connection.close() + + def do_POST(self): + try: + size = int(self.headers.get('Content-Length', '0')) + if not 0 < size <= MAX_BODY: + raise ValueError('bad body size') + body = json.loads(self.rfile.read(size)) + if self.path == '/release': + self.server.release.set() + return self.answer({'released': True}) + t = A6.AuthorityToken(**body['token']) + identity = encoded(asdict(t)) + except (KeyError, TypeError, ValueError, UnicodeError): + return self.answer({'error': 'invalid request'}, 400) + db = self.server.connect() + try: + if self.path == '/receipt': + rows = db.execute('SELECT receipt FROM effects WHERE token=? ORDER BY id', (identity,)).fetchall() + if len(rows) != 1: + # Missing data is NOT a negative receipt; duplicates are a conflict. + return self.answer({'status': 'UNKNOWN' if not rows else 'CONFLICT', 'count': len(rows)}) + return self.answer({'status': 'COMMITTED', 'receipt': json.loads(rows[0][0])}) + if self.path != '/effect': + return self.answer({'error': 'not found'}, 404) + db.execute('INSERT INTO requests(token) VALUES(?)', (identity,)) + fault = self.server.fault + if fault == 'drop_before_effect': + return self.drop() + receipt = fixture_receipt(t) + # This durable ledger insertion IS the effect; no cross-system atomicity. + db.execute('INSERT INTO effects(token,receipt) VALUES(?,?)', (identity, encoded(receipt))) + if fault == 'drop_after_effect': + return self.drop() + if fault == 'hold_after_effect' and not self.server.release.wait(10): + return self.drop() + if fault == 'http503_after_effect': + return self.answer({'error': 'injected after COMMIT'}, 503) + if fault == 'malformed_after_effect': + self.send_response(200) + self.send_header('Content-Length', '1') + self.end_headers() + self.wfile.write(b'{') + return + return self.answer({'status': 'COMMITTED', 'receipt': receipt}) + finally: + db.close() + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('mode', choices=('serve', 'init', 'dispatch', 'recover', 'reconcile', 'observe', 'context')) + parser.add_argument('--root', type=Path, required=True) + parser.add_argument('--url') + parser.add_argument('--engine', choices=('native', 'baseline'), default='native') + parser.add_argument('--fault', choices=FAULTS, default='normal') + parser.add_argument('--input', type=Path) + parser.add_argument('--timeout', type=float, default=2.0) + args = parser.parse_args() + if args.mode == 'serve': + server = Device(args.root, args.fault) + print(encoded({'url': f'http://127.0.0.1:{server.server_port}'}), flush=True) + server.serve_forever(poll_interval=0.05) + return + controller = HTTPController(args.root, args.engine, args.url) + value = json.loads(args.input.read_text()) if args.input else {} + if args.mode == 'dispatch': + result = controller.dispatch(value, timeout=args.timeout) + elif args.mode == 'recover': + result = controller.recover(value) + elif args.mode == 'reconcile': + result = controller.reconcile(value) + elif args.mode == 'context': + controller.context(**value) + result = controller.observation() + elif args.mode == 'init': + controller.initialize() + result = controller.observation() + else: + result = controller.observation() + print(encoded(result), flush=True) + + +if __name__ == '__main__': + main() diff --git a/experiments/capu_atman_http_recovery_v2/observer.py b/experiments/capu_atman_http_recovery_v2/observer.py new file mode 100644 index 0000000..0e5c22c --- /dev/null +++ b/experiments/capu_atman_http_recovery_v2/observer.py @@ -0,0 +1,32 @@ +"""Read-only, stdlib-only witness. Does not import or query controller code/state. + +Independent observation path on one trusted host, NOT independent attestation. +""" +import argparse +import hashlib +import json +from pathlib import Path +import sqlite3 + + +def observe(path: Path) -> dict: + db = sqlite3.connect(path.resolve().as_uri() + '?mode=ro', uri=True) + try: + db.execute('PRAGMA query_only=ON') + db.execute('BEGIN') + integrity = db.execute('PRAGMA integrity_check').fetchone()[0] + requests = db.execute('SELECT id,token FROM requests ORDER BY id').fetchall() + effects = db.execute('SELECT id,token,receipt FROM effects ORDER BY id').fetchall() + rows = {'requests': requests, 'effects': effects} + digest = hashlib.sha256(json.dumps(rows, sort_keys=True, separators=(',', ':')).encode()).hexdigest() + return {'integrity': integrity, 'request_count': len(requests), + 'effect_count': len(effects), 'rows_sha256': digest, 'rows': rows} + finally: + db.close() + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('database', type=Path) + args = parser.parse_args() + print(json.dumps(observe(args.database), sort_keys=True)) diff --git a/experiments/capu_atman_http_recovery_v2/run.py b/experiments/capu_atman_http_recovery_v2/run.py new file mode 100644 index 0000000..2376454 --- /dev/null +++ b/experiments/capu_atman_http_recovery_v2/run.py @@ -0,0 +1,67 @@ +"""Run the HTTP experiment; emit FAIL even if a test aborts, never stale PASS.""" +import argparse +import hashlib +import io +import json +from pathlib import Path +import platform +import sqlite3 +import sys +import unittest +import cryptography +import test_http_boundary as tests +from http_boundary import ROOT, PINS + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--output', type=Path, default=ROOT / 'evidence') + args = parser.parse_args() + args.output.mkdir(parents=True, exist_ok=True) + tests.TRACE_LOG.clear() + stream = io.StringIO() + suite = unittest.defaultTestLoader.loadTestsFromModule(tests) + result = unittest.TextTestRunner(stream=stream, verbosity=2).run(suite) + log = stream.getvalue() + (args.output / 'tests.txt').write_text(log) + print(log, flush=True) + native = {t['scenario']: t['steps'] for t in tests.TRACE_LOG if t['engine'] == 'native'} + baseline = {t['scenario']: t['steps'] for t in tests.TRACE_LOG if t['engine'] == 'baseline'} + equal = bool(native) and native == baseline + passed = result.wasSuccessful() and not result.skipped and result.testsRun == 28 and equal + manifest = { + 'schema': 'capu.atman.http-recovery.lab.v2', + 'status': 'BOUNDED_LOOPBACK_HTTP_RECOVERY_PASS' if passed else 'FAIL', + 'tests': {'run': result.testsRun, 'failures': len(result.failures), + 'errors': len(result.errors), 'skipped': len(result.skipped)}, + 'compared_scenarios': len(native), 'same_observations_as_baseline': equal, + 'observed_snapshots_per_arm': sum(len(steps) for steps in native.values()), + 'processes': ['controller subprocess', 'HTTP device subprocess', 'read-only observer subprocess'], + 'v1_head': '977864167c65f161e6db87b3d14257a11a67516f', + 'v1_source_sha256': PINS, + 'source_sha256': {p.name: hashlib.sha256(p.read_bytes()).hexdigest() + for p in sorted(ROOT.glob('*.py'))}, + 'environment': {'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version, + 'cryptography': cryptography.__version__, 'platform': platform.system()}, + 'traces': sorted(tests.TRACE_LOG, key=lambda t: (t['engine'], t['scenario'])), + 'non_claims': [ + 'A real loopback HTTP/process boundary, not a third-party service, physical device or WAN test.', + 'The durable device-ledger insertion IS the external effect; no distributed atomicity is claimed.', + 'Separate observer path, not an independent organization, trust anchor or source-completeness proof.', + 'Same trusted host; process isolation is not OS permission isolation or bypass resistance.', + 'A direct unguarded HTTP caller creates duplicate effects, intentionally demonstrated.', + 'Public fixture keys and synthetic A7 receipt tags remain non-production.', + 'UNKNOWN may remain blocked indefinitely; no general liveness or exactly-once guarantee.', + 'No physical power-loss, NVRAM, Byzantine-device, storage-rollback or production transport proof.', + 'No speed/energy or superiority claim over an equally capable conventional FSM.', + 'No full ATMAN runtime, Bardo/COSMIC integration, deployment or merge.' + ]} + canonical = json.dumps(manifest, sort_keys=True, separators=(',', ':')).encode() + manifest['result_digest_sha256'] = hashlib.sha256(canonical).hexdigest() + (args.output / 'result.json').write_text(json.dumps(manifest, indent=2) + '\n') + print(manifest['status'], manifest['result_digest_sha256'], flush=True) + return 0 if passed else 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/experiments/capu_atman_http_recovery_v2/test_http_boundary.py b/experiments/capu_atman_http_recovery_v2/test_http_boundary.py new file mode 100644 index 0000000..6a954dd --- /dev/null +++ b/experiments/capu_atman_http_recovery_v2/test_http_boundary.py @@ -0,0 +1,266 @@ +"""Real loopback sockets, separate processes and a read-only external observer.""" +from dataclasses import asdict +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import time +import unittest + +from http_boundary import ROOT, exchange +from proof import fixture_bundle, token + +TRACE_LOG = [] + + +class Harness: + def __init__(self, root: Path, engine: str, fault='normal'): + self.root, self.engine, self.serial = root, engine, 0 + self.children = [] + self.start_server(fault) + self.command('init') + + def start_server(self, fault): + self.server = subprocess.Popen( + [sys.executable, str(ROOT / 'http_boundary.py'), 'serve', + '--root', str(self.root / 'device'), '--fault', fault], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + self.children.append(self.server) + ready = self.server.stdout.readline() + if not ready: + raise RuntimeError('device failed: ' + self.server.stderr.read()) + self.url = json.loads(ready)['url'] + + def args(self, mode, value=None, timeout=2.0): + args = [sys.executable, str(ROOT / 'http_boundary.py'), mode, + '--root', str(self.root / 'controller'), '--engine', self.engine, + '--url', self.url, '--timeout', str(timeout)] + if value is not None: + self.serial += 1 + path = self.root / f'input-{self.serial}.json' + path.write_text(json.dumps(value)) + args += ['--input', str(path)] + return args + + def spawn(self, mode, value=None, timeout=2.0): + process = subprocess.Popen(self.args(mode, value, timeout), + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + self.children.append(process) + return process + + def command(self, mode, value=None, timeout=2.0): + result = subprocess.run(self.args(mode, value, timeout), + capture_output=True, text=True, timeout=15) + if result.returncode: + raise AssertionError(result.stderr) + return json.loads(result.stdout) + + def witness(self): + result = subprocess.run([sys.executable, str(ROOT / 'observer.py'), + str(self.root / 'device' / 'effects.sqlite')], + capture_output=True, text=True, check=True, timeout=10) + return json.loads(result.stdout) + + def wait_effect(self): + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + observed = self.witness() + if observed['effect_count'] == 1: + return observed + time.sleep(0.02) + raise AssertionError('external effect not observed') + + def stop_server(self): + self.server.kill() + self.server.communicate(timeout=5) + + def close(self): + for child in self.children: + if child.poll() is None: + child.kill() + child.communicate(timeout=5) + + +class Cases: + engine = None + + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.h = None + self.trace = {'engine': self.engine, 'scenario': self._testMethodName, 'steps': []} + + def begin(self, fault='normal'): + self.h = Harness(Path(self.temp.name), self.engine, fault) + self.addCleanup(self.h.close) + return self.h + + def tearDown(self): + if self.h: + self.h.close() + self.temp.cleanup() + TRACE_LOG.append(self.trace) + + def snapshot(self, label, effects, requests, outcome=None): + witness = self.h.witness() + control = self.h.command('observe') + self.assertEqual(witness['integrity'], 'ok') + self.assertEqual(witness['effect_count'], effects) + self.assertEqual(witness['request_count'], requests) + if outcome is not None: + self.assertEqual(control['outcome'], outcome) + self.trace['steps'].append({'event': label, 'controller': control, 'witness': witness}) + self.assertFalse((self.h.root / 'controller' / 'device.sqlite').exists()) + + def blocked(self, bundle=None, reason='OUTCOME_UNKNOWN'): + result = self.h.command('dispatch', bundle or fixture_bundle(token(1))) + self.assertFalse(result['forwarded']) + self.assertEqual(result['reason'], reason) + self.assertNotIn('http', result) + + def finish(self): + result = self.h.command('recover', asdict(token())) + self.assertTrue(result['applied']) + self.snapshot('LATE_POSITIVE_RECEIPT', 1, 1, 'COMMITTED') + self.blocked(reason='TERMINAL_COMMITTED') + self.snapshot('TERMINAL_RETRY_BLOCKED', 1, 1, 'COMMITTED') + + def lost(self, fault, transport): + h = self.begin(fault) + result = h.command('dispatch', fixture_bundle()) + self.assertTrue(result['forwarded']) + self.assertEqual(result['http']['transport'], transport) + self.snapshot('ACK_UNUSABLE', 1, 1, 'UNKNOWN') + for _ in range(3): + self.blocked() + self.snapshot('THREE_RETRIES_BLOCKED_NO_HTTP', 1, 1, 'UNKNOWN') + self.finish() + + def test_normal_ack_is_not_automatic_reconciliation(self): + h = self.begin() + result = h.command('dispatch', fixture_bundle()) + self.assertEqual(result['http']['transport'], 'RESPONSE_RECEIVED') + self.snapshot('HTTP_ACK_RECEIVED_NOT_YET_RECONCILED', 1, 1, 'UNKNOWN') + self.assertTrue(h.command('reconcile', result['http']['body']['receipt'])['applied']) + self.snapshot('EXACT_ACK_APPLIED', 1, 1, 'COMMITTED') + + def test_connection_dropped_after_effect(self): + self.lost('drop_after_effect', 'NO_USABLE_ACK') + + def test_http503_after_effect_is_not_negative(self): + self.lost('http503_after_effect', 'HTTP_ERROR') + + def test_malformed_ack_after_effect_is_not_negative(self): + self.lost('malformed_after_effect', 'NO_USABLE_ACK') + + def test_dropped_before_effect_stays_unknown(self): + h = self.begin('drop_before_effect') + self.assertEqual(h.command('dispatch', fixture_bundle())['http']['transport'], 'NO_USABLE_ACK') + self.snapshot('REQUEST_RECEIVED_NO_EFFECT', 0, 1, 'UNKNOWN') + result = h.command('recover', asdict(token())) + self.assertFalse(result['applied']) + self.assertEqual(result['reason'], 'NO_POSITIVE_RECEIPT') + self.blocked() + self.snapshot('MISSING_RECORD_NOT_NEGATIVE', 0, 1, 'UNKNOWN') + + def test_observer_triggers_actual_controller_kill(self): + h = self.begin('hold_after_effect') + worker = h.spawn('dispatch', fixture_bundle(), timeout=8) + witness = h.wait_effect() + self.assertEqual(witness['request_count'], 1) + self.assertIsNone(worker.poll(), 'controller must still be awaiting ACK') + worker.kill() + worker.communicate(timeout=5) + self.assertNotEqual(worker.returncode, 0) + self.snapshot('OBSERVER_SAW_EFFECT_THEN_KILLED_CONTROLLER', 1, 1, 'UNKNOWN') + exchange(h.url, '/release', {'release': True}) + self.blocked() + h.command('context', {'generation': 2, 'state_version': 'state/2'}) + self.assertTrue(h.command('recover', asdict(token()))['applied']) + self.blocked(fixture_bundle(token(1), generation=2, state_version='state/2'), + reason='TERMINAL_COMMITTED') + self.snapshot('RESTART_POLICY_CHANGED_HISTORICAL_RECEIPT_APPLIED', 1, 1, 'COMMITTED') + + def test_http_timeout_after_effect(self): + h = self.begin('hold_after_effect') + result = h.command('dispatch', fixture_bundle(), timeout=0.15) + self.assertEqual(result['http']['transport'], 'NO_USABLE_ACK') + h.wait_effect() + self.snapshot('TIMEOUT_AFTER_EFFECT', 1, 1, 'UNKNOWN') + exchange(h.url, '/release', {'release': True}) + self.blocked() + self.finish() + + def test_device_restart_preserves_positive_receipt(self): + h = self.begin('drop_after_effect') + h.command('dispatch', fixture_bundle()) + self.snapshot('BEFORE_DEVICE_KILL', 1, 1, 'UNKNOWN') + h.stop_server() + h.start_server('normal') + self.snapshot('DEVICE_RESTARTED_SAME_LEDGER', 1, 1, 'UNKNOWN') + self.finish() + + def test_tampered_receipt_does_not_release_unknown(self): + h = self.begin('drop_after_effect') + h.command('dispatch', fixture_bundle()) + receipt = exchange(h.url, '/receipt', {'token': asdict(token())})['body']['receipt'] + receipt['auth_tag'] ^= 1 + self.assertFalse(h.command('reconcile', receipt)['authenticated']) + self.snapshot('TAMPERED_RECEIPT_REJECTED', 1, 1, 'UNKNOWN') + self.blocked() + self.finish() + + def test_receipt_query_binds_full_token(self): + h = self.begin('drop_after_effect') + h.command('dispatch', fixture_bundle()) + result = h.command('recover', asdict(token(command_id=10))) + self.assertFalse(result['applied']) + self.snapshot('FOREIGN_OPERATION_NOT_CONFUSED_WITH_OWN_EFFECT', 1, 1, 'UNKNOWN') + self.finish() + + def test_two_dispatch_workers_create_one_effect(self): + h = self.begin() + workers = [h.spawn('dispatch', fixture_bundle()) for _ in range(2)] + results = [] + for worker in workers: + stdout, stderr = worker.communicate(timeout=15) + self.assertEqual(worker.returncode, 0, stderr) + results.append(json.loads(stdout)) + self.assertEqual(sum(r['forwarded'] for r in results), 1) + self.snapshot('TWO_WORKERS_ONE_HTTP_REQUEST', 1, 1, 'UNKNOWN') + self.finish() + + def test_stale_authorization_never_reaches_http(self): + h = self.begin() + h.command('context', {'generation': 2}) + self.blocked(fixture_bundle(), reason='AUTHORITY') + self.snapshot('STALE_AUTHORITY_NO_REQUEST_NO_EFFECT', 0, 0) + + def test_receipt_replay_is_rejected(self): + h = self.begin() + reply = h.command('dispatch', fixture_bundle())['http']['body']['receipt'] + self.assertTrue(h.command('reconcile', reply)['applied']) + self.assertFalse(h.command('reconcile', reply)['applied']) + self.snapshot('EXACT_RECEIPT_REPLAY_REJECTED', 1, 1, 'COMMITTED') + + def test_unguarded_direct_http_really_duplicates(self): + h = self.begin() + for _ in range(2): + self.assertEqual(exchange(h.url, '/effect', {'token': asdict(token())})['transport'], + 'RESPONSE_RECEIVED') + self.snapshot('BYPASS_CONTROL_TWO_REQUESTS_TWO_EFFECTS', 2, 2) + receipt = exchange(h.url, '/receipt', {'token': asdict(token())})['body'] + self.assertEqual(receipt['status'], 'CONFLICT') + self.assertNotIn('receipt', receipt) + + +class NativeTests(Cases, unittest.TestCase): + engine = 'native' + + +class BaselineTests(Cases, unittest.TestCase): + engine = 'baseline' + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/experiments/capu_atman_recovery_v1/.gitignore b/experiments/capu_atman_recovery_v1/.gitignore new file mode 100644 index 0000000..b5ee164 --- /dev/null +++ b/experiments/capu_atman_recovery_v1/.gitignore @@ -0,0 +1,4 @@ +.upstream/ +__pycache__/ +evidence/ +*.sqlite* diff --git a/experiments/capu_atman_recovery_v1/README.md b/experiments/capu_atman_recovery_v1/README.md new file mode 100644 index 0000000..5014e8c --- /dev/null +++ b/experiments/capu_atman_recovery_v1/README.md @@ -0,0 +1,140 @@ +# CaPU × ATMAN: process-crash recovery laboratory + +**Status:** bounded software composition experiment; not a production runtime. + +This experiment executes the **unmodified** CaPU A6/A7 Python models and +ATMAN's **unmodified** Ed25519 authority module, pinned to exact commits and +verified by Git blob ID before import. It supplies the missing integration +boundary: current authorization + durable dispatch reservation + a separately +stored mock effect + late outcome reconciliation after process termination. + +## Reproduce + +Python 3.11+ and `cryptography==46.0.4` are required. Tested here on Python +3.13.5. The archive includes exact upstream module snapshots; repository +checkouts download the same three pinned files with the bootstrap command. + +```sh +python -m pip install cryptography==46.0.4 +python bootstrap.py +python run.py --output evidence +``` + +No account, API key, paid service, FPGA, real actuator or private credential is +used. The keys and receipt secret in `proof.py` are **public test fixtures**. +Do not reuse them for any real system. + +The first bootstrap may access three fixed raw.githubusercontent.com URLs. +Every subsequent import validates the downloaded source before execution. +No automatic fallback to a newer revision is allowed. + +## Exact upstream boundary + +| Component | Commit | File | +|---|---|---| +| CaPU A6 | `5cdaa5280348841bf8448c5a7844c273df257c5d` | `tools/astra_capu_outcome_reconciliation_a6.py` | +| CaPU A7 | same | `tools/astra_capu_authenticated_receipt_a7.py` | +| ATMAN authority | `e62c279b9148a7ae9dd1a4654f6ddeea6add4a3f` | `model/authority.py` | + +A7's existing deterministic scenario must retain result digest +`6781dbfbd1b529866709980a3a85a38bd37f505daaddd53fd7c8e106ab863d2f`. +The A7 source is from draft PR #102, not an assertion that A7 is merged into +CaPU main. Only ATMAN authority is imported; its complete governance runtime +is not integrated. Bardo and COSMIC are deliberately outside this experiment. + +## The tested story + +```text +ATMAN signed action + exact state version + current policy generation + -> CaPU A6 reserves the attempt as UNKNOWN + -> controller commits authorization and reservation in control.sqlite + -> non-idempotent counter appends the effect to separate device.sqlite + -> child process exits without cleanup before acknowledgement + -> fresh process loads UNKNOWN and blocks same/successor attempts + -> forged negative receipt is rejected without consuming receipt sequence + -> policy changes; the historical effect still needs reconciliation + -> exact late A7 COMMITTED receipt closes the original attempt + -> even a newly authorized retry is blocked by terminal outcome +``` + +`AUTHORIZATION_COMMITTED` and effect `COMMITTED` are distinct events. The +counter deliberately has no deduplication constraint: a duplicate invocation +would create a second effect, making the tested failure observable. + +## Crash and concurrency coverage + +The tests terminate child processes with `os._exit` at five named cut points: +before controller commit, after controller commit, after device effect commit, +during receipt reconciliation, and after receipt reconciliation commit. +This is process-failure injection, **not a physical power-loss experiment**. + +A separate test runs two dispatch processes against one controller database. +The reservation, current-context verification and audit write share a +`BEGIN IMMEDIATE` transaction. Device effects are in a separate transaction +and separate database: there is no fictitious cross-system atomic commit. + +Receipt-sequence consumption and outcome reconciliation are committed together. +A7's existing rule is preserved: an authenticated but semantically rejected +receipt consumes its sequence; an authentication rejection does not. + +Negative cases cover stale policy, stale state, wrong role/scope, expired and +future-dated authorization, changed action, malformed authority, uncommitted +requests, foreign lineage/device/key epoch, wrong receipt sequence/attempt, +forged or malformed receipts, terminal conflict and missing storage. + +## Fair conventional baseline + +The second arm independently implements an ordinary finite-state machine. It +receives the same data and shares the same SQLite persistence, actual ATMAN +signature verifier and low-level synthetic tag calculation. It does **not** +call the native CaPU dispatch or reconciliation state-transition methods. + +Both arms execute the same 26 named scenarios. Two additional tests check the +existing A6 scenario and exact A7 result digest. One differential test compares +24 seeded traces of 12 actions each: **288 transition comparisons**. + +The total is **55 tests**, not 55 independent bugs or proofs. Equal results +show compatibility for these checks. They do not establish a correctness, +performance or novelty advantage over a well-built conventional system. + +## Trust and non-claims + +The controller, fixture key provisioning, local storage and mock device are +trusted. The device receipt primitive remains A7's transparent rotate/XOR +synthetic tag; it is not production authentication. The adapter assumes the +mock command cannot be duplicated or injected outside its dispatch method. +There is no real transport, external isolation or unbypassable hardware gate. + +Authorization is evaluated at **durable admission**, not at the exact physical +instant of the external effect. Revocation after admission does not retroactively +cancel an already admitted operation. Historical receipts are reconciled without +reusing old execution permission; new attempts still require current permission. + +If the process exits after reservation but before invoking the device, the +controller also sees UNKNOWN. It must not infer NOT_COMMITTED from missing +rows or a timeout. Only a definitive trusted receipt can release a successor. +Consequently indefinite HOLD is allowed: **no general liveness or exactly-once +guarantee is claimed**. + +The source uses SQLite rollback journals and synchronous=FULL. Persistence +remains conditional on filesystem, OS and storage behavior; no torn-write, +adversarial rollback, corrupt disk or power-loss guarantee is established here. +See https://www.sqlite.org/atomiccommit.html for SQLite's assumptions. + +The evidence JSON records source hashes, environment, decisions and counts. Its +digest detects changes relative to a known digest, not omitted events, false +inputs, an independent timestamp or an external trust anchor. + +No CPU/FPGA speed, energy, full ATMAN runtime, Bardo/COSMIC composition, +multi-organization governance, or real payment effect is demonstrated. + +## Files and next acceptance boundary + +`bootstrap.py` verifies upstream inputs; `proof.py` implements the adapter and +baseline; `test_proof.py` holds the cases; `run.py` produces the manifest and +human-readable log. `evidence/result.json` is the machine-readable local run. + +Before production or broader integration, replace the mock effect/receipt seam +with one real, independently observable device/API operation and test its +transport, idempotency, receipt finality and admission/revocation contract. +The current experiment must not be renamed a complete platform implementation. diff --git a/experiments/capu_atman_recovery_v1/bootstrap.py b/experiments/capu_atman_recovery_v1/bootstrap.py new file mode 100644 index 0000000..f78bb3e --- /dev/null +++ b/experiments/capu_atman_recovery_v1/bootstrap.py @@ -0,0 +1,72 @@ +"""Fetch pinned, unmodified upstream modules; verify Git blob IDs before use.""" +from __future__ import annotations +import hashlib +import importlib.util +from pathlib import Path +import sys +import types +import urllib.request + +ROOT = Path(__file__).resolve().parent +PINS = [ + ('capu', 'safal207/CaPU', '5cdaa5280348841bf8448c5a7844c273df257c5d', + 'tools/astra_capu_outcome_reconciliation_a6.py', '4686275e872e5f1348870f09f20fac560be31ff8'), + ('capu', 'safal207/CaPU', '5cdaa5280348841bf8448c5a7844c273df257c5d', + 'tools/astra_capu_authenticated_receipt_a7.py', 'b79dcf770248a8e8c667f488ebfc36b2d8b80999'), + ('atman', 'safal207/ATMAN-LATTICE', 'e62c279b9148a7ae9dd1a4654f6ddeea6add4a3f', + 'model/authority.py', '8bb0e7122c3d3acbe5e710ec21537d4413bfb69d'), +] + + +def verify(data: bytes, expected: str) -> None: + actual = hashlib.sha1(b'blob ' + str(len(data)).encode() + b'\0' + data).hexdigest() + if actual != expected: + raise ValueError(f'Upstream blob mismatch: expected {expected}, got {actual}') + + +def sources(fetch: bool = False) -> list[Path]: + paths = [] + for name, repo, ref, path, expected in PINS: + dest = ROOT / '.upstream' / name / path + if not dest.exists(): + if not fetch: + raise FileNotFoundError(f'{dest}: run python bootstrap.py first') + url = f'https://raw.githubusercontent.com/{repo}/{ref}/{path}' + with urllib.request.urlopen(url, timeout=30) as response: + data = response.read(100_000) + verify(data, expected) + dest.parent.mkdir(parents=True, exist_ok=True) + temporary = dest.with_suffix('.download') + temporary.write_bytes(data) + temporary.replace(dest) + verify(dest.read_bytes(), expected) + paths.append(dest) + return paths + + +def module(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise ImportError(str(path)) + value = importlib.util.module_from_spec(spec) + sys.modules[name] = value + spec.loader.exec_module(value) + return value + + +def load(): + a6_path, a7_path, authority_path = sources() + # This executable is an isolated experiment, not a replacement tools package. + if 'tools' not in sys.modules: + tools = types.ModuleType('tools') + tools.__path__ = [str(a6_path.parent)] + sys.modules['tools'] = tools + a6 = module('tools.astra_capu_outcome_reconciliation_a6', a6_path) + a7 = module('tools.astra_capu_authenticated_receipt_a7', a7_path) + authority = module('transition_proof_atman_authority', authority_path) + return a6, a7, authority + + +if __name__ == '__main__': + for path in sources(fetch=True): + print('PIN_VERIFIED', path.relative_to(ROOT)) diff --git a/experiments/capu_atman_recovery_v1/proof.py b/experiments/capu_atman_recovery_v1/proof.py new file mode 100644 index 0000000..5810157 --- /dev/null +++ b/experiments/capu_atman_recovery_v1/proof.py @@ -0,0 +1,320 @@ +"""Bounded CaPU A6/A7 + ATMAN authority composition with process-crash tests. + +Experiment only. One trusted controller/device/lineage. The A7 receipt tag is +synthetic. Dispatch admission, not physical actuation, is the policy cut-off. +""" +from __future__ import annotations +import argparse +from contextlib import contextmanager +from dataclasses import asdict +import json +import os +from pathlib import Path +import sqlite3 +from typing import Any, Iterator +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives import serialization +from bootstrap import load + +A6, A7, AUTH = load() +ROLE, SCOPE = 'effect.execute', 'mock/counter' +# Public, deterministic test fixtures. NEVER use these keys outside this lab. +ISSUER = Ed25519PrivateKey.from_private_bytes(bytes(range(32))) +ACTOR = Ed25519PrivateKey.from_private_bytes(bytes(range(32, 64))) +ISSUER_PUBLIC = ISSUER.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) +CRASH_CODES = {'before_commit': 91, 'after_commit': 92, 'after_effect': 93, + 'during_reconcile': 94, 'after_reconcile': 95} + + +def encoded(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(',', ':')) + + +def token(attempt: int = 0, **changes: Any): + fields = dict(authority_tag=167, incarnation=2, queue_epoch=7, slot_id=1, + command_id=9, attempt_id=attempt, effect_id=12, committed=True) + fields.update(changes) + return A6.AuthorityToken(**fields) + + +def action_for(t, state_version: str) -> dict: + return {'domain': 'capu-atman-recovery-lab/v1', 'operation': 'counter.increment', + 'delta': 1, 'state_version': state_version, 'token': asdict(t)} + + +def fixture_bundle(t=None, *, generation=1, state_version='state/1', + role=ROLE, scope=SCOPE, signed_at=10, valid_until=100) -> dict: + t = token() if t is None else t + action = action_for(t, state_version) + grant = AUTH.issue_authority_grant( + grant_id='lab-grant', subject_ref='lab-actor', subject_key_id='actor-key', + subject_public_key=ACTOR.public_key(), roles=(role,), scopes=(scope,), + policy_generation=generation, valid_from=1, valid_until=valid_until, + issuer_ref='lab-issuer', issuer_key_id='issuer-key', issuer_private_key=ISSUER) + proof = AUTH.sign_authorized_action(grant, private_key=ACTOR, role=role, + scope=scope, action=action, signed_at=signed_at) + return {'token': asdict(t), 'action': action, 'grant': asdict(grant), 'proof': asdict(proof)} + + +def fixture_receipt(t=None, *, seq=0, outcome='COMMITTED', **changes) -> dict: + t = token() if t is None else t + fields = {k: v for k, v in asdict(t).items() if k != 'committed'} + fields.update(device_id=60, key_epoch=2, receipt_seq=seq, outcome=A6.Outcome(outcome)) + fields.update(changes) + return asdict(A7.DeviceReceipt.signed(secret=0xBEEF, **fields)) + + +def decode_store(data: dict): + data = dict(data) + if data['lineage_value'] is not None: + data['lineage_value'] = tuple(data['lineage_value']) + data['last_outcome'] = A6.Outcome(data['last_outcome']) + return A6.PersistentOutcomeStore(**data) + + +def baseline_dispatch(s: dict, t) -> str: + """Independent ordinary FSM; same inputs and guarantees as the native arm.""" + if not t.committed: + return 'UNCOMMITTED' + if s['lineage_value'] is None: + return 'PERSISTENT_MISSING' + if tuple(s['lineage_value']) != t.lineage(): + return 'PERSISTENT_LINEAGE' + if s['terminal_committed']: + return 'TERMINAL_COMMITTED' + if s['terminal_conflict']: + return 'TERMINAL_CONFLICT' + if s['unresolved_valid']: + return 'OUTCOME_UNKNOWN' + if t.attempt_id != s['next_attempt']: + return 'PERSISTENT_FRONTIER' + if s['next_attempt'] == (1 << s['width_bits']) - 1: + return 'FRONTIER_EXHAUSTED' + s.update(unresolved_valid=True, unresolved_attempt=t.attempt_id, + next_attempt=t.attempt_id + 1, last_outcome='UNKNOWN') + return 'NONE' + + +def baseline_receipt(s: dict, trust: dict, r) -> dict: + checks = ((trust['valid'], 'TRUST_MISSING'), + (r.device_id == trust['trusted_device_id'], 'DEVICE_ID'), + (r.key_epoch == trust['trusted_key_epoch'], 'KEY_EPOCH'), + (r.receipt_seq == trust['next_receipt_seq'], 'RECEIPT_SEQUENCE'), + (r.auth_tag == r.expected_tag(trust['secret'], width_bits=trust['auth_width_bits']), 'AUTH_TAG')) + for ok, reason in checks: + if not ok: + return dict(authenticated=False, applied=False, auth=reason, semantic='NONE') + trust['next_receipt_seq'] += 1 # A7 accepts consume even on semantic reject. + checks = ((s['lineage_value'] is not None, 'PERSISTENT_MISSING'), + (tuple(s['lineage_value'] or []) == r.token().lineage(), 'PERSISTENT_LINEAGE'), + (not (s['terminal_committed'] or s['terminal_conflict']), 'TERMINAL'), + (s['unresolved_valid'], 'NO_UNRESOLVED_ATTEMPT'), + (r.attempt_id == s['unresolved_attempt'], 'ATTEMPT_MISMATCH'), + (r.outcome in (A6.Outcome.NOT_COMMITTED, A6.Outcome.COMMITTED, A6.Outcome.CONFLICT), 'INVALID_OUTCOME')) + for ok, reason in checks: + if not ok: + return dict(authenticated=True, applied=False, auth='NONE', semantic=reason) + s.update(unresolved_valid=False, last_outcome=r.outcome.value, + last_resolved_attempt=r.attempt_id, + terminal_committed=r.outcome is A6.Outcome.COMMITTED, + terminal_conflict=r.outcome is A6.Outcome.CONFLICT) + return dict(authenticated=True, applied=True, auth='NONE', semantic='NONE') + + +class Lab: + def __init__(self, root: str | Path, engine: str = 'native'): + if engine not in ('native', 'baseline'): + raise ValueError('unknown engine') + self.root, self.engine = Path(root), engine + self.control, self.device = self.root / 'control.sqlite', self.root / 'device.sqlite' + + @staticmethod + def connect(path: Path, *, create=False): + db = sqlite3.connect(str(path) if create else path.resolve().as_uri() + '?mode=rw', + uri=not create, timeout=10, isolation_level=None) + db.execute('PRAGMA synchronous=FULL') + return db + + @contextmanager + def transaction(self, path: Path) -> Iterator[sqlite3.Connection]: + db = self.connect(path) + try: + db.execute('BEGIN IMMEDIATE') + yield db + db.commit() + except BaseException: + db.rollback() + raise + finally: + db.close() + + def initialize(self): + self.root.mkdir(parents=True, exist_ok=True) + if self.control.exists() or self.device.exists(): + raise FileExistsError('Refusing to overwrite an existing experiment') + store = A6.PersistentOutcomeStore() + if not store.provision(token()): + raise RuntimeError('provision failed') + initial = dict(store=asdict(store), trust=asdict(A7.TrustedDeviceStore(60, 2, 0xBEEF)), + generation=1, state_version='state/1', engine=self.engine, + issuer_key=ISSUER_PUBLIC.hex()) + for path in (self.control, self.device): + db = self.connect(path, create=True) + try: + db.execute('PRAGMA journal_mode=DELETE') + if path == self.control: + db.execute('CREATE TABLE state(id INTEGER PRIMARY KEY CHECK(id=1), value TEXT NOT NULL)') + db.execute('CREATE TABLE audit(id INTEGER PRIMARY KEY, event TEXT NOT NULL, detail TEXT NOT NULL)') + db.execute('INSERT INTO state VALUES(1, ?)', (encoded(initial),)) + else: + # Deliberately NOT idempotent: duplicate calls make duplicate effects. + db.execute('CREATE TABLE effects(id INTEGER PRIMARY KEY, attempt INTEGER, detail TEXT NOT NULL)') + finally: + db.close() + + def _get(self, db): + row = db.execute('SELECT value FROM state WHERE id=1').fetchone() + if row is None: + raise RuntimeError('missing state; do not provision automatically') + data = json.loads(row[0]) + if data['engine'] != self.engine: + raise ValueError('cannot switch engine for existing state') + return data + + @staticmethod + def _save(db, data, event, detail): + db.execute('UPDATE state SET value=? WHERE id=1', (encoded(data),)) + db.execute('INSERT INTO audit(event,detail) VALUES(?,?)', (event, encoded(detail))) + + def state(self) -> dict: + db = self.connect(self.control) + try: + return self._get(db) + finally: + db.close() + + def effects(self) -> int: + db = self.connect(self.device) + try: + return db.execute('SELECT COUNT(*) FROM effects').fetchone()[0] + finally: + db.close() + + def context(self, *, generation=None, state_version=None): + with self.transaction(self.control) as db: + data = self._get(db) + if generation is not None: + if generation <= data['generation']: + raise ValueError('policy generation must increase') + data['generation'] = generation + if state_version is not None: + data['state_version'] = state_version + self._save(db, data, 'CONTEXT_CHANGED', {}) + + @staticmethod + def check_authority(bundle, t, data, now): + grant, proof = AUTH.AuthorityGrant(**bundle['grant']), AUTH.AuthorityProof(**bundle['proof']) + expected = action_for(t, data['state_version']) + if bundle['action'] != expected: + return ('CONTEXT_OR_ACTION_MISMATCH',) + if proof.role != ROLE or proof.scope != SCOPE: + return ('REQUIRED_ROLE_OR_SCOPE',) + if proof.signed_at > now: + return ('FUTURE_DATED_PROOF',) + ok, reasons = AUTH.verify_authority_proof( + grant, proof, action=expected, + trusted_issuer_keys={'issuer-key': bytes.fromhex(data['issuer_key'])}, + current_policy_generation=data['generation'], now=now) + return () if ok else reasons + + def dispatch(self, bundle: dict, *, now: int = 20, crash: str | None = None) -> dict: + if crash not in (None, 'before_commit', 'after_commit', 'after_effect'): + raise ValueError('invalid dispatch crash point') + with self.transaction(self.control) as db: + data = self._get(db) + try: + t = A6.AuthorityToken(**bundle['token']) + reasons = self.check_authority(bundle, t, data, now) + except (KeyError, TypeError, ValueError, AttributeError): + reasons = ('MALFORMED_AUTHORITY',) + if reasons: + result = dict(forwarded=False, reason='AUTHORITY', details=list(reasons)) + else: + if self.engine == 'native': + store = decode_store(data['store']) + controller = A6.A6Controller(store) + if not controller.load(t): + raise RuntimeError('fresh-controller load failed') + decision = controller.dispatch(t, commit_effect=False) + data['store'] = asdict(store) + reason = decision.reject_code.name + else: + reason = baseline_dispatch(data['store'], t) + result = dict(forwarded=reason == 'NONE', reason=reason) + event = 'AUTHORIZATION_COMMITTED' if result['forwarded'] else 'DISPATCH_REJECTED' + self._save(db, data, event, {'result': result, 'request': bundle}) + if crash == 'before_commit': + os._exit(CRASH_CODES[crash]) + # Policy is linearized at the preceding durable admission, not at actuation. + if not result['forwarded']: + return result + if crash == 'after_commit': + os._exit(CRASH_CODES[crash]) + with self.transaction(self.device) as db: + db.execute('INSERT INTO effects(attempt,detail) VALUES(?,?)', + (t.attempt_id, encoded(bundle['action']))) + if crash == 'after_effect': + os._exit(CRASH_CODES[crash]) + # No acknowledgement is invented here. Receipt reconciliation is separate. + return result + + def reconcile(self, receipt: dict, *, crash: str | None = None) -> dict: + if crash not in (None, 'during_reconcile', 'after_reconcile'): + raise ValueError('invalid reconciliation crash point') + with self.transaction(self.control) as db: + data = self._get(db) + try: + fields = dict(receipt) + fields['outcome'] = A6.Outcome(fields['outcome']) + r = A7.DeviceReceipt(**fields) + if self.engine == 'native': + store = decode_store(data['store']) + trust = A7.TrustedDeviceStore(**data['trust']) + decision = A7.A7Controller(A6.A6Controller(store), trust).process_receipt(r) + data.update(store=asdict(store), trust=asdict(trust)) + result = dict(authenticated=decision.authenticated, applied=decision.reconcile_accept, + auth=decision.auth_reject_code.name, semantic=decision.reconcile_reject_code.name) + else: + result = baseline_receipt(data['store'], data['trust'], r) + except (KeyError, TypeError, ValueError, AttributeError): + # The pure upstream model may reject malformed inputs by raising; + # don't persist any partial mutations from a malformed envelope. + data = self._get(db) + result = dict(authenticated=False, applied=False, auth='MALFORMED_RECEIPT', semantic='NONE') + self._save(db, data, 'RECEIPT', {'result': result, 'receipt': receipt}) + if crash == 'during_reconcile': + os._exit(CRASH_CODES[crash]) + if crash == 'after_reconcile': + os._exit(CRASH_CODES[crash]) + return result + + def observation(self) -> dict: + state = self.state() + return dict(outcome=state['store']['last_outcome'], + next_attempt=state['store']['next_attempt'], + next_receipt_seq=state['trust']['next_receipt_seq'], + terminal=state['store']['terminal_committed'], effects=self.effects()) + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('operation', choices=('dispatch', 'reconcile')) + parser.add_argument('root', type=Path) + parser.add_argument('input', type=Path) + parser.add_argument('--engine', choices=('native', 'baseline'), required=True) + parser.add_argument('--crash', choices=tuple(CRASH_CODES)) + args = parser.parse_args() + lab = Lab(args.root, args.engine) + result = getattr(lab, args.operation)(json.loads(args.input.read_text()), crash=args.crash) + print(encoded(result)) diff --git a/experiments/capu_atman_recovery_v1/requirements.txt b/experiments/capu_atman_recovery_v1/requirements.txt new file mode 100644 index 0000000..c011dd5 --- /dev/null +++ b/experiments/capu_atman_recovery_v1/requirements.txt @@ -0,0 +1 @@ +cryptography==46.0.4 diff --git a/experiments/capu_atman_recovery_v1/run.py b/experiments/capu_atman_recovery_v1/run.py new file mode 100644 index 0000000..3506304 --- /dev/null +++ b/experiments/capu_atman_recovery_v1/run.py @@ -0,0 +1,99 @@ +"""Run every test and emit a reproducible bounded evidence manifest.""" +from __future__ import annotations +import argparse +import hashlib +import io +import json +from pathlib import Path +import platform +import sqlite3 +import sys +import tempfile +import unittest +import cryptography +from bootstrap import PINS, ROOT, sources +from proof import Lab, fixture_bundle, fixture_receipt, token +from test_proof import crash +import test_proof + + +def sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def scenario(engine: str) -> dict: + with tempfile.TemporaryDirectory() as root: + lab = Lab(root, engine) + lab.initialize() + lab = crash(lab, 'after_effect', fixture_bundle()) + steps = [{'event': 'PROCESS_EXIT_AFTER_EFFECT_BEFORE_ACK', **lab.observation()}] + result = lab.dispatch(fixture_bundle(token(1))) + steps.append({'event': 'RETRY_WHILE_UNKNOWN', 'decision': result, **lab.observation()}) + forged = fixture_receipt(outcome='NOT_COMMITTED') + forged['auth_tag'] ^= 1 + result = lab.reconcile(forged) + steps.append({'event': 'FORGED_NEGATIVE', 'decision': result, **lab.observation()}) + # Current authority changes must not erase a historical external effect. + lab.context(generation=2, state_version='state/2') + result = lab.reconcile(fixture_receipt()) + steps.append({'event': 'LATE_EXACT_RECEIPT_AFTER_POLICY_CHANGE', + 'decision': result, **lab.observation()}) + result = lab.dispatch(fixture_bundle(token(1), generation=2, state_version='state/2')) + steps.append({'event': 'RETRY_AFTER_TERMINAL_COMMIT', 'decision': result, **lab.observation()}) + return {'engine': engine, 'steps': steps} + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument('--output', type=Path, default=ROOT / 'evidence') + args = parser.parse_args() + args.output.mkdir(parents=True, exist_ok=True) + upstream = sources() + log = io.StringIO() + suite = unittest.defaultTestLoader.loadTestsFromModule(test_proof) + result = unittest.TextTestRunner(stream=log, verbosity=2).run(suite) + (args.output / 'tests.txt').write_text(log.getvalue()) + print(log.getvalue()) + if not result.wasSuccessful(): + return 1 + native, baseline = scenario('native'), scenario('baseline') + if native['steps'] != baseline['steps']: + raise AssertionError('end-to-end baseline/native disagreement') + manifest = { + 'schema': 'capu.atman.process-recovery.lab.v1', + 'status': 'BOUNDED_SOFTWARE_COMPOSITION_PASS', + 'tests': {'run': result.testsRun, 'failures': len(result.failures), + 'errors': len(result.errors), 'skipped': len(result.skipped)}, + 'differential': {'seed': 20260905, 'traces': 24, 'steps_per_trace': 12, + 'compared_steps': 288, 'all_equal': True}, + 'process_crash_cut_points': ['before_commit', 'after_commit', 'after_effect', + 'during_reconcile', 'after_reconcile'], + 'baseline': 'independent ordinary FSM; shared I/O, ATMAN verifier and synthetic tag primitive', + 'native_scenario': native, 'baseline_scenario': baseline, + 'environment': {'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version, + 'cryptography': cryptography.__version__, 'platform': platform.system()}, + 'upstream': [dict(repository=p[1], commit=p[2], path=p[3], git_blob_sha1=p[4], + sha256=sha256(path)) for p, path in zip(PINS, upstream)], + 'source_sha256': {name: sha256(ROOT / name) + for name in ('bootstrap.py', 'proof.py', 'test_proof.py', 'run.py')}, + 'non_claims': [ + 'No advantage over the equally capable conventional baseline established.', + 'No CPU/FPGA speed or energy measurement.', + 'No physical device, network, power-loss, adversarial-storage or Byzantine proof.', + 'No production cryptographic claim for A7 synthetic receipts or public fixture keys.', + 'No exactly-once or general liveness guarantee; UNKNOWN can remain blocked indefinitely.', + 'No full ATMAN runtime or Bardo/COSMIC integration.', + 'Native inputs are three pinned modules, not the entire repositories or test suites.', + 'Dispatch policy linearizes at durable admission, not at physical actuation.', + 'Evidence digest is not an independent timestamp, trust anchor or completeness proof.', + ], + } + canonical = json.dumps(manifest, sort_keys=True, separators=(',', ':')).encode() + manifest['result_digest_sha256'] = hashlib.sha256(canonical).hexdigest() + (args.output / 'result.json').write_text(json.dumps(manifest, indent=2) + '\n') + print('BOUNDED_SOFTWARE_COMPOSITION_PASS', manifest['result_digest_sha256']) + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/experiments/capu_atman_recovery_v1/test_proof.py b/experiments/capu_atman_recovery_v1/test_proof.py new file mode 100644 index 0000000..0a6807c --- /dev/null +++ b/experiments/capu_atman_recovery_v1/test_proof.py @@ -0,0 +1,262 @@ +"""Real process exits, negative boundaries, native regressions and equal baselines.""" +from __future__ import annotations +from concurrent.futures import ThreadPoolExecutor +import copy +import json +from pathlib import Path +import random +import subprocess +import sys +import tempfile +import unittest +from proof import A6, A7, Lab, CRASH_CODES, fixture_bundle, fixture_receipt, token + +HERE = Path(__file__).resolve().parent + + +def crash(lab, point, value, operation='dispatch'): + request = lab.root / f'{point}.json' + request.write_text(json.dumps(value)) + run = subprocess.run([sys.executable, str(HERE / 'proof.py'), operation, + str(lab.root), str(request), '--engine', lab.engine, + '--crash', point], capture_output=True, text=True, timeout=15) + if run.returncode != CRASH_CODES[point]: + raise AssertionError(f'{point}: expected {CRASH_CODES[point]}, got {run.returncode}: {run.stderr}') + return Lab(lab.root, lab.engine) + + +class Cases: + def before_commit(self, lab): + lab = crash(lab, 'before_commit', fixture_bundle()) + self.assertEqual(lab.observation(), dict(outcome='NONE', next_attempt=0, + next_receipt_seq=0, terminal=False, effects=0)) + self.assertTrue(lab.dispatch(fixture_bundle())['forwarded']) + self.assertEqual(lab.effects(), 1) + + def after_commit(self, lab): + lab = crash(lab, 'after_commit', fixture_bundle()) + self.assertEqual(lab.observation(), dict(outcome='UNKNOWN', next_attempt=1, + next_receipt_seq=0, terminal=False, effects=0)) + self.assertEqual(lab.dispatch(fixture_bundle(token(1)))['reason'], 'OUTCOME_UNKNOWN') + self.assertEqual(lab.effects(), 0) # No receipt inferred from absence. + + def lost_ack_restart_late_receipt(self, lab): + lab = crash(lab, 'after_effect', fixture_bundle()) + self.assertEqual(lab.observation(), dict(outcome='UNKNOWN', next_attempt=1, + next_receipt_seq=0, terminal=False, effects=1)) + for attempt in (0, 1): + self.assertEqual(lab.dispatch(fixture_bundle(token(attempt)))['reason'], 'OUTCOME_UNKNOWN') + self.assertTrue(lab.reconcile(fixture_receipt())['applied']) + self.assertEqual(lab.dispatch(fixture_bundle(token(1)))['reason'], 'TERMINAL_COMMITTED') + self.assertEqual(lab.observation(), dict(outcome='COMMITTED', next_attempt=1, + next_receipt_seq=1, terminal=True, effects=1)) + + def atomic_reconcile_rollback(self, lab): + lab.dispatch(fixture_bundle()) + lab = crash(lab, 'during_reconcile', fixture_receipt(), 'reconcile') + self.assertEqual(lab.observation()['outcome'], 'UNKNOWN') + self.assertEqual(lab.observation()['next_receipt_seq'], 0) + self.assertTrue(lab.reconcile(fixture_receipt())['applied']) + self.assertEqual(lab.observation()['next_receipt_seq'], 1) + + def after_reconcile_commit(self, lab): + lab.dispatch(fixture_bundle()) + lab = crash(lab, 'after_reconcile', fixture_receipt(), 'reconcile') + self.assertEqual(lab.observation()['outcome'], 'COMMITTED') + self.assertEqual(lab.reconcile(fixture_receipt())['auth'], 'RECEIPT_SEQUENCE') + self.assertFalse(lab.dispatch(fixture_bundle(token(1)))['forwarded']) + self.assertEqual(lab.effects(), 1) + + def stale_generation(self, lab): + request = fixture_bundle() + lab.context(generation=2) + result = lab.dispatch(request) + self.assertEqual(result['reason'], 'AUTHORITY') + self.assertIn('stale_authority_policy_generation', result['details']) + self.assertEqual(lab.effects(), 0) + + def stale_state_version(self, lab): + request = fixture_bundle() + lab.context(state_version='state/2') + self.assertIn('CONTEXT_OR_ACTION_MISMATCH', lab.dispatch(request)['details']) + self.assertEqual(lab.effects(), 0) + + def wrong_role(self, lab): + self.assertIn('REQUIRED_ROLE_OR_SCOPE', lab.dispatch(fixture_bundle(role='effect.review'))['details']) + self.assertEqual(lab.effects(), 0) + + def wrong_scope(self, lab): + self.assertIn('REQUIRED_ROLE_OR_SCOPE', lab.dispatch(fixture_bundle(scope='mock/other'))['details']) + self.assertEqual(lab.effects(), 0) + + def expired_grant(self, lab): + self.assertIn('grant_expired', lab.dispatch(fixture_bundle(valid_until=15))['details']) + self.assertEqual(lab.effects(), 0) + + def future_proof(self, lab): + self.assertIn('FUTURE_DATED_PROOF', lab.dispatch(fixture_bundle(signed_at=30))['details']) + self.assertEqual(lab.effects(), 0) + + def changed_action(self, lab): + request = fixture_bundle() + request['token']['command_id'] += 1 + request['action']['token']['command_id'] += 1 + self.assertIn('action_digest_mismatch', lab.dispatch(request)['details']) + self.assertEqual(lab.effects(), 0) + + def missing_authority(self, lab): + self.assertIn('MALFORMED_AUTHORITY', lab.dispatch({'token': {}})['details']) + self.assertEqual(lab.effects(), 0) + + def uncommitted_request(self, lab): + self.assertEqual(lab.dispatch(fixture_bundle(token(committed=False)))['reason'], 'UNCOMMITTED') + self.assertEqual(lab.effects(), 0) + + def foreign_lineage(self, lab): + self.assertEqual(lab.dispatch(fixture_bundle(token(effect_id=13)))['reason'], 'PERSISTENT_LINEAGE') + self.assertEqual(lab.effects(), 0) + + def forged_receipt(self, lab): + lab.dispatch(fixture_bundle()) + receipt = fixture_receipt(outcome='NOT_COMMITTED') + receipt['auth_tag'] ^= 1 + self.assertEqual(lab.reconcile(receipt)['auth'], 'AUTH_TAG') + self.assertEqual(lab.observation()['next_receipt_seq'], 0) + self.assertEqual(lab.dispatch(fixture_bundle(token(1)))['reason'], 'OUTCOME_UNKNOWN') + self.assertEqual(lab.effects(), 1) + + def foreign_device(self, lab): + lab.dispatch(fixture_bundle()) + self.assertEqual(lab.reconcile(fixture_receipt(device_id=61))['auth'], 'DEVICE_ID') + self.assertEqual(lab.observation()['outcome'], 'UNKNOWN') + self.assertEqual(lab.observation()['next_receipt_seq'], 0) + + def wrong_key_epoch(self, lab): + lab.dispatch(fixture_bundle()) + self.assertEqual(lab.reconcile(fixture_receipt(key_epoch=3))['auth'], 'KEY_EPOCH') + self.assertEqual(lab.observation()['next_receipt_seq'], 0) + + def wrong_receipt_sequence(self, lab): + lab.dispatch(fixture_bundle()) + self.assertEqual(lab.reconcile(fixture_receipt(seq=1))['auth'], 'RECEIPT_SEQUENCE') + self.assertEqual(lab.observation()['next_receipt_seq'], 0) + + def wrong_attempt_consumes_sequence(self, lab): + lab.dispatch(fixture_bundle()) + self.assertEqual(lab.reconcile(fixture_receipt(token(1)))['semantic'], 'ATTEMPT_MISMATCH') + self.assertEqual(lab.observation()['outcome'], 'UNKNOWN') + self.assertEqual(lab.observation()['next_receipt_seq'], 1) + self.assertTrue(lab.reconcile(fixture_receipt(seq=1))['applied']) + + def malformed_receipt(self, lab): + lab.dispatch(fixture_bundle()) + receipt = fixture_receipt() + receipt['outcome'] = 'TIMEOUT_MEANS_FAILURE' + self.assertEqual(lab.reconcile(receipt)['auth'], 'MALFORMED_RECEIPT') + self.assertEqual(lab.observation()['next_receipt_seq'], 0) + self.assertEqual(lab.observation()['outcome'], 'UNKNOWN') + + def historical_receipt_after_revocation(self, lab): + lab.dispatch(fixture_bundle()) + lab.context(generation=2, state_version='state/2') + fresh = fixture_bundle(token(1), generation=2, state_version='state/2') + self.assertEqual(lab.dispatch(fresh)['reason'], 'OUTCOME_UNKNOWN') + self.assertTrue(lab.reconcile(fixture_receipt())['applied']) + self.assertEqual(lab.dispatch(fresh)['reason'], 'TERMINAL_COMMITTED') + self.assertEqual(lab.effects(), 1) + + def explicit_negative_only_releases_successor(self, lab): + lab = crash(lab, 'after_commit', fixture_bundle()) # No effect happened. + self.assertTrue(lab.reconcile(fixture_receipt(outcome='NOT_COMMITTED'))['applied']) + self.assertEqual(lab.dispatch(fixture_bundle())['reason'], 'PERSISTENT_FRONTIER') + self.assertTrue(lab.dispatch(fixture_bundle(token(1)))['forwarded']) + self.assertTrue(lab.reconcile(fixture_receipt(token(1), seq=1))['applied']) + self.assertEqual(lab.effects(), 1) + + def conflict_holds(self, lab): + lab.dispatch(fixture_bundle()) + self.assertTrue(lab.reconcile(fixture_receipt(outcome='CONFLICT'))['applied']) + self.assertEqual(lab.dispatch(fixture_bundle(token(1)))['reason'], 'TERMINAL_CONFLICT') + self.assertEqual(lab.effects(), 1) + + def concurrent_dispatch(self, lab): + request = lab.root / 'race.json' + request.write_text(json.dumps(fixture_bundle())) + command = [sys.executable, str(HERE / 'proof.py'), 'dispatch', str(lab.root), + str(request), '--engine', lab.engine] + def run(): + out = subprocess.run(command, capture_output=True, text=True, timeout=15) + self.assertEqual(out.returncode, 0, out.stderr) + return json.loads(out.stdout) + with ThreadPoolExecutor(2) as pool: + results = list(pool.map(lambda _: run(), range(2))) + self.assertEqual(sum(r['forwarded'] for r in results), 1) + self.assertEqual(lab.effects(), 1) + + def missing_storage_fails_closed(self, lab): + lab.control.unlink() + with self.assertRaises(Exception): + lab.dispatch(fixture_bundle()) + self.assertFalse(lab.control.exists()) + self.assertEqual(lab.effects(), 0) + + +class RecoveryTests(unittest.TestCase, Cases): + pass + + +def bind(case, engine): + def run(self): + with tempfile.TemporaryDirectory() as folder: + lab = Lab(folder, engine) + lab.initialize() + case(self, lab) + return run + +for _name, _case in Cases.__dict__.items(): + if callable(_case) and not _name.startswith('_'): + for _engine in ('native', 'baseline'): + setattr(RecoveryTests, f'test_{_engine}_{_name}', bind(_case, _engine)) + + +class SourceAndDifferentialTests(unittest.TestCase): + def test_upstream_a6_regression(self): + result = A6.scenario_result() + self.assertTrue(result['terminal_committed']) + self.assertTrue(result['restart_replay_blocked']) + self.assertEqual(result['external_effect_count'], 1) + + def test_upstream_a7_exact_result_digest(self): + self.assertEqual(A7.scenario_result()['result_digest_sha256'], + '6781dbfbd1b529866709980a3a85a38bd37f505daaddd53fd7c8e106ab863d2f') + + def test_seeded_transition_equivalence(self): + random_source = random.Random(20260905) + # 24 independent traces x 12 actions = 288 compared state-machine steps. + # No claim of exhaustive checking. Shared authority and tag primitive; + # separate lifecycle dispatch/reconciliation implementations. + with tempfile.TemporaryDirectory() as root: + for trace in range(24): + native, baseline = [Lab(Path(root) / f'{trace}-{engine}', engine) + for engine in ('native', 'baseline')] + native.initialize() + baseline.initialize() + for step in range(12): + if random_source.randrange(2): + item = fixture_bundle(token(random_source.randrange(3))) + left, right = native.dispatch(item), baseline.dispatch(item) + else: + item = fixture_receipt(token(random_source.randrange(3)), + seq=random_source.randrange(3), + outcome=random_source.choice(['COMMITTED', 'CONFLICT', 'UNKNOWN'])) + if random_source.randrange(4) == 0: + item['auth_tag'] ^= 1 + left, right = native.reconcile(item), baseline.reconcile(item) + self.assertEqual(left, right, (trace, step)) + self.assertEqual(native.observation(), baseline.observation(), (trace, step)) + self.assertEqual(native.state()['store'], baseline.state()['store']) + self.assertEqual(native.state()['trust'], baseline.state()['trust']) + + +if __name__ == '__main__': + unittest.main(verbosity=2)