import * as fs from "node:fs";
import * as path from "node:path";
import { logInternalError } from "../../utils/internal-error.ts";
import { atomicWriteFile } from "../atomic-write.ts";
import { type TeamEvent, withEventLogAppendLockSync, withEventLogLockSync } from "./event-log.ts";
import { poisonMarkerPath } from "./poison-marker.ts";

/**
 * ST-11: Stream a JSONL file line-by-line using a bounded read buffer (8 KB
 * chunks), invoking `onLine` for each non-empty line. Replaces the previous
 * `readFileSync` approach that loaded the ENTIRE event log (up to 4 MB /
 * 50 000 events) into a single string before splitting + JSON.parsing every
 * line into an array. Memory is now bounded by the chunk size plus the
 * longest single line, and parsing is interleaved with I/O so the append
 * lock is held for less wall-clock time.
 */
function forEachLineSync(filePath: string, onLine: (line: string) => void): void {
	const fd = fs.openSync(filePath, "r");
	try {
		const chunkSize = 8192;
		const buf = Buffer.alloc(chunkSize);
		let leftover = "";
		let offset = 0;
		let bytesRead: number;
		while ((bytesRead = fs.readSync(fd, buf, 0, chunkSize, offset)) > 0) {
			offset += bytesRead;
			leftover += buf.subarray(0, bytesRead).toString("utf-8");
			let nl: number;
			while ((nl = leftover.indexOf("\n")) >= 0) {
				const line = leftover.slice(0, nl);
				leftover = leftover.slice(nl + 1);
				if (line.length > 0) onLine(line);
			}
		}
		if (leftover.length > 0) {
			// Trailing line without a final newline.
			onLine(leftover);
		}
	} finally {
		fs.closeSync(fd);
	}
}

/**
 * ST-11: Stream the event log and keep only the last `maxKeep` valid (parseable)
 * events using a fixed-size ring buffer. Bounds memory to O(maxKeep) event
 * objects regardless of file size — previously the entire file was parsed
 * into a single array just to `slice(-compactToCount)` it. Corrupt lines
 * are skipped (matching `readEvents` behaviour). Returns the ordered
 * last-N events and the total count of valid events seen.
 */
function readLastEvents(filePath: string, maxKeep: number): { events: TeamEvent[]; totalCount: number } {
	if (maxKeep <= 0) {
		let totalCount = 0;
		forEachLineSync(filePath, (line) => {
			try {
				JSON.parse(line);
				totalCount++;
			} catch {
				/* skip corrupt */
			}
		});
		return { events: [], totalCount };
	}
	const ring: TeamEvent[] = new Array(maxKeep);
	let ringFilled = 0;
	let writeIdx = 0;
	let totalCount = 0;
	forEachLineSync(filePath, (line) => {
		let event: TeamEvent;
		try {
			event = JSON.parse(line) as TeamEvent;
		} catch {
			return; // skip corrupt — not counted (matches readEvents)
		}
		ring[writeIdx] = event;
		writeIdx = (writeIdx + 1) % maxKeep;
		if (ringFilled < maxKeep) ringFilled++;
		totalCount++;
	});
	const events: TeamEvent[] = [];
	if (ringFilled < maxKeep) {
		for (let i = 0; i < ringFilled; i++) events.push(ring[i]);
	} else {
		for (let i = 0; i < maxKeep; i++) events.push(ring[(writeIdx + i) % maxKeep]);
	}
	return { events, totalCount };
}

/**
 * ST-11: Stream the event log to count valid events and collect their seq
 * numbers, without materialising the full event array. Used by
 * `applyCompactionUnlocked` for the post-write recovery check. Memory is
 * bounded to the count + seq-set of the (small, post-compaction) file.
 */
function countEventsAndSeqs(filePath: string): { count: number; seqs: Set<number> } {
	const seqs = new Set<number>();
	let count = 0;
	forEachLineSync(filePath, (line) => {
		try {
			const event = JSON.parse(line) as TeamEvent;
			count++;
			const seq = event.metadata?.seq;
			if (typeof seq === "number") seqs.add(seq);
		} catch {
			/* skip corrupt */
		}
	});
	return { count, seqs };
}

export interface RotationConfig {
	maxFileSizeBytes: number;
	maxEventCount: number;
	compactToCount: number;
}

