"""Versioned Rust-friendly shadow records from retained debugger journals. Exports preserve raw journal bytes and all rejected/uncertain operations. A journal snapshot may end while a window is open; metadata says so without inventing an end. """ import hashlib import json import re from pathlib import Path import time from .debug_journal import durable_json, encode, verify KINDS={'window_start','episode_start','observation','controller_decision','operation_result', 'action_transition','attempt','contact','stream_gap','window_end','episode_end'} def observed(value, hook, stage, source_hash=None, reason=None): return dict(availability='observed' if reason is None else 'unavailable',value=value if reason is None else None, reason=reason,source=dict(hook=hook,stage=stage,source_hash=source_hash),observation_ref=None) def records(rows): owner=next((r['payload'] for r in rows if r['kind']=='owner'),{}) session=owner.get('session_id','unavailable'); process=owner.get('process'); build=owner.get('build') window=None; operation=None; generation=None; emitted=set() def record(row,kind,payload,native=None,suffix=''): raw=native or {}; data=raw.get('data'); data=data if isinstance(data,dict) else {} tick=raw.get('tick',data.get('tick',data.get('ticks'))) eng=raw.get('engine_time',raw.get('time',data.get('engine_time'))) stage=raw.get('stage','host_boundary') return dict(schema_version=1,kind=kind,record_id=f"{session}:{row['sequence']}:{kind}{suffix}", session_id=session,process=process,build=build,generation=raw.get('generation',generation), operation_id=operation,window_id=window,episode_id=None, clock=dict(tick=observed(tick,'native',stage,source_hash=owner.get('bridge_sha256'),reason=None if tick is not None else 'not_exposed'), engine_seconds=observed(eng,'native',stage,source_hash=owner.get('bridge_sha256'),reason=None if eng is not None else 'not_exposed'), host_monotonic_ns=row['host_monotonic_ns'],host_clock_id=owner.get('host_clock_id','unrecorded'),hook_stage=stage), stream=kind,sequence=row['sequence'],payload=payload,raw_ref=dict(path='journal.jsonl',sequence=row['sequence'])) for row in rows: p=row['payload'] if row['kind']=='intent': operation=p['operation_id'] elif row['kind']=='native_reply': generation=p.get('generation',generation) yield record(row,'observation',dict(native=p),p) data=p.get('data') if isinstance(data,dict) and isinstance(data.get('events'),list): for e in data['events']: key=(p.get('session'),e.get('generation',generation),e['sequence']) if key in emitted: continue emitted.add(key) if e.get('kind') in ('attempt','contact'): yield record(row,e['kind'],e,dict(e,generation=e.get('generation',generation)),str(e['sequence'])) elif row['kind']=='result': operation=p['operation_id'] result=p.get('result',{}) if p['kind']=='record-start' and p['status']=='completed': window=result.get('window_id',p['operation_id']) yield record(row,'window_start',dict(window_id=window)) yield record(row,'operation_result',p) collections=[] if isinstance(result,dict): if isinstance(result.get('events'),dict): collections.append(result['events']) for block in result.get('chunks',[]): collections.append(block['events']) for n,collection in enumerate(collections): for j,error in enumerate(collection.get('errors',[])): yield record(row,'stream_gap',dict(stream='combat_event',generation=collection['generation'],first=None,last=None,reason=error),suffix=f':{n}:error:{j}') for j,gap in enumerate(collection.get('gaps',[])): yield record(row,'stream_gap',dict(stream='combat_event',generation=collection['generation'],**gap),suffix=f':{n}:{j}') if p['kind'] in ('record-stop','close') and p['status']=='completed' and window is not None: yield record(row,'window_end',dict(window_id=window)); window=None operation=None def validate_record(r): fields={'schema_version','kind','record_id','session_id','process','build','generation','operation_id','window_id', 'episode_id','clock','stream','sequence','payload','raw_ref'} if set(r)!=fields or r['schema_version']!=1 or r['kind'] not in KINDS: raise ValueError('unsupported shadow record schema/kind/fields') if type(r['sequence']) is not int or r['sequence']<1: raise ValueError('invalid record sequence') if set(r['clock'])!={'tick','engine_seconds','host_monotonic_ns','host_clock_id','hook_stage'}: raise ValueError('unknown clock fields') if type(r['clock']['host_monotonic_ns']) is not int or r['clock']['host_monotonic_ns']<0: raise ValueError('invalid host clock') if r['raw_ref']!={'path':'journal.jsonl','sequence':r['sequence']}: raise ValueError('raw reference mismatch') for key in ('tick','engine_seconds'): o=r['clock'][key] if set(o)!={'availability','value','reason','source','observation_ref'}: raise ValueError('unknown availability fields') if set(o['source'])!={'hook','stage','source_hash'}: raise ValueError('unknown source fields') digest=o['source']['source_hash'] if digest is not None and not re.fullmatch('[0-9a-f]{64}',digest): raise ValueError('invalid source hash') if o['availability']=='observed': if o['value'] is None or o['reason'] is not None: raise ValueError('observed clock requires a value and no reason') types=(int,) if key=='tick' else (int,float) if type(o['value']) not in types or o['value']<0: raise ValueError('invalid native clock type/value') elif o['availability'] in ('unavailable','not_enabled'): if o['value'] is not None or not o['reason']: raise ValueError('missing clock requires null and a reason') else: raise ValueError('unknown availability') encode(r) # Reject non-finite JSON numerics. def export(manifest, output): manifest=Path(manifest); verify(manifest) metadata=json.loads(manifest.read_text()); raw=(manifest.parent/'journal.jsonl').read_bytes() if hashlib.sha256(raw).hexdigest()!=metadata['sha256']: raise ValueError('journal changed during export; retry a fresh snapshot, never game work') rows=[json.loads(line) for line in raw.splitlines()] output=Path(output) output.mkdir(parents=True,exist_ok=False,mode=0o700) (output/'journal.jsonl').write_bytes(raw) durable_json(output/'journal-manifest.json',metadata) baseline=manifest.parent/'mods-before.json' if baseline.exists(): (output/'mods-before.json').write_bytes(baseline.read_bytes()) values=list(records(rows)) for value in values: validate_record(value) (output/'shadow.jsonl').write_bytes(b''.join(encode(v)+b'\n' for v in values)) started={r['window_id'] for r in values if r['kind']=='window_start'} ended={r['window_id'] for r in values if r['kind']=='window_end'} owner=next((r['payload'] for r in rows if r['kind']=='owner'),{}) pending={r['payload']['operation_id'] for r in rows if r['kind']=='intent'}-{r['payload']['operation_id'] for r in rows if r['kind']=='result'} durable_json(output/'session.json',dict(schema_version=1,owner=owner,end_status='open' if started-ended or pending else 'no_open_windows',pending_operations=sorted(pending), open_windows=sorted(started-ended),native_equivalence_claim=False)) for p in output.iterdir(): p.chmod(0o600) files={p.name:dict(bytes=p.stat().st_size,sha256=hashlib.sha256(p.read_bytes()).hexdigest()) for p in output.iterdir()} result=dict(schema_version=1,kind='shadow_export_manifest',files=files,records=len(values),path=str(output), invalid_operations_retained=True,unsupported_task_lifecycle='See capability registry; no transitions are invented.') durable_json(output/'manifest.json',result) return dict(ok=True,**result) def verify_export(path): path=Path(path); m=json.loads(path.read_text()) if m.get('kind')!='shadow_export_manifest' or m.get('schema_version')!=1: raise ValueError('unsupported manifest') if set(m)!={'schema_version','kind','files','records','path','invalid_operations_retained','unsupported_task_lifecycle'}: raise ValueError('unknown export manifest fields') required={'journal.jsonl','journal-manifest.json','shadow.jsonl','session.json'} if not required.issubset(m['files']): raise ValueError('required export files must be hashed') for name,info in m['files'].items(): if Path(name).name!=name: raise ValueError('manifest traversal') data=(path.parent/name).read_bytes() if len(data)!=info['bytes'] or hashlib.sha256(data).hexdigest()!=info['sha256']: raise ValueError('export digest mismatch: '+name) source_rows=[json.loads(line) for line in (path.parent/'journal.jsonl').read_text().splitlines()] expected={r['record_id']:r for r in records(source_rows)} seen=set(); count=0 for line in (path.parent/'shadow.jsonl').read_text().splitlines(): r=json.loads(line); validate_record(r) if r['record_id'] in seen: raise ValueError('duplicate shadow record ID') if expected.get(r['record_id'])!=r: raise ValueError('shadow record does not match its raw journal source') seen.add(r['record_id']); count+=1 if count!=m['records'] or set(expected)!=seen: raise ValueError('record count/source coverage mismatch') verify(path.parent/'journal-manifest.json') return dict(ok=True,records=count,native_equivalence_claim=False)