import * as fs from "node:fs";
import * as path from "node:path";
import { type JsonlParseFailure, parseJsonlWindow, readJsonlSince, readJsonlTail } from "../../utils/incremental-reader.ts";
import { logInternalError } from "../../utils/internal-error.ts";
import type { TeamEvent } from "./event-log.ts";
import { currentGeneration } from "./event-log-rotation.ts";
import { commitPoisonFindings, planPoisonSkip } from "./poison-marker.ts";

// --- R18 / R16-B1 effect 2 (Phase 3.6): archive-tail readers ----------------
// Rotation stranding: a sync append that was mid-appendFileSync holding an fd
// on the RENAMED inode (ST-8 rename+create) lands in the archive file, not the
// live file. Round 18 measured 5.43% stranded events under max-contention
// zero-lock repro. Previously readers saw ONLY the live file, so stranded
// events were invisible — and sweepOldArchives unlinked them after 7 days
// (gone forever). Fix (Round 18 option (a)): mirror the mailbox
// safeReadMailboxFile archive-walk — readers also drain archive TAILS,
// deduped by seq, merged ahead of the live file's events.

/** List `<eventsPath>.<ts>.archive.jsonl` siblings (matches rotateEventLogUnlocked's
 *  naming), sorted by name (timestamp) so older generations come first. */
function listEventArchivePaths(eventsPath: string): string[] {
	const dir = path.dirname(eventsPath);
	const base = path.basename(eventsPath);
	try {
		return fs
			.readdirSync(dir)
			.filter((entry) => entry.startsWith(`${base}.`) && entry.endsWith(".archive.jsonl"))
			.sort()
			.map((entry) => path.join(dir, entry));
	} catch {
		// Directory missing — nothing to read.
		return [];
	}
}

/** Parse a JSONL file into TeamEvents with the unified U22 Phase 1 torn/poison
 *  classification: torn (unterminated) tail line held back + reported, complete
 *  poison lines skipped + sticky-marked, marker-hit lines skipped WITHOUT a
 *  re-parse. Returns empty events when the file is unreadable. */
interface EventsFileReadOutcome {
	events: TeamEvent[];
	tornTail: boolean;
	poisonSkipped: number;
	parseFailures: number;
	quarantined: boolean;
}

function emptyEventsFileReadOutcome(): EventsFileReadOutcome {
	return { events: [], tornTail: false, poisonSkipped: 0, parseFailures: 0, quarantined: false };
}

/** Number of complete (newline-terminated) lines — the marker reset-policy
 *  index basis. Byte-level scan, no decode. */
function countCompleteLines(buf: Buffer): number {
	let count = 0;
	let idx = 0;
	for (;;) {
		const newline = buf.indexOf(0x0a, idx);
		if (newline < 0) break;
		count++;
		idx = newline + 1;
	}
	return count;
}

function parseJsonlEventsWithIntegrity(filePath: string): EventsFileReadOutcome {
	let buf: Buffer;
	try {
		buf = fs.readFileSync(filePath);
	} catch {
		return emptyEventsFileReadOutcome();
	}
	const lineCount = countCompleteLines(buf);
	const plan = planPoisonSkip(filePath, buf.length, lineCount);
	const parsed = parseJsonlWindow<TeamEvent>(buf, 0, { poisonOffsets: plan.offsets });
	const quarantined = commitPoisonFindings(filePath, buf.length, lineCount, parsed.failures, parsed.staleMarkerOffsets);
	if (quarantined) {
		// Cap exceeded — the file was renamed to `<name>.corrupt-<ts>`; return
		// the safe empty value (never throw).
		return { ...emptyEventsFileReadOutcome(), parseFailures: parsed.failures.length, quarantined: true };
	}
	return {
		events: parsed.items.map((item) => item.value),
		tornTail: parsed.tornTail,
		poisonSkipped: parsed.poisonSkipped,
		parseFailures: parsed.failures.length,
		quarantined: false,
	};
}

/** Tail-window read of one events/archive JSONL file with the SAME torn/poison
 *  classification as the full parse above (marker lineIndex is unknown for a
 *  truncated window — byte offsets remain the stable match key). */
function readEventsFileTailWithIntegrity(filePath: string, tailBytes: number): EventsFileReadOutcome & { truncated: boolean } {
	let stat: fs.Stats;
	try {
		stat = fs.statSync(filePath);
	} catch {
		return { ...emptyEventsFileReadOutcome(), truncated: false };
	}
	const plan = planPoisonSkip(filePath, stat.size, undefined);
	const tail = readJsonlTail<TeamEvent>(filePath, tailBytes, { poisonOffsets: plan.offsets });
	const quarantined = commitPoisonFindings(filePath, stat.size, undefined, tail.failures, tail.staleMarkerOffsets);
	if (quarantined) {
		return { ...emptyEventsFileReadOutcome(), parseFailures: tail.failures.length, truncated: tail.truncated, quarantined: true };
	}
	return {
		events: tail.items,
		tornTail: tail.tornTail,
		poisonSkipped: tail.poisonSkipped,
		parseFailures: tail.failures.length,
		quarantined: false,
		truncated: tail.truncated,
	};
}

const ARCHIVE_TAIL_BYTES = 4 * 1024 * 1024; // 4 MB — mirrors readEventsCursor's TAIL_BYTES

/** Drain archive TAILS: events with seq > sinceSeq from every archive of
 *  eventsPath, deduped by seq, sorted by seq. Stranded in-flight-fd appends
 *  land at the END of the archive (the renamed pre-rotation inode), so a
 *  bounded tail read per archive is sufficient and keeps this O(tail bytes). */
