/**
 * The report outbox: the half of a tick report the model does not own.
 *
 * Authorship stays with the model — rendering a material report needs judgement
 * the daemon does not have. Delivery does not. On 2026-08-06 a suite release and
 * two tier-2 escalations were written and never sent, because a report reached
 * the operator only if the model remembered to call `telegram_send`, and an
 * unsent report was indistinguishable from a quiet tick (#123). `TICK_DELIVERY_RULE`
 * made that an instruction; this module makes it a mechanism.
 *
 * Three rules shape everything here:
 *
 *  1. **The row exists before the request does.** A report is persisted
 *     `pending` when it is handed over, and moved to `sending` with the id of
 *     the attempt about to run *before* the fetch. A crash therefore leaves an
 *     explicitly ambiguous row, not a silently lost one.
 *  2. **Delivery is at-least-once, and says so.** The Bot API accepts no
 *     client-supplied idempotency key and offers no bot-readable record of what
 *     it already sent, so "exactly once" cannot be built on this transport —
 *     `escalate.ts:258-266` is the incident from guessing the other way, where a
 *     delivered page was read as rejected and re-sent every tick. A recovered
 *     `sending` row is retried, and its message says it may be a repeat.
 *  3. **Not every throw means the same thing.** {@link sendTelegram} classifies
 *     its own failures, because only the code watching the socket can. A
 *     *definitive* failure — Telegram refused it, or the connection never
 *     opened — proves nobody has the message, so the row goes back to
 *     `pending` and the retry is an ordinary first attempt. An *unknown*
 *     outcome — cut off after the request left, or a 200 whose body could not
 *     be read — proves nothing, so the row stays `sending` and is flagged: the
 *     retry says it may be a repeat. Anything unclassifiable counts as unknown.
 *     Collapsing the two is the tempting simplification and the wrong one; it
 *     re-posts a delivered report as though it were new.
 *
 * The retry posture is `escalate.ts:151-160` generalised, not reinvented: the
 * ledger row is only advanced by an outcome, so an attempt that reaches no
 * conclusion leaves something for the next pass to pick up.
 */

import { availabilityDisposition, interruptDisposition } from "./availability.ts";
import {
  readTelegramToken,
  resolveProjectTopicId,
  sendTelegram,
  telegramTextParts,
  TelegramSendError,
} from "./escalate.ts";
import { createHash } from "node:crypto";
import { localDayKey } from "./digest-schedule.ts";
import {
  DIGEST_BACKLOG_LIMIT,
  INTERRUPT_CATEGORIES,
  type DigestBacklog,
  type Escalation,
  type HeldNotice,
  type InterruptCategory,
  type ProjectConfig,
  type ReportEnqueue,
  type ReportRecord,
  type Store,
} from "./types.ts";

/**
 * Attempts before a report is written off. Six attempts across the backoff
 * below is a little over half an hour — long enough to ride out a Telegram
 * blip or a router reboot, short enough that an operator learns their reporting
 * channel is broken while the day it broke is still today.
 */
export const REPORT_MAX_ATTEMPTS = 6;

/** Ceiling on the retry gap. Past this, waiting longer buys nothing: whatever
 *  is broken needs a human, and `status` plus the tier-2 page are how they hear. */
const REPORT_BACKOFF_CAP_MS = 15 * 60_000;

const REPORT_BACKOFF_BASE_MS = 30_000;

/**
 * How long a `sending` row may sit untouched before a pass reclaims it. A
 * process that died has no timer left to resolve its own row, and neither does
 * a `fetch` that never came back — the attempt id on every terminal transition
 * is what makes reclaiming safe, because a late answer from the reclaimed
 * attempt no longer matches the row and is discarded rather than applied.
 */
const SENDING_STALE_MS = 5 * 60_000;

/** Reports sent per pass. Bounded so a backlog cannot hold the pass — and the
 *  daemon's shutdown drain — open for an unpredictable stretch. */
const REPORT_BATCH = 5;

/** Fleet-scoped pages carry issue `0`; a report has no tracker issue at all. */
const NO_ISSUE = 0;

/** Bound on failure text in the operator-facing status block. */
const ERROR_SAMPLE = 90;

