/**
 * Background processing stage: orchestrates fire-and-forget message processing
 * after the synchronous HTTP response has been returned. Manages typing
 * indicators, approval prompt watchers, trusted contact notifications, and
 * the main agent loop invocation.
 *
 * Extracted from inbound-message-handler.ts to keep the top-level handler
 * focused on orchestration.
 */
import type { AssistantEvent } from "../../../api/index.js";
import { resolveGuardianPromptDelivery } from "../../../approvals/guardian-channel-delivery.js";
import {
  extractMessageTsFromCallbackUrl,
  extractThreadTsFromCallbackUrl,
  isSlackDeliveryCallbackUrl,
} from "../../../channels/slack-callback-url.js";
import type { ChannelId, InterfaceId } from "../../../channels/types.js";
import {
  getGuardianDelivery,
  guardianForChannel,
} from "../../../contacts/guardian-delivery-reader.js";
import { isConversationBusyError } from "../../../daemon/conversation-messaging.js";
import type { TrustContext } from "../../../daemon/trust-context-types.js";
import {
  sendChannelReaction,
  sendChannelTyping,
  setChannelThreadStatus,
  supportsChannelTyping,
} from "../../../messaging/providers/index.js";
import {
  getSiblingStreamedReplyTs,
  linkMessage,
  storeInboundSlackMetadata,
  storeReplyMessageId,
  storeStreamedReplyTs,
} from "../../../persistence/delivery-crud.js";
import {
  deferRetryUntilIdle,
  isDeduplicatedDeliveryOwnedBySibling,
  markProcessed,
  recordProcessingFailure,
} from "../../../persistence/delivery-status.js";
import { resolveGuardianName } from "../../../prompts/user-reference.js";
import { getLogger } from "../../../util/logger.js";
import { DAEMON_INTERNAL_ASSISTANT_ID } from "../../assistant-scope.js";
import {
  buildApprovalUIMetadata,
  getApprovalInfoByConversation,
  getChannelApprovalPrompt,
} from "../../channel-approvals.js";
import { deliverChannelReply } from "../../gateway-client.js";
import type {
  ApprovalCopyGenerator,
  MessageProcessor,
  SlackInboundMessageMetadata,
} from "../../http-types.js";
import { hasDeliverableAssistantText } from "../../no-response.js";
import { createSlackReplySession } from "../../slack-reply-session.js";
import type { TaskProgressData } from "../../slack-task-progress.js";
import {
  getTaskProgressDataFromSurfaceData,
  mergeTaskProgressData,
} from "../../slack-task-progress.js";
import { isContactTrustClass } from "../../trust-class.js";
import { resolveRoutingState } from "../../trust-context-resolver.js";
import { finalizeEventDelivery } from "../channel-delivery-routes.js";
import { deliverGeneratedApprovalPrompt } from "../guardian-approval-prompt.js";
import { withChannelTurnAdmission } from "./channel-turn-admission.js";

const log = getLogger("runtime-http");

export function isBoundGuardianActor(params: {
  trustClass: TrustContext["trustClass"];
  guardianExternalUserId?: string;
  requesterExternalUserId?: string;
}): boolean {
  const { trustClass, guardianExternalUserId, requesterExternalUserId } =
    params;

  return (
    trustClass === "guardian" &&
    !!guardianExternalUserId &&
    requesterExternalUserId === guardianExternalUserId
  );
}

// ---------------------------------------------------------------------------
// Public API
// ---------------------------------------------------------------------------

