import {  CommandId,  type EventId,  type ModelSelection,  type NodeId,  type OrchestrationV2AppThread,  type OrchestrationV2CheckpointScope,  type OrchestrationV2DomainEvent,  type OrchestrationV2ExecutionNode,  type OrchestrationV2ProviderFailure,  type OrchestrationV2ProviderThread,  type OrchestrationV2ProviderTurn,  type OrchestrationV2Run,  type OrchestrationV2RunAttempt,  type OrchestrationV2Subagent,  type OrchestrationV2TurnItem,  type ProviderSessionId,  type ProviderThreadId,  type ProviderTurnId,  type RunAttemptId,  type ThreadId,  type TurnItemId,} from "@t3tools/contracts";import * as Context from "effect/Context";import * as Cause from "effect/Cause";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Fiber from "effect/Fiber";import * as Layer from "effect/Layer";import * as Ref from "effect/Ref";import * as Schema from "effect/Schema";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,  type IdAllocatorV2AllocationError,  type IdAllocatorV2Shape,} from "./IdAllocator.ts";import type {  ProviderAdapterV2Event,  ProviderAdapterV2RuntimePolicy,  ProviderAdapterV2SessionRuntime,  ProviderAdapterV2TurnMessage,} from "./ProviderAdapter.ts";import { ProviderAdapterTurnStartError } from "./ProviderAdapter.ts";import { ProviderEventIngestorV2 } from "./ProviderEventIngestor.ts";import { makeProviderFailure, makeProviderFailureTurnItem } from "./ProviderFailure.ts";import { RunFinalizationObserver } from "./RunFinalizationService.ts"; export interface ProviderEventRoutingState {  readonly ownedThreadIds: ReadonlySet<ThreadId>;  readonly ownedProviderThreadIds: ReadonlySet<ProviderThreadId>;  readonly ownedProviderTurnIds: ReadonlySet<ProviderTurnId>;  readonly inheritedBackgroundTurnItems: ReadonlyMap<TurnItemId, OrchestrationV2Run["id"]>;  readonly rootProviderTurnId: ProviderTurnId | null;} export interface ProviderEventRouteIdentity {  readonly threadId: ThreadId;  readonly runId: OrchestrationV2Run["id"];  readonly attemptId: RunAttemptId;  readonly providerThreadId: ProviderThreadId;} export interface InheritedBackgroundTurnItemRoute {  readonly id: TurnItemId;  readonly runId: OrchestrationV2Run["id"];} type ProviderTerminalEvent = Extract<ProviderAdapterV2Event, { readonly type: "turn.terminal" }>; function isTerminalProviderTurnStatus(status: OrchestrationV2ProviderTurn["status"]): boolean {  return (    status === "completed" ||    status === "interrupted" ||    status === "failed" ||    status === "cancelled"  );} function isTerminalSubagentStatus(status: OrchestrationV2Subagent["status"]): boolean {  return (    status === "completed" ||    status === "interrupted" ||    status === "failed" ||    status === "cancelled"  );} // Turn item types whose lifecycle can outlive the root turn (background// commands, monitors/dynamic tools, subagent rows). Ingestion must not stop// while one of these is still non-terminal, or the late completion event is// dropped and the item spins forever in the projection.const backgroundCapableTurnItemTypes: ReadonlySet<OrchestrationV2TurnItem["type"]> = new Set([  "command_execution",  "dynamic_tool",  "subagent",]); function isTerminalTurnItemStatus(status: OrchestrationV2TurnItem["status"]): boolean {  return (    status === "completed" ||    status === "interrupted" ||    status === "failed" ||    status === "cancelled"  );} function isSettledRunEligibleForInheritedBackground(status: OrchestrationV2Run["status"]): boolean {  return status === "interrupted" || status === "failed" || status === "cancelled";} /** * Transfer delivery permission for exact live background items whose original * run no longer has a subscriber. Provider sessions are runtime containers and * can host multiple provider threads, so the durable provider-thread lineage is * the discriminator. Completed, rolled-back, and already-terminal items remain * excluded. */export function selectInheritedBackgroundTurnItems(input: {  readonly threadId: ThreadId;  readonly currentProviderThreadId: ProviderThreadId;  readonly currentRunOrdinal: number;  readonly runs: ReadonlyArray<OrchestrationV2Run>;  readonly turnItems: ReadonlyArray<OrchestrationV2TurnItem>;}): ReadonlyArray<InheritedBackgroundTurnItemRoute> {  const settledPriorRunIds = new Set(    input.runs      .filter(        (run) =>          run.threadId === input.threadId &&          run.ordinal < input.currentRunOrdinal &&          isSettledRunEligibleForInheritedBackground(run.status),      )      .map((run) => run.id),  );  return input.turnItems.flatMap((turnItem) =>    turnItem.threadId === input.threadId &&    turnItem.providerThreadId === input.currentProviderThreadId &&    turnItem.runId !== null &&    settledPriorRunIds.has(turnItem.runId) &&    backgroundCapableTurnItemTypes.has(turnItem.type) &&    !isTerminalTurnItemStatus(turnItem.status)      ? [{ id: turnItem.id, runId: turnItem.runId }]      : [],  );} type SubagentTurnItem = Extract<OrchestrationV2TurnItem, { readonly type: "subagent" }>; type OpenRunOwnedSubagentProjection = {  readonly subagents: ReadonlyMap<NodeId, OrchestrationV2Subagent>;  readonly turnItems: ReadonlyMap<NodeId, SubagentTurnItem>;  readonly childTurnItems: ReadonlyMap<TurnItemId, OrchestrationV2TurnItem>;  readonly nodes: ReadonlyMap<NodeId, OrchestrationV2ExecutionNode>;  /** Child threads once linked by a root-run subagent row; kept for cascade. */  readonly linkedChildThreadIds: ReadonlySet<ThreadId>;}; type RunOwnedSubagentTerminalStatus = Extract<  OrchestrationV2Subagent["status"],  "interrupted" | "failed" | "cancelled">; function isOpenExecutionNodeStatus(status: OrchestrationV2ExecutionNode["status"]): boolean {  return status === "pending" || status === "running" || status === "waiting";} function isRunOwnedSubagentTerminalStatus(  status: ProviderTerminalEvent["status"],): status is RunOwnedSubagentTerminalStatus {  return status === "interrupted" || status === "failed" || status === "cancelled";} export function canRouteRelatedSubagent(status: OrchestrationV2Subagent["status"]): boolean {  return status !== "interrupted" && status !== "failed" && status !== "cancelled";} function emptyOpenRunOwnedSubagentProjection(): OpenRunOwnedSubagentProjection {  return {    subagents: new Map(),    turnItems: new Map(),    childTurnItems: new Map(),    nodes: new Map(),    linkedChildThreadIds: new Set(),  };} function withLinkedChildThreadId(  current: OpenRunOwnedSubagentProjection,  childThreadId: ThreadId | null,): OpenRunOwnedSubagentProjection {  if (childThreadId === null || current.linkedChildThreadIds.has(childThreadId)) {    return current;  }  const linkedChildThreadIds = new Set(current.linkedChildThreadIds);  linkedChildThreadIds.add(childThreadId);  return { ...current, linkedChildThreadIds };} export function cascadeTerminalizeRunOwnedSubagents(input: {  readonly run: OrchestrationV2Run;  readonly open: OpenRunOwnedSubagentProjection;  readonly status: RunOwnedSubagentTerminalStatus;  readonly completedAt: DateTime.Utc;  readonly allocateEventId: () => Effect.Effect<EventId, IdAllocatorV2AllocationError>;}): Effect.Effect<ReadonlyArray<OrchestrationV2DomainEvent>, IdAllocatorV2AllocationError> {  return Effect.gen(function* () {    const events: Array<OrchestrationV2DomainEvent> = [];    // Prefer lifetime linkage over currently-open rows: subagent/turn-item    // snapshots may terminalize before the linked child-thread node settles.    const childThreadIds = new Set(input.open.linkedChildThreadIds);    for (const item of [...input.open.subagents.values(), ...input.open.turnItems.values()]) {      if (item.childThreadId !== null) {        childThreadIds.add(item.childThreadId);      }    }    const keys = new Set<NodeId>([      ...input.open.subagents.keys(),      ...input.open.turnItems.keys(),      ...input.open.nodes.keys(),    ]);    for (const key of keys) {      const subagent = input.open.subagents.get(key);      if (subagent !== undefined && !isTerminalSubagentStatus(subagent.status)) {        events.push({          id: yield* input.allocateEventId(),          type: "subagent.updated",          threadId: subagent.threadId,          runId: input.run.id,          nodeId: subagent.id,          driver: subagent.driver,          providerInstanceId: subagent.providerInstanceId,          occurredAt: input.completedAt,          payload: {            ...subagent,            status: input.status,            completedAt: input.completedAt,            updatedAt: input.completedAt,          },        });      }      const node = input.open.nodes.get(key);      if (        node !== undefined &&        ((node.threadId === input.run.threadId && node.runId === input.run.id) ||          childThreadIds.has(node.threadId)) &&        isOpenExecutionNodeStatus(node.status)      ) {        events.push({          id: yield* input.allocateEventId(),          type: "node.updated",          threadId: node.threadId,          runId: node.runId ?? input.run.id,          nodeId: node.id,          providerInstanceId: input.run.providerInstanceId,          occurredAt: input.completedAt,          payload: {            ...node,            status: input.status,            completedAt: input.completedAt,          },        });      }      const turnItem = input.open.turnItems.get(key);      if (        turnItem !== undefined &&        turnItem.runId === input.run.id &&        !isTerminalTurnItemStatus(turnItem.status)      ) {        events.push({          id: yield* input.allocateEventId(),          type: "turn-item.updated",          threadId: turnItem.threadId,          runId: input.run.id,          ...(turnItem.nodeId === null ? {} : { nodeId: turnItem.nodeId }),          providerInstanceId: input.run.providerInstanceId,          occurredAt: input.completedAt,          payload: {            ...turnItem,            status: input.status,            completedAt: input.completedAt,            updatedAt: input.completedAt,          },        });      }    }    for (const turnItem of input.open.childTurnItems.values()) {      if (!childThreadIds.has(turnItem.threadId) || isTerminalTurnItemStatus(turnItem.status)) {        continue;      }      events.push({        id: yield* input.allocateEventId(),        type: "turn-item.updated",        threadId: turnItem.threadId,        runId: turnItem.runId ?? input.run.id,        ...(turnItem.nodeId === null ? {} : { nodeId: turnItem.nodeId }),        providerInstanceId: input.run.providerInstanceId,        occurredAt: input.completedAt,        payload: {          ...turnItem,          ...("streaming" in turnItem ? { streaming: false } : {}),          status: input.status,          completedAt: input.completedAt,          updatedAt: input.completedAt,        },      });    }    return events;  });} export function finalProviderThreadStatus(  disposition: ProviderTerminalEvent["threadDisposition"],): OrchestrationV2ProviderThread["status"] {  return disposition === "broken" ? "error" : "idle";} export function makeProviderEventRoutingState(input: {  readonly identity: ProviderEventRouteIdentity;  readonly inheritedBackgroundTurnItems?: ReadonlyArray<InheritedBackgroundTurnItemRoute>;  readonly providerTurnId: ProviderTurnId | null;  readonly relatedThreadIds?: ReadonlyArray<ThreadId>;  readonly relatedProviderThreadIds?: ReadonlyArray<ProviderThreadId>;}): ProviderEventRoutingState {  return {    ownedThreadIds: new Set([input.identity.threadId, ...(input.relatedThreadIds ?? [])]),    ownedProviderThreadIds: new Set([      input.identity.providerThreadId,      ...(input.relatedProviderThreadIds ?? []),    ]),    ownedProviderTurnIds:      input.providerTurnId === null ? new Set() : new Set([input.providerTurnId]),    inheritedBackgroundTurnItems: new Map(      (input.inheritedBackgroundTurnItems ?? []).map((item) => [item.id, item.runId]),    ),    rootProviderTurnId: input.providerTurnId,  };} export function routeProviderEvent(  event: ProviderAdapterV2Event,  input: ProviderEventRouteIdentity,  state: ProviderEventRoutingState,): readonly [boolean, ProviderEventRoutingState] {  const ownsThread = (threadId: ThreadId): boolean => state.ownedThreadIds.has(threadId);  const ownsChildThread = (threadId: ThreadId): boolean =>    threadId !== input.threadId && ownsThread(threadId);  const ownsRun = (runId: string | null): boolean => runId === input.runId;  const addProviderThread = (providerThreadId: ProviderThreadId): ProviderEventRoutingState => ({    ...state,    ownedProviderThreadIds: new Set([...state.ownedProviderThreadIds, providerThreadId]),  });  const addProviderTurn = (    providerTurnId: ProviderTurnId,    root: boolean,  ): ProviderEventRoutingState => ({    ...state,    ownedProviderTurnIds: new Set([...state.ownedProviderTurnIds, providerTurnId]),    rootProviderTurnId: root ? providerTurnId : state.rootProviderTurnId,  });   switch (event.type) {    case "provider_session.updated":      // The session manager persists process-wide status once for every      // attached app thread before broadcasting the adapter event.      return [false, state];    case "app_thread.created": {      if (event.appThread.id === input.threadId) {        return [true, state];      }      const isOwnedSubagent =        event.appThread.lineage.relationshipToParent === "subagent" &&        event.appThread.lineage.parentThreadId !== null &&        ownsThread(event.appThread.lineage.parentThreadId);      if (!isOwnedSubagent) {        return [false, state];      }      return [        true,        {          ...state,          ownedThreadIds: new Set([...state.ownedThreadIds, event.appThread.id]),        },      ];    }    case "provider_thread.updated": {      const belongs =        state.ownedProviderThreadIds.has(event.providerThread.id) ||        (event.providerThread.appThreadId !== null && ownsThread(event.providerThread.appThreadId));      return belongs ? [true, addProviderThread(event.providerThread.id)] : [false, state];    }    case "provider_turn.updated": {      const isRoot = event.providerTurn.runAttemptId === input.attemptId;      const belongs =        isRoot ||        (event.providerTurn.providerThreadId !== input.providerThreadId &&          state.ownedProviderThreadIds.has(event.providerTurn.providerThreadId)) ||        state.ownedProviderTurnIds.has(event.providerTurn.id) ||        (event.threadId !== undefined && ownsChildThread(event.threadId));      return belongs ? [true, addProviderTurn(event.providerTurn.id, isRoot)] : [false, state];    }    case "node.updated": {      const belongs = ownsRun(event.node.runId) || ownsChildThread(event.node.threadId);      if (!belongs || event.node.providerThreadId === null) {        return [belongs, state];      }      return [true, addProviderThread(event.node.providerThreadId)];    }    case "subagent.updated":      return [ownsRun(event.subagent.runId) || ownsChildThread(event.subagent.threadId), state];    case "message.updated":      return [ownsRun(event.message.runId) || ownsChildThread(event.message.threadId), state];    case "turn_item.updated": {      if (ownsRun(event.turnItem.runId) || ownsChildThread(event.turnItem.threadId)) {        return [true, state];      }      const inheritedRunId = state.inheritedBackgroundTurnItems.get(event.turnItem.id);      // Preserve the item's original ownership while allowing the one live run      // to deliver an exact carryover identity selected from the projection.      const isInheritedBackgroundItem =        event.turnItem.threadId === input.threadId &&        event.turnItem.runId !== null &&        event.turnItem.runId === inheritedRunId &&        backgroundCapableTurnItemTypes.has(event.turnItem.type);      if (!isInheritedBackgroundItem) {        return [false, state];      }      if (!isTerminalTurnItemStatus(event.turnItem.status)) {        return [true, state];      }      const inheritedBackgroundTurnItems = new Map(state.inheritedBackgroundTurnItems);      inheritedBackgroundTurnItems.delete(event.turnItem.id);      return [true, { ...state, inheritedBackgroundTurnItems }];    }    case "plan.updated":      return [ownsRun(event.plan.runId) || ownsChildThread(event.plan.threadId), state];    case "runtime_request.updated":      return [        (event.threadId !== undefined && ownsChildThread(event.threadId)) ||          (event.runtimeRequest.providerTurnId !== null &&            state.ownedProviderTurnIds.has(event.runtimeRequest.providerTurnId)),        state,      ];    case "turn.terminal":      return [event.providerTurnId === state.rootProviderTurnId, state];  }} /** * ERRORS */export class RunExecutionStartError extends Schema.TaggedErrorClass<RunExecutionStartError>()(  "RunExecutionStartError",  {    commandId: CommandId,    runId: Schema.String,    cause: Schema.optional(Schema.Defect()),  },) {  override get message(): string {    return `Failed to start orchestration V2 run execution ${this.runId}.`;  }} export class RunExecutionIngestError extends Schema.TaggedErrorClass<RunExecutionIngestError>()(  "RunExecutionIngestError",  {    runId: Schema.String,    cause: Schema.optional(Schema.Defect()),  },) {  override get message(): string {    return `Failed while ingesting orchestration V2 run execution ${this.runId}.`;  }} export const RunExecutionServiceV2Error = Schema.Union([  RunExecutionStartError,  RunExecutionIngestError,]);export type RunExecutionServiceV2Error = typeof RunExecutionServiceV2Error.Type; /** * SERVICE DEFINITION */export interface RunExecutionServiceV2StartRootRunInput {  readonly commandId: CommandId;  readonly appThread: OrchestrationV2AppThread;  readonly providerSessionId: ProviderSessionId;  readonly session: ProviderAdapterV2SessionRuntime;  readonly run: OrchestrationV2Run;  readonly rootNode: OrchestrationV2ExecutionNode;  readonly checkpointScope: OrchestrationV2CheckpointScope;  readonly providerThread: OrchestrationV2ProviderThread;  readonly attempt: OrchestrationV2RunAttempt;

Lines 1–500 of 1409. Load more