/**
 * The within-run harness reliability a settlement now names where a human
 * reads it (#584). The row always carries the resolved model/provider and the
 * counts (written at settlement straight off the worker result); this is only
 * the *wording* — the sentence a settlement report and a tick digest tell an
 * operator who would otherwise have to read a transcript to learn that the
 * run finished on a fallback after a mid-run swap, or rode out a throttled
 * provider on retries and compaction.
 *
 * Deliberately empty for a clean run: a run that never swapped, retried or
 * compacted changes neither the settlement report nor the digest, so the
 * "additive" claim is the observable one — byte-for-byte today's output on a
 * quiet fleet. The resolved model is still recorded on the row every time (#584's
 * second fake), but a clean run's narrative has nothing to add.
 */
export interface ReliabilitySettlement {
  resolvedModel?: string;
  resolvedProvider?: string;
  retryFallbacks: { from: string; to: string }[];
  retryFallbackSucceeded: number;
  modelRecoveries: number;
  autoRetryCount: number;
  autoCompactionCount: number;
}

/**
 * The human sentence for the above reliability surface, or undefined when there
 * is nothing worth saying (a clean run). One line so it drops into both the
 * settlement report and a material event's summary unchanged.
 */
export function reliabilitySettlementLine(
  f: ReliabilitySettlement,
): string | undefined {
  const { resolvedModel, resolvedProvider } = f;
  if (f.retryFallbacks.length === 0 && f.retryFallbackSucceeded === 0 && f.modelRecoveries === 0 && f.autoRetryCount === 0 && f.autoCompactionCount === 0) {
    return undefined;
  }
  const activity = [
    f.retryFallbacks.length > 0
      ? `${f.retryFallbacks.length} within-run model swap${f.retryFallbacks.length === 1 ? "" : "s"}`
      : undefined,
    f.retryFallbackSucceeded > 0 ? `${f.retryFallbackSucceeded} recovered` : undefined,
    f.modelRecoveries > 0 ? `${f.modelRecoveries} model recovery` : undefined,
    f.autoRetryCount > 0 ? `${f.autoRetryCount} provider retry` : undefined,
    f.autoCompactionCount > 0 ? `${f.autoCompactionCount} compaction` : undefined,
  ].filter((part): part is string => part !== undefined);
  const finished =
    resolvedModel === undefined
      ? "finished on the fallback"
      : `finished on ${resolvedModel}${resolvedProvider === undefined ? "" : ` (${resolvedProvider})`}`;
  return `Reliability: ${activity.join(", ")}; ${finished}.`;
}

/**
 * One delivery attempt's transport for a *single message*. Injected so the
 * delivery contract — claim, split, send, record progress, retry — is testable
 * without a live bot, which is the same split `confineToolCall` and `verifyPr`
 * use: the decision is separable from the I/O. The outbox splits a long report
 * into labelled parts itself and drives one call per part (so partial progress
 * can be persisted and resumed, #566); a text passed here is always under the
 * wire limit. Resolves with Telegram's own message id when it returns one,
 * throws on any known failure — {@link TelegramSendError} carries the
 * definitive/unknown verdict the retry's honesty hangs on.
 */
export type ReportSend = (text: string) => Promise<number | undefined>;

/** What one pass did, by report id. Returned for the daemon's log and the tests. */
export interface ReportDeliveryPass {
  delivered: string[];
  /** Material reports preserved without sending after policy or availability changed. */
  deferred: string[];
  /** Known-failed, back in `pending` behind a backoff. */
  requeued: string[];
  /** The attempt ended without an answer. Left `sending` and flagged as a
   *  possible repeat; stale recovery hands it back when the window closes. */
  uncertain: string[];
  /** Out of attempts. Terminal, and escalated as tier 2 in its own right. */
  failed: string[];
  /** Reclaimed from `sending`: their outcome was never learned. */
  recovered: string[];
}

export interface ReportOutbox {
  /**
   * Return `sending` rows last touched at or before `staleAt` to `pending`,
   * flagged ambiguous. Called at daemon startup with "now" — every row still
   * `sending` when a daemon boots belonged to a process that is gone.
   */
  recover(staleAt: number): ReportRecord[];
  /** One bounded pass: reclaim, then send what is due, oldest first. */
  deliverDue(): Promise<ReportDeliveryPass>;
}