export interface BackgroundProcessingParams {
  processMessage: MessageProcessor;
  conversationId: string;
  eventId: string;
  content: string;
  displayContent?: string;
  attachmentIds?: string[];
  sourceChannel: ChannelId;
  sourceInterface: InterfaceId;
  externalChatId: string;
  trustCtx: TrustContext;
  metadataHints: string[];
  metadataUxBrief?: string;
  replyCallbackUrl?: string;
  assistantId?: string;
  approvalCopyGenerator?: ApprovalCopyGenerator;
  commandIntent?: Record<string, unknown>;
  sourceLanguageCode?: string;
  /** Chat type from the gateway (e.g. "private", "group", "supergroup"). */
  chatType?: string;
  /** IANA timezone reported by the active client for the current turn. */
  clientTimezone?: string;
  /** Slack app_mention/direct bot mention signal from the gateway. */
  slackBotMentioned?: boolean;
  /**
   * Slack-specific inbound metadata extracted at the HTTP boundary. Threaded
   * through to `persistUserMessage` so the row can be tagged with a
   * `slackMeta` envelope for the chronological renderer.
   */
  slackInbound?: SlackInboundMessageMetadata;
}

/**
 * Fire-and-forget: process the message and deliver the reply in the background.
 * The HTTP response returns immediately so the gateway webhook is not blocked.
 */
