/**
 * Extracted free-function form of DaemonServer.processMessage and its helpers.
 *
 * Route handlers import {@link processMessage} directly instead of receiving
 * it through DI. The DaemonServer methods delegate here.
 */

import { v7 as uuidv7 } from "uuid";

import { enrichMessageWithSourcePaths } from "../agent/attachments.js";
import {
  createAssistantMessage,
  createUserMessage,
} from "../agent/message-types.js";
import type { AssistantEvent } from "../api/index.js";
import {
  type ChannelId,
  type InterfaceId,
  parseChannelId,
  parseInterfaceId,
} from "../channels/types.js";
import {
  getAttachmentsByIds,
  getSourcePathsForAttachments,
} from "../persistence/attachments-store.js";
import {
  addMessage,
  getConversation,
  provenanceFromTrustContext,
  setConversationOriginChannelIfUnset,
  setConversationOriginInterfaceIfUnset,
} from "../persistence/conversation-crud.js";
import { updateMetaFile } from "../persistence/conversation-disk-view.js";
import { broadcastMessage } from "../runtime/assistant-event-hub.js";
import { DAEMON_INTERNAL_ASSISTANT_ID } from "../runtime/assistant-scope.js";
import { publishConversationMessagesChanged } from "../runtime/sync/resource-sync-events.js";
import {
  readTurnFailure,
  type TurnFailure,
} from "../telemetry/turn-outcome.js";
import { getLogger } from "../util/logger.js";
import type { Conversation } from "./conversation.js";
import {
  buildSlackMetaForPersistence,
  CONVERSATION_BUSY_MESSAGE,
  serializePersistedUserMessageContent,
} from "./conversation-messaging.js";
import {
  formatCleanResult,
  formatCompactResult,
} from "./conversation-process.js";
import { resolveChannelCapabilities } from "./conversation-runtime-assembly.js";
import {
  buildSlashContextForContent,
  resolveSlash,
} from "./conversation-slash.js";
import {
  getOrCreateConversation as getOrCreateActiveConversation,
  mergeConversationOptions,
} from "./conversation-store.js";
import { getDbMigrationReadiness } from "./daemon-readiness.js";
import type {
  ConversationCreateOptions,
  SlackInboundMessageMetadata,
} from "./handlers/shared.js";
import { HostAppControlProxy } from "./host-app-control-proxy.js";
import { HostCuProxy } from "./host-cu-proxy.js";
import {
  preactivateHostProxySkills,
  shouldAttachHostProxyForCapability,
} from "./host-proxy-preactivation.js";
import type { SubagentToolGateMode } from "./tool-setup-types.js";
import { restingTrust } from "./trust-context-types.js";

const log = getLogger("process-message");

