/**
 * ⛔ A HERDR SEAT WAS DEAF TO EVERY TMUX SEAT, AND THE REASON IS A SINGLETON.
 *
 * `deliverToHerdrSeats` runs at SEND time and returns early unless `activeTransport().kind` is
 * HERDR. `activeTransport()` returns a module-level singleton wired once at startup from the
 * SENDER's own config — so that early return can only ever see the SENDER's transport, and a tmux
 * seat's server could never type into a herdr pane. Measured both ways: `herdr → herdr` typed and
 * woke the seat; `tmux → herdr` reached the inbox file and was never typed.
 *
 * ⚠ THE EASIER FIX IS THE WRONG ONE: letting the SENDER drive herdr per recipient passes on a
 * one-machine fleet and fails silently and asymmetrically on two, because every sender would need
 * the `herdr` binary on its own host. So the duty goes to the seat that needs waking — the only
 * party guaranteed to be where it is.
 *
 * ⛔⛆ AND THE FIRST VERSION OF THIS FILE WAS INERT ON EVERY SEAT, INCLUDING ITS AUTHOR'S.
 * `8363968` started the tail only when `AGENT_COORD_BOUND_AGENT` was set at startup. No seat on
 * this fleet sets it — identity binds at `join` — so the coordinator measured "herdr tail NOT
 * started" on the PR's own server and qa read the variable unset on all six live seat servers.
 * Its eleven tests injected the agent id and so could not see it. The tail now starts from the
 * IDENTITY BINDING (`startTailOnBind`), idempotently, one per agent per process.
 *
 * ⛔ AND IT MUST NOT RUN THE REAPER. That version called `loadLiveTransports()` on every tick —
 * the loader that DELETES markers it judges dead, named an hour earlier as the vanishing-marker
 * mechanism. A reaper in a once-a-second loop on every seat is not shippable on the argument that
 * the current build usually keeps the marker. This file reads its own seat's marker FILE directly,
 * read-only, and never asks about anyone else's.
 *
 * THE CURSOR RULE IS THE PUSHER'S (⟨q-7be94b5e⟩): advance the PUSH cursor only past what was
 * VERIFIABLY typed. Two corrections from qa's read of `8363968`, both now driven by tests:
 *   · the cursor advances to the byte AFTER THE DELIVERED LINE, never to the file's size — sizing
 *     it skipped a later message in the same tick whose paste then failed;
 *   · a cursor write that FAILS is held, and an in-process high-water mark stops the same message
 *     being re-typed on the next tick — the directory fix closed one cause of a paste loop, this
 *     closes the class.
 *
 * IDLE COST, COUNTED BY TESTS RATHER THAN ASSERTED HERE: when no tailed file has grown, a tick is
 * one `stat()` per tailed file and nothing else. The marker and room membership are re-read every
 * `REFRESH_TICKS` ticks, and no herdr subprocess runs until there is something to type.
 */
import { readFileSync, statSync } from "node:fs";
import { inboxFile, roomFile, getRooms, transportFile } from "../store.js";
import { activeTransport, HERDR, targetOf, type Transport, type TransportMarker } from "../transports/index.js";
import { renderForPane, advancePushCursor, readPushCursorFor, type Message } from "./herdr-delivery.js";
import { newMessagesIn } from "./jsonl-offsets.js";

export const DEFAULT_POLL_MS = 1000;
/** Marker and room membership are re-read this often. A room joined mid-session is picked up within this many ticks. */
export const REFRESH_TICKS = 30;

type Source = { kind: "dm" | "room"; chan?: string; file: string };
export type TailOutcome = {
  delivered: { kind: "dm" | "room"; chan?: string; id: string }[];
  held: { kind: "dm" | "room"; chan?: string; id: string; why: string }[];
  idle?: string;
};

/** Per-agent memory that survives between ticks — what makes the idle path cheap and the paste loop impossible. */
export type TailContext = {
  tick: number;
  marker: TransportMarker | null;
  rooms: string[];
  /** In-process high-water mark per source: never re-type below this, even if the persistent cursor could not be written. */
  hw: Map<string, number>;
  /** Last seen size per source: a source that has not grown costs one stat. */
  sizes: Map<string, number>;
  /**
   * ⛔ START AT EOF, NEVER AT WHATEVER THE CURSOR FILE HAPPENS TO SAY. A tail is a LIVE push
   * mechanism, not a mail reader: nothing that predates its start belongs on the pane. Measured
   * 2026-09-17 — a daemon restart reset every push cursor to ~0, so a tail starting from disk
   * state would have typed a 3.4MB inbox into one seat's pane. `read_messages` still serves that
   * history on demand; that is the verb for it.
   */
  seedEof: boolean;
  /** Sources already seeded — a room discovered on a later refresh is seeded on ITS first sight, not skipped. */
  seeded: Set<string>;
};
export const newContext = (seedEof = false): TailContext => ({ tick: 0, marker: null, rooms: [], hw: new Map(), sizes: new Map(), seedEof, seeded: new Set() });