export function processChannelMessageInBackground(
  params: BackgroundProcessingParams,
): void {
  const {
    processMessage,
    conversationId,
    eventId,
    content,
    displayContent,
    attachmentIds,
    sourceChannel,
    sourceInterface,
    externalChatId,
    trustCtx,
    metadataHints,
    metadataUxBrief,
    replyCallbackUrl,
    assistantId,
    approvalCopyGenerator,
    commandIntent,
    sourceLanguageCode,
    chatType,
    clientTimezone,
    slackBotMentioned,
    slackInbound,
  } = params;

  // Capture the Slack ingress metadata onto the stored payload up front — before
  // the admission wait or any processing — so if the daemon dies mid-wait or
  // mid-turn, the retry sweep replays with the SAME `slackInbound` this turn
  // used. That keeps the derived idempotency key identical (the replay dedups
  // against a turn this attempt already persisted) and carries full slackMeta.
  if (slackInbound) {
    storeInboundSlackMetadata(eventId, slackInbound);
  }

  // Defer the whole turn + delivery until the conversation's processing lock is
  // free, serialized per conversation so same-conversation replies stay ordered.
  // A channel message routed to a busy conversation (e.g. a Slack
  // thread-participant reply arriving mid-session) is thereby processed when the
  // in-flight turn completes instead of being dropped. See
  // `channel-turn-admission.ts` for why channel turns defer rather than route
  // through the SSE-oriented conversation queue.
  void withChannelTurnAdmission(conversationId, async () => {
    const stopTypingHeartbeat =
      replyCallbackUrl && supportsChannelTyping(replyCallbackUrl)
        ? startTypingHeartbeat(replyCallbackUrl, externalChatId)
        : undefined;

    const slackThinkingStatus = createSlackThinkingStatusController({
      sourceChannel,
      replyCallbackUrl,
      chatId: externalChatId,
      startImmediately: shouldStartSlackThinkingStatusImmediately({
        sourceChannel,
        chatType,
        slackBotMentioned,
      }),
    });
    const stopApprovalWatcher = replyCallbackUrl
      ? startPendingApprovalPromptWatcher({
          conversationId,
          sourceChannel,
          externalChatId,
          trustClass: trustCtx.trustClass,
          guardianExternalUserId: trustCtx.guardianExternalUserId,
          guardianChatId: trustCtx.guardianChatId,
          requesterExternalUserId: trustCtx.requesterExternalUserId,
          replyCallbackUrl,
          assistantId,
          approvalCopyGenerator,
        })
      : undefined;
    const stopTcApprovalNotifier = replyCallbackUrl
      ? startTrustedContactApprovalNotifier({
          conversationId,
          sourceChannel,
          externalChatId,
          trustClass: trustCtx.trustClass,
          guardianExternalUserId: trustCtx.guardianExternalUserId,
          replyCallbackUrl,
          assistantId,
        })
      : undefined;

    try {
      const cmdIntent =
        commandIntent && typeof commandIntent.type === "string"
          ? {
              type: commandIntent.type as string,
              ...(typeof commandIntent.payload === "string"
                ? { payload: commandIntent.payload }
                : {}),
              ...(sourceLanguageCode
                ? { languageCode: sourceLanguageCode }
                : {}),
            }
          : undefined;
      let replyMessageId: string | undefined;
      const slackReplySession = createSlackReplySession({
        sourceChannel,
        chatType,
        replyCallbackUrl,
        chatId: externalChatId,
        recipientUserId: slackInbound?.actorExternalUserId,
        recipientTeamId: slackInbound?.actorTeamId,
        // Durably record the streamed message `ts` the instant the stream
        // opens, so a crash before `finalizeEventDelivery` leaves a breadcrumb
        // the redelivery path can reuse to edit the reply in place.
        onStreamOpen: (streamTs) => storeStreamedReplyTs(eventId, streamTs),
      });
      const observeAgentEvent = (msg: AssistantEvent): void => {
        if (
          msg.type === "message_complete" &&
          (msg.source === undefined || msg.source === "main") &&
          typeof msg.messageId === "string"
        ) {
          replyMessageId = msg.messageId;
        }
        slackReplySession?.observeEvent(msg);
        slackThinkingStatus?.observeEvent(msg);
      };

      let userMessageId: string | undefined;
      let deduplicatedIngress = false;
      try {
        const result = await processMessage(conversationId, content, {
          attachmentIds,
          transport: {
            channelId: sourceChannel,
            hints: metadataHints.length > 0 ? metadataHints : undefined,
            uxBrief: metadataUxBrief,
            chatType,
            ...(clientTimezone ? { clientTimezone } : {}),
          },
          assistantId,
          trustContext: trustCtx,
          isInteractive: resolveRoutingState(trustCtx).promptWaitingAllowed,
          ...(displayContent !== undefined ? { displayContent } : {}),
          ...(cmdIntent ? { commandIntent: cmdIntent } : {}),
          ...(slackInbound ? { slackInbound } : {}),
          onEvent: observeAgentEvent,
          sourceChannel,
          sourceInterface,
        });
        userMessageId = result.messageId;
        deduplicatedIngress = result.deduplicated === true;
        linkMessage(eventId, userMessageId);
        markProcessed(eventId);
        replyMessageId ??= result.assistantMessageId;
        if (replyMessageId) {
          storeReplyMessageId(eventId, replyMessageId);
        }
      } catch (err) {
        // Stop any live Slack stream cleanly. Its `ts` is already durably
        // recorded via `onStreamOpen`, so the retry sweep can reconcile
        // against that message rather than posting a duplicate.
        await slackReplySession?.finish();
        if (isConversationBusyError(err)) {
          // Admission observed the conversation idle, but a non-channel turn
          // (web / wake / voice) re-took the processing lock before this turn
          // could. Re-schedule for the retry sweep without burning a retry
          // attempt (`deferRetryUntilIdle`) so it reprocesses and delivers from
          // the stored payload once the lock frees — a plain processing-failure
          // record would classify the busy message as fatal and dead-letter it
          // (a silent drop), and even a retryable one could exhaust the budget
          // under sustained contention.
          log.info(
            { conversationId, eventId },
            "Channel turn lost the processing lock after admission; deferring to the retry sweep",
          );
          deferRetryUntilIdle(eventId);
          return;
        }
        log.error(
          { err, conversationId },
          "Background channel message processing failed",
        );
        recordProcessingFailure(eventId, err);
        return;
      }

      // An at-least-once redelivery that deduplicated against the original turn
      // must not blindly re-deliver: `finalizeEventDelivery` would re-emit the
      // reply via `sinceMessageId: userMessageId`. Consult the sibling events
      // linked to the same user message (they share `messageId` because the
      // deduped turn returns the original message id and this redelivery was
      // `linkMessage`d to it). `deliveryStatus` goes `pending` → `delivered` |
      // `failed` | `dead_letter`; only `pending` is non-terminal:
      //   - `delivered`            → reply already emitted → skip (would duplicate).
      //   - `failed`/`dead_letter` → a delivery attempt is recorded; the retry
      //     sweep (which selects `deliveryStatus='failed'`) or dead-letter replay
      //     owns recovery → skip to avoid racing it.
      //   - all siblings `pending` → the first process persisted the turn but
      //     died before recording a delivery outcome. The sweep never selects
      //     `pending`, so this redelivery is the only path that can recover the
      //     undelivered reply → fall through and deliver. If that first attempt
      //     had already streamed its reply live into Slack, its message `ts` is
      //     durably recorded on the sibling row (via `onStreamOpen`); reuse it so
      //     recovery edits that visible message in place instead of posting the
      //     persisted reply a second time.
      let priorDeduplicatedDeliveryOwned = false;
      let recoveredStreamMessageTs: string | undefined;
      if (deduplicatedIngress && userMessageId !== undefined) {
        if (isDeduplicatedDeliveryOwnedBySibling(userMessageId, eventId)) {
          priorDeduplicatedDeliveryOwned = true;
        } else {
          recoveredStreamMessageTs = getSiblingStreamedReplyTs(
            userMessageId,
            eventId,
          );
        }
      }

      if (priorDeduplicatedDeliveryOwned) {
        log.info(
          { conversationId, eventId },
          "Skipping channel reply delivery for deduplicated ingress event; a prior attempt owns delivery",
        );
      } else if (replyCallbackUrl) {
        try {
          await finalizeEventDelivery({
            eventId,
            conversationId,
            externalChatId,
            replyCallbackUrl,
            assistantId,
            replyMessageId,
            userMessageId,
            slackReplySession,
            ...(recoveredStreamMessageTs
              ? { priorStreamMessageTs: recoveredStreamMessageTs }
              : {}),
          });
        } catch (err) {
          log.error(
            { err, conversationId },
            "Background channel reply delivery failed",
          );
        }
      }
    } finally {
      stopTypingHeartbeat?.();
      slackThinkingStatus?.stop();
      stopApprovalWatcher?.();
      stopTcApprovalNotifier?.();
    }
  }).catch((err) => {
    log.error(
      { err, conversationId, eventId },
      "Channel turn admission failed unexpectedly",
    );
  });
}

