#!/usr/bin/env python3 """Bounded local mechanism experiment; never touches the checkout's Git refs.""" import argparse import hashlib import json import os from pathlib import Path import selectors import shutil import signal import sqlite3 import subprocess import sys import tempfile import time from datetime import datetime, timezone HERE = Path(__file__).resolve() def connect(path): db = sqlite3.connect(path) assert db.execute('PRAGMA journal_mode=WAL').fetchone()[0] == 'wal' db.execute('PRAGMA synchronous=FULL') assert db.execute('PRAGMA synchronous').fetchone()[0] == 2 return db def run(*args, cwd=None, check=True, input=None): # Ignore user Git configuration/hooks and inherited repository targeting. env = {k: v for k, v in os.environ.items() if not k.startswith('GIT_')} env.update(GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null', GIT_AUTHOR_NAME='Recovery Fixture', GIT_AUTHOR_EMAIL='fixture@example.invalid', GIT_COMMITTER_NAME='Recovery Fixture', GIT_COMMITTER_EMAIL='fixture@example.invalid', GIT_AUTHOR_DATE='2026-09-28T00:00:00Z', GIT_COMMITTER_DATE='2026-09-28T00:00:00Z') return subprocess.run(args, cwd=cwd, env=env, text=True, input=input, capture_output=True, check=check, timeout=20) def checkpoint(): print('checkpoint', flush=True) # Parent kills only this dedicated fixture subprocess, after its marker. signal.pause() def worker(mode, root): db = connect(root / 'local.db') if mode == 'send': oid = db.execute("SELECT oid FROM pending WHERE id='send'").fetchone()[0] run('git', 'push', str(root / 'receiver.git'), f'{oid}:refs/heads/main', cwd=root / 'client') checkpoint() # Accepted remotely, local acknowledgement deliberately absent. return db.execute('BEGIN IMMEDIATE') db.execute("UPDATE pending SET state='superseded' WHERE id='old'") if mode == 'split': db.commit() # Deliberately broken negative control. checkpoint() return db.execute("UPDATE document SET body='new'") if mode == 'middle': checkpoint() return db.execute("INSERT INTO pending VALUES ('new','new-identity','pending')") if mode == 'before': checkpoint() return db.commit() checkpoint() def kill_at(mode, root): p = subprocess.Popen([sys.executable, str(HERE), '--worker', mode, '--root', str(root)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) try: with selectors.DefaultSelector() as selector: selector.register(p.stdout, selectors.EVENT_READ) if not selector.select(timeout=20): raise RuntimeError(f'{mode}: checkpoint timeout') line = p.stdout.readline().strip() if line != 'checkpoint': raise RuntimeError(f'{mode}: no checkpoint: {line} {p.stderr.read()}') p.kill() p.communicate(timeout=5) assert p.returncode == -signal.SIGKILL return p.returncode finally: if p.poll() is None: p.kill() p.communicate(timeout=5) def initialize(root): root.mkdir() with connect(root / 'local.db') as db: db.executescript('CREATE TABLE document(body TEXT); ' 'CREATE TABLE pending(id TEXT PRIMARY KEY, oid TEXT, state TEXT); ' "INSERT INTO document VALUES ('old'); " "INSERT INTO pending VALUES ('old','old-identity','pending');") def state(root): with connect(root / 'local.db') as db: assert db.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' return {'body': db.execute('SELECT body FROM document').fetchone()[0], 'pending': db.execute('SELECT * FROM pending ORDER BY id').fetchall()} def sha(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def experiment(report): with tempfile.TemporaryDirectory(prefix='fep-storage-receiver-') as temp: root = Path(temp) old = {'body': 'old', 'pending': [('old', 'old-identity', 'pending')]} new = {'body': 'new', 'pending': [('new', 'new-identity', 'pending'), ('old', 'old-identity', 'superseded')]} for mode, expected in [('before', old), ('after', new), ('middle', old), ('split', {'body': 'old', 'pending': [('old', 'old-identity', 'superseded')]})]: case = root / mode initialize(case) code = kill_at(mode, case) observed = state(case) report['checks'].append({'id': f'local-{mode}', 'pass': observed == expected, 'negative_control': mode == 'split', 'worker_exit': code, 'observed': observed}) assert observed == expected case = root / 'receiver' initialize(case) client = case / 'client' receiver = case / 'receiver.git' run('git', 'init', '--quiet', '--initial-branch=main', str(client)) run('git', 'init', '--quiet', '--bare', '--initial-branch=main', str(receiver)) run('git', 'config', 'receive.denyNonFastForwards', 'true', cwd=receiver) tree = run('git', 'mktree', cwd=client, input='').stdout.strip() def commit(parent, text): args = ['git', 'commit-tree', tree] if parent: args += ['-p', parent] return run(*args, cwd=client, input=text + '\n').stdout.strip() def push(oid, force=False): return run('git', 'push', '--porcelain', str(receiver), ('+' if force else '') + f'{oid}:refs/heads/main', cwd=client, check=False) def tip(): return run('git', 'rev-parse', 'refs/heads/main', cwd=receiver).stdout.strip() base = commit(None, 'base') accepted = commit(base, 'operation A: synthetic intent') assert push(base).returncode == 0 with connect(case / 'local.db') as db: db.execute("INSERT INTO pending VALUES ('send',?,'pending')", (accepted,)) code = kill_at('send', case) with connect(case / 'local.db') as db: pending = db.execute("SELECT state FROM pending WHERE id='send'").fetchone()[0] report['checks'].append({'id': 'lost-local-ack', 'pass': tip() == accepted and pending == 'pending', 'worker_exit': code, 'receiver_tip': tip(), 'outbox_state': pending}) assert tip() == accepted and pending == 'pending' replay = push(accepted) count = int(run('git', 'rev-list', '--count', 'main', cwd=receiver).stdout) report['checks'].append({'id': 'same-identity-replay', 'pass': replay.returncode == 0 and count == 2, 'exit': replay.returncode, 'history_count': count, 'up_to_date': 'up to date' in replay.stdout}) assert replay.returncode == 0 and count == 2 with connect(case / 'local.db') as db: db.execute("UPDATE pending SET state='acknowledged' WHERE id='send'") concurrent = commit(accepted, 'operation B') assert push(concurrent).returncode == 0 stale = commit(base, 'stale operation C') with connect(case / 'local.db') as db: db.execute("INSERT INTO pending VALUES ('stale',?,'pending')", (stale,)) rejected = push(stale, force=True) # Force transmission; real server denies it. with connect(case / 'local.db') as db: pending = db.execute("SELECT state FROM pending WHERE id='stale'").fetchone()[0] server_refused = 'non-fast-forward' in rejected.stderr and 'remote rejected' in rejected.stdout report['checks'].append({'id': 'stale-ancestry', 'pass': rejected.returncode != 0 and server_refused and tip() == concurrent and pending == 'pending', 'exit': rejected.returncode, 'server_refused': server_refused, 'receiver_tip_unchanged': tip() == concurrent, 'outbox_state': pending}) assert report['checks'][-1]['pass'] run('git', 'gc', '--prune=now', cwd=receiver) retained = run('git', 'cat-file', '-t', accepted, cwd=receiver).stdout.strip() ancestor = run('git', 'merge-base', '--is-ancestor', accepted, 'main', cwd=receiver, check=False) report['checks'].append({'id': 'reachable-identity-after-gc', 'pass': retained == 'commit' and ancestor.returncode == 0}) assert report['checks'][-1]['pass'] # The same business intent under a NEW identity is another accepted commit. fresh = commit(concurrent, 'operation A: synthetic intent') assert push(fresh).returncode == 0 count = int(run('git', 'rev-list', '--count', 'main', cwd=receiver).stdout) report['checks'].append({'id': 'fresh-identity-duplicates-intent', 'negative_control': True, 'pass': fresh != accepted and count == 4, 'history_count': count}) assert report['checks'][-1]['pass'] report['identities'] = {'base': base, 'accepted': accepted, 'concurrent': concurrent, 'stale': stale, 'fresh': fresh} def main(): parser = argparse.ArgumentParser() parser.add_argument('--worker', choices=['before', 'after', 'middle', 'split', 'send']) parser.add_argument('--root', type=Path) parser.add_argument('--output', type=Path) args = parser.parse_args() if args.worker: worker(args.worker, args.root) return if not args.output: parser.error('--output is required (use a new file to preserve prior receipts)') # Reserve the receipt before execution and refuse to overwrite retained evidence. with args.output.open('x') as output: start = time.monotonic() import _sqlite3 report = {'started_utc': datetime.now(timezone.utc).isoformat(), 'checks': [], 'provenance': {'python': sys.version.split()[0], 'sqlite': sqlite3.sqlite_version, 'sqlite_source_id': sqlite3.connect(':memory:').execute('SELECT sqlite_source_id()').fetchone()[0], 'git': run('git', '--version').stdout.strip(), 'git_binary_sha256': sha(shutil.which('git')), 'sqlite_python_extension_sha256': sha(_sqlite3.__file__), 'experiment_sha256': sha(HERE)}, 'settings': {'sqlite_journal': 'wal', 'sqlite_synchronous': 'FULL', 'receiver': 'local bare git receive-pack', 'denyNonFastForwards': True}, 'scope': 'Process kills and discarded local acknowledgement; no power-loss or network-loss simulation.'} try: experiment(report) report['status'] = 'passed' except Exception as exc: report['status'] = 'failed' report['error'] = f'{type(exc).__name__}: {exc}' raise finally: report['execution_seconds'] = round(time.monotonic() - start, 6) json.dump(report, output, indent=2) output.write('\n') print(f"{len(report['checks'])} checks: {report['status']}") if __name__ == '__main__': main()