import { expect } from "e2e";
import { test, WEB } from "./t3.ts";
import { mkdirSync, writeFileSync } from 'node:fs';
import { execFileSync } from 'node:child_process';
import { join } from 'node:path';
import { APP_DIR, T3_HOME } from '../e2e.config.ts';

test('CodeRabbit acknowledgement stays idle; edited outcome wakes once', {...WEB, timeout: 1_200_000}, async ({ t3, app, screen, browser }) => {
  await fetch('http://127.0.0.1:18991/phase/initial');
  const project = await t3.project('Review-watch');
  await t3.thread({ project, title: 'Watch synthetic review', messages: ['Watch this review and wait.'], model: 'multi-turn' });
  await t3.open();
  await browser.locator('li[data-thread-item]').first().tap();
  const rpc = async (tag: string, payload: unknown) => browser.evaluate(async ({tag,payload}) => {
    return await new Promise<any>((resolve,reject)=>{
      const ws=new WebSocket(location.origin.replace(/^http/,'ws')+'/ws?orchestrationProtocol=2');
      const timer=setTimeout(()=>{ws.close();reject(new Error('RPC timeout '+tag));},15000);
      ws.onopen=()=>ws.send(JSON.stringify({_tag:'Request',id:'777',tag,payload,headers:[]}));
      ws.onmessage=(event)=>{for(const msg of String(event.data).trim().split('\n').map(x=>JSON.parse(x))){
        if(msg._tag==='Ping') ws.send(JSON.stringify({_tag:'Pong'}));
        if(msg.requestId==='777' && msg._tag==='Exit'){clearTimeout(timer);ws.close(); if(msg.exit._tag==='Success')resolve(msg.exit.value);else reject(new Error(JSON.stringify(msg)));}
      }};
      ws.onerror=()=>{clearTimeout(timer);reject(new Error('Websocket failed'));};
    });
  }, {tag,payload});
  const path=await browser.evaluate(()=>location.pathname);
  const output=join(T3_HOME,'..','review-watch-evidence');
  mkdirSync(output,{recursive:true});
  const commit=execFileSync('git',['rev-parse','HEAD'],{cwd:APP_DIR,encoding:'utf8'}).trim();
  const threadId=decodeURIComponent(path.split('/').at(-1)!);
  await rpc('orchestration.dispatchCommand',{type:'thread.pull-request.watch',commandId:crypto.randomUUID(),threadId,host:'github.com',repository:'scenario/demo',number:1,watching:true,link:{url:'https://github.com/scenario/demo/pull/1',source:'agent'}});
  const readWakeTexts = () => JSON.parse(execFileSync('sqlite3', [
    '-readonly', '-json', join(T3_HOME,'userdata/statev2.sqlite'),
    "SELECT json_extract(payload_json, '$.text') AS text FROM orchestration_v2_projection_messages WHERE role='user' ORDER BY created_at;",
  ], {encoding:'utf8'})).map((m:any)=>m.text).filter((text:string)=>text.startsWith('Update on pull request'));
  const waitFor = async (predicate: (s:any)=>boolean) => {
    for(let i=0;i<180;i++) {
      const s=await rpc('orchestration.getThreadProjection',{threadId});
      if(predicate(s)) return {...s, persistedWakeTexts:readWakeTexts()};
      await new Promise(r=>setTimeout(r,2000));
    }
    throw new Error('Watch milestone did not arrive');
  };
  const watchOf = (s:any) => s.thread.pullRequests[0]?.watch;
  const idle = (s:any) => s.runs.every((r:any)=>['completed','failed','interrupted','cancelled'].includes(r.status));
  await waitFor(s=>watchOf(s)?.headSha != null && idle(s));
  const initial = {...await rpc('orchestration.getThreadProjection',{threadId}), persistedWakeTexts:readWakeTexts()};
  await app.screenshot('watch-started');
  await fetch('http://127.0.0.1:18991/phase/ack');
  const ack = await waitFor(s=>watchOf(s)?.remarksThrough==='2090-01-01T00:00:00Z' && idle(s));
  await app.screenshot('acknowledgement');
  await fetch('http://127.0.0.1:18991/phase/outcome');
  const outcome = await waitFor(s=>watchOf(s)?.remarksThrough==='2090-01-01T00:01:00Z' && idle(s));
  await app.screenshot('finished-outcome');
  const counts = () => fetch('http://127.0.0.1:18991/counts').then(r=>r.json());
  const through = (await counts()).fingerprints;
  for(let i=0;i<90;i++) {
    if((await counts()).fingerprints > through) break;
    if(i===89) throw new Error('No unchanged watch sweep observed');
    await new Promise(r=>setTimeout(r,2000));
  }
  await new Promise(r=>setTimeout(r,2000));
  const unchanged = {...await rpc('orchestration.getThreadProjection',{threadId}), persistedWakeTexts:readWakeTexts()};
  await app.screenshot('unchanged-sweep');
  const summarize = (s:any) => ({
    commentWakes:watchOf(s).wakes,
    watermark:watchOf(s).remarksThrough,
    remarkIds:watchOf(s).remarkIds,
    wakes:s.persistedWakeTexts,
    turns:s.runs.length,
    idle:idle(s),
  });
  const observed={initial:summarize(initial),acknowledgement:summarize(ack),finished:summarize(outcome),unchanged:summarize(unchanged)};
  writeFileSync(`${output}/observed-${commit}.json`,JSON.stringify(observed,null,2));
  expect({
    ackWakes:watchOf(ack).wakes,
    ackTurns:ack.runs.length-initial.runs.length,
    outcomeWakes:watchOf(outcome).wakes,
    outcomeTurns:outcome.runs.length-initial.runs.length,
    unchangedWakes:watchOf(unchanged).wakes,
    unchangedTurns:unchanged.runs.length-initial.runs.length,
    outcomeDelivered:outcome.persistedWakeTexts.filter((text:string)=>text.includes('Review complete. No actionable findings.')).length,
    unchangedDelivered:unchanged.persistedWakeTexts.filter((text:string)=>text.includes('Review complete. No actionable findings.')).length,
  }).toEqual({ackWakes:0,ackTurns:0,outcomeWakes:1,outcomeTurns:1,unchangedWakes:1,unchangedTurns:1,outcomeDelivered:1,unchangedDelivered:1});
});