type ProcessMessageOptions = ConversationCreateOptions & {
  /** Per-turn observer for live agent-loop events. Does not replace SSE broadcast. */
  onEvent?: (msg: AssistantEvent) => void;
  /** IDs of user-uploaded attachments to resolve and include in the turn. */
  attachmentIds?: string[];
  /** Originating channel (e.g. "slack", "telegram"). Defaults to "vellum". */
  sourceChannel?: string;
  /** Originating interface (e.g. "cli", "web"). Defaults to "web". */
  sourceInterface?: string;
  /**
   * Origin tag of the turn (the conversation's `TitleOrigin`, e.g.
   * "memory_consolidation"), threaded from `runBackgroundJob`. Propagated
   * into the agent loop and tool context so the permission checker can scope
   * narrow non-interactive auto-grants to a specific internal background
   * origin. Unset for normal user-initiated turns.
   */
  requestOrigin?: string;
  /**
   * Firing's `cron_runs.id`, threaded through to the turn's usage rows so a
   * scheduled execute turn attributes its LLM spend to that firing. Per-turn:
   * not stored on the long-lived conversation.
   */
  cronRunId?: string | null;
  /**
   * Restrict this turn's tool surface to the given allowlist. Threaded from
   * `runBackgroundJob` so an unattended guardian-trust job (e.g. memory
   * consolidation) cannot reach tools it does not need. Applied to the live
   * conversation for the turn and restored afterward. Omitted = the full
   * conversation tool surface (unchanged for every normal caller).
   */
  allowedTools?: readonly string[];
  /**
   * How {@link allowedTools} is enforced (default `"wire"` — excluded tools
   * are not presented to the model). Ignored when `allowedTools` is absent.
   */
  toolGateMode?: SubagentToolGateMode;
  /**
   * Persist the inbound user message without indexing it — no memory
   * segments, no embeddings, no lexical-index entry. Threaded from
   * `runBackgroundJob` for machine-authored kickoff prompts (e.g. memory
   * consolidation) whose static instruction text must not enter memory or
   * search. Every other message in the turn indexes normally.
   */
  skipUserMessageIndexing?: boolean;
  /**
   * Stable idempotency key for this ingress turn. Threaded to
   * `persistUserMessage` as `clientMessageId` so an at-least-once redelivery of
   * the same logical turn dedups on the `client_message_id` partial unique index
   * instead of running the agent loop a second time. When omitted, a key is
   * synthesized from the transport's native per-message id where one exists (see
   * {@link deriveIngressIdempotencyKey}).
   */
  clientMessageId?: string;
};

/**
 * Derive a stable idempotency key for a server-side ingress turn so an
 * at-least-once redelivery (gateway/channel retry) dedups on the
 * `client_message_id` partial unique index rather than running a second turn.
 * An explicit caller key wins; otherwise it is synthesized from the transport's
 * native per-message id, namespaced so it can never collide with a web-client
 * nonce. Returns undefined when no stable id exists (behavior unchanged).
 */
export function deriveIngressIdempotencyKey(options?: {
  clientMessageId?: string;
  slackInbound?: SlackInboundMessageMetadata;
}): string | undefined {
  if (options?.clientMessageId) {
    return options.clientMessageId;
  }
  const slack = options?.slackInbound;
  if (slack) {
    return `slack:${slack.channelId}:${slack.channelTs}`;
  }
  return undefined;
}

function buildEventEmitter(
  observer?: (msg: AssistantEvent) => void,
): (msg: AssistantEvent) => void {
  if (!observer) {
    return broadcastMessage;
  }
  return (msg) => {
    broadcastMessage(msg);
    try {
      observer(msg);
    } catch (err) {
      log.warn({ err, messageType: msg.type }, "Agent event observer failed");
    }
  };
}

/**
 * Scope the live conversation's tool surface to `options.allowedTools` for the
 * duration of a turn, returning a closure that restores the prior scope. A
 * no-op (returns an empty restore) when no allowlist is supplied, so ordinary
 * turns are byte-for-byte unaffected.
 *
 * The `runBackgroundJob` → `processMessage` path does not run through the wake
 * machinery, so it cannot use {@link scopeWakeAllowedTools}; this mirrors that
 * helper's set-and-restore over the same two conversation fields. Default
 * `"wire"` gate mode filters the excluded tools out of the definitions sent to
 * the provider, so a scoped turn (e.g. unattended memory consolidation) can
 * never call a tool outside the allowlist — no reliance on the permission
 * checker's background auto-approve threshold.
 */
function applyTurnToolAllowlist(
  conversation: Conversation,
  options?: ProcessMessageOptions,
): () => void {
  if (!options?.allowedTools) {
    return () => {};
  }
  const previousTools = conversation.subagentAllowedTools;
  const previousGateMode = conversation.subagentToolGateMode;
  conversation.setSubagentAllowedTools(new Set(options.allowedTools));
  conversation.subagentToolGateMode = options.toolGateMode ?? "wire";
  return () => {
    conversation.setSubagentAllowedTools(previousTools);
    conversation.subagentToolGateMode = previousGateMode;
  };
}

// ---------------------------------------------------------------------------
// Turn-context resolution helpers
// ---------------------------------------------------------------------------

