/**
 * Escalation: the one path by which a stuck run reaches someone who can unstick
 * it. Tier 1 is the orchestrator session's problem, tier 2 is the human's, and
 * an issue comment is what is left when neither transport is reachable.
 *
 * Two rules shape everything here:
 *
 *  1. An escalation is never silently dropped. Either a transport delivered it,
 *     or `escalate()` throws so the dispatcher learns nobody is reachable. A
 *     swallowed escalation looks exactly like a healthy fleet. Tier 1 is where
 *     that is hardest: the orchestrator *accepts* an injection minutes before
 *     its turn settles, so the dedup marker waits for the settlement and a
 *     late failure falls back to an issue comment for that same escalation.
 *  2. An escalation is never repeated. The dispatcher re-notices the same
 *     unroutable issue on every poll, so the store's notification ledger — not
 *     the loop — is what stops a human being paged every five minutes.
 *
 * The bot token is read at send time and never travels further than the request
 * URL: it is redacted out of every error string this module can produce, since
 * those strings end up in daemon logs and, on the fallback path, in a public
 * issue comment.
 */
import { mkdirSync, readFileSync, renameSync, statSync, writeFileSync } from "node:fs";
import { homedir } from "node:os";
import { dirname, join } from "node:path";

import { availabilityOpen, interruptDisposition, type InterruptDisposition } from "./availability.ts";
import { stateDir } from "./config.ts";
import { heldNoticeId } from "./notices.ts";
import type { OrchestratorHandle } from "./orchestrator.ts";
import type { Escalation, ProjectConfig, Store, Tracker } from "./types.ts";

/**
 * Telegram rejects `sendMessage` over 4096 characters. Longer text is split
 * into labelled parts ({@link telegramTextParts}) instead of truncated, so the
 * tail of a digest or a tier-2 escalation arrives instead of being silently
 * dropped (#566).
 */
export const TELEGRAM_TEXT_LIMIT = 4096;

/**
 * Width reserved for the `(i/n)` label prepended to every part of a split
 * message, ported from omp-telegram's `PART_LABEL_RESERVE`. Parts are re-split
 * against `TELEGRAM_TEXT_LIMIT - TELEGRAM_PART_LABEL_RESERVE` so the label can
 * never push a part past the wire limit.
 */
export const TELEGRAM_PART_LABEL_RESERVE = 16;

/**
 * Split text into Telegram-sized parts. Ports omp-telegram 0.12.1's
 * `chunkLabeled` semantics (the plugin this transport bypasses, #566) so a
 * digest copies the same readability contract that plugin ships:
 *
 * - newline-preferred boundaries: a cut prefers the last paragraph break
 *   (`\n\n`), then line break, then the last space, each only when it lands in
 *   the second half of the window so parts stay large; an unsplittable run is
 *   hard-cut rather than dropped.
 * - numbered labels: when a message takes more than one part, every part is
 *   prefixed `(i/N)\n` so a reader sees the message continues and knows how
 *   many parts to expect. Under the label budget, so no part exceeds the limit.
 * - lossless: a split message reassembles byte-for-byte to its input
 *   (strip the labels and concatenate). Deliberately *not* ported from
 *   upstream: fence repair and leading-newline stripping are rendering
 *   flourishes for markdown-mode sends, and this transport sends plain text
 *   (`parse_mode` is never set) — both would rewrite bytes the operator never
 *   wrote.
 *
 * Text at or under the limit is returned whole and untouched, so the
 * single-message path keeps sending byte-identical payloads.
 */
export function telegramTextParts(text: string, limit: number = TELEGRAM_TEXT_LIMIT): string[] {
  if (text.length === 0 || text.length <= limit) return [text];
  // `splitToLimit` returns at least two pieces here (text is over the limit);
  // the guard is for future-proofing against a limit rewrite.
  const parts = splitToLimit(text, Math.max(1, limit));
  if (parts.length === 1) return parts;
  // Re-split under the label budget so `(i/N)\n` never pushes a part past the
  // wire limit.
  const labelled = splitToLimit(text, Math.max(1, limit - TELEGRAM_PART_LABEL_RESERVE));
  return labelled.map((part, index) => `(${index + 1}/${labelled.length})\n${part}`);
}

/**
 * The base splitter, ported from omp-telegram's `splitToLimit`. Prefers a
 * paragraph break, then a line break, then a space — each only when it sits in
 * the second half of the window, so a digest splits on natural boundaries
 * rather than mid-word. Never drops a byte: the fallback is a hard cut at
 * `limit`, which split a run with no whitespace in the window in the middle,
 * and the parts still reassemble to the input.
 */
function splitToLimit(text: string, limit: number): string[] {
  if (text.length <= limit) return [text];
  const out: string[] = [];
  let rest = text;
  while (rest.length > limit) {
    const para = rest.lastIndexOf("\n\n", limit);
    const line = rest.lastIndexOf("\n", limit);
    const space = rest.lastIndexOf(" ", limit);
    const cut = para > limit / 2 ? para : line > limit / 2 ? line : space > 0 ? space : limit;
    out.push(rest.slice(0, cut));
    rest = rest.slice(cut);
  }
  out.push(rest);
  return out;
}

/**
 * Whether a failed send settles the question of delivery. This distinction is
 * the whole of #123's honesty: both classes retry, but only one of them may be
 * producing a *second* copy of a message Telegram already accepted, and only
 * that one is allowed to say so.
 *
 * - `definitive` — Telegram answered and did not accept the message, or the
 *   request provably never left this host. Nobody has it. A retry is a first
 *   attempt, not a possible duplicate.
 * - `unknown` — the request left and its fate was never learned. Telegram may
 *   be holding the message. A retry may duplicate, and must carry that warning.
 */
export type TelegramSendOutcome = "definitive" | "unknown";

/** A send failure that knows which of the two it is. */
export class TelegramSendError extends Error {
  readonly outcome: TelegramSendOutcome;
  constructor(message: string, outcome: TelegramSendOutcome) {
    super(message);
    this.name = "TelegramSendError";
    this.outcome = outcome;
  }
}

/**
 * Error codes that prove no request was ever put on the wire. Bun reports both
 * a refused connection and a DNS failure as `ConnectionRefused`; the `E*` codes
 * are what Node/undici produce, kept so this classification does not silently
 * invert if the runtime underneath ever changes.
 */
const NEVER_CONNECTED: ReadonlySet<string> = new Set([
  "ConnectionRefused",
  "FailedToOpenSocket",
  "ECONNREFUSED",
  "ENOTFOUND",
  "EAI_AGAIN",
  "EHOSTUNREACH",
  "ENETUNREACH",
]);

/**
 * Classifies a `fetch` rejection, defaulting to `unknown`.
 *
 * The default is the entire point and must not be "tidied" into the simpler
 * branch: an aborted request, a socket reset or an `EPIPE` all happen *after*
 * the bytes went out, and Telegram may well have accepted the message before
 * the connection died. Guessing `definitive` there re-sends it as though
 * nothing had happened, which is how a delivered page is posted twice with
 * nothing saying so. Guessing `unknown` costs a possible-repeat marker on a
 * message that was never sent — a cosmetic wrong, traded against a silent
 * duplicate. When in doubt, doubt out loud.
 */
export function telegramFailureOutcome(cause: unknown): TelegramSendOutcome {
  let error: unknown = cause;
  for (let depth = 0; error instanceof Error && depth < 4; depth += 1) {
    const code: unknown = (error as { code?: unknown }).code;
    if (typeof code === "string" && NEVER_CONNECTED.has(code)) return "definitive";
    error = error.cause;
  }
  return "unknown";
}

export interface Escalator {
  escalate(e: Escalation): Promise<void>;
}

/**
 * The human-facing text, shared by both transports so a Telegram ping and its
 * issue-comment fallback are the same message — a human comparing the two
 * should never have to wonder whether they describe the same event.
 *
 * Deliberately plain: Telegram is called without `parse_mode`, because an issue
 * title containing `_` or `*` would otherwise make Telegram reject the whole
 * send, turning a cosmetic problem into a lost escalation. Plain text is also
 * valid Markdown, so the same string renders fine as an issue comment.
 */
/**
 * Fleet-scoped pages (integrity tripwire, spend-cap halt, stall) use issue `0`
 * as a sentinel — there is no tracker issue. Rendering that as `#0` made
 * Telegram pages and daemon logs look like a bug (#52). Keep the sentinel in
 * the typed field; only the human-facing label changes here.
 */
