stack/agents/steelbeak/work/queue-227/gate/association.py

# Role/run claims are checked against owned kernel facts and real launcher runtime records.
import os,time,json,hashlib
from pathlib import Path
class Association:
 def __init__(self,root,private,control,nonce,boot,inputs,owned,evidence):
  self.root=root;self.private=private;self.control=control;self.nonce=nonce;self.boot=boot;self.inputs=inputs;self.o=owned;self.e=evidence
  self.preload=str(root/'agents/steelbeak/work/queue-227/gate/priority.mjs');self.driver=str(root/'scripts/case2/e2e.mjs');self.runner=str(root/'packages/harness/src/runner.mjs');self.adapter=str(root/'scripts/case2/citation-adapter.mjs')
  self.launches={};self.registries={};self.accepted={};self.producers={};self.paths={};self.pending={};self.ends={};self.observed_raw=set();self.node_seen=set();self.proven_launches={};self.exited_launches={}
 def pending_failure(self,key,code):
  self.pending.setdefault(key,time.monotonic())
  if time.monotonic()-self.pending[key]>=5:self.e.fail(code)
 def archive_directory(self,directory,category):
  try:
   for path in directory.iterdir():
    if path.is_file() or path.is_symlink():self.e.raw(path,category,missing_ok=True)
  except FileNotFoundError:pass
 def host_role(self,key,row):
  if row['exe']!=self.inputs.execs['node']:return None
  a=row['argv'];node=self.inputs.execs['node']
  if len(a)<2 or a[0] not in [node,'node']:return None
  def module_path(value):return str((Path(row['cwd'])/value).resolve())
  if len(a)==3 and a[1]=='--check' and module_path(a[2])==self.driver:return dict(kind='host',role='host-syntax',run=None,module=self.driver,argv=[node,self.driver],execArgv=['--check'])
  if len(a)==3 and module_path(a[1])==self.driver:
   base=Path(a[2]).resolve()
   if not base.is_relative_to(self.private) or base.name!='root':return None
   return dict(kind='host',role='host-main',run=None,module=self.driver,argv=[node,self.driver,str(base)],execArgv=[])
  return None
 def live_host(self,fresh):
  return [(k,b) for k,b in self.accepted.items() if b['role']=='host-main' and k in fresh and self.o.valid(k)]
 def load_launches(self,fresh):
  self.archive_directory(self.control/'launches','launch-control')
  for path in (self.control/'launches').glob('*.json'):
   raw=self.e.raw(path,'launch-control',missing_ok=True)
   if raw is None:continue
   digest=hashlib.sha256(raw).hexdigest()
   if str(path) in self.paths and self.paths[str(path)]!=digest:self.e.fail('launch-announcement-conflict');continue
   self.paths[str(path)]=digest
   try:r=json.loads(raw);key=(r['wrapperPid'],r['wrapperStart']);run=Path(r['run'])
   except Exception as exc:self.e.fail('parse-launch-'+type(exc).__name__);continue
   if r.get('version')!=1 or r.get('nonce')!=self.nonce or path.name!=f'{key[0]}-{key[1]}.json':self.e.fail('launch-record-invalid');continue
   proven=self.proven_launches.get(str(path))
   if proven and proven['digest']==digest and proven['binding']['wrapper']==list(key) and proven['binding']['run']==str(run):
    current=self.o.valid(key)
    if current is None or current['state']=='Z':
     # This is historical attribution, not new authority. Final closure/survivor checks
     # must still prove normal managed exit; a dead wrapper never repeats startup proof.
     self.exited_launches[str(run)]=proven['binding']
     continue
   hosts=self.live_host(fresh);native=fresh.get(key)
   if not native or native['exe']!=self.inputs.execs['bwrap']:
    self.pending_failure(('launch',str(path)),'launch-kernel-unattributed');continue
   self.pending.pop(('launch',str(path)),None)
   a=native['argv'];owners=[(hk,hb) for hk,hb in hosts if self.o.descends(key,hk,fresh)]
   if len(owners)!=1:self.pending_failure(('owner',str(path)),'launch-host-lineage-refused');continue
   self.pending.pop(('owner',str(path)),None)
   hk,hb=owners[0];base=Path(hb['argv'][2]);expected=base/'data/launches/acme'/run.name
   if run!=expected or run.is_symlink() or a[-3:]!=[self.inputs.execs['node'],self.runner,str(run)] or '--chdir' not in a or a[a.index('--chdir')+1]!=str(run):self.e.fail('launch-kernel-argv-run-refused');continue
   def contains(seq):return any(a[i:i+len(seq)]==seq for i in range(len(a)-len(seq)+1))
   if not contains(['--setenv','NODE_OPTIONS','--import='+self.preload]) or not contains(['--ro-bind',self.preload,self.preload]):self.e.fail('launch-preload-route-missing');continue
   binding=dict(wrapper=list(key),host=list(hk),run=str(run),registry=str(base/'data/control/runtimes'))
   if str(run) in self.launches and self.launches[str(run)]!=binding:self.e.fail('launch-association-conflict');continue
   self.launches[str(run)]=binding
   self.proven_launches[str(path)]=dict(digest=digest,binding=binding)
 def load_registries(self):
  # Raw retained first; this host-only path is derived from actual kernel host/wrapper argv.
  for launch in self.launches.values():
   directory=Path(launch['registry']);self.archive_directory(directory,'runtime-control')
   for path in directory.glob('*.json'):
    r=self.e.parse(path,'runtime-control',missing_ok=True)
    if r is not None:self.registries[str(path)]=r
 def reconcile(self,fresh):
  self.load_registries()
  # A publication may have followed the initial snapshot. Drain fresh ancestry BEFORE
  # classification, then allow further fresh passes within the unchanged5s ceiling.
  fresh=self.o.traverse()
  for path,r in self.registries.items():
   if r.get('phase')=='ended':continue
   for pf,sf in [('pid','startTime'),('runnerPid','runnerStartTime')]:
    if r.get(pf) and r.get(sf) and (r[pf],r[sf]) not in self.o.rows:
     self.pending_failure(('registry',path,pf,r[pf],r[sf]),'runtime-identity-unattributed')
    elif r.get(pf) and r.get(sf):self.pending.pop(('registry',path,pf,r[pf],r[sf]),None)
  return fresh
 def runtime_for(self,launch,fresh):
  wk=tuple(launch['wrapper']);matches=[]
  for r in self.registries.values():
   if r.get('kind')!='session' or r.get('business')!='acme' or r.get('run')!=Path(launch['run']).name or r.get('boot')!=self.boot:continue
   if (r.get('pid'),r.get('startTime'))!=wk:continue
   rk=(r.get('runnerPid'),r.get('runnerStartTime'))
   if rk in fresh and self.o.descends(rk,wk,fresh):matches.append((rk,r))
  unique={k:row for k,row in matches}
  return next(iter(unique.items())) if len(unique)==1 else None
 def derive(self,key,native,directory,fresh):
  if native['exe']!=self.inputs.execs['node']:return None
  env=self.o.selected_env(key)
  if not env or env.get('NODE_OPTIONS')!='--import='+self.preload:return None
  if directory==self.control/'host':
   role=self.host_role(key,native)
   if role:return dict(**role,wrapper=None,runner=None,host=list(key))
   return None
  for launch in self.launches.values():
   run=launch['run']
   if directory!=Path(run)/'.row227-native':continue
   actual=self.runtime_for(launch,fresh)
   if not actual:return None
   rk,registry=actual;node=self.inputs.execs['node'];a=native['argv']
   common=dict(wrapper=launch['wrapper'],runner=list(rk),host=launch['host'],run=run)
   if key==rk and a==[node,self.runner,run]:return dict(**common,kind='runner',role='runner',module=self.runner,argv=a,execArgv=[])
   if key!=rk and self.o.descends(key,rk,fresh) and a==[node,'--import',self.preload,self.adapter] and str(Path(env.get('MOSAIC_TURN_MARKER','/')).parent)==run:
    return dict(**common,kind='adapter',role='adapter',module=self.adapter,argv=[node,self.adapter],execArgv=['--import',self.preload])
  return None
 def receipt(self,path,directory,fresh):
  raw=self.e.raw(path,'native-control',missing_ok=True)
  if raw is None:return
  digest=hashlib.sha256(raw).hexdigest()
  if str(path) in self.paths and self.paths[str(path)]!=digest:self.e.fail('native-receipt-rewritten');return
  self.paths[str(path)]=digest
  try:r=json.loads(raw)
  except Exception as exc:self.e.fail('parse-native-'+type(exc).__name__);return
  if set(r)!=set(['version','nonce','kind','pid','start','boot','nice','argv','module','execArgv','run']):self.e.fail('native-receipt-shape');return
  if r.get('version')!=1 or r.get('nonce')!=self.nonce or r.get('boot')!=self.boot or r.get('nice')!=19:self.e.fail('native-receipt-invalid');return
  matches=[(k,v) for k,v in fresh.items() if v['exe']==self.inputs.execs['node'] and v['start']==r.get('start') and v['nspid'] and v['nspid'][-1]==r.get('pid')]
  # Previously released producer may have exited; unchanged bytes alone preserve prior
  # attribution, never release a new record or conceal conflicts after its exit.
  old=[(k,p) for k,p in self.producers.items() if p==(str(path),digest)]
  if not matches and len(old)==1:return
  if len(matches)!=1:self.pending_failure(('receipt',str(path)),'native-innermost-identity-unattributed');return
  self.pending.pop(('receipt',str(path)),None)
  key,native=matches[0]
  if key in self.producers and self.producers[key]!=(str(path),digest):self.e.fail('native-producer-conflict-or-replay');return
  self.producers[key]=(str(path),digest)
  role=self.derive(key,native,directory,fresh)
  if role is None:self.pending_failure(('role',str(path)),'native-kernel-role-run-unattributed');return
  self.pending.pop(('role',str(path)),None)
  expected={k:role[k] for k in ['kind','run','module','argv','execArgv']}
  if any(r.get(k)!=v for k,v in expected.items()) or path.name!=f"{role['kind']}-{r['pid']}-{r['start']}.json":self.e.fail('native-role-module-run-argv-conflict');return
  if native['nice']!=19:self.e.fail('native-release-priority-refused');return
  binding=dict(**role,hostPid=key[0],hostStart=key[1],namespacePid=r['pid'],boot=self.boot,receiptSha256=digest)
  if key in self.accepted and self.accepted[key]!=binding:self.e.fail('native-release-association-conflict');return
  release=dict(receipt=r,binding=binding,receiptSha256=digest)
  ack=path.with_suffix('.release')
  if ack.exists():
   prior=self.e.parse(ack,'native-release')
   if prior!=release:self.e.fail('native-release-conflict');return
  else:
   # Revalidate latest association and ownership immediately before this exact release.
   latest=self.o.observe(key)
   if not latest or latest['nice']!=19 or self.derive(key,latest,directory,fresh)!=role:self.e.fail('native-release-latest-refused');return
   temp=ack.with_suffix('.pending')
   if temp.exists():self.e.raw(temp,'native-release-conflict');self.e.fail('native-release-pending-conflict');return
   with temp.open('x') as f:f.write(json.dumps(release)+'\n')
   temp.rename(ack)
  self.accepted[key]=binding
  if role['role']=='runner':
   permit=self.control/'authorizations'/f"{role['wrapper'][0]}-{role['wrapper'][1]}.json"
   authorization=dict(nonce=self.nonce,binding=binding)
   if permit.exists():
    if self.e.parse(permit,'host-authorization')!=authorization:self.e.fail('host-authorization-conflict');return
   else:
    temp=permit.with_suffix('.pending')
    if temp.exists():self.e.raw(temp,'authorization-conflict');self.e.fail('authorization-publication-conflict');return
    with temp.open('x') as f:f.write(json.dumps(authorization)+'\n')
    temp.rename(permit)
  self.e.write('native/'+f'{key[0]}-{key[1]}.json',dict(receipt=r,binding=binding))
 def poll(self,fresh):
  self.node_seen.update(k for k,r in fresh.items() if r['exe']==self.inputs.execs['node'])
  self.archive_directory(self.control/'host','host-control')
  for path in (self.control/'host').glob('*.json'):
   if not path.name.endswith('.end.json'):self.receipt(path,self.control/'host',fresh)
  self.load_launches(fresh);fresh=self.reconcile(fresh)
  for run in self.launches:
   directory=Path(run)/'.row227-native';self.archive_directory(directory,'sandbox-control')
   for path in directory.glob('*.json'):
    if not path.name.endswith('.end.json'):self.receipt(path,directory,fresh)
  return fresh
 def complete(self):
  if self.exited_launches:
   closure=self.e.parse(self.control/'fixture-cleanup.json','fixture-closure')
   for run,binding in self.exited_launches.items():
    if not isinstance(closure,dict) or closure.get('runtimeEntries')!=[] or not any(
      r.get('run')==Path(run).name and [r.get('pid'),r.get('startTime')]==binding['wrapper']
      for r in closure.get('runs',[]) if isinstance(r,dict)):
     self.e.fail('proven-launch-exit-closure-unverified');continue
    for key,accepted in self.accepted.items():
     if accepted['run']==run:
      current=self.o.valid(key)
      if current and current['state']!='Z':self.e.fail('proven-launch-exit-still-live')
  for directory in [self.control,self.control/'launches',self.control/'authorizations']+[Path(v['registry']) for v in self.launches.values()]:
   self.archive_directory(directory,'completion-control')
   if directory.exists() and any(p.name.endswith('.pending') or p.name.startswith('.receipt-') for p in directory.iterdir()):self.e.fail('control-publication-incomplete')
  if self.pending:self.e.fail('unresolved-association-input')
  if not any(b['role']=='host-main' for b in self.accepted.values()):self.e.fail('required-host-main-native-missing')
  if not self.launches:self.e.fail('required-real-launch-missing')
  for launch in self.launches.values():
   for role in ['runner','adapter']:
    if not any(b['role']==role and b['run']==launch['run'] and b['wrapper']==launch['wrapper'] for b in self.accepted.values()):self.e.fail('required-associated-'+role+'-missing')
  for key in self.node_seen:
   if key not in self.accepted:self.e.fail('observed-node-associated-receipt-missing')
  for directory in [self.control/'host']+[Path(run)/'.row227-native' for run in self.launches]:
   self.archive_directory(directory,'final-native-control')
   for path in directory.glob('*.pending'):self.e.fail('native-publication-incomplete')
   for path in directory.glob('*.json'):
    if path.name.endswith('.end.json'):continue
    if not path.with_suffix('.release').exists():self.e.fail('native-unreleased-input')