diff --git a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/seed_policy.py b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/seed_policy.py
index ae9c933..20d5f50 100644
--- a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/seed_policy.py
+++ b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/seed_policy.py
@@ -1,25 +1,109 @@
 """Literal owned direct-apphost transport; inherited canonical infolog semantics.
 Inert without the selected startup runner's concrete live custody context.
 """
-import os,time,json,hashlib,selectors,signal,subprocess,pathlib
+import os,time,json,hashlib,selectors,signal,subprocess,pathlib,re
 from growing_log import GrowingLog,_unique,_policy_observation,POLICY_TO_MECHANICAL,MAX_POLICY_INPUT
 from private_io import need,Refused
 from runtime_identity import mark_failure
+def policy_argv(policy,invocation):
+ need(re.fullmatch('[0-9a-f]{64}',invocation or ''),'exact policy invocation')
+ need(policy['environment']=={'PATH':'/usr/bin:/bin'},'literal policy environment')
+ return [policy['apphost'],'--closure-manifest',policy['closurePath'],'--closure-sha256',policy['closureSha256'],'--invocation-id',invocation]
+
+def _entry(context):
+ if type(context)is SeedPolicyContext:
+  import startup_runner as entry
+ elif type(context)is PolicyOnlyContext:
+  import policy_transport_runner as entry
+ else:raise Refused('closed concrete policy context')
+ return entry
+
+def _validate(context):
+ import mechanics as m
+ need(type(context.owner)is m.LinuxOwner,'concrete held Linux policy owner')
+ entry=_entry(context);a=m.load(context.admission)
+ entry.source_binding(a,context.profile)
+ need(a['operation']==('seed-attribution-startup-only'if type(context)is SeedPolicyContext else 'policy-transport-qualification-only'),'matching admitted policy composition')
+ need(context.profile==entry.profile(),'actual pinned concrete policy profile')
+ need(context.evidence_root==pathlib.Path(entry.ROOT)/'evidence','literal bound evidence root')
+ need(context.profile['policy']['environment']=={'PATH':'/usr/bin:/bin'},'literal bound policy environment')
+ return entry,a
+
+def launch_policy(context,invocation,deadline):
+ import mechanics as m
+ need(context.active is None and context.evaluations<3 and len(context.reaped)==context.evaluations,'one settled policy child before next evaluation')
+ entry,a=_validate(context)
+ command=policy_argv(context.profile['policy'],invocation)
+ need(context.pump(deadline)==(1 if type(context)is SeedPolicyContext else 0),'exact prelaunch task CLR count')
+ if type(context)is SeedPolicyContext:entry.runtime_socket().directory_check()
+ context.evaluations+=1;name='policy-'+str(context.evaluations)
+ end=min(deadline,context.deadline-8);rd,rw=os.pipe();ar,aw=os.pipe();child=None
+ try:
+  argv=[entry.PYTHON,'-I','-B',str(entry.HERE/entry.__file__.split('/')[-1]),'--shim','policy','--admission',str(context.admission),'--ready',str(rw),'--ack',str(ar),'--invocation',invocation]
+  child=subprocess.Popen(argv,env=context.profile['policy']['environment'],pass_fds=(rw,ar),start_new_session=True,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,umask=0o077)
+  context.active=child;context.children[name]=child;context.owner.leaders.add(child.pid)
+  os.close(rw);rw=-1;os.close(ar);ar=-1
+  with selectors.DefaultSelector()as sel:
+   sel.register(rd,selectors.EVENT_READ)
+   while True:
+    context.pump(end)
+    if sel.select(m.remaining(end,.02)):break
+   need(os.read(rd,64)==b'BAR-POLICY-READY\n','policy preexec handshake')
+  ident=m.identity(child.pid)
+  need(ident['pgid']==ident['session']==child.pid and ident['ppid']==os.getpid()and ident['uid']==os.geteuid(),'actual owned policy identity')
+  if child.pid not in context.owner.records:context.owner.acquire(ident)
+  need(m.same(ident,context.owner.records[child.pid])and child.pid in context.owner.handles,'held policy pidfd before GO')
+  need(os.sched_getaffinity(child.pid)=={a['cpu']},'policy serial affinity')
+  m.write_new(context.evidence_root/(name+'.process.json'),dict(identity=ident,argv=command,environment=context.profile['policy']['environment'],heldPidfdBeforeGO=True,sourceSHA256=a['sourceSHA256'],profileSHA256=a['profileSHA256']))
+  os.write(aw,b'GO\n');return child
+ except BaseException:
+  # Ownership is retained in the context even when the handshake failed.
+  # Caller/outer cleanup records the first failure and held settlement.
+  if child is not None and child.pid not in context.owner.records:
+   context.owner.observe()
+  raise
+ finally:
+  for fd in [rd,rw,ar,aw]:
+   if fd>=0:
+    try:os.close(fd)
+    except OSError:pass
+
+def policy_shim(args,entry,p,a):
+ import mechanics as m
+ import sys
+ kind='startup_runner'if entry is sys.modules.get('startup_runner')else 'policy_transport_runner'if entry is sys.modules.get('policy_transport_runner')else None
+ need(kind is not None,'literal policy shim caller')
+ entry.source_binding(a,p)
+ need(args.shim=='policy'and os.sched_getaffinity(0)=={a['cpu']},'literal policy shim and affinity')
+ parent=m.load(pathlib.Path(entry.ROOT)/'evidence/worker-start.json')
+ need(m.same(parent['worker'],m.identity(os.getppid()))and parent['admissionSHA256']==m.digest(args.admission),'source-bound actual policy parent')
+ if kind=='startup_runner':entry.runtime_socket().directory_check()
+ argv=policy_argv(p['policy'],args.invocation)
+ os.write(args.ready,b'BAR-POLICY-READY\n');os.close(args.ready)
+ need(os.read(args.ack,16)==b'GO\n','actual held policy release');os.close(args.ack)
+ entry.source_binding(a,p)
+ if kind=='startup_runner':entry.runtime_socket().directory_check()
+ os.chdir(entry.ROOT);os.execve(argv[0],argv,p['policy']['environment'])
+
+def settle_policy(context,child,deadline,force=False):
+ import mechanics as m
+ need(child is context.active,'exact policy child settlement')
+ if child.pid not in context.owner.records:context.owner.observe()
+ ident=context.owner.records[child.pid]
+ if force and child.poll()is None:context.owner.send(ident,signal.SIGKILL)
+ end=min(context.deadline,time.monotonic()+3)if force else deadline
+ remaining=end-time.monotonic();need(remaining>0,'policy settlement Unknown: deadline')
+ code=child.wait(timeout=remaining);context.owner.leaders.discard(child.pid)
+ context.reaped.append({'identity':ident,'exitCode':code,'reaped':True,'forced':force});context.active=None;return code
+
 class SeedPolicyContext:
  def __init__(self,profile,admission,owner,children,actor_selector,streams,counts,deadline):
-  self.profile=profile;self.admission=admission;self.owner=owner;self.children=children;self.actor_selector=actor_selector;self.streams=streams;self.counts=counts;self.deadline=deadline;self.evaluations=0;self.active=None;self.reaped=[]
+  import startup_runner as runner
+  self.profile=profile;self.admission=admission;self.owner=owner;self.children=children;self.actor_selector=actor_selector;self.streams=streams;self.counts=counts;self.deadline=deadline;self.evaluations=0;self.active=None;self.reaped=[];self.evidence_root=pathlib.Path(runner.ROOT)/'evidence'
  def pump(self,deadline):
-  # This is the real actor selector/owner/quota, not an injected success callback.
   import startup_runner as runner
   need(time.monotonic()<min(deadline,self.deadline-8),'policy/whole lifecycle deadline')
-  observed=self.owner.observe();runner.quota(self.owner)
-  clr=[]
-  for identity in observed:
-   try:
-    raw=__import__('runtime_identity').proc_bytes(identity['pid'],'maps',4194304)
-   except FileNotFoundError:continue
-   need(len(raw)<=4194304,'policy live maps cap')
-   if b'libcoreclr.so'in raw:clr.append(identity)
+  observed=self.owner.observe();runner.quota(self.owner);clr=_clr(observed)
   need(len(clr)<=2,'two task CLR cap')
   for key,_ in self.actor_selector.select(0):
    raw=os.read(key.fileobj.fileno(),65536)
@@ -27,28 +111,43 @@ def pump(self,deadline):
    self.counts[key.data]+=len(raw);need(self.counts[key.data]<=33554432,'individual actor stream bound');self.streams[key.data].write(raw);self.streams[key.data].flush()
   for role in ['host','engine']:need(self.children[role].poll()is None,'native actor exited during policy')
   return len(clr)