const DEFAULT_ROTATION_CONFIG: RotationConfig = {
	// 2.3: lowered from 5 MB to 4 MB so the file stays small enough that
	// `tail -c MAX_TAIL_BYTES` reads in run-snapshot-cache (default 32 KB)
	// always cover a useful slice and rotations happen earlier.
	maxFileSizeBytes: 4 * 1024 * 1024,
	maxEventCount: 50_000,
	compactToCount: 1_000,
};

const AVG_BYTES_PER_EVENT = 80;

function resolveConfig(config?: Partial<RotationConfig>): RotationConfig {
	return { ...DEFAULT_ROTATION_CONFIG, ...config };
}

/**
 * Check if an event file needs rotation/compaction.
 * M1: Uses file size estimation to avoid full-file read.
 */
export function needsRotation(eventsPath: string, config?: Partial<RotationConfig>): boolean {
	if (!fs.existsSync(eventsPath)) return false;
	const cfg = resolveConfig(config);
	try {
		const stat = fs.statSync(eventsPath);
		if (stat.size > cfg.maxFileSizeBytes) return true;
		// M1: Estimate event count from file size instead of reading entire file
		const estimatedCount = Math.floor(stat.size / AVG_BYTES_PER_EVENT);
		return estimatedCount > cfg.maxEventCount;
	} catch {
		return false;
	}
}

export interface CompactionResult {
	originalSize: number;
	compactedSize: number;
	eventsRemoved: number;
	eventsKept: number;
	recoveryFailed?: boolean;
}

/**
 * Compact an event log file:
 * C2: Fixed TOCTOU race — atomicWriteFile replaces in one step;
 * any events appended between readEvents and the write will be preserved
 * on the next compaction cycle because atomicWriteFile writes the full content.
 *
 * 1. Read all events
 * 2. Keep last `compactToCount` events
 * 3. Atomically write (atomicWriteFile handles temp-file + rename)
 * 4. Re-read to detect events appended during the window
 * 5. If events were lost, append them
 * 6. Bump the `.gen` generation sidecar (EP4/R32-M1) so byte-offset cursors
 *    echoing the pre-compaction generation re-read instead of resuming mid-file
 * 7. Return compaction stats
 */
export function compactEventLog(
	eventsPath: string,
	config?: Partial<RotationConfig>,
	options?: CompactionLockOptions,
): CompactionResult | undefined {
	// US-001 (2026-09-22): lock-scope reduction. The expensive part is the READ +
	// parse of the whole log (prepareCompaction). It is read-only, so it does NOT
	// need the append lock: hold the lock ONLY for the write+recover phase
	// (applyCompactionUnlocked), which is what must be atomic w.r.t. appends.
	//
	// Why this is safe: appenders take the SAME lock, so an append can only land
	// before our prepareCompaction read or after our applyCompactionUnlocked
	// write — never inside the write. applyCompactionUnlocked already re-reads
	// after the write and splices back any events that arrived in the window
	// (C2 recovery), so no event is lost; it only becomes slightly more likely to
	// splice, which is exactly the case it was built for.
	//
	// No concurrent-rotator guard is needed: this function is fully SYNCHRONOUS
	// (prepareCompaction and withEventLogLockSync both block), so two calls on the
	// same path cannot interleave within a process. An in-flight flag was tried
	// and removed — it was untestable dead code (mutation survived), and the
	// synchronous execution already serializes rotators. Cross-process rotators
	// are serialized by the on-disk lock in withEventLogLockSync.
	//
	// NOTE (Round 24 BUG 1): callers ALREADY holding the event-log lock (e.g.
	// appendEventInsideLock in event-log.ts) must still call applyCompactionUnlocked
	// directly — calling compactEventLog from inside the lock deadlocks (the
	// mkdir lock is not re-entrant → 5s timeout → compaction never ran → the
	// log grew unbounded until events were silently dropped past 50MB).
	const prepared = prepareCompaction(eventsPath, config);
	if (!prepared) return undefined;
	return compactPreparedEventLog(eventsPath, prepared, options);
}

/** Options for the locked write phase of compaction. `lock` is forwarded to
 * `withEventLogAppendLockSync` (and transitively `withEventLogLockSync`) so
 * callers/tests can bound the wait — the defaults (5s timeout / 10s stale)
 * mirror both existing lock families. */
export interface CompactionLockOptions {
	lock?: { timeoutMs?: number; staleMs?: number };
}

