import { assert, it, vi } from "@effect/vitest";import {  CheckpointScopeId,  ChatAttachmentId,  ChatFileAttachment,  CommandId,  EventId,  MessageId,  NodeId,  type OrchestrationV2AppThread,  type OrchestrationV2CheckpointScope,  type OrchestrationV2DomainEvent,  type OrchestrationV2ExecutionNode,  type OrchestrationV2ProviderThread,  type OrchestrationV2Run,  type OrchestrationV2RunAttempt,  type OrchestrationV2Subagent,  type OrchestrationV2TurnItem,  ProviderDriverKind,  ProviderInstanceId,  ProviderSessionId,  ProviderThreadId,  ProviderTurnId,  RunAttemptId,  RunId,  ThreadId,  TurnItemId,} from "@t3tools/contracts";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Ref from "effect/Ref";import * as Stream from "effect/Stream"; import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";import { ServerSettingsService } from "../serverSettings.ts";import { CheckpointServiceV2 } from "./CheckpointService.ts";import { EventSinkV2 } from "./EventSink.ts";import { IdAllocatorV2, layer as idAllocatorLayer } from "./IdAllocator.ts";import {  ProviderAdapterEventStreamError,  ProviderAdapterTurnStartError,  type ProviderAdapterV2Error,  type ProviderAdapterV2Event,  type ProviderAdapterV2SessionRuntime,} from "./ProviderAdapter.ts";import { ProviderEventIngestorV2 } from "./ProviderEventIngestor.ts";import {  canRouteRelatedSubagent,  cascadeTerminalizeRunOwnedSubagents,  finalProviderThreadStatus,  layer as runExecutionServiceLayer,  makeProviderEventRoutingState,  type ProviderEventRouteIdentity,  routeProviderEvent,  RunExecutionServiceV2,  selectInheritedBackgroundTurnItems,} from "./RunExecutionService.ts";import { RunFinalizationObserver } from "./RunFinalizationService.ts"; const driver = ProviderDriverKind.make("codex"); const RunExecutionTestLayer = runExecutionServiceLayer.pipe(  Layer.provide(    Layer.mergeAll(      Layer.mock(CheckpointServiceV2)({ captureBaseline: () => Effect.void }),      Layer.mock(EventSinkV2)({}),      idAllocatorLayer,      Layer.mock(ProviderEventIngestorV2)({ ingestNormalized: () => Effect.succeed([]) }),      ServerSettingsService.layerTest(),    ),  ),); it("keeps recoverable turn failures reusable and reserves error for broken threads", () => {  assert.equal(finalProviderThreadStatus("reusable"), "idle");  assert.equal(finalProviderThreadStatus("broken"), "error");}); it.effect("routes shared-runtime events only to their owning root run", () =>  Effect.gen(function* () {    const now = yield* DateTime.now;    const first: ProviderEventRouteIdentity = {      threadId: ThreadId.make("thread:shared-runtime:first"),      runId: RunId.make("run:shared-runtime:first"),      attemptId: RunAttemptId.make("attempt:shared-runtime:first"),      providerThreadId: ProviderThreadId.make("provider-thread:shared-runtime:first"),    };    const second: ProviderEventRouteIdentity = {      threadId: ThreadId.make("thread:shared-runtime:second"),      runId: RunId.make("run:shared-runtime:second"),      attemptId: RunAttemptId.make("attempt:shared-runtime:second"),      providerThreadId: ProviderThreadId.make("provider-thread:shared-runtime:second"),    };    const firstTurnId = ProviderTurnId.make("provider-turn:shared-runtime:first");    const turnEvent: ProviderAdapterV2Event = {      type: "provider_turn.updated",      driver,      threadId: first.threadId,      providerTurn: {        id: firstTurnId,        providerThreadId: first.providerThreadId,        nodeId: NodeId.make("node:shared-runtime:first"),        runAttemptId: first.attemptId,        nativeTurnRef: null,        ordinal: 1,        status: "running",        startedAt: now,        completedAt: null,      },    };    const messageEvent: ProviderAdapterV2Event = {      type: "message.updated",      driver,      message: {        createdBy: "agent",        creationSource: "provider",        id: MessageId.make("message:shared-runtime:first"),        threadId: first.threadId,        runId: first.runId,        nodeId: NodeId.make("node:shared-runtime:first"),        role: "assistant",        text: "first only",        attachments: [],        streaming: false,        createdAt: now,        updatedAt: now,      },    };    const terminalEvent: ProviderAdapterV2Event = {      type: "turn.terminal",      driver,      providerThreadId: first.providerThreadId,      providerTurnId: firstTurnId,      runOrdinal: 1,      status: "completed",      failure: null,      threadDisposition: "reusable",    };     const firstInitial = makeProviderEventRoutingState({      identity: first,      providerTurnId: null,    });    const secondInitial = makeProviderEventRoutingState({      identity: second,      providerTurnId: null,    });    const [firstTurnAccepted, firstAfterTurn] = routeProviderEvent(turnEvent, first, firstInitial);    const [secondTurnAccepted, secondAfterTurn] = routeProviderEvent(      turnEvent,      second,      secondInitial,    );     assert.isTrue(firstTurnAccepted);    assert.isFalse(secondTurnAccepted);    assert.isTrue(routeProviderEvent(messageEvent, first, firstAfterTurn)[0]);    assert.isFalse(routeProviderEvent(messageEvent, second, secondAfterTurn)[0]);    assert.isTrue(routeProviderEvent(terminalEvent, first, firstAfterTurn)[0]);    assert.isFalse(routeProviderEvent(terminalEvent, second, secondAfterTurn)[0]);  }),); it("does not route a superseded attempt through a reused provider thread", () => {  const threadId = ThreadId.make("thread:shared-runtime:restart");  const providerThreadId = ProviderThreadId.make("provider-thread:shared-runtime:restart");  const oldAttempt: ProviderEventRouteIdentity = {    threadId,    runId: RunId.make("run:shared-runtime:restart"),    attemptId: RunAttemptId.make("attempt:shared-runtime:restart:old"),    providerThreadId,  };  const newAttempt: ProviderEventRouteIdentity = {    ...oldAttempt,    attemptId: RunAttemptId.make("attempt:shared-runtime:restart:new"),  };  const oldTurnEvent: ProviderAdapterV2Event = {    type: "provider_turn.updated",    driver,    threadId,    providerTurn: {      id: ProviderTurnId.make("provider-turn:shared-runtime:restart:old"),      providerThreadId,      nodeId: NodeId.make("node:shared-runtime:restart:old"),      runAttemptId: oldAttempt.attemptId,      nativeTurnRef: null,      ordinal: 1,      status: "interrupted",      startedAt: null,      completedAt: null,    },  };   const newState = makeProviderEventRoutingState({ identity: newAttempt, providerTurnId: null });  assert.isFalse(routeProviderEvent(oldTurnEvent, newAttempt, newState)[0]);}); it("routes only exact same-thread background items inherited from settled runs", () => {  const threadId = ThreadId.make("thread:inherited-background-routing");  const otherThreadId = ThreadId.make("thread:inherited-background-routing:other");  const priorRunId = RunId.make("run:inherited-background-routing:prior");  const currentRunId = RunId.make("run:inherited-background-routing:current");  const itemId = TurnItemId.make("turn-item:inherited-background-routing");  const identity: ProviderEventRouteIdentity = {    threadId,    runId: currentRunId,    attemptId: RunAttemptId.make("attempt:inherited-background-routing:current"),    providerThreadId: ProviderThreadId.make("provider-thread:inherited-background-routing:current"),  };  const initial = makeProviderEventRoutingState({    identity,    inheritedBackgroundTurnItems: [{ id: itemId, runId: priorRunId }],    providerTurnId: null,  });  const inheritedRunning = {    type: "turn_item.updated",    driver,    turnItem: {      id: itemId,      threadId,      runId: priorRunId,      providerTurnId: null,      ordinal: 1,      type: "subagent",      status: "running",    },  } as Extract<ProviderAdapterV2Event, { type: "turn_item.updated" }>;  const unrelatedRunItem = {    ...inheritedRunning,    turnItem: {      ...inheritedRunning.turnItem,      runId: RunId.make("run:inherited-background-routing:unrelated"),    },  } as ProviderAdapterV2Event;  const unlistedPriorItem = {    ...inheritedRunning,    turnItem: {      ...inheritedRunning.turnItem,      id: TurnItemId.make("turn-item:inherited-background-routing:unrelated"),    },  } as ProviderAdapterV2Event;  const unrelatedThreadItem = {    ...inheritedRunning,    turnItem: { ...inheritedRunning.turnItem, threadId: otherThreadId },  } as ProviderAdapterV2Event;  const ordinaryItem = {    ...inheritedRunning,    turnItem: { ...inheritedRunning.turnItem, type: "reasoning" as const },  } as ProviderAdapterV2Event;  const inheritedTerminal = {    ...inheritedRunning,    turnItem: { ...inheritedRunning.turnItem, status: "completed" as const },  } as ProviderAdapterV2Event;   const [runningAccepted, afterRunning] = routeProviderEvent(inheritedRunning, identity, initial);  assert.isTrue(runningAccepted);  assert.isFalse(routeProviderEvent(unrelatedRunItem, identity, afterRunning)[0]);  assert.isFalse(routeProviderEvent(unlistedPriorItem, identity, afterRunning)[0]);  assert.isFalse(routeProviderEvent(unrelatedThreadItem, identity, afterRunning)[0]);  assert.isFalse(routeProviderEvent(ordinaryItem, identity, afterRunning)[0]);   const [terminalAccepted, afterTerminal] = routeProviderEvent(    inheritedTerminal,    identity,    afterRunning,  );  assert.isTrue(terminalAccepted);  assert.isFalse(    routeProviderEvent(inheritedRunning, identity, afterTerminal)[0],    "a nonterminal replay must not resurrect an inherited terminal",  );}); it("selects only live background items from non-completed settled prior runs", () => {  const threadId = ThreadId.make("thread:inherited-background-selection");  const otherThreadId = ThreadId.make("thread:inherited-background-selection:other");  const currentProviderThreadId = ProviderThreadId.make(    "provider-thread:inherited-background-selection:current",  );  const foreignProviderThreadId = ProviderThreadId.make(    "provider-thread:inherited-background-selection:foreign-session",  );  const interruptedRunId = RunId.make("run:inherited-background-selection:interrupted");  const failedRunId = RunId.make("run:inherited-background-selection:failed");  const cancelledRunId = RunId.make("run:inherited-background-selection:cancelled");  const completedRunId = RunId.make("run:inherited-background-selection:completed");  const rolledBackRunId = RunId.make("run:inherited-background-selection:rolled-back");  const currentRunId = RunId.make("run:inherited-background-selection:current");  const inheritedItemId = TurnItemId.make("turn-item:inherited-background-selection:live");  const failedItemId = TurnItemId.make("turn-item:inherited-background-selection:failed");  const cancelledItemId = TurnItemId.make("turn-item:inherited-background-selection:cancelled");  const makeRun = (    id: RunId,    ordinal: number,    status: OrchestrationV2Run["status"],    runThreadId = threadId,  ) =>    ({      id,      threadId: runThreadId,      ordinal,      status,    }) as OrchestrationV2Run;  const makeItem = (    id: TurnItemId,    runId: RunId,    status: OrchestrationV2TurnItem["status"],    type: OrchestrationV2TurnItem["type"] = "subagent",    itemThreadId = threadId,    providerThreadId = currentProviderThreadId,  ) =>    ({      id,      threadId: itemThreadId,      runId,      providerThreadId,      type,      status,    }) as OrchestrationV2TurnItem;   const selected = selectInheritedBackgroundTurnItems({    threadId,    currentProviderThreadId,    currentRunOrdinal: 6,    runs: [      makeRun(interruptedRunId, 1, "interrupted"),      makeRun(failedRunId, 2, "failed"),      makeRun(cancelledRunId, 3, "cancelled"),      makeRun(completedRunId, 4, "completed"),      makeRun(rolledBackRunId, 5, "rolled_back"),      makeRun(currentRunId, 6, "running"),      makeRun(        RunId.make("run:inherited-background-selection:other"),        1,        "interrupted",        otherThreadId,      ),    ],    turnItems: [      makeItem(inheritedItemId, interruptedRunId, "running"),      makeItem(failedItemId, failedRunId, "running"),      makeItem(cancelledItemId, cancelledRunId, "running"),      makeItem(        TurnItemId.make("turn-item:inherited-background-selection:completed-run"),        completedRunId,        "running",      ),      makeItem(        TurnItemId.make("turn-item:inherited-background-selection:terminal"),        interruptedRunId,        "completed",      ),      makeItem(        TurnItemId.make("turn-item:inherited-background-selection:ordinary"),        interruptedRunId,        "running",        "reasoning",      ),      makeItem(        TurnItemId.make("turn-item:inherited-background-selection:rolled-back"),        rolledBackRunId,        "running",      ),      makeItem(        TurnItemId.make("turn-item:inherited-background-selection:other-thread"),        interruptedRunId,        "running",        "subagent",        otherThreadId,      ),      makeItem(        TurnItemId.make("turn-item:inherited-background-selection:foreign-provider"),        interruptedRunId,        "running",        "subagent",        threadId,        foreignProviderThreadId,      ),    ],  });   assert.deepEqual(selected, [    { id: inheritedItemId, runId: interruptedRunId },    { id: failedItemId, runId: failedRunId },    { id: cancelledItemId, runId: cancelledRunId },  ]);}); it("does not carry interrupted child ownership into later attempts", () => {  assert.isFalse(canRouteRelatedSubagent("interrupted"));  assert.isFalse(canRouteRelatedSubagent("failed"));  assert.isFalse(canRouteRelatedSubagent("cancelled"));  assert.isTrue(canRouteRelatedSubagent("completed"));  assert.isTrue(canRouteRelatedSubagent("running"));   const threadId = ThreadId.make("thread:related-child:next-attempt");  const childThreadId = ThreadId.make("thread:related-child:interrupted");  const identity: ProviderEventRouteIdentity = {    threadId,    runId: RunId.make("run:related-child:next-attempt"),    attemptId: RunAttemptId.make("attempt:related-child:next-attempt"),    providerThreadId: ProviderThreadId.make("provider-thread:related-child:next-attempt"),  };  const state = makeProviderEventRoutingState({    identity,    providerTurnId: null,    relatedThreadIds: canRouteRelatedSubagent("interrupted") ? [childThreadId] : [],  });  const childNodeId = NodeId.make("node:related-child:interrupted");  const lateChildNode = {    type: "node.updated",    driver,    node: {      id: childNodeId,      threadId: childThreadId,      runId: null,      parentNodeId: null,      rootNodeId: childNodeId,      kind: "root_turn",      status: "completed",      countsForRun: false,      providerThreadId: null,      providerTurnId: null,      nativeItemRef: null,      runtimeRequestId: null,      checkpointScopeId: null,      startedAt: null,      completedAt: null,    },  } satisfies ProviderAdapterV2Event;   assert.isFalse(routeProviderEvent(lateChildNode, identity, state)[0]);}); it.effect("rechecks run ownership immediately before calling the provider", () =>  Effect.gen(function* () {    const runExecution = yield* RunExecutionServiceV2;    const guardCalls = yield* Ref.make(0);    const providerStarts = yield* Ref.make(0);    const threadId = ThreadId.make("thread:run-execution-start-guard");    const runId = RunId.make("run:run-execution-start-guard");    const attemptId = RunAttemptId.make("attempt:run-execution-start-guard");    const providerThreadId = ProviderThreadId.make("provider-thread:run-execution-start-guard");    const providerInstanceId = ProviderInstanceId.make("codex");    const providerSessionId = ProviderSessionId.make("session:run-execution-start-guard");    const rootNodeId = NodeId.make("node:run-execution-start-guard");    const run = {      id: runId,      threadId,      ordinal: 1,      providerInstanceId,    } as OrchestrationV2Run;    const rootNode = { id: rootNodeId } as OrchestrationV2ExecutionNode;    const providerThread = {      id: providerThreadId,      driver,    } as OrchestrationV2ProviderThread;    const attempt = {      id: attemptId,      providerTurnId: null,    } as OrchestrationV2RunAttempt;    const session = {      events: Stream.never,      startTurn: () => Ref.update(providerStarts, (count) => count + 1),    } as unknown as ProviderAdapterV2SessionRuntime;     yield* runExecution.startRootRun({      commandId: CommandId.make("command:run-execution-start-guard"),      appThread: { id: threadId } as OrchestrationV2AppThread,      providerSessionId,      session,      run,      rootNode,      checkpointScope: {        id: CheckpointScopeId.make("checkpoint-scope:run-execution-start-guard"),      } as OrchestrationV2CheckpointScope,      providerThread,      attempt,      attemptId,      providerTurnOrdinal: 1,      shouldStartProviderTurn: () =>        Ref.modify(guardCalls, (calls) => [calls === 0, calls + 1] as const),      message: {        messageId: MessageId.make("message:run-execution-start-guard"),        text: "Do not start after ownership changes.",        attachments: [],        createdBy: "user",        creationSource: "web",      },      modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },      runtimePolicy: {        runtimeMode: "full-access",        interactionMode: "default",        cwd: process.cwd(),        approvalPolicy: "never",        sandboxPolicy: {          type: "readOnly",

Lines 1–500 of 3489. Load more