/**
 * Orchestrator for mode 1 (terminal continuation) over injected ports.
 *
 * It owns the message pump: an authorized Telegram message becomes a prompt
 * turn, is queued, and is released to the agent as a follow-up only while the
 * agent is idle (so our lane queue keeps its edit-by-source window instead of
 * handing everything straight to Pi). When a turn finishes, the assistant's
 * reply is mirrored back to the bound chat and the next queued turn is pumped.
 *
 * All Pi/grammY specifics arrive as ports (isIdle, sendFollowUp, OutboundSender,
 * AbortRegistry), so this pump is unit-testable; `index.ts` wires the ports to
 * real `pi.on` handlers and command context.
 */
import type { Message, User } from "@grammyjs/types";
import { COMPLIANCE_LINKS, COMPLIANCE_NOTICE } from "../../constants";
import type { AbortRegistry } from "../../core/abort";
import { formatClock } from "../../core/datetime";
import {
	DEFAULT_FORWARD_POLICY,
	ForwardBursts,
	type ForwardPolicy,
	forwardLimitNote,
	limitForwardText,
} from "../../core/forwards";
import { TERMINAL_ORIGIN_MARKER } from "../../core/prompt-origin";
import { MessageQueue, type QueueItem } from "../../core/queue";
import { buildPromptTurn, type TurnSavedFile } from "../../core/turns";
import { buildRichMarkdownMessage } from "../../telegram/markdown";
import { isTopicAnchor } from "../../telegram/message-context";
import type { OutboundSender, OutboundTarget } from "../../telegram/outbound";
import {
	buildRichHtmlMessage,
	type RichHtml,
} from "../../telegram/rich-builder";
import {
	type ToolCallActivity,
	toolActivityMessage,
} from "../../telegram/tool-activity";
import type { TelegramEvent } from "../../telegram/updates";
import { bullet, card, link, note } from "./format";
import {
	type InboundImage,
	isServiceMessage,
	lastAssistantReply,
	messageText,
	messageToTurnInput,
	type PromptContent,
	parseSlashCommand,
} from "./messages";

