import { assert, it } from "@effect/vitest";
import * as NodeServices from "@effect/platform-node/NodeServices";
import {
  CheckpointId,
  CheckpointRef,
  CheckpointScopeId,
  MessageId,
  GitCommandError,
  type ModelSelection,
  NodeId,
  type OrchestrationV2AppThread,
  type OrchestrationV2DomainEvent,
  type OrchestrationV2ExecutionNode,
  type OrchestrationV2RuntimeRequest,
  type OrchestrationV2ProviderThread,
  type OrchestrationV2Run,
  type OrchestrationV2TurnItem,
  ProviderDriverKind,
  ProviderInstanceId,
  ProviderThreadId,
  ProviderTurnId,
  RunAttemptId,
  RunId,
  RuntimeRequestId,
  TurnItemId,
} from "@t3tools/contracts";
import * as Cause from "effect/Cause";
import * as DateTime from "effect/DateTime";
import * as Deferred from "effect/Deferred";
import * as Fiber from "effect/Fiber";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as SqlClient from "effect/sql/SqlClient";
import * as SqlError from "effect/sql/SqlError";
import * as Stream from "effect/Stream";

import { toolOutputImages } from "@t3tools/shared/toolOutput";
import * as SqlitePersistence from "../persistence/Sqlite.ts";
import * as ServerConfig from "../config.ts";
import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
import * as EventSink from "./EventSink.ts";
import * as EventStore from "./EventStore.ts";
import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
import * as ProjectionStore from "./ProjectionStore.ts";
import * as After from "./ProviderEventIngestor.ts";
import * as Before from "./EvidenceBefore.ts";
import { spawn } from "node:child_process";
import { createInterface } from "node:readline";
import { appendFileSync } from "node:fs";
import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
import { makeProviderFailure } from "@t3tools/provider-core/server/failure";
import {
  makeProviderEventRoutingState,
  type ProviderEventRouteIdentity,
  routeProviderEvent,
  selectInheritedBackgroundTurnItems,
} from "./RunExecutionService.ts";

const modelSelection = {
  instanceId: ProviderInstanceId.make("codex"),
  model: "gpt-5.4",
} satisfies ModelSelection;
const CODEX_DRIVER = ProviderDriverKind.make("codex");

function threadCreatedEvent(
  now: DateTime.Utc,
): Effect.Effect<
  OrchestrationV2DomainEvent,
  IdAllocator.IdAllocatorV2Error,
  IdAllocator.IdAllocatorV2
> {
  return Effect.gen(function* () {
    const idAllocator = yield* IdAllocator.IdAllocatorV2;
    const projectId = yield* idAllocator.allocate.project({
      fixtureName: "provider-event-ingestor",
    });
    const threadId = yield* idAllocator.allocate.thread({
      fixtureName: "provider-event-ingestor",
      projectId,
    });
    const providerThreadId = idAllocator.derive.providerThread({
      driver: CODEX_DRIVER,
      nativeThreadId: "native-thread",
    });
    const thread: OrchestrationV2AppThread = {
      createdBy: "user",
      creationSource: "web",
      id: threadId,
      projectId,
      title: "Provider event ingestor",
      providerInstanceId: modelSelection.instanceId,
      modelSelection: modelSelection,
      runtimeMode: "full-access",
      interactionMode: "default",
      branch: null,
      worktreePath: null,
      branchPullRequest: null,
      activeOrderKey: null,
      activeProviderThreadId: providerThreadId,
      lineage: {
        parentThreadId: null,
        relationshipToParent: null,
        rootThreadId: threadId,
      },
      forkedFrom: null,
      createdAt: now,
      updatedAt: now,
      archivedAt: null,
      settledOverride: null,
      settledAt: null,
      lastVisitedAt: null,
      deletedAt: null,
    };

    return {
      id: yield* idAllocator.allocate.event({ threadId }),
      type: "thread.created",
      threadId,
      occurredAt: now,
      payload: thread,
    };
  });
}