function readArchiveTailEvents(eventsPath: string, sinceSeq: number): TeamEvent[] {
	const archives = listEventArchivePaths(eventsPath);
	if (archives.length === 0) return [];
	const bySeq = new Map<number, TeamEvent>();
	for (const archivePath of archives) {
		// U22 Phase 1: archives get the same torn/poison classification (each
		// archive carries its own `<archive>.poison.jsonl` sidecar).
		const tail = readEventsFileTailWithIntegrity(archivePath, ARCHIVE_TAIL_BYTES);
		for (const event of tail.events) {
			const seq = event.metadata?.seq;
			if (typeof seq !== "number" || seq <= sinceSeq) continue;
			if (!bySeq.has(seq)) bySeq.set(seq, event);
		}
	}
	return [...bySeq.values()].sort((a, b) => (a.metadata?.seq ?? 0) - (b.metadata?.seq ?? 0));
}

/** Merge drained archive-tail events with live-file events: dedup by seq
 *  (live wins on collision — it is the persisted survivor), seq-sorted. */
function mergeArchiveTailEvents(archiveEvents: TeamEvent[], liveEvents: TeamEvent[]): TeamEvent[] {
	if (archiveEvents.length === 0) return liveEvents;
	const bySeq = new Map<number, TeamEvent>();
	for (const event of archiveEvents) bySeq.set(event.metadata?.seq ?? 0, event);
	for (const event of liveEvents) bySeq.set(event.metadata?.seq ?? 0, event);
	return [...bySeq.values()].sort((a, b) => (a.metadata?.seq ?? 0) - (b.metadata?.seq ?? 0));
}

export interface EventLogReadIntegrity {
	/** Any read file (archive or live) ended mid-line (crash-truncated tail). */
	tornTail: boolean;
	/** Lines skipped via sticky poison markers (not re-parsed this read). */
	poisonSkipped: number;
	/** Lines whose JSON.parse failed in this read (poison discovered now). */
	parseFailures: number;
	/** A file exceeded the poison-marker cap and was quarantined (last resort). */
	quarantined: boolean;
}

/** U22 Phase 1: readEvents plus the unified torn/poison metadata. `readEvents`
 *  (below) is the exact-event-behavior wrapper for existing callers. */
export function readEventsWithIntegrity(eventsPath: string): { events: TeamEvent[]; integrity: EventLogReadIntegrity } {
	// R18 / R16-B1 effect 2 (Phase 3.6): FULL-HISTORY semantics — merge every
	// archive (pre-rotation generations, including stranded in-flight appends)
	// with the live file, dedup by seq, ordered by seq. Callers needing the
	// full history (team-runner recovery, run-export, inspect, read, and the
	// event-reconstructor, whose only read entry is this function) now see
	// archived events too — this is the intended fix for "no event stranded".
	const archives = listEventArchivePaths(eventsPath);
	const events: TeamEvent[] = [];
	const seenSeqs = new Set<number>();
	const push = (event: TeamEvent): void => {
		const seq = event.metadata?.seq;
		if (typeof seq === "number" && seq > 0) {
			if (seenSeqs.has(seq)) return;
			seenSeqs.add(seq);
		}
		events.push(event);
	};
	const integrity: EventLogReadIntegrity = { tornTail: false, poisonSkipped: 0, parseFailures: 0, quarantined: false };
	const absorb = (outcome: EventsFileReadOutcome): void => {
		integrity.tornTail = integrity.tornTail || outcome.tornTail;
		integrity.poisonSkipped += outcome.poisonSkipped;
		integrity.parseFailures += outcome.parseFailures;
		integrity.quarantined = integrity.quarantined || outcome.quarantined;
		for (const event of outcome.events) push(event);
	};
	for (const archivePath of archives) {
		absorb(parseJsonlEventsWithIntegrity(archivePath));
	}
	if (fs.existsSync(eventsPath)) {
		absorb(parseJsonlEventsWithIntegrity(eventsPath));
	}
	events.sort((a, b) => (a.metadata?.seq ?? 0) - (b.metadata?.seq ?? 0));
	return { events, integrity };
}

export function readEvents(eventsPath: string): TeamEvent[] {
	return readEventsWithIntegrity(eventsPath).events;
}

export interface EventCursorOptions {
	sinceSeq?: number;
	/** Delivery cap for one read. When `limit` truncates the result, `events` is
	 *  a PREFIX of everything available — resume from the returned `nextSeq`
	 *  (never from `nextByteOffset`; see the continuation contract on
	 *  `readEventsCursor`). */
	limit?: number;
	fromByteOffset?: number;
	/** R-03: generation the caller captured on its previous read. When set, a
	 * mismatch with the live generation signals the file was rotated/truncated
	 * and the byte offset is stale — the cursor resets to 0 so the new file is
	 * re-read from its start instead of missing post-rotation events. */
	generation?: number;
}

export interface EventCursorResult {
	events: TeamEvent[];
	nextSeq: number;
	total: number;
	nextByteOffset?: number;
	/** R9-N2 (2026-10-17): true when the WIDE (default) path dropped a prefix
	 *  because the log exceeds the 4 MB tail window — callers paging by seq
	 *  alone cannot reach the dropped prefix; pass `fromByteOffset` to resume
	 *  scanning from a byte position instead. Additive/optional: callers that
	 *  ignore it keep the exact previous behavior. */
	truncated?: boolean;
	/** R-03: live generation of the events file at read time. Callers doing
	 * streaming byte-offset reads should echo this back as `generation` on the
	 * next call so rotation is detected and the cursor resets. */
	generation?: number;
	/** U22 Phase 1 (optional, additive): the live file's read ended on a torn
	 *  (unterminated) tail line — recoverable, the next append completes it. */
	tornTail?: boolean;
	/** U22 Phase 1 (optional, additive): lines skipped via sticky poison markers
	 *  (complete lines whose JSON.parse failed earlier — skipped without a
	 *  re-parse this read). */
	poisonSkipped?: number;
	/** U22 Phase 1 (optional, additive): lines whose JSON.parse failed in THIS
	 *  read (newly discovered poison). */
	parseFailures?: number;
}