export interface ConnectControllerDeps {
	/** Only messages from this Telegram user are accepted (their private chat id). */
	allowedUserId: number;
	/** Attachment size cap for describing inbound media. */
	maxBytes: number;
	/** Whether the agent is currently idle (not streaming). */
	isIdle: () => boolean;
	/** Deliver a prompt turn to the agent as a follow-up (no interruption). */
	sendFollowUp: (content: PromptContent) => Promise<void>;
	/**
	 * The moment the bot first SPEAKS in a turn (its first draft chunk, tool block or
	 * answer), with the messages that turn answers. This — not the moment the prompt is
	 * handed to the agent — is when a message typed outside the personal topic is copied
	 * into it (see the topic router's `mirrorStraysIntoPersonal`): a copy that lands
	 * while the model is still thinking sits alone in the topic for seconds, which reads
	 * as the bot echoing you rather than answering you.
	 */
	onTurnVisible?: (sourceMessageIds: readonly number[]) => Promise<void>;
	/** Download an inbound message's image attachments as base64 (best-effort). */
	loadImages?: (message: Message) => Promise<InboundImage[]>;
	/**
	 * Download an inbound message's non-image files to disk and report their
	 * paths (plus any per-file errors) so the model can open and reason about
	 * them. Images are delivered inline via {@link loadImages}, not saved here.
	 */
	saveAttachments?: (
		message: Message,
	) => Promise<{ savedFiles: TurnSavedFile[]; errors: string[] }>;
	/**
	 * Upload a file to the bound chat so the user receives it (the reverse of
	 * `saveAttachments`). Exactly one of `path`/`url`. Throws on failure so the
	 * calling tool can surface the exact error to the model.
	 */
	uploadFile?: (input: {
		path?: string;
		url?: string;
		caption?: string;
		/** Thread the upload under an existing message (the tool card it completes). */
		replyToMessageId?: number;
		/** Deliver it under this name instead of the one on disk. */
		filename?: string;
	}) => Promise<void>;
	/** Handle a `/clear` (or `/new`, `/reset`) request to wipe the agent's history. */
	onClear?: () => Promise<void>;
	/** Handle a `/esc` (or `/cancel`) request to interrupt the running turn. */
	onAbort?: () => Promise<void>;
	/**
	 * Handle a `/compact` request: summarise the history the model carries, so a long
	 * session keeps going instead of hitting the context window. The outcome is
	 * announced by the compaction cards, not by this call.
	 */
	onCompact?: () => Promise<void>;
	/**
	 * Build the `/status` card — what the session is doing right now (model, context,
	 * directory, queue). Synchronous on purpose: it reports state, it does not go and
	 * fetch any.
	 */
	onStatus?: () => string;
	/**
	 * Build the `/context` card — what the model is actually carrying, and what put it
	 * there. `/status` says how full the context is; this says what it is full OF.
	 */
	onContextReport?: () => string;
	/** Record/refresh the sender's profile in the contact store (best-effort). */
	onContact?: (user: User) => Promise<void>;
	/**
	 * Resolve what a replayed session-history card stood for, keyed by the id of the card
	 * the owner replied to: the text it showed and any images that message carried. The
	 * cards are display-only bot posts, so a reply to one otherwise carries an empty quote
	 * (`which said: ""`) and no picture; this hands back the real text (for the quote) and
	 * the pictures (re-delivered inline) so the model can answer about it. Undefined for a
	 * message that is not one of our anchored history cards.
	 */
	resolveHistoryAnchor?: (
		messageId: number,
	) => { text: string; images: InboundImage[] } | undefined;
	/**
	 * The owner's `personal` topic, resolved per send (it appears once the router has
	 * created it, and vanishes again on a fallback to the plain DM). EVERYTHING of this
	 * conversation goes there — prompts, replies, and the tool calls the model made for
	 * the owner, which are part of watching it work, not manager noise. Undefined → the
	 * plain DM, i.e. the pre-topics behaviour.
	 */
	chatThread?: () => number | undefined;
	/**
	 * How many images one turn may carry (an album folds into a single turn). `0` or
	 * undefined = no cap; the reason to cap is a small local context, not any Pi limit.
	 */
	maxImages?: number;
	/**
	 * Quiet window that closes an album (Telegram delivers one message per photo, a
	 * few tens of ms apart). Default 1500 ms — long enough for a slow group, short
	 * enough that a single photo is not noticeably delayed.
	 */
	albumWindowMs?: number;
	/**
	 * Budget for FORWARDED messages — how much of a forward is read, and how many of
	 * one batch. Defaults to {@link DEFAULT_FORWARD_POLICY} when not injected.
	 */
	forwards?: ForwardPolicy;
	/** Timer ports, injected so album batching is testable without real time. */
	setTimer?: (fn: () => void, ms: number) => unknown;
	clearTimer?: (handle: unknown) => void;
	/** Clock port, injected so forward batching is testable without real time. */
	clock?: { now(): number };
	/**
	 * IANA zone for the arrival time stamped into each turn's header (system zone
	 * when unset). The model is told when a message reached it inside the message
	 * itself — never as a message of its own; see `core/connect-context.ts`.
	 */
	timezone?: string;
	outbound: OutboundSender;
	abort: AbortRegistry;
}

/** See {@link ConnectControllerDeps.albumWindowMs}. */
export const DEFAULT_ALBUM_WINDOW_MS = 1500;

// Telegram bot commands the bridge handles itself instead of forwarding to the
// agent. Everything else (including /start) falls through as an ordinary prompt.
const CLEAR_COMMANDS = new Set(["clear", "new", "reset"]);
const ABORT_COMMANDS = new Set(["esc", "cancel"]);
const COMPACT_COMMANDS = new Set(["compact"]);
const STATUS_COMMANDS = new Set(["status"]);
const CONTEXT_COMMANDS = new Set(["context"]);

const HELP_COMMANDS = new Set(["help"]);
const START_COMMANDS = new Set(["start"]);
/**
 * The commands the bridge answers entirely from its own state: they never reach the
 * model, never touch the session, and never change what any mode is doing.
 *
 * Mixed mode is why this needs a name. Anything the owner writes in the bot DM takes
 * the brain back for coding — which ABORTS a manager turn in flight. That is right for
 * a message (the owner outranks a stranger who is waiting) and wrong for a question
 * ABOUT the bot: ask how full the context is, and a stranger's half-written reply dies
 * for it. `/esc`, `/clear` and `/compact` are not here on purpose — each one is the
 * owner reaching into the session, and taking it is the point.
 */
const READ_ONLY_COMMANDS = new Set([
	...HELP_COMMANDS,
	...STATUS_COMMANDS,
	...CONTEXT_COMMANDS,
]);

/** See {@link READ_ONLY_COMMANDS}. */
export function isReadOnlyBridgeCommand(text: string | undefined): boolean {
	const command = parseSlashCommand(text ?? "");
	return command !== null && READ_ONLY_COMMANDS.has(command.name);
}