- def launch(self,invocation,deadline):
-  import startup_runner as runner
-  need(self.active is None and self.evaluations<3 and len(self.reaped)==self.evaluations,'one settled policy child before next evaluation');need(self.pump(deadline)==1,'one host CLR before policy launch')
-  # Host is one CLR. Before policy GO, task ownership census is checked above;
-  # source/admission must separately reserve the two infrastructure CLR.
-  self.evaluations+=1
-  child=runner.launch('policy',self.profile,self.admission,self.owner,min(deadline,self.deadline),self.children,invocation=invocation,evaluation=self.evaluations,cleanup_reserve=0,policy_context=self)
-  self.active=child;return child
- def settle(self,child,deadline,force=False):
+ def launch(self,invocation,deadline):return launch_policy(self,invocation,deadline)
+ def settle(self,child,deadline,force=False):return settle_policy(self,child,deadline,force)
+
+class PolicyOnlyContext:
+ def __init__(self,profile,admission,owner,deadline):
+  import policy_transport_runner as runner
+  self.profile=profile;self.admission=admission;self.owner=owner;self.deadline=deadline;self.children={};self.evaluations=0;self.active=None;self.reaped=[];self.evidence_root=pathlib.Path(runner.ROOT)/'evidence';self.control=None
+ def pump(self,deadline):
   import mechanics as m
-  need(child is self.active,'exact policy child settlement');ident=self.owner.records[child.pid]
-  if force and child.poll()is None:self.owner.send(ident,signal.SIGKILL)
-  remaining=deadline-time.monotonic();need(remaining>0,'policy settlement Unknown: deadline')
-  code=child.wait(timeout=remaining);self.owner.leaders.discard(child.pid);self.reaped.append({'identity':ident,'exitCode':code,'reaped':True});self.active=None;return code
+  need(time.monotonic()<min(deadline,self.deadline-8),'policy-only/whole lifecycle deadline')
+  observed=self.owner.observe();need(not self.owner.uncertain,'policy-only ownership Unknown')
+  known={c.pid for c in self.children.values()}
+  need(all(x['pid']in known for x in observed),'unexpected policy-only descendant')
+  q=m.quota(self.evidence_root.parent,134217728,phase='policy-only',known_owned=list(self.owner.records.values()),live_owned=observed)
+  need(q['files']<=1024,'policy-only leaf bound');clr=_clr(observed)
+  need(len(clr)<=1,'one policy-only task CLR cap')
+  return len(clr)
+ def launch(self,invocation,deadline):return launch_policy(self,invocation,deadline)
+ def settle(self,child,deadline,force=False):return settle_policy(self,child,deadline,force)
+
+def _clr(observed):
+ import runtime_identity
+ clr=[]
+ for identity in observed:
+  try:raw=runtime_identity.proc_bytes(identity['pid'],'maps',4194304)
+  except FileNotFoundError:continue
+  need(len(raw)<=4194304,'policy live maps cap')
+  if b'libcoreclr.so'in raw:clr.append(identity)
+ return clr
 class SeedGrowingLog(GrowingLog):
  def bind(self,context):
-  need(type(context)is SeedPolicyContext and not hasattr(self,'context'),'closed concrete policy context');self.context=context;return self
+  need(type(context)in [SeedPolicyContext,PolicyOnlyContext] and not hasattr(self,'context'),'closed concrete policy context');_validate(context);self.context=context;return self
  def _run_policy(self,path,closure_path,closure_sha,encoded,deadline):
   ctx=self.context;p=ctx.profile['policy'];need((path,closure_path,closure_sha)==(p['apphost'],p['closurePath'],p['closureSha256']),'literal policy input mismatch');need(p['environment']=={'PATH':'/usr/bin:/bin'} and len(encoded)<=MAX_POLICY_INPUT,'literal policy environment/input cap')
   invocation=hashlib.sha256(os.urandom(32)).hexdigest();child=None;sel=None;diagnostic=bytearray();all_stdout=bytearray();output=bytearray();ready=None;offset=0;count=0;check='policy-child-start';stdin_open=True;input_registered=False
   try:
-   child=ctx.launch(invocation,deadline);ident=ctx.owner.records[child.pid]
+   child=ctx.launch(invocation,deadline);ident=dict(ctx.owner.records[child.pid]);ident['startTicks']=str(ident['startTicks'])
    inp,out,err=child.stdin.fileno(),child.stdout.fileno(),child.stderr.fileno()
    for fd in [inp,out,err]:os.set_blocking(fd,False)
    sel=selectors.DefaultSelector();sel.register(out,selectors.EVENT_READ,'stdout');sel.register(err,selectors.EVENT_READ,'stderr')
@@ -73,7 +172,8 @@ def _run_policy(self,path,closure_path,closure_sha,encoded,deadline):
       if ready is None and b'\n'in output:
        line,tail=bytes(output).split(b'\n',1);ready=json.loads(line.decode(),object_pairs_hook=_unique)
        expected={'schema':'fsbar.barc-runtime-evidence-policy-ready/v2','invocationId':invocation,'closureSha256':closure_sha,'pid':ident['pid'],'startTicks':ident['startTicks'],'uid':ident['uid'],'phase':'ready'}
-       need(ready==expected,'policy exact ready/current-process join');output=bytearray(tail);sel.register(inp,selectors.EVENT_WRITE,'stdin');input_registered=True
+       need(ready==expected,'policy exact ready/current-process join');output=bytearray(tail)
+       if type(ctx)is not PolicyOnlyContext or ctx.control!='withheld-input':sel.register(inp,selectors.EVENT_WRITE,'stdin');input_registered=True
     if child.poll()is not None and stdin_open:raise Refused('policy exited before full ready/request')
    need(ready is not None and offset==len(encoded),'ready before exact request');code=ctx.settle(child,deadline);need(code in [0,2],'policy natural exit')
    check='policy-completion';final=json.loads(bytes(output).decode(),object_pairs_hook=_unique)
@@ -81,26 +181,25 @@ def _run_policy(self,path,closure_path,closure_sha,encoded,deadline):
    self.last_policy_observation=_policy_observation(bytes(diagnostic))if diagnostic else None
    self.last_invocation={k:ready[k]for k in ['invocationId','closureSha256','pid','startTicks','uid']};self.last_invocation.update(readyPhase='ready',completedPhase='completed')
    from snapshot_publication import bytes_new,json_new
-   import startup_runner as runner
-   root=pathlib.Path(runner.ROOT)/'evidence';n=ctx.evaluations
+   root=ctx.evidence_root;n=ctx.evaluations
    frames=bytes_new(root/('policy-'+str(n)+'.stdout.raw'),bytes(all_stdout),131072)
    stderr=bytes_new(root/('policy-'+str(n)+'.stderr.raw'),bytes(diagnostic),131072)
    json_new(root/('policy-'+str(n)+'.transport.json'),{'requestSha256':hashlib.sha256(encoded).hexdigest(),'requestBytes':len(encoded),'ready':ready,'completed':final,'stdout':frames,'stderr':stderr,'ownedSettlement':ctx.reaped[-1],'deadlineMonotonic':deadline})
    return (json.dumps(final['result'],separators=(',',':'))+'\n').encode()
   except BaseException as error:
    from snapshot_publication import bytes_new,json_new
-   import startup_runner as runner
-   root=pathlib.Path(runner.ROOT)/'evidence';n=ctx.evaluations
+   root=ctx.evidence_root;n=ctx.evaluations
    try:
     stdout_ack=bytes_new(root/('policy-'+str(n)+'.failure.stdout.raw'),bytes(all_stdout),131072)
     stderr_ack=bytes_new(root/('policy-'+str(n)+'.failure.stderr.raw'),bytes(diagnostic),131072)
-    json_new(root/('policy-'+str(n)+'.failure.json'),{'primaryError':str(error),'checkpoint':check,'requestSha256':hashlib.sha256(encoded).hexdigest(),'requestBytes':len(encoded),'stdout':stdout_ack,'stderr':stderr_ack,'nativeAcceptance':False})
+    json_new(root/('policy-'+str(n)+'.failure.json'),{'primaryError':str(error),'checkpoint':check,'ready':ready,'requestWrittenBytes':offset,'requestSha256':hashlib.sha256(encoded).hexdigest(),'requestBytes':len(encoded),'stdout':stdout_ack,'stderr':stderr_ack,'nativeAcceptance':False})
    except BaseException as publication:error._barc_secondary_publication_failure=str(publication)
    observation=_policy_observation(bytes(diagnostic))if diagnostic else None
    if observation is not None:check=POLICY_TO_MECHANICAL[observation['checkpoint']]