export interface EventReadIntegrity {
	tornTail: boolean;
	poisonSkipped: number;
	parseFailures: number;
}

function positiveInteger(value: number | undefined): number | undefined {
	return value !== undefined && Number.isInteger(value) && value >= 0 ? value : undefined;
}

// --- Perf round 2 / Task 6 (fix round 1): VERIFIED watermark tail cache -------
//
// The first cut of this task scaled the tail window down with `limit` and
// accepted it whenever the window's first parsed seq <= sinceSeq — assuming
// file order == seq order. That assumption is FALSE:
//   (1) compaction recovery re-appends lost events at the END of the file
//       with their OLD seqs (event-log-rotation.ts applyCompaction);
//   (2) the sync (.mkdirlock) and async (.alok) lock families are disjoint,
//       so interleaved writers can append reserved seqs out of order;
//   (3) explicit baseMetadata.seq appends bypass reservation ordering.
// A file can therefore look like [1..200, 5000, 201..1700], and the window
// acceptance silently dropped event 5000 forever.
//
// This fix keeps NO order assumption. Instead each events path carries a
// cache entry with a VERIFIED watermark (transcript-cache.ts pattern):
//   verifiedOffset  byte offset just past the last VERIFIED complete line
//                  (a partial trailing line is held back, not consumed)
//   lastSeq         seq of the last event at verifiedOffset
//   ring            bounded, file-ordered suffix of the verified events
//   ino/dev         inode identity stamped from the SAME stat call as the
//                  size/mtimeMs stamps (fix round 2, below)
// The watermark is ESTABLISHED only by a full parse from offset 0 in which
// every parsed event carries a finite seq and seqs are non-decreasing, and
// EXTENDED only by deltas whose first seq >= lastSeq and whose seqs are
// non-decreasing. Any violation — out-of-order append, seq-less line, file
// shrink, stamp mismatch, INODE CHANGE, short delta read — rebuilds from a
// fresh full parse. With the watermark held, answering from the ring is sound:
//   - verified && ringStartSeq <= sinceSeq: every event BEFORE the ring
//     start lies inside the verified range, where seqs are non-decreasing,
//     so its seq <= ringStartSeq <= sinceSeq — the sinceSeq filter would
//     drop it in a full read too. The ring's post-filter set IS the full
//     read's post-filter set, so TAIL_EVENT_CAP slice, archive-tail merge
//     (R18/R16-B1), dedupe+sort, total/nextSeq are all identical.
//   - ringStartOffset === 0: the ring IS the whole file (nothing dropped).
// An UNVERIFIED lineage (an out-of-order file, e.g. after compaction
// recovery) may only answer via ringStartOffset === 0; anything else takes
// the full parse. Files larger than the wide 4MB window can return a strict
// SUPERSET of the old byte-window answer (older post-sinceSeq events the
// window used to drop) — never fewer events.
//
// Fix round 2 — INODE STAMPING (the rewrite-then-regrow hole): size+mtimeMs
// alone cannot distinguish "grew" from "was replaced and grew back past the
// old size". Every real REWRITER of the events file produces a NEW inode —
// compaction (applyCompactionUnlocked → atomicWriteFile temp+rename),
// rotation (renameSync + 'wx' create), forget+import — while appends
// PRESERVE it. So both the unchanged branch and the delta branch additionally
// require stat.ino === entry.ino && stat.dev === entry.dev; a mismatch is
// treated as a replacement and rebuilds from a full parse. Repro it closes:
// file [1..10] → cached read primes the entry → rewrite keeping [9,10]
// (shrink, new inode) → append 11..20 (regrow past the old size, mtimeMs now
// >= the primed mtimeMs) → the stamp-only check used to take the delta
// branch and serve [6,7,8,9,10,19,20] — 8 events permanently dropped, 3
// deleted phantom events served. With inode identity the read rebuilds and
// returns [9,10,11..20], byte-identical to a full parse.
//
// Two hygiene rules that keep the stamps trustworthy:
//   - the wide/FIND-05 path (sinceSeq=0 / no-limit / stat failures) DROPS
//     the entry for the path: mixed call patterns must not keep stale stamps
//     alive across reads that never re-validate them (an entry could then
//     outlive its inode up to the point of inode-number reuse);
//   - the delta read fstat's the fd it just opened and rejects a swap
//     (rotation between statSync and openSync) instead of attributing the
//     new inode's bytes to the old watermark.
// Residual (documented, empty in the current writer set): an IN-PLACE
// truncate rewrite — same inode, then regrowth past the old size — would
// still pass the delta branch. No such writer exists: appenders use
// appendFileSync, and every rewriter (compaction, rotation, forget+import)
// swaps the inode. Same accepted-risk class as transcript-cache.
// Accepted divergence, unchanged from fix round 1: a trailing fragment
// without "\n" is held back rather than parsed (a mid-append partial line),
// which can differ from the wide reader at the sub-line level.

const TAIL_BYTES = 4 * 1024 * 1024; // FIND-05: 4 MB wide tail window
const TAIL_EVENT_CAP = 5000;
const CURSOR_READ_CHUNK_BYTES = 64 * 1024;
const CURSOR_TAIL_CACHE_MAX_ENTRIES = 100;