/** The extension's repository and its Tangled mirror, shown in the help footer. */
export const REPO_URL = "https://github.com/m62624/pi-telegram-manager";
export const MIRROR_URL =
	"https://tangled.org/m62624.tngl.sh/pi-telegram-manager";

/** Static help shown for `/help`, mirroring the Telegram command menu. */
const HELP_TEXT = card("🧭", "Pi Telegram bridge", [
	bullet("/switch", "change mode — manager / personal / mixed"),
	bullet(
		"/resume",
		"pick which session personal runs in — current / new / resume",
	),
	bullet("/stop", "stop the bot entirely"),
	bullet("/esc", "cancel the current turn"),
	bullet("/clear", "clear the conversation history"),
	bullet("/compact", "summarise the history to free up context"),
	bullet("/status", "model, context, working directory, queue"),
	bullet("/context", "what I am carrying, and what filled it up"),
	bullet("/start", "privacy & terms — read before using"),
	bullet("/help", "show this help"),
	"",
	// In code, deliberately: with entity detection on, a bare `/telegram-personal` would
	// render as a button — and tapping it would send the text to the BOT, which does not
	// know that command and would hand it to the model as a prompt. A command that is not
	// yours to press should not look pressable.
	note(
		"Terminal commands (`/telegram-personal`, `-manager`, `-mixed`) run in Pi, not here.",
	),
	`⚠️ Terms you must follow: ${link("bot developers", COMPLIANCE_LINKS.botTerms)} · ${link("privacy", COMPLIANCE_LINKS.privacy)} · ${link("secretary/business", COMPLIANCE_LINKS.business)}`,
	`This bot runs ${link("pi-telegram-manager", REPO_URL)} · ${link("mirror", MIRROR_URL)}`,
]);

export class ConnectController {
	private readonly queue = new MessageQueue();
	private turnCounter = 0;
	/** Non-zero id of the current streaming draft; 0 when none is active. */
	private draftId = 0;
	/** Monotonic source of draft ids, so each message animates as its own draft. */
	private draftCounter = 0;
	/** Tool cards awaiting their outcome: toolCallId → the message to edit, and what it said. */
	private readonly toolCards = new Map<
		string,
		{ messageId: number; activity: ToolCallActivity }
	>();
	/** Albums still being collected: media_group_id → the queued turn it folds into. */
	private readonly albums = new Map<string, string>();
	private readonly albumTimers = new Map<string, unknown>();
	/** The open forward batch, so a burst of forwards reaches the model as one turn. */
	private readonly forwards: ForwardBursts;

	constructor(private readonly deps: ConnectControllerDeps) {
		this.forwards = new ForwardBursts(this.forwardPolicy);
	}

	private get forwardPolicy(): ForwardPolicy {
		return this.deps.forwards ?? DEFAULT_FORWARD_POLICY;
	}

	private now(): number {
		return this.deps.clock?.now() ?? Date.now();
	}

	/** Fold a line into an open batch, or open a new turn with it. */
	private async appendOrEnqueue(
		groupId: string,
		text: string,
		sourceMessageId: number,
	): Promise<void> {
		const open = this.albums.get(groupId);
		if (
			open !== undefined &&
			this.queue.appendToItem(open, { text, sourceMessageId })
		) {
			this.armAlbumFlush(groupId);
			return;
		}
		const id = `turn-${this.turnCounter++}`;
		this.queue.enqueue({
			id,
			lane: "default",
			text,
			sourceMessageIds: [sourceMessageId],
		});
		this.albums.set(groupId, id);
		this.armAlbumFlush(groupId);
	}

	private get target(): OutboundTarget {
		return {
			chatId: this.deps.allowedUserId,
			messageThreadId: this.deps.chatThread?.(),
		};
	}

	/**
	 * The running turn's source messages, until the bot first speaks (see
	 * {@link ConnectControllerDeps.onTurnVisible}). Null once they have been handed over
	 * — one turn copies its prompt once.
	 */
	private pendingMirror: readonly number[] | null = null;

	/** Hand the turn's prompt over right before the first thing the bot says. */
	private async flushMirror(): Promise<void> {
		const sources = this.pendingMirror;
		if (!sources) return;
		this.pendingMirror = null;
		await this.deps.onTurnVisible?.(sources);
	}

