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

# Admitted parent freeze, never a newly generated partial baseline.
import os,json,re,hashlib
from pathlib import Path
class Inputs:
 def __init__(self,root,admission,expected_sha,evidence):
  self.root=Path(root).resolve();self.e=evidence;self.admission=Path(admission);self.expected_sha=expected_sha
  raw=evidence.raw(self.admission,'admission')
  if raw is None or hashlib.sha256(raw).hexdigest()!=expected_sha:raise RuntimeError('admission-digest-refused')
  self.profile=json.loads(raw)
  if self.profile.get('version')!=1 or self.profile.get('admitted') is not True or self.profile.get('noRenicingDaemon') is not True or not self.profile.get('identity') or not self.profile.get('joinedIdentity'):raise RuntimeError('target-profile-unadmitted')
  if Path(self.profile['root']).resolve()!=self.root:raise RuntimeError('admitted-root-refused')
  self.manifest=Path(self.profile['inputManifest']['path']);content=evidence.raw(self.manifest,'frozen-manifest')
  if content is None or hashlib.sha256(content).hexdigest()!=self.profile['inputManifest']['sha256']:raise RuntimeError('frozen-manifest-digest-refused')
  self.expected={}
  for line in content.decode().splitlines():
   digest,name=line.split('  ',1);path=Path(name)
   if not re.fullmatch('[a-f0-9]{64}',digest) or path.is_absolute() or '..' in path.parts or name in self.expected:raise RuntimeError('frozen-manifest-entry-refused')
   self.expected[name]=digest
  self.manifest_sha=hashlib.sha256(content).hexdigest()
  self.execs={}
  for kind in ['node','bwrap','python']:
   row=self.profile['executables'][kind];p=Path(row['path']).resolve()
   if hashlib.sha256(p.read_bytes()).hexdigest()!=row['sha256']:raise RuntimeError('admitted-executable-changed')
   self.execs[kind]=str(p)
  self.required=self.required_files()
  if not self.required.issubset(self.expected):raise RuntimeError('frozen-execution-inputs-omitted')
 def required_files(self):
  root=self.root
  required={'packages/seat/src/sandbox.mjs','packages/harness/src/runner.mjs','packages/harness/src/git-tools.mjs','packages/harness/src/first-output.mjs','scripts/test-case2.sh','package.json','package-lock.json','packages/webui/package.json','packages/webui/package-lock.json','RELEASE'}
  sandbox=(root/'packages/seat/src/sandbox.mjs').read_text();block=sandbox.split('export const RUNNER_FILES = Object.freeze([',1)[1].split('].map',1)[0]
  required.update(re.findall(r'"(packages/[^"\n]+)"',block))
  # Real argv-launched and plugin/dynamic assets are seeds, not assumed literal host imports.
  for directory in ['scripts/case2','agents/steelbeak/work/queue-227/gate','packages/harness/src','packages/seat/src','packages/plugins','packages/webui/src/public','roles','contracts','skills','tools/unslop-hook']:
   p=root/directory
   if p.exists():required.update(str(f.relative_to(root)) for f in p.rglob('*') if f.is_file() and 'node_modules' not in f.parts and '__pycache__' not in f.parts)
  # Dependencies are mandatory own installed inputs; source-walk exclusions must not
  # filter every file from these separately admitted trees.
  for directory in ['node_modules','packages/webui/node_modules']:
   tree=root/directory
   if not tree.is_dir() or tree.is_symlink() or not tree.resolve().is_relative_to(root):raise RuntimeError('required-installed-dependency-tree-missing')
   installed={str(f.relative_to(root)) for f in tree.rglob('*') if f.is_file() and '__pycache__' not in f.parts}
   if not installed:raise RuntimeError('required-installed-dependency-tree-empty')
   required.update(installed)
  for p in (root/'packages').glob('*/tests/fixtures'):
   required.update(str(f.relative_to(root)) for f in p.rglob('*') if f.is_file())
  todo=list(required);seen=set()
  while todo:
   name=todo.pop()
   if name in seen:continue
   seen.add(name);p=root/name
   if p.suffix not in ['.mjs','.js']:continue
   for ref in re.findall(r'''(?:from\s*|import\s*\(\s*|new URL\(\s*)['"](\.[^'"]+)['"]''',p.read_text()):
    q=(p.parent/ref).resolve()
    if q.is_file() and q.is_relative_to(root):
     relative=str(q.relative_to(root))
     if relative not in required:required.add(relative);todo.append(relative)
  return required
 def measure(self):
  result={}
  for name in sorted(self.expected):
   try:
    path=self.root/name
    if not path.resolve().is_relative_to(self.root):raise RuntimeError('input-outside-admitted-root')
    result[name]=hashlib.sha256(path.read_bytes()).hexdigest()
   except Exception as exc:result[name]='ERROR:'+type(exc).__name__;self.e.fail('pinned-input-unreadable')
  for label,path,expected in [('@admission',self.admission,self.expected_sha),('@manifest',self.manifest,self.manifest_sha)]:
   try:result[label]=hashlib.sha256(path.read_bytes()).hexdigest()
   except Exception as exc:result[label]='ERROR:'+type(exc).__name__
   if result[label]!=expected:self.e.fail('consumed-list-or-admission-changed')
  for kind,row in self.profile['executables'].items():
   try:result['@exe:'+kind]=hashlib.sha256(Path(row['path']).resolve().read_bytes()).hexdigest()
   except Exception as exc:result['@exe:'+kind]='ERROR:'+type(exc).__name__
   if result['@exe:'+kind]!=row['sha256']:self.e.fail('admitted-executable-changed')
  if any(result.get(name)!=digest for name,digest in self.expected.items()):self.e.fail('admitted-frozen-input-changed')
  return result