server/admission.ts

Rama main · réplica de lectura del repositorio original.

// Purity is the contract: no IO, no store reads, no adapter calls. Callers gather state, call admit(), and apply the decision.

import type { FollowUp } from "../shared/follow-up.ts";

export type AdmissionSurface =
  | "direct-busy" // 1:1 send to a thread whose turn is live: steer-first
  | "direct" // 1:1 send on a settled thread: start now or park
  | "room" // room send while the channel works: always queue
  | "room-steer" // manual steer of a queued room message: head only
  | "guarded" // guarded external send: refuse when anything is busy
  | "unattended" // routines/webhooks: busy means a missed run
  | "peer" // delegations and resume drains: slot-aware admission
  | "opened-thread" // a bot's self-opened thread: queue behind a slot
  | "incident"; // a report to the Chief's incidents thread: never refused

export type AdmissionQueueReason = "capacity" | "group-turn";

export type AdmissionRefusalCode = "guarded_busy" | "queue-head-only" | "busy" | "missing";

export const DRAIN_COALESCE_WINDOW_MS = 120_000;

/** Room turns read only the room-context window's last N messages, so a bigger group would append lines the responder never sees. */
export const DRAIN_COALESCE_MAX_ITEMS = 30;

/** The first item always drains, so an empty identity never strands a queue. */
export function drainCoalesceHead<T>(
  items: readonly T[],
  identityOf: (item: T) => string,
  queuedAtOf: (item: T) => number,
  maxItems?: number,
): T[] {
  const head: T[] = [];
  let previousIdentity: string | undefined;
  let previousAt: number | undefined;
  for (const item of items) {
    if (maxItems !== undefined && head.length >= maxItems) break;
    const identity = identityOf(item);
    const at = queuedAtOf(item);
    const gap = at - (previousAt ?? at);
    if (head.length === 0 || (identity === previousIdentity && gap >= 0 && gap <= DRAIN_COALESCE_WINDOW_MS)) {
      head.push(item);
    } else {
      break;
    }
    previousIdentity = identity;
    previousAt = at;
  }
  return head;
}

export interface AdmissionMessage {
  carriesImages?: boolean;
  engineCanSteerImages?: boolean;
  /** Steering into a turn waiting on a person's surface choice would race the choice. */
  pendingComputerSelection?: boolean;
  engineCanSteer?: boolean;
  /** What the person chose for a message sent while the turn runs; absent means steer where the turn can take it. */
  followUp?: FollowUp;
  /** Room queues drain one item per turn, so only the head may jump into the live speaker's turn. */
  isQueueHead?: boolean;
  speakerPresent?: boolean;
}

export interface AdmissionState {
  botPresent?: boolean;
  /** Broader than threadBusy: guarded sends refuse on it, ordinary sends do not. */
  botBusy?: boolean;
  threadBusy?: boolean;
  atCapacity?: boolean;
  groupTurn?: boolean;
  parksBehindCoordination?: boolean;
  roomWorking?: boolean;
  threadWorking?: boolean;
}

export type AdmissionSteerBlock = "images";

export type AdmissionDecision =
  | { action: "steer" }
  | { action: "queue"; reason?: AdmissionQueueReason; blockedBy?: AdmissionSteerBlock; chose?: Exclude<FollowUp, "steer"> }
  | { action: "start" }
  | { action: "refuse"; code: AdmissionRefusalCode };

function imagesBlockSteer(message: AdmissionMessage): boolean {
  return message.carriesImages === true && message.engineCanSteerImages !== true;
}

/** Whether a live room turn holds this work back. A person's own send (sent now, or queued earlier) waits only for a
 * room turn on its own thread; every other kind of work waits for any room turn of the bot. */
export function roomTurnHolds(roomTurn: { threadId: string } | null | undefined, threadId: string, personSend: boolean): boolean {
  return Boolean(roomTurn) && (!personSend || roomTurn!.threadId === threadId);
}

export function admit(
  surface: AdmissionSurface,
  message: AdmissionMessage = {},
  state: AdmissionState = {},
): AdmissionDecision {
  switch (surface) {
    case "direct-busy": {
      // Restart waits for the Stop it asks for. A chosen queue counts only where the turn could take steering: one that
      // cannot queues anyway, as it always has (person-first may still stop it for the person's message).
      if (message.followUp === "restart") return { action: "queue", chose: "restart" };
      if (message.engineCanSteer === true && message.followUp === "queue") return { action: "queue", chose: "queue" };
      if (message.engineCanSteer !== true || message.pendingComputerSelection) return { action: "queue" };
      if (imagesBlockSteer(message)) return { action: "queue", blockedBy: "images" };
      return { action: "steer" };
    }
    case "direct": {
      // groupTurn here is roomTurnHolds(): a room turn on another thread does not hold a person's own thread.
      if (state.atCapacity || state.threadBusy || state.groupTurn || state.parksBehindCoordination) {
        // Capacity outranks a room turn in the receipt because the person can fix it.
        const reason = state.atCapacity ? "capacity" : state.groupTurn ? "group-turn" : undefined;
        return { action: "queue", reason };
      }
      return { action: "start" };
    }
    case "room":
      return (state.threadWorking ?? state.roomWorking) ? { action: "queue" } : { action: "start" };
    case "room-steer": {
      if (!message.speakerPresent || message.engineCanSteer !== true) {
        return { action: "queue" };
      }
      if (imagesBlockSteer(message)) return { action: "queue", blockedBy: "images" };
      if (!message.isQueueHead) {
        return { action: "refuse", code: "queue-head-only" };
      }
      return { action: "steer" };
    }
    case "guarded": {
      // A guarded send is an external contract: never queued, never steered, never started under stale permissions.
      if (state.botBusy || state.threadBusy || state.atCapacity || state.parksBehindCoordination || state.groupTurn) {
        return { action: "refuse", code: "guarded_busy" };
      }
      return { action: "start" };
    }
    case "unattended": {
      if (state.botPresent === false) return { action: "refuse", code: "missing" };
      if (state.atCapacity || state.groupTurn) return { action: "refuse", code: "busy" };
      return { action: "start" };
    }
    case "peer": {
      if (state.threadBusy || state.atCapacity || state.groupTurn) return { action: "refuse", code: "busy" };
      return { action: "start" };
    }
    case "opened-thread": {
      if (state.atCapacity || state.groupTurn) {
        return { action: "queue", reason: state.atCapacity ? "capacity" : "group-turn" };
      }
      return { action: "start" };
    }
    case "incident": {
      // Never refused (a refused report is a lost one), never parked behind handed-on work, never steered.
      if (state.atCapacity || state.threadBusy || state.groupTurn) {
        return { action: "queue", reason: state.atCapacity ? "capacity" : state.groupTurn ? "group-turn" : undefined };
      }
      return { action: "start" };
    }
  }
}