/** One parsed event plus the absolute byte offset of its line's start. */
interface CursorLineEvent {
	event: TeamEvent;
	startOffset: number;
}

interface CursorTailCacheEntry {
	/** Stamp pair from the last update; any mismatch forces a rebuild. */
	size: number;
	mtimeMs: number;
	/** Inode identity from the SAME stat call as the stamps above. Appends
	 *  preserve it; every rewriter (compaction's atomicWriteFile temp+rename,
	 *  rotation's renameSync+'wx' create, forget+import) produces a NEW one.
	 *  A mismatch means "the path now names a different file" — even when
	 *  size/mtimeMs alone would pass (rewrite-then-regrow past the old size
	 *  with a >= mtimeMs) — and forces a full-parse rebuild. */
	ino: number;
	dev: number;
	verifiedOffset: number;
	lastSeq: number;
	/** True only while every consumed line since offset 0 carried a finite,
	 *  non-decreasing seq. False => the ring may answer only from offset 0. */
	verified: boolean;
	ring: CursorLineEvent[];
	/** U22 Phase 1 metadata from the scan that built/extended this entry —
	 *  served on unchanged-stamp reads without re-reading event bytes. Marker
	 *  sidecar changes do not touch the events file's stamps, so a marker
	 *  written by another read path can make these counts lag by one read
	 *  (items are unaffected: a poison line yields no event either way). */
	tornTail: boolean;
	poisonSkipped: number;
}

const cursorTailCache = new Map<string, CursorTailCacheEntry>();

/** Fix round 2: inode identity check — the path must still name the exact
 *  file (same dev+ino) the entry's stamps describe. Appends preserve the
 *  inode; compaction (temp+rename), rotation (rename+'wx'), and forget+import
 *  all produce a new one, so a mismatch reliably means "replaced". */
function sameInode(stat: { ino: number; dev: number }, entry: CursorTailCacheEntry): boolean {
	return stat.ino === entry.ino && stat.dev === entry.dev;
}

/** Drop events from the ring's front (file order) past `bound`, keeping the
 *  newest suffix. Ring bound: max(TAIL_EVENT_CAP, limit) * 2 events — enough
 *  lookback behind the anchor for the provability check while bounding
 *  memory (the answer itself is capped to TAIL_EVENT_CAP events regardless,
 *  so a larger ring only widens provability, it never changes results). */
function evictCursorRing(ring: CursorLineEvent[], bound: number): CursorLineEvent[] {
	return ring.length > bound ? ring.slice(ring.length - bound) : ring;
}

/** Parse COMPLETE lines out of `buf`, which starts at absolute `baseOffset`
 *  (itself a line boundary). A trailing fragment without "\n" is held back
 *  (mid-append partial line). Corrupt/blank lines advance the verified
 *  offset but yield no event. `minSeq` seeds the watermark check: `ok` is
 *  false when any parsed event lacks a finite seq or goes backwards — i.e.
 *  the non-decreasing watermark is broken and the caller must rebuild.
 *  Parsed events are returned regardless (a full-parse answer must keep
 *  them, mirroring the wide reader). */
function scanEventLines(
	buf: Buffer,
	baseOffset: number,
	minSeq: number | undefined,
	poisonOffsets?: ReadonlySet<number>,
): {
	lines: CursorLineEvent[];
	verifiedBytes: number;
	ok: boolean;
	/** U22 Phase 1: new parse failures (poison) for sticky marking. */
	failures: JsonlParseFailure[];
	/** U22 Phase 1: held-back trailing fragment without "\n" (torn tail). */
	tornTail: boolean;
	/** U22 Phase 1: marker-hit lines skipped WITHOUT a parse. */
	poisonSkipped: number;
	/** U22 Phase 1: complete lines in the window (marker reset-policy index). */
	lineCount: number;
	/** U22 fix round re-parse guard: marker offsets whose line re-parsed OK
	 *  (stale markers) — forwarded for sidecar pruning. */
	staleMarkerOffsets: number[];
} {
	// U22 Phase 1: delegate the line scan to the shared torn/poison classifier
	// so the cached path classifies corrupt content IDENTICALLY to the wide
	// path and the full parser (marker hits skip the parse entirely).
	const parsed = parseJsonlWindow<TeamEvent>(buf, baseOffset, { poisonOffsets });
	let ok = true;
	let lastSeq = minSeq;
	for (const item of parsed.items) {
		const seq = item.value.metadata?.seq;
		if (typeof seq !== "number" || !Number.isFinite(seq) || (lastSeq !== undefined && seq < lastSeq)) {
			ok = false;
		} else {
			lastSeq = seq;
		}
	}
	return {
		lines: parsed.items.map((item) => ({ event: item.value, startOffset: item.byteOffset })),
		verifiedBytes: parsed.completeBytes,
		ok,
		failures: parsed.failures,
		tornTail: parsed.tornTail,
		poisonSkipped: parsed.poisonSkipped,
		lineCount: parsed.lineCount,
		staleMarkerOffsets: parsed.staleMarkerOffsets,
	};
}

/** Read exactly [start, end) from an already-open fd, or return null on any
 *  short read (concurrent shrink/rotation) so the caller rebuilds from a
 *  full parse. Reading from the caller's fd keeps the stat and the read on
 *  the SAME inode. */
