import { agentProviderDescriptor } from "@openbot/contracts/agent-providers";
import type {
  ConversationMessageSender,
  QueueDelivery,
  QueueSnapshot,
  ReorderQueueInput,
  SteerQueuedMessageInput,
  UpdateQueuedMessageInput,
} from "@openbot/contracts/ipc";
import { QueueEditRejectedError, type QueueEditRequest } from "@openbot/contracts/team-protocol/queue-edit-v1";
import { sourceText } from "@openbot/i18n/source";
import { Effect, Schema } from "effect";
import type { AgentProvider } from "../agent-client";
import type { AgentStore } from "../agent-store";
import type { ChannelAssignment } from "../channel-store";
import { promptMessage } from "../conversation-snapshots";
import { causeHelpers } from "../effect-boundary";
import type { MailboxStore } from "../mailbox-store";
import { decodeRecordResponse } from "../protocol";
import type { ConversationRuntime } from "./conversation-runtime";
import type { CustomEndpoints } from "./custom-endpoints";
import { agentNamesById, deliveryPromptInput, type TagReferenceSource } from "./delivery-content";
import { steerNote } from "./developer-instructions";
import { type DrainScheduler, REMOVED_ENDPOINT_MESSAGE } from "./drain-scheduler";
import type { MailboxSync } from "./mailbox-sync";
import type { ProviderRuntime } from "./provider-runtime";
import type { RoutineScheduler } from "./routine-scheduler";
import { providerForAgent } from "./thread-items";

export interface QueueControlsHooks {
  /** The channel task that owns a delivery. Read late: the channel service is built after this. */
  channelAssignment(deliveryId: string): ChannelAssignment | null;
  /**
   * The thread of the task that a delivery runs in, or null for the agent's own chat. It throws for a
   * deleted task.
   */
  taskThreadId(deliveryId: string): string | null;
  /** The tabs and threads of this agent, as the tags of a steered message name them. */
  /** `threadId` is the public thread that gets the prompt. */
  tagReferences(agentId: string, threadId: string | null): TagReferenceSource;
}

export interface QueueControlsOptions {
  store: AgentStore;
  mailbox: MailboxStore;
  mailboxSync: MailboxSync;
  conversation: ConversationRuntime;
  providers: ProviderRuntime;
  endpoints: CustomEndpoints;
  drain: DrainScheduler;
  routines: RoutineScheduler;
  hooks: QueueControlsHooks;
}

/**
 * Owns the user's changes to messages that wait in an agent's queue: cancel, edit, update, reorder
 * and steer into the running turn. Channel work is refused here; the channel task controls own it.
 *
 * It never imports the agent service facade.
 */
export class QueueControls {
  readonly #store: AgentStore;
  readonly #mailbox: MailboxStore;
  readonly #mailboxSync: MailboxSync;
  readonly #conversation: ConversationRuntime;
  readonly #providers: ProviderRuntime;
  readonly #endpoints: CustomEndpoints;
  readonly #drain: DrainScheduler;
  readonly #routines: RoutineScheduler;
  readonly #hooks: QueueControlsHooks;

  constructor(options: QueueControlsOptions) {
    this.#store = options.store;
    this.#mailbox = options.mailbox;
    this.#mailboxSync = options.mailboxSync;
    this.#conversation = options.conversation;
    this.#providers = options.providers;
    this.#endpoints = options.endpoints;
    this.#drain = options.drain;
    this.#routines = options.routines;
    this.#hooks = options.hooks;
  }