-   if child is not None and ctx.active is child:
+   if ctx.active is not None:
+    child=ctx.active
     try:ctx.settle(child,deadline,force=True)
-    except BaseException as settlement:raise mark_failure(Refused('policy held child settlement Unknown'),check,'unexpected',observation)from settlement
+    except BaseException as settlement:error._barc_secondary_settlement_failure=str(settlement)
    raise mark_failure(error,check,None,observation)
   finally:
    if sel is not None:sel.close()
diff --git a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_prearm_startup.py b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_prearm_startup.py
index 0235f73..e49fb22 100644
--- a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_prearm_startup.py
+++ b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_prearm_startup.py
@@ -2,24 +2,27 @@
 import unittest,pathlib,tempfile,os,json,time
 from unittest.mock import patch,Mock
 from seed_policy import SeedPolicyContext
+import seed_policy as s
 import startup_runner as r
 import mechanics as m
 import stock_bootstrap as b
 class Startup(unittest.TestCase):
- def p(self):return {'source':{'fsbarCommit':'361f5060aea0e340fb8a33da96a91fd234e7fc39','highbarCommit':'b'*40},'proposedPorts':{'grpc':5021,'gateway':3900,'unusedReceiverOrigin':3800},'runId':'controlled','nativeHost':'/controlled/host','engine':'/controlled/engine','dataRoot':'/controlled/data','startscript':'/controlled/startscript','runtimeEnvironment':{'DOTNET_EnableDiagnostics':'0'},'engineRuntimeEnvironment':{'XDG_RUNTIME_DIR':r.ROOT+'/engine/run'},'policy':{'apphost':'/controlled/policy','closurePath':'/controlled/closure','closureSha256':'c'*64}}
+ def p(self):return {'source':{'fsbarCommit':'361f5060aea0e340fb8a33da96a91fd234e7fc39','highbarCommit':'b'*40},'proposedPorts':{'grpc':5021,'gateway':3900,'unusedReceiverOrigin':3800},'runId':'controlled','nativeHost':'/controlled/host','engine':'/controlled/engine','dataRoot':'/controlled/data','startscript':'/controlled/startscript','runtimeEnvironment':{'DOTNET_EnableDiagnostics':'0'},'engineRuntimeEnvironment':{'XDG_RUNTIME_DIR':r.ROOT+'/engine/run'},'policy':{'apphost':'/controlled/policy','closurePath':'/controlled/closure','closureSha256':'c'*64,'environment':{'PATH':'/usr/bin:/bin'}}}
  def test_closed_raw_path_and_policy_environment(self):
   p=self.p();self.assertEqual(r.environment(p,'host')['BARC_ONE_UNIT_RAW_EVENTS'],r.ROOT+'/host/raw.jsonl');self.assertEqual(r.environment(p,'policy'),{'PATH':'/usr/bin:/bin'});self.assertEqual(set(r.commands(p)),{'host','engine'});self.assertEqual(len(r.commands(p,'d'*64)['policy']),7)
   for role in ['browser','receiver','actor']:
    with self.assertRaises(m.Refused):r.launch(role,p,'/controlled/a',None,time.monotonic()+1,{})
  def test_held_identity_before_go_and_literal_policy_command(self):
-  events=[];owner=Mock();owner.leaders=set();owner.acquire.side_effect=lambda x:events.append('held-pidfd');child=Mock(pid=456);identity={'pid':456,'ppid':os.getpid(),'pgid':456,'session':456,'uid':os.getuid(),'startTicks':'12','state':'S'}
+  events=[];owner=Mock();owner.leaders=set();owner.records={};owner.handles={}
+  def acquire(x):owner.records[x['pid']]=x;owner.handles[x['pid']]=1;events.append('held-pidfd')
+  owner.acquire.side_effect=acquire;child=Mock(pid=456);identity={'pid':456,'ppid':os.getpid(),'pgid':456,'session':456,'uid':os.getuid(),'startTicks':'12','state':'S'}
   original_write=os.write
   def write(fd,body):
    if body==b'GO\n':events.append('GO');return len(body)
    return original_write(fd,body)
-  def popen(argv,**kw):original_write(kw['pass_fds'][0],b'BAR-STARTUP-READY\n');events.append('shim');return child
-  with patch.object(r,'runtime_socket',return_value=Mock()),patch.object(r.subprocess,'Popen',side_effect=popen)as launch,patch.object(m,'identity',return_value=identity),patch.object(m,'load',return_value={'cpu':1}),patch.object(m,'write_new'),patch.object(os,'sched_getaffinity',return_value={1}),patch.object(os,'write',side_effect=write):
-   ctx=SeedPolicyContext(self.p(),'/controlled/grant',owner,{},None,{}, {},time.monotonic()+10);ctx.pump=Mock();registry={};result=r.launch('policy',self.p(),'/controlled/grant',owner,time.monotonic()+1,registry,invocation='d'*64,evaluation=1,cleanup_reserve=0,policy_context=ctx)
+  def popen(argv,**kw):original_write(kw['pass_fds'][0],b'BAR-POLICY-READY\n');events.append('shim');return child
+  with patch.object(r,'runtime_socket',return_value=Mock()),patch.object(r.subprocess,'Popen',side_effect=popen)as launch,patch.object(m,'identity',return_value=identity),patch.object(m,'load',return_value={'cpu':1}),patch.object(m,'write_new'),patch.object(os,'sched_getaffinity',return_value={1}),patch.object(os,'write',side_effect=write),patch.object(s,'_validate',side_effect=lambda ctx:(r,{'cpu':1,'sourceSHA256':'a'*64,'profileSHA256':'b'*64})):
+   ctx=SeedPolicyContext(self.p(),'/controlled/grant',owner,{},None,{}, {},time.monotonic()+10);ctx.pump=Mock(return_value=1);registry=ctx.children;result=r.launch('policy',ctx.profile,'/controlled/grant',owner,time.monotonic()+1,registry,invocation='d'*64,evaluation=1,cleanup_reserve=0,policy_context=ctx)
    self.assertIs(result,child);self.assertEqual(events,['shim','held-pidfd','GO']);self.assertIn('policy-1',registry);self.assertEqual(launch.call_args.kwargs['env'],{'PATH':'/usr/bin:/bin'});self.assertIs(launch.call_args.kwargs['stdin'],r.subprocess.PIPE)
  def test_ready_is_not_seed_predicate(self):
   with tempfile.TemporaryDirectory()as d:
diff --git a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_seed_policy.py b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_seed_policy.py
index 4d2c00c..ae27b37 100644
--- a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_seed_policy.py
+++ b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/helper/test_seed_policy.py
@@ -20,13 +20,13 @@ def context(self):
   ctx=s.SeedPolicyContext({'policy':{'apphost':'/controlled/policy','closurePath':'/controlled/closure','closureSha256':'c'*64,'environment':{'PATH':'/usr/bin:/bin'}}},'/controlled/grant',None,{},None,{}, {},time.monotonic()+5)
   return ctx
  def test_mismatch_refuses_before_launch(self):
-  ctx=self.context();log=object.__new__(s.SeedGrowingLog);log.bind(ctx)
+  ctx=self.context();log=object.__new__(s.SeedGrowingLog);log.context=ctx
   with patch.object(ctx,'launch')as launch:
    with self.assertRaises(Refused):log._run_policy('/different','/controlled/closure','c'*64,b'{}',time.monotonic()+1)
    launch.assert_not_called()
- def exercise(self,mutate=lambda x:x,mutate_final=lambda x:x):
+ def exercise(self,mutate=lambda x:x,mutate_final=lambda x:x,encode_final=lambda x:(json.dumps(x)+"\n").encode()):
   with tempfile.TemporaryDirectory()as d:
-   root=pathlib.Path(d);(root/'evidence').mkdir(mode=0o700);child=Child();ctx=self.context();identity={'pid':345,'startTicks':'23','uid':os.getuid()};ctx.owner=type('Owner',(),{'records':{345:identity}})();threads=[];requests=[]
+   root=pathlib.Path(d);(root/'evidence').mkdir(mode=0o700);child=Child();ctx=self.context();ctx.evidence_root=root/'evidence';identity={'pid':345,'startTicks':'23','uid':os.getuid()};ctx.owner=type('Owner',(),{'records':{345:identity}})();threads=[];requests=[]
    def launch(invocation,deadline):
     ctx.evaluations+=1;ctx.active=child
     def peer():
