nucleic/Sources/NucleicCore/Transcript.swift

import Foundation

/// Line 0 of every transcript file (BACKEND_PROTOCOL §6).
public struct SessionHeader: Sendable, Codable, Equatable {
    /// The current transcript format version stamped on new headers.
    public static let currentVersion = 1
    /// The highest header version this build can read. A session-transfer import rejects a
    /// transcript whose `version` exceeds this (the `versionMismatch` reject) — this is the
    /// first consumer of transcript versioning (mesh P5). Bump alongside `currentVersion` when
    /// a format change stays readable; leave it behind only for a genuinely breaking change.
    public static let maxSupported = 1

    public let version: Int
    public let sessionID: SessionID
    public let backend: BackendID
    public var backendSessionID: String?
    public let worktree: WorktreePath
    public let model: String?
    public let nativeTranscriptPath: String?
    public let createdAt: Date
    /// Monotonic count of reverts (undo) this transcript has undergone. A revert resets the seq
    /// counter (`TranscriptWriter.truncate`), so seqs only identify content *within* one epoch;
    /// stamping the epoch on line 0 makes the file self-describing — a mirror synced via a full
    /// fetch (which ships this header verbatim) holds the epoch on disk, a re-serving holder
    /// propagates it, and a session transfer carries it to the new owner. `nil` = written before
    /// the field existed (equivalent to 0: never reverted, or provenance unknown). Optional and
    /// version-neutral (`version` stays 1) so old builds read new files and vice versa.
    public var revertEpoch: UInt64?

    public init(
        version: Int = 1,
        sessionID: SessionID,
        backend: BackendID,
        backendSessionID: String? = nil,
        worktree: WorktreePath,
        model: String? = nil,
        nativeTranscriptPath: String? = nil,
        createdAt: Date,
        revertEpoch: UInt64? = nil
    ) {
        self.version = version
        self.sessionID = sessionID
        self.backend = backend
        self.backendSessionID = backendSessionID
        self.worktree = worktree
        self.model = model
        self.nativeTranscriptPath = nativeTranscriptPath
        self.createdAt = createdAt
        self.revertEpoch = revertEpoch
    }

    /// Encode/decode the header for the wire (full-transcript mesh sync carries it as opaque
    /// `Data` in `TranscriptFetchComplete.headerJSON`, since the wire layer can't name this type).
    /// The exact bytes needn't match line 0 — the receiver re-encodes via `TranscriptWriter` — so
    /// a plain iso8601 JSON coder suffices.
    public func encodedForWire() throws -> Data {
        let encoder = JSONEncoder()
        encoder.dateEncodingStrategy = .iso8601
        return try encoder.encode(self)
    }

    public static func decodeFromWire(_ data: Data) throws -> SessionHeader {
        let decoder = JSONDecoder()
        decoder.dateDecodingStrategy = .iso8601
        return try decoder.decode(SessionHeader.self, from: data)
    }
}

