"""Explicit isolated TLC execution; normal documentation builds never call this.""" from __future__ import annotations from datetime import datetime, timezone import json from pathlib import Path import shutil import subprocess import tempfile import time from .contracts import SourceInputs, ValidationError, checked_path, load_inputs, sha256, read_json from .evidence import classify_execution from .sany import TOOL_SHA256 TOOLS = Path(__file__).resolve().parents[2] / 'tools/tlc' RUNNER_SOURCE = Path(__file__).read_bytes() def execute_model(inputs: SourceInputs, model: dict, destination: Path, jar: Path) -> dict: destination = checked_path(destination.parent, destination.name) if destination.exists(): raise ValidationError('Attempt destination already exists; choose a new run ID') if sha256(jar.read_bytes()) != TOOL_SHA256: raise ValidationError('TLA+ tool SHA-256 hash mismatch') fresh = load_inputs(inputs.path) if fresh.sha256 != inputs.sha256 or model not in fresh.models: raise ValidationError('Input/model identity changed before execution') destination.parent.mkdir(parents=True, exist_ok=True) staging = Path(tempfile.mkdtemp(prefix='.formal-attempt-', dir=destination.parent)) run_id = destination.name if not run_id or any(c not in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_' for c in run_id): shutil.rmtree(staging) raise ValidationError('Unsafe run identity') receipt = {'schema_version': 1, 'kind': 'formal-check-execution', 'run_id': run_id, 'package_id': inputs.package_id, 'input_revision': inputs.revision, 'input_manifest_sha256': inputs.sha256, 'model_id': model['id'], 'tool_sha256': TOOL_SHA256, 'started_at': datetime.now(timezone.utc).isoformat(), 'admitted': False, 'actual_exit': None} # Retain the source loaded by this process, even if an editor later changes # the on-disk module before another run in the same matrix. tool_source=RUNNER_SOURCE (staging/'execution-tool.py').write_bytes(tool_source) receipt['runner_sha256']=sha256(tool_source) error = None try: with tempfile.TemporaryDirectory(prefix='formal-wiki-model-') as temp: work = Path(temp) (work / 'tools').mkdir() shutil.copy2(jar, work / 'tools/tla2tools.jar') if sha256((work/'tools/tla2tools.jar').read_bytes())!=TOOL_SHA256: raise ValidationError('Copied TLA+ tool hash mismatch') (work / 'adapter').mkdir() (work / 'adapter-src').mkdir() adapter_rows = [] for source in sorted(TOOLS.iterdir()): if source.suffix not in {'.java', '.tla'}: continue target = work / ('adapter-src' if source.suffix == '.java' else '') / source.name shutil.copy2(source, target) adapter_rows.append({'path': source.name, 'sha256': sha256(target.read_bytes())}) retained_adapter=staging/'adapter-src'/source.name retained_adapter.parent.mkdir(parents=True,exist_ok=True) retained_adapter.write_bytes(target.read_bytes()) receipt['adapter_sources'] = adapter_rows (staging / 'inputs').mkdir() (staging / 'inputs/inputs.json').write_bytes(inputs.path.read_bytes()) for source in inputs.files: if Path(source.path).name == 'FormalWikiTrace.tla': raise ValidationError('Package cannot replace the registered capture module') raw = checked_path(inputs.root, source.path).read_bytes() if sha256(raw) != source.sha256: raise ValidationError('Input hash changed while preparing execution') target = checked_path(work, source.path) target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(raw) retained = checked_path(staging / 'inputs', source.path) retained.parent.mkdir(parents=True, exist_ok=True) retained.write_bytes(raw) compile_command = ['javac', '-J-XX:+PerfDisableSharedMem', '-J-XX:ActiveProcessorCount=2', '-cp', 'tools/tla2tools.jar', '-d', 'adapter', 'adapter-src/FormalWikiTrace.java', 'adapter-src/FormalWikiOverrides.java'] compilation = subprocess.run(compile_command, cwd=work, text=True, capture_output=True, timeout=30) (staging / 'adapter-compile.stdout').write_text(compilation.stdout) (staging / 'adapter-compile.stderr').write_text(compilation.stderr) receipt['compile_command'] = compile_command if compilation.returncode: raise ValidationError(f'Capture adapter compilation failed: {compilation.stderr[-1000:]}') receipt['adapter_classes'] = [{'path': p.relative_to(work / 'adapter').as_posix(), 'sha256': sha256(p.read_bytes())} for p in sorted((work / 'adapter').rglob('*.class'))] runtime = subprocess.run(['java', '-XX:+PerfDisableSharedMem', '-version'], cwd=work, text=True, capture_output=True, timeout=10) receipt['java_identity'] = runtime.stdout + runtime.stderr heap_mib=model.get('bounds',{}).get('heap_mib',512) command = ['java', '-XX:+PerfDisableSharedMem', '-XX:+UseParallelGC', '-XX:ActiveProcessorCount=2', f'-Xmx{heap_mib}m', '-Xss8m', '-DTLA-Library=.', '-Dformalwiki.capture=capture.json', f'-Dformalwiki.run={run_id}', f'-Dformalwiki.input={inputs.sha256}', '-Dtlc2.overrides.TLCOverrides=FormalWikiOverrides', '-cp', 'tools/tla2tools.jar:adapter', 'tlc2.TLC', '-workers', '1', '-seed', '1', '-fp', '0', '-noGenerateSpecTE', # Only textual state-value printing is suppressed. The complete # typed CounterExample is retained by the postcondition adapter. '-suppressMessages', '2216,2217,2218', '-metadir', 'states', '-postCondition', 'FormalWikiTrace!Capture', '-config', model['config'], model['module']] receipt['command'] = command started = time.monotonic() try: result = subprocess.run(command, cwd=work, text=True, capture_output=True, timeout=min(model.get('max_seconds', 90), 300)) stdout, stderr = result.stdout, result.stderr receipt['actual_exit'] = result.returncode except subprocess.TimeoutExpired as expired: stdout = expired.stdout or b'' stderr = expired.stderr or b'' stdout = stdout.decode(errors='replace') if isinstance(stdout, bytes) else stdout stderr = stderr.decode(errors='replace') if isinstance(stderr, bytes) else stderr receipt['failure'] = 'timeout-incomplete' receipt['elapsed_seconds'] = round(time.monotonic() - started, 6) (staging / 'stdout.log').write_text(stdout) (staging / 'stderr.log').write_text(stderr) capture = {} if (work / 'capture.json').is_file(): shutil.copy2(work / 'capture.json', staging / 'capture.json') capture = read_json(staging / 'capture.json') if load_inputs(inputs.path).sha256 != inputs.sha256: raise ValidationError('Input manifest changed during execution') classification = classify_execution(model, receipt['actual_exit'], stdout + '\n' + stderr, capture, run_id, inputs.sha256) receipt['classification'] = {k: v for k, v in classification.items() if k != 'trace'} receipt['admitted'] = True except (ValidationError, OSError, subprocess.SubprocessError) as rejected: error = ValidationError(str(rejected)) receipt['admission_error'] = str(rejected) try: receipt['artifacts'] = [{'path': p.relative_to(staging).as_posix(), 'sha256': sha256(p.read_bytes())} for p in sorted(staging.rglob('*')) if p.is_file()] (staging / 'execution.json').write_text(json.dumps(receipt, indent=2) + '\n') staging.rename(destination) finally: if staging.exists(): shutil.rmtree(staging) if error: raise error return receipt