	/**
	 * Deliver the model's text into the personal topic.
	 *
	 * A message the owner typed in another topic is not answered where it was typed:
	 * Telegram cannot move it, and quoting it across topics is not something the
	 * clients agree on (phone, desktop and web each rendered the quote differently,
	 * and each was wrong in its own way). Instead the message itself is FORWARDED
	 * into the personal topic before the model sees it (see the topic router), so the
	 * conversation there stays whole and the answer needs no trick at all.
	 */
	private async sendAnswer(text: string): Promise<void> {
		await this.flushMirror();
		await this.deps.outbound.sendMarkdown(this.target, text);
	}

	/** Handle an inbound Telegram event. Returns true when it enqueued/edited a turn. */
	async onEvent(event: TelegramEvent): Promise<boolean> {
		if (event.kind !== "message" && event.kind !== "edited_message")
			return false;
		if (event.fromId !== this.deps.allowedUserId) return false;
		// A service message (you created a topic, pinned something) has no content: it
		// is not a prompt, and forwarding it woke the model with an empty turn.
		if (isServiceMessage(event.message)) return false;

		// Capture/refresh the sender's profile (name, username, …) for the contact
		// store — used now for a unified record and later relayed by the manager.
		if (this.deps.onContact && event.message.from) {
			void this.deps.onContact(event.message.from).catch(() => {});
		}

		// Intercept the bridge's own control commands (e.g. /clear) so they never
		// reach the agent as a prompt. Unknown commands (and /start, /help) fall
		// through and are treated as ordinary messages.
		if (await this.tryControlCommand(event.message)) return true;

		// Acknowledge receipt immediately with a "typing…" hint, before the agent
		// even starts (there is queue/dispatch latency in between).
		void this.sendTyping();

		// A FORWARD is a batch, not a message: forwarding five posts sends five
		// messages, each of any length. They are one act — so they fold into one turn
		// (like an album), and their bodies are capped by the forward budget, which is
		// separate from the ordinary message policy on purpose.
		const forward = this.forwards.track(
			String(this.deps.allowedUserId),
			event.message.forward_origin !== undefined,
			this.now(),
			event.message.media_group_id,
		);
		if (forward?.overLimit) {
			// Past the batch limit: the body is not read. Say so once, then drop the rest
			// in silence — repeating the note would be the flood the limit exists to stop.
			// The note joins the batch's own turn (its album, when it is one), so it is
			// read next to what WAS forwarded rather than as a turn of its own.
			if (forward.justHitLimit) {
				await this.appendOrEnqueue(
					event.message.media_group_id ?? forward.key,
					forwardLimitNote(this.forwardPolicy.maxMessages),
					event.message.message_id,
				);
			}
			return true;
		}

		// Save non-image files to disk (best-effort) so the model gets real paths;
		// images ride along inline via loadImages.
		const intake = await this.saveAttachments(event.message);
		const input = messageToTurnInput(event.message, this.deps.maxBytes);
		// A reply to a replayed history card carries an empty quote and no picture (the
		// cards are display-only). Fill the quote with what that message actually said,
		// and its images are re-delivered inline below, so the model answers about the
		// real thing rather than about "".
		const anchor = this.historyAnchorFor(event.message);
		const withReply = anchor
			? { ...input, reply: { text: anchor.text } }
			: input;
		const turn = buildPromptTurn({
			...withReply,
			text:
				forward && withReply.text
					? limitForwardText(withReply.text, this.forwardPolicy.maxChars)
					: withReply.text,
			receivedAt: formatClock(this.now(), this.deps.timezone),
			savedFiles: intake.savedFiles.length > 0 ? intake.savedFiles : undefined,
			attachmentErrors: intake.errors.length > 0 ? intake.errors : undefined,
		});

		if (
			event.kind === "edited_message" &&
			this.queue.editBySource(event.message.message_id, turn)
		) {
			return true; // still queued — rewrote it in place, no new dispatch
		}

		const ownImages = await this.loadImages(event.message);
		// Re-deliver the pictures of an anchored history card the owner replied to, so
		// "how many people are in this photo?" about a replayed image actually works.
		const images =
			anchor && anchor.images.length > 0
				? [...ownImages, ...anchor.images]
				: ownImages;

		// An ALBUM is not one message: Telegram splits it into one message per photo
		// (sharing a media_group_id), with your caption only on the first. Enqueued as
		// they arrive, five pictures became five turns — the model answering each
		// picture alone, four of them without your words. So an album is collected into
		// ONE turn: the first message opens it, the rest fold into it while it is still
		// queued, and it is dispatched once the group stops growing. A forward batch is
		// held open the same way, keyed by its burst instead of a media group.
		const groupId = event.message.media_group_id ?? forward?.key;
		if (groupId !== undefined) {
			const open = this.albums.get(groupId);
			if (open !== undefined) {
				const folded = this.queue.appendToItem(open, {
					maxImages: this.deps.maxImages,
					// Only the first message of an album carries your caption; the rest add
					// nothing but their picture, so their bare "[photo]" header is dropped.
					// Every forwarded message, by contrast, carries its own body — that IS
					// the content you forwarded, so all of them are kept.
					text:
						event.message.media_group_id === undefined || event.message.caption
							? turn
							: undefined,
					images,
					sourceMessageId: event.message.message_id,
				});
				if (folded) {
					this.armAlbumFlush(groupId);
					return true;
				}
				this.albums.delete(groupId);
			}
		}

		const id = `turn-${this.turnCounter++}`;
		this.queue.enqueue({
			id,
			lane: "default",
			text: turn,
			images: images.length > 0 ? images : undefined,
			sourceMessageIds: [event.message.message_id],
		});
		if (groupId !== undefined) {
			// Hold the album open: dispatch waits until no further photo has arrived for
			// `albumWindowMs`, so the whole group reaches the model in one turn.
			this.albums.set(groupId, id);
			this.armAlbumFlush(groupId);
			return true;
		}
		await this.dispatch();
		return true;
	}