function readCursorByteRange(fd: number, start: number, end: number): Buffer | null {
	if (end <= start) return Buffer.alloc(0);
	const length = end - start;
	const buf = Buffer.alloc(length);
	let totalRead = 0;
	while (totalRead < length) {
		const chunk = Math.min(CURSOR_READ_CHUNK_BYTES, length - totalRead);
		const n = fs.readSync(fd, buf, totalRead, chunk, start + totalRead);
		if (n <= 0) return null;
		totalRead += n;
	}
	return buf;
}

/** Full parse from offset 0 — the only way to ESTABLISH the watermark, and
 *  the answer basis when the ring cannot prove coverage (total coverage by
 *  construction). Returns undefined when the file cannot be read (the caller
 *  falls back to the wide window). O_NOFOLLOW matches readJsonlTail's
 *  symlink refusal. */
function rebuildCursorTailCache(
	eventsPath: string,
	bound: number,
): { entry: CursorTailCacheEntry; lines: CursorLineEvent[]; quarantined: boolean; parseFailures: number } | undefined {
	let fd: number | undefined;
	try {
		fd = fs.openSync(eventsPath, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW);
	} catch {
		return undefined;
	}
	try {
		const stat = fs.fstatSync(fd);
		const buf = readCursorByteRange(fd, 0, stat.size);
		if (buf === null) return undefined;
		// U22 Phase 1: exact lineCount (full file) makes the marker reset policy
		// precise here — stale markers (rotate/truncate) can never skip a line.
		const lineCount = countCompleteLines(buf);
		const plan = planPoisonSkip(eventsPath, buf.length, lineCount);
		const scan = scanEventLines(buf, 0, undefined, plan.offsets);
		const quarantined = commitPoisonFindings(eventsPath, buf.length, lineCount, scan.failures, scan.staleMarkerOffsets);
		const lastLine = scan.lines.at(-1)?.event.metadata?.seq;
		const entry: CursorTailCacheEntry = {
			size: stat.size,
			mtimeMs: stat.mtimeMs,
			ino: stat.ino,
			dev: stat.dev,
			verifiedOffset: scan.verifiedBytes,
			lastSeq: typeof lastLine === "number" ? lastLine : 0,
			verified: scan.ok,
			ring: evictCursorRing(scan.lines, bound),
			tornTail: scan.tornTail,
			poisonSkipped: scan.poisonSkipped,
		};
		return { entry, lines: scan.lines, quarantined, parseFailures: scan.failures.length };
	} finally {
		try {
			fs.closeSync(fd);
		} catch {
			/* ignore */
		}
	}
}

/** Shared answer pipeline (identical to the wide path's): TAIL_EVENT_CAP
 *  slice (+ its warning) BEFORE the sinceSeq filter, then the R18/R16-B1
 *  archive-tail merge, then the head-cap limit, then total/nextSeq. */
function cursorResultFromEvents(
	all: TeamEvent[],
	eventsPath: string,
	sinceSeq: number,
	limit: number | undefined,
	integrity?: EventReadIntegrity,
): EventCursorResult {
	let capped = all;
	if (capped.length > TAIL_EVENT_CAP) {
		logInternalError(
			"event-log.cursor-full-read",
			new Error(`readEventsCursor tail read dropped events from a larger log; pass fromByteOffset for incremental reads`),
			`eventsPath=${eventsPath}`,
		);
		capped = capped.slice(-TAIL_EVENT_CAP);
	}
	const filtered = capped.filter((event) => (event.metadata?.seq ?? 0) > sinceSeq);
	// R18 (Phase 3.6): rotation stranding — prepend archive-tail events (seq >
	// sinceSeq, deduped, seq-sorted) ahead of the live tail slice, so events
	// stranded into an archive by a rotation are still delivered to sinceSeq
	// streaming consumers. No-rotation case: no archives exist → behavior is
	// byte-identical to before (mergeArchiveTailEvents returns liveEvents).
	const merged = mergeArchiveTailEvents(readArchiveTailEvents(eventsPath, sinceSeq), filtered);
	// FS3 (fix): seq-sort before the limit slice. File-order slicing on a
	// documented non-monotonic log can deliver an early high seq inside the
	// first `limit` events; nextSeq=max-delivered then permanently skips every
	// undelivered lower-seq event for paging consumers resuming from nextSeq.
	// Sort ONLY under a limit: the no-limit path delivers every event (nextSeq
	// = true max ⇒ nothing skippable) and keeps its historical FILE-order
	// contract (pinned by event-log-torn-poison.test.ts AC7, U22 Phase 1).
	const events =
		limit !== undefined ? [...merged].sort((a, b) => (a.metadata?.seq ?? 0) - (b.metadata?.seq ?? 0)).slice(0, limit) : merged;
	const returnedMaxSeq = events.reduce((max, event) => Math.max(max, event.metadata?.seq ?? 0), sinceSeq);
	return {
		events,
		nextSeq: returnedMaxSeq,
		total: merged.length,
		...(integrity
			? { tornTail: integrity.tornTail, poisonSkipped: integrity.poisonSkipped, parseFailures: integrity.parseFailures }
			: {}),
	};
}

/** Can `ring` (under `entry`'s lineage) answer a read anchored at sinceSeq?
 *  Sound per the watermark docblock: either the ring starts at the verified
 *  seq watermark that sinceSeq has already passed, or the ring starts at
 *  offset 0 and is the whole file. */
function cursorRingProvable(entry: CursorTailCacheEntry, ring: CursorLineEvent[], sinceSeq: number): boolean {
	if (ring.length === 0) {
		return entry.verifiedOffset === 0 || (entry.verified && entry.lastSeq <= sinceSeq);
	}
	const start = ring[0];
	return start.startOffset === 0 || (entry.verified && (start.event.metadata?.seq ?? 0) <= sinceSeq);
}