import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
const scratch = "/tmp/ev-21092-capture-final";
const trace = (x: unknown) => appendFileSync(`${scratch}/runtime.jsonl`, JSON.stringify(x) + "\n");
const chain = (e: any): any => e == null ? null : ({tag:e._tag, message:e.message, code:e.code, cause:e.cause ? chain(e.cause):undefined, reason:e.reason?chain(e.reason):undefined});
const cases: any[] = [];
for (const revision of ['before','after']) {
 for (const route of ['P','R','O','S']) for (const mode of ['control','recover1','recover2','persistent','nonlock','mixed','injectedNonSql','injectedAbsent','injectedMixed']) cases.push({revision,route,mode,timeout:75});
 cases.push({revision,route:'P',mode:'recover1',timeout:5000});
 for (const route of ['R','O']) for (const mode of ['missing','wrongAttempt',...(route==='R'?['wrongStatus']:['wrongOrdinal']),'supersede']) cases.push({revision,route,mode,timeout:75});
 for (const mode of ['origin','nullChild','noModel','missingChild','equal','recheck','repeat','reportedSelection']) cases.push({revision,route:'S',mode,timeout:75});
 for (const mode of ['empty','ownerWins','ownerRejects']) cases.push({revision,route:'C',mode,timeout:75});
}
it.effect.each(cases)('$revision $route $mode busy_timeout=$timeout', (c) => { let holder: ReturnType<typeof spawn> | undefined; return Effect.gen(function*(){
 const key=`${c.revision}-${c.route}-${c.mode}-${c.timeout}`;
 const dbPath=`${scratch}/${key}.sqlite`;
 const ProviderEventIngestor=c.revision==='before'?Before:After;
 let active=false, attempts=0, primary=0;
 let request: ((x:any)=>Promise<any>)|undefined;
 const errors:any[]=[];
 const changes:string[]=[]; const concurrentIds:string[]=[];
 const startHolder=()=>Effect.promise(async()=>{
  holder=spawn('python3',[`${scratch}/holder.py`,dbPath],{stdio:['pipe','pipe','inherit']});
  const lines=createInterface({input:holder.stdout!});
  let waiter:((x:any)=>void)|undefined;
  lines.on('line',line=>{const row=JSON.parse(line);trace({key,holder:row});waiter?.(row);waiter=undefined;});
  request=(x)=>new Promise(resolve=>{waiter=resolve;holder!.stdin!.write(JSON.stringify(x)+'\n');});
  await request({op:'acquire'});
 });
 const release=()=>Effect.promise(async()=>{await request!({op:'release',sql:changes});});
 const targetType=c.route==='S'?'thread.model-selection-updated':c.route==='O'?'provider-thread.updated':'provider-turn.updated';
 const db=SqlitePersistence.layerFromPath(dbPath).pipe(Layer.provide(NodeServices.layer));
 const stores=Layer.merge(EventStore.layer,ProjectionStore.layer).pipe(Layer.provide(db));
 const sink=EventSink.layer.pipe(Layer.provide(Layer.mergeAll(stores,db)));
 const observed=Layer.effect(EventSink.EventSinkV2,Effect.gen(function*(){
  const real=yield* EventSink.EventSinkV2;
  const wrap=(method:string,input:any)=>Effect.gen(function*(){
   const targeted=active&&input.events.some((e:any)=>e.type===targetType);
   if(active&&input.events.some((e:any)=>e.type==='subagent.updated'))primary++;
   if(targeted){
    attempts++;
    if(attempts===1&&['recover1','recover2','persistent','mixed','supersede'].includes(c.mode))yield* startHolder();
    trace({key,milestone:'attempt',attempt:attempts,method,events:input.events.map((e:any)=>({id:e.id,type:e.type})),time:new Date().toISOString()});
   }
   const injected=c.mode.startsWith('injected')&&targeted;
   const injectedError=new EventSink.EventSinkWriteError({eventCount:1,...(c.mode==='injectedAbsent'?{}:{cause:c.mode==='injectedMixed'&&attempts===1?new SqlError.SqlError({reason:new SqlError.LockTimeoutError({cause:null})}):new Error('explicitly injected non-SQL proof')})});
   const result=yield* (injected?Effect.fail(injectedError):real[method](input)).pipe(Effect.tapError(e=>Effect.gen(function*(){
    if(targeted){errors.push(chain(e));trace({key,milestone:injected?'injected-error':'actual-error',attempt:attempts,error:chain(e),time:new Date().toISOString()});
     if((['recover1','mixed','supersede'].includes(c.mode)&&attempts===1)||(c.mode==='recover2'&&attempts===2))yield* release();
    }
   })));
   if(targeted)trace({key,milestone:'write-result',attempt:attempts,committed:result.committed,ids:(Array.isArray(result)?result:result.storedEvents).map((s:any)=>s.event.id)});
   return result;
  });
  return EventSink.EventSinkV2.of({...real,write:(x)=>wrap('write',x),writeIfRunCurrent:(x)=>wrap('writeIfRunCurrent',x),writeIfProviderThreadOwner:(x)=>wrap('writeIfProviderThreadOwner',x)});
 })).pipe(Layer.provide(sink));
 const git=GitVcsDriver.layer.pipe(Layer.provide(ServerConfig.layerTest(process.cwd(),`${scratch}/${key}-home`)),Layer.provideMerge(NodeServices.layer));
 const layers=Layer.mergeAll(db,stores,sink,IdAllocator.layer,ThreadCommandExecutor.layer,git);
 const recheckCommands=Layer.effect(ThreadCommandExecutor.ThreadCommandExecutor,Effect.gen(function*(){
 const lock=yield* ThreadCommandExecutor.ThreadCommandExecutor; const projection=yield* ProjectionStore.ProjectionStoreV2; const real=yield* EventSink.EventSinkV2;const ids=yield* IdAllocator.IdAllocatorV2;
 return {...lock,withLock:(threadId:any,effect:any)=>lock.withLock(threadId,Effect.gen(function*(){
 if(active&&c.mode==='recheck'){
 const child=yield* projection.getThread(threadId);const event={id:yield* ids.allocate.event({threadId}),type:'thread.model-selection-updated',threadId,occurredAt:yield* DateTime.now,payload:{...child,modelSelection:{instanceId:modelSelection.instanceId,model:'gpt-6.1-sol'}}};
 yield* real.write({events:[event]});concurrentIds.push(event.id);trace({key,milestone:'concurrent-selection-under-lock',event});
 }
 return yield* effect;
 }))};
 })).pipe(Layer.provide(layers));
 const ing=ProviderEventIngestor.layer.pipe(Layer.provide(Layer.mergeAll(stores,observed,IdAllocator.layer,recheckCommands,git)));
 yield* Effect.gen(function*(){
 const sql=yield* SqlClient.SqlClient;
 yield* sql.unsafe(`PRAGMA busy_timeout=${c.timeout}`);
 const busy=yield* sql`PRAGMA busy_timeout`;
 const id=yield* IdAllocator.IdAllocatorV2;
 const real=yield* EventSink.EventSinkV2;
 const projection=yield* ProjectionStore.ProjectionStoreV2;
 const store=yield* EventStore.EventStoreV2;
 const ingestor=yield* ProviderEventIngestor.ProviderEventIngestorV2;
 const now=yield* DateTime.now;
 const root=yield* threadCreatedEvent(now);
 const threadId=root.threadId;
 const runId=RunId.make('run:proof');const attemptId=RunAttemptId.make('attempt:proof');
 const providerThreadId=ProviderThreadId.make('provider-thread:proof');
 const providerSessionId=yield* id.allocate.providerSession({providerInstanceId:modelSelection.instanceId,threadId});
 const domain=(type:any,payload:any)=>Effect.gen(function*(){return {id:yield* id.allocate.event({threadId}),type,threadId,occurredAt:now,payload};});
 yield* real.write({events:[root,yield* domain('provider-session.updated',{id:providerSessionId,driver:CODEX_DRIVER,providerInstanceId:modelSelection.instanceId,status:'ready',cwd:scratch,model:modelSelection.model,capabilities:CodexProviderCapabilitiesV2,createdAt:now,updatedAt:now,lastError:null})]});
 const run={id:runId,threadId,ordinal:1,providerInstanceId:modelSelection.instanceId,modelSelection,providerThreadId,userMessageId:MessageId.make('message:proof'),rootNodeId:NodeId.make('node:proof'),activeAttemptId:attemptId,status:'running',requestedAt:now,startedAt:now,completedAt:null,checkpointId:null,contextHandoffId:null};
 const pt={id:providerThreadId,driver:CODEX_DRIVER,providerInstanceId:modelSelection.instanceId,providerSessionId,appThreadId:threadId,ownerNodeId:null,nativeThreadRef:{driver:CODEX_DRIVER,nativeId:'proof',strength:'strong'},nativeConversationHeadRef:null,status:'idle',firstRunOrdinal:1,lastRunOrdinal:1,handoffIds:[],forkedFrom:null,createdAt:now,updatedAt:now};
 if(c.mode!=='missing')yield* real.write({events:[yield* domain('run.created',{...run,...(c.mode==='wrongAttempt'?{activeAttemptId:'attempt:new'}:{}),...(c.mode==='wrongStatus'?{status:'completed'}:{})}),yield* domain('provider-thread.updated',{...pt,...(c.mode==='wrongOrdinal'?{lastRunOrdinal:2}:{})})]});
 if(c.mode==='supersede') changes.push(c.route==='R'?`UPDATE orchestration_v2_projection_runs SET status='completed',payload_json=json_set(payload_json,'$.activeAttemptId','attempt:new','$.status','completed') WHERE run_id='run:proof'`:`UPDATE orchestration_v2_projection_provider_threads SET last_run_ordinal=2,payload_json=json_set(payload_json,'$.lastRunOrdinal',2) WHERE provider_thread_id='provider-thread:proof'`);
 const childThreadId=id.derive.threadFromProviderThread({driver:CODEX_DRIVER,nativeThreadId:'proof-child'});
 if(c.route==='S'&&c.mode!=='missingChild')yield* real.write({events:[yield* domain('thread.created',{...root.payload,id:childThreadId,modelSelection:{...modelSelection,options:[{id:'reasoningEffort',value:'xhigh'}]},activeProviderThreadId:null,title:'proof child',lineage:{parentThreadId:threadId,relationshipToParent:'subagent',rootThreadId:threadId}})]});
 let event:any=c.route==='O'?{type:'provider_thread.updated',driver:CODEX_DRIVER,providerThread:{...pt,status:'active'}}:{type:'provider_turn.updated',driver:CODEX_DRIVER,providerTurn:{id:ProviderTurnId.make('turn:proof'),providerThreadId,nodeId:NodeId.make('node:proof'),runAttemptId:null,nativeTurnRef:null,ordinal:1,status:'completed',startedAt:now,completedAt:now}};
 if(c.route==='S')event={type:'subagent.updated',driver:CODEX_DRIVER,subagent:{id:NodeId.make('subagent:proof'),threadId,runId:null,parentNodeId:NodeId.make('node:root'),origin:c.mode==='origin'?'app_owned':'provider_native',createdBy:'agent',driver:CODEX_DRIVER,providerInstanceId:modelSelection.instanceId,providerThreadId:null,childThreadId:c.mode==='nullChild'?null:childThreadId,nativeTaskRef:null,prompt:'proof',title:'proof',model:c.mode==='noModel'?null:c.mode==='equal'?modelSelection.model:'gpt-6.1-sol',status:'running',result:null,startedAt:now,completedAt:null,updatedAt:now,...(c.mode==='reportedSelection'?{modelSelection:{instanceId:modelSelection.instanceId,model:'gpt-6.1-sol',options:[{id:'reasoningEffort',value:'low'}]}}:{})}};
 if(c.mode==='empty')event={type:'command_output.delta',driver:CODEX_DRIVER,commandId:'command:empty',delta:''};
 const rg={runId,activeAttemptId:attemptId,expectedStatus:'running'};
 const og={providerThreadId,runId,activeAttemptId:attemptId,expectedLastRunOrdinal:1};
 const guard=c.route==='R'?{writeIfRunCurrent:rg}:c.route==='O'?{writeIfProviderThreadOwner:og}:c.mode==='ownerWins'?{writeIfRunCurrent:{...rg,activeAttemptId:'wrong'},writeIfProviderThreadOwner:og}:c.mode==='ownerRejects'?{writeIfRunCurrent:rg,writeIfProviderThreadOwner:{...og,expectedLastRunOrdinal:99}}:{};
 if(['nonlock','mixed'].includes(c.mode))yield* sql.unsafe(`CREATE TRIGGER proof_abort BEFORE INSERT ON orchestration_events WHEN NEW.event_type='${targetType}' BEGIN SELECT RAISE(ABORT, 'proof non-lock constraint'); END`);
 const subscribeAfter=yield* store.latestSequence();
 const readyMarker=yield* domain('thread.metadata-updated',root.payload);
 const finishMarker=yield* domain('thread.metadata-updated',root.payload);
 const ready=yield* Deferred.make<void>();const finished=yield* Deferred.make<void>();
 const published:any[]=[];
 const subscriber=yield* Stream.runForEach(real.stream({afterSequence:subscribeAfter}),stored=>Effect.gen(function*(){
 if(stored.event.id===readyMarker.id){yield* Deferred.succeed(ready,undefined);return;}
 if(stored.event.id===finishMarker.id){yield* Deferred.succeed(finished,undefined);return;}
 published.push({id:stored.event.id,type:stored.event.type});
 })).pipe(Effect.forkScoped);
 yield* real.write({events:[readyMarker]});yield* Deferred.await(ready);
 const seq=yield* store.latestSequence();
 trace({key,milestone:'subscribed-before-ingestion',time:new Date().toISOString(),afterSequence:seq});
 trace({key,revision:c.revision==='before'?'3cc31ed27c8956b36cee35a50509fb7c6869592b':'83ef5cf0b67c915b0db262e8089c84c6eb71a669',dbPath,busy,route:c.route,mode:c.mode,milestone:'ingestion-start',time:new Date().toISOString()});
 active=true;
 const input={providerSessionId,providerInstanceId:modelSelection.instanceId,threadId,event,...guard};
 const exit=yield* ingestor.ingestNormalized(input).pipe(Effect.exit);
 active=false;
 if(holder){if(c.mode==='persistent'||(c.mode==='recover2'&&c.revision==='before'))yield* release();yield* Effect.promise(async()=>{await request!({op:'close'});await new Promise<void>(r=>holder!.once('exit',()=>r()));});holder=undefined;}
 yield* real.write({events:[finishMarker]});yield* Deferred.await(finished);yield* Fiber.interrupt(subscriber);
 const persisted=Array.from(yield* Stream.runCollect(store.read({afterSequence:seq}))).filter(s=>s.event.id!==finishMarker.id);
 const projected=yield* projection.getThreadProjection(threadId);
 const child=c.route==='S'&&c.mode!=='missingChild'?yield* projection.getThread(childThreadId):null;
 const target=persisted.filter(s=>s.event.type===targetType&&!concurrentIds.includes(s.event.id));
 const expectedFailure=['persistent','nonlock','mixed','injectedNonSql','injectedAbsent','injectedMixed'].includes(c.mode)||(c.revision==='before'&&['recover1','recover2','supersede'].includes(c.mode));
 trace({key,milestone:'ingestion-end',time:new Date().toISOString(),attempts,primary,errors,exit:Exit.isSuccess(exit)?{success:true,returned:exit.value.map(s=>({id:s.event.id,type:s.event.type}))}:{success:false,error:Cause.pretty(exit.cause), reasons:exit.cause.reasons?.map((r:any)=>chain(r.error??r.defect??r))},persisted:persisted.map(s=>({id:s.event.id,type:s.event.type})),published,projected:{turns:projected.providerTurns,providerThreads:projected.providerThreads,runs:projected.runs,childModel:child?.modelSelection}});
 assert.deepEqual(published,persisted.map(s=>({id:s.event.id,type:s.event.type})));
 assert.equal(new Set(published.map(s=>s.id)).size,published.length);
 assert.equal(Exit.isFailure(exit),expectedFailure);
 const noOp=['missing','wrongAttempt','wrongStatus','wrongOrdinal','supersede','ownerRejects','empty'].includes(c.mode);
 const secondarySkip=c.route==='S'&&['origin','nullChild','noModel','missingChild','equal','recheck'].includes(c.mode);
 assert.lengthOf(target,expectedFailure||noOp||secondarySkip?0:1);
 assert.equal(new Set(persisted.map(s=>s.event.id)).size,persisted.length);
 const duplicates=yield* sql`SELECT event_id,COUNT(*) AS n FROM orchestration_events GROUP BY event_id HAVING COUNT(*)>1`;
 assert.lengthOf(duplicates,0);
 if(c.route==='P'||c.route==='R'||c.route==='C')assert.equal(projected.providerTurns.length,target.length);
 if(c.route==='O'&&target.length)assert.equal(projected.providerThreads[0]?.status,'active');
 if(c.route==='S'&&child)assert.equal(child.modelSelection.model,target.length||c.mode==='recheck'?'gpt-6.1-sol':modelSelection.model);
 if(errors.length&&!c.mode.startsWith('injected')){
 assert.include(JSON.stringify(errors[0]),c.mode==='nonlock'?'ConstraintError':'LockTimeoutError');
 if(c.mode==='mixed'&&c.revision==='after')assert.notEqual(errors[1].cause?.reason?.tag,'LockTimeoutError');
 }

 if(['recover1','recover2','persistent','nonlock','mixed','injectedNonSql','injectedAbsent','injectedMixed'].includes(c.mode))assert.equal(attempts,c.revision==='before'?1:c.mode==='recover2'||c.mode==='persistent'?3:['nonlock','injectedNonSql','injectedAbsent'].includes(c.mode)?1:2);
 if(c.route==='S')assert.equal(persisted.filter(s=>s.event.type==='subagent.updated').length,1);
 if(c.mode==='reportedSelection')assert.deepEqual(child?.modelSelection.options,[{id:'reasoningEffort',value:'low'}]);
 if(c.mode==='repeat'||(c.route==='S'&&['recover1','recover2','reportedSelection'].includes(c.mode)&&Exit.isSuccess(exit))){
  const repeat=yield* ingestor.ingestNormalized(input);
  assert.deepEqual(repeat.map(s=>s.event.type),['subagent.updated']);
  const afterRepeat=Array.from(yield* Stream.runCollect(store.read({afterSequence:seq})));assert.equal(afterRepeat.filter(s=>s.event.type==='thread.model-selection-updated').length,1);assert.deepEqual((yield* projection.getThread(childThreadId)).modelSelection,child?.modelSelection);
  trace({key,milestone:'repeat',returned:repeat.map(s=>({id:s.event.id,type:s.event.type}))});
 }
 }).pipe(Effect.provide(Layer.mergeAll(layers,ing)));
}).pipe(Effect.scoped,Effect.ensuring(Effect.sync(()=>{if(holder)holder.kill();}))); },{timeout:30000});