/**
 * US-001 phase 3 — the LOCKED write phase, extracted so the read→write race is
 * deterministically testable: a caller (test) can run prepareCompaction, append
 * to the file (simulating another process landing events in the window), then
 * call this and assert nothing was lost.
 *
 * prepareCompaction reads OUTSIDE the lock, so an appender may have added events
 * between that read and this write. applyCompactionUnlocked writes exactly
 * `prepared.lines` (the kept window from the snapshot), which would CLOBBER those
 * newer events; its C2 recovery only re-splices events that were in `kept`, not
 * ones it never saw. So under the lock we read the bytes appended after the
 * snapshot offset and append them to the write. Appenders hold the same lock,
 * so this read is a complete tail snapshot.
 *
 * R4-2 (DR-02 Wave 2, 2026-10-15): the write phase now ADDITIONALLY acquires the
 * async family's `.alock` (sync-mode, via `withEventLogAppendLockSync`) BEFORE
 * `readTailSince`/the `atomicWriteFile` rename, releasing it only after the C2
 * recovery check completes. Rationale: async-family appends hold `.alock`
 * ACROSS await gaps (`open(eventsPath, "a")` → `appendFile`); running the rename
 * under `.mkdirlock` alone let it interleave with that gap, orphaning the
 * appender's fd — its event landed on the old inode and was silently lost (C2
 * only re-splices the `kept` window). Holding `.alock` across the whole write
 * phase excludes in-flight async appends cross-process (same on-disk lock dir
 * + pid-file protocol). Lock order stays strictly `.mkdirlock` → `.alock` (the
 * async family never takes `.mkdirlock`), so no deadlock. NOT archive-style by
 * lead decision: rotation's rename-to-archive already preserves in-flight fds
 * (ST-8), so only compaction's inode-replacing write needed the guard.
 */
export function compactPreparedEventLog(
	eventsPath: string,
	prepared: { lines: string; originalSize: number; originalCount: number; kept: TeamEvent[] },
	options?: CompactionLockOptions,
): CompactionResult | undefined {
	// R7-2 (DR-04, 2026-10-16): the write phase is now `applyCompactionAppendGuarded`
	// (below) — `.alock` nested inside the `.mkdirlock` held here, plus the
	// mandatory `readTailSince` splice. This wrapper keeps the historical
	// locked entry point (public API + tests) with identical semantics.
	return withEventLogLockSync(eventsPath, () => applyCompactionAppendGuarded(eventsPath, prepared, options), options?.lock);
}

/** R7-2 (DR-04, 2026-10-16): options for `applyCompactionAppendGuarded`.
 * `lock` is forwarded to `withEventLogAppendLockSync` so callers/tests can
 * bound the wait; `holdsAppendLock` declares the caller's lock state (see the
 * helper docblock). */
export interface ApplyCompactionGuardOptions extends CompactionLockOptions {
	/** Pass `true` ONLY when the caller ALREADY holds the async family's
	 * `.alock` (i.e. the guarded apply runs inside `withEventLogLockAsync`).
	 * The on-disk lock is NOT re-entrant — re-acquiring our own `.alock` in
	 * sync mode can only end in the bounded timeout (sleepSync starves the
	 * very in-process holder that must release it). With `true` the helper
	 * only tail-merges; with `false`/omitted it acquires `.alock` itself. */
	holdsAppendLock?: boolean;
}

/**
 * R7-2 (DR-04, 2026-10-16): the guarded write phase for INLINE compaction
 * call sites (the overflow/rotation-tick paths inside event-log.ts) that
 * previously called `prepareCompaction` + `applyCompactionUnlocked` bare —
 * no `.alock` and no tail-merge, so the inode-replacing write could interleave
 * with an in-flight async-family append parked between `open()` and
 * `appendFile()` (its fd kept pointing at the renamed-away inode → silent
 * loss; C2 recovery only re-splices the `kept` window), and sync-family bytes
 * appended between the prepare-read and the write were clobbered.
 *
 * Contract:
 *  - ALWAYS merges `readTailSince(eventsPath, prepared.originalSize)` into
 *    `prepared.lines` before `applyCompactionUnlocked` — sync-family appenders
 *    never hold `.alock`, so they can land in the prepare→apply window even
 *    while an async-family caller holds `.alock`; the splice rescues them.
 *  - `opts.holdsAppendLock === true` (caller is inside `withEventLogLockAsync`):
 *    tail-merge ONLY — never re-acquire `.alock` (non-reentrant → self-timeout).
 *  - `opts.holdsAppendLock` false/omitted (caller holds `.mkdirlock` — sync
 *    family — or no lock): acquire `.alock` in sync mode nested inside the
 *    held `.mkdirlock`, per the R4-2 lock order (`.mkdirlock` → `.alock`; the
 *    async family never takes `.mkdirlock`, so no inversion). NEVER call this
 *    with `false` while `.alock` is already held by this process.
 */
