import {  type ChatAttachment,  type CommandId,  MessageId,  type ModelSelection,  type OrchestrationV2Actor,  type OrchestrationV2Command,  type OrchestrationV2ConversationMessage,  type OrchestrationV2CreationSource,  type OrchestrationV2Run,  type OrchestrationV2ThreadShellSnapshot,  type OrchestrationV2ThreadProjection,  type OrchestrationV2ThreadShell,  type OrchestrationV2TurnItem,  ProjectId,  RunId,  ThreadId,} from "@t3tools/contracts";import * as Context from "effect/Context";import * as DateTime from "effect/DateTime";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Schema from "effect/Schema"; import {  OrchestratorDispatchError,  OrchestratorProjectionError,  OrchestratorV2,  type OrchestratorV2DispatchResult,  type OrchestratorV2Error,} from "./Orchestrator.ts";import {  LegacyV1ThreadImporter,  type LegacyV1ThreadImportError,} from "./LegacyV1ThreadImporter.ts"; export type ThreadManagementSendMode = "auto" | "queue" | "steer" | "restart"; export interface ThreadManagementProvenance {  readonly createdBy: OrchestrationV2Actor;  readonly creationSource: OrchestrationV2CreationSource;} export function withCreationProvenance(  command: OrchestrationV2Command,  provenance: ThreadManagementProvenance,): OrchestrationV2Command {  switch (command.type) {    case "thread.create":    case "message.dispatch":    case "thread.fork":    case "thread.merge_back":    case "delegated_task.request":      return { ...command, ...provenance };    default:      return command;  }} export function existingThreadIdsForCommand(  command: OrchestrationV2Command,): ReadonlyArray<ThreadId> {  switch (command.type) {    case "thread.create":      return [];    // Read-state commands only rewrite the thread payload's visited/unread    // watermark; they never touch messages, so they do not need the imported    // v1 transcript hydrated first. Visits fire on every thread-activity bump    // while a thread is open, so keeping them off the import path matters.    case "thread.visit":    case "thread.mark-unread":      return [];    case "thread.fork":      return [command.sourceThreadId];    case "thread.merge_back":      return command.sourceThreadId === command.targetThreadId        ? [command.sourceThreadId]        : [command.sourceThreadId, command.targetThreadId];    case "delegated_task.request":    case "delegated_task.wake-policy":    case "delegated_task.completion-delivery.acknowledge":    case "delegated_task.completion-delivery.dispose":      return [command.parentThreadId];    case "thread.created.record":      return command.parentThreadId === command.targetThreadId        ? [command.parentThreadId]        : [command.parentThreadId, command.targetThreadId];    default:      return [command.threadId];  }} export type ThreadManagementTerminalRunStatus = Extract<  OrchestrationV2Run["status"],  "completed" | "failed" | "cancelled" | "interrupted" | "rolled_back">; export interface ThreadManagementSendInput {  readonly projectId: ProjectId;  readonly commandId: CommandId;  readonly threadId: ThreadId;  readonly messageId: MessageId;  readonly text: string;  readonly attachments: ReadonlyArray<ChatAttachment>;  readonly modelSelection?: ModelSelection;  readonly mode: ThreadManagementSendMode;  readonly createdBy: OrchestrationV2Actor;  readonly creationSource: OrchestrationV2CreationSource;} export interface ThreadManagementSendResult {  readonly dispatch: OrchestratorV2DispatchResult;  readonly projection: OrchestrationV2ThreadProjection;  readonly message: OrchestrationV2ConversationMessage;  readonly run: OrchestrationV2Run;  /** Null for queued sends: the user turn item materializes when the queued turn starts. */  readonly turnItem: Extract<OrchestrationV2TurnItem, { readonly type: "user_message" }> | null;  readonly delivery: "started" | "queued" | "steered" | "restarted";} export interface ThreadManagementWaitInput {  readonly projectId: ProjectId;  readonly threadId: ThreadId;  readonly runId?: RunId;  readonly timeoutMs: number;  readonly pollIntervalMs?: number;} export interface ThreadManagementWaitResult {  readonly threadId: ThreadId;  readonly run: OrchestrationV2Run | null;  readonly timedOut: boolean;} export interface ThreadManagementInterruptInput {  readonly projectId: ProjectId;  readonly commandId: CommandId;  readonly threadId: ThreadId;  readonly runId?: RunId;  readonly reason?: string;} export type ThreadManagementInterruptResult =  | {      readonly type: "interrupt_requested";      readonly run: OrchestrationV2Run;      readonly dispatch: OrchestratorV2DispatchResult;    }  | { readonly type: "no_active_run" }  | {      readonly type: "already_terminal";      readonly run: OrchestrationV2Run & { readonly status: ThreadManagementTerminalRunStatus };    }; export class ThreadManagementThreadNotFoundError extends Schema.TaggedErrorClass<ThreadManagementThreadNotFoundError>()(  "ThreadManagementThreadNotFoundError",  {    projectId: ProjectId,    threadId: ThreadId,  },) {  override get message(): string {    return `Thread ${this.threadId} was not found in project ${this.projectId}.`;  }} export class ThreadManagementRunNotFoundError extends Schema.TaggedErrorClass<ThreadManagementRunNotFoundError>()(  "ThreadManagementRunNotFoundError",  {    threadId: ThreadId,    runId: RunId,  },) {  override get message(): string {    return `Run ${this.runId} does not belong to thread ${this.threadId}.`;  }} export class ThreadManagementThreadArchivedError extends Schema.TaggedErrorClass<ThreadManagementThreadArchivedError>()(  "ThreadManagementThreadArchivedError",  {    threadId: ThreadId,  },) {  override get message(): string {    return `Thread ${this.threadId} is archived and cannot receive messages.`;  }} export class ThreadManagementNoSteerableRunError extends Schema.TaggedErrorClass<ThreadManagementNoSteerableRunError>()(  "ThreadManagementNoSteerableRunError",  {    threadId: ThreadId,    mode: Schema.Literals(["steer", "restart"]),  },) {  override get message(): string {    return `Thread ${this.threadId} has no running turn that can be ${this.mode === "steer" ? "steered" : "restarted"}.`;  }} export class ThreadManagementThreadNotInterruptibleError extends Schema.TaggedErrorClass<ThreadManagementThreadNotInterruptibleError>()(  "ThreadManagementThreadNotInterruptibleError",  {    threadId: ThreadId,    runId: RunId,  },) {  override get message(): string {    return `Run ${this.runId} is not currently interruptible.`;  }} export class ThreadManagementProjectionLoadError extends Schema.TaggedErrorClass<ThreadManagementProjectionLoadError>()(  "ThreadManagementProjectionLoadError",  {    projectId: ProjectId,    threadId: ThreadId,    cause: Schema.Defect(),  },) {  override get message(): string {    return `Unable to load thread ${this.threadId} in project ${this.projectId}.`;  }} export class ThreadManagementProjectThreadsListError extends Schema.TaggedErrorClass<ThreadManagementProjectThreadsListError>()(  "ThreadManagementProjectThreadsListError",  {    projectId: ProjectId,    cause: Schema.Defect(),  },) {  override get message(): string {    return `Unable to list threads in project ${this.projectId}.`;  }} export class ThreadManagementDurableRunProjectionError extends Schema.TaggedErrorClass<ThreadManagementDurableRunProjectionError>()(  "ThreadManagementDurableRunProjectionError",  {    threadId: ThreadId,    messageId: MessageId,  },) {  override get message(): string {    return `Message ${this.messageId} was accepted on thread ${this.threadId} without a durable run projection.`;  }} export const ThreadManagementError = Schema.Union([  ThreadManagementThreadNotFoundError,  ThreadManagementRunNotFoundError,  ThreadManagementThreadArchivedError,  ThreadManagementNoSteerableRunError,  ThreadManagementThreadNotInterruptibleError,  ThreadManagementProjectionLoadError,  ThreadManagementProjectThreadsListError,  ThreadManagementDurableRunProjectionError,]);export type ThreadManagementError = typeof ThreadManagementError.Type; type ThreadManagementFailure = ThreadManagementError | OrchestratorV2Error; export interface ThreadManagementServiceShape {  readonly ensureLegacyTranscript: (    threadId: ThreadId,  ) => Effect.Effect<void, LegacyV1ThreadImportError>;  readonly dispatch: (    command: OrchestrationV2Command,  ) => Effect.Effect<OrchestratorV2DispatchResult, OrchestratorV2Error>;  readonly getCommandReceipt: OrchestratorV2["Service"]["getCommandReceipt"];  readonly getThreadProjection: (    threadId: ThreadId,  ) => Effect.Effect<OrchestrationV2ThreadProjection, OrchestratorV2Error>;  readonly getCheckpointContext: OrchestratorV2["Service"]["getCheckpointContext"];  readonly getThreadSnapshot: OrchestratorV2["Service"]["getThreadSnapshot"];  readonly getThreadSnapshotWindow: OrchestratorV2["Service"]["getThreadSnapshotWindow"];  readonly getProjectThread: (input: {    readonly projectId: ProjectId;    readonly threadId: ThreadId;  }) => Effect.Effect<OrchestrationV2ThreadProjection, ThreadManagementError>;  readonly getShellSnapshot: (options?: {    readonly location?: "active" | "archive";  }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, OrchestratorV2Error>;  readonly getThreadShell: OrchestratorV2["Service"]["getThreadShell"];  readonly listProjectThreads: (input: {    readonly projectId: ProjectId;    readonly includeSubagents: boolean;  }) => Effect.Effect<ReadonlyArray<OrchestrationV2ThreadShell>, ThreadManagementError>;  readonly sendToThread: (    input: ThreadManagementSendInput,  ) => Effect.Effect<ThreadManagementSendResult, ThreadManagementFailure>;  readonly waitForThread: (    input: ThreadManagementWaitInput,  ) => Effect.Effect<ThreadManagementWaitResult, ThreadManagementError>;  readonly interruptThread: (    input: ThreadManagementInterruptInput,  ) => Effect.Effect<ThreadManagementInterruptResult, ThreadManagementFailure>;  readonly getThreadEventSequence: OrchestratorV2["Service"]["getThreadEventSequence"];  readonly streamStoredEvents: OrchestratorV2["Service"]["streamStoredEvents"];  readonly streamStoredEventsFrom: OrchestratorV2["Service"]["streamStoredEventsFrom"];  readonly streamDomainEvents: OrchestratorV2["Service"]["streamDomainEvents"];} export class ThreadManagementService extends Context.Service<  ThreadManagementService,  ThreadManagementServiceShape>()("t3/orchestration-v2/ThreadManagementService") {} export function isActiveRun(run: OrchestrationV2Run): boolean {  return (    run.status === "preparing" ||    run.status === "starting" ||    run.status === "running" ||    run.status === "waiting"  );} export function isTerminalRunStatus(  status: OrchestrationV2Run["status"],): status is ThreadManagementTerminalRunStatus {  return (    status === "completed" ||    status === "failed" ||    status === "cancelled" ||    status === "interrupted" ||    status === "rolled_back"  );} export function latestRun(  projection: OrchestrationV2ThreadProjection,): OrchestrationV2Run | undefined {  return projection.runs.toSorted((left, right) => right.ordinal - left.ordinal)[0];} export function latestActiveRun(  projection: OrchestrationV2ThreadProjection,): OrchestrationV2Run | undefined {  return projection.runs    .filter(isActiveRun)    .toSorted((left, right) => right.ordinal - left.ordinal)[0];} export function latestSteerableRun(  projection: OrchestrationV2ThreadProjection,): OrchestrationV2Run | undefined {  return projection.runs    .filter(      (run) =>        run.status === "running" &&        run.activeAttemptId !== null &&        projection.providerTurns.some(          (turn) => turn.runAttemptId === run.activeAttemptId && turn.status === "running",        ),    )    .toSorted((left, right) => right.ordinal - left.ordinal)[0];} const make = Effect.gen(function* () {  const orchestrator = yield* OrchestratorV2;  const legacyImporter = yield* LegacyV1ThreadImporter;   const ensureLegacyTranscript = Effect.fn(    "orchestrationV2.threadManagement.ensureLegacyTranscript",  )(function* (threadId: ThreadId) {    yield* legacyImporter.ensureTranscript(threadId).pipe(      Effect.tapError((cause) =>        Effect.logWarning("Unable to hydrate migrated v1 thread transcript", {          threadId,          cause,        }),      ),    );  });   const ensureProjectionTranscript = (threadId: ThreadId) =>    ensureLegacyTranscript(threadId).pipe(      Effect.mapError(        (cause) =>          new OrchestratorProjectionError({            threadId,            cause,          }),      ),    );   const ensureCommandTranscripts = Effect.fn(    "orchestrationV2.threadManagement.ensureCommandTranscripts",  )(function* (command: OrchestrationV2Command) {    yield* Effect.forEach(      existingThreadIdsForCommand(command),      (threadId) => ensureLegacyTranscript(threadId),      { discard: true },    ).pipe(      Effect.mapError(        (cause) =>          new OrchestratorDispatchError({            commandId: command.commandId,            commandType: command.type,            cause,          }),      ),    );  });   const getThreadProjection: ThreadManagementServiceShape["getThreadProjection"] = (threadId) =>    ensureProjectionTranscript(threadId).pipe(      Effect.andThen(orchestrator.getThreadProjection(threadId)),    );   const getCheckpointContext: ThreadManagementServiceShape["getCheckpointContext"] = (threadId) =>    ensureProjectionTranscript(threadId).pipe(      Effect.andThen(orchestrator.getCheckpointContext(threadId)),    );   const getThreadSnapshot: ThreadManagementServiceShape["getThreadSnapshot"] = (threadId) =>    ensureProjectionTranscript(threadId).pipe(      Effect.andThen(orchestrator.getThreadSnapshot(threadId)),    );  const getThreadSnapshotWindow: ThreadManagementServiceShape["getThreadSnapshotWindow"] = (    threadId,    options,  ) =>    ensureProjectionTranscript(threadId).pipe(      Effect.andThen(orchestrator.getThreadSnapshotWindow(threadId, options)),    );   const dispatch: ThreadManagementServiceShape["dispatch"] = (command) =>    ensureCommandTranscripts(command).pipe(Effect.andThen(orchestrator.dispatch(command)));   const getProjectThread: ThreadManagementServiceShape["getProjectThread"] = (input) =>    getThreadProjection(input.threadId).pipe(      Effect.mapError(        (cause) =>          new ThreadManagementProjectionLoadError({            projectId: input.projectId,            threadId: input.threadId,            cause,          }),      ),      Effect.flatMap((projection) =>        projection.thread.projectId === input.projectId && projection.thread.deletedAt === null          ? Effect.succeed(projection)          : Effect.fail(              new ThreadManagementThreadNotFoundError({                projectId: input.projectId,                threadId: input.threadId,              }),            ),      ),    );   const listProjectThreads: ThreadManagementServiceShape["listProjectThreads"] = (input) =>    orchestrator.getShellSnapshot().pipe(      Effect.mapError(        (cause) =>          new ThreadManagementProjectThreadsListError({            projectId: input.projectId,            cause,          }),      ),      Effect.map((snapshot) =>        snapshot.threads          .filter((thread) => thread.projectId === input.projectId)          .filter(            (thread) =>              input.includeSubagents || thread.lineage.relationshipToParent !== "subagent",          )          .toSorted(            (left, right) =>              DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt) ||              right.id.localeCompare(left.id),          ),      ),    );   const sendToThread: ThreadManagementServiceShape["sendToThread"] = (input) =>    Effect.gen(function* () {      const target = yield* getProjectThread(input);      if (target.thread.archivedAt !== null) {        return yield* new ThreadManagementThreadArchivedError({          threadId: input.threadId,        });      }       const steerableRun = latestSteerableRun(target);      let dispatchMode: Extract<        OrchestrationV2Command,        { readonly type: "message.dispatch" }      >["dispatchMode"];      if (input.mode === "steer" || input.mode === "restart") {        if (steerableRun === undefined) {          return yield* new ThreadManagementNoSteerableRunError({            threadId: input.threadId,            mode: input.mode,          });

Lines 1–500 of 701. Load more