export function escalationIssueRef(issue: number): string {
  return issue === 0 ? "fleet" : `#${issue}`;
}

export function formatEscalation(e: Escalation, project: string): string {
  const lines = [
    `omp-conductor · tier ${e.tier} escalation`,
    `project: ${project}`,
    `issue: ${escalationIssueRef(e.issue)}`,
    `summary: ${e.summary}`,
  ];
  if (e.detail) lines.push(`detail: ${e.detail}`);
  if (e.runId) lines.push(`run: ${e.runId}`);
  return lines.join("\n");
}

/**
 * `orchestrator` is the tier-1 transport when one is running: a tier-1
 * escalation is the orchestrator's problem, not the human's, and an injected
 * prompt is the only form of it that can actually change anything. It stays
 * optional so a daemon whose orchestrator failed to start — and the unit suite —
 * still escalate, just to an issue comment.
 */
export function createEscalator(
  source: ProjectConfig | (() => ProjectConfig),
  tracker: Tracker,
  store: Store,
  orchestrator?: OrchestratorHandle,
  now: () => number = Date.now,
  deliveryAllowed: () => boolean = () => true,
  onTier1Diverted?: (e: Escalation) => void,
): Escalator {
  const currentProject = typeof source === "function" ? source : (): ProjectConfig => source;
  return {
    async escalate(e: Escalation): Promise<void> {
      // The daemon's provider resolves the config reloaded at the latest tick
      // boundary. Capture one snapshot for this delivery, including late
      // settlement callbacks, so a mid-call edit cannot split its policy.
      const p = currentProject();
      // Stable across daemon restarts: same project, issue, tier and summary is
      // the same event, however many times the loop rediscovers it.
      const key = `${p.name}:${e.issue}:${e.tier}:${e.summary}`;
      if (store.wasNotified(key)) return;

      const text = formatEscalation(e, p.name);
      const chatId = p.escalation.telegramChatId;

      if (e.tier === 1 && orchestrator) {
        // Resolves on acceptance, not on an answer, so this does not park the
        // tick behind a model. A rejection means the injection was never taken
        // at all: `undefined` falls through to the human-facing path below
        // rather than losing the escalation — a tier-1 event that reached
        // nobody is indistinguishable from a healthy fleet.
        const receipt = await orchestrator.deliver(e, p.name).catch(() => undefined);
        if (receipt) {
          /**
           * This escalation's own late failure, routed to this escalation's own
           * fallback. The key stays unmarked unless something actually lands:
           * an unmarked key means the next tick re-escalates, which is the
           * whole difference between a late escalation and a lost one. Marking
           * on acceptance is what used to drop it — "notified" written over an
           * event no human ever read, and nothing left to retry it.
           */
          const onTurnFailed = async (cause: unknown): Promise<void> => {
            if (!p.escalation.fallbackToIssueComment) {
              warn(
                `tier 1 escalation on ${escalationIssueRef(e.issue)} was accepted by the orchestrator but its ` +
                  `turn failed (${errText(cause)}), and no fallback is configured — left unmarked ` +
                  `so the next tick retries`,
              );
              return;
            }
            try {
              if (e.issue === 0) {
                throw new Error("fleet-scoped escalation has no issue to comment on");
              }
              await tracker.comment(e.issue, text);
              store.markNotified(key);
            } catch (err) {
              warn(
                `tier 1 escalation on ${escalationIssueRef(e.issue)} failed after acceptance ` +
                  `(${errText(cause)}) and its issue-comment fallback failed too ` +
                  `(${errText(err)}) — left unmarked so the next tick retries`,
              );
            }
          };

          // Acceptance is not delivery: the orchestrator's turn for *this*
          // injection can still fail minutes from now, so the marker waits for
          // `settled`. Deliberately not awaited — `settled` resolves only when
          // the model has finished answering, which is exactly the wait the
          // dispatcher tick must not take.
          void receipt.settled
            .then(() => {
              store.markNotified(key);
            }, onTurnFailed)
            .catch(() => {
              // `markNotified` is the only thing above that can still throw,
              // and a failing store write must not become an unhandled
              // rejection in a daemon that has nobody left to throw at.
            });
          return;
        }
      }

      if (e.tier === 2) {
        const category = e.category ?? "tier2";
        const at = now();
        let disposition: InterruptDisposition;
        if (!deliveryAllowed()) {
          disposition = "availability";
        } else if (e.urgent) {
          // Urgency may bypass category batching when the digest loop itself is
          // broken (#246), but it cannot invent an out-of-hours bypass the
          // operator did not configure (#273).
          const window = p.reporting?.availability;
          disposition =
            window !== undefined &&
            !availabilityOpen(window, at) &&
            !window.bypass.includes(category)
              ? "availability"
              : "interrupt";
        } else {
          disposition = interruptDisposition(p.reporting, category, at);
        }
        if (disposition !== "interrupt") {
          store.addHeldNotice({
            id: heldNoticeId(key),
            project: p.name,
            category,
            summary: e.summary,
            detail: text,
            createdAt: at,
            ...(disposition === "availability" ? { releaseOnAvailable: true } : {}),
            ...(e.urgent === true ? { urgent: true } : {}),
          });
          store.markNotified(key);
          return;
        }
      }

      if (e.tier === 2 && chatId) {
        const token = readTelegramToken();
        if (token) {
          // A send failure throws: `markNotified` stays uncalled so the next
          // poll retries instead of writing the event off as delivered. No
          // backoff in here — the dispatcher tick *is* the retry, and an
          // escalation the loop keeps rediscovering is re-derived from the
          // world rather than replayed from a queue.
          //
          // #123 built the durable outbox this comment used to name as the
          // upgrade path, and deliberately did not move escalations into it:
          // the condition that raises one is still true on the next tick, so
          // the tick re-raises it. A *report* has no such source — it exists
          // once, in the model's head, and nothing regenerates it — which is
          // why that one needed a ledger and this one does not.
          await sendTelegram(token, chatId, text, {
            topicId: resolveProjectTopicId(p),
            project: { name: p.name, workspaceRoot: p.workspaceRoot },
          });
          store.markNotified(key);
          return;
        }
      }

      // Tier 1 lands here with no orchestrator, or with one that would not take
      // the injection; tier 2 lands here when omp-telegram is not installed or
      // the project never configured a chat id.
      //
      // A tier-1 event reaching this point was diverted from the orchestrator to
      // an issue comment. That is the exact condition the down incident counts:
      // the daemon hooks this to accumulate the orchestrator-down incident's
      // diverted tally (a reader that owns its own dedupe and no-ops for a
      // healthy or external orchestrator).
      if (e.tier === 1) onTier1Diverted?.(e);
      if (!p.escalation.fallbackToIssueComment) {
        throw new Error(
          `no escalation transport configured for project "${p.name}": tier ${e.tier} ` +
            `escalation on ${escalationIssueRef(e.issue)} (${e.summary}) could not be delivered — ` +
            `set escalation.telegramChatId or escalation.fallbackToIssueComment`,
        );
      }
      // Fleet pages (issue 0) have nowhere to comment. Falling through here used
      // to call `issues/0/comments` and surface as a confusing "#0" delivery
      // failure after Telegram had already been tried (#52).
      if (e.issue === 0) {
        throw new Error(
          `fleet-scoped tier ${e.tier} escalation (${e.summary}) has no issue to comment on — ` +
            `configure escalation.telegramChatId so integrity/spend/stall pages can reach an operator`,
        );
      }
      await tracker.comment(e.issue, text);
      store.markNotified(key);
    },
  };
}

/**
 * The tier-1 settlement tail runs after `escalate()` has already returned, so a
 * failure there has no caller left to throw at. stderr is where it can still be
 * seen: the daemon's log *is* its stderr, so these land in `daemon.log` beside
 * every other conductor line.
 *
 * ponytail: duplicates daemon.ts's one-line format rather than sharing a
 * logger. Upgrade path is a `log.ts` both modules import.
 */
function warn(msg: string): void {
  process.stderr.write(`[conductor ${new Date().toISOString()}] ${msg}\n`);
}

function errText(e: unknown): string {
  return e instanceof Error ? e.message : String(e);
}