/** Perf round 2 / Task 6 (fix round 1): serve a sinceSeq+limit cursor read
 *  from the verified watermark cache. `sinceSeq > 0` gating keeps
 *  replay-from-zero calls on the wide path (seqs start at 1, so a sinceSeq=0
 *  anchor can never be proven passed by a watermark — every call would pay
 *  the full parse for nothing). Returns undefined when the file cannot be
 *  stat'd/read; the caller then runs the FIND-05 wide path unchanged. */
function readEventsCursorTailCached(eventsPath: string, sinceSeq: number, limit: number): EventCursorResult | undefined {
	let stat: fs.Stats;
	try {
		stat = fs.statSync(eventsPath);
	} catch {
		return undefined;
	}
	const previous = cursorTailCache.get(eventsPath);
	const bound = Math.max(TAIL_EVENT_CAP, limit) * 2;
	// U22 Phase 1: consult the sticky-poison sidecar up front (one existsSync
	// on the clean-file fast path — no event bytes touched). Marker-hit lines
	// are skipped without a parse in every branch below, keeping the cached
	// path's torn/poison classification identical to the wide path's.
	const plan = planPoisonSkip(eventsPath, stat.size, undefined);

	if (previous && stat.size === previous.size && stat.mtimeMs === previous.mtimeMs && sameInode(stat, previous)) {
		// Unchanged stamps + same inode: serve purely from the ring (no event
		// bytes read). The inode check is what makes "unchanged" trustworthy:
		// size+mtimeMs alone pass for a same-path replacement with recycled
		// stamps (compaction/rotation rewrites).
		const ring = evictCursorRing(previous.ring, bound);
		if (cursorRingProvable(previous, ring, sinceSeq)) {
			if (ring !== previous.ring) cursorTailCache.set(eventsPath, { ...previous, ring });
			return cursorResultFromEvents(
				ring.map((line) => line.event),
				eventsPath,
				sinceSeq,
				limit,
				{ tornTail: previous.tornTail, poisonSkipped: previous.poisonSkipped, parseFailures: 0 },
			);
		}
	} else if (
		previous &&
		stat.size > previous.size &&
		stat.mtimeMs >= previous.mtimeMs &&
		sameInode(stat, previous) &&
		previous.verifiedOffset < stat.size
	) {
		// Append-only growth on the SAME inode: read only [verifiedOffset,
		// size). Provability is checked BEFORE the read — appending only moves
		// the ring start forward, so an unprovable ring can never become
		// provable here. The inode requirement closes rewrite-then-regrow: a
		// replacement (new inode) that shrank and grew back past the old size
		// with mtimeMs >= previous.mtimeMs must NOT be read as a delta from
		// the dead watermark — it falls through to the full parse.
		const preRing = evictCursorRing(previous.ring, bound);
		if (cursorRingProvable(previous, preRing, sinceSeq)) {
			let delta: Buffer | null = null;
			let deltaFd: number | undefined;
			try {
				deltaFd = fs.openSync(eventsPath, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW);
				// TOCTOU guard: rotation/compaction can swap the path between
				// the statSync above and this open. The stamps describe the
				// stat'd inode; bytes from a different one must never extend
				// its watermark — re-check identity on the fd actually opened.
				const fdStat = fs.fstatSync(deltaFd);
				if (fdStat.ino === stat.ino && fdStat.dev === stat.dev) {
					delta = readCursorByteRange(deltaFd, previous.verifiedOffset, stat.size);
				}
			} catch {
				delta = null;
			} finally {
				if (deltaFd !== undefined) {
					try {
						fs.closeSync(deltaFd);
					} catch {
						/* ignore */
					}
				}
			}
			if (delta !== null) {
				const scan = scanEventLines(delta, previous.verifiedOffset, previous.lastSeq, plan.offsets);
				if (scan.ok) {
					const quarantined = commitPoisonFindings(eventsPath, stat.size, undefined, scan.failures, scan.staleMarkerOffsets);
					if (!quarantined) {
						const ring = evictCursorRing([...preRing, ...scan.lines], bound);
						const lastLineSeq = scan.lines.at(-1)?.event.metadata?.seq;
						const entry: CursorTailCacheEntry = {
							size: stat.size,
							mtimeMs: stat.mtimeMs,
							ino: stat.ino,
							dev: stat.dev,
							verifiedOffset: previous.verifiedOffset + scan.verifiedBytes,
							lastSeq: typeof lastLineSeq === "number" ? lastLineSeq : previous.lastSeq,
							// The delta verified against the watermark, but a
							// violation earlier in the file keeps the lineage
							// unverified (it can only answer from offset 0).
							verified: previous.verified,
							ring,
							tornTail: scan.tornTail,
							poisonSkipped: previous.poisonSkipped + scan.poisonSkipped,
						};
						cursorTailCache.set(eventsPath, entry);
						if (cursorRingProvable(entry, ring, sinceSeq)) {
							return cursorResultFromEvents(
								ring.map((line) => line.event),
								eventsPath,
								sinceSeq,
								limit,
								{ tornTail: entry.tornTail, poisonSkipped: entry.poisonSkipped, parseFailures: scan.failures.length },
							);
						}
					}
					// Quarantined (poison cap exceeded — file renamed): fall through
					// to the full-parse rebuild, which misses the renamed path and
					// hands the caller the wide path's safe-empty result.
				}
			}
		}
	}

	// Full parse: cache miss, stamp mismatch/shrink, INODE CHANGE (rewrite,
	// rotation, forget+import), watermark violation, or an anchor older than
	// the ring's provable start. Re-establishes the watermark and answers with
	// TOTAL coverage (exact full-read semantics for files within the wide
	// window; a superset — never fewer — beyond).
	const rebuilt = rebuildCursorTailCache(eventsPath, bound);
	if (rebuilt === undefined) return undefined;
	cursorTailCache.set(eventsPath, rebuilt.entry);
	// Bounded FIFO, not an LRU: Map.set on an existing key does NOT refresh
	// its insertion position, so the oldest-INSERTED entry is the one evicted
	// (the FIND-12 bound, with the comment telling the truth about what it is).
	if (cursorTailCache.size > CURSOR_TAIL_CACHE_MAX_ENTRIES) {
		const oldestKey = cursorTailCache.keys().next().value;
		if (oldestKey !== undefined) cursorTailCache.delete(oldestKey);
	}
	if (rebuilt.quarantined) {
		// U22 Phase 1 cap exceeded — the file was renamed to
		// `<name>.corrupt-<ts>`; return the safe empty answer (never throw).
		return cursorResultFromEvents([], eventsPath, sinceSeq, limit, {
			tornTail: false,
			poisonSkipped: 0,
			parseFailures: rebuilt.parseFailures,
		});
	}
	return cursorResultFromEvents(
		rebuilt.lines.map((line) => line.event),
		eventsPath,
		sinceSeq,
		limit,
		{ tornTail: rebuilt.entry.tornTail, poisonSkipped: rebuilt.entry.poisonSkipped, parseFailures: rebuilt.parseFailures },
	);
}