	/** (Re)start the quiet window that closes an album and releases it to the agent. */
	private armAlbumFlush(groupId: string): void {
		const setTimer =
			this.deps.setTimer ??
			((fn: () => void, ms: number) => setTimeout(fn, ms) as unknown);
		const clearTimer =
			this.deps.clearTimer ??
			((handle: unknown) =>
				clearTimeout(handle as ReturnType<typeof setTimeout>));
		const previous = this.albumTimers.get(groupId);
		if (previous !== undefined) clearTimer(previous);
		this.albumTimers.set(
			groupId,
			setTimer(() => {
				this.albums.delete(groupId);
				this.albumTimers.delete(groupId);
				void this.dispatch();
			}, this.deps.albumWindowMs ?? DEFAULT_ALBUM_WINDOW_MS),
		);
	}

	/**
	 * Run a bridge control command if the message is one we handle. Returns true
	 * when the message was consumed as a command (and must not be forwarded to
	 * the agent). `/clear`, `/new`, `/reset` wipe the agent's history.
	 */
	private async tryControlCommand(message: Message): Promise<boolean> {
		const command = parseSlashCommand(messageText(message));
		if (!command) return false;
		if (CLEAR_COMMANDS.has(command.name) && this.deps.onClear) {
			await this.deps.onClear();
			return true;
		}
		if (ABORT_COMMANDS.has(command.name) && this.deps.onAbort) {
			await this.deps.onAbort();
			return true;
		}
		if (COMPACT_COMMANDS.has(command.name) && this.deps.onCompact) {
			await this.deps.onCompact();
			return true;
		}
		if (STATUS_COMMANDS.has(command.name) && this.deps.onStatus) {
			await this.sendToChat(this.deps.onStatus());
			return true;
		}
		if (CONTEXT_COMMANDS.has(command.name) && this.deps.onContextReport) {
			await this.sendToChat(this.deps.onContextReport());
			return true;
		}
		if (HELP_COMMANDS.has(command.name)) {
			await this.sendToChat(HELP_TEXT);
			return true;
		}
		// /start (incl. the Secretary deep link /start bizChat…) always shows the
		// privacy/compliance reminder, so the terms are surfaced on first contact.
		//
		// In the live wiring the router answers /start before the controller ever sees
		// it (it also re-posts the mode pin, which only the router can do). This stays as
		// the controller's own guarantee: whatever drives it, /start is never handed to
		// the model as a prompt, and the terms are never skipped.
		if (START_COMMANDS.has(command.name)) {
			await this.sendToChat(COMPLIANCE_NOTICE);
			return true;
		}
		return false;
	}

	/**
	 * What the replayed history card a message replied to stood for — its real text and
	 * images — keyed by the id the owner replied to. Undefined when the reply is not to
	 * one of our anchored cards, so ordinary reply handling is untouched.
	 */
	private historyAnchorFor(
		message: Message,
	): { text: string; images: InboundImage[] } | undefined {
		const repliedId = message.reply_to_message?.message_id;
		if (repliedId === undefined || !this.deps.resolveHistoryAnchor)
			return undefined;
		return this.deps.resolveHistoryAnchor(repliedId);
	}