// ---------------------------------------------------------------------------
// Telegram typing heartbeat
// ---------------------------------------------------------------------------

/**
 * How often the working indicator is re-sent while a turn runs.
 *
 * Every channel that has one expires it after a few seconds and offers no
 * start/stop pair to hold it open, so showing it across a multi-second turn
 * means re-sending on a timer. The interval has to stay under the shortest
 * expiry of any channel that implements `typing`: Telegram is about five
 * seconds, Discord about ten. Anyone changing it should check the current
 * windows first.
 *
 * No explicit stop is needed: both clear the indicator when the bot posts.
 *
 * https://core.telegram.org/bots/api#sendchataction
 * https://discord.com/developers/docs/resources/channel#trigger-typing-indicator
 */
const TYPING_INTERVAL_MS = 4_000;

function startTypingHeartbeat(callbackUrl: string, chatId: string): () => void {
  let active = true;
  let inFlight = false;

  const emitTyping = (): void => {
    if (!active || inFlight) {
      return;
    }
    inFlight = true;
    void sendChannelTyping(callbackUrl, chatId)
      .catch((err) => {
        log.debug(
          { err, chatId },
          "Failed to deliver Telegram typing indicator",
        );
      })
      .finally(() => {
        inFlight = false;
      });
  };

  emitTyping();

  const interval = setInterval(emitTyping, TYPING_INTERVAL_MS);
  (interval as { unref?: () => void }).unref?.();

  return () => {
    active = false;
    clearInterval(interval);
  };
}

// ---------------------------------------------------------------------------
// Slack Assistants API thinking status indicator
// ---------------------------------------------------------------------------

type SlackThinkingStatusController = {
  observeEvent: (msg: AssistantEvent) => void;
  stop: () => void;
};

