#!/usr/bin/env python3 """Original deterministic LangGraph/SQLite exercise. No model, service or API. Install the pinned requirements first, then: python langgraph_hands_on.py run --output ./fresh-langgraph-run The output directory must not exist. Each action runs in a fresh Python process. SQLite effects are disposable fixtures; this is not a publishing integration. """ import argparse import dataclasses from datetime import datetime, timezone import hashlib import importlib.metadata import json import os from pathlib import Path import platform import resource import sqlite3 import subprocess import sys import time from typing import TypedDict def dump(value): return json.dumps(value, ensure_ascii=False, indent=2, default=lambda x: dataclasses.asdict(x) if dataclasses.is_dataclass(x) else str(x)) class State(TypedDict, total=False): job_id: str revision: int text: str mode: str decision: str approved_revision: int | None receipt_id: str | None def sink_connection(root): db = sqlite3.connect(root / 'sink.sqlite') db.executescript(''' CREATE TABLE IF NOT EXISTS attempts (id INTEGER PRIMARY KEY, action_id TEXT); CREATE TABLE IF NOT EXISTS receipts (id INTEGER PRIMARY KEY, action_id TEXT, receipt_id TEXT); CREATE TABLE IF NOT EXISTS unique_receipts (action_id TEXT PRIMARY KEY, receipt_id TEXT); CREATE TABLE IF NOT EXISTS gate_markers (id INTEGER PRIMARY KEY, job_id TEXT); ''') return db def counts(root, job): action = job + '/revision-1/save' with sink_connection(root) as db: result = {name: db.execute('SELECT count(*) FROM ' + name + ' WHERE action_id=?', (action,)).fetchone()[0] for name in ('attempts', 'receipts', 'unique_receipts')} result['gate_markers'] = db.execute('SELECT count(*) FROM gate_markers WHERE job_id=?', (job,)).fetchone()[0] return result def worker(args): began = time.perf_counter() blocked_network_events = [] def audit(event, _args): if event in ('socket.connect', 'socket.getaddrinfo'): blocked_network_events.append(event) raise RuntimeError('Network forbidden in deterministic exercise') sys.addaudithook(audit) os.environ['LANGGRAPH_STRICT_MSGPACK'] = 'true' os.environ['LANGSMITH_TRACING'] = 'false' os.environ['LANGCHAIN_TRACING_V2'] = 'false' from langgraph.graph import StateGraph, START, END from langgraph.types import Command, interrupt from langgraph.checkpoint.sqlite import SqliteSaver root = args.output.resolve(strict=True) def prepare(state): return {'decision': 'pending', 'approved_revision': None, 'receipt_id': None} def review(state): if state['mode'] == 'before-interrupt': with sink_connection(root) as db: db.execute('INSERT INTO gate_markers(job_id) VALUES (?)', (state['job_id'],)) response = interrupt({'draft_id': state['job_id'], 'revision': state['revision'], 'draft_text': state['text'], 'operation': 'save to isolated SQLite sink'}) approved = response.get('decision') == 'approve' and response.get('revision') == state['revision'] return {'decision': 'approved' if approved else 'rejected', 'approved_revision': state['revision'] if approved else None} def save(state): if state['decision'] != 'approved' or state['approved_revision'] != state['revision']: raise RuntimeError('Fixture revision guard rejected the write') action = state['job_id'] + '/revision-' + str(state['revision']) + '/save' receipt = 'receipt:' + action with sink_connection(root) as db: db.execute('INSERT INTO attempts(action_id) VALUES (?)', (action,)) if state['mode'] == 'deduplicated-failure': db.execute('INSERT INTO unique_receipts(action_id,receipt_id) VALUES (?,?) ON CONFLICT(action_id) DO NOTHING', (action, receipt)) else: db.execute('INSERT INTO receipts(action_id,receipt_id) VALUES (?,?)', (action, receipt)) marker = root / (state['job_id'] + '.failure-injected') if state['mode'] in ('unsafe-failure', 'deduplicated-failure') and not marker.exists(): marker.write_text('Synthetic exception after committed sink write, before node return.\n') raise RuntimeError('injected_after_sink_commit_before_node_return') return {'receipt_id': receipt} build_start = time.perf_counter() with SqliteSaver.from_conn_string(str(root / 'checkpoints.sqlite')) as saver: builder = StateGraph(State) builder.add_node('prepare', prepare) builder.add_node('review', review) builder.add_node('save', save) builder.add_edge(START, 'prepare') builder.add_edge('prepare', 'review') builder.add_conditional_edges('review', lambda state: 'save' if state['decision'] == 'approved' else END) builder.add_edge('save', END) graph = builder.compile(checkpointer=saver) compile_ms = (time.perf_counter() - build_start) * 1000 config = {'configurable': {'thread_id': args.job}} def snapshot(): state = graph.get_state(config) return {'values': state.values, 'next': list(state.next), 'interrupts': [dataclasses.asdict(i) for task in state.tasks for i in task.interrupts]} before = snapshot() value = {'job_id': args.job, 'revision': 1, 'text': 'Fictional research draft. No customer or publication destination.', 'mode': args.mode} if args.action == 'start' else ( Command(resume={'decision': args.decision, 'revision': args.review_revision}) if args.action == 'resume' else None) invoke_start = time.perf_counter() error = None result = None try: result = graph.invoke(value, config, durability='sync') except Exception as exc: error = {'type': type(exc).__name__, 'message': str(exc)} invoke_ms = (time.perf_counter() - invoke_start) * 1000 after = snapshot() graph_class = type(graph).__module__ + '.' + type(graph).__name__ record = {'job': args.job, 'action': args.action, 'mode': args.mode, 'decision_fixture': args.decision if args.action == 'resume' else None, 'review_revision_fixture': args.review_revision if args.action == 'resume' else None, 'process_id': os.getpid(), 'compiled_graph_class': graph_class, 'before': before, 'result': result, 'error': error, 'after': after, 'sink': counts(root, args.job), 'compile_ms': round(compile_ms, 3), 'invoke_ms': round(invoke_ms, 3), 'worker_elapsed_ms': round((time.perf_counter()-began)*1000, 3), 'peak_rss_bytes': resource.getrusage(resource.RUSAGE_SELF).ru_maxrss * (1 if sys.platform == 'darwin' else 1024), 'blocked_network_events': blocked_network_events} print(dump(record)) def run(args): root = args.output.resolve() root.mkdir(parents=True, exist_ok=False) script = Path(__file__).resolve() clean_env = {'PATH': os.environ.get('PATH', ''), 'LANG': 'C.UTF-8', 'PYTHONDONTWRITEBYTECODE': '1', 'PYTHONNOUSERSITE': '1', 'LANGSMITH_TRACING': 'false', 'LANGCHAIN_TRACING_V2': 'false', 'LANGGRAPH_STRICT_MSGPACK': 'true'} fixtures = [ ('approve', 'ordinary', [('start', ''), ('resume', 'approve'), ('resume', 'approve')]), ('reject', 'ordinary', [('start', ''), ('resume', 'reject')]), ('wrong-revision', 'ordinary', [('start', ''), ('resume', 'approve')]), ('before-interrupt', 'before-interrupt', [('start', ''), ('resume', 'approve')]), ('unsafe', 'unsafe-failure', [('start', ''), ('resume', 'approve'), ('retry', '')]), ('deduplicated', 'deduplicated-failure', [('start', ''), ('resume', 'approve'), ('retry', '')]), ] records = [] started = time.perf_counter() for job, mode, actions in fixtures: for index, (action, decision) in enumerate(actions): command = [sys.executable, str(script), 'worker', '--output', str(root), '--job', job, '--mode', mode, '--action', action, '--decision', decision, '--review-revision', '2' if job == 'wrong-revision' else '1'] result = subprocess.run(command, env=clean_env, capture_output=True, text=True, timeout=45) (root / (job + '-' + str(index) + '.stderr.txt')).write_text(result.stderr) if result.returncode: raise RuntimeError('Worker failed for ' + job + ': inspect retained stderr.') record = json.loads(result.stdout) (root / (job + '-' + str(index) + '.json')).write_text(dump(record) + '\n') records.append(record) by_job = {job: [record for record in records if record['job'] == job] for job, _, _ in fixtures} checks = [] def check(name, value): checks.append({'check': name, 'pass': bool(value)}) check('genuine compiled LangGraph used', all(r['compiled_graph_class'] == 'langgraph.graph.state.CompiledStateGraph' for r in records)) check('all initial calls pause at review with zero receipts', all(rs[0]['after']['next'] == ['review'] and rs[0]['after']['interrupts'] and rs[0]['sink']['receipts'] == rs[0]['sink']['unique_receipts'] == 0 for rs in by_job.values())) a = by_job['approve'] check('approved matching revision produces one receipt', a[1]['after']['values']['decision'] == 'approved' and a[1]['after']['values']['approved_revision'] == 1 and a[1]['sink']['receipts'] == 1 and a[1]['error'] is None) check('resume uses a new OS process and persisted pending state', a[0]['process_id'] != a[1]['process_id'] and a[1]['before']['values']['job_id'] == 'approve' and a[1]['before']['next'] == ['review']) check('duplicate resume after completion adds no second sink write in this graph', a[2]['sink']['receipts'] == a[1]['sink']['receipts'] == 1 and a[2]['sink']['attempts'] == a[1]['sink']['attempts'] == 1 and not a[2]['after']['next']) check('rejection ends with zero sink effects', by_job['reject'][-1]['after']['values']['decision'] == 'rejected' and by_job['reject'][-1]['sink']['attempts'] == 0 and not by_job['reject'][-1]['after']['next']) check('application revision mismatch is rejected with zero sink effects', by_job['wrong-revision'][-1]['after']['values']['decision'] == 'rejected' and by_job['wrong-revision'][-1]['sink']['attempts'] == 0) check('pre-interrupt local marker is repeated on resume', [r['sink']['gate_markers'] for r in by_job['before-interrupt']] == [1, 2]) for job in ('unsafe', 'deduplicated'): rs = by_job[job] check(job + ': injected failure occurs after first committed write', rs[1]['error'] is not None and rs[1]['error']['message'] == 'injected_after_sink_commit_before_node_return' and rs[1]['sink']['attempts'] == 1) check(job + ': restart recovery reruns unfinished save node', rs[2]['process_id'] != rs[1]['process_id'] and rs[2]['before']['next'] == ['save'] and rs[2]['sink']['attempts'] == 2 and rs[2]['error'] is None and not rs[2]['after']['next']) check('unsafe sink exposes duplicate effects', by_job['unsafe'][-1]['sink']['receipts'] == 2) check('sink-owned unique action key prevents duplicate accepted rows', by_job['deduplicated'][-1]['sink']['unique_receipts'] == 1 and by_job['deduplicated'][-1]['sink']['attempts'] == 2) check('no network connect or DNS attempts in worker execution', all(not r['blocked_network_events'] for r in records)) versions = {name: importlib.metadata.version(name) for name in ('langgraph','langgraph-checkpoint','langgraph-checkpoint-sqlite','langchain-core')} output = {'schema':'urdream-langgraph-hands-on-v1', 'status':'PASS' if all(c['pass'] for c in checks) else 'FAIL', 'observed_at':datetime.now(timezone.utc).isoformat(), 'versions':versions, 'environment':{'python':platform.python_version(), 'platform':platform.system(), 'release':platform.release(), 'machine':platform.machine(), 'sqlite':sqlite3.sqlite_version}, 'script_sha256':hashlib.sha256(script.read_bytes()).hexdigest(), 'process_actions':len(records), 'elapsed_seconds':round(time.perf_counter()-started,3), 'checks_passed':sum(c['pass'] for c in checks), 'checks_total':len(checks), 'checks':checks, 'records':records, 'limits':['Deterministic local Python nodes, not a model/agent quality benchmark.', 'Independent subprocess reopening tests process-exit recovery, not OS power loss or multi-host concurrency.', 'Duplicate resume observation is limited to this completed graph and pinned versions.', 'SQLite unique-key mitigation is application/sink behavior, not a framework exactly-once guarantee.', 'Injected error is an ordinary exception after a committed local write, not a SIGKILL or remote service failure.']} (root/'results.json').write_text(dump(output)+'\n') print(dump({k:v for k,v in output.items() if k!='records'})) if output['status'] != 'PASS': raise SystemExit(1) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('command', choices=['run','worker']) parser.add_argument('--output', type=Path, required=True) parser.add_argument('--job', default='approve') parser.add_argument('--mode', default='ordinary') parser.add_argument('--action', choices=['start','resume','retry'], default='start') parser.add_argument('--decision', default='approve') parser.add_argument('--review-revision', type=int, default=1) args = parser.parse_args() (run if args.command == 'run' else worker)(args) if __name__ == '__main__': main()