/// Append-only JSONL transcript — the UI's source of truth (locked decision).
/// The canonical, session-wide `seq` is (re)assigned here at append time, the
/// single serialization point (RUNTIME §2.1): adapters' provisional seq is only
/// an ordering hint, and synthetic events (setup logs, approval echoes) need
/// numbering too.
public actor TranscriptWriter {
    private let url: URL
    private let fileHandle: FileHandle
    private let encoder: JSONEncoder
    private var renderIndexAppender: TranscriptRenderIndexAppender?
    public private(set) var lastSeq: UInt64

    /// Run this actor on its own dispatch queue instead of the shared Swift cooperative
    /// pool. Every method here does BLOCKING file I/O (`write(contentsOf:)`, and
    /// `synchronize()` — an fsync that can stall for seconds on a struggling disk); on the
    /// width-limited cooperative pool (one thread per core, non-expandable) a few sessions
    /// writing to a slow volume parked enough pool threads to starve the entire concurrency
    /// runtime — decode loops, watchdogs, everything (the app-wide "frozen until stop"
    /// class, hang-report sweep 2026-07-28). On a private queue a blocked write costs one
    /// expendable GCD thread. Semantics are unchanged: the queue is serial, so actor
    /// isolation and operation atomicity hold exactly as before.
    ///
    /// Via ``SerialQueueExecutor`` rather than `DispatchSerialQueue`, which exists only in the
    /// Darwin Dispatch overlay — corelibs-libdispatch doesn't vend it, so the direct spelling
    /// doesn't compile on Windows or Linux. Same private serial queue either way.
    private nonisolated let ioExecutor = SerialQueueExecutor(
        label: "com.nucleic.transcript-io", qos: .utility)
    public nonisolated var unownedExecutor: UnownedSerialExecutor {
        ioExecutor.asUnownedSerialExecutor()
    }

    public init(url: URL, header: SessionHeader) throws {
        self.url = url
        try FileManager.default.createDirectory(
            at: url.deletingLastPathComponent(), withIntermediateDirectories: true)

        encoder = JSONEncoder()
        encoder.outputFormatting = [.sortedKeys, .withoutEscapingSlashes]
        encoder.dateEncodingStrategy = .iso8601
        renderIndexAppender = nil

        FileManager.default.createFile(atPath: url.path, contents: nil)
        fileHandle = try FileHandle(forWritingTo: url)
        lastSeq = 0

        var headerLine = try encoder.encode(header)
        headerLine.append(0x0A)
        try fileHandle.write(contentsOf: headerLine)
        renderIndexAppender = try? TranscriptRenderIndexAppender(
            newTranscriptAt: url, header: header, canonicalHeaderBytes: headerLine.count)
        if renderIndexAppender == nil { try? TranscriptRenderIndex.invalidate(for: url) }
    }

    /// Reopen an existing transcript for appending (resume after relaunch). Does not
    /// rewrite the header; continues numbering from `lastSeq`.
    public init(
        appendingTo url: URL, lastSeq: UInt64, initialHistory: [AgentEvent]? = nil
    ) throws {
        self.url = url
        encoder = JSONEncoder()
        encoder.outputFormatting = [.sortedKeys, .withoutEscapingSlashes]
        encoder.dateEncodingStrategy = .iso8601
        renderIndexAppender = nil
        fileHandle = try FileHandle(forWritingTo: url)
        try fileHandle.seekToEnd()
        self.lastSeq = lastSeq
        renderIndexAppender = TranscriptRenderIndexAppender(
            appendingTo: url, expectedLastSequence: lastSeq,
            initialHistory: initialHistory)
    }

    /// Appends and returns the event re-stamped with its canonical seq.
    @discardableResult
    public func append(_ event: AgentEvent) throws -> AgentEvent {
        lastSeq += 1
        let canonical = event.withSeq(lastSeq)
        var line = try encoder.encode(canonical)
        line.append(0x0A)
        try fileHandle.write(contentsOf: line)
        appendRenderIndex(canonical, canonicalByteLength: line.count)
        return canonical
    }

    /// Append an event **verbatim** — keeping its existing canonical `seq` instead of
    /// re-stamping. Used when mirroring another device's transcript (full-transcript mesh sync):
    /// the source already assigned canonical seqs, and re-stamping would clobber them and break
    /// the receiver's dedup/gap logic. `lastSeq` tracks the written seq so a reopened mirror
    /// resumes correctly. This is the second deliberate verbatim-seq path alongside `truncate`;
    /// steady-state owned writing still goes through `append` (RUNTIME §2.1).
    public func appendVerbatim(_ event: AgentEvent) throws {
        var line = try encoder.encode(event)
        line.append(0x0A)
        try fileHandle.write(contentsOf: line)
        appendRenderIndex(event, canonicalByteLength: line.count)
        lastSeq = event.seq
    }

    /// Rewrite the transcript so it ends at `events` (the session header on line 0 is
    /// preserved verbatim, or replaced by `header` when provided — the revert path stamps the
    /// bumped `revertEpoch` there in the same rewrite), discarding everything after — used when
    /// the user reverts the conversation to an earlier point. Events keep their existing
    /// canonical seqs; the append cursor is repositioned to the new end and `lastSeq` reset, so
    /// the next turn continues numbering cleanly. Append-only is the steady state (RUNTIME
    /// §2.1); this is the one deliberate, user-initiated rewrite.
    public func truncate(to events: [AgentEvent], header: SessionHeader? = nil) throws {
        let headerLine: Data
        if let header {
            headerLine = try encoder.encode(header)
        } else {
            let existing = try Data(contentsOf: url)
            headerLine =
                existing.firstIndex(of: 0x0A).map { existing.subdata(in: existing.startIndex..<$0) }
                ?? existing
        }
        var out = headerLine
        out.append(0x0A)
        for event in events {
            out.append(try encoder.encode(event))
            out.append(0x0A)
        }
        try fileHandle.seek(toOffset: 0)
        try fileHandle.write(contentsOf: out)
        try fileHandle.truncate(atOffset: UInt64(out.count))
        try fileHandle.synchronize()
        lastSeq = events.last?.seq ?? 0
        // Revert replaces the epoch/sequence namespace and every canonical offset after line 0.
        // Install a fresh derived index from the already-decoded retained events.
        try? renderIndexAppender?.close()
        renderIndexAppender = nil
        do {
            try TranscriptRenderIndex.rebuild(transcriptURL: url, events: events)
            renderIndexAppender = TranscriptRenderIndexAppender(
                appendingTo: url, expectedLastSequence: lastSeq,
                initialHistory: events)
        } catch {
            try? TranscriptRenderIndex.invalidate(for: url)
        }
    }

    /// Durability point — called at turn boundaries and approval persistence
    /// (RUNTIME open-Q #2: batched fsync).
    public func sync() throws {
        try fileHandle.synchronize()
        do {
            try renderIndexAppender?.sync()
        } catch {
            disableRenderIndex()
        }
    }

    public func close() throws {
        try? renderIndexAppender?.close()
        renderIndexAppender = nil
        try fileHandle.close()
    }

    private func appendRenderIndex(_ event: AgentEvent, canonicalByteLength: Int) {
        guard let renderIndexAppender else { return }
        do {
            try renderIndexAppender.append(event, canonicalByteLength: canonicalByteLength)
        } catch {
            disableRenderIndex()
        }
    }

    private func disableRenderIndex() {
        try? renderIndexAppender?.close()
        renderIndexAppender = nil
        try? TranscriptRenderIndex.invalidate(for: url)
    }
}