export function resolveTurnChannel(
  sourceChannel?: string,
  transportChannelId?: string,
): ChannelId {
  if (sourceChannel != null) {
    const parsed = parseChannelId(sourceChannel);
    if (!parsed) {
      throw new Error(`Invalid sourceChannel: ${sourceChannel}`);
    }
    return parsed;
  }
  if (transportChannelId != null) {
    const parsed = parseChannelId(transportChannelId);
    if (!parsed) {
      throw new Error(`Invalid transport.channelId: ${transportChannelId}`);
    }
    return parsed;
  }
  return "vellum";
}

export function resolveTurnInterface(sourceInterface?: string): InterfaceId {
  if (sourceInterface != null) {
    const parsed = parseInterfaceId(sourceInterface);
    if (!parsed) {
      throw new Error(`Invalid sourceInterface: ${sourceInterface}`);
    }
    return parsed;
  }
  return "web";
}

// ---------------------------------------------------------------------------
// prepareConversationForMessage
// ---------------------------------------------------------------------------

async function prepareConversationForMessage(
  conversationId: string,
  content: string,
  options?: ProcessMessageOptions,
): Promise<{
  conversation: Conversation;
  attachments: {
    id: string;
    filename: string;
    mimeType: string;
    data: string;
    filePath?: string;
  }[];
}> {
  const {
    attachmentIds,
    sourceChannel,
    sourceInterface,
    onEvent: _onEvent,
    cronRunId: _cronRunId,
    requestOrigin: _requestOrigin,
    allowedTools: _allowedTools,
    toolGateMode: _toolGateMode,
    ...conversationOptions
  } = options ?? {};
  const conversation = await getOrCreateActiveConversation(
    conversationId,
    Object.keys(conversationOptions).length > 0
      ? conversationOptions
      : undefined,
  );

  if (conversation.isProcessing()) {
    throw new Error(CONVERSATION_BUSY_MESSAGE);
  }

  const resolvedChannel = resolveTurnChannel(
    sourceChannel,
    conversationOptions.transport?.channelId,
  );
  const resolvedInterface = resolveTurnInterface(sourceInterface);
  conversation.setAssistantId(
    options?.assistantId ?? DAEMON_INTERNAL_ASSISTANT_ID,
  );
  conversation.taskRunId = options?.taskRunId;
  if (options?.trustContext !== undefined) {
    conversation.setTrustContext(options.trustContext);
  }
  if (options?.authContext !== undefined) {
    conversation.setAuthContext(options.authContext);
  }
  await conversation.ensureActorScopedHistory();

  mergeConversationOptions(conversationId, {
    trustContext: conversation.trustContext,
    authContext: conversation.authContext,
  });
  conversation.setChannelCapabilities(
    resolveChannelCapabilities(
      sourceChannel,
      sourceInterface,
      options?.transport?.chatType,
    ),
  );
  if (resolvedInterface === "chrome-extension") {
    throw new Error(
      "prepareConversationForMessage does not yet support chrome-extension transport — " +
        "use the conversation-routes.ts /v1/messages flow which routes host_browser through " +
        "the assistant event hub. If you need chrome-extension here, factor out the " +
        "wiring in conversation-routes.ts into a shared helper.",
    );
  }
  const sourceActorPrincipalId = conversation.authContext?.actorPrincipalId;
  // CU is per-conversation (owns step count, AX tree history, loop detection).
  if (
    shouldAttachHostProxyForCapability(
      "host_cu",
      resolvedInterface,
      sourceActorPrincipalId,
    )
  ) {
    if (!conversation.isProcessing() || !conversation.hostCuProxy) {
      conversation.setHostCuProxy(new HostCuProxy());
    }
  } else if (!conversation.isProcessing()) {
    conversation.setHostCuProxy(undefined);
  }
  // App-control mirrors CU's per-conversation lifecycle. The proxy attaches
  // unconditionally when the capability is reachable — feature-flag gating
  // is enforced by the skill-projection layer via SKILL.md frontmatter, so
  // an attached proxy is harmless when the flag is off.
  if (
    shouldAttachHostProxyForCapability(
      "host_app_control",
      resolvedInterface,
      sourceActorPrincipalId,
    )
  ) {
    if (!conversation.isProcessing() || !conversation.hostAppControlProxy) {
      conversation.setHostAppControlProxy(
        new HostAppControlProxy(conversationId),
      );
    }
  } else if (!conversation.isProcessing()) {
    conversation.setHostAppControlProxy(undefined);
  }
  // The early `isProcessing()` throw above guarantees the conversation is
  // idle here, so preactivation is unconditional once the proxies are wired.
  preactivateHostProxySkills(
    conversation,
    resolvedInterface,
    sourceActorPrincipalId,
  );
  conversation.setCommandIntent(options?.commandIntent ?? null);
  conversation.setTurnChannelContext({
    userMessageChannel: resolvedChannel,
    assistantMessageChannel: resolvedChannel,
  });
  conversation.setTurnInterfaceContext({
    userMessageInterface: resolvedInterface,
    assistantMessageInterface: resolvedInterface,
  });

  const attachments = attachmentIds
    ? (() => {
        const resolved = getAttachmentsByIds(attachmentIds, {
          hydrateFileData: true,
        });
        const sourcePaths = getSourcePathsForAttachments(attachmentIds);
        return resolved.map((a) => ({
          id: a.id,
          filename: a.originalFilename,
          mimeType: a.mimeType,
          data: a.dataBase64,
          ...(sourcePaths.has(a.id) ? { filePath: sourcePaths.get(a.id) } : {}),
        }));
      })()
    : [];

  return { conversation, attachments };
}