const keyOf = (s: { kind: string; chan?: string }) => (s.kind === "dm" ? "dm" : `room:${s.chan}`);
const sizeOf = (file: string): number => {
  try { return statSync(file).size; } catch { return 0; }
};

/** Read THIS seat's own marker, read-only. Never the reaping loader. */
export function readOwnMarker(agentId: string): TransportMarker | null {
  try { return JSON.parse(readFileSync(transportFile(agentId), "utf8")) as TransportMarker; } catch { return null; }
}

/** The shared line reader — re-exported so callers of this module keep one import. */
export { newMessagesIn } from "./jsonl-offsets.js";

async function offsetOf(agentId: string, src: Source, ctx: TailContext): Promise<number> {
  const c = await readPushCursorFor(agentId);
  const persisted = src.kind === "dm" ? Number(c.inboxOffset ?? 0) : Number(((c.roomOffsets as Record<string, number> | undefined) ?? {})[src.chan!] ?? 0);
  return Math.max(persisted, ctx.hw.get(keyOf(src)) ?? 0);
}

/** One pass. Everything that talks to the outside world is injectable, and nothing here schedules itself. */
export async function tailOnce(
  agentId: string,
  opts: {
    transport?: Transport;
    ctx?: TailContext;
    markerOf?: (agentId: string) => TransportMarker | null;
    roomsOf?: () => Promise<Record<string, { members?: string[] }>>;
    /**
     * ⟨q-1ce5bc97⟩ step 1 — EVERY push outcome, logged, not just the ones that end up in
     * `held`. Before this, bus-daemon.err.log recorded no push outcomes at all: a
     * `delivered` and a silently-dropped `typed-unconfirmed` looked identical from the
     * log, because nothing wrote either one. Injectable so a test can capture lines
     * without a real stderr; defaults to `console.error` — the same sink every other
     * diagnostic in this process already writes to.
     */
    log?: (line: string) => void;
  } = {},
): Promise<TailOutcome> {
  const out: TailOutcome = { delivered: [], held: [] };
  const log = opts.log ?? ((l: string) => console.error(l));
  const t = opts.transport ?? activeTransport();
  if (!t || t.kind !== HERDR) return { ...out, idle: "this server's transport is not herdr — the tail is for a herdr seat's own inbox" };
  const ctx = opts.ctx ?? newContext();
  const refresh = ctx.tick % REFRESH_TICKS === 0;
  ctx.tick += 1;

  // A seat with NO marker re-reads it only on the refresh cadence too. The coordinator's gate found
  // `|| !ctx.marker` here, which made a markerless seat — the exact state a vanished marker leaves —
  // read its marker file on every tick while the stated idle cost said otherwise.
  if (refresh) {
    ctx.marker = (opts.markerOf ?? readOwnMarker)(agentId);
    // ⟨q-d58ccdce⟩ ⛔⛆⛆ THIS BRANCH IS THE WHOLE DEFECT. `marker.rooms === false` makes
    // `ctx.rooms` PERMANENTLY EMPTY — every room this agent belongs to is silently excluded
    // from `sources` below, so the per-source loop never runs for a single one of them. That
    // is a total, structural skip with no `continue` to log: #395's per-outcome push logging
    // fires only for a source the loop actually reaches, and a room that is never a source
    // produces zero log lines forever — indistinguishable from a room with nothing owed.
    // Measured live 2026-09-19: every groundwork-kit-* seat's marker had `rooms:false` (a
    // side effect of `attach_agent {..., includeRoom:false}`, called believing it only
    // affected DM liveness), while every unaffected control seat had `rooms:true`. So THIS
    // is the log line the class needed — not one more line inside a loop that never starts.
    if (ctx.marker && ctx.marker.transport === HERDR && ctx.marker.rooms === false) {
      log(`[herdr-tail] rooms DISABLED for ${agentId} — marker.rooms=false, so no room is ever tailed regardless of membership (⟨q-d58ccdce⟩); re-attach with includeRoom:true (or omit it) to restore room delivery`);
    }
    if (ctx.marker && ctx.marker.transport === HERDR && ctx.marker.rooms !== false) {
      try {
        const all = await (opts.roomsOf ?? getRooms)();
        ctx.rooms = Object.entries(all ?? {}).filter(([, r]) => (r?.members ?? []).includes(agentId)).map(([chan]) => chan);
      } catch { ctx.rooms = []; }
    } else {
      ctx.rooms = [];
    }
  }
  const marker = ctx.marker;
  if (!marker || marker.transport !== HERDR) return { ...out, idle: `${agentId} has no herdr marker on this bus — nothing to type into (re-checked every ${REFRESH_TICKS} ticks)` };

  const sources: Source[] = [{ kind: "dm", file: inboxFile(agentId) }, ...ctx.rooms.map((chan) => ({ kind: "room" as const, chan, file: roomFile(chan) }))];

  for (const src of sources) {
    const size = sizeOf(src.file);
    const key = keyOf(src);
    // ⭐ THE IDLE PATH: a source that has not grown since the last tick costs this one stat.
    if (ctx.sizes.get(key) === size) continue;
    ctx.sizes.set(key, size);
    // ⛔ FIRST SIGHT OF A SOURCE UNDER seedEof: adopt EOF and type NOTHING. This is the only place
    // the starting offset is decided, and it is decided EXPLICITLY rather than inherited from a
    // cursor file that a restart may have reset. Per source, so a room joined later seeds on its
    // own first sight instead of replaying its history. Logged (low-frequency: once per source
    // per process) so a seat that never types into a newly-joined room is distinguishable from
    // one that was never told the room existed.
    if (ctx.seedEof && !ctx.seeded.has(key)) {
      ctx.seeded.add(key);
      ctx.hw.set(key, size);
      log(`[herdr-tail] seeded ${key} at EOF (${size} bytes) for ${agentId} — first sight this process, nothing predating the tail will be typed`);
      continue;
    }
    const from = await offsetOf(agentId, src, ctx);
    // ⭐ NEGATIVE CONTROL FOR THE CLASS THIS ROW IS ABOUT: nothing is logged here when
    // `size === from` — a source with nothing owed must stay silent, or every idle room on
    // every tick becomes a line and the log that is supposed to make a real stall visible
    // drowns in ones that are not.
    if (size <= from) continue;

    for (const { msg, end } of newMessagesIn(src.file, from)) {
      if (msg.from === agentId) { ctx.hw.set(key, end); continue; }
      const where = src.kind === "dm" ? ({ kind: "dm" } as const) : ({ kind: "room", chan: src.chan! } as const);
      const rendered = await renderForPane(msg as unknown as Message, agentId, where);
      let r: { delivered: boolean; error?: string; verified?: boolean; safeToRetry?: boolean; outcome?: string };
      try { r = await t.push(marker, rendered); } catch (e) { r = { delivered: false, safeToRetry: false, error: (e as Error).message }; }
      // ⟨q-1ce5bc97⟩ step 1(b) — EVERY push outcome, logged, before any branch decides what to do
      // with it. Not just the ones that end up `held`: a `delivered` needs a line too, or the
      // log can never distinguish "nothing was owed" from "something was pushed and it worked".
      log(
        `[herdr-tail] push id=${String(msg.id ?? "?")} target=${targetOf(marker) ?? "?"} outcome=${r.outcome ?? (r.delivered ? "delivered" : "unknown")}` +
          (r.error ? ` reason=${JSON.stringify(r.error)}` : ""),
      );
      if (!r.delivered && r.safeToRetry === true) {
        // HELD: the transport sent NO key (no ready input box, a draft, a dialog, an unreadable pane).
        // Nothing moves and the message is retried next tick, which costs only a screen read. Stop
        // this source so no later message is typed past one that did not land.
        out.held.push({ kind: src.kind, chan: src.chan, id: String(msg.id ?? ""), why: r.error ?? "held — nothing was typed" });
        ctx.sizes.delete(key); // force a re-read next tick
        break;
      }
      // ⛔ ANYTHING SHORT OF A VERIFIED DELIVERY DOES NOT MOVE THE PERSISTENT CURSOR (⟨q-15d763dc⟩).
      // Keys may have reached the pane (text typed, an Enter pressed) without the screen proving the
      // message was submitted, so it stays owed on disk: read_messages still serves it and a restart
      // re-offers it. The in-process high-water mark alone stops it being re-typed every tick, since
      // re-typing a message that may already be sitting in the box is the paste loop in another coat.
      // A throw lands here too: nothing says whether a key went out before it.
      if (!r.delivered || r.verified !== true) {
        ctx.hw.set(key, end);
        out.held.push({ kind: src.kind, chan: src.chan, id: String(msg.id ?? ""), why: `${r.outcome ?? (r.delivered ? "unverified" : "not delivered")} — ${r.error ?? "not verified"}; the persistent cursor was not moved (q-15d763dc)` });
        break;
      }
      // It landed and was verified. Record that in-process FIRST, so a cursor write that fails
      // cannot turn one delivery into a paste loop, then persist exactly past this line.
      ctx.hw.set(key, end);
      const persisted = await advancePushCursor(agentId, where, end);
      out.delivered.push({ kind: src.kind, chan: src.chan, id: String(msg.id ?? "") });
      if (!persisted) {
        out.held.push({ kind: src.kind, chan: src.chan, id: String(msg.id ?? ""), why: "delivered, but the push cursor could not be written — held in memory so it is not re-typed; a restart before the next successful write may type it once more" });
        break;
      }
    }
  }
  return out;
}