	/**
	 * Download a message's inline images — and, when it REPLIES to one, the replied-to
	 * picture too. Replying to a photo is how a person says "look at this one" (including
	 * a photo the bot itself sent), exactly as {@link saveAttachments} does for files.
	 * Without this the model got no picture on a reply and confabulated one from the
	 * caption. The topic's root message is not something the owner "replied to" — Telegram
	 * attaches it to every message in the topic — so it is skipped. Per-file failures are
	 * swallowed; the text header still notes the attachment.
	 */
	private async loadImages(message: Message): Promise<InboundImage[]> {
		if (!this.deps.loadImages) return [];
		const own = await this.deps.loadImages(message).catch(() => []);
		const repliedTo = message.reply_to_message;
		if (!repliedTo || isTopicAnchor(message, repliedTo)) return own;
		const quoted = await this.deps
			.loadImages(repliedTo as Message)
			.catch(() => []);
		return quoted.length > 0 ? [...own, ...quoted] : own;
	}

	/**
	 * Save non-image attachments to disk, swallowing a wholesale failure.
	 *
	 * A REPLIED-TO message's files are saved too. Replying to a file is how a person
	 * says "this one" — including to a file the BOT sent, like the full-output log of
	 * a tool call. Without this the model only ever saw the reply's caption ("📄 full
	 * output — bash") and had no way to open the thing being pointed at, which makes
	 * the reply meaningless. Mode 1 only: the manager never reads files this way.
	 */
	private async saveAttachments(
		message: Message,
	): Promise<{ savedFiles: TurnSavedFile[]; errors: string[] }> {
		if (!this.deps.saveAttachments) return { savedFiles: [], errors: [] };
		const own = await this.deps
			.saveAttachments(message)
			.catch(() => ({ savedFiles: [], errors: [] }));
		const repliedTo = message.reply_to_message;
		// A topic's root message is not something the owner "replied to" — Telegram
		// attaches it to every message in the topic.
		if (!repliedTo || isTopicAnchor(message, repliedTo)) return own;
		const quoted = await this.deps
			.saveAttachments(repliedTo as Message)
			.catch(() => ({ savedFiles: [], errors: [] }));
		if (quoted.savedFiles.length === 0 && quoted.errors.length === 0)
			return own;
		return {
			savedFiles: [...own.savedFiles, ...quoted.savedFiles],
			errors: [...own.errors, ...quoted.errors],
		};
	}

	/**
	 * Upload a file to the bound chat so the user receives it. Called by the
	 * `telegram_attach` tool; propagates errors so the tool reports them.
	 */
	async sendFile(input: {
		path?: string;
		url?: string;
		caption?: string;
	}): Promise<void> {
		if (!this.deps.uploadFile) {
			throw new Error("file upload is not available in this session");
		}
		await this.deps.uploadFile(input);
	}

	/** Build the follow-up content for a queued turn: images (if any) then text. */
	private toContent(item: QueueItem): PromptContent {
		if (!item.images || item.images.length === 0) return item.text;
		return [
			...item.images.map((img) => ({
				type: "image" as const,
				data: img.data,
				mimeType: img.mimeType,
			})),
			{ type: "text" as const, text: item.text },
		];
	}

	/**
	 * Release the next queued turn to the agent, but only while it is idle.
	 *
	 * The turn is dropped from the queue only once the agent has actually TAKEN it. A
	 * hand-off can fail — the session may refuse to go idle, and then it is not a turn
	 * that was delivered, it is a turn that was lost. It stays queued instead, and the
	 * next pump tries again; that is the difference between a bot that catches up when
	 * the session recovers and a bot that quietly ate your message.
	 */
	async dispatch(): Promise<void> {
		if (!this.deps.isIdle()) return;
		const next = this.queue.peek();
		if (!next) return;
		// An album still collecting its photos is not ready: releasing it now would
		// send the model the first picture and orphan the rest.
		if (this.isOpenAlbum(next.id)) return;
		this.pendingMirror = next.sourceMessageIds;
		try {
			await this.deps.sendFollowUp(this.toContent(next));
		} catch {
			// Not delivered. Take the mirror back too — the prompt is copied into the topic
			// when the turn really starts, not when we hoped it would.
			this.pendingMirror = null;
			return;
		}
		this.queue.dequeue();
	}

	private isOpenAlbum(itemId: string): boolean {
		for (const id of this.albums.values()) if (id === itemId) return true;
		return false;
	}

