import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
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 NodeCrypto from "@effect/platform-node/NodeCrypto";
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 OrchestrationV2ShellSnapshot,
  type OrchestrationV2RuntimeRequest,
  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 Option from "effect/Option";
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 ScratchThreadPromotion from "../project/ScratchThreadPromotion.ts";
import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
import * as ManagedPolicy from "../policy/ManagedPolicy.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";


import { EnvironmentId, type OrchestrationV2DomainEvent } from "@t3tools/contracts";
import * as EventStore from "./EventStore.ts";
import * as CommandReceiptStore from "./CommandReceiptStore.ts";
import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
import * as McpToolAccess from "../mcp/McpToolAccess.ts";
import * as ThreadHandlers from "../mcp/toolkits/thread/handlers.ts";
import { ThreadToolkit } from "../mcp/toolkits/thread/tools.ts";
import { mkdirSync, mkdtempSync, appendFileSync, writeFileSync } from "node:fs";
import { execFileSync } from "node:child_process";
const outputRoot = process.env.CAPTURE_OUTPUT!;
mkdirSync(outputRoot, {recursive:true});
const runDir = mkdtempSync(`${outputRoot}/run-${new Date().toISOString().replaceAll(":", "-")}-`);
process.env.TMPDIR = runDir;
const revision = execFileSync("git", ["rev-parse", "HEAD"], {encoding:"utf8"}).trim();
const emit = (variant:string, section:string, value:unknown) => {
 const line = `${section}: ${JSON.stringify(value)}\n`;
 appendFileSync(`${runDir}/${variant}.txt`, line);
 console.log(`CAPTURE ${variant} ${line.trimEnd()}`);
};
const layerPlatformTest = Layer.merge(
  Layer.mergeAll(NodeServices.layer, NodeCrypto.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 makeHarness = Effect.fn("capture.makeHarness")(function* (name:string) {
 const cwd = mkdtempSync(`${runDir}/${name}-workspace-`);
 execFileSync("git",["init","-q"],{cwd});
 execFileSync("git",["config","user.name","Synthetic evidence"],{cwd});
 execFileSync("git",["config","user.email","evidence@example.invalid"],{cwd});
 writeFileSync(`${cwd}/README.md`, "Synthetic queue evidence\n");
 execFileSync("git",["add","README.md"],{cwd});
 execFileSync("git",["commit","-qm","synthetic seed"],{cwd,env:{...process.env,GIT_AUTHOR_DATE:"2000-01-01T00:00:00Z",GIT_COMMITTER_DATE:"2000-01-01T00:00:00Z"}});
 const driver = ProviderDriverKind.make("codex");
 const instanceId = modelSelection.instanceId;
 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
 const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = [];
 const steered:string[]=[];
 const interrupted:ProviderTurnId[]=[];
 const capabilities = CodexProviderCapabilitiesV2;
  const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
    instanceId,
    driver,
    getCapabilities: () => Effect.succeed(capabilities),
    planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
    openSession: (input) =>
      Effect.gen(function* () {
        const now = yield* DateTime.now;
        const turnUpdated = (
          turn: ProviderAdapter.ProviderAdapterV2TurnInput,
          status: "running" | "interrupted",
        ) =>
          Queue.offer(events, {
            type: "provider_turn.updated",
            driver,
            providerTurn: {
              id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`),
              providerThreadId: turn.providerThread.id,
              nodeId: turn.rootNodeId,
              runAttemptId: turn.attemptId,
              nativeTurnRef: {
                driver,
                nativeId: `native:${turn.attemptId}`,
                strength: "strong",
              },
              ordinal: turn.providerTurnOrdinal,
              status,
              startedAt: now,
              completedAt: status === "running" ? null : now,
            },
          });
        const startTurn = (turn: ProviderAdapter.ProviderAdapterV2TurnInput) =>
          Effect.gen(function* () {
            started.push(turn);
            yield* turnUpdated(turn, "running");
          });
        return {
          instanceId,
          driver,
          providerSessionId: input.providerSessionId,
          providerSession: {
            id: input.providerSessionId,
            driver,
            providerInstanceId: instanceId,
            status: "ready",
            cwd,
            model: modelSelection.model,
            capabilities,
            createdAt: now,
            updatedAt: now,
            lastError: null,
          },
          events: Stream.fromQueue(events),
          ensureThread: ({ threadId }) =>
            Effect.succeed({
              id: ProviderThreadId.make(`provider-thread:${threadId}`),
              driver,
              providerInstanceId: instanceId,
              providerSessionId: input.providerSessionId,
              appThreadId: threadId,
              ownerNodeId: null,
              nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" },
              nativeConversationHeadRef: null,
              status: "idle",
              firstRunOrdinal: null,
              lastRunOrdinal: null,
              handoffIds: [],
              forkedFrom: null,
              createdAt: now,
              updatedAt: now,
            }),
          resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
          startTurn,
          compactThread: startTurn,
          steerTurn: (turn) =>
            Effect.sync(() => {
              steered.push(turn.message.text);
            }),
          interruptTurn: ({providerTurnId}) => Effect.sync(()=> {interrupted.push(providerTurnId);}),
          respondToRuntimeRequest: () => Effect.void,
          readThreadSnapshot: () => Effect.die("unused"),
          rollbackThread: () => Effect.die("unused"),
          forkThread: () => Effect.die("unused"),
        };
      }),
  };

const orchestrationAdapter = adapter;
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,
  },
);


const database = SqlitePersistence.layerMemory;
const layer = Layer.mergeAll(RuntimeLayer.layer,RuntimeLayer.layerEventSink,EventStore.layer,ProjectionStore.layer,ProjectStore.layer,ThreadCommandExecutor.layer).pipe(
 Layer.provide(McpSessionRegistryTestkit.layer),Layer.provide(McpProviderSessions.layer),
 Layer.provideMerge(database), Layer.provide(layerCheckpointStoreTest), Layer.provide(layerServerConfig),
 Layer.provide(ServerSettings.layerTest()), Layer.provide(layerTestProviderInstanceRegistry),
 Layer.provide(layerGitWorkflowTest),Layer.provide(layerProjectServiceTest),Layer.provideMerge(layerPlatformTest));
return {cwd,events,started,steered,interrupted,layer};
});

const variants = ["R01","R02","R03a","R03b","R03c","R04-absent-drain","R04-false-drain","R04-absent-rebuild","R04-false-rebuild","R05-set","R05-clear","R06-auto","R06-queue","R07-ordinary","R07-goal","R07-control","R08-steer","R08-guards","R09-resume","R09-selected"];
for (const variant of variants) {
 it.effect(`capture ${variant}`, () => Effect.scoped(Effect.gen(function*(){
  emit(variant,"ENVIRONMENT",{revision,fixture:variant,kind:"H: real runtime layer/event sink/projection, in-memory SQLite, registered fake Codex provider; synthetic data; controlled terminal events",clock:"TestClock epoch",visualDimensions:"not applicable: server-only transcript",mcp:"real toolkit, live fake-provider thread invocation context/access layer and ThreadManagementService"});
  const h=yield* makeHarness(variant);
  yield* Effect.gen(function*(){
   const o=yield* Orchestrator.OrchestratorV2;
   const sink=yield* EventSink.EventSinkV2;
   const store=yield* EventStore.EventStoreV2;
   const worker=yield* EffectWorker.OrchestrationEffectWorkerV2;
   const maintenance=yield* ProjectionMaintenance.ProjectionMaintenanceV2;
   const sql=yield* SqlClient.SqlClient;
   const threadId=ThreadId.make(`thread:${variant}`);
   let serial=0,lastSequence=0;
   const nextId=()=>CommandId.make(`capture:${variant}:${++serial}`);
   const mid=(key:string)=>MessageId.make(`message:${variant}:${key}`);
   const keyOf=(run:OrchestrationV2Run)=>run.userMessageId?.replace(`message:${variant}:`,"") ?? "no-message";
   const projection=o.getThreadProjection(threadId);
   const getRun=(key:string)=>Effect.map(projection,p=>p.runs.find(r=>r.userMessageId===mid(key))!);
   const summary=(r:OrchestrationV2Run)=>({id:keyOf(r),status:r.status,q:r.queuePosition,flag:Reflect.get(r,"queuedAheadOfCompletions")??"absent",held:r.queueHeld??false});
   const errorText=(e:unknown):unknown=>e && typeof e==="object" ? {_tag:Reflect.get(e,"_tag"),code:Reflect.get(e,"code"),message:Reflect.get(e,"message"),cause:Reflect.get(e,"cause")} : e;
   const command=(input:Record<string,unknown>)=>Effect.gen(function*(){
    emit(variant,"ACTION",input);
    const result=yield* o.dispatch({...input,threadId,commandId:input.commandId??nextId()} as never).pipe(Effect.result);
    emit(variant,"COMMAND_RESULT",result._tag==="Success"?{ok:true,sequence:result.success.sequence}:{ok:false,error:errorText(result.failure)});
    return result;
   });
   const dump= (label:string)=>Effect.gen(function*(){
    const p=yield* projection;
    emit(variant,"PERSISTED "+label,p.runs.map(summary));
    emit(variant,"DELIVERY_ORDER "+label,queuedRunsInDeliveryOrder(p).map(keyOf));
    const rows=yield* store.read({threadId,afterSequence:lastSequence}).pipe(Stream.runCollect);
    for(const row of rows){
     lastSequence=Math.max(lastSequence,row.sequence);
     if(row.event.type==="run.updated") emit(variant,"RUN_UPDATED "+label,summary(row.event.payload));
     appendFileSync(`${runDir}/${variant}-events.jsonl`,JSON.stringify(row)+"\n");
    }
    emit(variant,"PROVIDER "+label,{started:h.started.map(t=>t.message.messageId.replace(`message:${variant}:`,"")),steered:h.steered,interruptRequests:h.interrupted.length});
   });
   const waitEvent=(predicate:(e:OrchestrationV2DomainEvent)=>boolean)=>Effect.gen(function*(){
    const afterSequence=yield* store.latestSequence({threadId});
    return yield* sink.stream({threadId,afterSequence}).pipe(Stream.map(x=>x.event),Stream.filter(predicate),Stream.take(1),Stream.runDrain,Effect.forkChild({startImmediately:true}));
   });
   const send=(key:string,text=key,mode:"queue"|"auto"|"start"="queue",extra:Record<string,unknown>={})=>command({type:"message.dispatch",createdBy:"user",creationSource:"web",messageId:mid(key),text,attachments:[],modelSelection,...(mode==="auto"?{dispatchMode:{type:"start_immediately"},deliveryIntent:"auto"}:{dispatchMode:{type:mode==="start"?"start_immediately":"queue_after_active"}}),...extra});
   yield* o.dispatch({type:"thread.create",commandId:nextId(),threadId,projectId:ProjectId.make(`project:${variant}`),title:variant,modelSelection,runtimeMode:"full-access",interactionMode:"default",branch:null,worktreePath:h.cwd,createdBy:"user",creationSource:"web"});
   const observed:OrchestrationV2DomainEvent[]=[];
   const observation=yield* sink.stream({threadId,afterSequence:0}).pipe(Stream.map(x=>x.event),Stream.runForEach(e=>Effect.sync(()=>{observed.push(e)})),Effect.forkChild({startImmediately:true}));
   const start=(key:string,text=key)=>Effect.gen(function*(){
    const running=yield* waitEvent(e=>e.type==="provider-turn.updated"&&e.payload.status==="running");
    yield* send(key,text,"start");yield* worker.drain();yield* Fiber.join(running);
   });
   const finishActive=Effect.gen(function*(){
    const p=yield* projection;
    const active=p.runs.find(r=>r.status==="running"||r.status==="starting");
    assert.isDefined(active,"active run exists before natural terminal");
    const turn=p.providerTurns.find(t=>t.runAttemptId===active!.activeAttemptId&&t.status==="running")!;
    assert.isDefined(turn,"active fake provider turn exists");
    emit(variant,"ACTION",{naturalTerminal:keyOf(active!),status:"completed"});
    const settled=yield* waitEvent(e=>e.type==="run.updated"&&e.payload.id===active!.id&&e.payload.status==="waiting");
    yield* Queue.offer(h.events,{type:"provider_turn.updated",driver:ProviderDriverKind.make("codex"),providerTurn:{...turn,status:"completed",completedAt:yield* DateTime.now}});
    yield* Queue.offer(h.events,{type:"turn.terminal",driver:ProviderDriverKind.make("codex"),providerThreadId:turn.providerThreadId,providerTurnId:turn.id,runOrdinal:active!.ordinal,status:"completed",failure:null,threadDisposition:"reusable"});
    yield* Fiber.join(settled);
    const completed=yield* sink.stream({threadId,afterSequence:0}).pipe(Stream.filter(x=>x.event.type==="run.updated"&&x.event.payload.id===active!.id&&x.event.payload.status==="completed"),Stream.take(1),Stream.runDrain,Effect.forkChild({startImmediately:true}));
    yield* worker.drain();yield* Fiber.join(completed);
   });
   // Historical P owns completions for interruption fixtures; no Stop disposes their cohort.
   if(/R0[5-8]/.test(variant)&&variant!=="R08-guards") {yield* start("P");yield* finishActive;}
   yield* start("A",variant.startsWith("R07")?"/compact":"A blocks");
   const ownerKey=/R0[5-8]/.test(variant)&&variant!=="R08-guards"?"P":"A";
   let generation=0;
   const queueCompletion=(key:string)=>Effect.gen(function*(){
    const parent=yield* getRun(ownerKey); const g=++generation;
    const taskIds=[NodeId.make(`task:${variant}:${g}`)];
    emit(variant,"HARNESS_COHORT_SEED",{owner:ownerKey,message:key,generation:g,taskIds,disposition:"open"});
    yield* sink.write({events:[{id:EventId.make(`cohort:${variant}:${g}`),type:"run.updated",threadId,runId:parent.id,occurredAt:yield* DateTime.now,payload:{...parent,delegatedCompletion:{disposition:"open",nextGeneration:g+1,delivery:{messageId:mid(key),generation:g,taskIds}}}}]});
    yield* send(key,key,"queue",{createdBy:"agent",creationSource:"server",delegatedCompletion:{parentRunId:parent.id,generation:g,taskIds}});
   });
   const context=Layer.succeed(McpInvocationContext.McpInvocationContext,{environmentId:EnvironmentId.make("evidence"),requestNamespace:`evidence:${variant}`,thread:{threadId,providerSessionId:(yield* projection).providerSessions[0]!.id,providerInstanceId:modelSelection.instanceId},client:undefined,issuedAt:0,capabilities:new Set(["orchestration" as const])});
   const toolkit=yield* ThreadToolkit.pipe(Effect.provide(McpToolAccess.HandlersLayer.layer(ThreadHandlers.layer)));
   const mcp=(input:Record<string,unknown>)=>Effect.gen(function*(){
    emit(variant,"MCP_INPUT",input);
    const result=yield* toolkit.handle("t3_queue",{...input,threadId} as never).pipe(Stream.unwrap,Stream.runCollect,Effect.provide(context),Effect.result);
    emit(variant,"MCP_OUTPUT",result._tag==="Success"?result.success:{error:errorText(result.failure)});
    return result;
   });
   const reorder=(key:string,before:string|null)=>Effect.gen(function*(){
    yield* mcp({action:"reorder",queuedRunId:(yield* getRun(key)).id,beforeRunId:before===null?null:(yield* getRun(before)).id});yield* dump(`reorder ${key} before ${before}`);
   });
   const list=()=>mcp({action:"list",limit:20});
   const drain=Effect.gen(function*(){
    for(let step=0;step<16;step++) {
     let p=yield* projection;
     assert.isAtMost(p.runs.filter(r=>r.status==="starting"||r.status==="running").length,1,"one active at a time");
     let active=p.runs.find(r=>r.status==="starting"||r.status==="running");
     if(!active && !p.runs.some(r=>r.status==="queued"))break;
     if(!active){yield* o.resumeQueuedRuns; active=(yield* projection).runs.find(r=>r.status==="starting"||r.status==="running");}
     assert.isDefined(active,"a queued run was selected");
     const running=yield* sink.stream({threadId,afterSequence:0}).pipe(Stream.filter(x=>x.event.type==="provider-turn.updated"&&x.event.payload.runAttemptId===active!.activeAttemptId&&x.event.payload.status==="running"),Stream.take(1),Stream.runDrain,Effect.forkChild({startImmediately:true}));
     yield* worker.drain();yield* Fiber.join(running);yield* finishActive;
     yield* dump(`drain ${step}`);
    }
    const p=yield* projection;
    assert.equal(p.runs.filter(r=>r.status==="queued").length,0,"final queue empty");
    yield* dump("FINAL");
    const events=yield* store.read({threadId}).pipe(Stream.runCollect);
    const starts=events.filter(x=>(x.event.type==="run.updated"||x.event.type==="run.created")&&x.event.payload.status==="starting").map(x=>(x.event.type==="run.updated"||x.event.type==="run.created")?keyOf(x.event.payload):"");
    assert.equal(new Set(starts).size,starts.length,"each run starts exactly once");
    emit(variant,"ACTUAL_STARTING_SEQUENCE",starts);
    const expected=p.runs.filter(r=>r.status!=="cancelled").map(keyOf).sort();
    assert.deepEqual([...starts].sort(),expected,"no noncancelled run skipped");
    const orders:Record<string,[string[],string[]]>={
      R01:[["C1","C2","C3","U","V"],["U","C1","C2","C3","V"]],
      R02:[["C1","C2","C3","V","U","W"],["C1","V","U","C2","C3","W"]],
      R03a:[["C1","C2","U","V"],["C1","C2","U","V"]], R03b:[["C1","C2","V","U"],["C1","C2","V","U"]],R03c:[["C2","U","V"],["C2","U","V"]],
      "R05-set":[["C","G","U"],["C","G","U"]],"R05-clear":[["C","G","U"],["C","G","U"]],
      "R06-auto":[["C","G","U","V"],["C","G","U","V"]],"R06-queue":[["C","U","V","G"],["U","C","V","G"]],
      "R07-ordinary":[["C","S","W","V"],["S","C","W","V"]],"R07-goal":[["C","S","W","V"],["C","S","W","V"]],"R07-control":[["C","S","W","V"],["C","S","W","V"]],
      "R08-steer":[["C","V"],["C","V"]],"R08-guards":[["C1","C2","U","N"],["C1","C2","U","N"]],
      "R09-resume":[["C","U","V"],["U","C","V"]],"R09-selected":[["V","C","U"],["V","U","C"]]};
    const expectedOrder=variant.startsWith("R04")?["C1","C2","U"]:orders[variant]![revision.startsWith("a77de508")?1:0];
    assert.deepEqual(starts.filter(k=>k!=="P"&&k!=="A"),expectedOrder,"matrix expected actual start order");
    if(/^R0[567]/.test(variant)) {
      assert.deepEqual(h.steered,[],"goal/compaction never native-steered");
      const interrupts=variant==="R05-clear"||variant==="R06-queue"?0:1;
      assert.equal(h.interrupted.length,interrupts,"matrix interrupt request count");
    }
    emit(variant,"INVARIANTS",{oneActive:true,eachStartOnce:true,noSkippedIds:true,emptyQueue:true,matrixOrder:true});
   });
   const initial=variant==="R03c"?["C","U","V"]:variant.startsWith("R04")?["U","C1","C2"]:variant==="R01"?["C1","C2","U","V"]:variant==="R02"?["C1","C2","U","V","W"]:variant==="R03a"||variant==="R03b"?["C1","C2","U","V"]:variant.startsWith("R05")?["C","U","G"]:variant.startsWith("R07")?["C","W","S","V"]:variant==="R08-guards"?["C1","C2","U","N"]:["C","U","V"];
   for(const key of initial) {
    if(key.startsWith("C"))yield* queueCompletion(key);
    else if(key==="N")yield* send(key,key,"queue",{createdBy:"agent",creationSource:"server",notification:{source:{kind:"background_task"},outcome:"updated",summary:"synthetic activity"}});
    else yield* send(key,key==="G"?(variant==="R05-clear"?"/goal clear":"/goal ship it"):key==="S"&&variant==="R07-goal"?"/goal ship it":key);
   }
   yield* dump("SEEDED");
   if(variant==="R01") {
    yield* reorder("U","C1");yield* queueCompletion("C3");
    for(const cursor of [0,2,4])yield* mcp({action:"list",limit:2,cursor});
    yield* mcp({action:"read",queuedRunId:(yield* getRun("U")).id});
   } else if(variant==="R02") {
    yield* reorder("U","C2");yield* reorder("V","U");yield* queueCompletion("C3");yield* list();
   } else if(variant.startsWith("R03")) {
    yield* reorder("U",variant==="R03c"?"C":"C1");
    if(variant==="R03c") {yield* mcp({action:"cancel",queuedRunId:(yield* getRun("C")).id});yield* dump("after cancelling C; marker retained");yield* reorder("U","V");yield* queueCompletion("C2");}
    else yield* reorder("U",variant==="R03a"?"V":null);
   } else if(variant.startsWith("R04")) {
    const p=yield* projection; const flag=variant.includes("false")?{queuedAheadOfCompletions:false}:{};
    emit(variant,"EVENTSINK_INJECTION",{legacyPositions:{U:1,C1:2,C2:3},marker:variant.includes("false")?false:"absent",note:"explicit false is unknown to baseline codec; raw encoded JSON recorded"});
    yield* sink.write({events:p.runs.filter(r=>r.status==="queued").map(r=>({id:EventId.make(`legacy:${r.id}`),type:"run.updated" as const,threadId,runId:r.id,occurredAt:DateTime.makeUnsafe(0),payload:{...r,queuePosition:keyOf(r)==="U"?1:keyOf(r)==="C1"?2:3,...flag}}))});yield* dump("legacy injected");
    if(variant.endsWith("rebuild")) {
     const acceptedId=nextId();
     const cmd={type:"queued-run.reorder",commandId:acceptedId,runId:(yield* getRun("U")).id,beforeRunId:(yield* getRun("C1")).id};
     const result=yield* command(cmd);yield* dump("priority marker only");
     emit(variant,"REBUILD",yield* maintenance.rebuild);yield* dump("after true rebuild");
     const before=(yield* store.read({threadId,eventType:"run.updated"}).pipe(Stream.runCollect)).length;
     yield* command(cmd);
     const after=(yield* store.read({threadId,eventType:"run.updated"}).pipe(Stream.runCollect)).length;
     emit(variant,"REPLAY",{originalAccepted:result._tag==="Success",addedRunUpdates:after-before});
     assert.equal(after,before);
     yield* reorder("U",null);emit(variant,"REBUILD",yield* maintenance.rebuild);yield* dump("after false rebuild");
    }
    const json=yield* sql`SELECT payload_json FROM orchestration_v2_projection_runs ORDER BY ordinal`;
    emit(variant,"PERSISTED_JSON",json.map(row=>JSON.parse(row.payload_json as string)).map(r=>({message:r.userMessageId,q:r.queuePosition,flag:Reflect.get(r,"queuedAheadOfCompletions")??"absent"})));
   } else if(variant.startsWith("R05")) {
    yield* reorder("U","C");yield* reorder("G","U");
    yield* mcp({action:"promote_to_steer",queuedRunId:(yield* getRun("G")).id,targetRunId:(yield* getRun("A")).id});yield* dump("goal promotion before effects");yield* worker.drain();yield* dump("goal promotion");yield* list();
   } else if(variant.startsWith("R06")) {
    yield* reorder("U","C");yield* send("G","/goal ship it",variant==="R06-auto"?"auto":"queue");yield* dump("incoming goal before effects");yield* worker.drain();yield* dump("incoming goal");yield* list();
   } else if(variant.startsWith("R07")) {
    if(variant!=="R07-control"){yield* reorder("W","C");yield* reorder("S","W");}
    yield* mcp({action:"promote_to_steer",queuedRunId:(yield* getRun("S")).id,targetRunId:(yield* getRun("A")).id});yield* dump("compaction skip before effects");yield* worker.drain();yield* dump("compaction skipped");yield* list();
   } else if(variant==="R08-steer") {
    yield* reorder("U","C");yield* mcp({action:"promote_to_steer",queuedRunId:(yield* getRun("U")).id,targetRunId:(yield* getRun("A")).id});yield* worker.drain();yield* dump("steered U");
    const p=yield* projection;const r=yield* getRun("U");emit(variant,"CANCELLATION",{run:r.status,node:p.nodes.find(n=>n.id===r.rootNodeId)?.status,attempt:p.attempts.find(a=>a.id===r.activeAttemptId)?.status});assert.deepEqual(h.steered,["U"]);
   } else if(variant==="R08-guards") {
    const before=JSON.stringify((yield* projection).runs); const count=(yield* store.read({threadId,eventType:"run.updated"}).pipe(Stream.runCollect)).length;
    yield* reorder("C2","C1");yield* reorder("C2",null);yield* reorder("N","U");
    const after=JSON.stringify((yield* projection).runs);const nextCount=(yield* store.read({threadId,eventType:"run.updated"}).pipe(Stream.runCollect)).length;
    assert.equal(after,before);assert.equal(count,nextCount);emit(variant,"GUARDS",{runsUnchanged:true,runEventCountUnchanged:true,rejectedReceiptsMayPersist:true});
   } else if(variant.startsWith("R09")) {
    yield* reorder("U","C");const p=yield* projection;const active=yield* getRun("A");
    emit(variant,"EVENTSINK_INJECTION",{terminalizeA:"interrupted (EventSink injection; no Stop command)",holdAllQueued:true,atomic:true});
    yield* sink.write({events:p.runs.filter(r=>r.status==="queued"||r.id===active.id).map(r=>({id:EventId.make(`hold:${r.id}`),type:"run.updated" as const,threadId,runId:r.id,occurredAt:DateTime.makeUnsafe(0),payload:r.id===active.id?{...r,status:"interrupted" as const,completedAt:DateTime.makeUnsafe(0)}:{...r,queueHeld:true}}))});
    const prior=h.started.length;emit(variant,"RECOVERY_WHILE_HELD",yield* o.resumeQueuedRuns);yield* worker.drain();assert.equal(h.started.length,prior);yield* dump("held recovery did not start");
    const running=yield* waitEvent(e=>e.type==="provider-turn.updated"&&e.payload.status==="running");
    yield* command(variant==="R09-resume"?{type:"queue.resume"}:{type:"queued-run.start",runId:(yield* getRun("V")).id});yield* dump("unheld command committed");yield* worker.drain();yield* Fiber.join(running);yield* dump("unheld selected");
   }
   yield* list();yield* drain;
   emit(variant,"OBSERVATION_BARRIER",{subscribedBeforeActions:true,observedStarting:observed.filter(e=>(e.type==="run.updated"||e.type==="run.created")&&e.payload.status==="starting").length});
   yield* Fiber.interrupt(observation);
  }).pipe(Effect.provide(h.layer));
 })));
}