/// Rewrites the header (line 0) of an existing transcript file in place, preserving every event
/// line after it. Used by a session-transfer import to point the header at the destination's
/// recomputed worktree / native-transcript paths (mesh P5), and to drop `backendSessionID` when
/// memory-carry falls back to fresh context. The event bytes are copied verbatim so the file's
/// content — apart from line 0 — is byte-identical to what was received.
public enum TranscriptHeaderRewriter {
    /// Replace line 0 of the file at `url` with `header` (re-encoded with the canonical
    /// `TranscriptWriter` config so hashes stay reproducible), keeping all following lines.
    public static func rewrite(at url: URL, to header: SessionHeader) throws {
        let existing = try Data(contentsOf: url)
        // Everything after the first newline is the event body — preserved verbatim.
        let body: Data = existing.firstIndex(of: 0x0A)
            .map { existing.subdata(in: existing.index(after: $0)..<existing.endIndex) } ?? Data()

        let encoder = JSONEncoder()
        encoder.outputFormatting = [.sortedKeys, .withoutEscapingSlashes]
        encoder.dateEncodingStrategy = .iso8601
        var out = try encoder.encode(header)
        out.append(0x0A)
        out.append(body)
        try out.write(to: url, options: .atomic)
        // Header length and identity participate in sidecar validation. Removing this derived file
        // is cheap and prevents an equal-size rewrite from accidentally preserving stale metadata.
        try? TranscriptRenderIndex.invalidate(for: url)
    }

    /// Build the destination header from a received one: keep the identity fields (version,
    /// sessionID, backend, model, createdAt, revertEpoch — the new owner must *continue* the
    /// revert counter, not restart it, or every existing mirror would read as diverged), point
    /// `worktree`/`nativeTranscriptPath` at the destination's recomputed paths, and set
    /// `backendSessionID` (nil ⇒ fresh-context fallback).
    public static func destinationHeader(
        from source: SessionHeader, worktree: WorktreePath,
        nativeTranscriptPath: String?, backendSessionID: String?
    ) -> SessionHeader {
        SessionHeader(
            version: source.version,
            sessionID: source.sessionID,
            backend: source.backend,
            backendSessionID: backendSessionID,
            worktree: worktree,
            model: source.model,
            nativeTranscriptPath: nativeTranscriptPath,
            createdAt: source.createdAt,
            revertEpoch: source.revertEpoch)
    }
}

/// Reads a transcript back for resume/history paging. No CLI involved — the
/// transcript is canonical (RUNTIME §6).
public struct TranscriptReader: Sendable {
    /// One tail-first paging result. Events inside a batch remain chronological; successive calls
    /// move toward older history and their events should be prepended to the already-visible tail.
    public struct ReverseBatch: Sendable, Equatable {
        public let events: [AgentEvent]
        public let nextByteOffset: Int?

        public init(events: [AgentEvent], nextByteOffset: Int?) {
            self.events = events
            self.nextByteOffset = nextByteOffset
        }
    }

    public let url: URL

    public init(url: URL) {
        self.url = url
    }

    /// Decode just line 0 — for callers that need the header (e.g. the revert path stamping
    /// `revertEpoch`, the fetch server attaching it) without paying to decode every event.
    public func readHeader() throws -> SessionHeader {
        let decoder = JSONDecoder()
        decoder.dateDecodingStrategy = .iso8601
        let handle = try FileHandle(forReadingFrom: url)
        defer { try? handle.close() }
        var line = Data()
        while let chunk = try handle.read(upToCount: Self.readChunkSize), !chunk.isEmpty {
            if let newline = chunk.firstIndex(of: 0x0A) {
                line.append(chunk[..<newline])
                break
            }
            line.append(chunk)
            if Task.isCancelled { throw CancellationError() }
        }
        return try decoder.decode(SessionHeader.self, from: line)
    }