export interface ReportOutboxDeps {
  /** A provider lets a resident daemon apply config edits at the next tick. */
  project: ProjectConfig | (() => ProjectConfig);
  store: Store;
  /** A report nobody can deliver escalates through this. Optional so a unit
   *  test can exercise delivery without wiring an escalator. */
  escalate?: (e: Escalation) => Promise<void>;
  send?: ReportSend;
  now?: () => number;
  log?: (msg: string) => void;
  /** False while the resident daemon cannot validate live delivery policy. */
  deliveryAllowed?: () => boolean;
}

/**
 * Local day, matching how a human reads "one daily digest" and how the
 * dispatcher's own `startOfToday` reads "today". A UTC key would roll the
 * digest over mid-evening for anyone west of Greenwich; the zone is the
 * project's `reporting.digest.timezone` when set, else the host zone (#229).
 */
export function digestDedupeKey(at: number, timezone?: string): string {
  return `digest:${localDayKey(at, timezone)}`;
}

/** Exponential, capped. `attempts` is attempts *started*, so the first failure
 *  waits one base interval rather than none. */
export function reportBackoffMs(attempts: number): number {
  const exponent = Math.max(0, attempts - 1);
  if (exponent >= 31) return REPORT_BACKOFF_CAP_MS;
  return Math.min(REPORT_BACKOFF_BASE_MS * 2 ** exponent, REPORT_BACKOFF_CAP_MS);
}

/**
 * What actually lands in the chat.
 *
 * Plain text, no `parse_mode`, for the same reason {@link formatEscalation} is:
 * an underscore in a repo name would otherwise make Telegram reject the whole
 * send and turn a cosmetic problem into a lost report.
 *
 * The report id leads because it is the only thing this transport gives an
 * operator to recognise a duplicate by. When the row is ambiguous the message
 * says so in words rather than a symbol — the person reading it at 03:00 needs
 * to know whether to act twice, and "(2)" does not tell them that.
 */
export function formatReportMessage(r: ReportRecord, project: string): string {
  const lines = [
    `omp-conductor · report ${r.id}${r.ambiguous ? "  (POSSIBLE REPEAT)" : ""}`,
    `project: ${project}`,
  ];
  if (r.ambiguous) {
    lines.push(
      `An earlier attempt at this exact report left this host and never came back with an ` +
        `answer, so it may already be in this chat. Delivery is at-least-once: check for ` +
        `report ${r.id} above and ignore whichever copy is the duplicate.`,
    );
  }
  lines.push("", r.body);
  return lines.join("\n");
}

/**
 * The `reports` block in `status`.
 *
 * `pending` and `sending` are printed differently on purpose: they ask for
 * different things. A pending report is one nobody has — the daemon will retry
 * it, and if that keeps failing the error text says why. A sending report is one
 * whose outcome was never learned, so the operator may already have it, and the
 * retry that resolves it will arrive marked as a possible repeat. Collapsing the
 * two into "not delivered" would lose exactly the distinction that decides
 * whether they go and look in the chat.
 */
export function formatOpenReports(
  reports: readonly ReportRecord[],
  now: number = Date.now(),
): string[] {
  if (reports.length === 0) return [];
  const counts = { pending: 0, sending: 0, failed: 0 };
  for (const r of reports) {
    if (r.state === "pending") counts.pending += 1;
    else if (r.state === "sending") counts.sending += 1;
    else if (r.state === "failed") counts.failed += 1;
  }
  const lines = [
    "",
    `reports   ${counts.pending} pending · ${counts.sending} sending · ${counts.failed} failed ` +
      `(delivery is at-least-once — a retry may duplicate)`,
  ];
  for (const r of reports) {
    const head =
      `  ${r.id}  ${r.state === "pending" ? "pending" : r.state.toUpperCase()}`.padEnd(24) +
      `  ${r.kind.padEnd(8)}  ${humanAge(now - r.createdAt)} old`;
    lines.push(`${head}  ${openReportDetail(r, now)}`);
  }
  return lines;
}

/** Durable rows still owed to a future digest. Always rendered in `status`: a
 * quiet line proves the accumulator is empty, while a non-zero line makes loss
 * or backlog visible without asking the session what it remembers. Held
 * notices are split by what releases them: `releaseOnAvailable` rows are
 * interrupt-worthy notices waiting only for the operator's working window,
 * everything else is ordinary digest-bound backlog. One noun for both was how
 * a routine material backlog read as stranded escalations (#492). */
