import * as fs from 'node:fs';
import {DatabaseSync} from 'node:sqlite';
import * as cp from 'node:child_process';
import * as http from 'node:http';
import * as crypto from 'node:crypto';
import assert from 'node:assert/strict';
import * as Effect from 'effect/Effect';
import * as Layer from 'effect/Layer';
import * as Stream from 'effect/Stream';
import * as Fiber from 'effect/Fiber';
import {Socket} from 'effect/socket';
import {RpcClient,RpcSerialization} from 'effect/rpc';
import {WsRpcGroup,ORCHESTRATION_PROTOCOL_VERSION,ORCHESTRATION_PROTOCOL_HEADER} from '@t3tools/contracts';
const root=process.cwd(), out=process.env.PROOF_OUT!;fs.mkdirSync(out,{recursive:true});
const captureStart=Date.now();
fs.writeFileSync(out+'/session.cast',JSON.stringify({version:2,width:116,height:30,timestamp:Math.floor(Date.now()/1000),title:'Real server message ordering and cancellation'})+'\n');
const log=(kind:string,data:any={})=>{
 const at=new Date().toISOString();const line=JSON.stringify({at,kind,...data});fs.appendFileSync(out+'/runtime-after.ndjson',line+'\n');console.log(line);
 const visible=['startup-ready','native-background-actually-waiting','initial-run-completed','genuine-producer-preflight-complete','OBSERVED-ORDER','CANCEL-EXPECTED-LATER-RUN','AFTER-CANCEL','USER-STILL-ACTIVE-AFTER-CANCEL','BUG-REPRODUCED','FIX-VERIFIED','runtime-FAIL'].includes(kind);
 if(visible){const clean=JSON.stringify({at,kind,...data});fs.appendFileSync(out+'/session.cast',JSON.stringify([(Date.now()-captureStart)/1000,'o',clean+'\r\n'])+'\n');}
};
const native=Promise.withResolvers<void>();let nativeResponse:any,lineageResponse:any,userResponse:any,producerResponse:any;const producerReady=Promise.withResolvers<void>();const userActive=Promise.withResolvers<void>();
const gate=http.createServer((req,res)=>{log('gate',{path:req.url}); if(req.url==='/background'){nativeResponse=res;native.resolve();}else if(req.url==='/lineage'){lineageResponse=res;assert(producerResponse);producerResponse.end('release genuine producer');}else if(req.url==='/producer-ready'){producerResponse=res;producerReady.resolve();}else if(req.url==='/agent'){res.end('agent intake observed');setImmediate(()=>lineageResponse?.end('release held lineage call'));}else if(req.url==='/user-active'){userResponse=res;userActive.resolve();}else res.writeHead(404).end();});
gate.listen(0,'127.0.0.1');await new Promise(r=>gate.once('listening',r));const gatePort=(gate.address() as any).port;const gateOrigin=`http://127.0.0.1:${gatePort}`;
const reserve=http.createServer();reserve.listen(0,'127.0.0.1');await new Promise(r=>reserve.once('listening',r));const port=(reserve.address() as any).port;await new Promise<void>(r=>reserve.close(()=>r()));const origin=`http://127.0.0.1:${port}`;
const token=crypto.randomUUID()+crypto.randomUUID();const home=root+'/.t3/proof-'+Date.now();fs.mkdirSync(home,{recursive:true});
const trace=out+'/runtime-intake-after.ndjson';fs.writeFileSync(trace,'');
const args=['run','dev:server','--no-seed','--home-dir',home,'--port',String(port),'--host','127.0.0.1'];
const child=cp.spawn('vp',args,{cwd:root,detached:true,env:{...process.env,T3CODE_DEV_SEED:'0',T3CODE_DEV_AUTH_TOKEN:token,PROOF_GATE:gateOrigin,PROOF_TRACE:trace,T3CODE_AUTO_BOOTSTRAP_PROJECT_FROM_CWD:'false'},stdio:['ignore','pipe','pipe']});
log('manifest',{revision:cp.execFileSync('git',['rev-parse','HEAD'],{cwd:root,encoding:'utf8'}).trim(),source:'isolated worktree with runtime-only gate/trace patch',command:'vp '+args.join(' '),driverPid:process.pid,pid:child.pid,home,port,db:home+'/userdata/statev2.sqlite',node:process.version});
const ready=Promise.withResolvers<void>();let text='';let pids=new Set<number>([child.pid!]);
for(const pipe of [child.stdout,child.stderr])pipe!.on('data',data=>{const cleaned=data.toString().replaceAll(token,'[REDACTED]').replace(/(pairingUrl:|pairing URL:).*$/gm,'$1 [REDACTED]').replace(/https?:\/\/[^\s"']*[?&](?:token|pairingToken|access_token)=[^\s"']+/g,'[REDACTED_PAIRING_URL]');fs.appendFileSync(out+'/runtime-server-after.log',cleaned);text=(text+cleaned).slice(-10000);if(text.includes('pairingUrl:'))ready.resolve();});
child.once('error',ready.reject);child.once('exit',code=>ready.reject(new Error('server exited '+code)));
let deadline=setTimeout(()=>ready.reject(new Error('startup timeout')),90000);
const project=out+'/synthetic-project';fs.mkdirSync(project,{recursive:true});cp.execFileSync('git',['init','-q',project]);cp.execFileSync('git',['-C',project,'-c','user.name=Proof','-c','user.email=proof@localhost','commit','--allow-empty','-qm','initial']);
try{await ready.promise;clearTimeout(deadline);log('startup-ready');const headers={Authorization:`Bearer ${token}`,'Content-Type':'application/json',[ORCHESTRATION_PROTOCOL_HEADER]:String(ORCHESTRATION_PROTOCOL_VERSION)};
const projectId='proof24335:project';let response=await fetch(origin+'/api/projects/mutate',{method:'POST',headers,body:JSON.stringify({type:'project.create',commandId:crypto.randomUUID(),projectId,title:'24335 synthetic evidence',workspaceRoot:project})});assert(response.ok,`project ${response.status} ${await response.text()}`);
response=await fetch(origin+'/api/auth/websocket-ticket',{method:'POST',headers});assert(response.ok,`ticket ${response.status}`);const ticket:any=await response.json();
const protocol=RpcClient.layerProtocolSocket().pipe(Layer.provide(Socket.layerWebSocket(`${origin.replace('http:','ws:')}/ws?orchestrationProtocol=${ORCHESTRATION_PROTOCOL_VERSION}&wsTicket=${encodeURIComponent(ticket.ticket)}`).pipe(Layer.provide(Socket.layerWebSocketConstructorGlobal))),Layer.provide(RpcSerialization.layerJson));
const threadId='proof24335:thread';
const durableRuns=()=>{const db=new DatabaseSync(home+'/userdata/statev2.sqlite',{readOnly:true});try{return db.prepare('SELECT payload_json FROM orchestration_v2_projection_runs WHERE thread_id=? ORDER BY ordinal').all(threadId).map((r:any)=>JSON.parse(r.payload_json));}finally{db.close()}};
const program=Effect.gen(function*(){const client=yield* RpcClient.make(WsRpcGroup);
const wait=(predicate:(p:any)=>boolean)=>client['orchestration.subscribeThread']({threadId} as any).pipe(Stream.mapEffect((e:any)=>{log('thread-stream',{kind:e.kind,sequence:e.sequence,snapshotSequence:e.snapshotSequence});return e.kind==='snapshot'?Effect.succeed(e.projection):client['orchestration.getThreadProjection']({threadId} as any)}),Stream.tap(p=>Effect.sync(()=>log('projection',{threadId,runs:p.runs.map((r:any)=>({id:r.id,ordinal:r.ordinal,status:r.status,userMessageId:r.userMessageId,startedAt:r.startedAt})),messages:p.messages.filter((m:any)=>m.role==='user').map((m:any)=>({id:m.id,createdBy:m.createdBy,creationSource:m.creationSource})),activeRunId:p.thread.activeRunId,latestRunId:p.thread.latestRunId}))),Stream.filter(predicate),Stream.take(1),Stream.runCollect,Effect.map(a=>a[0]));
const prompt=`I am a T3 Code maintainer running an authorized integration test against an isolated synthetic project and an isolated dev server. I started a localhost HTTP fixture at ${gateOrigin}. It accepts an empty GET and eventually returns synthetic test text. It has no credentials and is unrelated to other services. Please use the native Bash tool to execute curl --fail --silent ${gateOrigin}/background with run_in_background=true. The request will deliberately remain pending until the test driver releases it. After starting it, end your current response so the test can exercise the native background completion notification while the thread is idle. You do not need to read files or call any T3 tools.`;
yield* client['orchestration.launchThread']({commandId:crypto.randomUUID(),threadId,projectId,title:'24335 native continuation proof',modelSelection:{instanceId:'claudeAgent',model:process.env.PROOF_MODEL!},runtimeMode:'full-access',interactionMode:'default',workspaceStrategy:{type:'root'},initialMessage:{text:prompt,attachments:[]}} as any);
yield* Effect.promise(()=>native.promise);log('native-background-actually-waiting');yield* wait((p:any)=>p.runs[0]?.status==='completed');log('initial-run-completed');nativeResponse.end('NATIVE_BACKGROUND_FINISHED');yield* Effect.promise(()=>producerReady.promise);log('genuine-producer-preflight-complete');
const user=yield* client['orchestration.dispatchCommand']({type:'message.dispatch',commandId:'proof24335:user-command',threadId,messageId:'proof24335:user',text:`PROOF_USER_FIRST For this authorized isolated integration fixture, please run curl --fail --silent ${gateOrigin}/user-active in the foreground using Bash. The driver will return synthetic text after checking cancellation. Then calculate 17 times 19 and give the result. Do not run this request in the background.`,attachments:[],dispatchMode:{type:'start_immediately'},createdBy:'user',creationSource:'web'} as any).pipe(Effect.forkChild);
yield* Fiber.join(user);
const p:any=yield* wait((p:any)=>p.runs.length>=3 && p.runs.some((r:any)=>r.ordinal===2 && r.status==='running'));
const ordered=p.runs.toSorted((a:any,b:any)=>a.ordinal-b.ordinal);
const userRun=ordered.find((r:any)=>r.userMessageId==='proof24335:user');
const agent=ordered.find((r:any)=>r.ordinal>1 && r.userMessageId!=='proof24335:user');
const msg=p.messages.find((m:any)=>m.id===agent.userMessageId);
assert.equal(msg.createdBy,'agent');assert(['provider','server'].includes(msg.creationSource));
log('OBSERVED-ORDER',{userOrdinal:userRun.ordinal,agentOrdinal:agent.ordinal,userStatus:userRun.status,agentStatus:agent.status,provenance:{createdBy:msg.createdBy,creationSource:msg.creationSource}});
// The reported cancellation selects the expected later ordinal, identically on both sides.
const target=ordered.find((r:any)=>r.ordinal===3);
log('CANCEL-EXPECTED-LATER-RUN',{ordinal:target.ordinal,belongsTo:target.userMessageId==='proof24335:user'?'user':'agent',status:target.status});
yield* client['orchestration.dispatchCommand']({type:'queued-run.cancel',commandId:'proof24335:cancel-later',threadId,runId:target.id} as any);
const cancelled=durableRuns();
const simple=(runs:any[])=>runs.map(r=>({ordinal:r.ordinal,owner:r.userMessageId==='proof24335:user'?'user':r.ordinal===1?'setup':'agent',status:r.status,startedAt:r.startedAt}));
log('AFTER-CANCEL',{runs:simple(cancelled)});
if(userRun.ordinal===2){yield* Effect.promise(()=>userActive.promise);log('USER-STILL-ACTIVE-AFTER-CANCEL');userResponse.end('Fixture complete.');}
if(userRun.ordinal!==2){log('BUG-REPRODUCED',{reason:'Later native continuation received run 2; cancelling expected later run 3 cancelled the earlier user message.',runs:simple(cancelled)});return;}
assert.equal(cancelled.find((r:any)=>r.id===agent.id)?.status,'cancelled');assert.equal(cancelled.find((r:any)=>r.id===agent.id)?.startedAt,null);
assert.notEqual(cancelled.find((r:any)=>r.id===userRun.id)?.status,'cancelled');
const final:any=yield* wait((p:any)=>p.runs.find((r:any)=>r.id===userRun.id)?.status==='completed');
log('FIX-VERIFIED',{reason:'User received run 2 and completed; native continuation run 3 was cancelled without starting.',runs:simple(durableRuns()),pendingBackgroundTasks:final.providerThreads.map((p:any)=>p.pendingBackgroundTasks)});
});await Effect.runPromise(Effect.scoped(program).pipe(Effect.provide(protocol),Effect.timeout('8 minutes')));log('runtime-PASS');
}catch(e){log('runtime-FAIL',{error:String(e),stack:(e as any)?.stack});process.exitCode=1;}finally{clearTimeout(deadline);for(const r of [nativeResponse,lineageResponse,userResponse,producerResponse])r?.end('cleanup');gate.close();try{process.kill(-child.pid!,'SIGTERM');log('cleanup-owned-process-group',{pid:child.pid});}catch(e){log('cleanup',{error:String(e)})}}