export function applyCompactionAppendGuarded(
	eventsPath: string,
	prepared: { lines: string; originalSize: number; originalCount: number; kept: TeamEvent[] },
	options?: ApplyCompactionGuardOptions,
): CompactionResult | undefined {
	const applyWithTailMerge = (): CompactionResult | undefined => {
		const merged = { ...prepared, lines: prepared.lines + readTailSince(eventsPath, prepared.originalSize) };
		return applyCompactionUnlocked(eventsPath, merged);
	};
	if (options?.holdsAppendLock) return applyWithTailMerge();
	return withEventLogAppendLockSync(eventsPath, applyWithTailMerge, options?.lock);
}

/**
 * US-001: return the raw text appended to `filePath` after `offset`, or "" when
 * nothing was appended / the file was replaced by a smaller one (a foreign
 * rotation — caller then proceeds without a splice, which is the pre-US-001
 * behaviour). Must be called while holding the event-log lock.
 */
/** R4-N1: typed abort — distinguishes "foreign rotation, skip this compaction
 *  cycle" from the benign transient read errors readTailSince degrades on. */
class ForeignRotationAbortError extends Error {
	constructor(size: number, offset: number) {
		super(
			`event-log.compaction-foreign-rotation: live file (${size}B) smaller than compaction snapshot offset (${offset}B) — aborting cycle`,
		);
		this.name = "ForeignRotationAbortError";
	}
}

function readTailSince(filePath: string, offset: number): string {
	try {
		const size = fs.statSync(filePath).size;
		// R4-N1 (2026-10-17): a file SMALLER than the snapshot offset means a
		// foreign rotation replaced the live file with a smaller one. The old
		// `size <= offset` proceed-no-splice let the caller's atomicWrite then
		// CLOBBER the new live file (post-rotation appends lost, ~1000 stale
		// events resurrected, C2 reported SUCCESS). size === offset is the
		// benign "nothing appended" case; size < offset ABORTS the compaction
		// cycle (typed throw — every caller of applyCompactionAppendGuarded
		// wraps in try/catch+log and skips this cycle; the next cycle
		// re-evaluates against the new file).
		if (size < offset) throw new ForeignRotationAbortError(size, offset);
		if (size === offset) return "";
		const fd = fs.openSync(filePath, "r");
		try {
			const length = size - offset;
			const buf = Buffer.alloc(length);
			const read = fs.readSync(fd, buf, 0, length, offset);
			const tail = buf.subarray(0, read).toString("utf-8");
			// Only splice complete lines (a partial trailing write would corrupt JSONL).
			return tail.endsWith("\n") ? tail : tail.slice(0, tail.lastIndexOf("\n") + 1);
		} finally {
			fs.closeSync(fd);
		}
	} catch (error) {
		// R4-N1: the foreign-rotation abort must ESCAPE the benign-error degrade
		// below — the whole point is that the caller must not proceed.
		if (error instanceof ForeignRotationAbortError) throw error;
		return "";
	}
}

/** Round 24 (BUG 1): the lock-free pre-read for compaction. Safe to run
 * outside the lock (read-only). Returns the compacted lines + stats needed
 * for the write phase.
 *
 * ST-11: streams the event log line-by-line through a ring buffer instead of
 * `readEvents` (full readFileSync + JSON.parse of every line into a single
 * array). Memory is bounded to O(compactToCount) events regardless of file
 * size, and the append lock is held for less time because parsing is
 * interleaved with I/O rather than deferred until the entire file is loaded. */