export function formatDigestBacklog(backlog: DigestBacklog, now: number = Date.now()): string[] {
  const age = (at: number | undefined): string =>
    at === undefined ? "" : ` (oldest ${humanAge(Math.max(0, now - at))})`;
  const heldForAvailability = backlog.availabilityHeldNoticeCount ?? 0;
  const heldForDigest = Math.max(0, backlog.heldNoticeCount - heldForAvailability);
  return [
    "",
    "digest backlog",
    `  material events         ${backlog.materialCount}${age(backlog.materialOldestAt)}`,
    `  held for digest         ${heldForDigest}${age(backlog.digestHeldNoticeOldestAt ?? backlog.heldNoticeOldestAt)}`,
    `  held for availability   ${heldForAvailability}${age(backlog.availabilityHeldNoticeOldestAt)}`,
  ];
}

function openReportDetail(r: ReportRecord, now: number): string {
  const flat = r.lastError?.replace(/\s+/g, " ").trim() ?? "";
  const error =
    flat.length === 0 ? "" : `  (${flat.length > ERROR_SAMPLE ? `${flat.slice(0, ERROR_SAMPLE)}…` : flat})`;
  if (r.state === "failed") {
    return `gave up after ${r.attempts} attempts — undeliverable${error}`;
  }
  if (r.state === "sending") {
    return (
      `attempt ${r.attempts}, outcome unknown — it left this host and never came back ` +
      `with an answer, so Telegram may already have it; the retry will say it may be a ` +
      `repeat${error}`
    );
  }
  const wait = r.nextAttemptAt <= now ? "due now" : `retry in ${humanAge(r.nextAttemptAt - now)}`;
  const repeat = r.ambiguous ? ", may be a repeat" : "";
  return `attempt ${r.attempts}/${REPORT_MAX_ATTEMPTS}, ${wait}${repeat}${error}`;
}

function humanAge(ms: number): string {
  if (ms < 60_000) return `${Math.max(0, Math.round(ms / 1_000))}s`;
  if (ms < 3_600_000) return `${Math.round(ms / 60_000)}m`;
  if (ms < 86_400_000) return `${Math.round(ms / 3_600_000)}h`;
  return `${Math.round(ms / 86_400_000)}d`;
}

/**
 * The default transport: the same bot token and the same `sendMessage` a tier-2
 * page uses. A missing chat id or token is treated as a known failure rather
 * than a permanent one, because both are fixable while the daemon runs — the
 * operator sets `escalation.telegramChatId`, or installs omp-telegram, and the
 * next retry lands. Burning the attempt budget is how they find out: `status`
 * names the reason on the row, and the exhausted report pages tier 2.
 */
export function telegramReportSend(p: ProjectConfig): ReportSend {
  return async (text: string): Promise<number | undefined> => {
    // Both of these are `definitive` because nothing left the host: there is no
    // message anywhere, so the retry is a first attempt and must not be
    // announced to the operator as a possible duplicate.
    const chatId = p.escalation.telegramChatId;
    if (chatId === undefined || chatId === "") {
      throw new TelegramSendError(
        `no report transport for project "${p.name}": set escalation.telegramChatId`,
        "definitive",
      );
    }
    const token = readTelegramToken();
    if (token === undefined) {
      throw new TelegramSendError(
        "no Telegram bot token readable — install and configure omp-telegram, or set OMP_TELEGRAM_STATE_DIR",
        "definitive",
      );
    }
    // `sendTelegram` splits internally and resolves with every part's id. A
    // caller on this seam always passes one message under the wire limit, so a
    // single-element array comes back; `[0]` is that one id. A text over the
    // limit would still be delivered whole (never truncated) — the seam would
    // simply under-report the extra ids, which is why the split lives at the
    // caller, not here.
    const ids = await sendTelegram(token, chatId, text, {
      topicId: resolveProjectTopicId(p),
      project: { name: p.name, workspaceRoot: p.workspaceRoot },
    });
    return ids[0];
  };
}