/**
 * omp-telegram owns `<state dir>/.env`; conductor only borrows the token, so a
 * user already running that bot gets tier-2 pings with no extra configuration.
 * Absence is not an error — it just means tier 2 degrades to the fallback.
 *
 * Exported for the report outbox (#123), which pages over the same bot and must
 * resolve the token the same way. Two readers of one `.env` is fine; two
 * *implementations* of the parse below is how they drift. The state dir is the
 * shared resolution both the token read and the interactive ask surface (#722)
 * need, so it is one function rather than a second copy of the override.
 */
export function telegramStateDir(): string {
  const override = process.env.OMP_TELEGRAM_STATE_DIR?.trim();
  return override ? override : join(homedir(), ".omp", "agent", "telegram");
}

export function readTelegramToken(): string | undefined {
  const dir = telegramStateDir();
  let raw: string;
  try {
    raw = readFileSync(join(dir, ".env"), "utf8");
  } catch {
    return undefined;
  }
  for (const line of raw.split("\n")) {
    const trimmed = line.trim();
    if (!trimmed || trimmed.startsWith("#")) continue;
    const match = /^(?:export\s+)?TELEGRAM_BOT_TOKEN\s*=\s*(.*)$/.exec(trimmed);
    if (!match) continue;
    let value = (match[1] ?? "").trim();
    const quote = value[0];
    if (value.length >= 2 && (quote === '"' || quote === "'") && value.endsWith(quote)) {
      value = value.slice(1, -1);
    }
    if (value) return value;
  }
  return undefined;
}

/**
 * One flat-chat delivery that happened only because the project's pinned
 * topic was unreachable (#1094).
 *
 * The fallback is a delivery, not a success: the message reached an audience
 * that did not ask for it while the topic the project reads stayed empty, and
 * the only record used to be a journal `warn`. This row is what makes the
 * mis-route visible where the operator looks: `doctor`'s topic-pin finding
 * and the project's `status` render the rows matching the *current* pin, so
 * re-pinning is what retires them. Nothing deletes rows — the file is bounded
 * and a retired row simply stops matching.
 */
export interface TelegramMisroute {
  /** The project whose message rode the flat chat. */
  project: string;
  /** The pinned `escalation.telegramTopicId` Telegram refused. */
  staleTopicId: number;
  /** When the flat delivery happened (epoch ms). */
  at: number;
}

const TELEGRAM_MISROUTES_FILE = "telegram-misroutes.json";
/** Diagnostic history, not a ledger: enough to see "again?", never unbounded. */
const TELEGRAM_MISROUTES_MAX = 20;

/**
 * Records one flat delivery beside the conductor state. Best-effort by
 * contract: the delivery has already happened, so a failed breadcrumb must
 * never turn a completed escalation into a thrown one.
 */
export function recordTelegramMisroute(projectName: string, staleTopicId: number, at: number = Date.now()): void {
  try {
    const next = [
      ...readTelegramMisroutes(),
      { project: projectName, staleTopicId, at },
    ].slice(-TELEGRAM_MISROUTES_MAX);
    const path = join(stateDir(), TELEGRAM_MISROUTES_FILE);
    mkdirSync(stateDir(), { recursive: true });
    const tmp = `${path}.${process.pid}.tmp`;
    writeFileSync(tmp, `${JSON.stringify(next, null, 2)}\n`);
    renameSync(tmp, path);
  } catch {
    // The journal warn still named the degradation; doctor still names the
    // stale pin. Losing the row costs history, never delivery.
  }
}

/**
 * Every recorded mis-route, oldest first. Unreadable or malformed state reads
 * empty — a broken breadcrumb file must not break the diagnostics that read
 * it, and per-row validation keeps a hand-edited entry from poisoning the
 * render (the same rule #1089 applied to the claim registry's fields).
 */
export function readTelegramMisroutes(): readonly TelegramMisroute[] {
  let raw: string;
  let parsed: unknown;
  try {
    raw = readFileSync(join(stateDir(), TELEGRAM_MISROUTES_FILE), "utf8");
    parsed = JSON.parse(raw);
  } catch {
    return [];
  }
  if (!Array.isArray(parsed)) return [];
  const rows: TelegramMisroute[] = [];
  for (const entry of parsed) {
    if (entry === null || typeof entry !== "object") continue;
    const project = Reflect.get(entry, "project");
    const staleTopicId = Reflect.get(entry, "staleTopicId");
    const at = Reflect.get(entry, "at");
    if (typeof project !== "string" || project.length === 0) continue;
    if (typeof staleTopicId !== "number" || !Number.isSafeInteger(staleTopicId)) continue;
    if (typeof at !== "number" || !Number.isFinite(at)) continue;
    rows.push({ project, staleTopicId, at });
  }
  return rows.sort((a, b) => a.at - b.at);
}

/**
 * The one-line notice prepended to a flat fallback (#1094): a reader in the
 * main chat must know why the message is there and which fleet sent it, not
 * reverse-engineer it from content — and the pinned id is named so the
 * operator can act without opening the journal. Exported because the tests
 * assert its wording and `status`'s misroute row quotes the same condition.
 */
export function staleTopicNotice(topicId: number, projectName?: string): string {
  const who = projectName === undefined ? "This project's" : `[${projectName}] its`;
  return (
    `${who} pinned Telegram topic (escalation.telegramTopicId=${topicId}) is stale — ` +
    `Telegram has no such thread, so this message was delivered to the flat chat instead.`
  );
}

/** One live omp-telegram topic claim, as much of it as conductor reads. */
export interface ClaimedTopic {
  threadId: number;
  /** The bridge's own title for the claim. */
  name: string;
  /**
   * The claiming pane's process id, when the bridge captured one. omp-telegram
   * treats a claim as live only while this pid is alive; a claim that captured
   * no pid is indistinguishable from a dead one for liveness purposes.
   */
  pid?: number;
  /** The herdr space the claiming pane sits in, when the bridge captured one. */
  workspaceLabel?: string;
  /**
   * The claiming pane's working directory, when the bridge captured one. The
   * bridge records it on every entry it writes; conductor reads it as the one
   * discriminator it owns for telling a project's own panes apart when several
   * sit in one herdr space (#1089).
   */
  cwd?: string;
  /**
   * The transcript the claiming pane actually writes, when the bridge captured
   * one. This is the *live* session file — herdr pins a restored pane to it
   * (`omp --resume=<that path>`) — so it can sit in a different session
   * directory than the pane's cwd implies (#600).
   */
  sessionFile?: string;
}

/**
 * The forum topic a project's Telegram actually lives in *now*.
 *
 * `escalation.telegramTopicId` is pinned by the wizard from the topics
 * omp-telegram had claimed at setup time — but the bridge re-claims a pane's
 * topic across restarts, and conductor restarts that service itself during
 * `upgrade` and `restart`. The pinned id therefore goes stale on the fleet's
 * own maintenance, and every topic-addressed send silently degrades to the main
 * chat: tier-2 pages, reports, digests, arm challenges, direct messages (#407).
 *
 * The operator's pin still wins whenever it is live. Only a pin that no *live*
 * claim carries is replaced, and only by the claim this project can be
 * identified with — so a deliberately separate alerts topic is never hijacked
 * by the pane's own thread. A dead row wearing the pin id does not keep it in
 * place: the row is the bridge's past, not a destination (#987). Unavailable
 * bridge state changes nothing, and #318's stale-topic retry remains the last
 * line of defence.
 *
 * Identity is read from the herdr space first, and only then from the claim's
 * title. The bridge titles a topic `ownAgentName ?? basename(cwd)`, and both
 * halves of that fall down on a real multi-project host: the agent-name lookup
 * is best-effort, and a fleet whose panes live under one state directory gives
 * every pane the same basename. `workspaceLabel` is the herdr space, which is
 * one-to-one with the project, so it is the identity that survives a re-claim.
 *
 * The substitution is logged, naming the project and which identity answered.
 * Never an id: a log line is a place these leak from.
 */
export function resolveProjectTopicId(project: ProjectConfig, alive?: (pid: number) => boolean): number | undefined {
  const pinned = project.escalation.telegramTopicId;
  if (pinned === undefined) return undefined;
  const result = claimedTelegramTopics();
  if (result.kind !== "ok" || result.claims.length === 0) return pinned;
  const claims = result.claims;
  // The pin wins only while a *live* claim carries it; a dead pinned row falls
  // through to the live-claim substitution below instead of winning the
  // exact-pin check (#987).
  if (claims.some((claim) => claim.threadId === pinned && claimIsLive(claim, alive === undefined ? undefined : killOf(alive)))) {
    return pinned;
  }
  const match = claimForProject(claims, project.name, alive, project.workspaceRoot);
  if (match === undefined) return pinned;
  warn(
    `escalation.telegramTopicId for ${project.name} is no longer a claimed topic; ` +
      `following omp-telegram's current "${project.name}" ` +
      `${match.workspaceLabel === project.name ? "herdr space claim" : "claim"} instead`,
  );
  return match.threadId;
}