@@ -35,7 +35,7 @@ def peer():
       ready={'schema':'fsbar.barc-runtime-evidence-policy-ready/v2','invocationId':invocation,'closureSha256':'c'*64,**identity,'phase':'ready'}
       os.write(out,(json.dumps(mutate(ready))+'\n').encode());requests.append(os.read(inp,4096))
       completed={**ready,'schema':'fsbar.barc-runtime-evidence-policy-completed/v2','phase':'completed','result':{'CONTROLLED':'not acceptance'}}
-      os.write(out,(json.dumps(mutate_final(completed))+'\n').encode())
+      os.write(out,encode_final(mutate_final(completed)))
      except OSError:pass
      finally:
       for fd in child.peer:
@@ -46,7 +46,7 @@ def peer():
    def settle(c,deadline,force=False):
     if force and not child.stdin.closed:child.stdin.close()
     code=c.wait(timeout=max(.001,deadline-time.monotonic()));ctx.active=None;ctx.reaped.append({'identity':identity,'exitCode':code,'reaped':True});return code
-   log=object.__new__(s.SeedGrowingLog);log.bind(ctx)
+   log=object.__new__(s.SeedGrowingLog);log.context=ctx
    try:
     with patch.object(ctx,'launch',side_effect=launch),patch.object(ctx,'pump')as pump,patch.object(ctx,'settle',side_effect=settle),patch.object(r,'ROOT',d):
      result=log._run_policy('/controlled/policy','/controlled/closure','c'*64,b'{"CONTROLLED":true}',time.monotonic()+1)
@@ -61,6 +61,15 @@ def test_ready_generation_refusals(self):
  def test_completed_generation_refusals(self):
   for key,value in [('pid',999),('startTicks','24'),('closureSha256','d'*64),('phase','ready'),('invocationId','e'*64)]:
    with self.subTest(key=key),self.assertRaises(Refused):self.exercise(mutate_final=lambda row:{**row,key:value})
+ def test_duplicate_completed_refuses(self):
+  with self.assertRaises(json.JSONDecodeError):self.exercise(encode_final=lambda row:(json.dumps(row)+'\n'+json.dumps(row)+'\n').encode())
+ def test_truncated_completed_refuses(self):
+  with self.assertRaises(json.JSONDecodeError):self.exercise(encode_final=lambda row:json.dumps(row)[:-1].encode())
+ def test_input_overflow_refuses_before_launch(self):
+  ctx=self.context();log=object.__new__(s.SeedGrowingLog);log.context=ctx
+  with patch.object(ctx,'launch')as launch:
+   with self.assertRaises(Refused):log._run_policy('/controlled/policy','/controlled/closure','c'*64,b'x'*(s.MAX_POLICY_INPUT+1),time.monotonic()+1)
+   launch.assert_not_called()
  def test_no_second_or_fourth_evaluation(self):
   ctx=self.context();ctx.active=object()
   with self.assertRaises(Refused):ctx.launch('a'*64,time.monotonic()+1)