export function prepareCompaction(
	eventsPath: string,
	config?: Partial<RotationConfig>,
):
	| {
			lines: string;
			originalSize: number;
			originalCount: number;
			kept: TeamEvent[];
	  }
	| undefined {
	if (!fs.existsSync(eventsPath)) return undefined;
	const cfg = resolveConfig(config);
	let originalSize: number;
	try {
		originalSize = fs.statSync(eventsPath).size;
	} catch {
		return undefined;
	}
	const { events: kept, totalCount: originalCount } = readLastEvents(eventsPath, cfg.compactToCount);
	if (originalCount <= cfg.compactToCount) return undefined;
	const lines = kept.map((e) => JSON.stringify(e)).join("\n") + "\n";
	return { lines, originalSize, originalCount, kept };
}

/** Round 24 (BUG 1): the write+recover phase of compaction. Assumes the
 * caller ALREADY holds the event-log lock (or accepts the unlocked race). */
export function applyCompactionUnlocked(
	eventsPath: string,
	prepared: {
		lines: string;
		originalSize: number;
		originalCount: number;
		kept: TeamEvent[];
	},
): CompactionResult | undefined {
	const { lines, originalSize, originalCount, kept } = prepared;
	try {
		atomicWriteFile(eventsPath, lines);
	} catch (err) {
		// Concurrent write conflict — skip compaction this cycle
		logInternalError("event-log-rotation.compact", err, `eventsPath=${eventsPath}`);
		return undefined;
	}
	// EP4 (R32-M1 ≡ R4-N1/FS10, fix-wave 6): compaction REWRITES the live log at
	// the same path (atomicWriteFile → temp+rename, a NEW inode) and drops the
	// events before the kept window. Without a generation bump, a byte-offset
	// cursor reader echoing the pre-compaction `generation` (readEventsCursor's
	// cache-key, cursor.ts) still matches the live sidecar and resumes from its
	// old offset — which now points into the MIDDLE of the rewritten file — so
	// it silently skips the kept-window head it never delivered. The `shrank`
	// fallback does NOT cover this: it only fires when the file is shorter than
	// the caller's offset. Bump here so EVERY compaction path (compactEventLog /
	// compactPreparedEventLog / applyCompactionAppendGuarded + the inline call
	// sites in event-log.ts) invalidates stale cursors exactly like rotation
	// (rotateEventLogUnlocked). A no-op compaction never reaches this point —
	// prepareCompaction returns undefined when originalCount <= compactToCount —
	// so the generation stays stable when nothing was truncated.
	bumpGenerationUnlocked(eventsPath);
	// U22 fix round: the compacted file is NEW content at the same path — every
	// old poison marker is meaningless (its byteOffset now names an arbitrary
	// line of the rewritten file). Drop the sidecar so a stale marker can never
	// resurrect once the file regrows past the old offsets; the re-parse guard
	// in the readers (parseJsonlWindow) is the backstop if this rm fails.
	try {
		fs.rmSync(poisonMarkerPath(eventsPath), { force: true });
	} catch (err) {
		logInternalError("event-log-rotation.compact-poison-sidecar", err, `eventsPath=${eventsPath}`, "debug");
	}
	// C2: Re-read to recover any events appended during the compaction window.
	// ST-11: stream the post-write file instead of `readEvents` (full
	// readFileSync + JSON.parse per line into an array). We only need the event
	// count and the set of seq numbers for the recovery check, so memory is
	// bounded without materialising every event object.
	try {
		const { count: afterWriteCount, seqs: afterSeqs } = countEventsAndSeqs(eventsPath);
		if (afterWriteCount >= kept.length) {
			return {
				originalSize,
				compactedSize: fs.statSync(eventsPath).size,
				eventsRemoved: originalCount - kept.length,
				eventsKept: kept.length + Math.max(0, afterWriteCount - kept.length),
			};
		}
		// afterWriteCount < kept.length — events were lost during compaction window.
		const missingEvents = kept.filter((e) => e.metadata?.seq === undefined || !afterSeqs.has(e.metadata.seq));
		let recoveredCount = 0;
		let recoveryFailed = false;
		if (missingEvents.length > 0) {
			const recoveryLines = missingEvents.map((e) => JSON.stringify(e) + "\n").join("");
			try {
				fs.appendFileSync(eventsPath, recoveryLines);
				recoveredCount = missingEvents.length;
			} catch (err) {
				recoveryFailed = true;
				logInternalError("event-log-rotation.recovery", err, `eventsPath=${eventsPath} lostEvents=${missingEvents.length}`);
			}
		}
		return {
			originalSize,
			compactedSize: fs.statSync(eventsPath).size,
			eventsRemoved: originalCount - kept.length,
			eventsKept: kept.length + recoveredCount,
			recoveryFailed,
		};
	} catch {
		// Post-write verification failed — compaction likely succeeded.
		return {
			originalSize,
			compactedSize: fs.statSync(eventsPath).size,
			eventsRemoved: originalCount - kept.length,
			eventsKept: kept.length,
		};
	}
}

