===== arm-c1.js ===== const child='thread:provider:codex:native-thread:01a1631f-7783-7d31-95f4-e01a0de72916'; cap.arm('before-lock',child);cap.log('fixture.C1.armed',{child});return {child}; ===== arm-c2.js ===== const {Effect,Ref}=cap.services; const sessions=await Effect.runPromise(Ref.get(cap.services.sessions)); for(const [sid,e] of sessions){ const runtime=e.exposedRuntime; const original=runtime.unloadThread; runtime.unloadThread=(input)=>original(input).pipe(Effect.tap(()=>Effect.promise(async()=>{ const owner=input.providerThread.appThreadId; cap.log('adapter.unload.native-complete',{owner,native:input.providerThread.nativeThreadRef}); await cap.gate('after-native-unload',owner); }))); } const lock=cap.services.threadAttachment; const originalLock=lock.withLock; let sequence=0; lock.withLock=(key,effect)=>Effect.suspend(()=>{const requestId=++sequence;cap.log('lock.request',{key,requestId});return originalLock(key,Effect.sync(()=>cap.log('lock.acquired',{key,requestId})).pipe(Effect.andThen(effect))).pipe(Effect.ensuring(Effect.sync(()=>cap.log('lock.finished',{key,requestId}))));}); cap.arm('after-native-unload',cap.child);cap.log('fixture.C2.armed',{child:cap.child,attached:[...sessions.values()].map(e=>[...e.attachedThreadIds])});return true; ===== arm-delegated-c1.js ===== const {Effect,Ref}=cap.services; const child='thread:delegated-task:command%3Amcp%3A5b553e09-b72e-4ab4-80e9-41c6a4a8e612%3Adelegate-task%3Adelegated-ready-fixture-1'; cap.child=child; const sessions=await Effect.runPromise(Ref.get(cap.services.sessions)); for(const [sid,e] of sessions){ const runtime=e.exposedRuntime;const original=runtime.resumeThread; runtime.resumeThread=(input)=>Effect.suspend(()=>{cap.log('resume.request',{owner:input.threadId??input.providerThread.appThreadId,native:input.providerThread.nativeThreadRef});return original(input).pipe(Effect.tap(()=>Effect.sync(()=>cap.log('resume.complete',{owner:input.threadId??input.providerThread.appThreadId,native:input.providerThread.nativeThreadRef}))));}); } cap.arm('before-lock',child);cap.log('fixture.C1.armed',{child});return {child,attached:[...sessions.values()].map(e=>[...e.attachedThreadIds])}; ===== control.py ===== #!/usr/bin/env python3 """Send an evidence-only service harness script to the isolated diagnostic server.""" import socket, http.client,sys,json class UnixHTTPConnection(http.client.HTTPConnection): def connect(self): self.sock=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM);self.sock.settimeout(180);self.sock.connect('/tmp/pr-23456-capture/control.sock') c=UnixHTTPConnection('localhost');c.request('POST','/',open(sys.argv[1]).read());r=c.getresponse();s=r.read().decode();print(s) if not json.loads(s)['ok']:sys.exit(1) ===== diagnostic-loader.mjs ===== // Evidence-only diagnostic instrumentation. No repository files are written. // Local Unix socket accepts trusted capture scripts, never exposed over HTTP. import { registerHooks } from 'node:module'; import fs from 'node:fs'; import http from 'node:http'; import crypto from 'node:crypto'; const root='/tmp/pr-23456-capture'; const log=(kind,data={})=>fs.appendFileSync(root+'/diagnostic.ndjson',JSON.stringify({time:new Date().toISOString(),pid:process.pid,kind,...data})+'\n'); const gates=new Map(); globalThis.__cap={log, async gate(stage,owner){ const key=stage+':'+owner; const g=gates.get(key); if(!g)return; log('gate.enter',{stage,owner});g.entered=true; await g.promise; log('gate.exit',{stage,owner});gates.delete(key); },arm(stage,owner){let release;const promise=new Promise(r=>release=r);gates.set(stage+':'+owner,{promise,release,entered:false});},release(stage,owner){gates.get(stage+':'+owner)?.release();},gates}; registerHooks({load(url,context,next){const result=next(url,context);if(!url.endsWith('/orchestration-v2/ProviderSessionManager.ts'))return result; let s=String(result.source);log('module.original',{url,sha256:crypto.createHash('sha256').update(s).digest('hex')}); function replace(a,b){if(!s.includes(a))throw new Error('missing instrumentation anchor: '+a);s=s.replace(a,b);} replace('return ProviderSessionManagerV2.of({','globalThis.__cap.services = { Effect, Ref, DateTime, Option, sessions, projectionStore, eventSink, idAllocator, threadAttachment };\n return (globalThis.__cap.manager = ProviderSessionManagerV2.of({'); replace('} satisfies ProviderSessionManagerV2Shape);','} satisfies ProviderSessionManagerV2Shape));'); replace('const ownerThreadId = providerThread.appThreadId ?? input.threadId;','const ownerThreadId = providerThread.appThreadId ?? input.threadId;\n globalThis.__cap.log("cleanup.owner", {parent:input.threadId,owner:ownerThreadId,providerThreadId:providerThread.id,native:providerThread.nativeThreadRef});\n yield* Effect.promise(()=>globalThis.__cap.gate("before-lock",ownerThreadId));'); replace('const entry = (yield* Ref.get(sessions)).get(key);\n if (','const entry = (yield* Ref.get(sessions)).get(key);\n globalThis.__cap.log("cleanup.recheck", {owner:ownerThreadId, attached:[...(entry?.attachedThreadIds ?? [])], sameRuntime:entry?.runtime === detached.value.runtime});\n if ('); replace('yield* unloadThread({ providerThread }).pipe(','yield* Effect.promise(()=>globalThis.__cap.gate("inside-lock",ownerThreadId));\n globalThis.__cap.log("unload.start",{owner:ownerThreadId,native:providerThread.nativeThreadRef});\n yield* unloadThread({ providerThread }).pipe(\n Effect.tap(()=>Effect.sync(()=>globalThis.__cap.log("unload.end",{owner:ownerThreadId,native:providerThread.nativeThreadRef}))),'); log('module.instrumented',{sha256:crypto.createHash('sha256').update(s).digest('hex')});return {...result,source:s}; }}); const socket=root+'/control.sock';try{fs.unlinkSync(socket);}catch{} http.createServer(async(req,res)=>{let body='';for await(const b of req)body+=b;try{const value=await (new Function('cap','return (async()=>{'+body+'})()'))(globalThis.__cap);res.end(JSON.stringify({ok:true,value}));}catch(e){res.end(JSON.stringify({ok:false,error:String(e),stack:e.stack}));}}).listen(socket,()=>fs.chmodSync(socket,0o600)); log('preload.start',{cwd:process.cwd(),argv:process.argv}); ===== extract-db.py ===== #!/usr/bin/env python3 """Read-only selected synthetic-fixture records; never copy the database.""" import sqlite3,json,pathlib root=pathlib.Path('/tmp/pr-23456-capture');c=sqlite3.connect('file:'+str(root/'state/userdata/statev2.sqlite')+'?mode=ro',uri=True);c.row_factory=sqlite3.Row threads=['mcp:e5042a93-73db-499e-93b6-99d8679c563c','thread:provider:codex:native-thread:01a1631f-7783-7d31-95f4-e01a0de72916','thread:delegated-task:command%3Amcp%3A5b553e09-b72e-4ab4-80e9-41c6a4a8e612%3Adelegate-task%3Adelegated-ready-fixture-1'] allowed={'READY','PARENT_DONE','DELEGATED_READY','DELEGATED_PARENT_DONE','PARENT_READY','C1_PONG','C1_AFTER_CLEANUP','C2_PARENT_READY','C2_PONG','FINAL_HEAD_PONG'} runs=[];messages=[] for tid in threads: runs.extend(dict(r) for r in c.execute('select thread_id,run_id,ordinal,provider_thread_id,status,requested_at,completed_at,provider_instance_id from orchestration_v2_projection_runs where thread_id=? order by ordinal',(tid,))) for r in c.execute("select message_id,thread_id,run_id,role,created_at,payload_json from orchestration_v2_projection_messages where thread_id=? and role='assistant'",(tid,)): d=dict(r);p=json.loads(d.pop('payload_json'));text=p.get('text'); if text in allowed:messages.append(d|{'text':text}) c.close();(root/'persisted-results.json').write_text(json.dumps({'access':'SQLite URI mode=ro','runs':runs,'assistantMessages':messages},indent=2)+'\n') print('selected runs',len(runs),'selected assistant messages',len(messages)) ===== extract-protocol.py ===== #!/usr/bin/env python3 """Allowlist protocol identity/lifecycle fields; never export prompts, auth, config or raw logs.""" import json,re,pathlib root=pathlib.Path('/tmp/pr-23456-capture');out=[] methods={'thread/start','thread/resume','thread/unsubscribe','turn/start','turn/completed','thread/started','thread/status/changed'} requests={} for f in sorted((root/'state/userdata/logs/provider').glob('*')): for line in f.open(): m=re.match(r'\[([^]]+)\] NTIVE: (.*)',line) if not m:continue try:d=json.loads(m[2]) except json.JSONDecodeError:continue if d.get('kind')!='protocol':continue e=d.get('event',{});p=e.get('payload',{}); method=p.get('method');direction=e.get('direction') if method in methods: x=p.get('params',{}); t=x.get('thread',{});turn=x.get('turn',{}); row={'time':m[1],'direction':direction,'method':method,'id':p.get('id'),'threadId':x.get('threadId') or t.get('id') or turn.get('threadId'),'session':d.get('providerSessionId')} if method=='turn/completed':row.update(turnId=turn.get('id'),status=turn.get('status')) out.append(row) if direction=='outgoing' and p.get('id') is not None:requests[p['id']]=(method,row.get('threadId')) elif p.get('id') in requests and ('result' in p or 'error' in p): method,tid=requests[p['id']];r=p.get('result') or {};t=r.get('thread') or {};out.append({'time':m[1],'direction':direction,'replyTo':method,'id':p['id'],'threadId':t.get('id') or tid,'error':p.get('error') is not None}) (root/'protocol.ndjson').write_text(''.join(json.dumps(x)+'\n' for x in out)); print(f'Extracted {len(out)} allowlisted records') ===== final-snapshot.js ===== const {Effect,Ref,projectionStore}=cap.services; const parent='mcp:e5042a93-73db-499e-93b6-99d8679c563c'; const native='thread:provider:codex:native-thread:01a1631f-7783-7d31-95f4-e01a0de72916'; const allowed=new Set(['READY','PARENT_DONE','DELEGATED_READY','DELEGATED_PARENT_DONE','PARENT_READY','C1_PONG','C1_AFTER_CLEANUP','C2_PARENT_READY','C2_PONG']); return await Effect.runPromise(Effect.gen(function*(){const rows=[];const sessions=yield* Ref.get(cap.services.sessions); for(const tid of [parent,native,cap.child]){const p=yield* projectionStore.getThreadRecords(tid,['providerThreads','subagents','runs','messages']);const thread=yield* projectionStore.getThread(tid);rows.push({threadId:tid,thread,providerThreads:p.providerThreads.map(t=>({id:t.id,appThreadId:t.appThreadId,providerSessionId:t.providerSessionId,nativeThreadRef:t.nativeThreadRef,status:t.status})),subagents:p.subagents.map(a=>({id:a.id,origin:a.origin,childThreadId:a.childThreadId,providerThreadId:a.providerThreadId,status:a.status,result:allowed.has(a.result)?a.result:undefined})),runs:p.runs.map(r=>({id:r.id,ordinal:r.ordinal,status:r.status,requestedAt:r.requestedAt,completedAt:r.completedAt})),assistantMessages:p.messages.filter(m=>m.role==='assistant'&&allowed.has(m.text)).map(m=>({id:m.id,runId:m.runId,role:m.role,text:m.text,createdAt:m.createdAt}))});} return {sessions:[...sessions].map(([id,e])=>({id,attached:[...e.attachedThreadIds]})),rows};})); ===== finish-c1.js ===== const {Effect,Ref}=cap.services; const sessions=await Effect.runPromise(Ref.get(cap.services.sessions)); const before=[...sessions].map(([id,e])=>({id,attached:[...e.attachedThreadIds]})); cap.log('fixture.C1.release',{before});cap.release('before-lock',cap.child);return before; ===== finish-c2.js ===== const fs=await import('node:fs');const path='/tmp/pr-23456-capture/diagnostic.ndjson'; const relevant=()=>{const rows=fs.readFileSync(path,'utf8').trim().split('\n').map(JSON.parse);const start=rows.findLastIndex(r=>r.kind==='gate.enter'&&r.stage==='after-native-unload'&&r.owner===cap.child);if(start<0)return null;const after=rows.slice(start);return after.some(r=>r.kind==='lock.request'&&r.key.includes(cap.child))?after:null;}; let rows=relevant();if(rows===null)rows=await new Promise(resolve=>{const watcher=fs.watch(path,()=>{const r=relevant();if(r){watcher.close();resolve(r);}});const r=relevant();if(r){watcher.close();resolve(r);}}); const {Effect,Ref}=cap.services;const sessions=await Effect.runPromise(Ref.get(cap.services.sessions)); const state=[...sessions].map(([id,e])=>({id,attached:[...e.attachedThreadIds],loaded:[...e.loadedProviderThreadKeyByThread.keys()]})); cap.log('fixture.C2.contention-observed',{state});cap.release('after-native-unload',cap.child);return {rows,state}; ===== gate-status.js ===== return [...cap.gates].map(([key,g])=>({key,entered:g.entered})); ===== mcp-cli-loader.mjs ===== // Reuse the CLI's short-lived local MCP auth to invoke an existing shipped tool. // This changes only the diagnostic client's command dispatch, not server code. import {registerHooks} from 'node:module'; registerHooks({load(url,ctx,next){const r=next(url,ctx);if(!url.endsWith('/cli/thread.ts'))return r;let s=String(r.source);const a='Effect.fn("cli.thread.send")(function* (tools) {';if(!s.includes(a))throw Error('missing CLI anchor');s=s.replace(a,a+`\n if (message.startsWith("CAPTURE_ORGANIZE:")) {\n const result = yield* tools.call("t3_thread_organize", {threadId,action:message.slice("CAPTURE_ORGANIZE:".length)}, Schema.Unknown);\n yield* printJson(result); return;\n }`);return {...r,source:s};}}); ===== observe-sessions.js ===== const {Effect,Ref}=cap.services;return await Effect.runPromise(Effect.map(Ref.get(cap.services.sessions),m=>[...m].map(([id,e])=>({id,attached:[...e.attachedThreadIds],loaded:[...e.loadedProviderThreadKeyByThread.keys()]})))); ===== prewarm-cli-loader.mjs ===== // Readiness signal only, so the real CLI can be ready before the 10s unload timeout starts. import {registerHooks} from 'node:module'; registerHooks({load(url,ctx,next){const r=next(url,ctx);if(!url.endsWith('/cli/thread.ts'))return r;let s=String(r.source);const a='const readMessageArgument = Effect.fn("cli.thread.readMessage")(function* (argument: string) {';if(!s.includes(a))throw Error('missing readiness anchor');s=s.replace(a,a+'\n if(argument === "-") process.stderr.write("CAPTURE_MESSAGE_READY\\n");');return {...r,source:s};}}); ===== provenance-loader.mjs ===== // Observation only: returns each module byte-for-byte unchanged. import {registerHooks} from 'node:module'; import fs from 'node:fs';import crypto from 'node:crypto'; const path='/tmp/pr-23456-capture/unmodified-modules.ndjson'; registerHooks({load(url,ctx,next){const r=next(url,ctx);if(url.includes('/home/ubuntu/wt/pr-23456/')&&r.source!=null)fs.appendFileSync(path,JSON.stringify({time:new Date().toISOString(),pid:process.pid,url,format:r.format,sha256:crypto.createHash('sha256').update(r.source).digest('hex')})+'\n');return r;}}); ===== provenance.py ===== #!/usr/bin/env python3 import os,json,hashlib,pathlib,subprocess,datetime,sys p=pathlib.Path('/tmp/pr-23456-capture');r=json.loads((p/'state/userdata/server-runtime.json').read_text());pid=r['pid'] files=['apps/server/src/bin.ts','apps/server/src/orchestration-v2/ProviderSessionManager.ts','apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts'] record={'time':datetime.datetime.now(datetime.timezone.utc).isoformat(),'runtime':r,'cwd':os.readlink(f'/proc/{pid}/cwd'),'exe':os.readlink(f'/proc/{pid}/exe'),'cmdline':pathlib.Path(f'/proc/{pid}/cmdline').read_bytes().decode().split('\0'),'proc_start_ticks':pathlib.Path(f'/proc/{pid}/stat').read_text().split()[21],'HEAD':subprocess.check_output(['git','rev-parse','HEAD'],text=True).strip(),'branch':subprocess.check_output(['git','branch','--show-current'],text=True).strip(),'dirty':subprocess.check_output(['git','status','--porcelain'],text=True),'files':{f:hashlib.sha256(pathlib.Path(f).read_bytes()).hexdigest() for f in files},'git_blob_files':{f:hashlib.sha256(subprocess.check_output(['git','show','HEAD:'+f])).hexdigest() for f in files}} (p/sys.argv[1]).write_text(json.dumps(record,indent=2)+'\n');print(json.dumps({'pid':pid,'HEAD':record['HEAD'],'dirty':record['dirty'],'filesMatchHead':record['files']==record['git_blob_files']})) if '--stop' in sys.argv:os.kill(pid,15) ===== release-c1.js ===== cap.log('fixture.C1.native-child-cannot-take-app-message');cap.release('before-lock','thread:provider:codex:native-thread:01a1631f-7783-7d31-95f4-e01a0de72916');return true; ===== run-c2.py ===== #!/usr/bin/env python3 """Real application CLI/MCP traffic plus explicitly injected service unload gate; not a unit test.""" import subprocess,json,pathlib,sys root=pathlib.Path('/tmp/pr-23456-capture');cwd='/home/ubuntu/wt/pr-23456' parent='mcp:e5042a93-73db-499e-93b6-99d8679c563c';child='thread:delegated-task:command%3Amcp%3A5b553e09-b72e-4ab4-80e9-41c6a4a8e612%3Adelegate-task%3Adelegated-ready-fixture-1' commands=[]; children=[] def spawn(cmd,**kwargs): commands.append(cmd);p=subprocess.Popen(cmd,cwd=cwd,**kwargs);children.append(p);return p def control(name,out): cmd=['python3',str(root/'control.py'),str(root/name)];commands.append(cmd) r=subprocess.run(cmd,cwd=cwd,capture_output=True,text=True,timeout=180);(root/out).write_text(r.stdout+r.stderr);r.check_returncode() try: childcmd=['node','--import',str(root/'prewarm-cli-loader.mjs'),'apps/server/src/bin.ts','thread','send',child,'-','--base-dir',str(root/'state'),'--wait','--json'] with (root/'C2-child.txt').open('w') as output: client=spawn(childcmd,stdin=subprocess.PIPE,stdout=output,stderr=subprocess.PIPE,text=True) line=client.stderr.readline();(root/'C2-client-ready.txt').write_text(line) if line.strip()!='CAPTURE_MESSAGE_READY':raise RuntimeError('client not prewarmed: '+line) control('arm-c2.js','C2-arm.json') with (root/'C2-settle-parent.txt').open('w') as settleout: settle=spawn(['node','--import',str(root/'mcp-cli-loader.mjs'),'apps/server/src/bin.ts','thread','send',parent,'CAPTURE_ORGANIZE:settle','--base-dir',str(root/'state'),'--json'],stdout=settleout,stderr=subprocess.STDOUT,text=True) control('wait-for-gate.js','C2-gate-entered.json') client.stdin.write('Reply exactly C2_PONG. No tools or file writes.\n');client.stdin.close() control('finish-c2.js','C2-contention.json') client.wait(timeout=150);settle.wait(timeout=30) (root/'C2-client-stderr.txt').write_text(client.stderr.read()) if client.returncode or settle.returncode:raise RuntimeError(f'client={client.returncode} settle={settle.returncode}') finally: (root/'C2-commands.json').write_text(json.dumps(commands,indent=2)+'\n') for p in children: if p.poll() is None:p.terminate();p.wait(timeout=15) ===== snapshot.js ===== const {Effect,Ref,projectionStore}=cap.services; return await Effect.runPromise(Effect.gen(function*(){ const sessions=yield* Ref.get(cap.services.sessions); const rows=[]; for(const [id,e] of sessions){ for(const tid of e.attachedThreadIds){ const p=yield* projectionStore.getThreadRecords(tid,['providerThreads','subagents','runs']); rows.push({session:id,attached:[...e.attachedThreadIds],threadId:tid,projection:p}); } } return rows; })); ===== verify-evidence.py ===== #!/usr/bin/env python3 """Check ordering and identities in captured evidence, not implementation tests.""" import json,pathlib,hashlib p=pathlib.Path('/tmp/pr-23456-capture');diag=[json.loads(x) for x in (p/'diagnostic.ndjson').read_text().splitlines()];protocol=[json.loads(x) for x in (p/'protocol.ndjson').read_text().splitlines()];db=json.loads((p/'persisted-results.json').read_text());child='thread:delegated-task:command%3Amcp%3A5b553e09-b72e-4ab4-80e9-41c6a4a8e612%3Adelegate-task%3Adelegated-ready-fixture-1';native='01a16322-0b4b-7710-a0f2-fff2bd820179' def one(kind,**where):return next(x for x in diag if x['kind']==kind and all(x.get(k)==v for k,v in where.items())) start=one('gate.enter',stage='before-lock',owner=child)['time'];end=one('gate.exit',stage='before-lock',owner=child)['time'];check=next(x for x in diag if x['kind']=='cleanup.recheck' and x['owner']==child and x['time']>end) assert child in check['attached']; messages={x['text']:x for x in db['assistantMessages']};assert messages['C1_AFTER_CLEANUP']['created_at']>check['time'];assert not any(x.get('method')=='thread/unsubscribe' and x.get('threadId')==native and start<=x['time']<=messages['C1_AFTER_CLEANUP']['created_at'] for x in protocol) gate=one('gate.enter',stage='after-native-unload',owner=child);release=one('gate.exit',stage='after-native-unload',owner=child);held=one('lock.acquired',requestId=3);request=one('lock.request',requestId=4);finish=one('lock.finished',requestId=3);acquire=one('lock.acquired',requestId=4) assert [diag.index(x) for x in [held,gate,request,release,finish,acquire]] == sorted(diag.index(x) for x in [held,gate,request,release,finish,acquire]); assert acquire['time'][...cap.gates].some(([key,g])=>key.endsWith(':'+expected)&&g.entered); if(ready())return {entered:true,owner:expected}; const fs=await import('node:fs'); return await new Promise((resolve)=>{const watcher=fs.watch('/tmp/pr-23456-capture/diagnostic.ndjson',()=>{if(ready()){watcher.close();resolve({entered:true,owner:expected});}});if(ready()){watcher.close();resolve({entered:true,owner:expected});}});