diff --git a/src/handoff-operations.ts b/src/handoff-operations.ts index 775abf4..b97da5b 100644 --- a/src/handoff-operations.ts +++ b/src/handoff-operations.ts @@ -1,6 +1,7 @@ // Opt-in integration only. Default CLI imports this without opening storage or // starting an observer/collector. Production receipt bootstrap is unqualified. import { homedir, hostname } from 'node:os'; +import { channel } from 'node:diagnostics_channel'; import { dirname, isAbsolute, join, parse, resolve } from 'node:path'; import { closeSync, constants, fstatSync, lstatSync, mkdirSync, openSync, readSync, realpathSync } from 'node:fs'; import { HandoffLedger, payloadDigest, type HandoffBinding } from './handoff-ledger.js'; @@ -11,10 +12,51 @@ import { metadataCapability, observeMetadata, resumeMetadataObservation, metadat import { checkMessage, queueId, send, type SendContext, type SendOptions } from './send.js'; import { Refusal } from './discovery.js'; import { HomeRefusal } from './writer.js'; -import { executable, run, UnknownOutcome } from './process.js'; +import { executable, run, UnknownOutcome, type Done } from './process.js'; import { currentPlatform, selectNativeProfile, validateNativeProfiles, type NativeProfile } from './native-profiles.js'; import { codexStoragePin } from './codex-storage.js'; +// Internal observations only: no I/O or public protocol changes. Never publish +// paths, identifiers, generations, native output or arguments on this channel. +const ownerProbeChannel = channel('session-peer.handoff.owner-probe'); +type ProbeOutcome = 'matched' | 'generation_mismatch' | 'invalid_reply' | 'unverified_reply' | + 'timeout' | 'output_limit' | 'process_error' | 'nonzero_exit' | 'interrupted' | 'stopped' | 'aborted' | 'budget_exhausted'; +async function retainedOwnerProbe(scope: MetadataScope, budget: HandoffBudget, ownerArgs: string[], + phase: 'send' | 'wait', stopped: () => boolean, signal?: AbortSignal): Promise { + const observed = ownerProbeChannel.hasSubscribers, started = observed ? performance.now() : 0; + let timeoutMs = 0; + const report = (outcome: ProbeOutcome, result?: Done) => { + if (!observed) return; + ownerProbeChannel.publish(Object.freeze({schemaVersion:1,phase,outcome, + elapsedMs:Math.max(0,Math.round((performance.now()-started)*1000)/1000), + timeoutMs:Math.max(0,timeoutMs),spawned:result?.spawned ?? false, + code:result?.code ?? null,interrupted:result?.interrupted ?? false,stopReason:result?.stopReason ?? null})); + }; + if (stopped()) { report('stopped'); return false; } + if (signal?.aborted) { report('aborted'); return false; } + if (budget.observationRemainingMs() < 1) { report('budget_exhausted'); return false; } + const worker = fileURLToPath(new URL('./handoff-writer-probe.js', import.meta.url)); + timeoutMs = Math.min(800,budget.observationRemainingMs()); + let result: Done; + try { result = await run(process.execPath,[worker,scope.home,scope.threadId,...ownerArgs], + {timeout:timeoutMs,limit:4096,env:{...process.env,NODE_OPTIONS:undefined,NODE_PATH:undefined}}); } + catch (error) { report('process_error'); throw error; } + if (stopped()) { report('stopped',result); return false; } + if (signal?.aborted) { report('aborted',result); return false; } + if (result.code !== 0 || result.interrupted) { + report(result.stopReason ?? (result.interrupted ? 'interrupted' : 'nonzero_exit'),result); return false; + } + try { + const value = JSON.parse(result.stdout); + const valid = value.schemaVersion === 1 && value.ok === true && + Object.keys(value).every(key=>['schemaVersion','ok','generation'].includes(key)) && /^[0-9a-f]{64}$/.test(value.generation); + const matched = valid && value.generation === scope.generation; + report(matched ? 'matched' : valid ? 'generation_mismatch' : + value.schemaVersion === 1 && value.ok === false ? 'unverified_reply' : 'invalid_reply',result); + return matched; + } catch { report('invalid_reply',result); return false; } +} + export type HandoffOptions = { correlationId?: string; requestAck?: boolean; observeDelivery?: boolean; remote?: boolean; waitFor?: Exclude; seconds?: number; ledgerPath?: string; payload?: string; budget?: HandoffBudget; resultOverhead?: number }; export type OperationResult = { value: Record; exitCode: number }; @@ -138,13 +180,8 @@ export async function handoffSend(options: SendOptions, handoffOptions: HandoffO }; const persist = (action: () => T) => persistWithinBudget(budget, action); const authorize = async () => { if (stopped) throw new Refusal('stopped_by_operator', 130); budget.beforeEffect(); handoff = await persist(() => { if (stopped) throw new Refusal('stopped_by_operator', 130); return ledger.commitEffect(handoff!.correlationId, original!, budget); }); fenced = true; ownWait(); }; - const verifyOwner = async (target: MetadataScope, signal?: AbortSignal) => { - if (stopped || signal?.aborted || budget.observationRemainingMs() < 1) return false; - const worker = fileURLToPath(new URL('./handoff-writer-probe.js', import.meta.url)); - const result = await run(process.execPath, [worker, target.home, target.threadId, ...ownerArgs], { timeout: Math.min(800, budget.observationRemainingMs()), limit: 4096, env: { ...process.env, NODE_OPTIONS: undefined, NODE_PATH: undefined } }); - if (signal?.aborted || stopped || result.code !== 0 || result.interrupted) return false; - try { const value = JSON.parse(result.stdout); return value.schemaVersion === 1 && value.ok === true && Object.keys(value).every(key => ['schemaVersion','ok','generation'].includes(key)) && /^[0-9a-f]{64}$/.test(value.generation) && value.generation === target.generation; } catch { return false; } - }; + const verifyOwner = (target: MetadataScope, signal?: AbortSignal) => + retainedOwnerProbe(target,budget,ownerArgs,'send',()=>stopped,signal); const finish = (value: Record, exitCode: number): OperationResult => ({ value: { ...value, ...(handoff ? { handoff } : {}) }, exitCode }); if (handoffOptions.waitFor) process.on('SIGINT', stop); try { @@ -275,17 +312,8 @@ export async function waitForHandoff(ledger: HandoffLedger, id: string, goal: 'd const persist = (action: () => T) => persistWithinBudget(budget, action); const ownWait = () => { if (operationId) handoff = ledger.operationStatus(id, operationId); }; const ownerArgs = ownerProbeArgs(context?.qualification); - const verifyOwner = async (scope: MetadataScope, signal: AbortSignal) => { - if (stopped || signal.aborted || budget.observationRemainingMs() < 1) return false; - const worker = fileURLToPath(new URL('./handoff-writer-probe.js', import.meta.url)); - const result = await run(process.execPath, [worker, scope.home, scope.threadId, ...ownerArgs], { timeout: Math.min(800, budget.observationRemainingMs()), limit: 4096, env: { ...process.env, NODE_OPTIONS: undefined, NODE_PATH: undefined } }); - if (stopped || signal.aborted || result.code !== 0 || result.interrupted) return false; - try { - const value = JSON.parse(result.stdout); - return value.schemaVersion === 1 && value.ok === true && Object.keys(value).every(key => ['schemaVersion','ok','generation'].includes(key)) && - /^[0-9a-f]{64}$/.test(value.generation) && value.generation === scope.generation; - } catch { return false; } - }; + const verifyOwner = (scope: MetadataScope, signal: AbortSignal) => + retainedOwnerProbe(scope,budget,ownerArgs,'wait',()=>stopped,signal); process.on('SIGINT', stop); try { handoff = await persist(() => ledger.beginWait(id, goal, budget, { observationSupported: context !== undefined })); diff --git a/test/fixtures/owner-probe-diagnostics.mjs b/test/fixtures/owner-probe-diagnostics.mjs new file mode 100644 index 0000000..4dfd431 --- /dev/null +++ b/test/fixtures/owner-probe-diagnostics.mjs @@ -0,0 +1,37 @@ +// Explicit test preload only. Production never reads this environment key or +// writes diagnostic files. Per process: <=64 attempts, <=512 bytes per record. +// The shared-file quota is best-effort under multiple writers, not atomic. +// An absent terminal capture summary means completeness is unknown. +import { channel } from 'node:diagnostics_channel'; +import { appendFileSync, lstatSync } from 'node:fs'; +const file=process.env.SESSION_PEER_TEST_OWNER_PROBE_LOG; +if(file){ + let observed=0,captured=0,dropped=0; + const append=value=>{ + try{ + const line=JSON.stringify(value)+'\n',bytes=Buffer.byteLength(line),stat=lstatSync(file); + if(bytes>512||!stat.isFile()||stat.isSymbolicLink()||stat.nlink!==1||stat.size+bytes>32768)return false; + appendFileSync(file,line,{mode:0o600});return true; + }catch{return false;} + }; + const keys=['schemaVersion','phase','outcome','elapsedMs','timeoutMs','spawned','code','interrupted','stopReason']; + const outcomes=['matched','generation_mismatch','invalid_reply','unverified_reply','timeout','output_limit','process_error','nonzero_exit','interrupted','stopped','aborted','budget_exhausted']; + const safe=value=>value!==null&&typeof value==='object'&&!Array.isArray(value)&& + Object.keys(value).length===keys.length&&keys.every(k=>Object.hasOwn(value,k))&& + value.schemaVersion===1&&['send','wait'].includes(value.phase)&&outcomes.includes(value.outcome)&& + Number.isFinite(value.elapsedMs)&&value.elapsedMs>=0&&Number.isFinite(value.timeoutMs)&&value.timeoutMs>=0&&value.timeoutMs<=800&& + typeof value.spawned==='boolean'&&typeof value.interrupted==='boolean'&& + (value.code===null||Number.isInteger(value.code))&&[null,'timeout','output_limit','process_error'].includes(value.stopReason); + channel('session-peer.handoff.owner-probe').subscribe(value=>{ + observed++; + try{ + if(observed>64||!safe(value)){dropped++;return;} + // Project into a fresh plain record; never serialize unknown properties + // or a supplied object's custom toJSON method. + const record=Object.fromEntries(keys.map(k=>[k,value[k]])); + if(!append(record)){dropped++;return;} + captured++; + }catch{dropped++;} + }); + process.once('exit',()=>append({captureSummary:{observed,captured,dropped}})); +} diff --git a/test/handoff-delivery.test.ts b/test/handoff-delivery.test.ts index 3df8709..77e8519 100644 --- a/test/handoff-delivery.test.ts +++ b/test/handoff-delivery.test.ts @@ -7,20 +7,26 @@ import fs from 'node:fs'; import {DatabaseSync} from 'node:sqlite'; import {chmodSync,existsSync,mkdirSync,mkdtempSync,readFileSync,realpathSync,rmSync,statSync,writeFileSync,watch} from 'node:fs'; import {join,delimiter,resolve} from 'node:path'; +import {fileURLToPath} from 'node:url'; import {tmpdir} from 'node:os'; import {validateHandoff} from '../dist/handoff.js'; import {HandoffLedger,payloadDigest} from '../dist/handoff-ledger.js'; import {handoffSend} from '../dist/handoff-operations.js'; const qualified={skip:process.platform!=='darwin'||process.arch!=='arm64'}; const thread='aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa',cli=resolve('dist/cli.js'); +const diagnosticPreload=new URL('./fixtures/owner-probe-diagnostics.mjs',import.meta.url).href; const schema=`CREATE TABLE thread_turns(thread_id TEXT NOT NULL,turn_id TEXT NOT NULL,status TEXT NOT NULL,PRIMARY KEY(thread_id,turn_id)); CREATE TABLE thread_items(thread_id TEXT NOT NULL,turn_id TEXT NOT NULL,item_id TEXT NOT NULL,rollout_ordinal INTEGER NOT NULL,item_json TEXT NOT NULL,item_type TEXT NOT NULL,PRIMARY KEY(thread_id,turn_id,item_id)); CREATE INDEX idx_thread_items_user_messages ON thread_items(thread_id,rollout_ordinal) WHERE item_type='userMessage'; CREATE TABLE thread_history_projection_state(thread_id TEXT PRIMARY KEY,next_rollout_byte_offset INTEGER NOT NULL,next_rollout_ordinal INTEGER NOT NULL);`; async function fixture(t:TestContext,mode='inject',version='0.160.1',warm=true){ const root=realpathSync(mkdtempSync(join(process.env.TASK_TEMP??tmpdir(),'codex-hd-'))),home=join(root,'home'),bin=join(root,'bin');chmodSync(root,0o700);mkdirSync(home,{mode:0o700});mkdirSync(bin,{mode:0o700});mkdirSync(join(home,'thread-writer-locks'),{mode:0o700}); + const probeLog=join(root,'owner-probes.jsonl');writeFileSync(probeLog,'',{mode:0o600}); + const probeRecords=()=>readFileSync(probeLog,'utf8').trim().split('\n').filter(Boolean).map(line=>JSON.parse(line)); + const probes=()=>probeRecords().filter(value=>!Object.hasOwn(value,'captureSummary')); + const probeCaptures=()=>probeRecords().filter(value=>Object.hasOwn(value,'captureSummary')).map(value=>value.captureSummary); let history:DatabaseSync|undefined; - t.after(()=>{try{history?.close();}finally{rmSync(root,{recursive:true,force:true});}}); + t.after(()=>{try{const events=probes();t.diagnostic(JSON.stringify({ownerProbeSummary:{count:events.length,maxElapsedMs:events.length?Math.max(...events.map(e=>e.elapsedMs)):null,outcomes:Object.fromEntries([...new Set(events.map(e=>e.outcome))].map(k=>[k,events.filter(e=>e.outcome===k).length])),stopReasons:Object.fromEntries([...new Set(events.map(e=>e.stopReason??'none'))].map(k=>[k,events.filter(e=>(e.stopReason??'none')===k).length]))},ownerProbeTrace:events.slice(0,64),traceOmitted:Math.max(0,events.length-64),captureSummaries:probeCaptures().slice(0,16),captureSummariesOmitted:Math.max(0,probeCaptures().length-16)}));}finally{try{history?.close();}finally{rmSync(root,{recursive:true,force:true});}}}); const state=new DatabaseSync(join(home,'state_5.sqlite'));state.exec('CREATE TABLE threads(id TEXT, cwd TEXT)');state.prepare('INSERT INTO threads VALUES(?,?)').run(thread,home);state.close(); // Model a live warm WAL store: keep the owner connection open so other // fixture connections cannot remove/recreate WAL and SHM between probes. @@ -60,11 +66,11 @@ if(mode==='ledger-failure')fs.chmodSync(path.join(process.env.SESSION_PEER_HANDO if(mode==='inject'){const {DatabaseSync}=require('node:sqlite');const db=new DatabaseSync(path.join(process.env.CODEX_HOME,'thread_history_1.sqlite'));db.prepare('INSERT INTO thread_items VALUES(?,?,?,?,?,?)').run(r.params.threadId,'turn-fixture','native-item-not-client-id',1,JSON.stringify({type:'userMessage',id:'native-item-not-client-id',clientId:r.params.clientUserMessageId,content:[{text:'PRIVATE_BODY_SENTINEL'}]}),'userMessage');db.close();} console.log(JSON.stringify({id:r.id,result:{queuedSubmission:{id:'queue-fixture',clientUserMessageId:r.params.clientUserMessageId,input:r.params.input}}})); }});`,{mode:0o700}); - const env={...process.env,HOME:root,USERPROFILE:root,CODEX_HOME:home,SESSION_PEER_CODEX_HOMES:'[]',CLAUDE_CONFIG_DIR:join(root,'.claude'),SESSION_PEER_TAILSCALE:'off',SESSION_PEER_UPDATE_NOTICE:'',CODEX_THREAD_ID:'',CODEX_SESSION_ID:'',PATH:bin+delimiter+(process.env.PATH??''),SESSION_PEER_HANDOFF_HOME:join(root,'state','handoff')}; + const env={...process.env,SESSION_PEER_TEST_OWNER_PROBE_LOG:probeLog,HOME:root,USERPROFILE:root,CODEX_HOME:home,SESSION_PEER_CODEX_HOMES:'[]',CLAUDE_CONFIG_DIR:join(root,'.claude'),SESSION_PEER_TAILSCALE:'off',SESSION_PEER_UPDATE_NOTICE:'',CODEX_THREAD_ID:'',CODEX_SESSION_ID:'',PATH:bin+delimiter+(process.env.PATH??''),SESSION_PEER_HANDOFF_HOME:join(root,'state','handoff')}; const args=['send','--to','codex:'+thread,'--codex-home',home,'--codex-bin',codex,'--message','fixture request','--no-from','--no-reply-to']; let activeSend:undefined|{child:ReturnType;closed:()=>boolean;output:()=>{stdout:string;stderr:string}}; - function start(options:string[]){const child=spawn(process.execPath,[cli,...options,'--json'],{env});let stdout='',stderr='',closedOutput=false;child.once('close',()=>{closedOutput=true;});child.stdout.on('data',c=>stdout+=c);child.stderr.on('data',c=>stderr+=c);if(options[0]==='send')activeSend={child,closed:()=>closedOutput,output:()=>({stdout,stderr})};child.stdin.end();const timer=setTimeout(()=>child.kill('SIGKILL'),15000);const closed=once(child,'close');const result=closed.then(([code,signal])=>{clearTimeout(timer);assert.equal(signal,null,stderr);return{code,value:JSON.parse(stdout),stdout,stderr};});void result.catch(()=>{});t.after(async()=>{if(child.exitCode===null&&child.signalCode===null){child.kill('SIGKILL');await closed;}});return{child,result};} - async function invoke(options:string[]){return start(options).result;} + function start(options:string[],preload?:string){const child=spawn(process.execPath,['--import',diagnosticPreload,...(preload?['--import',preload]:[]),cli,...options,'--json'],{env});let stdout='',stderr='',closedOutput=false;child.once('close',()=>{closedOutput=true;});child.stdout.on('data',c=>stdout+=c);child.stderr.on('data',c=>stderr+=c);if(options[0]==='send')activeSend={child,closed:()=>closedOutput,output:()=>({stdout,stderr})};child.stdin.end();const timer=setTimeout(()=>child.kill('SIGKILL'),15000);const closed=once(child,'close');const result=closed.then(([code,signal])=>{clearTimeout(timer);assert.equal(signal,null,stderr);return{code,value:JSON.parse(stdout),stdout,stderr};});void result.catch(()=>{});t.after(async()=>{if(child.exitCode===null&&child.signalCode===null){child.kill('SIGKILL');await closed;}});return{child,result};} + async function invoke(options:string[],preload?:string){return start(options,preload).result;} assert.equal((await invoke(['handoff','init'])).code,0); const calls=()=>readFileSync(log,'utf8').trim().split('\n').filter(Boolean).map(line=>JSON.parse(line)); const nativeCalls=()=>readFileSync(nativeLog,'utf8').trim().split('\n').filter(Boolean).map(line=>JSON.parse(line)); @@ -122,7 +128,7 @@ console.log(JSON.stringify({id:r.id,result:{queuedSubmission:{id:'queue-fixture' }); } function inject(clientId:string){const db=new DatabaseSync(join(home,'thread_history_1.sqlite'));try{db.prepare('INSERT INTO thread_items VALUES(?,?,?,?,?,?)').run(thread,'turn-fixture','after-submit-item',1,JSON.stringify({type:'userMessage',id:'after-submit-item',clientId,content:[{text:'PRIVATE_BODY_SENTINEL'}]}),'userMessage');}finally{db.close();}} - return{root,home,codex,env,calls,nativeCalls,args,invoke,start,queued,initialized,ledger,ledgerFile,pendingWait,inject,foreignOwner:()=>writeFileSync(lsof+'.data','p'+held.pid+'\0ccodex-fixture\0u'+((process.getuid?.()??0)+1)+'\0',{mode:0o600}),releaseInitialize:()=>writeFileSync(initializeRelease,'release')}; + return{root,home,codex,env,calls,nativeCalls,probes,probeCaptures,args,invoke,start,queued,initialized,ledger,ledgerFile,pendingWait,inject,foreignOwner:()=>writeFileSync(lsof+'.data','p'+held.pid+'\0ccodex-fixture\0u'+((process.getuid?.()??0)+1)+'\0',{mode:0o600}),releaseInitialize:()=>writeFileSync(initializeRelease,'release')}; } async function directEnvironment(f:{root:string;env:NodeJS.ProcessEnv},action:()=>Promise):Promise{ const saved={...process.env},cwd=process.cwd();Object.assign(process.env,f.env);process.chdir(f.root); @@ -363,13 +369,14 @@ test('independent wait SIGINT records its persisted operation as stopped without test('independent wait cannot adopt another owner or reconstruct missing retained qualification',qualified,async t=>{ const wrong=await fixture(t,'none');const submitted=await wrong.invoke([...wrong.args,'--observe-delivery']); - assert.equal(submitted.code,0,submitted.stdout);const original=validateHandoff(submitted.value.handoff),wrongNative=wrong.nativeCalls();wrong.foreignOwner(); + assert.equal(submitted.code,0,JSON.stringify({result:submitted.value,queueCalls:wrong.calls(),nativeCalls:wrong.nativeCalls(),ownerProbes:wrong.probes(),ownerProbeCaptures:wrong.probeCaptures()}));const original=validateHandoff(submitted.value.handoff),wrongNative=wrong.nativeCalls();wrong.foreignOwner(); const refused=await wrong.invoke(['handoff','wait','--correlation-id',original.correlationId,'--wait-for','delivered','--wait-timeout','10']); assert.equal(refused.code,1,refused.stdout);const changed=validateHandoff(refused.value.handoff); assert.equal(changed.wait.status,'unsupported');assert.equal(changed.submission.status,'submitted');assert.equal(changed.targetGeneration,original.targetGeneration); assert.equal(changed.observation.injectionObserved,false);assert.equal(wrong.calls().length,1);assert.deepEqual(wrong.nativeCalls(),wrongNative); + assert.ok(wrong.probes().some(event=>event.phase==='wait'&&event.outcome==='unverified_reply')); const old=await fixture(t,'none');const sent=await old.invoke([...old.args,'--observe-delivery']); - assert.equal(sent.code,0,sent.stdout);const handoff=validateHandoff(sent.value.handoff),oldNative=old.nativeCalls(); + assert.equal(sent.code,0,JSON.stringify({result:sent.value,queueCalls:old.calls(),nativeCalls:old.nativeCalls(),ownerProbes:old.probes(),ownerProbeCaptures:old.probeCaptures()}));const handoff=validateHandoff(sent.value.handoff),oldNative=old.nativeCalls(); const retained=old.ledger();delete retained.data.records[handoff.correlationId].observationQualification; retained.digest=payloadDigest(JSON.stringify(retained.data)); writeFileSync(old.ledgerFile,JSON.stringify(retained),{mode:0o600});old.inject(handoff.correlationId); @@ -430,3 +437,56 @@ test('exact qualified 0.162.1 native profile retains its own source when resumin assert.equal(result.value.handoff.wait.status,'satisfied');assert.equal(result.value.handoff.ack.status,'not_requested'); assert.equal(f.calls().length,1);assert.deepEqual(f.nativeCalls(),nativeBefore); }); + +for(const mode of ['timeout','invalid_reply','generation_mismatch','process_error'] as const){ + test('owner probe diagnostic distinguishes '+mode+' without authorizing a queue',qualified,async t=>{ + const f=await fixture(t,'none'),loader=join(f.root,'owner-probe-control.mjs'),reply=join(f.root,'owner-probe-reply.mjs'); + writeFileSync(reply,mode==='timeout'?'await new Promise(resolve=>setTimeout(resolve,1200));': + `process.stdout.write(JSON.stringify(${JSON.stringify(mode==='invalid_reply'?{schemaVersion:1,ok:true,generation:'f'.repeat(64),unexpected:'PRIVATE_PROBE_SENTINEL'}:{schemaVersion:1,ok:true,generation:'f'.repeat(64)})})+'\\n');`,{mode:0o600}); + writeFileSync(loader,`import cp from 'node:child_process';import{syncBuiltinESMExports}from'node:module'; +const spawn=cp.spawn;let count=0; +cp.spawn=function(binary,args,options){ + if(binary===process.execPath&&Array.isArray(args)&&String(args[0]).endsWith('/handoff-writer-probe.js')&&++count===3){ + ${mode==='process_error'?"return spawn.call(this,binary+'-missing',args,options);":mode==='timeout'?`return spawn.call(this,binary,['--import',${JSON.stringify(reply)},...args],options);`:`return spawn.call(this,binary,[${JSON.stringify(reply)}],options);`} + } + return spawn.call(this,binary,args,options); +};syncBuiltinESMExports();`,{mode:0o600}); + const sent=await f.invoke([...f.args,'--observe-delivery'],loader),events=f.probes(); + const diagnostic=JSON.stringify({result:sent.value,queueCalls:f.calls(),ownerProbes:events}); + assert.equal(sent.code,1,diagnostic);assert.equal(sent.value.error,'app_server_owner_changed',diagnostic); + assert.equal(sent.value.submitted,false);assert.deepEqual(f.calls(),[]);assert.equal(events.length,3,diagnostic); + assert.deepEqual(events.slice(0,2).map(e=>e.outcome),['matched','matched']); + assert.equal(events[2].outcome,mode);assert.equal(events[2].timeoutMs,800); + if(mode==='timeout'){assert.equal(events[2].stopReason,'timeout');assert.equal(events[2].interrupted,true);assert.ok(events[2].elapsedMs>=700);} + if(mode==='process_error'){assert.equal(events[2].stopReason,'process_error');assert.equal(events[2].spawned,false);} + for(const event of events){ + assert.deepEqual(Object.keys(event).sort(),['schemaVersion','phase','outcome','elapsedMs','timeoutMs','spawned','code','interrupted','stopReason'].sort()); + assert.ok(Buffer.byteLength(JSON.stringify(event))<=512);assert.equal(event.phase,'send'); + assert.equal(JSON.stringify(event).includes(f.root),false);assert.equal(JSON.stringify(event).includes(thread),false); + } + assert.equal(diagnostic.includes('PRIVATE_PROBE_SENTINEL'),false);assert.equal(diagnostic.includes('f'.repeat(64)),false); + }); +} + +test('test-only owner capture reports dropped records at the per-process bound',async t=>{ + const root=realpathSync(mkdtempSync(join(process.env.TASK_TEMP??tmpdir(),'codex-probe-capture-')));chmodSync(root,0o700); + t.after(()=>rmSync(root,{recursive:true,force:true}));const file=join(root,'capture.jsonl');writeFileSync(file,'',{mode:0o600}); + const script="import{channel}from'node:diagnostics_channel';const c=channel('session-peer.handoff.owner-probe');for(let i=0;i<80;i++)c.publish({schemaVersion:1,phase:'send',outcome:'matched',elapsedMs:1,timeoutMs:800,spawned:true,code:0,interrupted:false,stopReason:null});"; + const done=spawnSync(process.execPath,['--import',diagnosticPreload,'--input-type=module','--eval',script],{env:{...process.env,SESSION_PEER_TEST_OWNER_PROBE_LOG:file},encoding:'utf8',timeout:5000}); + assert.equal(done.status,0,done.stderr);assert.equal(done.signal,null);assert.equal(done.stdout,''); + const rows=readFileSync(file,'utf8').trim().split('\n').map(line=>JSON.parse(line)); + assert.equal(rows.filter(row=>!Object.hasOwn(row,'captureSummary')).length,64); + assert.deepEqual(rows.at(-1),{captureSummary:{observed:80,captured:64,dropped:16}}); + assert.ok(statSync(file).size<=32768); +}); + +test('test-only owner capture rejects unexpected private fields without writing them',async t=>{ + const root=realpathSync(mkdtempSync(join(process.env.TASK_TEMP??tmpdir(),'codex-probe-private-')));chmodSync(root,0o700); + t.after(()=>rmSync(root,{recursive:true,force:true}));const file=join(root,'capture.jsonl');writeFileSync(file,'',{mode:0o600}); + const script="import{channel}from'node:diagnostics_channel';const c=channel('session-peer.handoff.owner-probe');const valid={schemaVersion:1,phase:'send',outcome:'matched',elapsedMs:1,timeoutMs:800,spawned:true,code:0,interrupted:false,stopReason:null};c.publish({...valid,privatePath:'PRIVATE_CAPTURE_SENTINEL'});c.publish(valid);const disguised={...valid};Object.defineProperty(disguised,'toJSON',{value:()=>({privatePath:'PRIVATE_CAPTURE_SENTINEL'})});c.publish(disguised);"; + const done=spawnSync(process.execPath,['--import',diagnosticPreload,'--input-type=module','--eval',script],{env:{...process.env,SESSION_PEER_TEST_OWNER_PROBE_LOG:file},encoding:'utf8',timeout:5000}); + assert.equal(done.status,0,done.stderr);assert.equal(done.signal,null);assert.equal(done.stdout,''); + const raw=readFileSync(file,'utf8'),rows=raw.trim().split('\n').map(line=>JSON.parse(line)); + assert.equal(rows.length,3);assert.deepEqual(rows.at(-1),{captureSummary:{observed:3,captured:2,dropped:1}}); + assert.equal(raw.includes('PRIVATE_CAPTURE_SENTINEL'),false);assert.equal(raw.includes('privatePath'),false); +}); diff --git a/test/ssh-handoff.test.ts b/test/ssh-handoff.test.ts index 888999c..b8768c5 100644 --- a/test/ssh-handoff.test.ts +++ b/test/ssh-handoff.test.ts @@ -112,12 +112,16 @@ test('SSH remaining duration charges setup/frame time and never renews cleanup o async function fixture(t: TestContext) { const root = realpathSync(mkdtempSync(join(process.env.TASK_TEMP ?? tmpdir(), 'codex-ssh-handoff-'))); chmodSync(root, 0o700); const destination = join(root, 'destination'), origin = join(root, 'origin'), socket = join(root, 'socket'), log = join(root, 'calls.jsonl'); - const preparationLog = join(root, 'preparations.jsonl'); + const preparationLog = join(root, 'preparations.jsonl'), probeLog=join(root,'owner-probes.jsonl');writeFileSync(probeLog,'',{mode:0o600}); + const diagnosticPreload=fileURLToPath(new URL('./fixtures/owner-probe-diagnostics.mjs',import.meta.url)); + const probeRecords=()=>readFileSync(probeLog,'utf8').trim().split('\n').filter(Boolean).map(line=>JSON.parse(line)); + const ownerProbes=()=>probeRecords().filter(value=>!Object.hasOwn(value,'captureSummary')); + const probeCaptures=()=>probeRecords().filter(value=>Object.hasOwn(value,'captureSummary')).map(value=>value.captureSummary); mkdirSync(origin); mkdirSync(join(destination, '.claude', 'sessions'), { recursive: true }); writeFileSync(log, ''); writeFileSync(preparationLog, ''); const messages: string[] = []; const server = createServer(s => { let body = ''; s.on('data', part => { body += part; }); s.on('end', () => { messages.push(body); s.end(); }); }); server.listen(socket); await once(server, 'listening'); - t.after(async () => { await new Promise(resolve => server.close(() => resolve())); rmSync(root, { recursive: true, force: true }); }); + t.after(async () => { try{const events=ownerProbes();t.diagnostic(JSON.stringify({ownerProbeSummary:{count:events.length,maxElapsedMs:events.length?Math.max(...events.map(e=>e.elapsedMs)):null,outcomes:Object.fromEntries([...new Set(events.map(e=>e.outcome))].map(k=>[k,events.filter(e=>e.outcome===k).length])),stopReasons:Object.fromEntries([...new Set(events.map(e=>e.stopReason??'none'))].map(k=>[k,events.filter(e=>(e.stopReason??'none')===k).length]))},ownerProbeTrace:events.slice(0,64),traceOmitted:Math.max(0,events.length-64),captureSummaries:probeCaptures().slice(0,16),captureSummariesOmitted:Math.max(0,probeCaptures().length-16)}));}finally{try{await new Promise(resolve => server.close(() => resolve()));}finally{rmSync(root, { recursive: true, force: true });}} }); writeFileSync(join(destination, '.claude', 'sessions', `${process.pid}.json`), JSON.stringify({ pid: process.pid, name: 'fixture', messagingSocketPath: socket })); writeFileSync(join(root, 'ssh'), `#!${process.execPath} const fs=require('node:fs'),cp=require('node:child_process'),args=process.argv.slice(2); @@ -128,23 +132,23 @@ fs.appendFileSync(process.env.FIXTURE_LOG,JSON.stringify(request)+'\\n'); if(process.env.FIXTURE_PARTIAL_PREPARE&&request.handoff?.preparation){process.stdout.write('{');process.exit(0);} const env={...process.env,HOME:process.env.FIXTURE_DEST,USERPROFILE:process.env.FIXTURE_DEST,CLAUDE_CONFIG_DIR:process.env.FIXTURE_DEST+'/.claude',SESSION_PEER_HANDOFF_HOME:process.env.FIXTURE_DEST+'/state/handoff'}; const selector=args.at(-1).match(/ --handoff-remaining-ms=([1-9][0-9]{0,4})$/); -const child=cp.spawn(process.execPath,[process.env.FIXTURE_CLI,'--stdio-request',...(selector?['--handoff-remaining-ms='+selector[1]]:[])],{env,cwd:env.HOME,stdio:['pipe','pipe','inherit']}); +const child=cp.spawn(process.execPath,['--import',process.env.FIXTURE_DIAGNOSTIC_PRELOAD,process.env.FIXTURE_CLI,'--stdio-request',...(selector?['--handoff-remaining-ms='+selector[1]]:[])],{env,cwd:env.HOME,stdio:['pipe','pipe','inherit']}); let output='';child.stdout.on('data',part=>{output+=part;process.stdout.write(part);}); child.stdin.end(input);child.on('close',(code,signal)=>{if(request.handoff?.preparation)fs.appendFileSync(process.env.FIXTURE_PREPARATIONS,output);if(signal)process.kill(process.pid,signal);else process.exitCode=request.handoff?.preparation&&process.env.FIXTURE_PREPARE_EXIT255?255:code;}); `, { mode: 0o700 }); - const common = { ...process.env, CODEX_HOME: '', CODEX_THREAD_ID: '', CODEX_SESSION_ID: '', SESSION_PEER_TAILSCALE: 'off', SESSION_PEER_UPDATE_NOTICE: '', + const common = { ...process.env, SESSION_PEER_TEST_OWNER_PROBE_LOG:probeLog,FIXTURE_DIAGNOSTIC_PRELOAD:diagnosticPreload,CODEX_HOME: '', CODEX_THREAD_ID: '', CODEX_SESSION_ID: '', SESSION_PEER_TAILSCALE: 'off', SESSION_PEER_UPDATE_NOTICE: '', PATH: root + delimiter + process.env.PATH, FIXTURE_VERSION: VERSION_LINE, FIXTURE_LOG: log, FIXTURE_PREPARATIONS: preparationLog, FIXTURE_CLI: cli, FIXTURE_DEST: destination }; const destEnv = { ...common, HOME: destination, USERPROFILE: destination, CLAUDE_CONFIG_DIR: join(destination, '.claude'), SESSION_PEER_HANDOFF_HOME: join(destination, 'state', 'handoff') }; const originEnv = { ...common, HOME: origin, USERPROFILE: origin, CLAUDE_CONFIG_DIR: join(origin, '.claude'), SESSION_PEER_HANDOFF_HOME: join(origin, 'state', 'handoff') }; async function invoke(args: string[], destinationSide = false, input?: string, extra: NodeJS.ProcessEnv = {}) { - const child = spawn(process.execPath, [cli, ...args], { env: { ...(destinationSide ? destEnv : originEnv), ...extra }, cwd: destinationSide ? destination : origin }); + const child = spawn(process.execPath, ['--import',diagnosticPreload,cli, ...args], { env: { ...(destinationSide ? destEnv : originEnv), ...extra }, cwd: destinationSide ? destination : origin }); let stdout = '', stderr = ''; child.stdout.on('data', part => { stdout += part; }); child.stderr.on('data', part => { stderr += part; }); child.stdin.end(input); // Harness margin permits cleanup/reporting; the request budget is unchanged. const timer = setTimeout(() => child.kill('SIGKILL'), 20000); const [code, signal] = await once(child, 'close'); clearTimeout(timer); assert.equal(signal, null, stderr); return { code, value: JSON.parse(stdout), stdout, stderr }; } - return { root, destination, origin, destEnv, messages, invoke, calls: () => readFileSync(log, 'utf8').trim().split('\n').filter(Boolean).map(line => JSON.parse(line)), + return { root, destination, origin, destEnv, messages, invoke, ownerProbes, probeCaptures, calls: () => readFileSync(log, 'utf8').trim().split('\n').filter(Boolean).map(line => JSON.parse(line)), preparations: () => readFileSync(preparationLog, 'utf8').trim().split('\n').filter(Boolean).map(line => JSON.parse(line)), args: ['send', '--host', 'fixture', '--to', String(process.pid), '--message=fixture body', '--no-from', '--no-reply-to', '--json'] }; } @@ -195,8 +199,9 @@ for (const mode of ['inject', 'none', 'loss'] as const) { test(`qualified SSH Codex ${mode} keeps destination preparation binding and at most one native queue`, qualified, async t => { const f = await codexFixture(t, mode), started = performance.now(); const sent = await f.invoke(f.args, false, undefined, f.env); - assert.equal(sent.code, mode === 'inject' ? 0 : 1, sent.stdout); - const calls = f.calls(); assert.equal(calls.length, 2); + const diagnostic=JSON.stringify({result:sent.value,queueCalls:f.queueCalls(),nativeCalls:f.nativeCalls(),ownerProbes:f.ownerProbes(),ownerProbeCaptures:f.probeCaptures()}); + assert.equal(sent.code, mode === 'inject' ? 0 : 1, diagnostic); + const calls = f.calls(); assert.equal(calls.length, 2,diagnostic); assert.equal(calls[0].handoff.preparation, true); assert.equal(calls[1].handoff.preparation, undefined); assert.ok(calls[1].handoff.remainingMs < calls[0].handoff.remainingMs); const original = f.preparations()[0], initial = validateHandoff(original.handoff); @@ -206,7 +211,7 @@ for (const mode of ['inject', 'none', 'loss'] as const) { assert.equal(calls[1].args[calls[1].args.indexOf('--correlation-id') + 1], initial.correlationId); assert.equal(calls[1].args[calls[1].args.indexOf('--codex-home') + 1], original.codexHome); assert.equal(calls[1].args[calls[1].args.indexOf('--to') + 1], `codex:${original.target.id}`); - assert.deepEqual(f.queueCalls(), [{ method: 'thread/queue/add', clientId: initial.correlationId, threadId: thread, codexHome: f.home }]); + assert.deepEqual(f.queueCalls(), [{ method: 'thread/queue/add', clientId: initial.correlationId, threadId: thread, codexHome: f.home }],diagnostic); assert.equal(f.nativeCalls().some(call => call.command === 'queue'), false); assert.equal(f.messages.length, 0); assert.equal(existsSync(join(f.origin, 'state', 'handoff')), false); const retained = f.ledger().data.records[initial.correlationId];