stack/packages/discord/src/connector.mjs

// The loop: gateway event -> authorize -> journal -> engine -> deliver.
//
// Ordering rules that make a restart safe:
//   1. An accepted message id is appended to the inbox before anything else
//      happens. On start the inbox is read back; a replayed id is ignored.
//   2. Every delivery journals an intent (with nonce, channel and content)
//      before the POST and a confirmed, refused or unknown line after it.
//   3. `start()` reconciles every unresolved intent by sending the same nonce
//      again with enforce_nonce, which makes Discord return the existing
//      message instead of posting twice. An intent older than the dedupe
//      window is marked refused, not re-sent, because a re-send could post a
//      second reply. If any intent is still unknown after that, start refuses.
//   4. A STOP file refuses new turns; the current one finishes or times out.
//   5. A reload swaps the binding between turns: authorization and limits
//      read the current binding at admission; a turn in flight keeps the
//      values it started with. Fixed keys (see binding.mjs) are refused.
//   7. Approvals (approvals.mjs): a turn whose tool calls opened an approval
//      request ends with the connector posting the request message with an
//      Approve button and binding its id at the record service through
//      `api`; the button, or a reply that is exactly `approve`, is resolved
//      here against the ledger and submitted by the connector, never by the
//      model. Every step is journaled before the call, so `start()` retries
//      unfinished binds and approvals under their original idempotency keys.
//   6. The daily ceiling counts admissions (admissions.jsonl) on the current
//      UTC date; an admission is written before the engine runs, so turns in
//      flight and turns cut short by a crash count too.
//      Over the ceiling, inbound messages are journaled as drops, one fixed
//      line is posted per day, and the process stays up idle.
//
// Everything with a side effect is passed in: rest, gateway, engine, clock,
// timers. The offline suite drives this with fakes.

import { authorize, isThreadType } from "./authorize.mjs";
import { reloadDiff } from "./binding.mjs";
import { envelope, splitReply } from "./context.mjs";
import {
  ensureJournal, appendInbox, readInboxIds, appendOutbox, unresolvedOutbox, appendDrop,
  writeTurn, appendAdmission, countAdmissionsOn, appendNotice, noticeOn, utcDate, stopRequested,
} from "./journal.mjs";
import { RestOutcome } from "./rest.mjs";
import { DiscordError } from "./errors.mjs";
import {
  APPROVAL_LINES, CALLBACK_TYPE, EPHEMERAL, INTERACTION_TYPE, appendApproval, approveComponents, interactionAuthorId,
  loadOpenRequests, renderRequest, resolveInteraction, resolveReply, validateRequest,
} from "./approvals.mjs";
import { SETSPARK_REFUSAL, SetsparkRefusal, connectorKey } from "./setspark.mjs";

// The reaction placed on every admitted message as a read receipt.
export const READ_RECEIPT = "\u{1F440}"; // eyes

export const FIXED_LINES = Object.freeze({
  failed: "I could not finish an answer to that message. The attempt is recorded and someone will look at it.",
  oversize: "That message is longer than I take in one go. Please send it in a shorter form.",
  ceiling: "I have reached my daily limit for replies here and will answer again after midnight UTC.",
});

export const RECONCILE_WINDOW_MS = 5 * 60 * 1000;