/**
 * The one claim answering to this project — or the reason there is none.
 *
 * `kind: "match"` is the unique claim a substitution can follow. `ambiguous`
 * is the multi-answer case: several claims wearing the project's space (or,
 * when no claim wears a space at all, several wearing its title) means the
 * bridge cannot say which pane belongs to this project. A caller that must
 * not certify a guess — doctor's arm probe, which would have to pick a
 * session file — needs this spelled out, not collapsed into "none".
 */
export type ProjectClaim =
  | { readonly kind: "match"; readonly claim: ClaimedTopic }
  | { readonly kind: "none" }
  | { readonly kind: "ambiguous"; readonly claimants: readonly ClaimedTopic[] };

/**
 * How the live claims answer to a project, identity spellings in order:
 * unique herdr space, then unique title.
 *
 * Only live claims take part ({@link claimIsLive}): a dead row neither
 * answers on its own nor turns this project's one live pane into an
 * ambiguity (#987). Ambiguity is not a coin toss: paging the wrong project's
 * topic is worse than the flat-chat degrade #318 already handles, and
 * scanning the wrong pane's session misses the reply entirely. The title is
 * a fallback for bridges that never captured a space, so it runs only when
 * *no* live claim carries the project's space — several live claims wearing
 * the space is an ambiguity the title cannot resolve, because the titled
 * claim may be a sibling pane, not this project (#626).
 *
 * One same-space ambiguity conductor settles itself (#1089): it provisioned
 * both panes from its own config, so when the caller supplies the project's
 * orchestrator cwd — its workspace root, against the console's
 * `<stateDir>/console/<name>` — exactly one live claimant sitting there is
 * identification, not a guess. Neither or both matching stays the ambiguity.
 */
export function resolveProjectClaim(
  claims: readonly ClaimedTopic[],
  projectName: string,
  alive?: (pid: number) => boolean,
  orchestratorCwd?: string,
): ProjectClaim {
  // Dead claims do not vote: a closed pane's row must not turn this project's
  // one live claim into an ambiguity, and must not answer by title on its own
  // (#987). Ambiguity among genuinely live claims is narrowed by the one
  // discriminator conductor owns (#1089), never removed (#626).
  const live = claims.filter((claim) => claimIsLive(claim, alive === undefined ? undefined : killOf(alive)));
  const bySpace = live.filter((claim) => claim.workspaceLabel === projectName);
  if (bySpace.length > 0) {
    if (bySpace.length === 1) return { kind: "match", claim: bySpace[0]! };
    // Conductor provisions every same-space pane from its own config: the
    // orchestrator pane's cwd is the project's workspace root, the console's
    // is `<stateDir>/console/<name>` (#1089). Exactly one claimant sitting at
    // the orchestrator cwd is identification; neither or both matching stays
    // the coin toss #626 refuses to spin.
    const orchestrator =
      orchestratorCwd === undefined ? [] : bySpace.filter((claim) => claim.cwd === orchestratorCwd);
    return orchestrator.length === 1
      ? { kind: "match", claim: orchestrator[0]! }
      : { kind: "ambiguous", claimants: bySpace };
  }
  const byTitle = live.filter((claim) => claim.name === projectName);
  if (byTitle.length === 0) return { kind: "none" };
  return byTitle.length === 1
    ? { kind: "match", claim: byTitle[0]! }
    : { kind: "ambiguous", claimants: byTitle };
}

/**
 * The unique live claim answering to a project, if exactly one does.
 *
 * `undefined` covers both "no claim answers" and "several answer" — {@link
 * resolveProjectClaim} is the one that tells them apart. Callers that may
 * only fall back (a substitution keeps the pin, a session dir falls back to
 * the tick cwd) read that as "nothing to follow"; callers that must not
 * certify a guess need the richer outcome.
 */
export function claimForProject(
  claims: readonly ClaimedTopic[],
  projectName: string,
  alive?: (pid: number) => boolean,
  orchestratorCwd?: string,
): ClaimedTopic | undefined {
  const match = resolveProjectClaim(claims, projectName, alive, orchestratorCwd);
  return match.kind === "match" ? match.claim : undefined;
}

/**
 * The transcript this project's live omp-telegram claim names — where the
 * orchestrator pane's reply actually lands.
 *
 * `arm` derives its scan directory from the tick cwd, but a pane resumed from
 * a session created elsewhere keeps the original transcript path (herdr pins
 * the pane to it), so the cwd-derived directory is the one place the reply is
 * guaranteed *not* to be (#600). The claim's `sessionFile` is the live answer;
 * identity resolves exactly as {@link resolveProjectTopicId}'s substitution
 * does, via {@link claimForProject}. Absent when the bridge has no claim or
 * the claim names no file — the caller then falls back to the cwd-derived
 * directory, which is the honest answer for a host that predates claims.
 */
export function resolveClaimedSessionFile(
  project: ProjectConfig,
  alive?: (pid: number) => boolean,
): string | undefined {
  const result = claimedTelegramTopics();
  if (result.kind !== "ok" || result.claims.length === 0) return undefined;
  return claimForProject(result.claims, project.name, alive, project.workspaceRoot)?.sessionFile;
}

/**
 * omp-telegram's canonical thread-id key spelling: a plain positive decimal
 * integer ("4001"). `Number(id)` alone would coerce `""`, whitespace, `"0"`,
 * `"01"`, `"1e2"` or `"0x10"` into integers the bridge never wrote — and a
 * thread id the bridge never wrote is a malformed registry, not a readable
 * topic (#626).
 */
function parseClaimedTopicId(id: string): number | undefined {
  if (!/^[1-9][0-9]*$/.test(id)) return undefined;
  const parsed = Number(id);
  return Number.isSafeInteger(parsed) ? parsed : undefined;
}

/**
 * What omp-telegram's live claim registry says, or why it could not be read.
 *
 * `kind: "unavailable"` carries the failure instead of collapsing it to an
 * empty list, so a caller that must tell "no claims" from "cannot tell" — the
 * #626 doctor check — can stay fail-closed. `kind: "missing"` is the one
 * absence that is not an operator error: a bridge that has never run a forum
 * pane has no file at all, exactly as a missing token is not one. Callers that
 * must not certify a route keep both non-`ok` kinds diagnostic; the setup
 * wizard is allowed to keep the missing-file silence (#318) while surfacing a
 * present-but-broken registry.
 */
export type ClaimedTopicsResult =
  | { kind: "ok"; claims: readonly ClaimedTopic[] }
  | { kind: "missing" }
  | { kind: "unavailable"; problem: string };

/**
 * omp-telegram's live topic claims, from
 * `{ threads: { "<threadId>": { name, workspaceLabel } } }`. Borrowed exactly as
 * the token is. A registry that cannot be read or does not parse as the bridge
 * shape is `unavailable` with a human reason, never a silent empty list — the
 * empty list is reserved for a registry that really carries no claims. A single
 * malformed claim member makes the whole registry `unavailable`: a partial list
 * would let doctor certify a route it never verified (#626). A registry that
 * does not exist is `missing`, not an empty list: the bridge may simply never
 * have claimed, and the caller decides whether that is silence or signal.
 *
 * `stateDir` is passed by the setup wizard, which has already probed for the
 * bridge; send-time callers let it resolve the same way the token does.
 */
