tricu

generation_io.py (3653B)

---

 """Declared generation framing, never inferred from execution or tree shape.
 
 Missing metadata denotes isolated historical v1. New records carry both versions:
 a transition can execute v1 while producing v2, using a separately selected output
 adapter. These are caller assertions bound to recorded artifact bytes, not a proof
 of arbitrary worker semantics or an authority sandbox.
 """
 import hashlib
 import argparse
 
 FIELDS = ('execution_io_version', 'produced_io_version', 'produced_adapter_sha256')
 
 
 def version(value):
     if type(value) is not int or value not in (1, 2):
         raise ValueError('IO version must be integer 1 or 2')
     return value
 
 
 def cli_version(text):
     if text not in ('1', '2'):
         raise argparse.ArgumentTypeError('IO version must be 1 or 2')
     return int(text)
 
 
 def versions(record):
     if not isinstance(record, dict):
         raise ValueError('invalid generation IO metadata')
     present = [key in record for key in FIELDS]
     if any(present) and not all(present):
         raise ValueError('incomplete generation IO metadata')
     if not any(present):
         return 1, 1
     pin = record['produced_adapter_sha256']
     if not isinstance(pin, str) or len(pin) != 64 or any(c not in '0123456789abcdef' for c in pin):
         raise ValueError('invalid produced adapter digest')
     return version(record['execution_io_version']), version(record['produced_io_version'])
 
 
 def produced_adapter(directory, record):
     versions(record)
     name = 'produced-io-adapter.arboricx' if 'produced_io_version' in record else 'io-adapter.arboricx'
     payload = (directory / name).read_bytes()
     expected = record.get('produced_adapter_sha256', record.get('adapter_sha256'))
     if hashlib.sha256(payload).hexdigest() != expected:
         raise ValueError('produced adapter identity mismatch')
     return payload
 
 
 def execution_flags(io_version, host='haskell'):
     value = version(io_version)
     if host not in ('haskell', 'haskell-ffi'):
         raise ValueError('selected execution host does not support IO')
     if host == 'haskell-ffi' and value != 2:
         raise ValueError('Haskell embedding requires IO version 2')
     return ['--unsafe-io'] if value == 1 else []
 
 
 def check_pair(first, second):
     _, output = versions(first)
     execution, reproduced = versions(second)
     if output != execution or output != reproduced:
         raise ValueError('generation IO version chain mismatch')
     a = first.get('produced_adapter_sha256', first['adapter_sha256'])
     b = second.get('produced_adapter_sha256', second['adapter_sha256'])
     if a != second['adapter_sha256'] or a != b:
         raise ValueError('generation adapter chain mismatch')
 
 
 def check_invocation(record, command, host='haskell'):
     execution, _ = versions(record)
     execution_flags(execution, host)
     if not isinstance(command, list) or not all(isinstance(arg, str) for arg in command):
         raise ValueError('invalid generation invocation')
     if host != 'haskell':
         return
     grants = {'--unsafe-io', '--allow-read', '--allow-write', '--allow-read-all', '--allow-write-all'}
     if execution == 2 and any(arg.split('=', 1)[0] in grants for arg in command):
         raise ValueError('invocation IO authority mismatch')
     if execution == 1 and '--unsafe-io' not in command:
         raise ValueError('invocation IO authority mismatch')
 
 
 def compiler_flags(io_version):
     # Retained v1 runtimes predate the selector. Their documented default is v1;
     # do not probe failure and retry. HostIO always requires the explicit new flag.
     return [] if version(io_version) == 1 else ['--toolchain-io-version', '2']