	/** Arm interruption for the turn that just started. */
	onAgentStart(abortTurn: () => void): void {
		this.deps.abort.set(abortTurn);
	}

	/**
	 * Deliver one assistant message as it completes.
	 *
	 * A run is not one message: working through a task the model narrates ("Google
	 * blocked me, trying Bing"), calls tools, answers, and often adds a trailing
	 * "done, browser closed". Mirroring only the run's LAST assistant text — what
	 * agent_end used to do — delivered that trailing line and silently dropped the
	 * answer itself. So each assistant message goes out when it ends, in order, which
	 * is also the live trace of the model working.
	 */
	async deliverAssistant(text: string): Promise<void> {
		if (!text.trim()) return;
		await this.sendAnswer(text);
	}

	/**
	 * Close out the finished run and ask for the next queued turn.
	 *
	 * `fallbackReply` delivers the run's last assistant text, and is a SAFETY NET
	 * only: `index.ts` passes false once anything was already mirrored message by
	 * message (see {@link deliverAssistant}), and for a run that is not the owner's
	 * (mixed: a manager moderation turn shares this session, and its text belongs to
	 * the interlocutor).
	 *
	 * ASK, not take: {@link dispatch} hands nothing over unless `isIdle()` says the
	 * session can accept it, and when Pi calls this it cannot — `agent_end` is awaited
	 * from inside the run that is ending. So the queue really drains a moment later,
	 * from `agent_settled` (see `index.ts`), which is the first instant a prompt can be
	 * taken. Asking here anyway costs one comparison and keeps this method honest for
	 * every other caller.
	 */
	async onAgentEnd(
		messages: readonly unknown[],
		fallbackReply = true,
	): Promise<void> {
		this.deps.abort.clear();
		const reply = fallbackReply ? lastAssistantReply(messages) : null;
		if (reply) await this.sendAnswer(reply);
		await this.dispatch();
	}

	/** Show the Telegram "typing…" indicator on the bound chat (repeat while busy). */
	async sendTyping(): Promise<void> {
		await this.deps.outbound.chatAction(this.target, "typing").catch(() => {});
	}

	/**
	 * Send one of the BRIDGE's own messages to the bound chat — help, status, a card.
	 *
	 * These go out with entity detection on, so a `/command` in them is tappable rather
	 * than being text you have to retype. That is the whole difference between the help
	 * card and the pinned mode message, which was tappable all along simply because it
	 * is sent as plain text. Model prose is not sent through here (see
	 * {@link deliverAssistant}) and keeps its own rendering.
	 */
	async sendToChat(markdown: string): Promise<number[]> {
		return await this.deps.outbound.sendMarkdown(this.target, markdown, {
			detectEntities: true,
		});
	}

	/**
	 * Mirror a prompt typed at the Pi terminal into the bound chat, clearly
	 * marked, so the Telegram history reflects everything asked — from either
	 * side. Best-effort; the real reply is still mirrored on `agent_end`.
	 */
	async mirrorTerminalInput(text: string): Promise<void> {
		const trimmed = text.trim();
		if (!trimmed) return;
		await this.deps.outbound
			.sendMarkdown(this.target, `_${TERMINAL_ORIGIN_MARKER}_\n\n${trimmed}`)
			.catch(() => {});
	}

	/**
	 * Open a streaming-draft id for the assistant message about to stream. An id
	 * already open — the thinking placeholder of this same turn — is REUSED, so the
	 * placeholder animates into the streaming text instead of being replaced by a
	 * second draft (which would read as a flicker). A closed draft gets a fresh id,
	 * distinct per message, so drafts never animate across replies.
	 */
	beginDraft(): void {
		if (this.draftId !== 0) return;
		this.draftCounter = (this.draftCounter % 1_000_000) + 1;
		this.draftId = this.draftCounter;
	}

	/**
	 * Push the turn's live trace into the draft: finished steps as plain lines, the
	 * current one animated. This is NOT the model's reasoning — the SDK exposes none;
	 * it is what the agent is DOING, filling the silence between the prompt and the
	 * first token, and it disappears with the draft. Best-effort, like
	 * {@link streamDraft}.
	 */
	async streamThinking(html: RichHtml): Promise<void> {
		if (!html.html.trim()) return;
		this.beginDraft();
		await this.flushMirror();
		await this.deps.outbound
			.draft(this.target, this.draftId, buildRichHtmlMessage(html))
			.catch(() => {});
	}