  readonly cancel = Effect.fn("QueueControls.cancel")(function* (
    this: QueueControls,
    agentId: string,
    deliveryId: string,
  ): Effect.fn.Return<void, QueueOperationFailed> {
    if (this.#hooks.channelAssignment(deliveryId))
      return yield* new QueueOperationFailed({ cause: new Error(sourceText("error.backend.useChannelTaskControls")) });
    const sender = this.#mailbox.getDelivery(deliveryId)?.delivery.sender;
    yield* this.#mailbox.cancel(agentId, deliveryId).pipe(toQueueOperationFailed);
    this.#mailboxSync.emitQueue(agentId);
    this.#drain.scheduleDrain(agentId);
    // The requester may hold the other answers until this request ends.
    if (sender?.kind === "agent") this.#drain.scheduleDrain(sender.agentId);
  }, Effect.uninterruptible);

  readonly edit = Effect.fn("QueueControls.edit")(function* (
    this: QueueControls,
    agentId: string,
    input: QueueEditRequest,
    sender?: ConversationMessageSender,
  ): Effect.fn.Return<QueueSnapshot, QueueOperationFailed> {
    const finished = yield* queueStep(() =>
      this.#mailbox.finishedQueueEditAction(agentId, input.deliveryId, input.editId),
    );
    if (finished) {
      if (input.action === "begin" || input.action === "retain-attachments")
        return yield* new QueueOperationFailed({
          cause: new QueueEditRejectedError(sourceText("error.backend.editFinished")),
        });
      // The uploads belong to an edit that is over, so they never stay behind.
      if (input.action === "save")
        yield* Effect.forEach(
          input.attachmentDraftIds,
          (id) => this.#mailbox.discardDraft(id).pipe(toQueueOperationFailed),
          {
            concurrency: "unbounded",
            discard: true,
          },
        );
      // Only a retry of the action that finished can report success. A Save that follows a
      // finished Cancel never reached the message, so the client must keep its text.
      if (input.action !== finished)
        return yield* new QueueOperationFailed({
          cause: new QueueEditRejectedError(
            finished === "cancel" ? sourceText("error.backend.editCancelled") : sourceText("error.backend.editSaved"),
          ),
        });
      if (
        input.action === "save" &&
        !this.#mailbox.matchesFinishedQueueSave(
          agentId,
          input.deliveryId,
          input.editId,
          input.text,
          input.keepAttachmentIds,
          input.attachmentDraftIds,
        )
      )
        return yield* new QueueOperationFailed({
          cause: new QueueEditRejectedError(sourceText("error.backend.editSavedDifferent")),
        });
      this.#drain.scheduleDrain(agentId);
      this.#mailboxSync.emitQueue(agentId);
      return this.#mailboxSync.queueSnapshot(agentId);
    }
    if (this.#hooks.channelAssignment(input.deliveryId))
      return yield* new QueueOperationFailed({ cause: new Error(sourceText("error.backend.useChannelTaskControls")) });
    if (input.action === "begin")
      yield* queueStep(() => this.#mailbox.beginQueueEdit(agentId, input.deliveryId, input.editId));
    else {
      if (input.action === "retain-attachments")
        yield* queueStep(() =>
          this.#mailbox.retainQueueEditAttachments(agentId, input.deliveryId, input.editId, input.attachmentDraftIds),
        );
      if (input.action === "save") {
        yield* this.#mailbox
          .updateQueuedMessage(
            agentId,
            input.deliveryId,
            input.text,
            input.keepAttachmentIds,
            input.attachmentDraftIds,
            input.editId,
            sender,
          )
          .pipe(toQueueOperationFailed);
        const snapshot = this.#conversation.snapshotToUpdate(agentId);
        if (snapshot) {
          this.#mailboxSync.syncMailboxMessages(snapshot);
          this.#conversation.emitConversation(snapshot, "queue.message-updated");
        }
      }
      if (input.action === "cancel")
        yield* queueStep(() => this.#mailbox.finishQueueEdit(agentId, input.deliveryId, input.editId));
      this.#drain.scheduleDrain(agentId);
    }
    this.#mailboxSync.emitQueue(agentId);
    return this.#mailbox.listQueue(agentId);
  }, Effect.uninterruptible);

  readonly update = Effect.fn("QueueControls.update")(function* (
    this: QueueControls,
    input: UpdateQueuedMessageInput,
    sender?: ConversationMessageSender,
  ): Effect.fn.Return<void, QueueOperationFailed> {
    if (this.#hooks.channelAssignment(input.deliveryId))
      return yield* new QueueOperationFailed({ cause: new Error(sourceText("error.backend.useChannelTaskControls")) });
    yield* this.#mailbox
      .updateQueuedMessage(
        input.agentId,
        input.deliveryId,
        input.text,
        input.keepAttachmentIds,
        input.attachmentDraftIds,
        undefined,
        sender,
      )
      .pipe(toQueueOperationFailed);
    const snapshot = this.#conversation.snapshotToUpdate(input.agentId);
    if (snapshot) this.#mailboxSync.syncMailboxMessages(snapshot);
    this.#mailboxSync.emitQueue(input.agentId);
    if (snapshot) this.#conversation.emitConversation(snapshot, "queue.message-updated");
    this.#drain.scheduleDrain(input.agentId);
  }, Effect.uninterruptible);

  readonly reorder = Effect.fn("QueueControls.reorder")(function* (
    this: QueueControls,
    input: ReorderQueueInput,
  ): Effect.fn.Return<void, QueueOperationFailed> {
    if (input.deliveryIds.some((id) => this.#hooks.channelAssignment(id)))
      return yield* new QueueOperationFailed({
        cause: new Error(sourceText("error.backend.useChannelTaskControlsWork")),
      });
    // A side thread orders only its own messages. They take the places that they hold now in the
    // whole queued order, so the other work of the agent does not move. A message that an editor
    // holds is not in the order the panel sends, and keeps its place.
    const { taskId } = input;
    if (taskId !== undefined) {
      const thread = new Set(this.#mailbox.laneQueuedIds(input.agentId, taskId));
      const moved = new Set(input.deliveryIds);
      // A stale order names a message that left the thread, or leaves out one that is not edited.
      const left = [...thread].filter((id) => !moved.has(id));
      if (
        input.deliveryIds.some((id) => !thread.has(id)) ||
        left.some((id) => !this.#mailbox.getDelivery(id)?.delivery.editing)
      )
        return yield* new QueueOperationFailed({ cause: new Error(sourceText("error.backend.queueOrderStale")) });
      let next = 0;
      const order = this.#mailbox
        .queuedDeliveryIds(input.agentId)
        .map((id) => (moved.has(id) ? (input.deliveryIds[next++] ?? id) : id));
      yield* this.#mailbox.reorderQueue(input.agentId, order).pipe(toQueueOperationFailed);
      this.#mailboxSync.emitQueue(input.agentId);
      return;
    }
    // The queue the user reads holds no channel or messaging work, so the order it sends names the
    // normal messages alone, and the mailbox reads the whole queued order. That work stays at the
    // head: it reserved the agent before these messages arrived.
    const executionDeliveryIds = this.#mailbox.queuedExecutionDeliveryIds(input.agentId);
    yield* this.#mailbox
      .reorderQueue(input.agentId, [...executionDeliveryIds, ...input.deliveryIds])
      .pipe(toQueueOperationFailed);
    this.#mailboxSync.emitQueue(input.agentId);
  }, Effect.uninterruptible);

  readonly steer = Effect.fn("QueueControls.steer")(function* (this: QueueControls, input: SteerQueuedMessageInput) {
    const agent = yield* this.#store.existing(input.agentId).pipe(toQueueOperationFailed);
    const { client, session, snapshot, context, turnId, routineSession, publicThreadId } = yield* queueStep(() => {
      const client = this.#providers.requireReadyClientForAgent(agent);
      // A task message steers the task's turn, in the task's own provider session.
      const taskThreadId = this.#hooks.taskThreadId(input.deliveryId);
      const publicThreadId = taskThreadId ?? agent.threadId;
      const routineSession = publicThreadId ? this.#conversation.routineSession(agent.id, publicThreadId) : undefined;
      const session =
        routineSession ??
        (taskThreadId
          ? this.#store.database.activeProviderSession(taskThreadId, agent.provider)?.externalSessionId
          : this.#store.activeProviderSession(agent.id)?.externalSessionId);
      const snapshot = this.#conversation.ensureSnapshot(agent.id, taskThreadId ?? agent.threadId);
      if (!session || !snapshot.activeTurnId || snapshot.activeTurnId !== input.expectedTurnId)
        throw new Error(sourceText("error.backend.steerTurnChanged"));
      if (this.#hooks.channelAssignment(input.deliveryId))
        throw new Error(sourceText("error.backend.useChannelTaskControls"));
      const context = this.#mailbox.getDelivery(input.deliveryId);
      if (!context || context.delivery.recipientAgentId !== agent.id || context.delivery.status !== "queued")
        throw new Error(sourceText("error.backend.steerQueuedOnly"));
      const turnId = snapshot.activeTurnId;
      // Steering uses the running turn's endpoint even when the saved model has changed.
      if (!this.#endpoints.serves(this.#drain.modelForTurn(agent.id, turnId) ?? agent.model, client.provider))
        throw new Error(REMOVED_ENDPOINT_MESSAGE);
      return { client, session, snapshot, context, turnId, routineSession, publicThreadId };
    });
    // A routine run has its own session, which ends with the run. A message steered into it would not
    // stay in the session of its thread, so it waits for the next turn.
    if (routineSession) {
      yield* this.#mailbox.markSteerFallback(input.deliveryId, "steer-failed").pipe(toQueueOperationFailed);
      this.#mailboxSync.emitQueue(agent.id);
      return yield* new QueueOperationFailed({
        cause: new SteerUnsupportedError(sourceText("error.backend.steerRoutineRun")),
      });
    }
    if (agentProviderDescriptor(providerForAgent(agent)).steer !== "native") {
      yield* this.#mailbox.markSteerFallback(input.deliveryId, "provider-unsupported").pipe(toQueueOperationFailed);
      this.#mailboxSync.emitQueue(agent.id);
      return yield* new QueueOperationFailed({ cause: new SteerUnsupportedError() });
    }
    yield* this.#mailbox.markSteering(input.deliveryId, turnId).pipe(toQueueOperationFailed);
    this.#mailboxSync.emitQueue(agent.id);
    const note = steerNoteFor(providerForAgent(agent), context.delivery.sender, context.delivery.text);
    yield* client
      .request(
        "turn/steer",
        {
          threadId: session,
          expectedTurnId: turnId,
          clientUserMessageId: input.deliveryId,
          input: [
            // A block of its own, as the turn context blocks are, so history readers remove it.
            ...(note ? [{ type: "text" as const, text: note }] : []),
            ...deliveryPromptInput(context, {
              agentNames: agentNamesById(this.#store.list()),
              snapshot,
              routineRun:
                context.delivery.sender.kind === "routine" ? this.#routines.runForDelivery(input.deliveryId) : null,
              wokenWaits: context.delivery.sender.kind === "routine" ? this.#routines.wokenWaits(input.deliveryId) : [],
              answeredRoutine: this.#drain.answeredRoutine(context.delivery),
              tags: this.#hooks.tagReferences(agent.id, publicThreadId),
            }),
          ],
        },
        decodeRecordResponse,
      )
      .pipe(
        toQueueOperationFailed,
        // The turn may have ended while the request was in flight, and found nothing else to start
        // with this message out of the queue, so the drain is asked again.
        Effect.tapError(() =>
          this.#mailbox.restoreUnsteered(input.deliveryId, turnId).pipe(
            toQueueOperationFailed,
            Effect.andThen(
              queueStep(() => {
                this.#mailboxSync.syncMailboxMessages(snapshot);
                this.#mailboxSync.emitQueue(agent.id);
                this.#drain.scheduleDrain(agent.id);
              }),
            ),
          ),
        ),
      );
    yield* this.#mailbox.markRunning(input.deliveryId, turnId).pipe(toQueueOperationFailed);
    yield* queueStep(() => {
      // A side thread reads no mailbox copies, so the steered message joins it here, as its provider
      // history gives it after a restart (#5263).
      if (
        this.#conversation.isExecutionThread(snapshot.threadId) &&
        !snapshot.messages.some((message) => message.id === input.deliveryId)
      ) {
        snapshot.messages.push(promptMessage(context, { ...context.delivery, turnId }));
        this.#mailboxSync.syncDeliveryMessage(snapshot, input.deliveryId);
      }
      this.#mailboxSync.syncMailboxMessages(snapshot);
      this.#mailboxSync.emitQueue(agent.id);
      this.#conversation.emitConversation(snapshot, "queue.message-steered", { deliveryId: input.deliveryId });
    });
  }, Effect.uninterruptible);
}

/**
 * Claude Code and Grok Build tell their model that a message arrived while it worked, and to
 * continue the turn after it. Grok also reads only the first text block of a steer.
 */
const PROVIDERS_THAT_FRAME_STEERS: ReadonlySet<AgentProvider> = new Set(["claude", "grok"]);

/**
 * The note that goes in front of a steered message, or null when the provider writes its own or the
 * message is not from a person or a teammate. A routine run's prompt already gives its own task.
 * A slash command from the user gets none, as at the start of a turn: the agent reads a command
 * only at the start of the input.
 */
export function steerNoteFor(provider: AgentProvider, sender: QueueDelivery["sender"], text: string): string | null {
  if (PROVIDERS_THAT_FRAME_STEERS.has(provider)) return null;
  if (sender.kind === "user") return text.startsWith("/") ? null : steerNote("user");
  return sender.kind === "agent" ? steerNote("teammate") : null;
}

/** The message cannot steer the running turn, so it stays queued. A refusal, not a failure. */
export class SteerUnsupportedError extends Error {
  constructor(message = sourceText("error.backend.steerUnsupported")) {
    super(message);
  }
}

export class QueueOperationFailed extends Schema.TaggedError<QueueOperationFailed>()("QueueOperationFailed", {
  cause: Schema.Defect(),
}) {}

const { sync: queueStep, rewrap: toQueueOperationFailed } = causeHelpers(QueueOperationFailed);