import {  ApprovalRequestId,  type AssistantDeliveryMode,  CommandId,  MessageId,  type OrchestrationEvent,  type OrchestrationMessage,  type OrchestrationProposedPlanId,  CheckpointRef,  classifyTaskAgentKind,  EventId,  isToolLifecycleItemType,  ThreadId,  type ThreadTokenUsageSnapshot,  TurnId,  type OrchestrationCheckpointSummary,  type OrchestrationProposedPlan,  type OrchestrationThread,  type OrchestrationThreadActivity,  type ProviderRuntimeEvent,} from "@t3tools/contracts";import * as Cache from "effect/Cache";import * as Cause from "effect/Cause";import * as Crypto from "effect/Crypto";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 Stream from "effect/Stream";import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { ProviderService } from "../../provider/Services/ProviderService.ts";import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts";import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts";import { isGitRepository } from "../../git/Utils.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ThreadBackgroundLivenessService } from "../ThreadBackgroundLiveness.ts";import { ThreadPlanProgressService } from "../ThreadPlanProgress.ts";import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import {  ProviderRuntimeIngestionService,  type ProviderRuntimeIngestionShape,} from "../Services/ProviderRuntimeIngestion.ts";import { projectActivityPayload } from "../ActivityPayloadProjection.ts";import { forkParked } from "../../serverActivation.ts";import { ServerSettingsService } from "../../serverSettings.ts";import { canReplaceThreadTitle } from "../threadTitles.ts"; const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`;const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`; // Fallback when the in-memory description cache no longer has the task name// (server restart, session-exit sweep, TTL/capacity eviction): earlier// task.started/task.progress activities for the task are persisted with it.function findTaskTitleInActivities(  activities: ReadonlyArray<OrchestrationThreadActivity> | undefined,  taskId: string,): string | undefined {  if (!activities) {    return undefined;  }  for (let index = activities.length - 1; index >= 0; index -= 1) {    const activity = activities[index];    if (!activity || (activity.kind !== "task.started" && activity.kind !== "task.progress")) {      continue;    }    const payload =      activity.payload && typeof activity.payload === "object"        ? (activity.payload as { taskId?: unknown; title?: unknown; detail?: unknown })        : undefined;    if (payload?.taskId !== taskId) {      continue;    }    const title =      typeof payload.title === "string"        ? payload.title        : activity.kind === "task.started" && typeof payload.detail === "string"          ? payload.detail          : undefined;    if (title && title.trim().length > 0) {      return title;    }  }  return undefined;} interface AssistantSegmentState {  baseKey: string;  nextSegmentIndex: number;  activeMessageId: MessageId | null;} const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000;const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);const BUFFERED_PROPOSED_PLAN_BY_ID_CACHE_CAPACITY = 10_000;const BUFFERED_PROPOSED_PLAN_BY_ID_TTL = Duration.minutes(120);const TASK_DESCRIPTION_BY_TASK_CACHE_CAPACITY = 10_000;const TASK_DESCRIPTION_BY_TASK_TTL = Duration.minutes(120);const MAX_BUFFERED_ASSISTANT_CHARS = 24_000;const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0"; type TurnStartRequestedDomainEvent = Extract<  OrchestrationEvent,  { type: "thread.turn-start-requested" }>; type RuntimeIngestionInput =  | {      source: "runtime";      event: ProviderRuntimeEvent;    }  | {      source: "domain";      event: TurnStartRequestedDomainEvent;    }; function toTurnId(value: TurnId | string | undefined): TurnId | undefined {  return value === undefined ? undefined : TurnId.make(String(value));} function toApprovalRequestId(value: string | undefined): ApprovalRequestId | undefined {  return value === undefined ? undefined : ApprovalRequestId.make(value);} function sameId(left: string | null | undefined, right: string | null | undefined): boolean {  if (left === null || left === undefined || right === null || right === undefined) {    return false;  }  return left === right;} function hasAssistantMessageForTurn(  messages: ReadonlyArray<OrchestrationMessage>,  turnId: TurnId,  options?: { readonly streamingOnly?: boolean },): boolean {  for (let index = 0; index < messages.length; index += 1) {    const message = messages[index];    if (!message) {      continue;    }    if (message.role !== "assistant" || message.turnId !== turnId) {      continue;    }    if (options?.streamingOnly === true && !message.streaming) {      continue;    }    return true;  }  return false;} function findMessageById(  messages: ReadonlyArray<OrchestrationMessage>,  messageId: MessageId,): OrchestrationMessage | undefined {  for (let index = 0; index < messages.length; index += 1) {    const message = messages[index];    if (message?.id === messageId) {      return message;    }  }  return undefined;} function findProposedPlanById(  proposedPlans: ReadonlyArray<    Pick<OrchestrationProposedPlan, "id" | "createdAt" | "implementedAt" | "implementationThreadId">  >,  planId: string,):  | Pick<OrchestrationProposedPlan, "id" | "createdAt" | "implementedAt" | "implementationThreadId">  | undefined {  for (let index = 0; index < proposedPlans.length; index += 1) {    const proposedPlan = proposedPlans[index];    if (proposedPlan?.id === planId) {      return proposedPlan;    }  }  return undefined;} function hasCheckpointForTurn(  checkpoints: ReadonlyArray<OrchestrationCheckpointSummary>,  turnId: TurnId,): boolean {  for (let index = 0; index < checkpoints.length; index += 1) {    if (checkpoints[index]?.turnId === turnId) {      return true;    }  }  return false;} function maxCheckpointTurnCount(  checkpoints: ReadonlyArray<OrchestrationCheckpointSummary>,): number {  let maxTurnCount = 0;  for (let index = 0; index < checkpoints.length; index += 1) {    const checkpoint = checkpoints[index];    if (checkpoint && checkpoint.checkpointTurnCount > maxTurnCount) {      maxTurnCount = checkpoint.checkpointTurnCount;    }  }  return maxTurnCount;} function truncateDetail(value: string, limit = 180): string {  return value.length > limit ? `${value.slice(0, limit - 3)}...` : value;} function normalizeProposedPlanMarkdown(planMarkdown: string | undefined): string | undefined {  const trimmed = planMarkdown?.trim();  if (!trimmed) {    return undefined;  }  return trimmed;} function hasRenderableAssistantText(text: string | undefined): boolean {  return (text?.trim().length ?? 0) > 0;} function proposedPlanIdForTurn(threadId: ThreadId, turnId: TurnId): string {  return `plan:${threadId}:turn:${turnId}`;} function proposedPlanIdFromEvent(event: ProviderRuntimeEvent, threadId: ThreadId): string {  const turnId = toTurnId(event.turnId);  if (turnId) {    return proposedPlanIdForTurn(threadId, turnId);  }  if (event.itemId) {    return `plan:${threadId}:item:${event.itemId}`;  }  return `plan:${threadId}:event:${event.eventId}`;} function assistantSegmentBaseKeyFromEvent(event: ProviderRuntimeEvent): string {  return String(event.itemId ?? event.turnId ?? event.eventId);} function assistantSegmentMessageId(baseKey: string, segmentIndex: number): MessageId {  return MessageId.make(    segmentIndex === 0 ? `assistant:${baseKey}` : `assistant:${baseKey}:segment:${segmentIndex}`,  );}function buildContextWindowActivityPayload(  event: ProviderRuntimeEvent,): ThreadTokenUsageSnapshot | undefined {  if (event.type !== "thread.token-usage.updated" || event.payload.usage.usedTokens <= 0) {    return undefined;  }  return event.payload.usage;} function normalizeRuntimeTurnState(  value: string | undefined,): "completed" | "failed" | "interrupted" | "cancelled" {  switch (value) {    case "failed":    case "interrupted":    case "cancelled":    case "completed":      return value;    default:      return "completed";  }} function orchestrationSessionStatusFromRuntimeState(  state: "starting" | "running" | "waiting" | "ready" | "interrupted" | "stopped" | "error",): "starting" | "running" | "ready" | "interrupted" | "stopped" | "error" {  switch (state) {    case "starting":      return "starting";    case "running":    case "waiting":      return "running";    case "ready":      return "ready";    case "interrupted":      return "interrupted";    case "stopped":      return "stopped";    case "error":      return "error";  }} function sessionStatusAllowsActiveTurn(  status: ReturnType<typeof orchestrationSessionStatusFromRuntimeState>,): boolean {  return status === "starting" || status === "running";} function requestKindFromCanonicalRequestType(  requestType: string | undefined,): "command" | "file-read" | "file-change" | undefined {  switch (requestType) {    case "command_execution_approval":    case "exec_command_approval":      return "command";    case "file_read_approval":      return "file-read";    case "file_change_approval":    case "apply_patch_approval":      return "file-change";    default:      return undefined;  }} /** * Copies the optional TaskAgentLinkage bundle from a task.* runtime payload * into the persisted activity payload. Identity fields ride on every row so * client folds survive activity retention; absent fields stay absent. */function taskLinkageActivityFields(payload: Record<string, unknown>): Record<string, unknown> {  const fields: Record<string, unknown> = {    // Server-stamped classification: persisted rows are self-describing, so    // clients trust the stamp instead of re-deriving agent-vs-background    // from taskType denylists and marker heuristics (legacy rows without a    // stamp keep the client fallback).    agentKind: classifyTaskAgentKind({      taskType: typeof payload.taskType === "string" ? payload.taskType : undefined,      agentId: typeof payload.agentId === "string" ? payload.agentId : undefined,    }),  };  for (const key of [    "taskType",    "agentId",    "title",    "role",    "model",    "effort",    "toolUseId",    "parentAgentId",    "workflowName",    "agentIndex",    "phaseIndex",    "phaseTitle",    "phases",    "attempt",    "runHandles",    "outputFile",    "agentPath",    "timelineBypass",    "typedUsage",    "status",    "error",  ] as const) {    if (payload[key] !== undefined) {      fields[key] = payload[key];    }  }  return fields;} export function runtimeEventToActivities(  event: ProviderRuntimeEvent,  taskTitle?: string,): ReadonlyArray<OrchestrationThreadActivity> {  const maybeSequence = (() => {    const eventWithSequence = event as ProviderRuntimeEvent & { sessionSequence?: number };    return eventWithSequence.sessionSequence !== undefined      ? { sequence: eventWithSequence.sessionSequence }      : {};  })();  switch (event.type) {    case "request.opened": {      if (event.payload.requestType === "tool_user_input") {        return [];      }      const requestKind = requestKindFromCanonicalRequestType(event.payload.requestType);      return [        {          id: event.eventId,          createdAt: event.createdAt,          tone: "approval",          kind: "approval.requested",          summary:            requestKind === "command"              ? "Command approval requested"              : requestKind === "file-read"                ? "File-read approval requested"                : requestKind === "file-change"                  ? "File-change approval requested"                  : "Approval requested",          payload: {            requestId: toApprovalRequestId(event.requestId),            ...(requestKind ? { requestKind } : {}),            requestType: event.payload.requestType,            ...(event.payload.detail ? { detail: event.payload.detail } : {}),          },          turnId: toTurnId(event.turnId) ?? null,          ...maybeSequence,        },      ];    }     case "request.resolved": {      if (event.payload.requestType === "tool_user_input") {        return [];      }      const requestKind = requestKindFromCanonicalRequestType(event.payload.requestType);      return [        {          id: event.eventId,          createdAt: event.createdAt,          tone: "approval",          kind: "approval.resolved",          summary: "Approval resolved",          payload: {            requestId: toApprovalRequestId(event.requestId),            ...(requestKind ? { requestKind } : {}),            requestType: event.payload.requestType,            ...(event.payload.decision ? { decision: event.payload.decision } : {}),          },          turnId: toTurnId(event.turnId) ?? null,          ...maybeSequence,        },      ];    }     case "runtime.error": {      return [        {          id: event.eventId,          createdAt: event.createdAt,          tone: "error",          kind: "runtime.error",          summary: "Runtime error",          payload: {            message: truncateDetail(event.payload.message),          },          turnId: toTurnId(event.turnId) ?? null,          ...maybeSequence,        },      ];    }     case "tool.denied": {      return [        {          id: event.eventId,          createdAt: event.createdAt,          tone: "error",          kind: "tool.denied",          summary: `Tool denied: ${event.payload.toolName}`,          payload: {            toolName: event.payload.toolName,            ...(event.payload.toolUseId ? { toolUseId: event.payload.toolUseId } : {}),            ...(event.payload.reason ? { detail: truncateDetail(event.payload.reason) } : {}),            ...(event.payload.agentId ? { agentId: event.payload.agentId } : {}),          },          turnId: toTurnId(event.turnId) ?? null,          ...maybeSequence,        },      ];    }     case "runtime.warning": {      return [        {          id: event.eventId,          createdAt: event.createdAt,          tone: "info",          kind: "runtime.warning",          // Use the adapter-supplied message as the row label so the work log          // shows what the warning was about, not a generic "Runtime warning".          summary: truncateDetail(event.payload.message, 120),          payload: {            message: truncateDetail(event.payload.message),            ...(event.payload.detail !== undefined ? { detail: event.payload.detail } : {}),          },          turnId: toTurnId(event.turnId) ?? null,          ...maybeSequence,        },      ];    }     case "turn.plan.updated": {      return [        {          id: event.eventId,          createdAt: event.createdAt,          tone: "info",          kind: "turn.plan.updated",          summary: "Plan updated",          payload: {            plan: event.payload.plan,            ...(event.payload.explanation !== undefined              ? { explanation: event.payload.explanation }              : {}),          },          turnId: toTurnId(event.turnId) ?? null,          ...maybeSequence,

Lines 1–500 of 2082. Load more