#!/usr/bin/env python3 """DN recovery baseline v0.1. Local LangGraph; deterministic nodes, no model/network. Run: python recovery_test.py run --repeats 10 --output results.json Install pinned dependencies from requirements-lock.txt in a fresh environment. """ import argparse import csv import hashlib import importlib.metadata import json import os from pathlib import Path import platform import sqlite3 import subprocess import sys import tempfile import uuid from datetime import datetime, timezone from typing import TypedDict # Disable optional tracing. This program makes no external calls. os.environ['LANGSMITH_TRACING'] = 'false' os.environ['LANGCHAIN_TRACING_V2'] = 'false' CASES = [ ('restart', 'memory-only'), ('restart', 'sqlite-checkpoint'), ('uncertain-outcome', 'new-key-on-retry'), ('uncertain-outcome', 'stable-key'), ('revocation', 'cached-allow'), ('revocation', 'receiver-recheck'), ] EXIT_INJECTED = 86 class State(TypedDict, total=False): task_id: str prepared: bool cached_allowed: bool result: str def connect(path): c = sqlite3.connect(path) c.execute('PRAGMA synchronous=FULL') return c def initialize(directory): with connect(directory / 'receiver.db') as c: c.executescript(''' CREATE TABLE effects (request_key TEXT PRIMARY KEY, task_id TEXT NOT NULL); CREATE TABLE policy (id INTEGER PRIMARY KEY, allowed INTEGER, epoch INTEGER); INSERT INTO policy VALUES (1,1,0); CREATE TABLE prep (id INTEGER PRIMARY KEY AUTOINCREMENT); ''') def worker(directory, scenario, variant, phase): from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.checkpoint.memory import InMemorySaver receiver = directory / 'receiver.db' def prepare(state): with connect(receiver) as c: c.execute('INSERT INTO prep DEFAULT VALUES') allowed = bool(c.execute('SELECT allowed FROM policy WHERE id=1').fetchone()[0]) return {'prepared': True, 'cached_allowed': allowed} def action(state): # A separate simulated receiver transaction commits the effect before the # worker can acknowledge/checkpoint success. No wallet or remote API. with connect(receiver) as c: c.execute('BEGIN IMMEDIATE') if scenario == 'revocation': if variant == 'receiver-recheck': allowed = bool(c.execute('SELECT allowed FROM policy WHERE id=1').fetchone()[0]) else: allowed = state['cached_allowed'] if not allowed: return {'result': 'blocked'} key = state['task_id'] if variant != 'new-key-on-retry' else str(uuid.uuid4()) c.execute('INSERT OR IGNORE INTO effects VALUES (?,?)', (key, state['task_id'])) if scenario == 'uncertain-outcome' and phase == 1: os._exit(EXIT_INJECTED) # committed effect; node has not returned return {'result': 'accepted'} builder = StateGraph(State) if scenario == 'uncertain-outcome': builder.add_node('action', action) builder.add_edge(START, 'action') else: builder.add_node('prepare', prepare) builder.add_node('action', action) builder.add_edge(START, 'prepare') builder.add_edge('prepare', 'action') builder.add_edge('action', END) checkpoint_conn = None if variant == 'memory-only': saver = InMemorySaver() else: # SqliteSaver serializes access; LangGraph persists from a worker thread. checkpoint_conn = sqlite3.connect(directory / 'checkpoints.db', check_same_thread=False) checkpoint_conn.execute('PRAGMA synchronous=FULL') saver = SqliteSaver(checkpoint_conn) graph = builder.compile(checkpointer=saver, interrupt_after=['prepare'] if scenario != 'uncertain-outcome' else None) config = {'configurable': {'thread_id': 'dn-synthetic-task'}} if phase == 1: graph.invoke({'task_id': 'synthetic-task-001'}, config, durability='sync') os._exit(EXIT_INJECTED) # paused after prepared state was checkpointed snapshot = graph.get_state(config) if not snapshot.values: outcome = {'result': 'state-unavailable', 'hasCheckpoint': False} else: result = graph.invoke(None, config, durability='sync') outcome = {'result': result.get('result'), 'hasCheckpoint': True} (directory / 'outcome.json').write_text(json.dumps(outcome)) if checkpoint_conn: checkpoint_conn.close() def run(repeats, output): if not 1 <= repeats <= 100: raise ValueError('repeats must be between 1 and 100') rows = [] with tempfile.TemporaryDirectory(prefix='dn-recovery-') as temp: for scenario, variant in CASES: for repeat in range(1, repeats + 1): directory = Path(temp) / f'{scenario}-{variant}-{repeat}' directory.mkdir() initialize(directory) command = [sys.executable, str(Path(__file__).resolve()), 'worker', '--directory', str(directory), '--scenario', scenario, '--variant', variant] before = subprocess.run(command + ['--phase', '1'], capture_output=True, text=True, timeout=20) if before.returncode != EXIT_INJECTED: raise RuntimeError(f'injection failed: {scenario}/{variant}: {before.returncode}: {before.stderr}') if scenario == 'revocation': with connect(directory / 'receiver.db') as c: c.execute('UPDATE policy SET allowed=0,epoch=1 WHERE id=1') after = subprocess.run(command + ['--phase', '2'], capture_output=True, text=True, timeout=20) if after.returncode != 0: raise RuntimeError(f'restart failed: {scenario}/{variant}: {after.stderr}') outcome = json.loads((directory / 'outcome.json').read_text()) with connect(directory / 'receiver.db') as c: effects = c.execute('SELECT count(*) FROM effects').fetchone()[0] prep = c.execute('SELECT count(*) FROM prep').fetchone()[0] target = 0 if scenario == 'revocation' else 1 accepted = (effects == target and outcome['result'] == ('blocked' if scenario == 'revocation' else 'accepted') and (scenario == 'uncertain-outcome' or prep == 1)) rows.append({'scenario': scenario, 'configuration': variant, 'repeat': repeat, 'injectedExitCode': before.returncode, 'restartExitCode': after.returncode, 'committedEffects': effects, 'preparationExecutions': prep, 'outcome': outcome['result'], 'checkpointAvailableAfterRestart': outcome['hasCheckpoint'], 'acceptanceCriterionMet': accepted}) summary = [] for scenario, variant in CASES: subset = [r for r in rows if r['scenario'] == scenario and r['configuration'] == variant] summary.append({'scenario': scenario, 'configuration': variant, 'trials': len(subset), 'criteriaMet': sum(r['acceptanceCriterionMet'] for r in subset), 'committedEffects': sum(r['committedEffects'] for r in subset), 'expectedEffectsPerTrial': 0 if scenario == 'revocation' else 1}) versions = {n: importlib.metadata.version(n) for n in ['langgraph', 'langgraph-checkpoint', 'langgraph-checkpoint-sqlite', 'langchain-core']} data = {'name': 'DN Runtime Recovery Local Baseline', 'version': '0.1', 'executedAt': datetime.now(timezone.utc).isoformat(), 'evidenceType': 'local-framework-execution-with-synthetic-tools', 'framework': {'name': 'LangGraph', 'packages': versions}, 'environment': {'python': platform.python_version(), 'os': platform.system(), 'architecture': platform.machine(), 'sqlite': sqlite3.sqlite_version}, 'sourceSha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), 'repeatsPerConfiguration': repeats, 'trialCount': len(rows), 'modelCalls': 0, 'managedProviderRuns': 0, 'realTransactions': 0, 'latencyBenchmark': None, 'productionFailureRate': None, 'limitations': [ 'Deterministic synthetic workload; intentionally weak application controls are not vendor defaults.', 'One local environment, one framework version, serial runs and selected injection points.', 'Separate local SQLite receiver database; no real remote service, model, wallet or cloud deployment.', 'Abrupt process exit is real; OS crash, power loss, network partition and concurrent revocation are not tested.', 'Receiver recheck and effect share one local transaction; arbitrary external tools may not support this boundary.', 'Repeated deterministic cases are not independent evidence of a population failure rate or security certification.' ], 'summary': summary, 'trials': rows} output.write_text(json.dumps(data, indent=2) + '\n') with output.with_suffix('.csv').open('w', newline='') as f: writer = csv.DictWriter(f, fieldnames=list(rows[0])); writer.writeheader(); writer.writerows(rows) print(json.dumps({'trialCount': len(rows), 'summary': summary, 'versions': versions}, indent=2)) if __name__ == '__main__': p = argparse.ArgumentParser(description=__doc__) subs = p.add_subparsers(dest='command', required=True) r = subs.add_parser('run'); r.add_argument('--repeats', type=int, default=10) r.add_argument('--output', type=Path, default=Path('results.json')) w = subs.add_parser('worker'); w.add_argument('--directory', type=Path, required=True) w.add_argument('--scenario', choices=sorted({c[0] for c in CASES}), required=True) w.add_argument('--variant', required=True); w.add_argument('--phase', type=int, choices=[1,2], required=True) args = p.parse_args() if args.command == 'run': run(args.repeats, args.output) else: worker(args.directory, args.scenario, args.variant, args.phase)