export function claimedTelegramTopics(stateDir?: string): ClaimedTopicsResult {
  const override = stateDir?.trim() ?? process.env.OMP_TELEGRAM_STATE_DIR?.trim();
  const dir = override ? override : join(homedir(), ".omp", "agent", "telegram");
  const registry = join(dir, "threads.json");
  let raw: unknown;
  try {
    raw = JSON.parse(readFileSync(registry, "utf8"));
  } catch (err) {
    if ((err as NodeJS.ErrnoException).code === "ENOENT") return { kind: "missing" };
    return {
      kind: "unavailable",
      problem: `cannot read ${registry}: ${err instanceof Error ? err.message : String(err)}`,
    };
  }
  if (typeof raw !== "object" || raw === null || Array.isArray(raw) || !("threads" in raw)) {
    return {
      kind: "unavailable",
      problem: `${registry} is not a { threads: { "<threadId>": { name, workspaceLabel } } } registry`,
    };
  }
  const threads = raw.threads;
  if (typeof threads !== "object" || threads === null || Array.isArray(threads)) {
    return {
      kind: "unavailable",
      problem: `${registry}: "threads" is not a map of claims`,
    };
  }
  const out: ClaimedTopic[] = [];
  for (const [id, entry] of Object.entries(threads)) {
    const threadId = parseClaimedTopicId(id);
    if (threadId === undefined) {
      return {
        kind: "unavailable",
        problem: `${registry}: claim "${id}" is not a thread id`,
      };
    }
    if (typeof entry !== "object" || entry === null || Array.isArray(entry)) {
      return {
        kind: "unavailable",
        problem: `${registry}: claim "${id}" is not a { name, workspaceLabel } claim`,
      };
    }
    // A claim member is only as good as its strings: a non-string field means
    // the registry is not the shape the bridge writes, and coercing it would
    // let doctor certify a route it never verified (#626). Same for the pid —
    // the bridge records a number or nothing, so any other spelling fails the
    // registry instead of being dressed up as a live owner. Empty strings are
    // the bridge's own spelling for an absent field, so they stay readable.
    if ("name" in entry && typeof entry.name !== "string") {
      return { kind: "unavailable", problem: `${registry}: claim "${id}" has a non-string name` };
    }
    if ("workspaceLabel" in entry && typeof entry.workspaceLabel !== "string") {
      return { kind: "unavailable", problem: `${registry}: claim "${id}" has a non-string workspaceLabel` };
    }
    if ("sessionFile" in entry && typeof entry.sessionFile !== "string") {
      return { kind: "unavailable", problem: `${registry}: claim "${id}" has a non-string sessionFile` };
    }
    if ("cwd" in entry && typeof entry.cwd !== "string") {
      return { kind: "unavailable", problem: `${registry}: claim "${id}" has a non-string cwd` };
    }
    if ("pid" in entry && (typeof entry.pid !== "number" || !Number.isFinite(entry.pid))) {
      return { kind: "unavailable", problem: `${registry}: claim "${id}" has a non-numeric pid` };
    }
    let name = id;
    let workspaceLabel: string | undefined;
    let sessionFile: string | undefined;
    let pid: number | undefined;
    let cwd: string | undefined;
    if ("name" in entry && entry.name.trim() !== "") name = entry.name.trim();
    if ("workspaceLabel" in entry && entry.workspaceLabel.trim() !== "") workspaceLabel = entry.workspaceLabel.trim();
    if ("sessionFile" in entry && entry.sessionFile.trim() !== "") sessionFile = entry.sessionFile.trim();
    if ("cwd" in entry && entry.cwd.trim() !== "") cwd = entry.cwd.trim();
    if ("pid" in entry) pid = entry.pid;
    const claim: ClaimedTopic = { threadId, name };
    if (workspaceLabel !== undefined) claim.workspaceLabel = workspaceLabel;
    if (sessionFile !== undefined) claim.sessionFile = sessionFile;
    if (pid !== undefined) claim.pid = pid;
    if (cwd !== undefined) claim.cwd = cwd;
    out.push(claim);
  }
  return { kind: "ok", claims: out };
}

/**
 * Whether the paired bridge runs with per-session topic tidy on (access.json
 * `topicsTidy: true` — `/telegram topics tidy on`).
 *
 * On a tidy host every pane exit closes (forum) or deletes (DM) its own topic,
 * so a pinned `escalation.telegramTopicId` is structurally stale: it can never
 * be among the live claims again, and treating it as a live pin is the false
 * belief #600 names. The bridge's own setting is read here so `doctor` can say
 * "vestigial" rather than "moved once".
 */
export function telegramTopicsTidy(stateDirPath?: string): boolean {
  const override = stateDirPath?.trim() ?? process.env.OMP_TELEGRAM_STATE_DIR?.trim();
  const dir = override ? override : join(homedir(), ".omp", "agent", "telegram");
  try {
    const parsed = JSON.parse(readFileSync(join(dir, "access.json"), "utf8")) as {
      readonly topicsTidy?: unknown;
    };
    return parsed.topicsTidy === true;
  } catch {
    return false;
  }
}

/**
 * Whether a pid is a live process as `topics.ts` judges a claim
 * (`src/topics.ts:isAlive`): `process.kill(pid, 0)` throws `ESRCH` for a gone
 * process, and **any** other error — including `EPERM`, which means the
 * process exists but is owned by another uid — is dead here. The bridge
 * declines a claim it cannot prove is alive, so conductor must too; a claim
 * the bridge would treat as dead is never dressed up as a live recipient.
 *
 * `kill` is injectable so a test can deterministically throw `EPERM` without
 * owning another uid's process. Distinct from {@link lockPidAlive}, which
 * applies `api.ts`'s opposite rule to the poll lock.
 */
export function pidAlive(
  pid: number,
  kill: (target: number, signal: number) => void = (target, signal) => {
    process.kill(target, signal);
  },
): boolean {
  try {
    kill(pid, 0);
    return true;
  } catch {
    return false;
  }
}

/**
 * Whether one claimed topic is live — **the** liveness fact about a claim
 * (#987).
 *
 * Two doctor rows used to answer this question separately: `topic-pin` treated
 * every row in the registry as live (the reader returns rows, it never filtered
 * them), while `telegram-plumbing` applied `pidAlive` and called the same claim
 * `claim-dead`. One fact, two rules, opposite answers — and whichever row the
 * reader happened to trust decided whether the fleet looked healthy. A doctor
 * that contradicts itself is worse than a silent one, because it teaches the
 * reader to discount its rows.
 *
 * A claim with no pid is not live: the bridge records a number or nothing, and
 * `telegramPlumbingVerdict` already refuses a pidless claim as dead. `kill` is
 * injectable for the same reason it is on {@link pidAlive}.
 *
 * Being dead is **not** a licence to delete the row. No Bot API call lists
 * forum topics, so a claim row is the only index to a remote topic
 * (`omp-telegram/src/topics.ts:149-151`): dropping it strands the topic with no
 * supported way to remove it — 70 were stranded that way in one day, and every
 * recovery depended on an artifact nothing guarantees. Callers report dead
 * claims; the bridge's own `/cleanup` removes them, remote side first.
 */
export function claimIsLive(
  claim: Pick<ClaimedTopic, "pid">,
  kill?: (target: number, signal: number) => void,
): boolean {
  if (claim.pid === undefined) return false;
  return kill === undefined ? pidAlive(claim.pid) : pidAlive(claim.pid, kill);
}

/**
 * Whether a bot.lock owner pid is a live process as the bridge's `api.ts`
 * judges it (`src/api.ts:pidAlive`): `EPERM` means the process exists but is
 * owned by another uid, which is *live* — the bridge reads its own lock with
 * this rule (`acquireLock`), so a lock owner is judged differently from a
 * topic claim or the DM owner (see {@link pidAlive}).
 */
export function lockPidAlive(
  pid: number,
  kill: (target: number, signal: number) => void = (target, signal) => {
    process.kill(target, signal);
  },
): boolean {
  try {
    kill(pid, 0);
    return true;
  } catch (err) {
    return (err as NodeJS.ErrnoException).code === "EPERM";
  }
}

/**
 * The omp-telegram bridge's single-poller lock cadence, ported from
 * `src/api.ts` (`LOCK_HEARTBEAT_MS`): the elected poller refreshes `bot.lock`'s
 * mtime on this interval while it polls.
 */
export const TELEGRAM_LOCK_HEARTBEAT_MS = 15_000;

/**
 * A `bot.lock` whose mtime is younger than this is live regardless of its
 * recorded pid, ported from `src/api.ts` (`LOCK_FRESH_MS`). This is the
 * contract that makes daemon-pid liveness insufficient: on a `getMe` or poller
 * failure the daemon releases `bot.lock`, stops its heartbeat and sleeps while
 * its own pid and `daemon.json` remain live — so only the lock (or its absence)
 * says whether anything is actually polling.
 */
export const TELEGRAM_LOCK_FRESH_MS = 45_000;