// --- R-03: generation sidecar for rotation cursor invalidation ---

/**
 * R-03: Path of the generation sidecar for an events file. Mirrors the
 * `.seq` sidecar pattern. The generation is bumped every time the events
 * file is REWRITTEN in place with different content — rotation (truncated +
 * archived) or compaction (prefix dropped, kept window rewritten) — so
 * readers holding a byte-offset cursor can detect their offset is stale and
 * re-read from the start of the new file instead of missing events.
 */
export function generationPath(eventsPath: string): string {
	return `${eventsPath}.gen`;
}

/**
 * R-03: Read the current generation of an events file.
 *
 * Primary (and sole authoritative) mechanism = the `.gen` sidecar, which is
 * cross-platform. Inode (`fs.statSync().ino`) was evaluated as a secondary
 * fast-path signal but is platform-dependent: on Windows `ino` is always 0,
 * so it cannot be relied upon to detect rotation. The `.gen` sidecar is
 * therefore used exclusively (verified Windows path: ino=0 falls back to
 * the sidecar here).
 *
 * Returns 0 when the sidecar is absent (backward-compat: logs created before
 * this fix have no sidecar and are treated as generation 0, so a streaming
 * cursor that never tracked generation behaves as before on its first read).
 */
export function currentGeneration(eventsPath: string): number {
	try {
		const raw = fs.readFileSync(generationPath(eventsPath), "utf-8");
		const value = Number.parseInt(raw.trim(), 10);
		return Number.isInteger(value) && value >= 0 ? value : 0;
	} catch {
		return 0;
	}
}

/**
 * R-03: Atomically bump and persist the generation sidecar. Called by
 * `rotateEventLogUnlocked` after truncating the events file, by
 * `applyCompactionUnlocked` after rewriting it (EP4/R32-M1), and by
 * `quarantinePoisonedEventsFile` after renaming the poisoned file away
 * (EP4/FS10 residual, wave-6 M4-R2 — a rename is the same file-instance
 * change at the SAME path, so byte-offset cursors must reset). Assumes the
 * caller already holds the event-log lock (or accepts the unlocked race).
 * Returns the new generation value.
 */
export function bumpGenerationUnlocked(eventsPath: string): number {
	const next = currentGeneration(eventsPath) + 1;
	try {
		atomicWriteFile(generationPath(eventsPath), String(next));
	} catch (error) {
		logInternalError("event-log.bump-generation", error, `eventsPath=${eventsPath}`);
	}
	return next;
}

/**
 * Rotate an event log file by archiving it with a timestamp.
 * The current file is renamed to `<eventsPath>.<timestamp>.archive.jsonl`
 * and a fresh empty file is created in its place.
 * Readers using `readEvents`/`readEventsCursor` see the new file; archive
 * files are read back by the archive-tail readers in event-log.ts
 * (R18 / R16-B1 effect 2, Phase 3.6): on a detected `.gen` generation bump
 * the cursor drains the previous generations' archive TAILS (events with
 * seq > sinceSeq, deduped by seq) ahead of the fresh live file, so events
 * stranded into an archive by an in-flight fd append during the rename are
 * still delivered instead of being swept away after the retention window.
 */
export function rotateEventLog(eventsPath: string): boolean {
	if (!fs.existsSync(eventsPath)) return false;
	// FIX: Wrap rotation in lock to prevent race conditions with concurrent readers.
	// Order of operations: (1) rename old live file to archive, (2) create new
	// empty live file. The rename is atomic (POSIX) so eventsPath is briefly
	// absent between (1) and (2); readers using readEvents tolerate a missing
	// file (return []). See rotateEventLogUnlocked for the ST-8 rationale.
	//
	// NOTE (Round 24 BUG 1): callers ALREADY holding the lock must call
	// rotateEventLogUnlocked directly — this locked variant is NOT re-entrant.
	return withEventLogLockSync(eventsPath, () => rotateEventLogUnlocked(eventsPath));
}

/** H2 retention: delete `<eventsPath>.*.archive.jsonl` files older than the window so
 * they don't accumulate forever (mirrors notification/metric-sink rotateOldFiles). */
