import { SourceControlProviderKind } from "@t3tools/contracts";
import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts";
import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";
import * as NodeServices from "@effect/platform-node/NodeServices";
import { assert, it } from "@effect/vitest";
import { vi } from "vite-plus/test";
import {
  type ApplicationStoredEvent,
  CheckpointId,
  CheckpointRef,
  CommandId,
  ContextTransferId,
  EventId,
  MessageId,
  NodeId,
  RuntimeRequestId,
  TurnItemId,
  type ModelSelection,
  type OrchestrationV2Run,
  type OrchestrationV2ThreadProjection,
  ProjectId,
  type PullRequestDetail,
  type PullRequestComment,
  PullRequestOperationError,
  ProviderDriverKind,
  ProviderInstanceId,
  ProviderSessionId,
  ProviderThreadId,
  ProviderTurnId,
  RunId,
  ScheduledTaskId,
  ThreadId,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
import * as DateTime from "effect/DateTime";
import * as Deferred from "effect/Deferred";
import * as Fiber from "effect/Fiber";
import * as Layer from "effect/Layer";
import * as Path from "effect/Path";
import * as Queue from "effect/Queue";
import * as Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";
import * as SqlClient from "effect/sql/SqlClient";

import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
import * as GitWorkflow from "../git/GitWorkflowService.ts";
import * as ServerConfig from "../config.ts";
import * as SqlitePersistence from "../persistence/Sqlite.ts";
import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
import * as ProjectService from "../project/ProjectService.ts";
import * as ServerSettings from "../serverSettings.ts";
import * as McpSessionRegistryTestkit from "../mcp/McpSessionRegistry.testkit.ts";
import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts";
import type { ProviderInstance } from "@t3tools/provider-core/server/driver";
import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
import * as VcsProcess from "../vcs/VcsProcess.ts";
import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
import * as Orchestrator from "./Orchestrator.ts";
import { queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts";
import { ROLLBACK_FAILED_MESSAGE } from "./CheckpointRollbackService.ts";
import * as EffectWorker from "./EffectWorker.ts";
import * as EffectOutbox from "./EffectOutbox.ts";
import * as EventSink from "./EventSink.ts";
import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryService.ts";
import * as ProjectionMaintenance from "./ProjectionMaintenance.ts";
import * as ProjectionStore from "./ProjectionStore.ts";
import * as PullRequestWatchReactor from "./PullRequestWatchReactor.ts";
import { PULL_REQUEST_WATCH_WAKE_LIMIT } from "./pullRequestWatch.ts";
import * as PullRequestService from "../pullRequest/PullRequestService.ts";
import * as ProjectStore from "./ProjectStore.ts";
import * as ProviderSessionManager from "./ProviderSessionManager.ts";
import * as RuntimeLayer from "./runtimeLayer.ts";
import { shellStreamItemFromThreadShell } from "./ShellStream.ts";
import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
import * as ThreadManagementService from "./ThreadManagementService.ts";
import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts";
import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
import * as Scheduler from "../scheduling/Scheduler.ts";
import * as SecretRequests from "../secrets/SecretRequests.ts";
import * as PreviewManager from "../preview/Manager.ts";
import * as TerminalManager from "../terminal/Manager.ts";
import * as ResourceCleanupService from "./ResourceCleanupService.ts";
import * as ThreadLaunchService from "./ThreadLaunchService.ts";
import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";

const layerPlatformTest = Layer.merge(
  NodeServices.layer,
  Layer.mock(SourceControlProviderRegistry.SourceControlProviderRegistry)({
    resolveLink: () => Effect.die("unused title link"),
  }),
);

const layerServerConfig = ServerConfig.layerTest(process.cwd(), {
  prefix: "t3-orchestration-v2-runtime-layer-",
});

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

const layerVcsDriverRegistryTest = VcsDriverRegistry.layer.pipe(
  Layer.provide(VcsProcess.layer),
  Layer.provide(layerServerConfig),
  Layer.provide(layerPlatformTest),
);

const layerCheckpointStoreTest = CheckpointStore.layer.pipe(
  Layer.provide(layerVcsDriverRegistryTest),
);
const layerGitWorkflowTest = Layer.mock(GitWorkflow.GitWorkflowService)({
  pruneWorktrees: () => Effect.void,
  createWorktree: () => Effect.succeed({} as never),
});
const layerProjectServiceTest = Layer.mock(ProjectService.ProjectService)({
  getById: () => Effect.succeedNone,
});

const driver = ProviderDriverKind.make("codex");
const orchestrationAdapter = {
  instanceId: modelSelection.instanceId,
  driver,
  getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
  planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
  openSession: () => Effect.die("sessions are not used by lifecycle tests"),
} as ProviderAdapter.ProviderAdapterV2["Service"];
const providerInstance = {
  instanceId: modelSelection.instanceId,
  driverKind: driver,
  continuationIdentity: {
    driverKind: driver,
    continuationKey: "codex:test",
  },
  displayName: "Codex test",
  enabled: true,
  snapshot: { getSnapshot: Effect.succeed({}) } as unknown as ProviderInstance["snapshot"],
  orchestrationAdapter,
  textGeneration: {} as ProviderInstance["textGeneration"],
} satisfies ProviderInstance;
const alternateProviderInstance = {
  ...providerInstance,
  instanceId: alternateInstanceId,
  continuationIdentity: {
    driverKind: driver,
    continuationKey: "codex:test:alternate",
  },
  displayName: "Codex alternate test",
  orchestrationAdapter: {
    ...orchestrationAdapter,
    instanceId: alternateInstanceId,
  },
} satisfies ProviderInstance;

const layerTestProviderInstanceRegistry = Layer.succeed(
  ProviderInstanceRegistry.ProviderInstanceRegistry,
  {
    getInstance: (instanceId) =>
      Effect.succeed(
        [providerInstance, alternateProviderInstance].find(
          (instance) => instanceId === instance.instanceId,
        ),
      ),
    listInstances: Effect.succeed([providerInstance, alternateProviderInstance]),
    listUnavailable: Effect.succeed([]),
    streamChanges: Stream.empty,
    subscribeChanges: Effect.never,
  },
);


import { EnvironmentId, type OrchestrationV2Command } from "@t3tools/contracts";
import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
import * as Exit from "effect/Exit";
import * as Cause from "effect/Cause";
import * as McpToolAccess from "../mcp/McpToolAccess.ts";
import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
import * as ThreadHandlers from "../mcp/toolkits/thread/handlers.ts";
import { ThreadToolkit } from "../mcp/toolkits/thread/tools.ts";
const layerTest = Layer.mergeAll(
  RuntimeLayer.layer,
  RuntimeLayer.layerEventSink,
  ProjectStore.layer,
  ProjectionStore.layer,
  EffectOutbox.layer,
  ThreadCommandExecutor.layer,
).pipe(
  Layer.provide(McpSessionRegistryTestkit.layer),
  Layer.provide(McpProviderSessions.layer),
  Layer.provide(SqlitePersistence.layerMemory),
  Layer.provide(layerCheckpointStoreTest),
  Layer.provide(layerServerConfig),
  Layer.provide(ServerSettings.layerTest()),
  Layer.provide(layerTestProviderInstanceRegistry),
  Layer.provide(layerGitWorkflowTest),
  Layer.provide(layerProjectServiceTest),
  Layer.provide(layerPlatformTest),
);


const label = "harness capture; real orchestrator/event sink/projection store; stub provider; injected terminal events; no real model execution";
const log = (kind: string, value: unknown) => console.log(`CAPTURE ${kind} ${JSON.stringify(value)}`);
const makeFixture = Effect.fn("evidence.makeFixture")(function* (row: string) {
  log("SECTION", {row, label});
  const o = yield* Orchestrator.OrchestratorV2;
  const sink = yield* EventSink.EventSinkV2;
  const threadId = ThreadId.make(`evidence:${row}`);
  let serial = 0;
  const mid = (name: string) => MessageId.make(`${threadId}:${name}`);
  const projection = o.getThreadProjection(threadId);
  const nameOf = (run: OrchestrationV2Run) => run.userMessageId.slice(`${threadId}:`.length);
  const get = (name: string) => projection.pipe(Effect.map(p => p.runs.find(r => r.userMessageId === mid(name))!));
  const runEvents = (commandId: CommandId) => sink.readByCommandId({commandId}).pipe(Stream.runCollect, Effect.map(es => es.filter(e => e.event.type.startsWith("run.")).map(e => ({sequence:e.sequence,type:e.event.type,payload:e.event.payload}))));
  const command = (fields: Omit<OrchestrationV2Command,"commandId">, fixedId?: string) => Effect.gen(function* () {
    const c = {...fields, commandId: CommandId.make(fixedId ?? `${threadId}:cmd:${++serial}`)} as OrchestrationV2Command;
    log("COMMAND",c);
    const result = yield* o.dispatch(c).pipe(Effect.exit);
    if (Exit.isFailure(result)) log("ERROR",Cause.pretty(result.cause));
    else log("RESULT",result.value);
    log("DIAGNOSTIC command run events",yield* runEvents(c.commandId));
    return Exit.isSuccess(result);
  });
  yield* command({type:"thread.create",threadId,createdBy:"user",creationSource:"web",projectId:ProjectId.make(`project:${row}`),title:row,modelSelection,runtimeMode:"full-access",interactionMode:"default",branch:null,worktreePath:process.cwd()});
  const update = (name: string, patch: Partial<OrchestrationV2Run>, why: string) => Effect.gen(function* () {
    const r = yield* get(name); const now = yield* DateTime.now;
    log("INJECTED run.updated",{name,why,patch});
    yield* sink.write({events:[{id:EventId.make(`${threadId}:inject:${++serial}`),type:"run.updated",threadId,runId:r.id,occurredAt:now,payload:{...r,...patch}}]});
  });
  const send = (name:string, text=name, extra={}) => command({type:"message.dispatch",threadId,messageId:mid(name),text,attachments:[],modelSelection,createdBy:"user",creationSource:"web",dispatchMode:{type:name === "A" || name === "P" ? "start_immediately":"queue_after_active"},...extra});
  const completion = (generation:number, owner="A", name=`C${generation}`) => Effect.gen(function* () {
    const parent = yield* get(owner); const taskIds=[NodeId.make(`${threadId}:task:${generation}`)];
    yield* update(owner,{delegatedCompletion:{disposition:"open",nextGeneration:generation+1,delivery:{messageId:mid(name),generation,taskIds}}},"synthetic valid completion cohort");
    yield* send(name,name,{createdBy:"agent",creationSource:"server",delegatedCompletion:{parentRunId:parent.id,generation,taskIds}});
  });
  const reorder = (name:string,before:string|null) => Effect.gen(function* () {
    return yield* command({type:"queued-run.reorder",threadId,runId:(yield* get(name)).id,beforeRunId:before === null ? null : (yield* get(before)).id});
  });
  const mcpDependencies = Layer.mergeAll(NodeCrypto.layer, Layer.succeed(McpInvocationContext.McpInvocationContext,{
    environmentId:EnvironmentId.make("evidence-environment"),requestNamespace:`evidence:${row}`,thread:undefined,client:{sessionId:"synthetic",label:"evidence",access:"full-access"},issuedAt:0,capabilities:new Set(["orchestration"] as const)
  }), Layer.mock(ThreadManagementService.ThreadManagementService)({
    getThreadShell:o.getThreadShell,
    getProjectThreadRecords:({threadId})=>o.getThreadProjection(threadId),
    dispatch:(c)=>Effect.gen(function*(){log("MCP DISPATCH COMMAND",c);return yield* o.dispatch(c);}),
  }));
  const toolkit = yield* ThreadToolkit.pipe(Effect.provide(McpToolAccess.HandlersLayer.layer(ThreadHandlers.layer).pipe(Layer.provide(mcpDependencies))));
  const mcp = (input: Parameters<typeof toolkit.handle<"t3_queue">>[1]) => Effect.gen(function* () {
    log("t3_queue INPUT",input);
    const response = yield* toolkit.handle("t3_queue",input).pipe(Stream.unwrap,Stream.runCollect,Effect.provide(mcpDependencies),Effect.exit);
    if (Exit.isFailure(response)) log("t3_queue ERROR",Cause.pretty(response.cause));
    else log("t3_queue OUTPUT",response.value);
    return response;
  });
  const inspect = (stage:string) => Effect.gen(function* () {
    const p=yield* projection;
    log("DIAGNOSTIC projection",{stage,deliveryOrder:queuedRunsInDeliveryOrder(p).map(nameOf),runs:p.runs.map(r=>({name:nameOf(r),id:r.id,status:r.status,q:r.queuePosition,priority:(r as OrchestrationV2Run & {queuedAheadOfCompletions?:boolean}).queuedAheadOfCompletions,held:r.queueHeld,rootNodeId:r.rootNodeId})),interrupts:p.turnItems.filter(i=>i.type === "run_interrupt_request")});
    yield* mcp({action:"list",threadId,limit:100});
  });
  const afterSequence = yield* o.getThreadEventSequence(threadId);
  const registered = yield* Deferred.make<void>();
  const promotions = yield* Queue.unbounded<RunId>();
  yield* sink.stream({threadId,afterSequence:afterSequence-1}).pipe(Stream.runForEach(stored=>Effect.gen(function*(){
    yield* Deferred.succeed(registered,undefined);
    if(stored.event.type === "run.updated" && stored.event.payload.status === "starting") {
      log("PERSISTED PROMOTION",{sequence:stored.sequence,runId:stored.event.payload.id,name:nameOf(stored.event.payload),status:"starting"});
      yield* Queue.offer(promotions,stored.event.payload.id);
    }
  })),Effect.forkScoped);
  yield* Deferred.await(registered);
  log("BARRIER","persisted event stream registered through replayed thread event");
  const finish = (name:string) => Effect.gen(function*(){yield* update(name,{status:"completed",completedAt:yield* DateTime.now},"injected terminal event; releases blocker without Stop disposal");});
  const drain = (expected?:string[]) => Effect.gen(function* () {
    const initial=yield* projection;
    const count=queuedRunsInDeliveryOrder(initial).length;
    const actual:string[]=[];
    const active=initial.runs.filter(r=>r.status === "starting" || r.status === "running");
    assert.isAtMost(active.length,1);
    if(active[0]) yield* finish(nameOf(active[0]));
    else {log("COMMAND","orchestrator.resumeQueuedRuns"); log("RESULT",yield* o.resumeQueuedRuns);}
    for(let i=0;i<count;i++) {
      const id=yield* Queue.take(promotions); const p=yield* projection; const r=p.runs.find(r=>r.id===id)!;
      assert.equal(r.status,"starting"); assert.equal(p.runs.filter(r=>r.status === "starting" || r.status === "running").length,1);
      assert.isFalse(actual.includes(nameOf(r))); actual.push(nameOf(r));
      log("ACTUAL PROMOTION",{name:nameOf(r),runId:id,status:r.status,remaining:queuedRunsInDeliveryOrder(p).map(nameOf)});
      yield* finish(nameOf(r));
    }
    const final=yield* projection;
    assert.equal(queuedRunsInDeliveryOrder(final).length,0);
    log("ACTUAL PROMOTIONS COMPLETE",{actual,expected,emptyFinalQueue:true,oneLiveRunMaximum:true});
    if(expected) assert.deepEqual(actual,expected);
    return actual;
  });
  return {o,sink,threadId,mid,projection,get,nameOf,command,send,completion,reorder,update,mcp,inspect,finish,drain,promotions};
});
it.layer(layerTest)("queue evidence", it=>{
  for(const row of ["R01","R02","R03","R19","R09","R04","R05","R06","R07","R16","R22"]){
    it.effect(row,()=>Effect.gen(function*(){
      const f=yield* makeFixture(row);
      const {send,completion,reorder,inspect,drain}=f;
      yield* send("A");
      yield* completion(1); yield* completion(2);
      yield* send("U");
      if(row!=="R06" && row!=="R07") yield* send("V");
      if(["R03","R04"].includes(row)) yield* send("W");
      yield* inspect("seed");
      if(row === "R01") {
        const accepted=yield* reorder("U","C1");
        yield* completion(3); yield* inspect("after correction and later completion");
        yield* drain(accepted?["U","C1","C2","C3","V"]:["C1","C2","C3","U","V"]);
        log("REGRESSION ASSERTION",{accepted,requirement:"reorder must succeed ahead of completions"});
        assert.isTrue(accepted,"regression: old anchor rejection must be removed");
      } else if(row === "R02") {
        const accepted=yield* reorder("U","C2"); yield* inspect("between completions");
        yield* drain(accepted?["C1","U","C2","V"]:["C1","C2","U","V"]);
      } else if(row === "R03") {
        const accepted=yield* reorder("U","C1");
        yield* reorder("V","U"); yield* reorder("W","V"); yield* inspect("ordinary anchors in prefix");
        yield* drain(accepted?["W","V","U","C1","C2"]:undefined);
      } else if(row === "R19") {
        yield* f.mcp({action:"reorder",threadId:f.threadId,queuedRunId:(yield* f.get("U")).id,beforeRunId:(yield* f.get("C1")).id});
        yield* completion(3);
        for(const cursor of [0,2,4]) yield* f.mcp({action:"list",threadId:f.threadId,limit:2,cursor});
        yield* f.mcp({action:"read",threadId:f.threadId,queuedRunId:(yield* f.get("U")).id});
        yield* f.mcp({action:"reorder",threadId:f.threadId,queuedRunId:(yield* f.get("C2")).id,beforeRunId:null});
        yield* f.command({type:"queued-run.edit",threadId:f.threadId,runId:(yield* f.get("U")).id,text:"U".repeat(1001)});
        yield* f.mcp({action:"list",threadId:f.threadId,limit:100});
        yield* f.mcp({action:"read",threadId:f.threadId,queuedRunId:(yield* f.get("U")).id});
        yield* inspect("MCP boundary final"); yield* drain();
      } else if(row === "R09") {
        yield* send("N","N",{createdBy:"agent",creationSource:"server",notification:{source:{kind:"monitor"},outcome:"updated",summary:"N",detail:"synthetic"}});
        for(const [from,to] of [["C2","C1"],["C2",null],["N","U"]] as const) yield* reorder(from,to);
        yield* inspect("automatic moves rejected"); yield* drain();
      } else if(row === "R04" || row === "R05") {
        const reached=yield* reorder("U","C1");
        log("PRECONDITION",{markedStateReached:reached,baselineLimitation:reached?null:"unreachable through original reorder"});
        yield* inspect("before reverse"); yield* reorder("U",row === "R04"?"W":null);
        yield* inspect("after reverse");
        yield* drain(row === "R04"?["C1","C2","V","U","W"]:["C1","C2","V","U"]);
      } else if(row === "R06" || row === "R07") {
        for(const [name,q] of [["U",1],["C1",2],["C2",3]] as const) yield* f.update(name,{queuePosition:q},"injected legacy q; missing marker");
        yield* inspect("legacy effective order");
        if(row === "R06") {const ok=yield* reorder("U","C1");yield* inspect("marker-only unchanged q");yield* drain(ok?["U","C1","C2"]:["C1","C2","U"]);}
        else {
          yield* f.update("U",{queuedAheadOfCompletions:false},"explicit false legacy variant");
          yield* inspect("false equivalent to missing");
          const p=yield* f.projection; const u=yield* f.get("U");const c1=yield* f.get("C1");const c2=yield* f.get("C2");
          log("STATIC HARNESS comparator boundary",queuedRunsInDeliveryOrder({...p,runs:[{...u,queuePosition:null,ordinal:10},{...c1,queuePosition:2,ordinal:3},{...c2,queuePosition:2,ordinal:2},{...u,id:RunId.make("nonqueued"),status:"completed"}]}).map(r=>({id:r.id,ordinal:r.ordinal,q:r.queuePosition})));
          yield* drain(["C1","C2","U"]);
        }
      } else if(row === "R16") {
        yield* reorder("U","C1"); yield* inspect("before in-memory projection rebuild true");
        const maintenance=yield* ProjectionMaintenance.ProjectionMaintenanceV2;
        log("DIAGNOSTIC rebuild true",yield* maintenance.rebuild);yield* inspect("after rebuild true");
        const fields={type:"queued-run.reorder" as const,threadId:f.threadId,runId:(yield* f.get("U")).id,beforeRunId:null};
        yield* f.command(fields,`${f.threadId}:replay`);yield* f.command(fields,`${f.threadId}:replay`);
        log("DIAGNOSTIC rebuild false",yield* maintenance.rebuild);yield* inspect("after rebuild false and replay");yield* drain();
      } else if(row === "R22") {
        yield* f.command({type:"queued-run.cancel",threadId:f.threadId,runId:(yield* f.get("C1")).id});
        yield* reorder("U","C2");yield* inspect("before last completion cancellation");
        yield* f.command({type:"queued-run.cancel",threadId:f.threadId,runId:(yield* f.get("C2")).id});
        yield* inspect("marker survives cancellation");yield* reorder("U","V");yield* inspect("same-position clear with no completions");
        yield* completion(3,"A","C2fresh");yield* inspect("fresh completion priority");yield* drain(["C2fresh","U","V"]);
      }
      log("ROW COMPLETE",row);
    }));
  }
});