	/**
	 * Push a partial assistant reply as an ephemeral animated draft. Best-effort:
	 * a draft is a transient preview, so any failure (bot not eligible, too long)
	 * is ignored and never blocks the turn or the final send.
	 */
	async streamDraft(text: string): Promise<void> {
		if (this.draftId === 0 || !text.trim()) return;
		// The first chunk IS the bot starting to speak — the prompt it answers belongs
		// above it, not seconds earlier while the model was still silent.
		await this.flushMirror();
		await this.deps.outbound
			.draft(this.target, this.draftId, buildRichMarkdownMessage(text))
			.catch(() => {});
	}

	/** Close the current streaming draft (the real reply is sent separately). */
	endDraft(): void {
		this.draftId = 0;
	}

	/**
	 * Erase the draft in place with an empty one. A draft only expires ~30s after its
	 * last update, so a turn that ends while the placeholder is up — an abort (/esc),
	 * an error, a tool-only run — would otherwise leave "Thinking…" animating long
	 * after the agent stopped. Best-effort: if the API refuses an empty draft, the
	 * placeholder still expires on its own.
	 */
	async clearDraft(): Promise<void> {
		if (this.draftId === 0) return;
		const draftId = this.draftId;
		this.draftId = 0;
		await this.deps.outbound
			.draft(this.target, draftId, { html: "" })
			.catch(() => {});
	}

	/**
	 * Surface an agent tool invocation to the bound chat as a collapsible block
	 * (tool name + folded parameters) — it belongs with the conversation it serves, so
	 * you can watch the model work. Best-effort: a formatting/send failure must never
	 * interrupt the agent's turn.
	 *
	 * The card is posted `running` and remembered by `callId`, so
	 * {@link completeToolActivity} can finish it in place once the tool returns.
	 */
	async sendToolActivity(
		activity: ToolCallActivity,
		callId?: string,
	): Promise<void> {
		await this.flushMirror();
		const ids = await this.deps.outbound
			.sendMessages(this.target, [toolActivityMessage(activity)])
			.catch(() => [] as number[]);
		const messageId = ids[0];
		if (callId === undefined || messageId === undefined) return;
		this.toolCards.set(callId, { messageId, activity });
	}

	/**
	 * Finish the card for a call that has returned: rewrite it with a ✅ or ❌ and its
	 * output folded in, rather than posting a second message about it. Re-rendered
	 * from the REMEMBERED activity, because the end event carries no arguments — a
	 * card rebuilt without them would silently lose the command it ran.
	 *
	 * An unknown call id (cards disabled mid-turn, a card that failed to send) is a
	 * no-op, and so is a failed edit — the running card stays, which is still true.
	 */
	async completeToolActivity(
		callId: string,
		result: unknown,
		isError: boolean,
	): Promise<number | undefined> {
		const card = this.toolCards.get(callId);
		if (!card) return undefined;
		this.toolCards.delete(callId);
		await this.deps.outbound
			.editRich(
				this.target,
				card.messageId,
				toolActivityMessage({
					...card.activity,
					status: isError ? "error" : "ok",
					result,
				}),
			)
			.catch(() => {});
		// The caller may hang the full-output file off this card.
		return card.messageId;
	}

	/**
	 * Attach a tool's own full-output file to the chat — the whole log behind a card
	 * that had to stop at "… (75 earlier lines)". WHETHER it should be attached is
	 * decided upstream (was it truncated? is it within the owner's byte cap?); this
	 * only sends it. Best-effort: a failed upload leaves the card, which still names
	 * the path.
	 */
	async attachToolOutput(
		path: string,
		caption: string,
		replyToMessageId?: number,
		filename?: string,
	): Promise<void> {
		if (!this.deps.uploadFile) return;
		await this.deps
			.uploadFile({ path, caption, replyToMessageId, filename })
			.catch(() => {});
	}

	/**
	 * Close out every card still waiting for a result. A call interrupted by /esc
	 * never fires its end event, so its card would otherwise keep the running state
	 * forever — claiming work that stopped. Mark them cancelled and forget them.
	 */
	async cancelOpenToolCards(): Promise<void> {
		const open = [...this.toolCards.values()];
		this.toolCards.clear();
		for (const card of open) {
			await this.deps.outbound
				.editRich(
					this.target,
					card.messageId,
					toolActivityMessage({ ...card.activity, status: "cancelled" }),
				)
				.catch(() => {});
		}
	}

	/** Pending (not yet dispatched) turn count — for footer/status. */
	pendingCount(): number {
		return this.queue.size();
	}

	/** Drop every queued turn (used by /stop). */
	clearQueue(): void {
		this.queue.clear();
	}
}