type SlackThinkingStatusHandle = {
  updateLoadingMessages: (loadingMessages?: string[]) => void;
  clear: () => void;
};

export function shouldStartSlackThinkingStatusForText(text: string): boolean {
  return hasDeliverableAssistantText(text);
}

function shouldEmitSlackThinkingStatus(
  sourceChannel: ChannelId,
  replyCallbackUrl?: string,
): boolean {
  return (
    sourceChannel === "slack" && isSlackDeliveryCallbackUrl(replyCallbackUrl)
  );
}

export function shouldStartSlackThinkingStatusImmediately(params: {
  sourceChannel: ChannelId;
  chatType?: string;
  slackBotMentioned?: boolean;
}): boolean {
  if (params.sourceChannel !== "slack") {
    return false;
  }
  return params.chatType === "im" || params.slackBotMentioned === true;
}

function createSlackThinkingStatusController(params: {
  sourceChannel: ChannelId;
  replyCallbackUrl?: string;
  chatId: string;
  startImmediately?: boolean;
}): SlackThinkingStatusController | undefined {
  const { sourceChannel, replyCallbackUrl, chatId, startImmediately } = params;
  if (
    !replyCallbackUrl ||
    !shouldEmitSlackThinkingStatus(sourceChannel, replyCallbackUrl)
  ) {
    return undefined;
  }
  const callbackUrl = replyCallbackUrl;

  let stopped = false;
  let slackThinkingStatus: SlackThinkingStatusHandle | undefined;
  let observedAssistantText = "";
  let currentLoadingMessages: string[] | undefined = startImmediately
    ? [...SLACK_GENERIC_LOADING_MESSAGES]
    : undefined;
  let lastSentLoadingMessageKey: string | undefined;
  const taskProgressBySurfaceId = new Map<string, TaskProgressData>();

  const start = (): void => {
    if (stopped || slackThinkingStatus) {
      return;
    }
    slackThinkingStatus = setSlackThinkingStatus(
      callbackUrl,
      chatId,
      currentLoadingMessages,
    );
    lastSentLoadingMessageKey = getLoadingMessagesKey(currentLoadingMessages);
  };

  const maybeUpdateLoadingMessages = (): void => {
    const nextLoadingMessageKey = getLoadingMessagesKey(currentLoadingMessages);
    if (nextLoadingMessageKey === lastSentLoadingMessageKey) {
      return;
    }
    lastSentLoadingMessageKey = nextLoadingMessageKey;
    slackThinkingStatus?.updateLoadingMessages(currentLoadingMessages);
  };

  const observeTaskProgress = (msg: AssistantEvent): void => {
    if (msg.type === "ui_surface_show") {
      const progress = getTaskProgressDataFromSurfaceData(msg.data);
      if (!progress) {
        return;
      }
      taskProgressBySurfaceId.set(msg.surfaceId, progress);
    } else if (msg.type === "ui_surface_update") {
      const existing = taskProgressBySurfaceId.get(msg.surfaceId);
      const progress = mergeTaskProgressData(existing, msg.data);
      if (!progress) {
        return;
      }
      taskProgressBySurfaceId.set(msg.surfaceId, progress);
    } else {
      return;
    }

    currentLoadingMessages =
      getTaskProgressLoadingMessage(
        taskProgressBySurfaceId.get(msg.surfaceId),
      ) ?? [];
    maybeUpdateLoadingMessages();
  };

  if (startImmediately) {
    start();
  }

  return {
    observeEvent(msg) {
      if (stopped) {
        return;
      }

      if (msg.type === "ui_surface_show" || msg.type === "ui_surface_update") {
        observeTaskProgress(msg);
        return;
      }

      if (slackThinkingStatus || msg.type !== "assistant_text_delta") {
        return;
      }

      observedAssistantText += msg.text;
      if (shouldStartSlackThinkingStatusForText(observedAssistantText)) {
        start();
      }
    },
    stop() {
      stopped = true;
      slackThinkingStatus?.clear();
    },
  };
}