export function createConnector({
  binding: initialBinding, journalDir, rest, gateway, engine,
  // The record service client for the connector's own calls (bind an
  // approval message, add an approval). Absent means approvals are off:
  // an opened request is journaled as unbound and nothing is posted.
  api = null,
  now = () => Date.now(),
  setTimeoutImpl = globalThis.setTimeout, clearTimeoutImpl = globalThis.clearTimeout,
  typingIntervalMs = 8000,
  readReceipt = READ_RECEIPT,
  log = () => {},
} = {}) {
  for (const [k, v] of Object.entries({ binding: initialBinding, journalDir, rest, gateway, engine })) {
    if (!v) throw new DiscordError(`connector: ${k} required`, 1);
  }
  ensureJournal(journalDir);
  let binding = initialBinding;

  const state = {
    inbox: new Set(), channels: new Map(), inFlight: 0, typingTimer: null, typingChannel: null,
    pending: new Set(), stopping: false, started: false, deliveryChain: Promise.resolve(), turnPromises: new Set(),
    approvals: loadOpenRequests(journalDir), interactions: new Set(),
  };
  const principal = () => (binding.tools && binding.tools.setspark ? binding.tools.setspark.principal : binding.seat);
  const nameOf = (id) => {
    const u = binding.users.find((x) => x.id === id);
    return u ? u.name : null;
  };

  function iso() {
    return new Date(now()).toISOString();
  }

  function rememberChannel(c) {
    if (c && typeof c.id === "string") {
      state.channels.set(c.id, { type: c.type, parentId: c.parent_id ?? null, name: c.name ?? null, guildId: c.guild_id ?? binding.guildId });
    }
  }

  const channelInfo = (id) => state.channels.get(id);

  // --- delivery ---

  async function sendChunk({ channelId, replyTo, content, nonce, components = null }) {
    const base = { nonce, channelId, replyTo, at: iso() };
    appendOutbox(journalDir, { ...base, status: "intent", content, ...(components ? { components } : {}) });
    try {
      const r = await rest.createMessage(channelId, { content, nonce, replyTo, components });
      appendOutbox(journalDir, { ...base, at: iso(), status: "confirmed", messageId: r.messageId });
      return { nonce, status: "confirmed", messageId: r.messageId };
    } catch (err) {
      if (err instanceof RestOutcome) {
        appendOutbox(journalDir, { ...base, at: iso(), status: err.kind, error: err.message });
        log(`delivery ${nonce}: ${err.kind}: ${err.message}`);
        return { nonce, status: err.kind, error: err.message };
      }
      appendOutbox(journalDir, { ...base, at: iso(), status: "unknown", error: err.message });
      log(`delivery ${nonce}: unknown: ${err.message}`);
      return { nonce, status: "unknown", error: err.message };
    }
  }

  // Chunks go out in order; a chunk that is not confirmed stops the rest of
  // that reply so reconcile never produces an out-of-order tail.
  function deliver({ channelId, replyTo, text, noncePrefix }) {
    const chunks = splitReply(text, binding.limits.replyChunkChars);
    const run = async () => {
      const results = [];
      for (let i = 0; i < chunks.length; i++) {
        const r = await sendChunk({ channelId, replyTo: i === 0 ? replyTo : null, content: chunks[i], nonce: `${noncePrefix}-${i}` });
        results.push(r);
        if (r.status !== "confirmed") break;
      }
      return { chars: text.length, chunkCount: chunks.length, chunks: results };
    };
    const p = state.deliveryChain.then(run, run);
    state.deliveryChain = p.catch(() => {});
    return p;
  }

  // --- typing ---

  function typingTick() {
    state.typingTimer = null;
    if (state.inFlight === 0 || state.typingChannel === null) return;
    rest.typing(state.typingChannel);
    state.typingTimer = setTimeoutImpl(typingTick, typingIntervalMs);
  }

  function typingStart(channelId) {
    state.typingChannel = channelId;
    if (state.typingTimer === null) typingTick();
  }

  function typingStop() {
    if (state.inFlight > 0) return;
    if (state.typingTimer !== null) clearTimeoutImpl(state.typingTimer);
    state.typingTimer = null;
    state.typingChannel = null;
  }

  // --- approvals ---

  // A refusal the service stated is `refused`; anything where the command
  // may have run (timeout, socket error, 5xx, an unreadable 2xx) is
  // `unknown` and is retried under the same key on the next start.
  const UNKNOWN_REASONS = new Set([SETSPARK_REFUSAL.SERVER, SETSPARK_REFUSAL.TIMEOUT, SETSPARK_REFUSAL.NETWORK, SETSPARK_REFUSAL.TOO_BIG, SETSPARK_REFUSAL.NOT_JSON]);
  function apiOutcome(err) {
    if (err instanceof SetsparkRefusal) return { status: UNKNOWN_REASONS.has(err.reason) ? "unknown" : "refused", error: `${err.reason}${err.code ? ` (code ${err.code})` : ""}` };
    return { status: "unknown", error: err.message };
  }

  // Bind the posted message to the request at the service. Journaled as
  // intent before the call; the key is fixed by the message id, so a retry
  // after a crash repeats the same command.
  async function bindRequest(rec) {
    const base = { kind: "bind", requestId: rec.request.requestId, messageId: rec.messageId, channelId: rec.channelId };
    appendApproval(journalDir, { ...base, at: iso(), status: "intent" });
    rec.bind = { status: "intent", at: iso(), error: null };
    try {
      await api.bindApprovalMessage({ requestId: rec.request.requestId, messageId: rec.messageId, channelId: rec.channelId, idempotencyKey: connectorKey(principal(), rec.messageId, "bind") });
      appendApproval(journalDir, { ...base, at: iso(), status: "done" });
      rec.bind = { status: "done", at: iso(), error: null };
    } catch (err) {
      const o = apiOutcome(err);
      appendApproval(journalDir, { ...base, at: iso(), status: o.status, error: o.error });
      rec.bind = { status: o.status, at: iso(), error: o.error };
      log(`bind ${rec.request.requestId}: ${o.status}: ${o.error}`);
    }
    return rec.bind.status;
  }

  // Post the request message for a request the model opened during a turn.
  // Returns what the turn record keeps about it.
  async function postRequest(rawRequest, { channelId, replyTo }) {
    let request;
    try {
      request = validateRequest(rawRequest);
    } catch (err) {
      log(`approval request rejected: ${err.message}`);
      return { requestId: rawRequest && rawRequest.requestId, status: "refused", error: err.message };
    }
    if (!api) {
      log(`approval request ${request.requestId}: no api client, not posted`);
      return { requestId: request.requestId, status: "refused", error: "approvals are not enabled on this binding" };
    }
    const names = request.approvers.map(nameOf).filter((n) => n !== null);
    const content = renderRequest(request, names);
    const run = () => sendChunk({ channelId, replyTo, content, nonce: `${replyTo}-a`, components: approveComponents(request.requestId) });
    const p = state.deliveryChain.then(run, run);
    state.deliveryChain = p.catch(() => {});
    const sent = await p;
    if (sent.status !== "confirmed") return { requestId: request.requestId, status: sent.status, error: sent.error };
    const rec = { request, messageId: sent.messageId, channelId, content, openedAt: iso(), bind: null, approvals: new Map() };
    appendApproval(journalDir, { kind: "opened", at: rec.openedAt, ...request, messageId: rec.messageId, channelId, content });
    state.approvals.set(rec.messageId, rec);
    const bind = await bindRequest(rec);
    return { requestId: request.requestId, status: "posted", messageId: rec.messageId, bind };
  }

  // Submit one approver's approval. Journaled as intent before the call.
  // The message id is the event's own (the reply, or the button's message);
  // the bound message id is the request message, so the service can check
  // a reply targets the bound proposal. The evidence is one Discord message
  // unique to the approver: the reply itself, or for a button press the
  // confirmation line the connector posted (evidenceId) before calling.
  // Its url and exact text go to the service as source_url and statement.
  // Resolves the ledger status: done, refused or unknown.
  const messageUrl = (channelId, messageId) => `https://discord.com/channels/${binding.guildId}/${channelId}/${messageId}`;
  async function submitApproval(rec, { authorId, eventId, messageId, how, evidenceId, statement }) {
    const base = { kind: "approval", requestId: rec.request.requestId, authorId, eventId, messageId, how, evidenceId, statement };
    appendApproval(journalDir, { ...base, at: iso(), status: "intent" });
    rec.approvals.set(authorId, { status: "intent", at: iso(), eventId, messageId, how, evidenceId, statement, error: null });
    let outcome;
    try {
      await api.addApproval({
        requestId: rec.request.requestId, kind: how, authorId, messageId, boundMessageId: rec.messageId,
        sourceUrl: messageUrl(rec.channelId, evidenceId), statement,
        idempotencyKey: connectorKey(principal(), eventId, "approval"),
      });
      outcome = { status: "done", error: null };
    } catch (err) {
      outcome = apiOutcome(err);
      log(`approval ${rec.request.requestId} by ${authorId}: ${outcome.status}: ${outcome.error}`);
    }
    appendApproval(journalDir, { ...base, at: iso(), status: outcome.status, ...(outcome.error ? { error: outcome.error } : {}) });
    rec.approvals.set(authorId, { ...outcome, at: iso(), eventId, messageId, how, evidenceId, statement });
    return outcome.status;
  }

  // The confirmation line posted for a button press, in the request's
  // channel, before the service is asked. Sage authored it, so it is one
  // fetchable message unique to this approver and this press.
  function confirmationText(rec, authorId) {
    return `Approval: ${nameOf(authorId) || "a listed approver"} approved ${rec.request.decisionId} v${rec.request.proposalVersion} (digest ${rec.request.digest.slice(0, 8)}) by button.`;
  }
  async function postConfirmation(rec, authorId, interactionId) {
    const content = confirmationText(rec, authorId);
    const run = () => sendChunk({ channelId: rec.channelId, replyTo: rec.messageId, content, nonce: `${interactionId}-c` });
    const p = state.deliveryChain.then(run, run);
    state.deliveryChain = p.catch(() => {});
    const sent = await p;
    return sent.status === "confirmed" ? { evidenceId: sent.messageId, statement: content } : null;
  }

  const allApproved = (rec) => rec.request.approvers.every((a) => (rec.approvals.get(a) || {}).status === "done");

  function lineFor(reason) {
    if (reason === "not-approver") return APPROVAL_LINES.notApprover;
    if (reason === "already" || reason === "pending") return APPROVAL_LINES.already;
    return APPROVAL_LINES.closed;
  }

  // A reply that is exactly the approve word on a request message.
  async function handleApprovalReply(message, decision) {
    const channelId = message.channel_id;
    if (!decision.ok) {
      appendDrop(journalDir, { at: iso(), reason: `approval-${decision.reason}`, messageId: message.id, channelId, authorId: message.author.id });
      await track(deliver({ channelId, replyTo: message.id, text: lineFor(decision.reason), noncePrefix: `${message.id}-x` }));
      return { accepted: false, reason: `approval-${decision.reason}` };
    }
    const status = await submitApproval(decision.rec, { authorId: decision.authorId, eventId: message.id, messageId: message.id, how: "reply", evidenceId: message.id, statement: message.content });
    const text = status === "done" ? `Approved by ${nameOf(decision.authorId) || "a listed approver"}.` : APPROVAL_LINES.failed;
    await track(deliver({ channelId, replyTo: message.id, text, noncePrefix: `${message.id}-x` }));
    return { accepted: true, approval: status };
  }

  // The Approve button. Discord wants an answer within seconds: a refusal
  // is answered at once with a private line; an approval is deferred, the
  // service is asked, and the request message is then edited in place.
  async function handleInteraction(interaction) {
    if (!interaction || typeof interaction !== "object" || typeof interaction.id !== "string") return { accepted: false, reason: "not-an-interaction" };
    if (interaction.type !== INTERACTION_TYPE.MESSAGE_COMPONENT) return { accepted: false, reason: "not-a-component" };
    if (state.interactions.has(interaction.id)) return { accepted: false, reason: "duplicate" };
    state.interactions.add(interaction.id);
    const authorId = interactionAuthorId(interaction);
    const messageId = interaction.message && interaction.message.id;
    const decision = resolveInteraction(interaction, state.approvals);
    if (!decision.ok) {
      appendDrop(journalDir, { at: iso(), reason: `approval-${decision.reason}`, interactionId: interaction.id, messageId, channelId: interaction.channel_id, authorId });
      await rest.interactionCallback(interaction.id, interaction.token, { type: CALLBACK_TYPE.CHANNEL_MESSAGE, data: { content: lineFor(decision.reason), flags: EPHEMERAL, allowed_mentions: { parse: [] } } });
      return { accepted: false, reason: `approval-${decision.reason}` };
    }
    const deferred = await rest.interactionCallback(interaction.id, interaction.token, { type: CALLBACK_TYPE.DEFERRED_UPDATE_MESSAGE });
    const rec = decision.rec;
    // No confirmation message, no call: the press is dropped and the
    // approver can press again.
    const evidence = await postConfirmation(rec, authorId, interaction.id);
    const status = evidence
      ? await submitApproval(rec, { authorId, eventId: interaction.id, messageId, how: "button", ...evidence })
      : "not-posted";
    if (!evidence) {
      appendDrop(journalDir, { at: iso(), reason: "approval-confirmation-not-posted", interactionId: interaction.id, messageId, channelId: interaction.channel_id, authorId });
      log(`interaction ${interaction.id}: confirmation not posted, approval not submitted`);
    }
    if (!deferred) return { accepted: true, approval: status, edited: false };
    const names = rec.request.approvers.filter((a) => (rec.approvals.get(a) || {}).status === "done").map((a) => nameOf(a) || "a listed approver");
    const content = status === "done"
      ? `${rec.content}
Approved by ${names.join(", ")}.`
      : `${rec.content}
${status === "not-posted" ? APPROVAL_LINES.pressAgain : APPROVAL_LINES.failed}`;
    let edited = false;
    try {
      await rest.editInteractionMessage(interaction.application_id, interaction.token, { content: content.slice(0, 2000), components: approveComponents(rec.request.requestId, { disabled: allApproved(rec) }) });
      edited = true;
    } catch (err) {
      log(`interaction ${interaction.id}: edit failed: ${err.message}`);
    }
    return { accepted: true, approval: status, edited };
  }

  // Retry binds and approvals that were journaled but never confirmed.
  // Safe by construction: the keys are fixed by the event ids.
  async function reconcileApprovals() {
    const results = [];
    if (!api) return results;
    for (const rec of state.approvals.values()) {
      if (rec.bind && (rec.bind.status === "intent" || rec.bind.status === "unknown")) {
        results.push({ kind: "bind", requestId: rec.request.requestId, status: await bindRequest(rec) });
      }
      for (const [authorId, a] of rec.approvals) {
        if (a.status === "intent" || a.status === "unknown") {
          if (typeof a.evidenceId !== "string" || typeof a.statement !== "string") {
            log(`approval ${rec.request.requestId} by ${authorId}: ledger entry has no evidence, not retried`);
            continue;
          }
          results.push({ kind: "approval", requestId: rec.request.requestId, authorId, status: await submitApproval(rec, { authorId, eventId: a.eventId, messageId: a.messageId, how: a.how, evidenceId: a.evidenceId, statement: a.statement }) });
        }
      }
    }
    return results;
  }

  // --- turns ---

  async function runTurn(message, auth) {
    const startedAt = iso();
    const t0 = now();
    const targetChannel = auth.thread ? auth.thread.id : auth.channel.id;
    const record = {
      turnVersion: 1, binding: binding.name, seat: binding.seat,
      messageId: message.id, channelId: auth.channel.id, channelName: auth.channel.name,
      threadId: auth.thread ? auth.thread.id : null, threadName: auth.thread ? auth.thread.name : null,
      authorId: message.author.id, requester: auth.user.name, startedAt, inboundChars: message.content.length,
      engine: { provider: binding.engine.provider, model: binding.engine.model, thinking: binding.engine.thinking, usage: null, turns: null },
      tools: binding.tools ? [] : null,
    };
    const prompt = envelope({
      guildName: binding.guildName, channelName: auth.channel.name, threadName: auth.thread ? auth.thread.name : null,
      authorId: message.author.id, requester: auth.user.name, messageId: message.id, text: message.content,
    });
    state.inFlight += 1;
    typingStart(targetChannel);
    // Read receipt: the reaction goes on the inbound message as soon as the
    // turn is admitted, so the author sees it was received before the reply
    // lands. It runs alongside the engine; its outcome goes in the record.
    const receipt = rest.react(message.channel_id, message.id, readReceipt).then((ok) => ({ emoji: readReceipt, ok }));
    let status = "ok";
    let error = null;
    let reply = null;
    try {
      const result = await engine.prompt(prompt, { timeoutMs: binding.limits.turnTimeoutSeconds * 1000 });
      record.engine.usage = result.usage;
      record.engine.turns = result.turns ?? null;
      if (binding.tools) record.tools = result.tools || [];
      if (result.model) record.engine.model = result.model;
      if (result.text.length === 0) throw new DiscordError("engine returned no text", 1, { code: "engine-empty" });
      reply = await deliver({ channelId: targetChannel, replyTo: message.id, text: result.text, noncePrefix: message.id });
      const opened = (result.tools || []).filter((t) => t && t.name === "open_approval_request" && t.ok && t.request);
      if (opened.length > 0) {
        record.approvalRequests = [];
        for (const t of opened) record.approvalRequests.push(await postRequest(t.request, { channelId: targetChannel, replyTo: message.id }));
      }
    } catch (err) {
      status = "failed";
      error = { code: (err.details && err.details.code) || "error", message: err.message };
      log(`turn ${message.id} failed: ${error.code}: ${error.message}`);
      // Q19: one fixed line, never model output, on failure.
      reply = await deliver({ channelId: targetChannel, replyTo: message.id, text: FIXED_LINES.failed, noncePrefix: `${message.id}-f` });
    } finally {
      state.inFlight -= 1;
      typingStop();
    }
    record.receipt = await receipt;
    const endedAt = iso();
    writeTurn(journalDir, message.id, { ...record, status, error, reply, endedAt, latencyMs: now() - t0 });
    return status;
  }

  function track(p) {
    state.turnPromises.add(p);
    p.finally(() => state.turnPromises.delete(p)).catch(() => {});
    return p;
  }

  function drop(message, reason, extra = {}) {
    appendDrop(journalDir, {
      at: iso(), reason, messageId: message && message.id, channelId: message && message.channel_id,
      authorId: message && message.author && message.author.id, ...extra,
    });
    log(`drop ${reason} message=${message && message.id}`);
    return { accepted: false, reason };
  }

  // Exposed for tests. Returns {accepted, reason?, turn?}. The id is
  // reserved in `pending` for the whole call, so a duplicate event arriving
  // while the thread lookup below is awaiting cannot be admitted twice.
  async function handleMessage(message) {
    if (!message || typeof message !== "object" || typeof message.id !== "string") return drop(message, "not-a-message");
    if (state.inbox.has(message.id) || state.pending.has(message.id)) return drop(message, "duplicate");
    state.pending.add(message.id);
    try {
      return await admit(message);
    } finally {
      state.pending.delete(message.id);
    }
  }

  async function admit(message) {
    // A thread the connector has not seen: one REST lookup, only when the
    // cheap checks (guild, listed author) would let the message through.
    const listed = binding.channels.some((c) => c.id === message.channel_id);
    if (!listed && !state.channels.has(message.channel_id) && message.guild_id === binding.guildId
      && message.author && binding.users.some((u) => u.id === message.author.id)) {
      try {
        rememberChannel(await rest.getChannel(message.channel_id));
      } catch (err) {
        log(`channel lookup ${message.channel_id} failed: ${err.message}`);
      }
    }
    const auth = authorize(binding, message, channelInfo);
    if (!auth.ok) return drop(message, auth.reason);
    if (typeof message.content !== "string") return drop(message, "no-content");

    appendInbox(journalDir, { id: message.id, at: iso(), channelId: message.channel_id, authorId: message.author.id, chars: message.content.length });
    state.inbox.add(message.id);
    const targetChannel = auth.thread ? auth.thread.id : auth.channel.id;

    if (state.stopping || stopRequested(journalDir)) return drop(message, "stopped");
    // A reply that is exactly the approve word on a request message is an
    // approval, not a prompt; any other reply to that message goes to the
    // model as usual.
    const approval = resolveReply(message, state.approvals);
    if (approval.reason !== "not-a-request" && approval.reason !== "not-approve") return handleApprovalReply(message, approval);
    if (auth.oversize) {
      await track(deliver({ channelId: targetChannel, replyTo: message.id, text: FIXED_LINES.oversize, noncePrefix: `${message.id}-b` }));
      return drop(message, "oversize", { chars: message.content.length });
    }
    const today = utcDate(now());
    // Admissions are durable and appended before the engine runs, so a burst
    // in flight and a turn interrupted by a crash both count.
    if (countAdmissionsOn(journalDir, today) >= binding.limits.turnsPerDay) {
      // One notice per UTC day, durable across restarts: journaled before the attempt.
      if (!noticeOn(journalDir, "ceiling", today)) {
        appendNotice(journalDir, { kind: "ceiling", date: today, at: iso(), messageId: message.id });
        await track(deliver({ channelId: targetChannel, replyTo: message.id, text: FIXED_LINES.ceiling, noncePrefix: `${message.id}-c` }));
      }
      return drop(message, "ceiling", { limit: binding.limits.turnsPerDay });
    }
    appendAdmission(journalDir, { id: message.id, at: iso(), channelId: targetChannel });
    const turn = track(runTurn(message, auth));
    return { accepted: true, turn };
  }

  function onDispatch({ t, d }) {
    switch (t) {
      case "GUILD_CREATE":
        if (d && d.id === binding.guildId) {
          for (const c of d.channels || []) rememberChannel({ ...c, guild_id: d.id });
          for (const c of d.threads || []) rememberChannel({ ...c, guild_id: d.id });
        }
        return;
      case "THREAD_CREATE":
      case "THREAD_UPDATE":
      case "CHANNEL_CREATE":
      case "CHANNEL_UPDATE":
        rememberChannel(d);
        return;
      case "THREAD_LIST_SYNC":
        for (const c of (d && d.threads) || []) rememberChannel({ ...c, guild_id: d.guild_id });
        return;
      case "MESSAGE_CREATE":
        handleMessage(d).catch((err) => log(`handleMessage failed: ${err.message}`));
        return;
      case "INTERACTION_CREATE":
        track(handleInteraction(d)).catch((err) => log(`handleInteraction failed: ${err.message}`));
        return;
      default:
        return;
    }
  }

  // Re-send unresolved intents with their original nonce. Returns the list
  // of outcomes. Throws if anything is still unknown afterwards.
  async function reconcile() {
    const results = [];
    for (const e of unresolvedOutbox(journalDir)) {
      const base = { nonce: e.nonce, channelId: e.channelId, replyTo: e.replyTo ?? null, at: iso() };
      // Age from the first line for this nonce; a retry never refreshes it.
      const age = now() - Date.parse(e.intentAt || e.at || 0);
      if (!(age < RECONCILE_WINDOW_MS) || typeof e.content !== "string" || typeof e.channelId !== "string") {
        appendOutbox(journalDir, { ...base, status: "refused", error: "stale intent not re-sent (outside the nonce dedupe window)" });
        results.push({ nonce: e.nonce, status: "refused", stale: true });
        continue;
      }
      try {
        const r = await rest.createMessage(e.channelId, { content: e.content, nonce: e.nonce, replyTo: e.replyTo ?? null, components: e.components ?? null });
        appendOutbox(journalDir, { ...base, status: "confirmed", messageId: r.messageId, reconciled: true });
        results.push({ nonce: e.nonce, status: "confirmed", messageId: r.messageId });
      } catch (err) {
        const status = err instanceof RestOutcome ? err.kind : "unknown";
        appendOutbox(journalDir, { ...base, status, error: err.message, reconciled: true });
        results.push({ nonce: e.nonce, status, error: err.message });
      }
    }
    const stillUnknown = results.filter((r) => r.status === "unknown");
    if (stillUnknown.length > 0) {
      throw new DiscordError(`outbox has ${stillUnknown.length} unreconciled delivery(ies); resolve by hand before starting`, 1, { nonces: stillUnknown.map((r) => r.nonce) });
    }
    return results;
  }

  return {
    handleMessage,
    handleInteraction,
    onDispatch,
    reconcile,
    reconcileApprovals,
    get approvals() {
      return state.approvals;
    },
    rememberChannel,
    get binding() {
      return binding;
    },
    // Swap the binding in place. Throws (exit 2) and changes nothing when a
    // fixed key differs. Returns the summary of what changed.
    reload(next) {
      const diff = reloadDiff(binding, next);
      binding = next;
      return diff;
    },
    get inFlight() {
      return state.inFlight;
    },
    get inboxSize() {
      return state.inbox.size;
    },

    async start() {
      if (state.started) throw new DiscordError("connector already started", 1);
      if (stopRequested(journalDir)) throw new DiscordError(`STOP is present in ${journalDir}; remove it to run`, 1);
      const reconciled = await reconcile();
      const approvals = await reconcileApprovals();
      if (approvals.length > 0) log(`approvals: ${approvals.map((a) => `${a.kind} ${a.requestId} ${a.status}`).join("; ")}`);
      state.inbox = readInboxIds(journalDir);
      state.started = true;
      engine.start();
      gateway.on("dispatch", onDispatch);
      gateway.connect();
      log(`started: ${state.inbox.size} inbox id(s), ${reconciled.length} reconciled delivery(ies)`);
      return { inbox: state.inbox.size, reconciled };
    },

    // Finish in-flight turns (each has its own timeout), then close.
    async stop() {
      state.stopping = true;
      gateway.close(1000, "stop");
      await Promise.allSettled([...state.turnPromises]);
      await state.deliveryChain;
      typingStop();
      await engine.stop();
    },
  };
}