const ARCHIVE_RETENTION_DAYS = 7;
function sweepOldArchives(eventsPath: string, now = Date.now()): void {
	const dir = path.dirname(eventsPath);
	const base = path.basename(eventsPath);
	let entries: string[];
	try {
		entries = fs.readdirSync(dir);
	} catch {
		return;
	}
	const cutoff = now - ARCHIVE_RETENTION_DAYS * 24 * 60 * 60 * 1000;
	for (const name of entries) {
		if (!name.startsWith(`${base}.`)) continue;
		// U22 fix round: sweep each archive's poison-marker sidecar
		// (`<archive>.poison.jsonl`) with the same retention — including
		// orphans whose archive was already swept (rotateEventLogUnlocked
		// moves sidecars alongside their archives).
		const isArchive = name.endsWith(".archive.jsonl");
		const isArchiveSidecar = name.endsWith(".archive.jsonl.poison.jsonl");
		if (!isArchive && !isArchiveSidecar) continue;
		const archivePath = path.join(dir, name);
		try {
			if (fs.statSync(archivePath).mtimeMs < cutoff) fs.unlinkSync(archivePath);
		} catch (error) {
			logInternalError("event-log.archive-sweep", error, `archivePath=${archivePath}`, "debug");
		}
	}
}

/** Round 24 (BUG 1): the lock-free core of rotation. Assumes the caller
 * already holds the event-log lock (or accepts the unlocked race). */
export function rotateEventLogUnlocked(eventsPath: string): boolean {
	if (!fs.existsSync(eventsPath)) return false;
	try {
		const ts = new Date().toISOString().replace(/[:.]/g, "-");
		let archivePath = `${eventsPath}.${ts}.archive.jsonl`;
		// Round 12: avoid timestamp collisions when two rotations happen within
		// the same millisecond (copyFileSync would silently overwrite the
		// first archive). Append a counter until the path is free.
		let collision = 1;
		while (fs.existsSync(archivePath)) {
			archivePath = `${eventsPath}.${ts}.${collision}.archive.jsonl`;
			collision++;
		}
		// ST-8: rename+create instead of copy+truncate. The old approach
		// (copyFileSync + atomicWriteFile("")) had two data-loss windows:
		//
		//   1. Copy window: events appended between copyFileSync and the
		//      atomicWriteFile truncate landed on the OLD inode. After
		//      atomicWriteFile replaced eventsPath with a new (empty) inode,
		//      those appends were on an orphaned inode with no path → lost
		//      forever once the writer's fd closed.
		//   2. atomicWriteFile replaces the inode → any in-flight writer fd
		//      (opened before rotation) continues writing to the old inode,
		//      which is now orphaned.
		//
		// rename+create fixes both:
		//   - rename(2) is atomic: the live log's inode moves to the archive
		//     path in one step, carrying ALL pre-rotation content. No copy
		//     window exists — there is nothing to lose between "read" and
		//     "truncate" because both happen in the single rename syscall.
		//   - In-flight writer fds that opened the old inode before the rename
		//     continue writing to it — but it is now the ARCHIVE (it has a
		//     path), so their appends are preserved, not orphaned. The next
		//     append (all write paths open per-call) naturally opens the new
		//     inode at eventsPath.
		//   - The new live file is created with O_CREAT|O_EXCL ("wx"): if a
		//     concurrent writer (different lock class — sync `.mkdirlock` vs
		//     async `.alock` are separate) already recreated the file between
		//     our rename and here, we leave their data intact (EEXIST → skip).
		fs.renameSync(eventsPath, archivePath);
		// U22 fix round: the poison-marker sidecar describes the OLD file's
		// bytes — move it alongside the archive so its markers keep naming the
		// exact lines they were recorded against. The fresh live file starts
		// with NO sidecar, so a stale marker can never resurrect on it (the
		// re-parse guard in the readers is the backstop if this rename fails).
		try {
			const sidecar = poisonMarkerPath(eventsPath);
			if (fs.existsSync(sidecar)) fs.renameSync(sidecar, poisonMarkerPath(archivePath));
		} catch (error) {
			logInternalError("event-log.rotate-poison-sidecar", error, `eventsPath=${eventsPath}`, "debug");
		}
		try {
			const fd = fs.openSync(eventsPath, "wx", 0o600);
			fs.closeSync(fd);
		} catch (err) {
			if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err;
			// A concurrent writer recreated the file — their data is preserved.
		}
		// R-03: bump the generation sidecar so byte-offset cursor readers
		// detect the truncation and reset (re-read from offset 0 of the new
		// file), instead of reading past EOF and missing post-rotation events.
		bumpGenerationUnlocked(eventsPath);
		// H2: retention — sweep archive files older than the window so they don't
		// accumulate forever (mirrors notification/metric-sink rotateOldFiles).
		sweepOldArchives(eventsPath);
		return true;
	} catch (error) {
		// R17-S1 (Phase 3.8): severity "error" (was default "debug", PI_TEAMS_DEBUG-
		// gated) — a failed rotation leaves the file over the size limit and the
		// next non-terminal appends are silently skipped; the chain must signal.
		logInternalError("event-log.rotate", error, `eventsPath=${eventsPath}`, "error");
		return false;
	}
}