/**
 * One direct message to the operator, addressed from config rather than from
 * whichever chat last spoke to the session (#366).
 *
 * `telegram_send` keeps the active forum topic only while it names no chat, and
 * a locally injected tick has no active message to inherit one from — so a
 * fleet whose Telegram is a topic had no path from an autonomous turn into its
 * own thread, and answers surfaced in the main chat instead. This addresses the
 * project's configured chat and topic, and it is deliberately *not* a way past
 * the operator's interrupt policy: the same availability decision an autonomous
 * Telegram tool call gets is applied here, and a deferred message is durably
 * held for the digest or the working-hours catch-up rather than sent anyway.
 *
 * The `QUESTION:` marker is the floor's own spelling for "this needs an
 * answer", so one prefix means one category whichever surface it leaves by.
 */
export type OperatorMessageOutcome =
  | { kind: "sent"; category: InterruptCategory }
  | { kind: "held"; category: InterruptCategory; noticeId: string; reason: "availability" | "digest" };

export function operatorMessageCategory(text: string): InterruptCategory {
  return /^\s*QUESTION:\s/i.test(text) ? "decision-needed" : "material";
}

export async function deliverOperatorMessage(
  project: ProjectConfig,
  text: string,
  deps: {
    store: Store;
    at: number;
    noticeId: string;
    send?: ReportSend;
    /** Declared escalation category; overrides the `QUESTION:`-prefix read. */
    category?: InterruptCategory;
  },
): Promise<OperatorMessageOutcome> {
  const category = deps.category ?? operatorMessageCategory(text);
  const disposition = interruptDisposition(project.reporting, category, deps.at);
  if (disposition !== "interrupt") {
    deps.store.addHeldNotice({
      id: deps.noticeId,
      project: project.name,
      category,
      summary: text.split("\n", 1)[0]!.slice(0, 240),
      detail: text,
      createdAt: deps.at,
      ...(disposition === "availability" ? { releaseOnAvailable: true } : {}),
    });
    return { kind: "held", category, noticeId: deps.noticeId, reason: disposition };
  }
  const send = deps.send ?? telegramReportSend(project);
  // A long operator message is split and driven part-by-part through the same
  // seam the outbox uses, so a custom transport sees the same per-part calls
  // and the default one never truncates (#566).
  for (const part of telegramTextParts(text)) {
    await send(part);
  }
  return { kind: "sent", category };
}

/** Keep the mechanical catch-up comfortably inside Telegram's report wrapper. */
const AVAILABILITY_REPORT_BODY_LIMIT = 3_200;
const AVAILABILITY_REPORT_KEY_PREFIX = "availability/";

interface AvailabilityReportMarker {
  categories: InterruptCategory[];
  urgent: boolean;
}

function availabilityReportMarker(report: ReportRecord): AvailabilityReportMarker | undefined {
  const key = report.dedupeKey;
  if (key === undefined || !key.startsWith(AVAILABILITY_REPORT_KEY_PREFIX)) return undefined;
  const parts = key.slice(AVAILABILITY_REPORT_KEY_PREFIX.length).split("/");
  const urgent = parts[0] === "urgent";
  const encoded = (urgent ? parts[1] : parts[0]) ?? "";
  return {
    urgent,
    categories: encoded
      .split(",")
      .filter((category): category is InterruptCategory =>
        INTERRUPT_CATEGORIES.includes(category as InterruptCategory),
      ),
  };
}

function availabilityNoticeLine(notice: HeldNotice): string {
  const flat = (text: string, limit: number): string => {
    const value = text.replace(/\s+/g, " ").trim();
    return value.length <= limit ? value : `${value.slice(0, limit - 1)}…`;
  };
  return (
    `- [${notice.category}] ${flat(notice.summary, 180)} ` +
    `(${new Date(notice.createdAt).toISOString()}) — ${flat(notice.detail, 240)}`
  );
}

/**
 * Hand availability-held notices to the existing durable outbox when the live
 * policy permits them again. Association and report creation are one SQLite
 * transaction, so daemon downtime or a send failure cannot lose a notice.
 */