    public func read() throws -> (header: SessionHeader, events: [AgentEvent]) {
        let decoder = JSONDecoder()
        decoder.dateDecodingStrategy = .iso8601

        // Stream JSONL in bounded chunks instead of materializing the file, its line-slice array,
        // and every decoded event at once. The returned event array is the controller's canonical
        // history and is unavoidable today; eliminating the other two full-file representations
        // keeps several concurrent long-session hydrations from multiplying peak memory.
        let handle = try FileHandle(forReadingFrom: url)
        defer { try? handle.close() }
        var buffer = Data()
        var header: SessionHeader?
        var events: [AgentEvent] = []
        var decodedLineCount = 0

        func decodeLine(_ bytes: Data.SubSequence) throws {
            guard !bytes.isEmpty else { return }
            let line: Data.SubSequence
            if bytes.last == 0x0D { line = bytes.dropLast() } else { line = bytes }
            guard !line.isEmpty else { return }
            if header == nil {
                header = try decoder.decode(SessionHeader.self, from: Data(line))
            } else {
                events.append(try decoder.decode(AgentEvent.self, from: Data(line)))
            }
            decodedLineCount += 1
            if decodedLineCount.isMultiple(of: 256), Task.isCancelled {
                throw CancellationError()
            }
        }

        while let chunk = try handle.read(upToCount: Self.readChunkSize), !chunk.isEmpty {
            buffer.append(chunk)
            var lineStart = buffer.startIndex
            while lineStart < buffer.endIndex,
                  let newline = buffer[lineStart...].firstIndex(of: 0x0A)
            {
                try decodeLine(buffer[lineStart..<newline])
                lineStart = buffer.index(after: newline)
            }
            if lineStart > buffer.startIndex { buffer.removeSubrange(buffer.startIndex..<lineStart) }
        }
        if !buffer.isEmpty { try decodeLine(buffer[buffer.startIndex..<buffer.endIndex]) }
        guard let header else {
            throw CocoaError(.fileReadCorruptFile, userInfo: [NSFilePathErrorKey: url.path])
        }
        return (header, events)
    }

    /// Decode one bounded page from the end of the JSONL file. Unlike `read()`, this makes the
    /// newest useful history available without visiting or decoding every older line first. The
    /// file is mapped read-only, so creating each cursor page does not copy the transcript into
    /// heap memory; `beforeByteOffset` resumes immediately before the prior page.
    ///
    /// `maximumEncodedBytes` is a soft bound because one canonical JSON event cannot be split. A
    /// single oversized event therefore occupies its own page, while ordinary pages stay within
    /// both limits. Work is deliberately synchronous and bounded so callers can run one page per
    /// detached task and publish/yield between pages.
    public func readReverseBatch(
        beforeByteOffset: Int? = nil,
        maximumEvents: Int = 24,
        maximumEncodedBytes: Int = 128 * 1024
    ) throws -> ReverseBatch {
        let eventLimit = max(1, maximumEvents)
        let byteLimit = max(1, maximumEncodedBytes)
        let data = try Data(contentsOf: url, options: .mappedIfSafe)
        let start = data.startIndex
        var upper = beforeByteOffset.map {
            data.index(start, offsetBy: min(max(0, $0), data.count))
        } ?? data.endIndex

        // A canonical writer ends each line with LF. Ignore trailing empty lines without making
        // the first page an empty page.
        while upper > start, data[data.index(before: upper)] == 0x0A {
            upper = data.index(before: upper)
        }

        let decoder = JSONDecoder()
        decoder.dateDecodingStrategy = .iso8601
        var newestFirst: [AgentEvent] = []
        newestFirst.reserveCapacity(eventLimit)
        var encodedBytes = 0
        var exhausted = false

        while upper > start {
            if Task.isCancelled { throw CancellationError() }
            guard let newline = data[start..<upper].lastIndex(of: 0x0A) else {
                // The only newline-free prefix is line 0 (the SessionHeader), never an event.
                exhausted = true
                break
            }
            let lower = data.index(after: newline)
            var line = data[lower..<upper]
            if line.last == 0x0D { line = line.dropLast() }
            upper = newline
            guard !line.isEmpty else { continue }

            newestFirst.append(try decoder.decode(AgentEvent.self, from: Data(line)))
            encodedBytes += line.count
            if newestFirst.count >= eventLimit || encodedBytes >= byteLimit { break }
        }

        if upper <= start { exhausted = true }
        let next = exhausted ? nil : data.distance(from: start, to: upper)
        return ReverseBatch(events: Array(newestFirst.reversed()), nextByteOffset: next)
    }

    private static let readChunkSize = 256 * 1024
}