/** Test/invalidation hook: drop the verified watermark cache for one events
 *  path (or every path when omitted). */
export function clearEventsCursorTailCache(eventsPath?: string): void {
	if (!eventsPath) {
		cursorTailCache.clear();
		return;
	}
	cursorTailCache.delete(eventsPath);
}

/**
 * Read team events with an optional incremental anchor (`sinceSeq`) plus a
 * delivery cap (`limit`).
 *
 * Continuation / backpressure contract (BR-08):
 *  - `nextSeq` is the continuation token for events that CARRY `metadata.seq`
 *    (scheduler-emitted events): the max seq among the events DELIVERED by
 *    this call — never a watermark past an undelivered event. Resume with
 *    `{ sinceSeq: nextSeq }` (`>` comparison) to receive the rest.
 *    LIMITS (cold-verify correction — documented, not fixed here): events
 *    WITHOUT `metadata.seq` (e.g. the worker progress channel) are filtered
 *    out of the sinceSeq path entirely, and DUPLICATE seqs cannot be paged
 *    through by seq. Those cases need the `fromByteOffset` anchor.
 *  - `total` is the number of events AVAILABLE for this read (after the
 *    sinceSeq filter, before the `limit` slice) — EXCEPT that the tail cap
 *    (TAIL_EVENT_CAP) may drop a prefix first, in which case `total` counts
 *    only the retained suffix.
 *  - `nextByteOffset` is an OPTIONAL incremental-read accelerator, and is only
 *    returned when it is a SAFE resume anchor: on the `fromByteOffset` path it
 *    is OMITTED whenever `limit` truncated the delta, because the underlying
 *    offset then points past the ENTIRE delta (undelivered tail included) —
 *    resuming from it would silently skip events. The default (non-byte)
 *    path never returns it at all.
 */