export function enqueueAvailableHeldNotices(
  project: ProjectConfig,
  store: Store,
  now: number,
): ReportEnqueue | undefined {
  // The Store arbitrates the due-digest lease and this catch-up in the same
  // SQLite transaction. A preflight here would reopen the snapshot race.
  const categories = INTERRUPT_CATEGORIES.filter(
    (category) => interruptDisposition(project.reporting, category, now) === "interrupt",
  );
  const urgentCategories = INTERRUPT_CATEGORIES.filter(
    (category) =>
      availabilityDisposition(project.reporting?.availability, category, now) === "interrupt",
  );
  const urgent = store.undigestedNotices(
    project.name,
    DIGEST_BACKLOG_LIMIT,
    true,
    urgentCategories,
    true,
  );
  const eligible =
    urgent.length > 0
      ? urgent
      : store.undigestedNotices(
          project.name,
          DIGEST_BACKLOG_LIMIT,
          true,
          categories,
          false,
        );
  if (eligible.length === 0) return undefined;

  const lines = [
    "Working-hours catch-up",
    "These interruptions were held durably while the operator was outside the configured availability window:",
  ];
  const selected: HeldNotice[] = [];
  for (const notice of eligible) {
    const line = availabilityNoticeLine(notice);
    const next = [...lines, line].join("\n");
    if (selected.length > 0 && next.length > AVAILABILITY_REPORT_BODY_LIMIT) break;
    lines.push(line);
    selected.push(notice);
  }

  const selectedCategories = INTERRUPT_CATEGORIES.filter((category) =>
    selected.some((notice) => notice.category === category),
  );
  const urgentMarker = selected[0]?.urgent === true ? "urgent/" : "";
  return store.enqueueAvailabilityReport(
    {
      project: project.name,
      kind: "digest",
      body: lines.join("\n"),
      dedupeKey: `${AVAILABILITY_REPORT_KEY_PREFIX}${urgentMarker}${selectedCategories.join(",")}/${selected[0]!.id}`,
      at: now,
    },
    selected.map((notice) => notice.id),
  );
}