/** The owner record omp-telegram writes into `bot.lock`. */
export interface TelegramLockOwner {
  pid: number;
  startedAt: number;
  nonce?: string;
  name?: string;
  sessionId?: string;
  sessionFile?: string;
}

/**
 * Outcome of reading the bridge's `bot.lock`, told apart so absent or
 * malformed state is a named failure, never a silent "no poller" (#612).
 */
export type TelegramPollState =
  | { kind: "ok"; owner: TelegramLockOwner; mtimeMs: number }
  | { kind: "unreadable" };

/**
 * Parse a `bot.lock` owner record, mirroring omp-telegram's own
 * `parseLockOwner` (`src/api.ts`) exactly: the first line is the owning pid,
 * read with `Number.parseInt(firstLine, 10)` — a positive decimal only, so
 * `0x10` parses to 0 and is no owner rather than becoming 16, and PID 1 is a
 * valid owner (the bridge excludes only `pid <= 0`). A legacy bare-pid lock
 * is still an owner (`{ pid, startedAt: 0 }`); a v2 record whose JSON object
 * repeats the first-line pid is returned whole — carrying the poller's
 * session identity — and a malformed v2 record falls back to the first-line
 * pid exactly as the bridge does.
 */
function parseTelegramLockOwner(content: string): TelegramLockOwner | undefined {
  const trimmed = content.trim();
  if (trimmed === "") return undefined;
  const newline = trimmed.indexOf("\n");
  const firstLine = newline === -1 ? trimmed : trimmed.slice(0, newline);
  const pid = Number.parseInt(firstLine, 10) || 0;
  if (pid <= 0) return undefined;
  if (newline !== -1) {
    try {
      const owner: unknown = JSON.parse(trimmed.slice(newline + 1));
      if (typeof owner === "object" && owner !== null && !Array.isArray(owner)) {
        const record = owner as { readonly pid?: unknown };
        if (record.pid === pid) return owner as TelegramLockOwner;
      }
    } catch {
      // Malformed v2 record — fall back to the first-line pid, as the bridge does.
    }
  }
  return { pid, startedAt: 0 };
}

/**
 * The omp-telegram bridge's poll ownership, from the same state dir as
 * `threads.json` (`bot.lock`). `unreadable` when the file is absent,
 * unreadable or not a valid owner record — a bridge that is not polling holds
 * no lock, and a malformed record is a named failure, never a pass. The
 * `mtimeMs` is the file's own mtime, so the caller can apply the bridge's
 * freshness rule (a heartbeat younger than {@link TELEGRAM_LOCK_FRESH_MS} is
 * live even when the recorded pid is gone).
 */
export function readTelegramPollState(stateDir?: string): TelegramPollState {
  const override = stateDir?.trim() ?? process.env.OMP_TELEGRAM_STATE_DIR?.trim();
  const dir = override ? override : join(homedir(), ".omp", "agent", "telegram");
  const lockPath = join(dir, "bot.lock");
  let mtimeMs: number;
  let raw: string;
  try {
    mtimeMs = statSync(lockPath).mtimeMs;
    raw = readFileSync(lockPath, "utf8");
  } catch {
    return { kind: "unreadable" };
  }
  const owner = parseTelegramLockOwner(raw);
  if (owner === undefined) return { kind: "unreadable" };
  return { kind: "ok", owner, mtimeMs };
}

/**
 * The live untopiced-DM recipient omp-telegram pins (`dm-owner.json`), as much
 * as conductor reads. The bridge routes an *unthreaded* private reply to
 * whatever session owns this file — never to a project-matched
 * `threads.json` claim, which is the forum route's owner (`src/bridge.ts`).
 */
export interface TelegramDmOwner {
  pid: number;
  name?: string;
  sessionId?: string;
  sessionFile?: string;
  workspaceLabel?: string;
}

/**
 * Outcome of reading the bridge's `dm-owner.json`. Mirrors omp-telegram's own
 * `loadDmOwner` (`src/topics.ts`): the record is a live `ThreadEntry`, so a
 * `pid` that is not a finite number — an absent file, or a malformed record —
 * is no owner at all. Absent and unreadable are deliberately one state: for
 * the flat route both mean the private reply has no pinned recipient, which
 * is a named failure, never a silent pass.
 */
export type TelegramDmOwnerState =
  | { kind: "ok"; owner: TelegramDmOwner }
  | { kind: "unreadable" };

/**
 * The bridge's untopiced-DM recipient, from the same state dir as
 * `threads.json` (`dm-owner.json`).
 */
export function readTelegramDmOwner(stateDir?: string): TelegramDmOwnerState {
  const override = stateDir?.trim() ?? process.env.OMP_TELEGRAM_STATE_DIR?.trim();
  const dir = override ? override : join(homedir(), ".omp", "agent", "telegram");
  let raw: string;
  try {
    raw = readFileSync(join(dir, "dm-owner.json"), "utf8");
  } catch {
    return { kind: "unreadable" };
  }
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch {
    return { kind: "unreadable" };
  }
  if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
    return { kind: "unreadable" };
  }
  const record = parsed as { readonly [key: string]: unknown };
  const pid = record["pid"];
  if (typeof pid !== "number" || !Number.isFinite(pid)) return { kind: "unreadable" };
  const owner: TelegramDmOwner = { pid };
  for (const key of ["name", "sessionId", "sessionFile", "workspaceLabel"] as const) {
    const value = record[key];
    if (typeof value === "string" && value.trim() !== "") owner[key] = value.trim();
  }
  return { kind: "ok", owner };
}

/**
 * The paired inbound channel as the tick defines it, probed exactly the way
 * `armTicks` probes it (the access file must parse, the bridge must be
 * enabled and `allowFrom` must hold exactly one paired owner). Anything else
 * — a missing or unreadable file, a disabled bridge, a missing or foreign
 * pair — is down with the arm's own reason.
 */
export type TelegramChannelState =
  | { kind: "up"; owner: string }
  | { kind: "down"; reason: string };

/**
 * The paired inbound channel verdict, byte-identical to the one `armTicks`
 * reads through its own `readPairedChannel` (fleet.ts keeps that one private;
 * this port lives with the other bridge-state readers so `doctor` and the arm
 * challenge can never disagree): the access file parses, the bridge is
 * enabled and `allowFrom` holds exactly one paired owner id. Every failure
 * reason is the arm's own, so a doctor finding and the arm's refusal name the
 * same fact.
 */
export function readTelegramChannel(path: string): TelegramChannelState {
  let parsed: unknown;
  try {
    parsed = JSON.parse(readFileSync(path, "utf8"));
  } catch {
    return { kind: "down", reason: "unreadable or missing access.json" };
  }
  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
    return { kind: "down", reason: "access.json is not an object" };
  }
  const access = parsed as { readonly [key: string]: unknown };
  if (access["enabled"] !== true) return { kind: "down", reason: "bridge disabled (enabled !== true)" };
  const allowFrom = access["allowFrom"];
  if (!Array.isArray(allowFrom) || allowFrom.length !== 1) {
    return { kind: "down", reason: "allowFrom must hold exactly one paired owner id" };
  }
  const owner = allowFrom[0];
  if (typeof owner !== "string" && typeof owner !== "number") {
    return { kind: "down", reason: "paired owner id is not a string or number" };
  }
  return { kind: "up", owner: String(owner) };
}

/** A fact a live-plumbing verdict verified; a pass names which kinds it proved. */
export type TelegramPlumbingProof = "claim" | "dm-owner" | "poller";

/**
 * The bounded reasons a live-plumbing verdict fails, each naming the fact that
 * failed. Forum-route failures name the topic claim the challenge would send
 * into; flat-route failures name the reply's recipient. An unreadable or
 * malformed state is always its own named failure — never folded into "no
 * claim"/"no owner", which would read as a pass.
 */
export type TelegramPlumbingFailureReason =
  | "channel-down"
  | "registry-unreadable"
  | "no-topic-claim"
  | "claim-dead"
  | "claim-session-outside"
  | "no-dm-owner"
  | "dm-owner-dead"
  | "dm-owner-unrelated"
  | "no-poller-state"
  | "poller-dead";

/**
 * The liveness-checked verdict on omp-telegram plumbing, reusable by `arm`,
 * `doctor` and recovery. A pass names the proof kinds only — never a chat or
 * topic identifier, which are exactly the values that leak from log lines.
 */
