import { expect } from "e2e";
import { test, WEB } from "./t3.ts";
import { APP_DIR, T3_HOME } from "../e2e.config.ts";
import { execFile, spawn } from "node:child_process";
import { promisify } from "node:util";
import { mkdir, writeFile, readFile } from "node:fs/promises";
import http from "node:http";
import { once } from "node:events";
const exec = promisify(execFile);
const delay = (ms: number) => new Promise(r => setTimeout(r, ms));
const root = process.env.HANDSHAKE_PROOF_SCRATCH ?? "/tmp/linked-handshake-proof";
const cli = async (args: string[]) => (await exec(process.execPath, ["apps/server/src/bin.ts", ...args], {cwd: APP_DIR, env: {...process.env, T3CODE_DEV_SEED: "0"}, maxBuffer: 2e6})).stdout;
async function auth(home: string, kind: "pair" | "mcp") {
  const script = `import * as E from 'effect/Effect'; import * as L from 'effect/Layer'; import * as N from '@effect/platform-node/NodeServices'; import * as Crypto from '@effect/platform-node/NodeCrypto'; import * as C from './src/config.ts'; import * as A from './src/auth/EnvironmentAuth.ts';
  const value = await E.runPromise(E.gen(function*(){const a=yield* A.EnvironmentAuth; return yield* ${kind === "pair" ? "a.createPairingLink({subject:'one-time-token'})" : "a.issueMcpClientSession({label:'Handshake proof',access:'full-access',peerLink:false,client:{deviceType:'bot'}})"};}).pipe(E.provide(A.layerRuntime.pipe(L.provide(C.layerTest(process.cwd(),${JSON.stringify(home)})))),E.provide([N.layer,Crypto.layer]),E.scoped)); console.log(JSON.stringify(value));`;
  const result = await exec(process.execPath, ["--input-type=module", "-e", script], {cwd: `${APP_DIR}/apps/server`, maxBuffer: 2e6});
  return JSON.parse(result.stdout.trim().split("\n").at(-1)!);
}
async function mcp(origin: string, token: string) {
  const headers: Record<string,string> = {authorization:`Bearer ${token}`, "content-type":"application/json",accept:"application/json, text/event-stream"};
  const post = (body: unknown, signal?: AbortSignal) => fetch(`${origin}/mcp`, {method:"POST",headers,body:JSON.stringify(body), signal});
  const initialized = await post({jsonrpc:"2.0",id:0,method:"initialize",params:{protocolVersion:"2025-06-18",capabilities:{},clientInfo:{name:"handshake-proof",version:"1"}}});
  headers["mcp-session-id"] = initialized.headers.get("mcp-session-id")!;
  headers["mcp-protocol-version"] = "2025-06-18";
  await post({jsonrpc:"2.0",method:"notifications/initialized"});
  return post;
}
const agentSource = (environmentId: string, run: string) => `
import readline from 'node:readline';
import {writeFile} from 'node:fs/promises';
import crypto from 'node:crypto';
const sessions=new Map();
const send=message=>process.stdout.write(JSON.stringify({jsonrpc:'2.0',...message})+'\\n');
async function handle(message){
 const p=message.params??{};
 if(message.method==='initialize')return {protocolVersion:1,agentCapabilities:{loadSession:true,promptCapabilities:{embeddedContext:true}},authMethods:[],agentInfo:{name:'Handshake replay',version:'1'}};
 if(message.method==='session/new'||message.method==='session/load'){
  const id=p.sessionId??crypto.randomUUID();
  const server=p.mcpServers.find(s=>s.name==='t3-code');
  const env=Object.fromEntries((server?.env??[]).map(e=>[e.name,e.value]));
  sessions.set(id,{endpoint:env.T3_ACP_MCP_ENDPOINT??server?.url,authorization:env.T3_ACP_MCP_AUTHORIZATION??server?.headers?.find(h=>h.name.toLowerCase()==='authorization')?.value});
  return {sessionId:id,models:{currentModelId:'showcase',availableModels:[{modelId:'showcase',name:'Handshake replay'}]}};
 }
 if(message.method==='session/prompt'){
  const session=sessions.get(p.sessionId);
  const prompt=p.prompt.map(c=>c.text??'').join('');
  const stage=prompt.includes('notification')?'notifications-initialized':'initialize';
  const controller=new AbortController();session.cancel=()=>controller.abort();
  const update=u=>send({method:'session/update',params:{sessionId:p.sessionId,update:u}});
  const toolCallId=crypto.randomUUID();
  const args={environmentId:${JSON.stringify(environmentId)},allProjects:true};
  update({sessionUpdate:'tool_call',toolCallId,title:'List threads on Peer B',kind:'fetch',status:'in_progress',rawInput:args});
  const started=Date.now();
  try{
   if(!session.endpoint||!session.authorization)throw Error('No injected MCP endpoint');
   const headers={authorization:session.authorization,'content-type':'application/json',accept:'application/json, text/event-stream'};
   const post=body=>fetch(session.endpoint,{method:'POST',headers,body:JSON.stringify(body),signal:controller.signal});
   const init=await post({jsonrpc:'2.0',id:0,method:'initialize',params:{protocolVersion:'2025-06-18',capabilities:{},clientInfo:{name:'handshake-replay',version:'1'}}});
   headers['mcp-session-id']=init.headers.get('mcp-session-id');headers['mcp-protocol-version']='2025-06-18';
   await post({jsonrpc:'2.0',method:'notifications/initialized'});
   const response=await post({jsonrpc:'2.0',id:1,method:'tools/call',params:{name:'t3_thread_list',arguments:args}});
   const raw=await response.text();
   const json=JSON.parse(raw.startsWith('data:')?raw.split('\\n').filter(l=>l.startsWith('data:')).map(l=>l.slice(5)).join(''):raw);
   const text=json.result.content.map(c=>c.text??'').join('');
   let detail;try{detail=JSON.parse(text).message??text}catch{detail=text}
   await writeFile(${JSON.stringify(run)}+'/'+stage+'-agent.json',JSON.stringify({outcome:detail,raw,callElapsedMs:Date.now()-started}));
   update({sessionUpdate:'tool_call_update',toolCallId,status:json.result.isError?'failed':'completed',content:[{type:'content',content:{type:'text',text:detail}}],rawOutput:json.result});
   update({sessionUpdate:'agent_message_chunk',content:{type:'text',text:detail}});
   return {stopReason:'end_turn'};
  }catch(error){
   if(controller.signal.aborted)return {stopReason:'cancelled'};
   update({sessionUpdate:'agent_message_chunk',content:{type:'text',text:String(error)}});
   throw error;
  }
 }
 if(message.method==='session/cancel'){sessions.get(p.sessionId)?.cancel?.();return {};}
 return {};
}
readline.createInterface({input:process.stdin}).on('line',line=>{const message=JSON.parse(line);if(!message.method)return;Promise.resolve(handle(message)).then(result=>{if(message.id!==undefined)send({id:message.id,result})},error=>{if(message.id!==undefined)send({id:message.id,error:{code:-32603,message:String(error)}})});});
`;
async function stableCapture(app: {screenshot: (label: string) => Promise<string>}, label: string) {
  for (let n=0;n<10;n++) {
    const first=await app.screenshot(`${label}-stability-${n}`);
    await delay(500);
    const second=await app.screenshot(label);
    if ((await readFile(first)).equals(await readFile(second))) return;
  }
  throw new Error("Settings screenshots did not stabilize");
}
test("Linked MCP handshake stalls record a timeout", WEB, async ({t3, app, screen}) => {
  const run = `${root}/${Date.now()}`;
  await mkdir(`${run}/peer/userdata`, {recursive:true});
  await mkdir(`${run}/bin`, {recursive:true});
  await writeFile(`${run}/bin/hostnamectl`, '#!/bin/sh\necho "Peer B"\n', {mode:0o755});
  await writeFile(`${run}/peer/userdata/settings.json`, JSON.stringify({linkedEnvironmentsEnabled:true}));
  const peerPort = 22000 + Math.floor(Math.random()*10000);
  const log = await import("node:fs").then(fs => fs.openSync(`${run}/peer.log`, "w"));
  const peer = spawn(process.execPath, ["scripts/dev-runner.ts","dev:server","--home-dir",`${run}/peer`,"--no-seed","--port",String(peerPort)], {cwd: APP_DIR,env:{...process.env,PATH:`${run}/bin:${process.env.PATH}`,T3CODE_DEV_SEED:"0",T3CODE_AUTO_BOOTSTRAP_PROJECT_FROM_CWD:"0"},stdio:["ignore",log,log],detached:true});
  let stage = "forward";
  const arrivals: string[] = [];
  let arrived: (() => void) | undefined;
  const sockets = new Set<any>();
  const proxy = http.createServer(async (req,res) => {
    const chunks=[]; for await(const chunk of req) chunks.push(chunk);
    const body=Buffer.concat(chunks);
    const method = req.url === "/mcp" ? JSON.parse(body.toString() || "{}").method : "descriptor";
    arrivals.push(method);
    if(method===stage) { arrived?.(); return; }
    if(method === "tools/call" && stage === "slow-tool") await delay(4500);
    const upstream=http.request({hostname:"127.0.0.1",port:peerPort,path:req.url,method:req.method,headers:req.headers}, answer => {res.writeHead(answer.statusCode!,answer.headers);answer.pipe(res);});
    upstream.on("error", e=>{res.writeHead(502);res.end(String(e));});
    upstream.end(body);
  });
  proxy.on("connection", socket=>{sockets.add(socket);socket.on("close",()=>sockets.delete(socket));});
  proxy.listen(0,"127.0.0.1"); await once(proxy,"listening");
  const proxyOrigin=`http://127.0.0.1:${(proxy.address() as any).port}`;
  const controllers: AbortController[]=[];
  try {
    let ready=false; for(let n=0;n<120;n++){try{if((await fetch(`http://127.0.0.1:${peerPort}/.well-known/t3/environment`)).ok){ready=true;break;}}catch{} await delay(500);}
    if(!ready) throw new Error("Peer server did not start");
    const pairing=await auth(`${run}/peer`,"pair");
    await cli(["environment","link",proxyOrigin,"--pairing-code",pairing.credential,"--base-dir",T3_HOME]);
    const linked=JSON.parse(await cli(["environment","list","--json","--base-dir",T3_HOME]));
    const environmentId=linked[0].environmentId;
    const project=await t3.project("Handshake demo");
    await t3.thread({project,title:"List threads on Peer B",messages:["Prepare to list threads on the linked environment."]});
    const runtime=JSON.parse(await readFile(`${T3_HOME}/userdata/server-runtime.json`,"utf8"));
    const credentials=await auth(T3_HOME,"mcp");
    const post=await mcp(runtime.origin,credentials.token);
    await t3.open();
    await screen.getByText("Handshake demo",{exact:true}).first().tap();
    await app.open("/settings/connections");
    await screen.getByText("Peer B",{exact:true}).first().tap();
    const localId=JSON.parse(await (await fetch(`${runtime.origin}/.well-known/t3/environment`)).text()).environmentId;
    await writeFile(`${run}/agent.mjs`,agentSource(environmentId,run));
    const settingsPath=`${T3_HOME}/userdata/settings.json`;
    const settings=JSON.parse(await readFile(settingsPath,"utf8"));
    settings.providerInstances.devReplay.config={source:"local",commandPath:process.execPath,commandArgs:[`${run}/agent.mjs`]};
    await writeFile(settingsPath,JSON.stringify(settings));
    await delay(1500);
    const outcomes=[];
    for(const handshake of ["initialize","notifications/initialized"]){
      stage=handshake;
      const key=handshake === "initialize" ? "initialize" : "notifications-initialized";
      const reached = new Promise<void>(resolve=>{arrived=resolve;});
      const message=handshake === "initialize" ? "List threads on Peer B with initialize stalled" : "List threads on Peer B with initialized notification stalled";
      const launched=JSON.parse((await cli(["thread","start",project,message,"--title",message,"--provider","devReplay","--model","showcase","--json","--base-dir",T3_HOME])).trim().split("\n").at(-1)!);
      await app.open(`/${localId}/${launched.threadId}`);
      await delay(1000);
      await Promise.race([reached,delay(30000).then(()=>{throw new Error("Replay agent never reached the stalled handshake");})]);
      await delay(5500);
      const agent=await readFile(`${run}/${key}-agent.json`,"utf8").then(JSON.parse).catch(()=>({outcome:"pending",callElapsedMs:null}));
      await app.screenshot(`${key}-call-outcome`);
      await app.open("/settings/connections");
      await screen.getByText("Peer B",{exact:true}).first().tap();
      await delay(1000);
      await stableCapture(app,handshake === "initialize" ? "initialize-stalled" : "initialized-notification-stalled");
      const links=JSON.parse(await cli(["environment","list","--json","--base-dir",T3_HOME]));
      outcomes.push({stage:handshake,...agent,lastError:links[0].lastError,arrivals:[...arrivals]});
      if(agent.outcome === "pending"){
        await post({jsonrpc:"2.0",id:10,method:"tools/call",params:{name:"t3_thread_interrupt",arguments:{threadId:launched.threadId}}});
      }
    }
    stage="slow-tool";
    const slowStarted=Date.now();
    const slowResponse=await post({jsonrpc:"2.0",id:2,method:"tools/call",params:{name:"t3_thread_list",arguments:{environmentId,allProjects:true}}});
    const slowText=await slowResponse.text();
    await writeFile(`${run}/observations.json`,JSON.stringify({outcomes,slowTool:{elapsedMs:Date.now()-slowStarted,response:slowText}},null,2));
    expect(slowText).toContain("threads");
    for(const result of outcomes){
      expect(result.outcome).toContain("The linked environment stopped answering.");
      expect(result.lastError).toContain("timed out");
    }
  } finally {
    for(const controller of controllers)controller.abort();
    for(const socket of sockets)socket.destroy();proxy.close();
    if(peer.pid)process.kill(-peer.pid,"SIGTERM");
  }
});