export function createReportOutbox(deps: ReportOutboxDeps): ReportOutbox {
  const { store } = deps;
  const currentProject =
    typeof deps.project === "function" ? deps.project : (): ProjectConfig => deps.project as ProjectConfig;
  const now = deps.now ?? Date.now;
  const injectedSend = deps.send;
  const log = deps.log ?? ((): void => {});

  /**
   * A report that exhausted its retries is itself news, so it pages tier 2 —
   * the explicit answer to the question #123 left open. The obvious objection is
   * that this page rides the transport that just failed six times, and that is
   * true and accepted: tier 2 has an issue-comment fallback only for
   * issue-scoped events, and a report has no issue, so a total Telegram outage
   * degrades this to a log line and the `reports` block in `status`. It is still
   * worth sending. The common failure is not "Telegram is down" but "this chat
   * id is wrong" or "the bot was kicked from this chat", and in both of those
   * the page reaches an operator who is otherwise being told nothing at all.
   *
   * It goes through the ordinary escalator, so the `notifications` ledger
   * deduplicates it: the summary carries the report id, which is unique, so one
   * undeliverable report pages exactly once and never every five minutes.
   */
  const pageUndeliverable = async (r: ReportRecord, error: string): Promise<void> => {
    const project = currentProject();
    if (deps.escalate === undefined) return;
    try {
      await deps.escalate({
        tier: 2,
        project: project.name,
        issue: NO_ISSUE,
        summary: `Report ${r.id} could not be delivered after ${r.attempts} attempts — ${project.name} is reporting into a void`,
        detail: [
          `Kind: ${r.kind}. Handed over ${new Date(r.createdAt).toISOString()}.`,
          `Last error: ${error}`,
          "",
          "The report itself is still in the outbox and is shown by `omp-conductor status`.",
          "Nothing retries it again: fix the transport, then treat the status block as the backlog.",
          "",
          r.body,
        ].join("\n"),
      });
    } catch (err) {
      log(
        `report ${r.id} is undeliverable and its tier-2 page failed too ` +
          `(${err instanceof Error ? err.message : String(err)}) — ` +
          `it stays in \`status\` as the only surface left`,
      );
    }
  };

  const giveUp = async (
    claimed: ReportRecord,
    attemptId: string,
    error: string,
    pass: ReportDeliveryPass,
  ): Promise<void> => {
    store.markReportFailed(claimed.id, attemptId, error, now());
    pass.failed.push(claimed.id);
    log(`report ${claimed.id} failed after ${claimed.attempts} attempts: ${error}`);
    await pageUndeliverable({ ...claimed, state: "failed" }, error);
  };

  const attempt = async (r: ReportRecord, pass: ReportDeliveryPass): Promise<void> => {
    const project = currentProject();
    if (deps.deliveryAllowed?.() === false) {
      pass.deferred.push(r.id);
      return;
    }
    const availabilityMarker = availabilityReportMarker(r);
    if (availabilityMarker !== undefined) {
      if (availabilityMarker.categories.length === 0) {
        pass.deferred.push(r.id);
        log(`availability catch-up ${r.id} has no valid category marker and was held fail-closed`);
        return;
      }
      const dispositions = availabilityMarker.categories.map((category) =>
        availabilityMarker.urgent
          ? availabilityDisposition(project.reporting?.availability, category, now())
          : interruptDisposition(project.reporting, category, now()),
      );
      if (dispositions.some((disposition) => disposition !== "interrupt")) {
        if (store.releasePendingAvailabilityReport(r.id, project.name, r.ambiguous)) {
          const reason = dispositions.includes("digest")
            ? "reporting policy changed"
            : "the availability window closed";
          log(`availability catch-up ${r.id} returned to the digest because ${reason}`);
        }
        pass.deferred.push(r.id);
        return;
      }
    }
    if (r.kind === "material") {
      const at = now();
      const disposition = interruptDisposition(project.reporting, "material", at);
      if (disposition !== "interrupt") {
        const firstLine = r.body.split("\n", 1)[0]!;
        const deferred = store.deferPendingReportToNotice(r.id, {
          id: r.id,
          project: project.name,
          category: "material",
          summary: (
            r.ambiguous ? `POSSIBLE REPEAT of report ${r.id}: ${firstLine}` : firstLine
          ).slice(0, 240),
          detail: r.ambiguous
            ? `This report may already have reached Telegram before its outcome was lost.\n\n${r.body}`
            : r.body,
          createdAt: at,
          ...(disposition === "availability" ? { releaseOnAvailable: true } : {}),
        });
        if (deferred) {
          pass.deferred.push(r.id);
          log(
            `report ${r.id} preserved for ${
              disposition === "availability" ? "the next availability window" : "a digest"
            } after reporting policy changed`,
          );
          return;
        }
      }
    }
    const send = injectedSend ?? telegramReportSend(project);
    const attemptId = crypto.randomUUID();
    // Losing this race is ordinary: another pass, or another daemon, already
    // owns the attempt. Returning is what keeps one report from being in flight
    // twice at once — the achievable half of "never double-posts", and the half
    // that actually prevents a retry storm.
    const claimed = store.claimReport(r.id, attemptId, now());
    if (claimed === undefined) return;

    // A report whose attempts all ended *unknown* never produces a definitive
    // failure to trip the budget below, so without this it would cycle through
    // stale recovery forever, posting a possible duplicate every window. A
    // budget nothing can exhaust is not a budget.
    if (claimed.attempts > REPORT_MAX_ATTEMPTS) {
      await giveUp(
        claimed,
        attemptId,
        claimed.lastError ?? "every attempt ended without an answer from Telegram",
        pass,
      );
      return;
    }

    const text = formatReportMessage(claimed, project.name);
    const parts = telegramTextParts(text);
    // Fingerprint of the exact text the split below is computed from. The
    // per-part watermark on the row is only meaningful while this is unchanged:
    // the resume point is a count into a specific split, so any drift in the
    // text invalidates it (#566).
    const messageHash = createHash("sha256").update(text).digest("hex");
    // Resume at the first part this attempt has not yet confirmed. A count is
    // only a valid resume point while the text — and therefore the split — is
    // byte-identical to the attempt that wrote the watermark. Any change,
    // notably the possible-repeat banner an unknown outcome adds to an
    // ambiguous report, reshapes the parts, so the watermark is deliberately
    // not carried across it and the whole report is re-sent amber-flagged —
    // the same at-least-once semantics as the single-message path (#566).
    const resumeAt = claimed.sentPartsHash === messageHash ? (claimed.sentParts ?? 0) : 0;
    const start = Math.min(resumeAt, parts.length);

    const messageIds: number[] = [];
    try {
      // Strictly in order, one `sendMessage` at a time: parts never interleave
      // with each other or with a message sent concurrently to the same chat.
      for (let i = start; i < parts.length; i += 1) {
        const id = await send(parts[i]!);
        if (id !== undefined) messageIds.push(id);
        // Persist progress before the next part ships. Only a multi-part send
        // has a crossing point to protect: a crash or a definitive failure can
        // then only ever re-send the part that was in flight, never the
        // confirmed prefix. Guarded by attempt id like every other transition.
        if (parts.length > 1) {
          store.markReportPartsSent(claimed.id, attemptId, i + 1, messageHash, now());
        }
      }
    } catch (err) {
      const at = now();
      const error = err instanceof Error ? err.message : String(err);

      /**
       * Which of the two failures this was is decided at the transport, by the
       * code that watched the socket — never here, and never by inspecting the
       * message text. {@link sendTelegram} throws `TelegramSendError` carrying
       * the verdict; anything else reaching this catch is a `ReportSend` that
       * did not classify, and is treated as `unknown` on purpose.
       *
       * Do not "simplify" this into the requeue branch below. Every throw is
       * not a known failure: an aborted request, a socket reset, or a body read
       * that failed after a 200 all leave Telegram possibly holding the
       * message. Requeuing those as `pending` would drop the possible-repeat
       * marker and post a second copy of a delivered report with nothing
       * anywhere saying it might be one — the precise dishonesty #123's
       * correction exists to forbid.
       */
      const outcome = err instanceof TelegramSendError ? err.outcome : "unknown";
      if (outcome === "unknown") {
        // Stays `sending`: the row's own state is the honest record that nobody
        // knows. Stale recovery hands it back when the window closes, and every
        // message from then on says it may be a repeat.
        store.markReportUncertain(claimed.id, attemptId, error);
        pass.uncertain.push(claimed.id);
        log(
          `report ${claimed.id} attempt ${claimed.attempts} left this host and never came back ` +
            `with an answer (${error}) — held as sending; the retry will say it may be a repeat`,
        );
        return;
      }

      if (claimed.attempts >= REPORT_MAX_ATTEMPTS) {
        await giveUp(claimed, attemptId, error, pass);
        return;
      }
      const delay = reportBackoffMs(claimed.attempts);
      store.markReportPending(claimed.id, attemptId, at + delay, error, at);
      pass.requeued.push(claimed.id);
      log(
        `report ${claimed.id} attempt ${claimed.attempts}/${REPORT_MAX_ATTEMPTS} was rejected, ` +
          `${start + messageIds.length}/${parts.length} parts accepted, ` +
          `retrying in ${Math.round(delay / 1_000)}s: ${error}`,
      );
      return;
    }

    store.markReportDelivered(claimed.id, attemptId, messageIds, now());
    pass.delivered.push(claimed.id);
    log(
      `report ${claimed.id} delivered` +
        (parts.length === 1
          ? messageIds[0] === undefined
            ? ""
            : ` as telegram message ${messageIds[0]}`
          : ` as ${parts.length} telegram messages (${messageIds.join(", ")})`) +
        `${claimed.ambiguous ? " (marked as a possible repeat)" : ""}`,
    );
  };

  return {
    recover(staleAt: number): ReportRecord[] {
      return store.recoverSendingReports(currentProject().name, staleAt, now());
    },

    async deliverDue(): Promise<ReportDeliveryPass> {
      const project = currentProject();
      const pass: ReportDeliveryPass = {
        delivered: [],
        deferred: [],
        requeued: [],
        uncertain: [],
        failed: [],
        recovered: [],
      };

      // Before picking up new work: a row left `sending` past the stale window
      // belongs to a process that is gone, or to a request that will never
      // answer. Either way nobody is going to resolve it but us.
      for (const recovered of store.recoverSendingReports(
        project.name,
        now() - SENDING_STALE_MS,
        now(),
      )) {
        pass.recovered.push(recovered.id);
        log(
          `report ${recovered.id} was left mid-send and its outcome is unknown — ` +
            `retrying; the message will say it may be a repeat`,
        );
      }

      // Sequential, not parallel. Telegram rate-limits a bot per chat, and the
      // operator reads these in the order they arrive — a burst that lands out
      // of order costs more to read than it saves to send.
      for (const r of store.dueReports(project.name, now(), REPORT_BATCH)) {
        await attempt(r, pass);
      }
      return pass;
    },
  };
}