// ─── the per-process registry of running tails ─────────────────────────────

type TailRecord = {
  agentId: string;
  startedAt: string;
  ticks: number;
  delivered: number;
  held: number;
  lastTickAt: string | null;
  lastWhy: string | null;
  stop: () => void;
};
const running = new Map<string, TailRecord>();

/** What `capabilities` reports: the tails THIS PROCESS is actually running, not the ones it could run. */
export function herdrTailState(): { agentId: string; running: true; startedAt: string; ticks: number; delivered: number; held: number; lastTickAt: string | null; lastWhy: string | null }[] {
  return [...running.values()].map(({ stop: _stop, ...r }) => ({ ...r, running: true as const }));
}

export function stopHerdrTail(agentId: string): boolean {
  const r = running.get(agentId);
  if (!r) return false;
  r.stop();
  running.delete(agentId);
  return true;
}
export function stopAllHerdrTails(): void {
  for (const id of [...running.keys()]) stopHerdrTail(id);
}

/**
 * Start the tail for `agentId` if this server's transport is herdr. IDEMPOTENT: a second bind of
 * the same agent in the same process returns `running: true, started: false`.
 *
 * ⛔ NEVER OVERLAPS ITSELF: a tick still typing when the timer fires would interleave two pastes
 * into one pane; the guard is one in-flight flag, because the next tick re-reads the cursor.
 */
