import { adjustCursors } from "./admin.js";
import { RECORD_AUTHORITY, isHuman, recordAuthorityFor, resolveRole, roleMatches } from "../roles.js";
import {
  TYPED_RECORD_CUTOVER_ISO,
  suggestRecordType,
  typedRecordGuidance,
  typedRecordMode,
} from "../typed-records.js";
import { ARCHIVE_STATUS_FILE, ARCHIVE_INBOX_DIR, ARCHIVE_ROOMS_DIR, archiveJsonl, archiveInboxFile, archiveRoomFile } from "../store.js";
import { randomUUID } from "node:crypto";
import { existsSync, openSync, watch } from "node:fs";
import { promises as fsp } from "node:fs";
import { spawn, spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import { z } from "zod";
import { checkClosingLine } from "../closing-line.js";
import { readLog } from "./logwatch.js";
// The replay grammar is single-sourced in hooks/replay.mjs and shared with both
// pushers — see the note at the annotation site below. `hooks/` ships beside
// `dist/`, so this resolves the same in the built package as in source.
// @ts-expect-error — untyped .mjs sibling, deliberately not duplicated in TS
import { priorDeliveries, replayInfo } from "../../hooks/replay.mjs";
// ⟨q-e5cb3538⟩ The pane-unsafe byte class is single-sourced there too, shared with the renderer.
// @ts-expect-error — untyped .mjs sibling, deliberately not duplicated in TS
import { findControlByte } from "../../hooks/control-bytes.mjs";
import { renderRecord } from "./render.js";
import path from "node:path";
import {
  AGENTS_FILE,
  CURSOR_DIR,
  DEFAULT_ROOM,
  INBOX_DIR,
  ROOT,
  ROOM_FILE,
  ROOMS_DIR,
  ROOMS_FILE,
  STATUS_FILE,
  addMember,
  appendJsonl,
  cursorFile,
  deleteFile,
  ensureRoom,
  fileSize,
  getRooms,
  inboxFile,
  listCursorFiles,
  listInboxFiles,
  listTransportFiles,
  logFile,
  memberRooms,
  normalizeRoom,
  pidFile,
  readJson,
  readJsonl,
  receiptFile,
  listReceiptFiles,
  removeMember,
  rewriteJsonl,
  roomFile,
  rotateAgentToken,
  setRoomMeta,
  stashHistory,
  retrieveHistory,
  pruneHistory,
  transportFile,
  TRANSPORT_DIR,
  updateJson,
  type RoomRegistry,
} from "../store.js";
import {
  type AgentEntry,
  type AgentRegistry,
  type Message,
  type MessageRecord,
  type StatusEntry,
  type Cursor,
  type Source,
  type TransportMarker,
  sourceFile,
  getOffset,
  setOffset,
  sysMsg,
  moveFile,
  STALE_MS,
  EVICT_MS,
  MAX_WAIT_MS,
  isDecision,
} from "./shared.js";
import { isKnownHuman } from "../store.js";
import { deliverToHerdrSeats } from "./herdr-delivery.js";
import { verifyCommitCite } from "../commit-cite.js";

// ---------- send_message ----------

// Typed protocol record (Phase 8). Additive: omitting it reproduces v1
// behavior exactly.
const citationSchema = z.object({
  kind: z.enum(["pr", "file", "commit", "url"]),
  ref: z.string().min(1),
});

// Payloads are LOOSE objects: a v3 sender's extra keys ride through to disk
// untouched rather than being silently stripped. That is safe precisely
// because payload is nested — it is caller data, and nothing in it is ever
// read as a top-level Message field (which is what keeps a forged `tag` or
// `urgent` out; see the source-level lock in test/tier.test.mjs).
const summaryPayload = z.looseObject({ summary: z.string().min(1) });

// All five fields, or none. A `decision` carrying three of them is
// structurally wrong for the type it claims — and would render as a truncated
// packet, which is worse than no packet.
const decisionPayload = z.looseObject({
  title: z.string().min(1),
  context: z.string().min(1),
  options: z.array(z.string().min(1)).min(1),
  recommendation: z.string().min(1),
  ifNoAction: z.string().min(1),
});

// ⟨q-dcbaf544⟩ — `gatedBy` is the seat that JUDGED when it is not the sender;
// `scribe` is the sender that TRANSCRIBED it. Fields, not prose: a name in
// prose is a mention, a name in a field is a position, and only the position
// survives a scanner. Authority does not move (David's ruling 2026-09-14):
// the roles that may emit `verdict` are unchanged; this is how a gate routed
// to any other seat reaches the record without lying about who gated.
const verdictPayload = z.looseObject({
  result: z.enum(["pass", "fail"]),
  headRefOid: z.string().min(1),
  notes: z.string().optional(),
  gatedBy: z.string().min(1).optional(),
  scribe: z.string().min(1).optional(),
});

// Discriminated on `type`, so an unknown type is rejected outright while a
// known type is checked only against its own shape. `payload` is optional on
// every arm: Phase 8 is additive and may not put a new required field on the
// wire. `cites` is optional here too — `done` needs a PR citation, but that is
// enforced in sendMessageTool as a plain {ok:false,error}, mirroring the
// identity-binding rejection, rather than as a schema throw.
export const messageRecordSchema = z.discriminatedUnion("type", [
  z.object({ type: z.literal("decision"), payload: decisionPayload.optional(), cites: z.array(citationSchema).optional() }),
  z.object({ type: z.literal("verdict"), payload: verdictPayload.optional(), cites: z.array(citationSchema).optional() }),
  z.object({ type: z.literal("done"), payload: summaryPayload.optional(), cites: z.array(citationSchema).optional() }),
  z.object({ type: z.literal("blocker"), payload: summaryPayload.optional(), cites: z.array(citationSchema).optional() }),
  z.object({ type: z.literal("risk"), payload: summaryPayload.optional(), cites: z.array(citationSchema).optional() }),
  z.object({ type: z.literal("fyi"), payload: summaryPayload.optional(), cites: z.array(citationSchema).optional() }),
  z.object({ type: z.literal("action"), payload: summaryPayload.optional(), cites: z.array(citationSchema).optional() }),
  z.object({ type: z.literal("go"), payload: summaryPayload.optional(), cites: z.array(citationSchema).optional() }),
  z.object({ type: z.literal("scope"), payload: summaryPayload.optional(), cites: z.array(citationSchema).optional() }),
]);

// ---------- record authority (Phase 8 Task 4) ----------

// The table (RECORD_AUTHORITY, ../roles.ts) is a floor on the three types
// other agents ACT on; everything else is unrestricted.
//
// NOT A TRUST BOUNDARY. A role is self-declared at register/join (there is no
// authority issuing them), so this is a CONSISTENCY check: it stops a worker
// from accidentally emitting a `verdict` or countersigning its own `scope`,
// the same way a linter stops a typo. Anything that must actually be
// authenticated has to resolve identity-bound tokens (see tokens.json), never
// this table.
// Rejection shape mirrors the identity-binding rejection in server.ts: the
// caller gets a plain `{ok: false, error}`, and nothing is written.
export async function checkRecordAuthority(
  from: string,
  record: MessageRecord | undefined,
): Promise<{ ok: true } | { ok: false; error: string }> {
  if (!record) return { ok: true };
  const rule = RECORD_AUTHORITY[record.type];
  if (!rule) return { ok: true };

  const reg = await readJson<AgentRegistry>(AGENTS_FILE, {});
  const entry = reg[from];
  if (roleMatches(entry, rule.roles)) return { ok: true };

  const held = resolveRole(entry);
  return {
    ok: false,
    error:
      `record.type '${record.type}' is restricted to ${rule.label} roles ` +
      `(${[...rule.roles].join(", ")}); sender '${from}' holds ` +
      `${held ? `role '${held.roleId}'` : entry ? "no role" : "no registry entry"}. ` +
      `Send it as text, or register with the role that owns this record type.`,
  };
}

/** ⟨q-dcbaf544⟩ — `scribe`, when present, must be the sender; `gatedBy` may name anyone (that is its point). */
export function checkVerdictScribe(from: string, record: MessageRecord | undefined): { ok: true } | { ok: false; error: string } {
  if (!record || record.type !== "verdict") return { ok: true };
  const p = record.payload as { scribe?: string; gatedBy?: string } | undefined;
  if (p?.scribe && p.scribe !== from) {
    return {
      ok: false,
      error:
        `verdict payload.scribe is '${p.scribe}' but the sender is '${from}' — the scribe is the seat that SENDS the record. ` +
        `Put the seat that judged in payload.gatedBy and either omit scribe or set it to '${from}'.`,
    };
  }
  return { ok: true };
}

// ---------- typed records obligatory (Phase 5.1 Task 12) ----------

// An untyped agent→agent message must not be able to EXIST. Enforced HERE, at
// the send, and not at the render: a rule applied where the message is read
// leaves the untyped message on disk, and the next reader re-derives the type
// from prose. See src/typed-records.ts for the staging, the suggestion rules,
// and why `fyi` stays an honest catch-all.
//
// Returns `undefined` when the send is fine, a WARNING string while the rule is
// staged, or a REFUSAL after the cutover.
//
// TWO EXEMPTIONS, AND THEY ARE DIFFERENT IN KIND:
//   - the RECIPIENT is a human. Canon: "David-facing messages may use normal
//     prose". Scoped to agent→agent traffic, so the one channel whose reader is
//     a person is untouched. Nothing is declared for this — it is a property of
//     who is being written to.
//   - the SENDER holds a prose-only exemption, declared per agent at join and
//     visible in list_agents. That one is a statement about a model's ability
//     to pick a type, and it is paid for by every reader.
async function typedRecordCheck(args: {
  from: string;
  to?: string;
  text: string;
  record?: MessageRecord;
}): Promise<{ ok: true; warning?: string } | { ok: false; error: string }> {
  if (args.record?.type) return { ok: true };

  const reg = await readJson<AgentRegistry>(AGENTS_FILE, {});

  // David-facing prose stays prose. An UNREGISTERED recipient is treated as an
  // agent, not as a human: the safe reading of "I cannot tell" is the rule, and
  // a human on this bus has a registry entry (that is how the pane is found).
  // ⟨q-178878aa⟩ — and the human need not be a REGISTERED agent: a registry entry for a
  // human is evicted after EVICT_MS (no heartbeat), so the exemption keys on the durable
  // human set the server knows (humans.json + AGENT_COORD_HUMANS), read here, at send time.
  if (args.to && (isHuman(reg[args.to]) || (await isKnownHuman(args.to)))) return { ok: true };

  const sender = reg[args.from];
  if (sender?.proseOnly) return { ok: true };

  const suggestion = suggestRecordType(args.text, recordAuthorityFor(sender).mayNotEmit);
  const guidance = typedRecordGuidance(suggestion);

  if (typedRecordMode() === "warn") {
    return {
      ok: true,
      warning:
        `UNTYPED — stored, but this send is REFUSED from ${TYPED_RECORD_CUTOVER_ISO}. ` +
        `${guidance} Until then an untyped multi-line message arrives in full in every reader's context ` +
        `instead of one line plus a retrieve_message handle. ` +
        `A model that cannot pick a type declares proseOnly:true at join (per-agent, visible in list_agents).`,
    };
  }
  return {
    ok: false,
    error:
      `agent→agent messages must carry a typed record (since ${TYPED_RECORD_CUTOVER_ISO}) — nothing was written. ` +
      `${guidance} ` +
      `Types: decision · verdict · done · blocker · risk · fyi · action · go · scope; 'fyi' is the honest ` +
      `catch-all — do not force a false 'decision'/'risk' to get past this. ` +
      `Messages TO a human are exempt (a recipient registered with a human role, or an id the server knows as human: ` +
      `humans.json / AGENT_COORD_HUMANS — see list_agents.humans), and an agent that cannot pick a type declares proseOnly:true at join.`,
  };
}

export const sendMessageSchema = {
  from: z.string().min(1),
  to: z.string().optional(),
  room: z.string().optional(),
  // Optional ONLY so a record can fill it (Task 3.3). This relaxes a
  // constraint rather than adding one, so v1 senders are unaffected; a call
  // with neither `text` nor a renderable `record` is rejected in the tool.
  text: z.string().min(1).optional(),
  kind: z.enum(["decision", "status", "chatter"]).optional(),
  record: messageRecordSchema.optional(),
  // Additive reply link. Validated in the tool (malformed → {ok:false};
  // unknown → store + warn) so the rejection shape matches identity-binding
  // / unknown-recipient, not a schema throw. Existing records unchanged.
  inReplyTo: z.string().optional(),
  // ⟨q-cc0819dc⟩ — needed only for a `done` cited by COMMIT: the repository whose
  // origin/main must carry the sha (one local git call, no network).
  repo: z.string().optional(),
};

const MESSAGE_ID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

async function messageIdExists(id: string): Promise<boolean> {
  const files: string[] = [];
  for (const n of await listInboxFiles()) files.push(path.join(INBOX_DIR, n));
  files.push(ROOM_FILE);
  if (existsSync(ROOMS_DIR)) {
    for (const n of await fsp.readdir(ROOMS_DIR)) {
      if (n.endsWith(".jsonl")) files.push(path.join(ROOMS_DIR, n));
    }
  }
  if (existsSync(ARCHIVE_INBOX_DIR)) {
    for (const n of await fsp.readdir(ARCHIVE_INBOX_DIR)) {
      if (n.endsWith(".jsonl")) files.push(path.join(ARCHIVE_INBOX_DIR, n));
    }
  }
  if (existsSync(ARCHIVE_ROOMS_DIR)) {
    for (const n of await fsp.readdir(ARCHIVE_ROOMS_DIR)) {
      if (n.endsWith(".jsonl")) files.push(path.join(ARCHIVE_ROOMS_DIR, n));
    }
  }
  for (const f of files) {
    const entries = await readJsonl<Message>(f);
    if (entries.some((m) => m.id === id)) return true;
  }
  return false;
}

// q-314e0187: `process.pid` is this MCP process's own pid — the process
// writing the entry right now, always known, no lookup needed. It is what a
// disputed message can be cross-referenced against (sessions/*.json,
// `ps -p <pid>`) when two live sessions hold one identity. In stdio mode this
// pid uniquely identifies the session (one process per session); in HTTP mode
// many sessions can share a process, so it identifies the SERVER handling the
// call rather than the caller — still strictly more than nothing, and the
// same scoping `trackSession` already applies to session markers.
function messageProvenance(): { pid: number; tmuxPane?: string } {
  return {
    pid: process.pid,
    ...(process.env.TMUX_PANE ? { tmuxPane: process.env.TMUX_PANE } : {}),
  };
}

export async function sendMessageTool(args: {
  from: string;
  to?: string;
  room?: string;
  text?: string;
  kind?: "decision" | "status" | "chatter";
  record?: MessageRecord;
  inReplyTo?: string;
  repo?: string;
}) {
  // Record authority first — a sender who may not emit this type is refused
  // before any other check runs, so a rejected record writes nothing anywhere.
  const authority = await checkRecordAuthority(args.from, args.record);
  if (!authority.ok) return { ok: false as const, error: authority.error };
  // ⟨q-dcbaf544⟩ — a scribe is the sender by definition; a verdict whose
  // `scribe` names someone else has its two positions crossed and would be
  // attributed wrongly by every reader. Refused as a value, nothing written.
  const scribeCheck = checkVerdictScribe(args.from, args.record);
  if (!scribeCheck.ok) return { ok: false as const, error: scribeCheck.error };

  // ⟨q-e5cb3538⟩ — INGRESS POLICY: REFUSE, don't rewrite. A control byte in content is a
  // keystroke in every pane the message is typed into (CR submits, ESC interrupts, ETX is
  // Ctrl-C). Refused rather than silently escaped because the stored message is what gets
  // cited and replayed; a rewrite would make it differ from what the sender wrote, and the
  // sender is the only party who can say what the byte was for. Line feed and tab pass.
  // Render-time escaping (hooks/tier.mjs) still covers history stored before this check.
  const control = findControlByte({ from: args.from, to: args.to, room: args.room, text: args.text, record: args.record, repo: args.repo });
  if (control) {
    return {
      ok: false as const,
      error:
        `refused: '${control.path}' carries a control character (${control.escape}) at offset ${control.offset}. ` +
        "Delivered to a pane it would be typed as a keystroke in the recipient's session. " +
        "Remove it, or write it out visibly (e.g. \\r); quoted external text (PR bodies, commit messages) is the usual carrier. Nothing was written.",
    };
  }

  // A `done` must cite the work it claims. Presence and shape only — resolving
  // the ref against gh/git is a consumer's job, and the send path makes no
  // network calls. Rejected as a value, not a throw, mirroring the
  // identity-binding rejection in src/server.ts.
  if (args.record?.type === "done") {
    // ⟨q-fee7239f⟩ — the same check that reads the closing citation reads the
    // closing GRAMMAR: a merge closing may not assert a deletion it has not
    // read. Refused as a value, naming the two accepted forms; nothing written.
    const closing = checkClosingLine(String(args.text ?? ""));
    if (!closing.ok) return { ok: false as const, error: closing.error };
    const cites = args.record.cites ?? [];
    const hasPr = cites.some((c) => c.kind === "pr" && c.ref.trim().length > 0);
    const commitCites = cites.filter((c) => c.kind === "commit");
    if (!hasPr && commitCites.length === 0) {
      return {
        ok: false as const,
        error:
          "a 'done' record must carry at least one {kind:'pr'} citation — an uncited DONE is an unverifiable claim. " +
          "Work that has no PR by rule (docs pushed straight to the shared branch) cites {kind:'commit', ref:<full 40-hex sha>} " +
          "and passes `repo`, the repository whose origin/main carries it.",
      };
    }
    // ⟨q-cc0819dc⟩ — a COMMIT cite satisfies a `done` only when the commit is real and on the
    // shared branch: every commit cite is verified, so a fabricated or short sha is refused
    // by name and nothing is written. A PR cite beside it does not excuse a bad commit cite.
    for (const c of commitCites) {
      const v = verifyCommitCite(c.ref, args.repo);
      if (!v.ok) return { ok: false as const, error: `a 'done' record cited by commit was refused — ${v.why}` };
    }
  }

  // `text` is what every consumer reads, so it must exist. The author's
  // wording ALWAYS wins: a record renders only to fill an absent text, never
  // to overwrite one.
  const text = args.text ?? (args.record ? renderRecord(args.record) : null);
  if (!text) {
    return {
      ok: false as const,
      error: args.record
        ? `record type '${args.record.type}' has no payload to render — supply 'text', or a payload the type's layout can render`
        : "'text' is required when no record is supplied",
    };
  }

  // After `text` is resolved (a record can fill it) and before anything is
  // written, so a refusal leaves nothing on disk.
  const typed = await typedRecordCheck({ from: args.from, to: args.to, text, record: args.record });
  if (!typed.ok) return { ok: false as const, error: typed.error };
  const typedWarning = typed.warning;

  let replyWarning: string | undefined;
  if (args.inReplyTo !== undefined) {
    if (!MESSAGE_ID_RE.test(args.inReplyTo)) {
      return {
        ok: false as const,
        error: `inReplyTo '${args.inReplyTo}' is not a message id (expected uuid)`,
      };
    }
    if (!(await messageIdExists(args.inReplyTo))) {
      replyWarning =
        `inReplyTo '${args.inReplyTo}' is not a known message — reply stored but no matching parent may exist`;
    }
  }

  // The tmux pusher's injection guard (hooks/tmux-pusher.mjs `shouldInject`)
  // never types leading-slash text into a pane — only `send_command`'s
  // allowlist goes in raw. The message is still stored and readable, so say so
  // here instead of letting the sender assume it landed on screen.
  const slashWarning = /^\s*\//.test(text)
    ? "text starts with '/' — stored, but tmux-push transports never type leading-slash text into a pane (injection guard); prefix it (e.g. `AGENT_ACTION: run /<skill> …`) or use `send_command` for /clear and /compact"
    : undefined;

  // DM → inbox. Otherwise resolve the channel (default `general`), make sure it
  // exists in the registry, and tag the message with its channel.
  if (args.to) {
    // A ROOM NAME IN `to:` IS REFUSED, AND THE GUARD SITS ABOVE THE WRITE.
    //
    // `to: "<room>"` used to append the message to `inbox/<room>.jsonl` — a file
    // nothing reads — and return `ok: true`. A coordinator's CANON post landed
    // nowhere and it took a stranded file on disk to find it. The ordering IS the
    // defect: the append ran BEFORE the registry lookup, so a refusal added after
    // the existing check would still have left the message written.
    //
    // REFUSE rather than warn-and-store, and the distinction is not fussiness: a
    // typo'd agent id is a plausible mistake worth a warning, because the id might
    // become real. A room name is a CATEGORY ERROR the server can identify with
    // certainty — rooms and agents are different namespaces and the confusion is
    // one keystroke. Measured when this shipped: zero overlap between room names
    // and registered agent ids, so a refusal cannot misfire on a real recipient.
    const rooms = await getRooms();
    if (Object.prototype.hasOwnProperty.call(rooms, normalizeRoom(args.to))) {
      return {
        ok: false as const,
        error:
          `'${args.to}' is a ROOM, not an agent — nothing was written. ` +
          `A DM to a room name lands in an inbox no one reads and would have returned ok. ` +
          `Post to the channel with room:'${normalizeRoom(args.to)}' (omit 'to'), or name an agent id.`,
      };
    }
    const msg: Message = {
      id: randomUUID(),
      ts: Date.now(),
      from: args.from,
      to: args.to,
      text,
      ...(args.record ? { record: args.record } : {}),
      ...(args.inReplyTo ? { inReplyTo: args.inReplyTo } : {}),
      provenance: messageProvenance(),
    };
    const target = inboxFile(args.to);
    await appendJsonl(target, msg);
    // Phase 5.4 Task 4 — a herdr-attached recipient has no pusher: deliver now, in-process.
    const herdr = await deliverToHerdrSeats(msg, [args.to], { kind: "dm" });
    void herdr;
    // Offline delivery is intentional (the inbox is created on demand), but a
    // typo'd recipient shouldn't vanish silently — surface a warning when the
    // target isn't a known agent so the caller can catch the mistake.
    const reg = await readJson<AgentRegistry>(AGENTS_FILE, {});
    const recipientWarning = reg[args.to]
      ? undefined
      : (await isKnownHuman(args.to))
        ? undefined // a human is not an agent and is not expected to be registered; the inbox is theirs to read
        : `recipient '${args.to}' is not a registered agent — message stored in their inbox but no one may be listening`;
    const warning = [typedWarning, recipientWarning, replyWarning, slashWarning].filter(Boolean).join("; ") || undefined;
    return { ok: true, id: msg.id, target, room: undefined, warning };
  }

  const chan = normalizeRoom(args.room);
  if (chan !== DEFAULT_ROOM) await ensureRoom(chan, args.from);
  const msg: Message = {
    id: randomUUID(),
    ts: Date.now(),
    from: args.from,
    room: chan,
    text,
    ...(args.kind ? { kind: args.kind } : {}),
    ...(args.record ? { record: args.record } : {}),
    ...(args.inReplyTo ? { inReplyTo: args.inReplyTo } : {}),
    provenance: messageProvenance(),
  };
  const target = roomFile(chan);
  await appendJsonl(target, msg);
  // Phase 5.4 Task 4 — herdr-attached members with rooms on receive the post now, in-process.
  const roomMembers = (await getRooms())[chan]?.members ?? [];
  const herdrRoom = await deliverToHerdrSeats(msg, roomMembers, { kind: "room", chan });
  void herdrRoom;
  await maybeCompactRoom(chan);
  const roomWarning = [typedWarning, replyWarning, slashWarning].filter(Boolean).join("; ") || undefined;
  return { ok: true, id: msg.id, target, room: chan, ...(roomWarning ? { warning: roomWarning } : {}) };
}

// ---------- live compaction (self-limiting streams) ----------

// Rooms and the status stream compact themselves on write: once a live file
// grows past its threshold, the oldest entries move to the archive (never
// deleted) and every cursor shifts down. Fresh decisions (< decisionDays old)
// are exempt — they stay in the live file. Cursor adjustment subtracts the
// full removed count, so an agent parked behind a kept decision may re-read
// it once; at-least-once beats silently skipping.
function envInt(name: string, fallback: number): number {
  const n = Number(process.env[name]);
  return Number.isFinite(n) && n > 0 ? Math.floor(n) : fallback;
}

const ROOM_COMPACT_THRESHOLD = envInt("AGENT_COORD_ROOM_MAX", 1000);
const ROOM_COMPACT_KEEP = envInt("AGENT_COORD_ROOM_KEEP", 500);
const STATUS_COMPACT_THRESHOLD = envInt("AGENT_COORD_STATUS_MAX", 2000);
const STATUS_COMPACT_KEEP = envInt("AGENT_COORD_STATUS_KEEP", 1000);
// Skip the entry count entirely while the file is small — the common case on
// every send. ~150 bytes/entry means the threshold can't be hit below this.
const COMPACT_SIZE_GATE = 100 * 1024;
const DECISION_FRESH_MS = 30 * 24 * 60 * 60 * 1000;

async function maybeCompactRoom(chan: string): Promise<void> {
  const file = roomFile(chan);
  if ((await fileSize(file)) < COMPACT_SIZE_GATE) return;
  const entries = await readJsonl<Message>(file);
  if (entries.length <= ROOM_COMPACT_THRESHOLD) return;
  const boundaryTs = entries[entries.length - ROOM_COMPACT_KEEP]!.ts;
  const decisionCutoff = Date.now() - DECISION_FRESH_MS;
  const r = await archiveJsonl<Message>(
    file,
    archiveRoomFile(chan),
    (e) => e.ts >= boundaryTs || (isDecision(e) && e.ts > decisionCutoff)
  );
  if (r.removed > 0) await adjustCursors({ roomRemovedByChan: { [chan]: r.removed } });
}

async function maybeCompactStatus(): Promise<void> {
  if ((await fileSize(STATUS_FILE)) < COMPACT_SIZE_GATE) return;
  const entries = await readJsonl<StatusEntry>(STATUS_FILE);
  if (entries.length <= STATUS_COMPACT_THRESHOLD) return;
  const boundaryTs = entries[entries.length - STATUS_COMPACT_KEEP]!.ts;
  const r = await archiveJsonl<StatusEntry>(
    STATUS_FILE,
    ARCHIVE_STATUS_FILE,
    (e) => e.ts >= boundaryTs
  );
  if (r.removed > 0) await adjustCursors({ statusRemoved: r.removed });
}

// ---------- read_messages ----------

export const readMessagesSchema = {
  agentId: z.string().min(1),
  source: z.enum(["inbox", "room", "status"]),
  room: z.string().optional(),
  limit: z.number().int().positive().max(500).optional(),
  peek: z.boolean().optional(),
  sinceTs: z.number().optional(),
};

export async function readMessagesTool(args: {
  agentId: string;
  source: Source;
  room?: string;
  limit?: number;
  peek?: boolean;
  sinceTs?: number;
}) {
  const file = sourceFile(args.source, args.agentId, args.room);
  const all = await readJsonl<Message | StatusEntry>(file);

  let entries: (Message | StatusEntry)[] = [];
  let totalNew = 0;

  // Room and status reads default to 50 entries to prevent agents flooding
  // themselves with full history on join — the status stream grows unbounded
  // across the fleet. Inbox drains fully since it is targeted by nature.
  const effectiveLimit = args.limit ?? (args.source === "inbox" ? undefined : 50);

  if (args.peek) {
    const cursor = await readJson<Cursor>(cursorFile(args.agentId), {});
    const startOffset = getOffset(cursor, args.source, args.room);
    entries = all.slice(startOffset);
    if (args.sinceTs !== undefined) entries = entries.filter((e) => e.ts > args.sinceTs!);
    totalNew = entries.length;
  } else {
    await updateJson<Cursor>(cursorFile(args.agentId), {}, (current) => {
      const startOffset = getOffset(current, args.source, args.room);
      let e = all.slice(startOffset);
      if (args.sinceTs !== undefined) e = e.filter((x) => x.ts > args.sinceTs!);
      totalNew = e.length;
      entries = e;
      // Advance past EVERYTHING we account for here (recent window + any
      // overflow we stash below). The overflow is recoverable via the history
      // hash, so it must not requeue for the next read — that would re-flood.
      if (e.length > 0) setOffset(current, args.source, args.room, startOffset + e.length);
      return current;
    });
  }

  // CCR overflow handling (room and status sources). When the backlog exceeds
  // the window, return the RECENT slice raw and replace the older overflow with
  // a compact digest carrying a retrieval hash. The agent expands it on demand
  // via retrieve_room_history. Peek is side-effect-free, so it never stashes —
  // it reports the count and tells the agent to do a real read to get a hash.
  let recent = entries;
  let history: { digest: string; hash?: string; older: number } | undefined;
  if (args.source !== "inbox" && effectiveLimit && entries.length > effectiveLimit) {
    const overflow = entries.slice(0, entries.length - effectiveLimit);
    recent = entries.slice(entries.length - effectiveLimit);
    const stashKey = args.source === "room" ? normalizeRoom(args.room) : "status";
    if (args.peek) {
      history = { digest: digestOverflow(overflow, undefined), older: overflow.length };
    } else {
      const hash = await stashHistory(stashKey, args.agentId, overflow);
      history = { digest: digestOverflow(overflow, hash), hash, older: overflow.length };
    }
  } else if (effectiveLimit && entries.length > effectiveLimit) {
    // Inbox keeps the legacy oldest-first chunking (no stash) — targeted
    // messages must never be skipped over.
    recent = entries.slice(0, effectiveLimit);
  }

  // Drop the agent's own posts on shared channels — reading your own broadcast
  // back is never useful and confuses turn-based agents into self-replies.
  // Cursor has already advanced past them, so they won't reappear.
  const visible =
    args.source === "room" || args.source === "status"
      ? recent.filter((e) => entryAuthor(e) !== args.agentId)
      : recent;

  // Phase 5.3 Task 21.2 — ANNOTATE A MESSAGE THAT HAS ALREADY BEEN DELIVERED.
  //
  // The REMOTE pusher (scripts/coord-pusher.mjs) consumes the bus over the wire
  // and cannot read this host's receipts/, so it cannot see for itself that a
  // message it is about to paste was pasted before. Without this, the replay
  // marker would appear in local panes only — HALF THE FLEET, and the half a
  // reader could not identify from inside the pane, which is worse than no
  // marker because its absence would read as "live".
  //
  // The two routes compute the same DATA and share ONE renderer
  // (hooks/replay.mjs `replayMarker`), so neither pane can be handed a
  // different vocabulary for the same fact.
  const priorText = await fsp.readFile(receiptFile(args.agentId), "utf8").catch(() => "");
  const prior = priorDeliveries(priorText);
  const annotated = visible.map((e) => {
    const replay = "id" in e ? replayInfo(prior, (e as Message).id) : undefined;
    return replay ? { ...e, replay } : e;
  });

  return {
    ok: true,
    messages: annotated,
    totalNew,
    returned: annotated.length,
    room: args.source === "room" ? normalizeRoom(args.room) : undefined,
    ...(history ? { history } : {}),
  };
}

// Lossless summary of a stashed backlog slice: surfaces error/failure posts
// verbatim (the lines that usually matter most in a flood) and collapses the
// rest to counts. Mirrors headroom's content-aware digest, kept deliberately
// simple — the full originals are one retrieve_room_history call away.
function digestOverflow(over: (Message | StatusEntry)[], hash: string | undefined): string {
  const authors = new Set(over.map(entryAuthor).filter(Boolean));
  const errorRe = /\b(error|fatal|fail(ed|ure)?|panic|exception)\b/i;
  const errors = over.filter((m) => errorRe.test(JSON.stringify(m)));
  const first = over[0]?.ts;
  const last = over[over.length - 1]?.ts;
  const span =
    first && last && last > first ? ` over ${Math.round((last - first) / 60000)}m` : "";
  const parts = [
    `[${over.length} earlier message${over.length === 1 ? "" : "s"} compressed`,
    `${authors.size} agent${authors.size === 1 ? "" : "s"}${span}`,
  ];
  if (errors.length) parts.push(`${errors.length} error post${errors.length === 1 ? "" : "s"}`);
  const head = parts.join(", ");
  const tail = hash
    ? ` hash=${hash}] — call retrieve_room_history(hash="${hash}") to expand`
    : `] — read without peek to get an expandable hash`;
  // Decisions are the one thing a digest must not bury — quote them verbatim
  // (capped) below the summary line.
  const decisions = over.filter((m): m is Message => isDecision(m as Message));
  const quoted = decisions
    .slice(-5)
    .map((d) => `  [decision] ${d.from}: ${d.text.length > 200 ? d.text.slice(0, 200) + "…" : d.text}`);
  const decisionBlock = decisions.length
    ? `\n${quoted.join("\n")}${decisions.length > 5 ? `\n  (+${decisions.length - 5} earlier decisions in hash)` : ""}`
    : "";
  return head + tail + decisionBlock;
}

// ---------- retrieve_message (Phase 8 Task 6) ----------

export const retrieveMessageSchema = {
  agentId: z.string().min(1),
  id: z.string().min(1),
};

// Expand a digest handle back into the full typed record.
//
// NOT a cache. The handle is the message's own `id`, and this reads the source
// of truth — rooms/<chan>.jsonl or inbox/<agent>.jsonl. That matters for two
// reasons the CCR history cache could not offer: nothing is duplicated, and
// nothing expires. A stashed copy would have carried HISTORY_TTL_MS and become
// permanently unrecoverable after 30 minutes, which is exactly when a handle
// that outlived a /clear would be expanded.
//
// It also sidesteps the cursor: the pusher SHARES the cursor file with
// read_messages, so anything already delivered to a pane is behind the cursor
// and re-reading the channel would not return it. A by-id lookup never
// consults a cursor.
//
// AUTHORITY BY CONSTRUCTION: we only ever open files this agent is entitled to
// read — its own inbox, and the rooms it is a member of. There is no separate
// permission check that could disagree with the search, and a handle for a
// message delivered somewhere else simply is not found. That is the same
// property the history cache spent `forAgent` scoping to get.
export async function retrieveMessageTool(args: { agentId: string; id: string }) {
  const rooms = await memberRooms(args.agentId);
  const scopes = [
    {
      source: "inbox" as const,
      room: undefined as string | undefined,
      live: inboxFile(args.agentId),
      archived: archiveInboxFile(args.agentId),
    },
    ...rooms.map((r) => ({
      source: "room" as const,
      room: r as string | undefined,
      live: roomFile(r),
      archived: archiveRoomFile(r),
    })),
  ];

  // Live files first. The archive is opened ONLY on a miss, so no normal
  // retrieval touches it and compaction/prune semantics are unchanged — but a
  // record that compaction moved is still recoverable, because archive/ is
  // append-only and complete.
  for (const pass of ["live", "archived"] as const) {
    for (const s of scopes) {
      const entries = await readJsonl<Message>(s[pass]);
      const msg = entries.find((m) => m.id === args.id);
      if (msg) {
        return {
          ok: true as const,
          id: msg.id,
          source: s.source,
          room: s.room,
          archived: pass === "archived",
          message: msg,
          record: msg.record,
        };
      }
    }
  }
  return {
    ok: false as const,
    error: `no message '${args.id}' in any channel you can read — it may never have been delivered to you`,
  };
}

// ---------- retrieve_room_history ----------

export const retrieveRoomHistorySchema = {
  agentId: z.string().min(1),
  hash: z.string().min(1),
  query: z.string().optional(),
};

export async function retrieveRoomHistoryTool(args: {
  agentId: string;
  hash: string;
  query?: string;
}) {
  const res = await retrieveHistory<Message | StatusEntry>(args.hash, args.agentId, args.query);
  if (!res.ok) {
    const reason =
      res.reason === "expired"
        ? "That history entry has expired (30m TTL). Re-read the channel with a higher limit to fetch it again."
        : res.reason === "forbidden"
          ? "That history hash was produced for a different agent and cannot be retrieved by you."
          : "No history entry for that hash. It may have expired or never existed.";
    return { ok: false, reason: res.reason, message: reason };
  }
  return {
    ok: true,
    room: res.room,
    hash: args.hash,
    total: res.total,
    returned: res.messages.length,
    messages: res.messages,
  };
}

function entryAuthor(e: Message | StatusEntry): string | undefined {
  return "from" in e ? e.from : e.agentId;
}

// ---------- post_status ----------

export const postStatusSchema = {
  agentId: z.string().min(1),
  status: z.string().min(1),
  detail: z.string().optional(),
};

export async function postStatusTool(args: { agentId: string; status: string; detail?: string }) {
  const entry: StatusEntry = {
    id: randomUUID(),
    ts: Date.now(),
    agentId: args.agentId,
    status: args.status,
    detail: args.detail,
  };
  await appendJsonl(STATUS_FILE, entry);
  await maybeCompactStatus();
  return { ok: true, id: entry.id };
}

// ---------- wait_for_message ----------

export const waitForMessageSchema = {
  agentId: z.string().min(1),
  source: z.enum(["inbox", "room", "status"]),
  room: z.string().optional(),
  timeoutMs: z.number().int().positive().max(MAX_WAIT_MS).optional(),
};

export async function waitForMessageTool(args: {
  agentId: string;
  source: Source;
  room?: string;
  timeoutMs?: number;
}) {
  const totalTimeout = Math.min(args.timeoutMs ?? 30_000, MAX_WAIT_MS);
  const file = sourceFile(args.source, args.agentId, args.room);
  const deadline = Date.now() + totalTimeout;

  // Loop so that file growth caused only by the agent's own self-posts (which
  // readMessagesTool now filters out for room/status) doesn't return an empty
  // result — keep waiting until we have something to deliver or time out.
  while (Date.now() < deadline) {
    // ALREADY-UNREAD CONTENT WAKES THIS IMMEDIATELY.
    //
    // The loop below waits for the file to GROW from the size captured at entry,
    // so anything that arrived before the wait started could not wake it: the
    // agent sat through the full timeout with messages readable the whole time.
    // Seven real posts landed across consecutive 60s waits that all returned
    // `timedOut`, while `read_messages` returned every one.
    //
    // Growth is the wrong question on its own — "is there anything for me" is
    // the right one, and it is also the cheaper check.
    const pending = await readMessagesTool({ agentId: args.agentId, source: args.source, room: args.room });
    if (pending.messages.length > 0) return { ...pending, waited: true, timedOut: false };

    // 8.4 — THE SAME POSITION THE WATCH USES, not a second mechanism beside it.
    //
    // This compared byte SIZE, which answers "did the file grow" — a proxy for
    // "is there a new line". The watch keys on LINE OFFSET, which is the thing
    // itself, and reusing it means the wait and the watch cannot disagree about
    // what "new" means. A second mechanism beside the first is how two answers
    // to one question start drifting.
    //
    // It still consumes nothing by itself: the offset is read, never written
    // back to the cursor (8.2).
    const startLines = readLog(file).total;
    const startSize = await fileSize(file);
    const remaining = deadline - Date.now();
    if (remaining <= 0) break;

    const changed = await new Promise<boolean>((resolve) => {
      let settled = false;
      const finish = (v: boolean) => {
        if (settled) return;
        settled = true;
        clearInterval(poll);
        try {
          watcher?.close();
        } catch {
          // ignore
        }
        clearTimeout(t);
        resolve(v);
      };

      const check = async () => {
        // Line count first — a write that changes bytes without adding a line
        // (a rewrite, a truncation) is not new traffic, and size alone reports
        // it as such.
        if (readLog(file).total > startLines) return finish(true);
        const sz = await fileSize(file);
        if (sz > startSize) finish(true);
      };

      let watcher: ReturnType<typeof watch> | undefined;
      try {
        watcher = watch(file, () => {
          void check();
        });
      } catch {
        // file may not exist; polling will handle
      }
      const poll = setInterval(() => void check(), 500);
      const t = setTimeout(() => finish(false), remaining);
    });

    if (!changed) break;
    const result = await readMessagesTool({ agentId: args.agentId, source: args.source, room: args.room });
    if (result.returned > 0) return result;
    // otherwise, only self-posts arrived; keep waiting on the remaining budget
  }

  return { ok: false, timedOut: true };
}