// ---------------------------------------------------------------------------
// processMessage — main entry point for channel inbound + daemon callers
// ---------------------------------------------------------------------------

/**
 * Defense-in-depth: never run a turn against a not-yet-migrated schema.
 * The HTTP send path is already gated by dbMigrationUnavailableForPath, the
 * IPC transport by dbMigrationGateResponse, and the background sweeps are
 * deferred until migrations settle — but {@link processMessage} and
 * {@link processMessageInBackground} are the two sinks every message caller
 * funnels through (sweeps, scheduler, pointer, signal handlers, the
 * conversation launcher), and the signal-file transport reaches them without
 * passing either transport gate. Migrations run async during startup, so a
 * caller can get here before the tables exist.
 */
function assertDbMigrationsReadyForTurn(): void {
  const migrationReadiness = getDbMigrationReadiness();
  if (!migrationReadiness.ready) {
    throw new Error(
      `Cannot process message: database migrations ${migrationReadiness.state}`,
    );
  }
}

export async function processMessage(
  conversationId: string,
  content: string,
  options?: ProcessMessageOptions,
): Promise<{
  messageId: string;
  assistantMessageId?: string;
  /**
   * True when this ingress deduplicated against a prior turn (same idempotency
   * key) and the agent loop was skipped. Channel finalization must skip reply
   * delivery on this signal so an at-least-once redelivery does not re-emit the
   * original turn's reply.
   */
  deduplicated?: boolean;
  /**
   * The agent turn's failure outcome, or `null` when it replied normally. Set
   * when the turn failed (e.g. its LLM call failed with an invalid provider) —
   * that path persists a synthetic error message and returns normally rather
   * than throwing, so this is the only way an awaiting caller can tell. Omitted
   * by the slash-command branches, which never run the agent loop.
   */
  turnFailure?: TurnFailure | null;
}> {
  assertDbMigrationsReadyForTurn();

  const { conversation, attachments } = await prepareConversationForMessage(
    conversationId,
    content,
    options,
  );
  const emitEvent = buildEventEmitter(options?.onEvent);

  const serverInterfaceCtx = conversation.getTurnInterfaceContext();
  const slashContext = buildSlashContextForContent(content, {
    conversationId,
    messageCount: conversation.getMessages().length,
    inputTokens: conversation.usageStats.inputTokens,
    outputTokens: conversation.usageStats.outputTokens,
    estimatedCost: conversation.usageStats.estimatedCost,
    userMessageInterface: serverInterfaceCtx?.userMessageInterface,
  });
  const slashResult = await resolveSlash(content, slashContext);

  const slackMeta = buildSlackMetaForPersistence({
    slackInbound: options?.slackInbound,
    turnChannel: conversation.getTurnChannelContext()?.userMessageChannel,
  });

  if (slashResult.kind === "unknown") {
    const serverTurnCtx = conversation.getTurnChannelContext();
    const serverProvenance = provenanceFromTrustContext(
      // Ingress persists before any per-turn stamp; the slot was just written
      // by this message's own resolution, and the per-turn field may still
      // hold the previous turn's actor.
      restingTrust(conversation),
    );
    const imageSourcePaths: Record<string, string> = {};
    for (let i = 0; i < attachments.length; i++) {
      const a = attachments[i];
      if (a.filePath && a.mimeType.toLowerCase().startsWith("image/")) {
        imageSourcePaths[`${i}:${a.filename}`] = a.filePath;
      }
    }
    const serverChannelMeta = {
      ...serverProvenance,
      ...(serverTurnCtx
        ? {
            userMessageChannel: serverTurnCtx.userMessageChannel,
            assistantMessageChannel: serverTurnCtx.assistantMessageChannel,
          }
        : {}),
      ...(serverInterfaceCtx
        ? {
            userMessageInterface: serverInterfaceCtx.userMessageInterface,
            assistantMessageInterface:
              serverInterfaceCtx.assistantMessageInterface,
          }
        : {}),
      ...(Object.keys(imageSourcePaths).length > 0 ? { imageSourcePaths } : {}),
    };
    const userMetaWithSlack = slackMeta
      ? { ...serverChannelMeta, slackMeta }
      : serverChannelMeta;
    const cleanMsg = await createUserMessage(content, attachments);
    const llmMsg = enrichMessageWithSourcePaths(cleanMsg, attachments);
    const persisted = await addMessage(
      conversationId,
      "user",
      await serializePersistedUserMessageContent(
        content,
        options?.displayContent,
        attachments,
      ),
      { metadata: userMetaWithSlack },
    );
    conversation.getMessages().push(llmMsg);

    if (serverTurnCtx) {
      try {
        setConversationOriginChannelIfUnset(
          conversationId,
          serverTurnCtx.userMessageChannel,
        );
      } catch (err) {
        log.warn(
          { err, conversationId },
          "Failed to set origin channel (best-effort)",
        );
      }
    }
    if (serverInterfaceCtx) {
      try {
        setConversationOriginInterfaceIfUnset(
          conversationId,
          serverInterfaceCtx.userMessageInterface,
        );
      } catch (err) {
        log.warn(
          { err, conversationId },
          "Failed to set origin interface (best-effort)",
        );
      }
    }

    if (serverTurnCtx || serverInterfaceCtx) {
      try {
        const convForMeta = getConversation(conversationId);
        if (convForMeta) {
          updateMetaFile(convForMeta);
        }
      } catch (err) {
        log.warn(
          { err, conversationId },
          "Failed to update disk meta (best-effort)",
        );
      }
    }

    const assistantMsg = createAssistantMessage(slashResult.message);
    const persistedAssistant = await addMessage(
      conversationId,
      "assistant",
      JSON.stringify(assistantMsg.content),
      { metadata: serverChannelMeta },
    );
    conversation.getMessages().push(assistantMsg);
    publishConversationMessagesChanged(conversationId);
    return {
      messageId: persisted.id,
      assistantMessageId: persistedAssistant.id,
    };
  }

  if (slashResult.kind === "compact") {
    const serverTurnCtx = conversation.getTurnChannelContext();
    const serverProvenance = provenanceFromTrustContext(
      // Ingress persists before any per-turn stamp; the slot was just written
      // by this message's own resolution, and the per-turn field may still
      // hold the previous turn's actor.
      restingTrust(conversation),
    );
    const compactChannelMeta = {
      ...serverProvenance,
      ...(serverTurnCtx
        ? {
            userMessageChannel: serverTurnCtx.userMessageChannel,
            assistantMessageChannel: serverTurnCtx.assistantMessageChannel,
          }
        : {}),
      ...(serverInterfaceCtx
        ? {
            userMessageInterface: serverInterfaceCtx.userMessageInterface,
            assistantMessageInterface:
              serverInterfaceCtx.assistantMessageInterface,
          }
        : {}),
    };
    const compactUserMeta = slackMeta
      ? { ...compactChannelMeta, slackMeta }
      : compactChannelMeta;
    const cleanMsg = await createUserMessage(content, attachments);
    const persisted = await addMessage(
      conversationId,
      "user",
      await serializePersistedUserMessageContent(
        content,
        options?.displayContent,
        attachments,
      ),
      { metadata: compactUserMeta },
    );
    conversation.getMessages().push(cleanMsg);

    conversation.emitActivityState("thinking", "context_compacting");
    const result = await conversation.forceCompact();
    const responseText = formatCompactResult(result);
    const assistantMsg = createAssistantMessage(responseText);
    const persistedAssistant = await addMessage(
      conversationId,
      "assistant",
      JSON.stringify(assistantMsg.content),
      { metadata: compactChannelMeta },
    );
    conversation.getMessages().push(assistantMsg);
    publishConversationMessagesChanged(conversationId);
    return {
      messageId: persisted.id,
      assistantMessageId: persistedAssistant.id,
    };
  }

  if (slashResult.kind === "clean") {
    const serverTurnCtx = conversation.getTurnChannelContext();
    const serverProvenance = provenanceFromTrustContext(
      // Ingress persists before any per-turn stamp; the slot was just written
      // by this message's own resolution, and the per-turn field may still
      // hold the previous turn's actor.
      restingTrust(conversation),
    );
    const cleanChannelMeta = {
      ...serverProvenance,
      ...(serverTurnCtx
        ? {
            userMessageChannel: serverTurnCtx.userMessageChannel,
            assistantMessageChannel: serverTurnCtx.assistantMessageChannel,
          }
        : {}),
      ...(serverInterfaceCtx
        ? {
            userMessageInterface: serverInterfaceCtx.userMessageInterface,
            assistantMessageInterface:
              serverInterfaceCtx.assistantMessageInterface,
          }
        : {}),
    };
    const cleanUserMeta = slackMeta
      ? { ...cleanChannelMeta, slackMeta }
      : cleanChannelMeta;
    const cleanMsg = await createUserMessage(content, attachments);
    const persisted = await addMessage(
      conversationId,
      "user",
      await serializePersistedUserMessageContent(
        content,
        options?.displayContent,
        attachments,
      ),
      { metadata: cleanUserMeta },
    );
    conversation.getMessages().push(cleanMsg);

    const result = await conversation.forceClean();
    const responseText = formatCleanResult(result);
    const assistantMsg = createAssistantMessage(responseText);
    const persistedAssistant = await addMessage(
      conversationId,
      "assistant",
      JSON.stringify(assistantMsg.content),
      { metadata: cleanChannelMeta },
    );
    conversation.getMessages().push(assistantMsg);
    publishConversationMessagesChanged(conversationId);
    return {
      messageId: persisted.id,
      assistantMessageId: persistedAssistant.id,
    };
  }

  const resolvedContent = slashResult.content;

  const requestId = uuidv7();
  const persistMetadata = options?.slackInbound
    ? { slackInbound: options.slackInbound }
    : undefined;
  const ingressKey = deriveIngressIdempotencyKey(options);
  const { id: messageId, deduplicated } = await conversation.persistUserMessage(
    {
      content: resolvedContent,
      attachments,
      requestId,
      metadata: persistMetadata,
      displayContent: options?.displayContent,
      ...(options?.skipUserMessageIndexing ? { skipIndexing: true } : {}),
      ...(ingressKey ? { clientMessageId: ingressKey } : {}),
    },
  );
  publishConversationMessagesChanged(conversationId);

  if (deduplicated) {
    // This exact ingress (same idempotency key) already ran a turn; skip the
    // loop so an at-least-once redelivery does not emit a second reply. The
    // `deduplicated` signal also stops channel finalization from re-delivering
    // the original reply.
    log.info(
      { conversationId, messageId },
      "Skipping agent loop for deduplicated ingress message",
    );
    return {
      messageId,
      deduplicated: true,
      turnFailure: readTurnFailure(messageId),
    };
  }

  const restoreToolScope = applyTurnToolAllowlist(conversation, options);
  try {
    await conversation.runAgentLoop(resolvedContent, messageId, {
      onEvent: emitEvent,
      isInteractive: options?.isInteractive ?? false,
      isUserMessage: true,
      ...(options?.callSite ? { callSite: options.callSite } : {}),
      ...(options?.overrideProfile
        ? { overrideProfile: options.overrideProfile }
        : {}),
      ...(options?.requestOrigin
        ? { requestOrigin: options.requestOrigin }
        : {}),
      ...(options?.cronRunId ? { cronRunId: options.cronRunId } : {}),
    });
  } finally {
    restoreToolScope();
  }

  // Read the just-finished turn's outcome from the stamp `runAgentLoop`'s
  // `finally` wrote onto this user-message row. A failed turn (e.g. an LLM
  // call that failed with an invalid provider) ends without throwing, so this
  // is the only signal an awaiting caller gets.
  const turnFailure = readTurnFailure(messageId);
  return { messageId, turnFailure };
}