const SLACK_THINKING_MAX_DURATION_MS = 120_000;
const SLACK_GENERIC_LOADING_MESSAGES = ["Thinking…"] as const;
const SLACK_THINKING_STATUSES = ["is on it", "is working hard"] as const;

function getRandomSlackThinkingStatus(): string {
  return SLACK_THINKING_STATUSES[
    Math.floor(Math.random() * SLACK_THINKING_STATUSES.length)
  ]!;
}

function getLoadingMessagesKey(loadingMessages?: string[]): string | undefined {
  return loadingMessages?.join("\n");
}

function getTaskProgressLoadingMessage(
  progress: TaskProgressData | undefined,
): string[] | undefined {
  if (!progress) {
    return undefined;
  }

  const activeStepIndex = progress.steps.findIndex(
    (step) => step.status === "in_progress",
  );
  if (activeStepIndex < 0) {
    return undefined;
  }

  const activeStep = progress.steps[activeStepIndex]!;
  return [
    `In progress (${activeStepIndex + 1}/${progress.steps.length}): ${
      activeStep.label
    }`,
  ];
}

/**
 * Set Slack Assistants API status on the thread and return a handle for
 * updating loading messages or clearing the indicator.
 *
 * A safety timer auto-clears the status after {@link SLACK_THINKING_MAX_DURATION_MS}
 * to prevent a stuck indicator when `processMessage` hangs.
 */
function setSlackThinkingStatus(
  callbackUrl: string,
  chatId: string,
  loadingMessages?: string[],
): SlackThinkingStatusHandle {
  let cleared = false;

  // Extract the thread timestamp from the callback URL so we can target
  // the correct thread for the Assistants API status.
  const threadTs = extractThreadTsFromCallbackUrl(callbackUrl);

  // For non-threaded DMs, fall back to emoji reaction on the original message.
  if (!threadTs) {
    const messageTs = extractMessageTsFromCallbackUrl(callbackUrl);
    if (!messageTs) {
      return {
        updateLoadingMessages: () => {},
        clear: () => {},
      };
    }

    const addPromise = sendChannelReaction(callbackUrl, {
      chatId,
      messageId: messageTs,
      emoji: "eyes",
      action: "add",
    }).catch((err) => {
      log.debug({ err, chatId, messageTs }, "Failed to add eyes reaction");
    });

    const clearReaction = (): void => {
      if (cleared) {
        return;
      }
      cleared = true;
      clearTimeout(safetyTimer);
      void addPromise.then(() =>
        sendChannelReaction(callbackUrl, {
          chatId,
          messageId: messageTs,
          emoji: "eyes",
          action: "remove",
        }).catch((err) => {
          log.debug(
            { err, chatId, messageTs },
            "Failed to remove eyes reaction",
          );
        }),
      );
    };

    const safetyTimer = setTimeout(
      clearReaction,
      SLACK_THINKING_MAX_DURATION_MS,
    );
    (safetyTimer as { unref?: () => void }).unref?.();

    return {
      updateLoadingMessages: () => {},
      clear: clearReaction,
    };
  }

  // Track the set promise so clear waits for it to settle first,
  // preventing a race where clear arrives at Slack before set.
  let statusPromise = setChannelThreadStatus(callbackUrl, {
    chatId,
    threadTs,
    status: getRandomSlackThinkingStatus(),
    ...(loadingMessages ? { loadingMessages } : {}),
  }).catch((err) => {
    log.debug({ err, chatId, threadTs }, "Failed to set Slack thinking status");
  });

  const updateLoadingMessages = (nextLoadingMessages?: string[]): void => {
    if (cleared) {
      return;
    }
    statusPromise = statusPromise.then(() =>
      setChannelThreadStatus(callbackUrl, {
        chatId,
        threadTs,
        status: getRandomSlackThinkingStatus(),
        ...(nextLoadingMessages
          ? { loadingMessages: nextLoadingMessages }
          : {}),
      }).catch((err) => {
        log.debug(
          { err, chatId, threadTs },
          "Failed to update Slack thinking status",
        );
      }),
    );
  };

  const clearStatus = (): void => {
    if (cleared) {
      return;
    }
    cleared = true;
    clearTimeout(safetyTimer);
    void statusPromise.then(() =>
      setChannelThreadStatus(callbackUrl, {
        chatId,
        threadTs,
        status: "",
      }).catch((err) => {
        log.debug(
          { err, chatId, threadTs },
          "Failed to clear Slack thinking status",
        );
      }),
    );
  };

  const safetyTimer = setTimeout(clearStatus, SLACK_THINKING_MAX_DURATION_MS);
  (safetyTimer as { unref?: () => void }).unref?.();

  return {
    updateLoadingMessages,
    clear: clearStatus,
  };
}