export type TelegramPlumbingVerdict =
  | { ok: true; proof: readonly TelegramPlumbingProof[] }
  | { ok: false; reason: TelegramPlumbingFailureReason };

/** Pure-code wiring so a test injects live/dead pids and freshness without touching the host. */
export interface TelegramPlumbingProbe {
  /**
   * The typed claim-registry read. Consulted only on the forum route — a flat
   * challenge never reads a claim. Any non-`ok` kind is a named failure.
   */
  registry: ClaimedTopicsResult;
  /** The typed `bot.lock` poll-ownership read — absent/malformed is a named failure. */
  poll: TelegramPollState;
  /**
   * The typed `dm-owner.json` read — the untopiced-DM recipient a flat
   * challenge's reply routes through. Consulted only on the flat route.
   */
  dmOwner: TelegramDmOwnerState;
  /**
   * The paired inbound channel state, exactly as `armTicks` reads it. Down is
   * a named failure on both routes: the arm challenge sends to the paired
   * owner, so no channel means no challenge can go out at all.
   */
  channel: TelegramChannelState;
  /**
   * Whether a recorded claim or dm-owner pid is a live process, with `topics.ts`
   * semantics (`src/topics.ts:isAlive`): `EPERM` is dead. Distinct from
   * {@link lockPidAlive} — the bridge itself uses two rules.
   */
  alive: (pid: number) => boolean;
  /**
   * Whether a recorded bot.lock owner pid is a live process, with `api.ts`
   * semantics (`src/api.ts:pidAlive`): `EPERM` means the process exists but is
   * owned by another uid, so it is *live*. The bridge reads its own lock with
   * this rule (`acquireLock`), so a lock owner is judged differently from a
   * topic claim or the DM owner.
   */
  lockAlive: (pid: number) => boolean;
  /**
   * Whether a lock-file mtime is younger than {@link TELEGRAM_LOCK_FRESH_MS} —
   * the bridge treats a fresh heartbeat as live even when the recorded pid is
   * gone (its own `acquireLock` liveness rule).
   */
  fresh: (mtimeMs: number) => boolean;
}

/**
 * The transcript-surface input for {@link telegramPlumbingVerdict}'s session
 * identity checks: the directories the claim-only proof judges recipients
 * against (a claim's, the DM owner's, or the local poll owner's session file
 * must land inside, or a flat reply could land where no challenge would have
 * looked). Since #614 the *challenge* proof reads conductor acknowledgement
 * state and never opens a transcript, so callers judging challenge-mode
 * plumbing omit this — passing it there could report arming impossible while
 * `arm` succeeds.
 */
export type TelegramPlumbingScan = { dirs: readonly string[] };

/**
 * Whether Telegram plumbing is live *now*, without the 300-second challenge and
 * without any network call, on the route {@link armTicks} actually sends on.
 *
 * The paired inbound channel must be up on both routes. Forum route (a
 * send-time topic was resolved): the claim for that *exact* topic must be
 * live — never an unrelated project-name match, which is what lets a live
 * sibling claim false-pass a stale pin — and its session identity must land
 * inside the scan surface or the reply can never be seen. Flat route (no
 * topic): the challenge is unthreaded and the reply routes through the DM
 * owner, so the pinned owner must be live and inside the scan surface; with
 * no pinned owner (`/telegram own clear`) the bridge delivers locally in the
 * polling session, so the lock's v2 owner record must carry a session
 * identity inside the scan surface — a daemon-only or unrelated owner is not a
 * recipient. On both routes the bridge's poll ownership must be live too:
 * `bot.lock` holds a valid owner whose pid is alive, or whose heartbeat is
 * fresh. Any unreadable or malformed state is a deterministic typed failure
 * naming the failed fact — never a pass. Poll liveness deliberately follows
 * the lock, not `daemon.json`: on a fatal the daemon releases `bot.lock` and
 * stops its heartbeat while its own pid stays live (#612 ecosystem
 * correction).
 */
/**
 * Adapts an injected pid-liveness predicate to the `kill` seam
 * {@link claimIsLive} owns, so the verdict can read the shared claim fact
 * without a second pid rule and without changing its own probe shape (#987).
 * Throwing is how `pidAlive` spells "dead", which is the contract it applies to
 * whatever this returns.
 */
export function killOf(alive: (pid: number) => boolean): (target: number, signal: number) => void {
  return (target) => {
    if (!alive(target)) throw new Error("not alive");
  };
}

export function telegramPlumbingVerdict(
  sendTopic: number | undefined,
  scan: TelegramPlumbingScan | undefined,
  probe: TelegramPlumbingProbe,
): TelegramPlumbingVerdict {
  const { registry, poll, dmOwner, channel, alive, lockAlive, fresh } = probe;
  if (channel.kind === "down") return { ok: false, reason: "channel-down" };
  let recipient: TelegramPlumbingProof;
  if (sendTopic !== undefined) {
    if (registry.kind !== "ok") return { ok: false, reason: "registry-unreadable" };
    const claim = registry.claims.find((c) => c.threadId === sendTopic);
    if (claim === undefined) return { ok: false, reason: "no-topic-claim" };
    // The shared fact, not a second copy of the pid rule (#987): `topic-pin`
    // reads the same predicate, so the two rows cannot disagree about a claim.
    if (!claimIsLive(claim, killOf(alive))) return { ok: false, reason: "claim-dead" };
    if (
      scan !== undefined &&
      claim.sessionFile !== undefined &&
      !sessionWithinScan(claim.sessionFile, scan.dirs)
    ) {
      return { ok: false, reason: "claim-session-outside" };
    }
    recipient = "claim";
  } else {
    if (dmOwner.kind === "ok") {
      if (dmOwner.owner.pid === undefined || !alive(dmOwner.owner.pid)) return { ok: false, reason: "dm-owner-dead" };
      if (
        scan !== undefined &&
        (dmOwner.owner.sessionFile === undefined || !sessionWithinScan(dmOwner.owner.sessionFile, scan.dirs))
      ) {
        return { ok: false, reason: "dm-owner-unrelated" };
      }
      recipient = "dm-owner";
    } else {
      // No pinned DM owner — after `/telegram own clear` the bridge delivers
      // local replies in the polling session itself, so the lock's v2 owner
      // record must name that session. A daemon-only or unrelated owner is no
      // recipient: the reply would land in a transcript arm never scans.
      if (poll.kind !== "ok") return { ok: false, reason: "no-dm-owner" };
      if (
        scan !== undefined &&
        (poll.owner.sessionFile === undefined || !sessionWithinScan(poll.owner.sessionFile, scan.dirs))
      ) {
        return { ok: false, reason: "no-dm-owner" };
      }
      recipient = "poller";
    }
  }
  if (poll.kind !== "ok") return { ok: false, reason: "no-poller-state" };
  if (!lockAlive(poll.owner.pid) && !fresh(poll.mtimeMs)) return { ok: false, reason: "poller-dead" };
  // When the poller's own session is the flat recipient, the live lock is the
  // whole proof — the recipient and the poll are one fact, not two kinds.
  return { ok: true, proof: recipient === "poller" ? ["poller"] : [recipient, "poller"] };
}

/** Whether a session file sits directly inside one of the arm scan directories. */
function sessionWithinScan(sessionFile: string, dirs: readonly string[]): boolean {
  return dirs.includes(dirname(sessionFile));
}

/**
 * The one Telegram send in this package. Exported so the report outbox (#123)
 * reuses it rather than forking it: the response handling below is load-bearing
 * and was paid for once already (see the comment on the parse). Resolves with
 * Telegram's own message id for every accepted part, in order — one id for a
 * message under the limit, one per message once a split is needed. Throws on
 * every *known* failure — connection refused, HTTP error, `{"ok":false}` —
 * which is what lets a caller treat a throw as "nobody has this" and a crash
 * as "nobody knows".
 *
 * Splitting happens here, at the transport, so every message class that rides
 * this seam — tier-2 escalations, report parts, fleet pages, operator
 * messages — is delivered whole: a report's tail is the fleet state and open
 * items, and a tier-2 escalation's tail may be the question itself (#566).
 * Parts are sent strictly in order, one `sendMessage` at a time, so they never
 * interleave with each other.
 *
 * Optional `topicId` pins the message to a forum topic (`message_thread_id`).
 * A definitive missing-thread reject degrades in two ordered steps (#1094):
 * first to the project's one live claim, when `project` names who the send
 * answers to and the registry identifies exactly one (#1089) — a resend into
 * the project's own current topic. Only when no topic can be resolved does it
 * retry as a flat chat, prepending a notice that names the stale pin and the
 * project, and recording a mis-route row for doctor/status (#318's silent
 * shape). Every degrade re-sends the whole split: parts already accepted went
 * into a thread Telegram has just told us is gone, so nothing readable is
 * ever duplicated.
 */