export interface EventLogStats {
	fileSizeBytes: number;
	eventCount: number;
	oldestTimestamp?: string;
	newestTimestamp?: string;
}

/**
 * L3: Get event log stats using optimized reads.
 * Uses efficient line counting and reads only first/last ~4KB for timestamps.
 */
export function getEventLogStats(eventsPath: string): EventLogStats | undefined {
	if (!fs.existsSync(eventsPath)) return undefined;
	try {
		const stat = fs.statSync(eventsPath);
		const fileSizeBytes = stat.size;
		if (fileSizeBytes === 0) {
			return { fileSizeBytes: 0, eventCount: 0 };
		}

		// NEW-9 fix: stream-scan for line count (no full-file load).
		// Read last up-to-1KB for newest timestamp.
		let newestTimestamp: string | undefined;
		let lastLine = "";
		const tailSize = Math.min(fileSizeBytes, 1024);
		{
			const tailBuf = Buffer.alloc(tailSize);
			const fd = fs.openSync(eventsPath, "r");
			try {
				fs.readSync(fd, tailBuf, 0, tailSize, fileSizeBytes - tailSize);
			} finally {
				fs.closeSync(fd);
			}
			const tailStr = tailBuf.toString("utf-8");
			// JSONL files end with "\n", so the last newline bounds an empty string.
			// Walk backwards to find the last non-empty line.
			let searchFrom = tailStr.length;
			for (;;) {
				const nl = tailStr.lastIndexOf("\n", searchFrom - 1);
				if (nl < 0) {
					lastLine = tailStr.trim();
					break;
				}
				const candidate = tailStr.slice(nl + 1, searchFrom).trim();
				if (candidate) {
					lastLine = candidate;
					break;
				}
				searchFrom = nl;
			}
			try {
				if (lastLine) {
					newestTimestamp = (JSON.parse(lastLine) as { time: string }).time;
				}
			} catch {
				/* corrupt tail */
			}
		}

		// Stream-scan to count newlines and find first line boundary.
		let eventCount = 0;
		let firstLineBytes = 0;
		const buf = Buffer.alloc(8192);
		let offset = 0;
		let newlineCount = 0;
		const scanFd = fs.openSync(eventsPath, "r");
		try {
			let bytesRead: number;
			while ((bytesRead = fs.readSync(scanFd, buf, 0, buf.length, offset)) > 0) {
				for (let i = 0; i < bytesRead; i++) {
					if (buf[i] === 10) {
						if (newlineCount === 0) firstLineBytes = offset + i + 1;
						newlineCount++;
					}
				}
				offset += bytesRead;
			}
		} finally {
			fs.closeSync(scanFd);
		}
		eventCount = newlineCount;

		// Read first line for oldest timestamp.
		let oldestTimestamp: string | undefined;
		if (firstLineBytes > 0) {
			try {
				const firstBuf = Buffer.alloc(firstLineBytes);
				const fd = fs.openSync(eventsPath, "r");
				try {
					fs.readSync(fd, firstBuf, 0, firstLineBytes, 0);
				} finally {
					fs.closeSync(fd);
				}
				const firstLine = firstBuf.toString("utf-8").trim();
				if (firstLine) {
					oldestTimestamp = (JSON.parse(firstLine) as { time: string }).time;
				}
			} catch {
				/* corrupt head */
			}
		}

		return {
			fileSizeBytes,
			eventCount,
			oldestTimestamp,
			newestTimestamp,
		};
	} catch {
		return undefined;
	}
}