// ---------------------------------------------------------------------------
// Pending approval prompt watcher
// ---------------------------------------------------------------------------

const PENDING_APPROVAL_POLL_INTERVAL_MS = 300;

function startPendingApprovalPromptWatcher(params: {
  conversationId: string;
  sourceChannel: ChannelId;
  externalChatId: string;
  trustClass: TrustContext["trustClass"];
  guardianExternalUserId?: string;
  guardianChatId?: string;
  requesterExternalUserId?: string;
  replyCallbackUrl: string;
  assistantId?: string;
  approvalCopyGenerator?: ApprovalCopyGenerator;
}): () => void {
  const {
    conversationId,
    sourceChannel,
    externalChatId,
    trustClass,
    guardianExternalUserId,
    guardianChatId,
    requesterExternalUserId,
    replyCallbackUrl,
    assistantId,
    approvalCopyGenerator,
  } = params;

  // Approval prompt delivery is guardian-only. Non-guardian and unverified
  // actors must never receive approval prompt broadcasts for the conversation.
  // We also require an explicit identity match against the bound guardian to
  // avoid broadcasting prompts when trustClass is stale/mis-scoped.
  if (
    !isBoundGuardianActor({
      trustClass,
      guardianExternalUserId,
      requesterExternalUserId,
    })
  ) {
    return () => {};
  }

  let active = true;
  const deliveredRequestIds = new Set<string>();

  const poll = async (): Promise<void> => {
    while (active) {
      try {
        const prompt = getChannelApprovalPrompt(conversationId);
        const pending = getApprovalInfoByConversation(conversationId);
        const info = pending[0];
        if (prompt && info && !deliveredRequestIds.has(info.requestId)) {
          deliveredRequestIds.add(info.requestId);
          // Addressed to the guardian's own chat, not the chat the turn is
          // running in, which can be a room that reads the tool and its
          // command preview.
          const promptDelivery = resolveGuardianPromptDelivery({
            turnChatId: externalChatId,
            turnCallbackUrl: replyCallbackUrl,
            guardianChatId,
          });
          const delivered = await deliverGeneratedApprovalPrompt({
            replyCallbackUrl: promptDelivery.callbackUrl,
            chatId: promptDelivery.chatId,
            sourceChannel,
            assistantId: assistantId ?? DAEMON_INTERNAL_ASSISTANT_ID,
            prompt,
            uiMetadata: buildApprovalUIMetadata(prompt, info),
            messageContext: {
              scenario: "standard_prompt",
              toolName: info.toolName,
              channel: sourceChannel,
            },
            approvalCopyGenerator,
          });
          if (!delivered) {
            // Delivery can fail transiently (network or gateway outage).
            // Keep polling and retry prompt delivery for the same request.
            deliveredRequestIds.delete(info.requestId);
          }
        }
      } catch (err) {
        log.warn(
          { err, conversationId },
          "Pending approval prompt watcher failed",
        );
      }
      await delay(PENDING_APPROVAL_POLL_INTERVAL_MS);
    }
  };

  void poll();
  return () => {
    active = false;
  };
}