diff --git a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/policy_transport_runner.py b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/policy_transport_runner.py
new file mode 100644
index 0000000..99a5293
--- /dev/null
+++ b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/policy_transport_runner.py
@@ -0,0 +1,194 @@
+"""Finite NO-GAME qualification of the same owned apphost transport as startup.
+Inert until fresh exact source/profile/closure/control-input root admission.
+The Python worker is explicitly the controlled file writer, never a game actor.
+"""
+import sys
+sys.dont_write_bytecode=True
+import os,json,pathlib,time,datetime,argparse,subprocess,selectors,ctypes,hashlib,stat
+HERE=pathlib.Path(__file__).resolve().parent;sys.path[:0]=[str(HERE),str(HERE/'helper')]
+import mechanics as m,outer_custody as outer
+from seed_policy import SeedGrowingLog,PolicyOnlyContext,policy_shim
+from growing_log import _policy_observation
+from runtime_identity import start_ticks
+from snapshot_publication import bytes_new,json_new
+ROOT='/tmp/bar-policy-transport-qualification-20261004';PYTHON='/usr/bin/python3.14'
+CASES=['stable-consume','malformed-request','withheld-input']
+LIMITS=dict(wholeSeconds=180,cleanupReserve=8,termSeconds=3,killSeconds=3,attemptBytes=134217728,attemptFiles=1024,mapsBytes=4194304,ownedIdentities=128,maximumEvaluations=3,boundarySeconds=5,outputBytes=131072)
+need=m.need
+
+def reviewed_source():
+ s=m.load(HERE/'review-packet.json');need(s['schema']=='bar.seed-attribution-source-review/v2','complete source schema')
+ need(len(s['files'])==len({x['path']for x in s['files']}),'unique source roster')
+ need(set(x['path']for x in s['files'])=={str(f.relative_to(HERE))for f in HERE.rglob('*')if f.is_file()and f.name!='review-packet.json'},'complete sealed source')
+ for x in s['files']:
+  need(set(x)=={'path','bytes','sha256','mode'}and not pathlib.PurePosixPath(x['path']).is_absolute()and '..'not in pathlib.PurePosixPath(x['path']).parts,'safe source pin')
+  f=HERE/x['path'];need(not f.is_symlink()and f.stat().st_size==x['bytes']and stat.S_IMODE(f.stat().st_mode)==x['mode']and m.digest(f)==x['sha256'],'exact sealed source pin')
+ return m.digest(HERE/'review-packet.json')
+
+def profile():
+ p=m.load(HERE/'policy-transport-profile.json')
+ need(set(p)=={'schema','preparationReady','operation','attemptRoot','limits','cases','policy','controlInput','physicalPins','sourceHelperPins'},'closed policy-only profile fields')
+ need(p['schema']=='bar.policy-transport-profile/v1'and p['preparationReady']is True and p['operation']=='policy-transport-qualification-only'and p['attemptRoot']==ROOT,'prepared policy-only profile')
+ need(p['limits']==LIMITS and p['cases']==CASES,'literal policy-only caps/cases')
+ need(set(p['policy'])=={'apphost','closurePath','closureSha256','environment'}and p['policy']['environment']=={'PATH':'/usr/bin:/bin'},'literal policy-only apphost/environment')
+ need(set(p['controlInput'])=={'schema','label','logBase64','sha256','configuration'}and p['controlInput']['schema']=='bar.controlled-log-input/v1'and p['controlInput']['label']=='controlled worker FD; no engine identity','explicit controlled input')
+ import base64
+ raw=base64.b64decode(p['controlInput']['logBase64'],validate=True);need(len(raw)<=10485760 and hashlib.sha256(raw).hexdigest()==p['controlInput']['sha256'],'exact finite controlled log')
+ config=p['controlInput']['configuration'];need(config['roots']=={'attemptRoot':ROOT}and config['commands']['engine']==['controlled-descriptive-fixture','--isolation','--isolation-dir',ROOT+'/fixture/data','--write-dir',ROOT+'/fixture/write','--config','not-executed','not-executed'],'descriptive fixture fields are never executable')
+ need(config['artifacts']['runtimeEvidencePolicy']=={'path':p['policy']['apphost'],'sha256':next(x['sha256']for x in p['physicalPins']if x['path']==p['policy']['apphost'])}and config['artifacts']['runtimeEvidencePolicyClosure']=={'path':p['policy']['closurePath'],'sha256':p['policy']['closureSha256']},'control input exact actual apphost/closure')
+ need(0<len(p['physicalPins'])<=1024 and sum(x['bytes']for x in p['physicalPins'])<=1073741824,'finite physical policy roster')
+ need(set(p['sourceHelperPins'])=={'helper/seed_policy.py','helper/growing_log.py','mechanics.py','outer_custody.py','helper/runtime_identity.py','helper/private_io.py','helper/snapshot_publication.py','policy_transport_runner.py'},'closed no-game modules')
+ for name,sha in p['sourceHelperPins'].items():need(m.digest(HERE/name)==sha,'exact policy-only helper source')
+ return p
+
+def source_binding(a,p):
+ need(a['schema']=='bar.policy-transport-admission/v1'and a['operation']=='policy-transport-qualification-only','policy-only admission separation')
+ need(a['sourceSHA256']==reviewed_source()and a['profileSHA256']==m.digest(HERE/'policy-transport-profile.json')and a['policy']==p['policy']and a['controlInputSHA256']==p['controlInput']['sha256']and a['cases']==CASES,'exact policy-only source/profile/input')
+
+def admission(a,p,operator,now,root_absent,affinity):
+ need(set(a)=={'schema','operation','sourceSHA256','profileSHA256','operator','cpu','expiresUTC','root','oneUse','policy','controlInputSHA256','cases','capacity'},'closed policy-only admission')
+ source_binding(a,p)
+ need(a['root']==ROOT and a['oneUse']is True and root_absent and m.same(a['operator'],operator),'fresh admitted operator/root')
+ need(type(a['cpu'])is int and a['cpu']in affinity,'one admitted CPU')
+ expiry=datetime.datetime.fromisoformat(a['expiresUTC']);need(expiry.tzinfo is not None and 0<(expiry-now).total_seconds()<=600,'finite fresh admission')
+ c=a['capacity'];need(c['maximumTaskCLR']==2 and c['maximumInfrastructureCLR']==2 and c['cpu']==a['cpu']and c['reservationSource'],'actual reservation required');m.checked_pin(c['currentCensusPin'])
+ return a
+
+def originals(p,deadline):
+ for pin in p['physicalPins']:m.remaining(deadline,1);m.checked_pin(pin)
+ return {'pinsChecked':len(p['physicalPins'])}
+
+def quota(owner=None):
+ live=owner.observe()if owner else[]
+ return m.quota(pathlib.Path(ROOT),134217728,phase='policy-only',known_owned=list(owner.records.values())if owner else[],live_owned=live)
+
+def shim(args):
+ p=profile();a=m.load(args.admission);source_binding(a,p)
+ return policy_shim(args,sys.modules['policy_transport_runner'],p,a)
+
+def require_malformed_refusal(retained,settlement,encoded,evaluation):
+ need(evaluation==2 and encoded==b'{malformed','literal second malformed control')
+ need(retained['requestWrittenBytes']==len(encoded)and retained['requestSha256']==hashlib.sha256(encoded).hexdigest()and settlement['exitCode']==2 and not settlement['forced'],'malformed request must naturally refuse after exact full request')
+ ack=retained['stderr'];expected=pathlib.Path(ROOT)/'evidence'/('policy-'+str(evaluation)+'.failure.stderr.raw')
+ need(ack['path']==str(expected)and ack['fullWriteFsyncAcknowledged']is True and ack['bytes']<=2048,'exact acknowledged bounded malformed diagnostic')
+ m.checked_pin(ack);raw=expected.read_bytes()
+ need(len(raw)==ack['bytes']and hashlib.sha256(raw).hexdigest()==ack['sha256'],'retained malformed diagnostic byte join');m.checked_pin(ack)
+ observation=_policy_observation(raw)
+ need(observation=={'schema':'fsbar.barc-runtime-evidence-failure-observation/v1','checkpoint':'request-evaluation','kind':'exception'},'exact genuine malformed request-evaluation exception diagnostic')
+ return observation
+
+def run_cases(p,a_path,owner,deadline):
+ import base64
+ A=pathlib.Path(ROOT);f=A/'fixture';f.mkdir(mode=0o700)
+ for name in ['data','write']:(f/name).mkdir(mode=0o700)
+ # Keep exactly one genuine writable worker descriptor for inherited census.
+ fd=os.open(f/'infolog.txt',os.O_WRONLY|os.O_APPEND|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
+ log=None;ctx=PolicyOnlyContext(p,a_path,owner,deadline);results=[]
+ try:
+  raw=base64.b64decode(p['controlInput']['logBase64'],validate=True)
+  need(os.write(fd,raw)==len(raw),'complete controlled log write');os.fsync(fd)
+  writer={'pid':os.getpid(),'startTicks':start_ticks(os.getpid()),'uid':os.geteuid()}
+  json_new(A/'evidence/controlled-writer.json',{'label':'controlled input, not engine infolog','identity':writer,'openWritableDescriptor':fd,'logSha256':hashlib.sha256(raw).hexdigest()})
+  log=SeedGrowingLog.acquire(str(f/'infolog.txt'),writer).bind(ctx)
+  start=time.monotonic();positive=log.consume('browser',p['controlInput']['configuration'],min(deadline-8,start+5))
+  need(ctx.evaluations==1 and ctx.active is None and len(ctx.reaped)==1 and ctx.reaped[0]['exitCode']==0,'positive must settle exactly one natural successful invocation')
+  need(log.state['phase']=='consumed','inherited consumed state required')
+  json_new(A/'evidence/positive-consume.json',{'controlledInput':True,'result':positive,'lastInvocation':log.last_invocation,'ownedSettlement':ctx.reaped[0],'canonicalCurrentProcessQualified':True,'actorEligible':False})
+  results.append({'case':CASES[0],'qualified':True,'evaluations':1})
+  for case,encoded in [('malformed-request',b'{malformed'),('withheld-input',b'{}')]:
+   ctx.control=case;before=ctx.evaluations;error=None
+   try:log._run_policy(p['policy']['apphost'],p['policy']['closurePath'],p['policy']['closureSha256'],encoded,min(deadline-8,time.monotonic()+5))
+   except BaseException as caught:error=caught
+   need(error is not None and ctx.evaluations==before+1 and ctx.active is None and len(ctx.reaped)==ctx.evaluations,'negative requires actual one invocation and held settlement')
+   retained=m.load(A/'evidence'/('policy-'+str(ctx.evaluations)+'.failure.json'))
+   need(retained['ready']is not None,'negative control ready-before-failure coverage required')
+   if case=='withheld-input':need(retained['requestWrittenBytes']==0 and ctx.reaped[-1]['forced']and 'deadline'in str(error),'withheld-input deadline must force held settlement')
+   else:require_malformed_refusal(retained,ctx.reaped[-1],encoded,ctx.evaluations)
+   need(not getattr(error,'_barc_secondary_settlement_failure',None)and not getattr(error,'_barc_secondary_publication_failure',None),'negative cleanup/publication complete')
+   json_new(A/'evidence'/('control-'+case+'.json'),{'case':case,'expectedFailure':str(error),'firstFailure':retained,'observedSettlement':ctx.reaped[-1],'failureIsNotNaturalSuccess':True,'actorEligible':False})
+   results.append({'case':case,'qualified':True,'expectedNegative':True})
+  need(ctx.evaluations==3 and ctx.active is None,'exact three sequential evaluations')
+  return {'cases':results,'evaluations':ctx.evaluations,'transportQualified':True,'controlledInput':True,'nativeSeedQualified':False,'actorEligible':False,'nativeUsefulPlay':'0/6'}
+ finally:
+  if log is not None:log.close()
+  os.close(fd)
+
+def worker(args):
+ p=profile();a=m.load(args.admission);source_binding(a,p);os.sched_setaffinity(0,{a['cpu']});A=pathlib.Path(ROOT);start=m.load(A/'evidence/outer-start.json')
+ need(m.same(start['outer'],m.identity(os.getppid()))and start['admissionSHA256']==m.digest(args.admission),'independent outer identity')
+ os.write(args.ready,b'BAR-POLICY-WORKER\n');os.close(args.ready);need(os.read(args.ack,16)==b'GO\n','independent worker release');os.close(args.ack)
+ json_new(A/'evidence/worker-start.json',dict(worker=m.identity(os.getpid()),admissionSHA256=m.digest(args.admission)))
+ owner=m.LinuxOwner();result={'transportQualified':False,'failure':None,'secondaryFailures':[]}
+ try:result.update(run_cases(p,pathlib.Path(args.admission),owner,start['deadline']))
+ except BaseException as error:result['failure']=str(error)
+ finally:
+  try:result['innerCleanup']=m.settle(owner,start['deadline']);need(result['innerCleanup']['status']=='settled','inner settlement Unknown')
+  except BaseException as error:result['secondaryFailures'].append(str(error))
+  result['transportQualified']=bool(result.get('transportQualified')and not result['failure']and not result['secondaryFailures']);owner.close();json_new(A/'evidence/transport-result.json',result)
+ return 0 if result['transportQualified']else 3
+
+def execute(a_path):
+ info=a_path.lstat();need(stat.S_ISREG(info.st_mode)and stat.S_IMODE(info.st_mode)==0o600 and info.st_uid==os.geteuid()and info.st_nlink==1,'private single-link actual admission')
+ consumed=a_path.parent/(a_path.name+'.consumed');need(not os.path.lexists(consumed),'one-use grant already consumed')
+ started=time.monotonic();deadline=started+180;p=profile();a=m.load(a_path)
+ admission(a,p,m.identity(os.getppid()),datetime.datetime.now(datetime.timezone.utc),not os.path.lexists(ROOT),os.sched_getaffinity(0));os.sched_setaffinity(0,{a['cpu']});pre=originals(p,deadline-8)
+ json_new(consumed,{'sourceSHA256':a['sourceSHA256'],'profileSHA256':a['profileSHA256'],'operationRoot':ROOT,'operator':a['operator'],'noRetry':True})
+ A=pathlib.Path(ROOT);A.mkdir(mode=0o700)
+ for name in ['logs','evidence']:(A/name).mkdir(mode=0o700)
+ json_new(A/'evidence/outer-start.json',dict(outer=m.identity(os.getpid()),admissionSHA256=m.digest(a_path),deadline=deadline,preOriginals=pre))
+ need(ctypes.CDLL(None,use_errno=True).prctl(36,1,0,0,0)==0,'outer subreaper')
+ rd,rw=os.pipe();ar,aw=os.pipe();proc=None;owned=None;failure=None;result={'transportQualified':False,'secondaryFailures':[]};buffers={'stdout':bytearray(),'stderr':bytearray()}
+ def fail(check,error):
+  nonlocal failure
+  observation={'check':check,'errorType':type(error).__name__,'error':str(error)}
+  if failure is None:failure=error;result['firstFailure']=observation
+  else:result['secondaryFailures'].append(observation)
+ try:
+  proc=subprocess.Popen([PYTHON,'-I','-B',str(HERE/'policy_transport_runner.py'),'--worker','--admission',str(a_path),'--ready',str(rw),'--ack',str(ar)],env={'PATH':'/usr/bin:/bin','PYTHONDONTWRITEBYTECODE':'1'},pass_fds=(rw,ar),start_new_session=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,umask=0o077)
+  os.close(rw);rw=-1;os.close(ar);ar=-1;owned=outer.OwnedChild(proc,adopted=True)
+  with selectors.DefaultSelector()as sel:
+   sel.register(rd,selectors.EVENT_READ);need(sel.select(m.remaining(deadline,5,8))and os.read(rd,64)==b'BAR-POLICY-WORKER\n','real independent worker handshake')
+  need(os.sched_getaffinity(proc.pid)=={a['cpu']},'independent worker affinity');json_new(A/'evidence/worker-independent-identity.json',{'leader':owned.leader,'cpu':a['cpu'],'heldPidfdBeforeGO':True});os.write(aw,b'GO\n');os.close(aw);aw=-1
+  with selectors.DefaultSelector()as sel:
+   for name in buffers:sel.register(getattr(proc,name),selectors.EVENT_READ,name)
+   while sel.get_map()or owned.exited()is None:
+    m.remaining(deadline,1,8);owned.observe();quota()
+    for key,_ in sel.select(.02):
+     raw=os.read(key.fileobj.fileno(),65536)
+     if not raw:sel.unregister(key.fileobj);continue
+     buffers[key.data].extend(raw);need(len(buffers[key.data])<=524288,'outer output cap')
+  status=owned.exited();result['workerExitCode']=status.si_status if status.si_code==os.CLD_EXITED else-status.si_status
+ except BaseException as error:fail('outer-supervision',error)
+ finally:
+  for fd in [rd,rw,ar,aw]:
+   if fd>=0:
+    try:os.close(fd)
+    except OSError:pass
+  if owned:
+   try:result['independentCustody']=owned.settle(deadline);need(result['independentCustody']['leaderReaped']and not result['independentCustody']['remaining']and not result['independentCustody']['errors'],'independent complete settlement')
+   except BaseException as error:fail('outer-cleanup',error)
+  else:fail('outer-cleanup',m.CleanupUnknown('independent owner unavailable'))
+  try:
+   result['postOriginals']=originals(p,deadline);inner=m.load(A/'evidence/transport-result.json');result['inner']=inner
+   need(inner['transportQualified']and inner['innerCleanup']['status']=='settled'and result.get('workerExitCode')==0,'inner transport and cleanup required')
+  except BaseException as error:fail('inner/pins',error)
+  for name,raw in buffers.items():
+   try:bytes_new(A/'logs'/('outer-'+name+'.raw'),bytes(raw),524288)
+   except BaseException as error:fail('raw-publication',error)
+  try:json_new(A/'evidence/full-output-census.json',{'files':m.census(A),'scope':'before final outer receipt; root final census required'});result['finalQuota']=quota()
+  except BaseException as error:fail('final-census',error)
+  result.update(transportQualified=failure is None and not result['secondaryFailures'],failure=str(failure)if failure else None,elapsedSeconds=time.monotonic()-started,operationCount=1,noRetry=True,nativeSeedQualified=False,actorEligible=False,nativeUsefulPlay='0/6',noGameBrowserArmOrCommands=True,sourceLinkRemoteContent='Unknown',rootVerdict='Pending root full review')
+  encoded=(json.dumps(result,sort_keys=True,separators=(',',':'))+'\n').encode();q=result.get('finalQuota',{})
+  if 'files'not in q or q['files']+1>1024 or q['bytes']+len(encoded)>134217728:result.update(transportQualified=False,failure=result.get('failure')or'final receipt quota reserve')
+  json_new(A/'evidence/outer-result.json',result,1048576)
+ return 0 if result['transportQualified']else 3
+
+def main():
+ parser=argparse.ArgumentParser();parser.add_argument('--admission',required=True);parser.add_argument('--worker',action='store_true');parser.add_argument('--shim',choices=['policy']);parser.add_argument('--ready',type=int);parser.add_argument('--ack',type=int);parser.add_argument('--invocation');args=parser.parse_args()
+ need(not(args.worker and args.shim),'one closed entry')
+ if args.shim:return shim(args)
+ if args.worker:return worker(args)
+ need(args.ready is None and args.ack is None and args.invocation is None,'no outer command/payload selection');return execute(pathlib.Path(args.admission))
+if __name__=='__main__':
+ sys.modules['policy_transport_runner']=sys.modules[__name__]
+ raise SystemExit(main())
diff --git a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/run-source-controls.py b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/run-source-controls.py
index 55185b3..1a32265 100644
--- a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/run-source-controls.py
+++ b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/run-source-controls.py
@@ -13,11 +13,11 @@ def forbidden(*args,**kwargs):raise RuntimeError('portable source controls prohi
 os.system=forbidden
 for name in ['fork','forkpty','posix_spawn','posix_spawnp','execv','execve','execvp','execvpe']:
  if hasattr(os,name):setattr(os,name,forbidden)
-FILES=('test_qualify_prearm.py','test_native_evidence.py','helper/test_prearm_startup.py','helper/test_seed_policy.py','helper/test_capture_custody.py','helper/test_loaded_custody.py','helper/test_settings_custody.py','helper/test_proc_bytes.py','fixture/tests/test_prearm_observation.py')
+FILES=('test_policy_transport_runner.py','test_qualify_prearm.py','test_native_evidence.py','helper/test_prearm_startup.py','helper/test_seed_policy.py','helper/test_capture_custody.py','helper/test_loaded_custody.py','helper/test_settings_custody.py','helper/test_proc_bytes.py','fixture/tests/test_prearm_observation.py')
 suite=unittest.TestSuite()
 for index,name in enumerate(FILES):
  spec=importlib.util.spec_from_file_location('portable_seed_controls_'+str(index),ROOT/name);module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module);suite.addTests(unittest.defaultTestLoader.loadTestsFromModule(module))
-expected=71
+expected=89
 if suite.countTestCases()!=expected:raise SystemExit('portable source control roster changed: '+str(suite.countTestCases()))
 result=unittest.TextTestRunner(verbosity=2).run(suite)
 if not result.wasSuccessful()or result.testsRun!=expected or result.skipped:raise SystemExit(1)
diff --git a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/startup_runner.py b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/startup_runner.py
index 83c907f..47e86ba 100644
--- a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/startup_runner.py
+++ b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/startup_runner.py
@@ -25,7 +25,7 @@ def profile():
  validate_schema(p)
  need(p['requiredHostMappings']==required and len(required)==3,'literal pinned apphost/coreclr/clrjit')
  need(p['policy']['environment']=={'PATH':'/usr/bin:/bin'} and p['policy']['maximumSimultaneousChildren']==1 and p['policy']['maximumEvaluations']==3 and p['policy']['deadlineSeconds']==5 and p['policy']['inputBytes']==16777216 and p['policy']['outputBytes']==131072,'literal closed policy caps')
- need(set(p['sourceHelperPins'])=={'helper/runtime_identity.py','helper/stock_bootstrap.py','helper/dynamic_mapping.py','helper/snapshot_publication.py','helper/mapping_profile.py','helper/seed_policy.py','helper/growing_log.py','helper/loaded_custody.py','helper/settings_custody.py','qualify_prearm.py','native_evidence.py'},'closed successor source helpers')
+ need(set(p['sourceHelperPins'])=={'helper/runtime_identity.py','helper/stock_bootstrap.py','helper/dynamic_mapping.py','helper/snapshot_publication.py','helper/mapping_profile.py','helper/seed_policy.py','helper/growing_log.py','helper/loaded_custody.py','helper/settings_custody.py','qualify_prearm.py','native_evidence.py','policy_transport_runner.py'},'closed successor source helpers')
  for path,sha in p['sourceHelperPins'].items():need(m.digest(HERE/path)==sha,'exact profile helper byte pin')
  return p
 
@@ -71,7 +71,8 @@ def commands(p,invocation=None):
  ports=p['proposedPorts'];A=pathlib.Path(ROOT)
  result={'host':[p['nativeHost'],'--stock-tactical-live-host',f"127.0.0.1:{ports['grpc']}",f"http://127.0.0.1:{ports['gateway']}",f"http://127.0.0.1:{ports['unusedReceiverOrigin']}",str(A/'host'),p['source']['fsbarCommit']], 'engine':[p['engine'],'--isolation','--isolation-dir',p['dataRoot'],'--write-dir',str(A/'engine'),'--config',str(A/'engine/springsettings.cfg'),p['startscript']]}
  if invocation is not None:
-  need(re.fullmatch('[0-9a-f]{64}',invocation),'exact policy invocation');result['policy']=[p['policy']['apphost'],'--closure-manifest',p['policy']['closurePath'],'--closure-sha256',p['policy']['closureSha256'],'--invocation-id',invocation]
+  from seed_policy import policy_argv
+  result['policy']=policy_argv(p['policy'],invocation)
  return result
 
 def runtime_socket():
@@ -140,25 +141,25 @@ def record_tail(path,label):
 
 def launch(role,p,a_path,owner,deadline,registered,invocation=None,evaluation=None,cleanup_reserve=8,policy_context=None):
  need(role in ['host','engine','policy'],'literal launch role')
- need((role!='policy' and invocation is None and evaluation is None) or (role=='policy' and re.fullmatch('[0-9a-f]{64}',invocation or '') and evaluation in [1,2,3]),'literal invocation/evaluation')
- name=role if role!='policy'else 'policy-'+str(evaluation)
+ if role=='policy':
+  from seed_policy import SeedPolicyContext,launch_policy
+  need(type(policy_context)is SeedPolicyContext and policy_context.owner is owner and policy_context.children is registered and policy_context.profile is p and str(policy_context.admission)==str(a_path),'exact native policy caller')
+  return launch_policy(policy_context,invocation,deadline)
+ need(role in ['host','engine']and invocation is None and evaluation is None,'literal native actor launch')
+ name=role
  end=min(deadline-cleanup_reserve,clock()+5);rd,rw=os.pipe();ar,aw=os.pipe();child=None
  try:
   runtime_socket().directory_check()
   shim_argv=[PYTHON,'-I','-B',str(HERE/'startup_runner.py'),'--shim',role,'--admission',str(a_path),'--ready',str(rw),'--ack',str(ar)]
-  if role=='policy':shim_argv+=['--invocation',invocation]
-  child=subprocess.Popen(shim_argv,env=environment(p,role),pass_fds=(rw,ar),start_new_session=True,stdin=subprocess.PIPE if role=='policy'else None,stdout=subprocess.PIPE,stderr=subprocess.PIPE,umask=0o077)
+  child=subprocess.Popen(shim_argv,env=environment(p,role),pass_fds=(rw,ar),start_new_session=True,stdin=None,stdout=subprocess.PIPE,stderr=subprocess.PIPE,umask=0o077)
   registered[name]=child;owner.leaders.add(child.pid);os.close(rw);rw=-1;os.close(ar);ar=-1
   with selectors.DefaultSelector()as sel:
    sel.register(rd,selectors.EVENT_READ)
    while True:
-    if role=='policy':
-     from seed_policy import SeedPolicyContext
-     need(type(policy_context)is SeedPolicyContext,'concrete policy preexec supervision context');policy_context.pump(end)
-    if sel.select(m.remaining(end,.02 if role=='policy'else 5)):break
+    if sel.select(m.remaining(end,5)):break
    need(os.read(rd,64)==b'BAR-STARTUP-READY\n','actor preexec handshake')
   ident=m.identity(child.pid);need(ident['pgid']==ident['session']==child.pid and ident['ppid']==os.getpid(),'actual owned actor identity');owner.acquire(ident);need(os.sched_getaffinity(child.pid)=={m.load(a_path)['cpu']},'actor serial affinity')
-  m.write_new(pathlib.Path(ROOT)/'evidence'/(name+'.process.json'),dict(identity=ident,argv=commands(p,invocation)[role],environment=environment(p,role)));os.write(aw,b'GO\n');return child
+  m.write_new(pathlib.Path(ROOT)/'evidence'/(name+'.process.json'),dict(identity=ident,argv=commands(p)[role],environment=environment(p,role)));os.write(aw,b'GO\n');return child
  finally:
   for fd in [rd,rw,ar,aw]:
    if fd>=0:
@@ -312,7 +313,11 @@ def source_binding(a,p):
  need(a['sourceSHA256']==reviewed_source()and a['profileSHA256']==m.digest(HERE/'startup-profile.json'),'shim/worker exact source')
 
 def shim(args):
- p=profile();a=m.load(args.admission);source_binding(a,p);need(args.shim in ['host','engine','policy'],'literal roles');need((args.shim=='policy' and re.fullmatch('[0-9a-f]{64}',args.invocation or ''))or(args.shim!='policy' and args.invocation is None),'shim literal invocation');need(os.sched_getaffinity(0)=={a['cpu']},'shim affinity')
+ p=profile();a=m.load(args.admission);source_binding(a,p)
+ if args.shim=='policy':
+  from seed_policy import policy_shim
+  return policy_shim(args,sys.modules[__name__],p,a)
+ need(args.shim in ['host','engine','policy'],'literal roles');need((args.shim=='policy' and re.fullmatch('[0-9a-f]{64}',args.invocation or ''))or(args.shim!='policy' and args.invocation is None),'shim literal invocation');need(os.sched_getaffinity(0)=={a['cpu']},'shim affinity')
  parent=m.load(pathlib.Path(ROOT)/'evidence/worker-start.json');need(m.same(parent['worker'],m.identity(os.getppid()))and parent['admissionSHA256']==m.digest(args.admission),'owned source-bound parent')
  os.write(args.ready,b'BAR-STARTUP-READY\n');os.close(args.ready);need(os.read(args.ack,16)==b'GO\n','actual preexec observer release');os.close(args.ack);runtime_socket().directory_check();os.chdir(ROOT);c=commands(p,args.invocation)[args.shim];os.execve(c[0],c,environment(p,args.shim))
 
@@ -413,4 +418,6 @@ def main():
  if a.shim:return shim(a)
  if a.worker:return worker(a)
  return execute(pathlib.Path(a.admission))
-if __name__=='__main__':raise SystemExit(main())
+if __name__=='__main__':
+ sys.modules['startup_runner']=sys.modules[__name__]
+ raise SystemExit(main())
diff --git a/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/test_policy_transport_runner.py b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/test_policy_transport_runner.py
new file mode 100644
index 0000000..a6671f8
--- /dev/null
+++ b/tests/Broker.NativeProof/RuntimeEvidence/fixtures/seed-attribution-helper/test_policy_transport_runner.py
@@ -0,0 +1,102 @@
+"""Pure closed-composition/source-routing controls; no policy/CLR/native execution."""
+import unittest,time,pathlib,tempfile,os,ast,json,hashlib
+from unittest.mock import patch,Mock
+import seed_policy as s
+import policy_transport_runner as r
+import startup_runner as native
+import mechanics as m
+from private_io import Refused
+class Composition(unittest.TestCase):
+ def context(self):return s.PolicyOnlyContext({'policy':{'apphost':'/controlled/apphost','closurePath':'/controlled/closure','closureSha256':'c'*64,'environment':{'PATH':'/usr/bin:/bin'}}},'/controlled/a',Mock(),time.monotonic()+20)
+ def test_outer_one_use_and_private_grant_before_effects(self):
+  source=pathlib.Path(r.__file__).read_text();self.assertIn("stat.S_IMODE(info.st_mode)==0o600",source);self.assertIn("info.st_nlink==1",source);self.assertLess(source.index('json_new(consumed,'),source.index('A.mkdir(mode=0o700)'))
+  with tempfile.TemporaryDirectory()as d:
+   grant=pathlib.Path(d)/'grant';grant.write_text('{}');grant.chmod(0o600);(grant.parent/(grant.name+'.consumed')).write_text('{}')
+   with patch.object(r,'profile')as profile,patch.object(r.subprocess,'Popen')as process:
+    with self.assertRaises(m.Refused):r.execute(grant)
+    profile.assert_not_called();process.assert_not_called()
+ def test_malformed_requires_retained_exact_exception_diagnostic(self):
+  with tempfile.TemporaryDirectory()as d:
+   root=pathlib.Path(d);(root/'evidence').mkdir();file=root/'evidence/policy-2.failure.stderr.raw'
+   encoded=b'{malformed';observation={'schema':'fsbar.barc-runtime-evidence-failure-observation/v1','checkpoint':'request-evaluation','kind':'exception'}
+   def retained(value):
+    raw=(json.dumps(value)+'\n').encode();file.write_bytes(raw);file.chmod(0o600);pin=m.physical(file);pin['fullWriteFsyncAcknowledged']=True
+    return {'requestWrittenBytes':len(encoded),'requestSha256':hashlib.sha256(encoded).hexdigest(),'stderr':pin}
+   with patch.object(r,'ROOT',d):
+    settlement={'exitCode':2,'forced':False};self.assertEqual(r.require_malformed_refusal(retained(observation),settlement,encoded,2),observation)
+    for value in [{**observation,'checkpoint':'final-closure'},{**observation,'kind':'refused'},{**observation,'extra':'forged'},{'unknown':True}]:
+     with self.subTest(value=value),self.assertRaises(m.Refused):r.require_malformed_refusal(retained(value),settlement,encoded,2)
+    for code,force in [(0,False),(2,True),(-9,True)]:
+     with self.subTest(code=code,force=force),self.assertRaises(m.Refused):r.require_malformed_refusal(retained(observation),{'exitCode':code,'forced':force},encoded,2)
+    row=retained(observation);file.write_bytes(b'changed')
+    with self.assertRaises(m.Refused):r.require_malformed_refusal(row,settlement,encoded,2)
+ def test_closed_binding(self):
+  with patch.object(s,'_validate')as validation:
+   for kind in [s.SeedPolicyContext,s.PolicyOnlyContext]:
+    log=object.__new__(s.SeedGrowingLog);ctx=object.__new__(kind);self.assertIs(log.bind(ctx),log)
+    with self.assertRaises(Refused):log.bind(ctx)
+   validation.assert_called()
+  class Extension(s.PolicyOnlyContext):pass
+  for ctx in [object.__new__(Extension),object()]:
+   with self.assertRaises(Refused):object.__new__(s.SeedGrowingLog).bind(ctx)
+ def test_exact_seven_element_argv(self):
+  ctx=self.context();argv=s.policy_argv(ctx.profile['policy'],'d'*64)
+  self.assertEqual(argv,['/controlled/apphost','--closure-manifest','/controlled/closure','--closure-sha256','c'*64,'--invocation-id','d'*64])
+  for invocation in ['',None,'d'*63,'d'*64+';']:
+   with self.assertRaises(Refused):s.policy_argv(ctx.profile['policy'],invocation)
+ def test_unsettled_and_fourth_refuse_before_source_or_launch(self):
+  ctx=self.context()
+  with patch.object(s,'_validate')as validate,patch.object(s.subprocess,'Popen')as process:
+   ctx.active=object()
+   with self.assertRaises(Refused):ctx.launch('a'*64,time.monotonic()+1)
+   ctx.active=None;ctx.evaluations=3;ctx.reaped=[1,2,3]
+   with self.assertRaises(Refused):ctx.launch('a'*64,time.monotonic()+1)
+   validate.assert_not_called();process.assert_not_called()
+ def test_native_and_policy_only_delegate_same_launch_settlement(self):
+  ctx=self.context();native_ctx=object.__new__(s.SeedPolicyContext)
+  with patch.object(s,'launch_policy',return_value='CONTROLLED')as launch:
+   self.assertEqual(ctx.launch('a'*64,1),'CONTROLLED');self.assertEqual(native_ctx.launch('a'*64,1),'CONTROLLED');self.assertEqual(launch.call_count,2)
+  with patch.object(s,'settle_policy',return_value=0)as settle:
+   self.assertEqual(ctx.settle('child',1),0);self.assertEqual(native_ctx.settle('child',1),0);self.assertEqual(settle.call_count,2)
+ def test_policy_only_does_not_import_or_invoke_native_profile_socket_commands(self):
+  tree=ast.parse(pathlib.Path(r.__file__).read_text());names={x.module for x in ast.walk(tree)if isinstance(x,ast.ImportFrom)}|{a.name for x in ast.walk(tree)if isinstance(x,ast.Import)for a in x.names}
+  self.assertNotIn('startup_runner',names);self.assertNotIn('runtime_socket',names)
+  source=pathlib.Path(r.__file__).read_text();self.assertNotIn('nativeHost',source);self.assertNotIn('engineRuntimeEnvironment',source)
+ def test_native_shim_and_policy_only_shim_use_shared_function(self):
+  for module in [native,r]:
+   tree=ast.parse(pathlib.Path(module.__file__).read_text());shim=next(x for x in tree.body if isinstance(x,ast.FunctionDef)and x.name=='shim')
+   self.assertTrue(any(isinstance(x,ast.Call)and isinstance(x.func,ast.Name)and x.func.id=='policy_shim'for x in ast.walk(shim)))
+ def test_wrong_admission_does_not_cross_composition(self):
+  with self.assertRaises(m.Refused):r.source_binding({'schema':'bar.seed-attribution-admission/v2','operation':'seed-attribution-startup-only'}, {})
+  ctx=self.context();ctx.owner=object.__new__(m.LinuxOwner)
+  with patch.object(m,'load',return_value={'operation':'seed-attribution-startup-only'}),patch.object(r,'source_binding'):
+   with self.assertRaises(Refused):s._validate(ctx)
+ def test_unprepared_profile_refuses(self):
+  with self.assertRaises(m.Refused):r.profile()
+ def test_policy_only_quota_has_no_socket_exception_and_rejects_foreign_descendant(self):
+  ctx=self.context();ctx.owner.uncertain=False;ctx.owner.records={};ctx.owner.observe.return_value=[]
+  with patch.object(m,'quota',return_value={'files':0})as quota,patch.object(s,'_clr',return_value=[]):
+   self.assertEqual(ctx.pump(time.monotonic()+1),0);self.assertNotIn('runtime_socket',quota.call_args.kwargs)
+   ctx.owner.observe.return_value=[{'pid':999}]
+   with self.assertRaises(Refused):ctx.pump(time.monotonic()+1)
+ def test_one_policy_clr_cap(self):
+  ctx=self.context();ctx.owner.uncertain=False;ctx.owner.records={};ctx.owner.observe.return_value=[]
+  with patch.object(m,'quota',return_value={'files':0}),patch.object(s,'_clr',return_value=[1,2]):
+   with self.assertRaises(Refused):ctx.pump(time.monotonic()+1)
+ def test_native_pump_still_requires_both_live_actors(self):
+  owner=Mock();owner.observe.return_value=[];selector=Mock();selector.select.return_value=[]
+  children={'host':Mock(),'engine':Mock()}
+  ctx=s.SeedPolicyContext({'policy':{}},'/controlled/a',owner,children,selector,{}, {},time.monotonic()+20)
+  with patch.object(native,'quota'),patch.object(s,'_clr',return_value=[]):
+   for role in ['host','engine']:
+    for child in children.values():child.poll.return_value=None
+    children[role].poll.return_value=1
+    with self.subTest(role=role),self.assertRaises(Refused):ctx.pump(time.monotonic()+1)
+ def test_exact_policy_source_profile_required(self):
+  ctx=self.context();ctx.owner=object.__new__(m.LinuxOwner)
+  with patch.object(m,'load',return_value={'operation':'policy-transport-qualification-only'}),patch.object(r,'source_binding'),patch.object(r,'profile',return_value={'different':'CONTROLLED'}):
+   with self.assertRaises(Refused):s._validate(ctx)
+ def test_controlled_worker_writer_and_exact_case_order_are_source_wired(self):
+  source=pathlib.Path(r.__file__).read_text();self.assertIn('os.O_WRONLY|os.O_APPEND|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW',source);self.assertIn("start_ticks(os.getpid())",source);self.assertIn("log.consume('browser'",source);self.assertIn("ctx.evaluations==1",source);self.assertIn("ctx.evaluations==3",source)
+  self.assertEqual(r.CASES,['stable-consume','malformed-request','withheld-input'])
+if __name__=='__main__':unittest.main()