/**
 * Fire-and-forget variant of {@link processMessage}. Persists the user
 * message and kicks off the agent loop in the background, returning the
 * `messageId` immediately without waiting for completion.
 *
 * Used by signal handlers and the conversation-launcher where the caller
 * does not await the full agent turn.
 */
export async function processMessageInBackground(
  conversationId: string,
  content: string,
  options?: ProcessMessageOptions,
): Promise<{ messageId: string }> {
  assertDbMigrationsReadyForTurn();

  const { conversation, attachments } = await prepareConversationForMessage(
    conversationId,
    content,
    options,
  );
  const emitEvent = buildEventEmitter(options?.onEvent);

  const requestId = uuidv7();
  const persistMetadata = options?.slackInbound
    ? { slackInbound: options.slackInbound }
    : undefined;
  const ingressKey = deriveIngressIdempotencyKey(options);
  const { id: messageId, deduplicated } = await conversation.persistUserMessage(
    {
      content,
      attachments,
      requestId,
      metadata: persistMetadata,
      displayContent: options?.displayContent,
      ...(ingressKey ? { clientMessageId: ingressKey } : {}),
    },
  );
  publishConversationMessagesChanged(conversationId);

  if (deduplicated) {
    // At-least-once redelivery of a turn that already ran — skip the loop.
    log.info(
      { conversationId, messageId },
      "Skipping background agent loop for deduplicated ingress message",
    );
    return { messageId };
  }

  const restoreToolScope = applyTurnToolAllowlist(conversation, options);
  conversation
    .runAgentLoop(content, messageId, {
      onEvent: emitEvent,
      isInteractive: options?.isInteractive ?? false,
      isUserMessage: true,
      ...(options?.callSite ? { callSite: options.callSite } : {}),
      ...(options?.overrideProfile
        ? { overrideProfile: options.overrideProfile }
        : {}),
      ...(options?.cronRunId ? { cronRunId: options.cronRunId } : {}),
    })
    .finally(() => {
      restoreToolScope();
    })
    .catch((err) => {
      log.error({ err, conversationId }, "Background agent loop failed");
    });

  return { messageId };
}