export function ensureHerdrTail(
  agentId: string,
  opts: { transport?: Transport; pollMs?: number; seedEof?: boolean; tail?: (id: string, ctx: TailContext) => Promise<TailOutcome> } = {},
): { started: boolean; running: boolean; why: string } {
  const t = opts.transport ?? activeTransport();
  if (!t || t.kind !== HERDR) return { started: false, running: false, why: `transport is ${t?.kind ?? "unwired"}, not herdr — a tmux seat is woken by its pusher` };
  if (running.has(agentId)) return { started: false, running: true, why: "already tailing this agent in this process" };
  const ctx = newContext(opts.seedEof === true);
  const run = opts.tail ?? ((id: string, c: TailContext) => tailOnce(id, { transport: t, ctx: c }));
  const rec: TailRecord = { agentId, startedAt: new Date().toISOString(), ticks: 0, delivered: 0, held: 0, lastTickAt: null, lastWhy: null, stop: () => {} };
  let inFlight = false;
  const timer = setInterval(() => {
    if (inFlight) return;
    inFlight = true;
    void run(agentId, ctx)
      .then((o) => {
        rec.ticks += 1;
        rec.lastTickAt = new Date().toISOString();
        rec.delivered += o.delivered.length;
        rec.held += o.held.length;
        rec.lastWhy = o.held.at(-1)?.why ?? o.idle ?? null;
      })
      .catch((e) => { rec.lastWhy = `tick threw: ${(e as Error).message}`; })
      .finally(() => { inFlight = false; });
  }, opts.pollMs ?? DEFAULT_POLL_MS);
  timer.unref?.(); // optional-call: `unref` is absent on some host timer shims, and a server that cannot unref its tail must still start it
  rec.stop = () => clearInterval(timer);
  running.set(agentId, rec);
  return { started: true, running: true, why: `tailing ${agentId}'s inbox and rooms every ${opts.pollMs ?? DEFAULT_POLL_MS}ms` };
}

/**
 * The hook the server calls at EVERY identity binding — `join`, a gated first claim, the env var,
 * or (HTTP) an authenticated request from a PRE-BOUND identity.
 *
 * ⛔ THE BINDING TRANSITION IS NOT REACHABLE UNDER HTTP, WHICH IS WHY THIS ALSO HANGS OFF AUTH.
 * The `bound === undefined` sites below only fire when a session CLAIMS an id. A token-authenticated
 * daemon resolves identity per request and reports `pre-bound (N agents)`, so `bound` is never
 * undefined and NO tail ever started — measured 2026-09-17 as `herdrTail: []` on a freshly
 * restarted daemon whose seats had re-joined after it. With the send-time path deferring owed
 * messages to "the recipient's own tail", a tail that never exists makes that deferral a permanent
 * silent drop: one message behind and the seat is deaf for good (⟨q-cdb5b007⟩).
 *
 * Idempotent, so the per-request call costs one Map lookup once the tail is up.
 */
export function startTailOnBind(agentId: string, log: (line: string) => void = (l) => console.error(l)): void {
  const r = ensureHerdrTail(agentId, { seedEof: true });
  if (r.started) log(`[agent-coord-mcp] herdr tail: ${r.why}`);
}