export async function sendTelegram(
  token: string,
  chatId: string,
  text: string,
  opts?: {
    topicId?: number;
    /**
     * Who this send answers to, when it rides a pinned topic (#1094): the
     * project whose escalation, report, or challenge it carries. Only a
     * caller that resolved `topicId` for a project passes it, and that pair
     * is what lets a missing-thread fallback prefer the project's one live
     * claim, label the flat delivery with the project's name, and record the
     * mis-route where the operator looks.
     */
    project?: { name: string; workspaceRoot?: string };
  },
): Promise<number[]> {
  const topicId =
    opts?.topicId !== undefined && Number.isFinite(opts.topicId) && Number.isSafeInteger(opts.topicId)
      ? opts.topicId
      : undefined;
  const parts = telegramTextParts(text);
  try {
    return await postTelegramParts(token, chatId, parts, topicId);
  } catch (err) {
    if (
      topicId === undefined ||
      !(err instanceof TelegramSendError) ||
      err.outcome !== "definitive" ||
      !isMissingTelegramThread(err.message)
    ) {
      throw err;
    }
    warn(
      `escalation.telegramTopicId=${topicId} is stale (${err.message}); retrying flat chat`,
    );
    // #318's flat retry is the last resort, not the first (#1094). When the
    // claim registry names exactly one live claim for this project — #1089's
    // cwd discriminator included — the resend follows it: a page landing in
    // the project's own current topic is not a mis-route at all. A
    // substitution that itself fails falls through to the labelled flat send
    // below; delivery outranks tidiness.
    const project = opts?.project;
    if (project !== undefined) {
      try {
        const registry = claimedTelegramTopics();
        if (registry.kind === "ok") {
          const claim = claimForProject(registry.claims, project.name, undefined, project.workspaceRoot);
          if (claim !== undefined && claim.threadId !== topicId) {
            warn(`following omp-telegram's live "${project.name}" claim instead of the flat chat`);
            return await postTelegramParts(token, chatId, parts, claim.threadId);
          }
        }
      } catch (subErr) {
        warn(
          `live-claim substitution failed (${subErr instanceof Error ? subErr.message : String(subErr)}); delivering flat`,
        );
      }
    }
    // Flat, but never silent (#1094): the delivered text names the stale pin
    // and the project, so the wrong audience knows what it is reading, and
    // doctor/status gain the evidence against this pin. The record is written
    // only after the send resolves — a flat send that threw delivered
    // nothing, and the next tick retries it as a first attempt.
    const labelled = `${staleTopicNotice(topicId, project?.name)}\n\n${text}`;
    const ids = await postTelegramParts(token, chatId, telegramTextParts(labelled), undefined);
    if (project?.name !== undefined) recordTelegramMisroute(project.name, topicId);
    return ids;
  }
}

/** One part per `sendMessage`, in order, collecting the ids Telegram returns. */
async function postTelegramParts(
  token: string,
  chatId: string,
  parts: readonly string[],
  topicId: number | undefined,
): Promise<number[]> {
  const ids: number[] = [];
  for (const part of parts) {
    const id = await postTelegramMessage(token, chatId, part, topicId);
    if (id !== undefined) ids.push(id);
  }
  return ids;
}

/** Telegram's definitive "that forum topic is gone" answers. */
function isMissingTelegramThread(diagnostic: string): boolean {
  return /message thread not found|topic_id_invalid/i.test(diagnostic);
}

async function postTelegramMessage(
  token: string,
  chatId: string,
  text: string,
  topicId: number | undefined,
): Promise<number | undefined> {
  const url = `https://api.telegram.org/bot${token}/sendMessage`;
  const payload: Record<string, unknown> = {
    chat_id: chatId,
    // `sendTelegram` splits long text into labelled parts above this call, so
    // the text handed here is always a whole message: a part under the wire
    // limit, or a message that never needed splitting. Nothing is truncated —
    // a hard slice here is how the digest lost its tail (#566).
    text,
    disable_web_page_preview: true,
  };
  if (topicId !== undefined) payload.message_thread_id = topicId;
  const body = JSON.stringify(payload);

  let res: Response;
  try {
    res = await fetch(url, { method: "POST", headers: { "content-type": "application/json" }, body });
  } catch (cause) {
    // The URL embeds the token, and runtimes love to quote the failing URL back
    // at you — redact before this string reaches a log or an issue comment.
    const nested = cause instanceof Error && cause.cause instanceof Error ? `: ${cause.cause.message}` : "";
    const reason = cause instanceof Error ? `${cause.message}${nested}` : String(cause);
    throw new TelegramSendError(
      `telegram sendMessage failed: ${redact(reason, token)}`,
      telegramFailureOutcome(cause),
    );
  }

  // The raw body is what gets parsed; `diagnostic` is only ever for humans.
  // These were one variable once, and the bug that produced was expensive: the
  // 400-char cap meant for a log line was applied first, so `JSON.parse` was
  // handed a truncated object and threw. Telegram echoes the whole message back
  // inside `result.text`, so every page long enough to matter overflowed and was
  // reported as rejected *after being delivered* — with the dedup marker only
  // written on success, that also re-sent the same page every tick. Redaction
  // stays out of the parse for the same reason: it rewrites the very bytes the
  // decision is read from.
  let raw: string;
  let bodyUnread: unknown;
  try {
    raw = await res.text();
  } catch (cause) {
    raw = "";
    bodyUnread = cause;
  }
  const diagnostic = redact(raw, token).slice(0, 400);
  if (!res.ok) {
    throw new TelegramSendError(
      `telegram sendMessage failed: HTTP ${res.status} ${diagnostic}`,
      "definitive",
    );
  }
  // A 200 whose body never arrived is the one case where the status reads as
  // encouraging and proves nothing: Telegram had already decided, and the answer
  // was lost coming back. Reported as `unknown` rather than rejected so the
  // retry carries a possible-repeat marker instead of quietly posting a second
  // copy of a message that very probably landed. Checked after the status,
  // because a non-ok status settles the question on its own.
  if (bodyUnread !== undefined) {
    const reason = bodyUnread instanceof Error ? bodyUnread.message : String(bodyUnread);
    throw new TelegramSendError(
      `telegram sendMessage left but its response body could not be read: ${redact(reason, token)}`,
      "unknown",
    );
  }

  // Telegram answers 200 with `{"ok":false}` for plenty of real failures
  // (kicked from the chat, bad chat_id), so the status alone proves nothing.
  let ok = false;
  let messageId: number | undefined;
  try {
    const parsed: unknown = JSON.parse(raw);
    ok = typeof parsed === "object" && parsed !== null && "ok" in parsed && parsed.ok === true;
    // Read from the same parse the ok/not-ok decision comes from, and only from
    // the body — the outbox records this id as the durable proof of *which*
    // attempt landed, and this transport offers no other reconciliation. A
    // missing or odd `result` is not a delivery failure: `ok: true` is the
    // verdict, the id is a courtesy, and inventing one would be worse than
    // recording none.
    if (typeof parsed === "object" && parsed !== null && "result" in parsed) {
      const result: unknown = parsed.result;
      if (typeof result === "object" && result !== null && "message_id" in result) {
        const id: unknown = result.message_id;
        if (typeof id === "number" && Number.isSafeInteger(id)) messageId = id;
      }
    }
  } catch {
    ok = false;
  }
  if (!ok) {
    throw new TelegramSendError(`telegram sendMessage rejected: ${diagnostic}`, "definitive");
  }
  return messageId;
}

/**
 * Strips the bot token — and its secret half, which some proxies log on its
 * own — out of any string that is about to be thrown, logged or commented.
 */
function redact(text: string, token: string): string {
  let out = token ? text.split(token).join("<redacted>") : text;
  const colon = token.indexOf(":");
  const secret = colon >= 0 ? token.slice(colon + 1) : "";
  if (secret.length >= 8) out = out.split(secret).join("<redacted>");
  return out;
}