// ---------------------------------------------------------------------------
// Trusted contact approval notifier
// ---------------------------------------------------------------------------

// Module-level map tracking which approval requestIds have already been
// notified to trusted contacts. Maps requestId -> conversationId so that
// cleanup can be scoped to the owning conversation's poller, preventing
// concurrent pollers from different conversations from evicting each
// other's entries.
const globalNotifiedApprovalRequestIds = new Map<string, string>();

/**
 * Start a poller that sends a one-shot "waiting for guardian approval" message
 * to the trusted/unverified contact when a confirmation_request enters guardian
 * approval wait. Deduplicates by requestId so each request only produces one
 * message.
 *
 * Only activates for trusted_contact and unverified_contact actors with a
 * resolvable guardian route.
 */
function startTrustedContactApprovalNotifier(params: {
  conversationId: string;
  sourceChannel: ChannelId;
  externalChatId: string;
  trustClass: TrustContext["trustClass"];
  guardianExternalUserId?: string;
  replyCallbackUrl: string;
  assistantId?: string;
}): () => void {
  const {
    conversationId,
    sourceChannel,
    externalChatId,
    trustClass,
    guardianExternalUserId,
    replyCallbackUrl,
    assistantId,
  } = params;

  // Only notify identity-known non-guardian contacts (trusted_contact and
  // unverified_contact) who have a resolvable guardian route.
  if (!isContactTrustClass(trustClass) || !guardianExternalUserId) {
    return () => {};
  }

  let active = true;

  const poll = async (): Promise<void> => {
    while (active) {
      try {
        const pending = getApprovalInfoByConversation(conversationId);
        const info = pending[0];

        // Clean up resolved requests from the module-level dedupe map.
        // Only remove entries that belong to THIS conversation — other
        // conversations' pollers own their own entries. Without this
        // scoping, concurrent pollers would evict each other's request
        // IDs and cause duplicate notifications.
        const currentPendingIds = new Set(pending.map((p) => p.requestId));
        for (const [rid, cid] of globalNotifiedApprovalRequestIds) {
          if (cid === conversationId && !currentPendingIds.has(rid)) {
            globalNotifiedApprovalRequestIds.delete(rid);
          }
        }

        if (info && !globalNotifiedApprovalRequestIds.has(info.requestId)) {
          globalNotifiedApprovalRequestIds.set(info.requestId, conversationId);
          // Gateway-resolved guardian display name (display-only).
          const guardians = await getGuardianDelivery({
            channelTypes: [sourceChannel],
          });
          const displayName = guardians
            ? (guardianForChannel(guardians, sourceChannel)?.displayName ??
              undefined)
            : undefined;
          const guardianName = resolveGuardianName(displayName);
          const waitingText = `Waiting for ${guardianName}'s approval...`;
          try {
            await deliverChannelReply(replyCallbackUrl, {
              chatId: externalChatId,
              text: waitingText,
              assistantId: assistantId ?? DAEMON_INTERNAL_ASSISTANT_ID,
            });
          } catch (err) {
            log.warn(
              { err, conversationId },
              "Failed to deliver trusted-contact pending-approval notification",
            );
            // Remove from notified set so delivery is retried on next poll
            globalNotifiedApprovalRequestIds.delete(info.requestId);
          }
        }
      } catch (err) {
        log.warn(
          { err, conversationId },
          "Trusted-contact approval notifier poll failed",
        );
      }
      await delay(PENDING_APPROVAL_POLL_INTERVAL_MS);
    }
  };

  void poll();
  return () => {
    active = false;

    // Evict all dedupe entries owned by this conversation so the
    // module-level map doesn't grow unboundedly after the poller stops.
    for (const [rid, cid] of globalNotifiedApprovalRequestIds) {
      if (cid === conversationId) {
        globalNotifiedApprovalRequestIds.delete(rid);
      }
    }
  };
}

// ---------------------------------------------------------------------------
// Utility
// ---------------------------------------------------------------------------

function delay(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}