export function readEventsCursor(eventsPath: string, options: EventCursorOptions = {}): EventCursorResult {
	// Incremental byte-offset path: read only new bytes since last known offset
	if (options.fromByteOffset !== undefined) {
		// R-03: detect file rotation/truncation via the generation sidecar BEFORE
		// reusing the byte offset. If the file was rotated since the caller last
		// read, it was truncated to empty (pre-rotation content archived to
		// `<eventsPath>.<ts>.archive.jsonl`) and is growing again from 0 — the
		// caller's offset now points past EOF, so post-rotation events would be
		// silently missed. Reset to offset 0 to re-read the current file from its
		// start.
		// R18 (Phase 3.6): re-reading from 0 re-delivers no previously-returned
		// events FROM THE LIVE FILE — but events STRANDED into the archive by the
		// rotation (in-flight fd appends on the renamed inode) were never
		// delivered and, before this fix, were lost forever once sweepOldArchives
		// unlinked them. On a detected generation bump we therefore FIRST drain
		// the previous generations' archive TAILS (events with seq > sinceSeq,
		// deduped by seq — mailbox safeReadMailboxFile archive-walk pattern)
		// ahead of the fresh live file's events.
		const liveGen = currentGeneration(eventsPath);
		const staleCursor = options.generation !== undefined && options.generation !== liveGen;
		const sinceSeq = positiveInteger(options.sinceSeq) ?? 0;
		const archiveEvents = staleCursor ? readArchiveTailEvents(eventsPath, sinceSeq) : [];
		const requestedOffset = staleCursor ? 0 : (positiveInteger(options.fromByteOffset) ?? 0);
		// U22 Phase 1: consult the sticky-poison sidecar (marker-hit lines are
		// skipped without a re-parse — same classification as the other paths).
		let fileSize = 0;
		try {
			fileSize = fs.statSync(eventsPath).size;
		} catch {
			/* missing file — markers are vacuous; readJsonlSince returns empty */
		}
		const plan = planPoisonSkip(eventsPath, fileSize, undefined);
		let read = readJsonlSince<TeamEvent>(eventsPath, { byteOffset: requestedOffset, lineCount: 0 }, { poisonOffsets: plan.offsets });
		let drainedArchiveEvents = archiveEvents;
		if (!staleCursor && read.shrank) {
			// U22 Phase 1 rotation tail-drain: the live file is now SHORTER than
			// the caller's offset (rotate/truncate WITHOUT a generation echo —
			// e.g. a reader that never tracked generation). Reading forward from
			// the clamped offset would silently skip the new file's head, and the
			// pre-rotation tail (stranded into the archive) would never drain.
			// Same contract as the staleCursor branch: drain the archive TAILS
			// first (seq > sinceSeq, deduped), THEN re-read the live file from 0 —
			// the reader never advances past a rotation without draining the old
			// window, and the seq-dedupe merge kills the archive/live overlap.
			drainedArchiveEvents = readArchiveTailEvents(eventsPath, sinceSeq);
			read = readJsonlSince<TeamEvent>(eventsPath, { byteOffset: 0, lineCount: 0 }, { poisonOffsets: plan.offsets });
		}
		const quarantined = commitPoisonFindings(eventsPath, fileSize, undefined, read.failures, read.staleMarkerOffsets);
		const safeItems = quarantined ? [] : read.items;
		const filtered = safeItems.filter((event) => (event.metadata?.seq ?? 0) > sinceSeq);
		const merged = mergeArchiveTailEvents(drainedArchiveEvents, filtered);
		const limit = positiveInteger(options.limit);
		// FS3 (fix): seq-sort before the limit slice — see cursorResultFromEvents.
		// Sort ONLY under a limit (no-limit keeps the historical file-order
		// contract; nothing is skippable when everything is delivered).
		const events =
			limit !== undefined ? [...merged].sort((a, b) => (a.metadata?.seq ?? 0) - (b.metadata?.seq ?? 0)).slice(0, limit) : merged;
		const returnedMaxSeq = events.reduce((max, event) => Math.max(max, event.metadata?.seq ?? 0), sinceSeq);
		// BR-08: `read.state.byteOffset` sits past the ENTIRE delta read above —
		// including the events the `limit` slice just dropped. Handing that back
		// as `nextByteOffset` let a caller resume PAST events it never saw (silent
		// loss: the delta read then returns an empty tail). Only expose the
		// accelerator when this call delivered the whole delta; under truncation
		// the caller must continue from `nextSeq` (see the doc comment on
		// readEventsCursor).
		const deliveredWholeDelta = limit === undefined || merged.length <= limit;
		return {
			events,
			nextSeq: returnedMaxSeq,
			total: merged.length,
			...(deliveredWholeDelta ? { nextByteOffset: read.state.byteOffset } : {}),
			generation: liveGen,
			tornTail: quarantined ? false : read.tornTail,
			poisonSkipped: quarantined ? 0 : read.poisonSkipped,
			parseFailures: read.failures.length,
		};
	}

	// FIND-05 default path: byte-level tail read (last 4MB) instead of
	// full-file read. Bounds CPU to O(tail bytes) instead of O(total
	// events). The legacy readEvents() full parse path is preserved for
	// callers that explicitly need the full history (e.g. tests that
	// assert exact contents) and as a small-file fallback.
	//
	// The 5000-event tail cap and the "event-log.cursor-full-read"
	// warning are preserved. A separate cursor-tail-truncated warning is
	// emitted whenever the file exceeds the 4MB tail budget, signalling
	// that a prefix was dropped and callers should pass fromByteOffset for
	// streaming reads.
	const sinceSeq = positiveInteger(options.sinceSeq) ?? 0;
	const limit = positiveInteger(options.limit);
	// Perf round 2 / Task 6 (fix round 1): streaming ticks anchored with
	// sinceSeq and capped with limit (run-event-bus onWithReplay, broker
	// events.since/events.subscribe resync) are served by the verified
	// watermark cache — delta-only reads for append growth, zero event-byte
	// reads when nothing changed, and a full-parse rebuild on any watermark
	// violation, shrink, or stale anchor. Warning suppression there is
	// backed by proof, not by an order assumption: a ring answer happens
	// only when no post-sinceSeq event can have been dropped. sinceSeq=0 /
	// no-limit calls keep this wide path unchanged.
	if (limit !== undefined && sinceSeq > 0) {
		const cached = readEventsCursorTailCached(eventsPath, sinceSeq, limit);
		if (cached !== undefined) return cached;
	}
	// Fix round 2: the wide path never re-validates the watermark cache, so it
	// must DROP the entry instead of leaving its stamps behind. Without this,
	// a rewrite that lands between sinceSeq=0 / no-limit calls keeps stale
	// (size, mtimeMs, ino) stamps in the map until some later cached-path call
	// re-validates them — the entry would outlive its inode up to the point of
	// inode-number reuse. The next sinceSeq+limit call re-establishes the
	// watermark with one full parse.
	cursorTailCache.delete(eventsPath);
	// U22 Phase 1: the wide tail read goes through the same torn/poison
	// classification (marker sidecar consulted; poison lines sticky-marked;
	// torn tail held back and reported) as the full parse and the cached path.
	const tail = readEventsFileTailWithIntegrity(eventsPath, TAIL_BYTES);
	if (tail.truncated) {
		logInternalError("event-log.cursor-tail-truncated", {
			eventsPath,
			returned: tail.events.length,
			tailBytes: TAIL_BYTES,
		});
	}
	const result = cursorResultFromEvents(tail.events, eventsPath, sinceSeq, limit, {
		tornTail: tail.tornTail,
		poisonSkipped: tail.poisonSkipped,
		parseFailures: tail.parseFailures,
	});
	// R9-N2: surface the dropped-prefix condition so replay consumers can
	// switch to the fromByteOffset anchor instead of silently missing the head.
	return tail.truncated ? { ...result, truncated: true } : result;
}