diff --git a/Packages/Shared/CmuxAgentChat/Sources/CmuxAgentChat/Parsing/TranscriptJSONValue.swift b/Packages/Shared/CmuxAgentChat/Sources/CmuxAgentChat/Parsing/TranscriptJSONValue.swift index a67d936ae2d4..b4cfcff2c8ae 100644 --- a/Packages/Shared/CmuxAgentChat/Sources/CmuxAgentChat/Parsing/TranscriptJSONValue.swift +++ b/Packages/Shared/CmuxAgentChat/Sources/CmuxAgentChat/Parsing/TranscriptJSONValue.swift @@ -25,13 +25,59 @@ enum TranscriptJSONValue: Sendable, Equatable, Codable { /// /// - Parameter jsonLine: The raw line text. init?(jsonLine: String) { - let data = Data(jsonLine.utf8) - guard let value = try? JSONDecoder().decode(TranscriptJSONValue.self, from: data) else { - return nil + // Foundation's parser, not `init(from:)`: probing each value's shape + // through `Decoder` costs a failed decode per guess at every level, + // which dominates large records. The pool frees the Foundation objects + // per line, because a parse loop never returns to a runloop. + let value: TranscriptJSONValue? = autoreleasepool { + guard let parsed = try? JSONSerialization.jsonObject( + with: Data(jsonLine.utf8), + options: .fragmentsAllowed + ) else { + return nil + } + return TranscriptJSONValue(foundationValue: parsed) } + guard let value else { return nil } self = value } + /// Converts `JSONSerialization` output, or answers `nil` for anything else. + private init?(foundationValue value: Any) { + switch value { + case let string as String: + self = .string(string) + case let number as NSNumber: + if CFGetTypeID(number) == CFBooleanGetTypeID() { + self = .bool(number.boolValue) + } else if let integer = Int(exactly: number) { + self = .integer(integer) + } else { + self = .number(number.doubleValue) + } + case let object as [String: Any]: + var converted: [String: TranscriptJSONValue] = [:] + converted.reserveCapacity(object.count) + for (key, member) in object { + guard let member = TranscriptJSONValue(foundationValue: member) else { return nil } + converted[key] = member + } + self = .object(converted) + case let array as [Any]: + var converted: [TranscriptJSONValue] = [] + converted.reserveCapacity(array.count) + for element in array { + guard let element = TranscriptJSONValue(foundationValue: element) else { return nil } + converted.append(element) + } + self = .array(converted) + case is NSNull: + self = .null + default: + return nil + } + } + /// Creates a value by trying each JSON shape in turn. /// /// - Parameter decoder: The decoder to read from. diff --git a/Packages/Shared/CmuxAgentChat/Tests/CmuxAgentChatTests/TranscriptJSONValueParityTests.swift b/Packages/Shared/CmuxAgentChat/Tests/CmuxAgentChatTests/TranscriptJSONValueParityTests.swift new file mode 100644 index 000000000000..399c5034b6b6 --- /dev/null +++ b/Packages/Shared/CmuxAgentChat/Tests/CmuxAgentChatTests/TranscriptJSONValueParityTests.swift @@ -0,0 +1,43 @@ +import Foundation +import Testing + +@testable import CmuxAgentChat + +/// `init?(jsonLine:)` must read every line the way the `Decodable` path does. +@Suite("TranscriptJSONValue parity") +struct TranscriptJSONValueParityTests { + private func decoded(_ line: String) -> TranscriptJSONValue? { + try? JSONDecoder().decode(TranscriptJSONValue.self, from: Data(line.utf8)) + } + + @Test( + "the line parser matches the Decodable path", + arguments: [ + #"null"#, #"true"#, #"false"#, #"0"#, #"1"#, #"-1"#, #"1.5"#, #"1.0"#, #"1e3"#, #"-0"#, + #"9007199254740993"#, #"9223372036854775807"#, #"-9223372036854775808"#, + #"9223372036854775808"#, #"18446744073709551616"#, #"1e30"#, + #""""#, #""text""#, #""é\u00e9\n\"\\""#, #""\ud83d\ude00""#, #""\ud83d""#, #""\ude00x""#, "\"\u{1F600}\"", + #"{}"#, #"[]"#, #"[1,"a",null,true,{"k":[2.5]}]"#, + #"{"a":1,"a":2}"#, #"{"bool":true,"one":1,"zero":0,"float":0.25}"#, + #"{"type":"response_item","payload":{"type":"message","content":[{"text":"hi"}]}}"#, + #" {"spaced" : [ 1 , 2 ] } "#, + "", "{", "[1,", "{\"a\":}", "tru", "01", "'a'", "[1,]", "{\"a\":1,}", "\u{1}", + ] + ) + func parity(line: String) { + #expect(TranscriptJSONValue(jsonLine: line) == decoded(line), "line: \(line)") + } + + @Test("booleans and the integers zero and one stay distinct") + func booleansAreNotNumbers() { + #expect(TranscriptJSONValue(jsonLine: "true") == .bool(true)) + #expect(TranscriptJSONValue(jsonLine: "1") == .integer(1)) + #expect(TranscriptJSONValue(jsonLine: "0") == .integer(0)) + } + + @Test("an integer past Int's range reads as a number without trapping") + func integerPastIntRange() { + #expect(TranscriptJSONValue(jsonLine: "9223372036854775808")?.int == nil) + #expect(TranscriptJSONValue(jsonLine: "1e30")?.int == nil) + } +} diff --git a/Packages/macOS/CmuxMobileHost/Package.swift b/Packages/macOS/CmuxMobileHost/Package.swift index 8f6528961e2f..274857abab78 100644 --- a/Packages/macOS/CmuxMobileHost/Package.swift +++ b/Packages/macOS/CmuxMobileHost/Package.swift @@ -34,7 +34,10 @@ let package = Package( ), .testTarget( name: "CmuxMobileHostTests", - dependencies: ["CmuxMobileHost"], + dependencies: [ + "CmuxMobileHost", + .product(name: "CmuxAgentChat", package: "CmuxAgentChat"), + ], swiftSettings: [ .swiftLanguageMode(.v6), ] diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift index 8bece1be1e40..d3aa413e05ef 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift @@ -32,7 +32,7 @@ public actor AgentChatArtifactIndex { case notInSet } - private struct CacheKey: Sendable, Equatable { + struct CacheKey: Sendable, Equatable { let transcriptPath: String let workingDirectory: String? let fileSize: UInt64 @@ -48,7 +48,33 @@ public actor AgentChatArtifactIndex { let snapshot: Snapshot } + /// What must match for two requests to share a build. + struct BuildIdentity: Sendable, Hashable { + let sessionID: String + let transcriptPath: String + let workingDirectory: String? + } + + /// A snapshot being built, shared by every request with the same identity. + struct InFlightBuild: Sendable { + let id = UUID() + let key: CacheKey + let task: Task + var waiterCount = 0 + } + + /// How much of a transcript's tail one snapshot reads. Paths referenced + /// only before this window leave the scope, which fails closed. + static let scanLimits = TranscriptTailLimits( + maxLines: 200_000, + maxBytes: 256 * 1024 * 1024, + maxRecordBytes: 8 * 1024 * 1024 + ) + private var cacheBySessionID = ChatArtifactLRUCache(capacity: 8) + private(set) var builds: [BuildIdentity: InFlightBuild] = [:] + /// Builds started since creation; concurrent misses must share one. + private(set) var startedBuildCount = 0 public func snapshot( sessionID: String, @@ -56,18 +82,83 @@ public actor AgentChatArtifactIndex { transcriptPath: String, workingDirectory: String? ) async throws -> Snapshot { - let key = try Self.cacheKey(transcriptPath: transcriptPath, workingDirectory: workingDirectory) - if let cached = cacheBySessionID.value(forKey: sessionID), cached.key == key { - return cached.snapshot - } - let snapshot = try Self.buildSnapshot( - agentKind: agentKind, + // Builds are shared per session, path and directory, so scope never + // crosses transcripts. After one wait, the build in flight read the file + // after this request arrived and answers it even if the file grew again, + // which stops a transcript growing during every build from holding a + // request forever. + let identity = BuildIdentity( + sessionID: sessionID, transcriptPath: transcriptPath, - workingDirectory: workingDirectory, - generation: key.generation + workingDirectory: workingDirectory ) - cacheBySessionID.insert(CacheEntry(key: key, snapshot: snapshot), forKey: sessionID) - return snapshot + var mayJoinAnyBuild = false + while true { + let key = try Self.cacheKey(transcriptPath: transcriptPath, workingDirectory: workingDirectory) + if let cached = cacheBySessionID.value(forKey: sessionID), cached.key == key { + return cached.snapshot + } + if let build = builds[identity] { + if build.key == key || mayJoinAnyBuild { + return try await join(buildID: build.id, identity: identity) + } + // Older than this request: wait for it rather than scan alongside it. + _ = try? await join(buildID: build.id, identity: identity) + try Task.checkCancellation() + mayJoinAnyBuild = true + continue + } + // Off the actor, so a slow transcript doesn't hold up other sessions. + let task = Task.detached(priority: .utility) { + try Self.buildSnapshot( + agentKind: agentKind, + transcriptPath: transcriptPath, + workingDirectory: workingDirectory, + generation: key.generation + ) + } + let build = InFlightBuild(key: key, task: task) + builds[identity] = build + startedBuildCount += 1 + return try await join(buildID: build.id, identity: identity) + } + } + + /// Waits for a session's build. A cancelled waiter stays until the build + /// ends unless it is the last one, whose departure cancels the build. + private func join(buildID: UUID, identity: BuildIdentity) async throws -> Snapshot { + guard var build = builds[identity], build.id == buildID else { + throw CancellationError() + } + build.waiterCount += 1 + builds[identity] = build + let task = build.task + let key = build.key + let result = await withTaskCancellationHandler { + await task.result + } onCancel: { + Task { await self.leave(buildID: buildID, identity: identity) } + } + // The first waiter to resume publishes the result for later requests. + if builds[identity]?.id == buildID { + builds[identity] = nil + if case .success(let snapshot) = result { + cacheBySessionID.insert(CacheEntry(key: key, snapshot: snapshot), forKey: identity.sessionID) + } + } + return try result.get() + } + + private func leave(buildID: UUID, identity: BuildIdentity) { + guard var build = builds[identity], build.id == buildID else { return } + build.waiterCount -= 1 + guard build.waiterCount == 0 else { + builds[identity] = build + return + } + // Later requests start fresh instead of joining a cancelled build. + builds[identity] = nil + build.task.cancel() } public func canonicalPath( @@ -122,16 +213,25 @@ public actor AgentChatArtifactIndex { workingDirectory: String?, generation: String ) throws -> Snapshot { - let data = try Data(contentsOf: URL(fileURLWithPath: transcriptPath), options: .mappedIfSafe) - let text = String(decoding: data, as: UTF8.self) - let lines = text.split(separator: "\n", omittingEmptySubsequences: false).map(String.init) + guard let reader = TranscriptTailReader(path: transcriptPath) else { + throw CocoaError(.fileReadNoSuchFile) + } + let limits = scanLimits + let window = try reader.scan(limits: limits) + // A last line with no newline yet still counts as a line here. + let lines = reader.lines( + in: window.lineRanges + [window.trailingFragment], + maxRecordBytes: limits.maxRecordBytes + ) let parseResult: ChatTranscriptParseResult switch agentKind { case .codex: - parseResult = CodexTranscriptParser().parse(lines: lines, startingSeq: 0) + parseResult = CodexTranscriptParser().parse(lines: lines, startingSeq: window.firstLine) case .claude, .other: - parseResult = ClaudeTranscriptParser().parse(lines: lines, startingSeq: 0) + parseResult = ClaudeTranscriptParser().parse(lines: lines, startingSeq: window.firstLine) } + // The line sequence ends early on cancellation; don't cache a partial scope. + try Task.checkCancellation() let artifacts = ChatArtifactIndexedReference.derive( from: parseResult.messages, supplementalReferences: parseResult.artifactReferences, diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatTranscriptTailer.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatTranscriptTailer.swift index 0753b5d30527..02c3b204dede 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatTranscriptTailer.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatTranscriptTailer.swift @@ -9,6 +9,11 @@ import Foundation /// Seq stability: `seq` equals the absolute transcript line index. The /// initial backfill may skip a long head (bounded memory), in which case /// pages before the cache report `hasMore` honestly. +/// +/// Reads are bounded in bytes as well as lines: a transcript can hold gigabytes +/// in few lines, so ``TranscriptTailLimits`` caps both the window and any one +/// record. A record over the cap keeps its `seq` and appears as an +/// ``ChatMessageKind/unsupported(_:)`` row. The transcript is never modified. public actor AgentChatTranscriptTailer { /// A live transcript change: newly appended messages and in-place /// updates (tool results that completed earlier messages). @@ -29,7 +34,11 @@ public actor AgentChatTranscriptTailer { private let path: String private let onBatch: @Sendable (Batch) async -> Void - private let maxInitialLines: Int + /// The `rawType` of the row standing in for a record over the size limit. + static let oversizedRecordRawType = "oversized_record" + private static let appendChunkBytes = 1024 * 1024 + + private let limits: TranscriptTailLimits private let maxCachedMessages: Int private var cache: [ChatMessage] = [] @@ -40,20 +49,33 @@ public actor AgentChatTranscriptTailer { /// rotation is detected even when the new file is the same size or /// larger (seeking to the old offset would otherwise skip its head). private var fileInode: UInt64? - private var pendingFragment = Data() + private var assembler: TranscriptLineAssembler private var headTruncated = false private var watchTask: Task? private var watcher: FileWatcher? - private var started = false + private var startTask: Task? + private var loadTask: Task? + private var initialLoadFinished = false + private var stopped = false private var reportedTitle = false + /// Whether ``stop()`` ended this tailer before its first read completed, so + /// its cache says nothing about the transcript. A caller wanting history + /// must use a new tailer. + public var didAbandonInitialLoad: Bool { + stopped && !initialLoadFinished + } + /// Creates a tailer. /// /// - Parameters: /// - sessionID: The session this transcript belongs to. /// - agentKind: Selects the parser (claude or codex). /// - path: Absolute transcript JSONL path. - /// - maxInitialLines: Backfill bound for the first read. + /// - maxInitialLines: Backfill line bound for the first read. + /// - maxInitialBytes: Backfill byte bound for the first read, and the + /// most appended bytes parsed into one batch. + /// - maxRecordBytes: Size above which one record is not decoded. /// - maxCachedMessages: In-memory cache cap; oldest fall out. /// - onBatch: Receives live change batches after the initial load. public init( @@ -61,35 +83,39 @@ public actor AgentChatTranscriptTailer { agentKind: ChatAgentKind, path: String, maxInitialLines: Int = 2000, + maxInitialBytes: Int = 32 * 1024 * 1024, + maxRecordBytes: Int = 8 * 1024 * 1024, maxCachedMessages: Int = 4000, onBatch: @escaping @Sendable (Batch) async -> Void ) { self.sessionID = sessionID self.agentKind = agentKind self.path = path - self.maxInitialLines = maxInitialLines + self.limits = TranscriptTailLimits( + maxLines: maxInitialLines, + maxBytes: maxInitialBytes, + maxRecordBytes: maxRecordBytes + ) + self.assembler = TranscriptLineAssembler(maxRecordBytes: maxRecordBytes) self.maxCachedMessages = maxCachedMessages self.onBatch = onBatch } /// Performs the initial backfill (idempotent) and starts watching for - /// growth. + /// growth. Every caller returns once the backfill has finished or + /// ``stop()`` has cancelled it; the read runs off this actor. public func start() async { - guard !started else { return } - started = true - loadInitialTail() - let watcher = FileWatcher(path: path, throttle: .milliseconds(200)) - self.watcher = watcher - watchTask = Task { [weak self] in - for await _ in watcher.events { - guard let self else { return } - await self.drainNewContent() - } + guard !stopped else { return } + if startTask == nil { + startTask = Task { await self.loadThenWatch() } } + await startTask?.value } - /// Stops watching and releases resources. + /// Cancels any read in progress, stops watching and releases resources. public func stop() async { + stopped = true + loadTask?.cancel() watchTask?.cancel() watchTask = nil if let watcher { @@ -98,6 +124,26 @@ public actor AgentChatTranscriptTailer { watcher = nil } + private func loadThenWatch() async { + let loaded = await loadTail() + guard !stopped else { return } + if let loaded { + commit(loaded) + } + initialLoadFinished = true + let watcher = FileWatcher(path: path, throttle: .milliseconds(200)) + self.watcher = watcher + watchTask = Task { [weak self] in + // Lines written after the scan and before the watcher attached have + // no event of their own, and an idle agent may never write again. + await self?.drainNewContent() + for await _ in watcher.events { + guard let self else { return } + await self.drainNewContent() + } + } + } + /// Serves one history page from the cache, keeping equal-seq groups /// whole at page boundaries. /// @@ -142,45 +188,103 @@ public actor AgentChatTranscriptTailer { // MARK: - Reading - private func loadInitialTail() { - guard let handle = FileHandle(forReadingAtPath: path) else { return } - defer { try? handle.close() } - // Memory-mapped read: newline scanning walks the file without - // copying it; only the bounded tail is decoded into strings. - guard let data = try? Data(contentsOf: URL(fileURLWithPath: path), options: .mappedIfSafe) else { - return - } - var lineStarts: [Int] = [0] - data.withUnsafeBytes { (raw: UnsafeRawBufferPointer) in - for index in 0.. 0 - var lines: [String] = [] - lines.reserveCapacity(completeLineCount - parseStartLine) - for lineIndex in parseStartLine.. LoadedTail? { + let path = path + let agentKind = agentKind + let limits = limits + let task = Task.detached(priority: .utility) { + try Self.readTail(path: path, agentKind: agentKind, limits: limits) } - let outcome = parse(lines: lines, startingSeq: parseStartLine) - cache = outcome.messages - parseState = outcome.state + loadTask = task + defer { loadTask = nil } + // `stop()` may have run between creating the task and publishing it. + if stopped { task.cancel() } + return try? await task.value + } + + /// Reads and parses the tail on the calling task, one record at a time. + private static func readTail( + path: String, + agentKind: ChatAgentKind, + limits: TranscriptTailLimits + ) throws -> LoadedTail? { + guard let reader = TranscriptTailReader(path: path) else { return nil } + let window = try reader.scan(limits: limits) + let outcome = parse( + lines: reader.lines(in: window.lineRanges, maxRecordBytes: limits.maxRecordBytes), + startingSeq: window.firstLine, + state: ChatTranscriptParseState(), + agentKind: agentKind + ) + // The line sequence ends early on cancellation, leaving a partial parse. + try Task.checkCancellation() + return LoadedTail( + window: window, + messages: insertingOversizedRows( + into: outcome.messages, + oversizedSeqs: window.oversizedLines, + fallbackTimestamp: outcome.state.lastTimestamp + ), + state: outcome.state, + assembler: TranscriptLineAssembler( + maxRecordBytes: limits.maxRecordBytes, + fragmentByteCount: window.trailingFragment.upperBound - window.trailingFragment.lowerBound, + fragment: { reader.data(in: window.trailingFragment) } + ) + ) + } + + private func commit(_ loaded: LoadedTail) { + cache = loaded.messages + parseState = loaded.state + assembler = loaded.assembler + byteOffset = loaded.window.byteCount + lineCount = loaded.window.completeLineCount + fileInode = loaded.window.inode + // A positive first line means older lines exist that this tailer won't serve. + headTruncated = loaded.window.firstLine > 0 trimCacheIfNeeded() } + /// Adds a row for each skipped record, keeping ascending `seq` order. + private static func insertingOversizedRows( + into messages: [ChatMessage], + oversizedSeqs: [Int], + fallbackTimestamp: Date? + ) -> [ChatMessage] { + guard !oversizedSeqs.isEmpty else { return messages } + var merged: [ChatMessage] = [] + merged.reserveCapacity(messages.count + oversizedSeqs.count) + var next = messages.startIndex + for seq in oversizedSeqs { + while next < messages.endIndex, messages[next].seq < seq { + merged.append(messages[next]) + next += 1 + } + merged.append( + ChatMessage( + id: "line-\(seq)", + seq: seq, + role: .system, + timestamp: merged.last?.timestamp + ?? fallbackTimestamp + ?? Date(timeIntervalSince1970: 0), + kind: .unsupported(ChatUnsupportedPayload(rawType: oversizedRecordRawType)) + ) + ) + } + merged.append(contentsOf: messages[next...]) + return merged + } + private func drainNewContent() async { guard let handle = FileHandle(forReadingAtPath: path) else { return } defer { try? handle.close() } @@ -193,9 +297,12 @@ public actor AgentChatTranscriptTailer { // new file's head). Reset, re-read from scratch, and tell // clients explicitly: the seq space restarted, and id-based // heuristics can't always detect that (codex line-N ids repeat). + // The old state keeps serving history until the new read is ready. + let loaded = await loadTail() + guard !stopped else { return } byteOffset = 0 lineCount = 0 - pendingFragment = Data() + assembler = TranscriptLineAssembler(maxRecordBytes: limits.maxRecordBytes) cache = [] parseState = ChatTranscriptParseState() headTruncated = false @@ -203,62 +310,97 @@ public actor AgentChatTranscriptTailer { // the file) carries a new first prompt; allow it to be rediscovered // and re-emitted as the title instead of keeping the stale one. reportedTitle = false - loadInitialTail() + if let loaded { + commit(loaded) + } await onBatch(Batch(appended: [], updated: [], discoveredTitle: nil, didReset: true)) return } guard size > byteOffset else { return } try? handle.seek(toOffset: byteOffset) - guard let newData = try? handle.readToEnd(), !newData.isEmpty else { return } - byteOffset += UInt64(newData.count) - - var buffer = pendingFragment - buffer.append(newData) - var lines: [String] = [] - var sliceStart = buffer.startIndex - for index in buffer.indices where buffer[index] == 0x0A { - lines.append(String(decoding: buffer[sliceStart..= limits.maxBytes { + await publish(lines) + lines = [] + bufferedBytes = 0 + } } - pendingFragment = Data(buffer[sliceStart...]) - guard !lines.isEmpty else { return } + await publish(lines) + } + private func publish(_ lines: [TranscriptLineAssembler.Line]) async { + guard !lines.isEmpty, !stopped else { return } let startingSeq = lineCount lineCount += lines.count - let outcome = parse(lines: lines, startingSeq: startingSeq) + var records: [String] = [] + records.reserveCapacity(lines.count) + var oversizedSeqs: [Int] = [] + for (offset, line) in lines.enumerated() { + switch line { + case .record(let text): + records.append(text) + case .oversized: + // An empty line consumes its seq and the parser drops it. + records.append("") + oversizedSeqs.append(startingSeq + offset) + } + } + let outcome = Self.parse( + lines: records, + startingSeq: startingSeq, + state: parseState, + agentKind: agentKind + ) parseState = outcome.state - var updated = outcome.updatedMessages - for update in updated { + let appended = Self.insertingOversizedRows( + into: outcome.messages, + oversizedSeqs: oversizedSeqs, + fallbackTimestamp: outcome.state.lastTimestamp + ) + for update in outcome.updatedMessages { if let index = cache.firstIndex(where: { $0.id == update.id }) { cache[index] = update } } - cache.append(contentsOf: outcome.messages) + cache.append(contentsOf: appended) trimCacheIfNeeded() // Updates for messages that already fell out of the cache are still // pushed: a live client may hold them in its window. - guard !outcome.messages.isEmpty || !updated.isEmpty else { return } + guard !appended.isEmpty || !outcome.updatedMessages.isEmpty else { return } var discoveredTitle: String? if !reportedTitle, let title { reportedTitle = true discoveredTitle = title } - updated = outcome.updatedMessages await onBatch( Batch( - appended: outcome.messages, - updated: updated, + appended: appended, + updated: outcome.updatedMessages, discoveredTitle: discoveredTitle ) ) } - private func parse(lines: [String], startingSeq: Int) -> ChatTranscriptParseResult { + private static func parse( + lines: some Sequence, + startingSeq: Int, + state: ChatTranscriptParseState, + agentKind: ChatAgentKind + ) -> ChatTranscriptParseResult { switch agentKind { case .codex: - return CodexTranscriptParser().parse(lines: lines, startingSeq: startingSeq, state: parseState) + return CodexTranscriptParser().parse(lines: lines, startingSeq: startingSeq, state: state) case .claude, .other: - return ClaudeTranscriptParser().parse(lines: lines, startingSeq: startingSeq, state: parseState) + return ClaudeTranscriptParser().parse(lines: lines, startingSeq: startingSeq, state: state) } } diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift new file mode 100644 index 000000000000..2c11f391e0e7 --- /dev/null +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift @@ -0,0 +1,278 @@ +import Foundation + +/// Limits for one bounded read of a transcript's tail. +struct TranscriptTailLimits: Sendable, Equatable { + /// The most recent complete lines a window may hold. + let maxLines: Int + /// Total bytes of the records a window may decode. + let maxBytes: Int + /// Bytes above which one record is skipped instead of decoded. Must not + /// exceed `maxBytes`, so a kept trailing fragment stays within the budget. + let maxRecordBytes: Int +} + +/// The located tail of a transcript. Holds offsets only, so its size follows +/// `maxLines` and never the size of the file. +struct TranscriptTailWindow: Sendable, Equatable { + /// Identity of the file that was scanned. + let inode: UInt64 + /// Bytes scanned, which is where an append reader continues. + let byteCount: UInt64 + /// Complete (newline-terminated) lines in the file. + let completeLineCount: Int + /// Absolute line index of the first line in `lineRanges`. + let firstLine: Int + /// Byte ranges of the window's lines, oldest first, without their newlines. + let lineRanges: [Range] + /// Absolute line indexes in the window that exceed `maxRecordBytes`. + let oversizedLines: [Int] + /// The bytes after the last newline. + let trailingFragment: Range +} + +/// Reads a transcript's tail without mapping or buffering the whole file. +/// +/// `seq` is an absolute line index, so locating the tail still counts every +/// newline once. The count walks a fixed buffer and keeps a fixed ring of +/// offsets, and both it and record decoding stop when the task is cancelled. +final class TranscriptTailReader { + private static let scanChunkBytes = 64 * 1024 + + private let descriptor: Int32 + + /// Opens a transcript for reading, or answers `nil` when it can't be opened. + init?(path: String) { + let descriptor = open(path, O_RDONLY | O_CLOEXEC) + guard descriptor >= 0 else { return nil } + self.descriptor = descriptor + // The scan touches every page once; keep it out of the file cache. + _ = fcntl(descriptor, F_NOCACHE, 1) + } + + deinit { + close(descriptor) + } + + /// Locates the newest lines that fit `limits`. + /// + /// A line over `maxRecordBytes` stays in the window, costs no budget and is + /// listed in `oversizedLines`, so one huge record can't hide its neighbors. + /// The trailing fragment, which a caller may decode as a last record, is + /// charged first unless it exceeds `maxRecordBytes`. The window ends where + /// the next older line would exceed `maxBytes`, and holds at most + /// `maxLines` lines. + /// + /// - Throws: `CancellationError` when the task is cancelled, or the read + /// error from the file. + func scan(limits: TranscriptTailLimits) throws -> TranscriptTailWindow { + var info = stat() + guard fstat(descriptor, &info) == 0 else { throw Self.lastError() } + + let capacity = max(1, limits.maxLines) + var lineStarts = [UInt64](repeating: 0, count: capacity) + var completeLines = 0 + var currentStart: UInt64 = 0 + var offset: UInt64 = 0 + + let buffer = UnsafeMutableRawBufferPointer.allocate( + byteCount: Self.scanChunkBytes, + alignment: 1 + ) + defer { buffer.deallocate() } + let base = UnsafeRawPointer(buffer.baseAddress!) + while true { + try Task.checkCancellation() + let count = try read(into: buffer, at: offset) + if count == 0 { break } + var cursor = 0 + while cursor < count, let hit = memchr(base + cursor, 0x0A, count - cursor) { + let index = base.distance(to: UnsafeRawPointer(hit)) + lineStarts[completeLines % capacity] = currentStart + completeLines += 1 + currentStart = offset + UInt64(index) + 1 + cursor = index + 1 + } + offset += UInt64(count) + } + + // Newest to oldest, stopping at the first line the budget can't hold. + let oldestRetained = completeLines - min(completeLines, capacity) + var firstLine = completeLines + let fragmentLength = Int(clamping: offset - currentStart) + var usedBytes = fragmentLength > limits.maxRecordBytes ? 0 : fragmentLength + var oversizedNewestFirst: [Int] = [] + var line = completeLines - 1 + while line >= oldestRetained, completeLines - line <= limits.maxLines { + let start = lineStarts[line % capacity] + let end = line + 1 < completeLines ? lineStarts[(line + 1) % capacity] : currentStart + let length = Int(clamping: end - 1 - start) + if length > limits.maxRecordBytes { + oversizedNewestFirst.append(line) + } else { + guard usedBytes + length <= limits.maxBytes else { break } + usedBytes += length + } + firstLine = line + line -= 1 + } + + var ranges: [Range] = [] + ranges.reserveCapacity(completeLines - firstLine) + for line in firstLine..], + maxRecordBytes: Int + ) -> TranscriptTailLineSequence { + TranscriptTailLineSequence(reader: self, ranges: ranges, maxRecordBytes: maxRecordBytes) + } + + /// Reads a byte range, answering what could be read. A short result is a + /// file that shrank underneath the scan; parsers drop the partial record. + func data(in range: Range) -> Data { + let wanted = Int(clamping: range.upperBound - range.lowerBound) + guard wanted > 0 else { return Data() } + var data = Data(count: wanted) + let received = data.withUnsafeMutableBytes { (raw: UnsafeMutableRawBufferPointer) -> Int in + var total = 0 + while total < wanted { + let slice = UnsafeMutableRawBufferPointer(rebasing: raw[total...]) + guard let count = try? read(into: slice, at: range.lowerBound + UInt64(total)), + count > 0 else { + break + } + total += count + } + return total + } + if received < wanted { + data.removeSubrange(received.. Int { + while true { + let count = pread(descriptor, buffer.baseAddress, buffer.count, off_t(offset)) + if count >= 0 { return count } + if errno != EINTR { throw Self.lastError() } + } + } + + private static func lastError() -> any Error { + POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO) + } +} + +/// The lazily decoded lines of a ``TranscriptTailWindow``. +struct TranscriptTailLineSequence: Sequence, IteratorProtocol { + fileprivate let reader: TranscriptTailReader + fileprivate let ranges: [Range] + fileprivate let maxRecordBytes: Int + private var index = 0 + + fileprivate init(reader: TranscriptTailReader, ranges: [Range], maxRecordBytes: Int) { + self.reader = reader + self.ranges = ranges + self.maxRecordBytes = maxRecordBytes + } + + mutating func next() -> String? { + guard index < ranges.count, !Task.isCancelled else { return nil } + let range = ranges[index] + index += 1 + guard range.upperBound - range.lowerBound <= UInt64(maxRecordBytes) else { return "" } + return String(decoding: reader.data(in: range), as: UTF8.self) + } +} + +/// Splits appended transcript bytes into lines, holding at most one record. +/// +/// A record that outgrows `maxRecordBytes` stops being buffered: the rest of it +/// is discarded as it arrives, and it comes out as one ``Line/oversized`` line +/// so it still consumes its `seq`. +struct TranscriptLineAssembler: Sendable { + /// One complete transcript line. + enum Line: Sendable, Equatable { + /// A record within the limit. + case record(String) + /// A record over the limit, whose bytes were not kept. + case oversized + } + + private let maxRecordBytes: Int + private var fragment = Data() + private var isDiscarding = false + + /// Creates an assembler with an empty fragment. + init(maxRecordBytes: Int) { + self.maxRecordBytes = maxRecordBytes + } + + /// Creates an assembler continuing a partial record of a known length, + /// whose bytes are supplied only when it is within the limit. + init(maxRecordBytes: Int, fragmentByteCount: UInt64, fragment: () -> Data) { + self.maxRecordBytes = maxRecordBytes + if fragmentByteCount > UInt64(maxRecordBytes) { + isDiscarding = true + } else { + self.fragment = fragment() + } + } + + /// Consumes a chunk of appended bytes. + /// + /// - Returns: The lines the chunk completed, in file order. + mutating func consume(_ chunk: Data) -> [Line] { + var lines: [Line] = [] + chunk.withUnsafeBytes { (raw: UnsafeRawBufferPointer) in + guard let base = raw.baseAddress else { return } + var cursor = 0 + while cursor < raw.count, let hit = memchr(base + cursor, 0x0A, raw.count - cursor) { + let index = base.distance(to: UnsafeRawPointer(hit)) + lines.append(completeLine(UnsafeRawBufferPointer(rebasing: raw[cursor.. Line { + accumulate(tail) + defer { + fragment = Data() + isDiscarding = false + } + return isDiscarding ? .oversized : .record(String(decoding: fragment, as: UTF8.self)) + } + + private mutating func accumulate(_ bytes: UnsafeRawBufferPointer) { + guard !isDiscarding else { return } + if fragment.count + bytes.count > maxRecordBytes { + fragment = Data() + isDiscarding = true + } else { + fragment.append(contentsOf: bytes) + } + } +} diff --git a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/AgentChatTranscriptTailerBoundsTests.swift b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/AgentChatTranscriptTailerBoundsTests.swift new file mode 100644 index 000000000000..88ee6cb47f3c --- /dev/null +++ b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/AgentChatTranscriptTailerBoundsTests.swift @@ -0,0 +1,154 @@ +import CmuxAgentChat +import Foundation +import Testing +@testable import CmuxMobileHost + +/// Regressions for https://github.com/manaflow-ai/cmux/issues/31300: the first +/// read of a transcript must be bounded in bytes, not only in lines. +@Suite("Agent chat transcript tailer bounds", .timeLimit(.minutes(2))) +struct AgentChatTranscriptTailerBoundsTests { + private static let mebibyte = 1024 * 1024 + + private func codexLine(_ text: String) -> String { + let object: [String: Any] = [ + "timestamp": "2026-06-11T21:38:05.381Z", + "type": "response_item", + "payload": [ + "type": "message", + "role": "user", + "content": [["type": "input_text", "text": text]], + ], + ] + let data = try! JSONSerialization.data(withJSONObject: object) + return String(decoding: data, as: UTF8.self) + } + + private func makeTranscript(_ contents: String) throws -> URL { + let url = FileManager.default.temporaryDirectory + .appendingPathComponent("tailer-bounds-\(UUID().uuidString).jsonl") + try contents.write(to: url, atomically: false, encoding: .utf8) + return url + } + + private func makeTailer(url: URL) -> AgentChatTranscriptTailer { + AgentChatTranscriptTailer(sessionID: "session", agentKind: .codex, path: url.path) { _ in } + } + + private func isOversizedRow(_ message: ChatMessage) -> Bool { + if case .unsupported = message.kind { return true } + return false + } + + @Test("one valid record over the size limit is skipped with a visible row and keeps its seq") + func oversizedRecordKeepsSeq() async throws { + let huge = codexLine(String(repeating: "x", count: 9 * Self.mebibyte)) + let url = try makeTranscript(codexLine("before") + "\n" + huge + "\n" + codexLine("after") + "\n") + defer { try? FileManager.default.removeItem(at: url) } + + let tailer = makeTailer(url: url) + await tailer.start() + let page = await tailer.history(beforeSeq: nil, limit: 100) + await tailer.stop() + + #expect(page.messages.map(\.seq) == [0, 1, 2]) + #expect(isOversizedRow(page.messages[1])) + #expect(!isOversizedRow(page.messages[2])) + #expect(page.hasMore == false) + } + + @Test("few lines holding many bytes stop at the byte budget with absolute seqs and honest paging") + func byteBudgetBoundsFewLongLines() async throws { + // Each record is under the single-record limit; together they are far + // over the window, and far under the 2,000 line limit. + let record = codexLine(String(repeating: "y", count: 2 * Self.mebibyte)) + let lineCount = 40 + let contents = String(repeating: record + "\n", count: lineCount) + let url = try makeTranscript(contents) + defer { try? FileManager.default.removeItem(at: url) } + + let tailer = makeTailer(url: url) + await tailer.start() + let page = await tailer.history(beforeSeq: nil, limit: 100) + await tailer.stop() + + let seqs = page.messages.map(\.seq) + #expect(seqs.last == lineCount - 1) + #expect(seqs.count <= 16) + #expect((seqs.first ?? 0) > 0) + #expect(seqs == Array((seqs.first ?? 0)...(lineCount - 1))) + #expect(page.hasMore == true) + } + + @Test("a small transcript loads whole, pages by seq and reports no older history") + func smallTranscriptIsUnchanged() async throws { + let contents = (0..<5).map { codexLine("line \($0)") + "\n" }.joined() + let url = try makeTranscript(contents) + defer { try? FileManager.default.removeItem(at: url) } + + let tailer = makeTailer(url: url) + await tailer.start() + let newest = await tailer.history(beforeSeq: nil, limit: 2) + let older = await tailer.history(beforeSeq: 3, limit: 100) + await tailer.stop() + + #expect(newest.messages.map(\.seq) == [3, 4]) + #expect(newest.hasMore == true) + #expect(older.messages.map(\.seq) == [0, 1, 2]) + #expect(older.hasMore == false) + } + + @Test("a trailing partial record joins the bytes appended after the first read") + func trailingPartialCompletesOnAppend() async throws { + let last = codexLine("completed later") + let splitAt = last.index(last.startIndex, offsetBy: last.count / 2) + let url = try makeTranscript(codexLine("first") + "\n" + String(last[.. URL { + let url = FileManager.default.temporaryDirectory + .appendingPathComponent("tail-reader-\(UUID().uuidString).jsonl") + try Data(contents.utf8).write(to: url) + return url + } + + private func decodedLines(_ reader: TranscriptTailReader, _ window: TranscriptTailWindow, maxRecordBytes: Int) -> [String] { + Array(reader.lines(in: window.lineRanges, maxRecordBytes: maxRecordBytes)) + } + + @Test("the window holds the newest lines within the line limit, with absolute indexes") + func lineLimit() throws { + let url = try makeFile("a\nbb\nccc\ndddd\n") + defer { try? FileManager.default.removeItem(at: url) } + let reader = try #require(TranscriptTailReader(path: url.path)) + let limits = TranscriptTailLimits(maxLines: 2, maxBytes: 100, maxRecordBytes: 100) + let window = try reader.scan(limits: limits) + #expect(window.completeLineCount == 4) + #expect(window.firstLine == 2) + #expect(window.byteCount == 14) + #expect(decodedLines(reader, window, maxRecordBytes: 100) == ["ccc", "dddd"]) + } + + @Test("the window stops at the byte budget but a skipped record costs none of it") + func byteBudget() throws { + let url = try makeFile("aaaa\nbbbb\n\(String(repeating: "h", count: 50))\ncccc\n") + defer { try? FileManager.default.removeItem(at: url) } + let reader = try #require(TranscriptTailReader(path: url.path)) + let limits = TranscriptTailLimits(maxLines: 100, maxBytes: 8, maxRecordBytes: 10) + let window = try reader.scan(limits: limits) + #expect(window.firstLine == 1) + #expect(window.oversizedLines == [2]) + #expect(decodedLines(reader, window, maxRecordBytes: 10) == ["bbbb", "", "cccc"]) + } + + @Test("bytes after the last newline are a fragment, not a line") + func trailingFragment() throws { + let url = try makeFile("line\npartial") + defer { try? FileManager.default.removeItem(at: url) } + let reader = try #require(TranscriptTailReader(path: url.path)) + let limits = TranscriptTailLimits(maxLines: 10, maxBytes: 100, maxRecordBytes: 100) + let window = try reader.scan(limits: limits) + #expect(window.completeLineCount == 1) + #expect(window.byteCount == 12) + #expect(String(decoding: reader.data(in: window.trailingFragment), as: UTF8.self) == "partial") + } + + @Test("the trailing fragment is charged to the byte budget") + func fragmentChargedToBudget() throws { + let url = try makeFile("aaaa\nbbbb\npartial") + defer { try? FileManager.default.removeItem(at: url) } + let reader = try #require(TranscriptTailReader(path: url.path)) + let limits = TranscriptTailLimits(maxLines: 10, maxBytes: 12, maxRecordBytes: 100) + let window = try reader.scan(limits: limits) + // 7 fragment bytes leave room for "bbbb" but not "aaaa". + #expect(window.firstLine == 1) + #expect(decodedLines(reader, window, maxRecordBytes: 100) == ["bbbb"]) + } + + @Test("an oversized trailing fragment costs no budget") + func oversizedFragmentCostsNothing() throws { + let url = try makeFile("aaaa\nbbbb\n\(String(repeating: "p", count: 50))") + defer { try? FileManager.default.removeItem(at: url) } + let reader = try #require(TranscriptTailReader(path: url.path)) + let limits = TranscriptTailLimits(maxLines: 10, maxBytes: 8, maxRecordBytes: 10) + let window = try reader.scan(limits: limits) + #expect(window.firstLine == 0) + } + + @Test("a zero line limit yields an empty window") + func zeroLineLimit() throws { + let url = try makeFile("a\nb\n") + defer { try? FileManager.default.removeItem(at: url) } + let reader = try #require(TranscriptTailReader(path: url.path)) + let limits = TranscriptTailLimits(maxLines: 0, maxBytes: 100, maxRecordBytes: 100) + let window = try reader.scan(limits: limits) + #expect(window.completeLineCount == 2) + #expect(window.firstLine == 2) + #expect(window.lineRanges.isEmpty) + } + + @Test("an empty file has an empty window") + func emptyFile() throws { + let url = try makeFile("") + defer { try? FileManager.default.removeItem(at: url) } + let reader = try #require(TranscriptTailReader(path: url.path)) + let limits = TranscriptTailLimits(maxLines: 10, maxBytes: 100, maxRecordBytes: 100) + let window = try reader.scan(limits: limits) + #expect(window.completeLineCount == 0) + #expect(window.lineRanges.isEmpty) + #expect(window.trailingFragment.isEmpty) + } + + @Test("a multi-gigabyte file yields a window sized by the limits") + func multiGigabyteFile() throws { + // Sparse, so the test exercises real offsets without the disk cost. + let url = try makeFile("") + defer { try? FileManager.default.removeItem(at: url) } + let handle = try FileHandle(forWritingTo: url) + let holeBytes: UInt64 = 3 * 1024 * 1024 * 1024 + try handle.truncate(atOffset: holeBytes) + try handle.seek(toOffset: holeBytes) + try handle.write(contentsOf: Data("\nsecond\nthird\n".utf8)) + try handle.close() + + let reader = try #require(TranscriptTailReader(path: url.path)) + let limits = TranscriptTailLimits(maxLines: 4, maxBytes: 1024, maxRecordBytes: 1024) + let window = try reader.scan(limits: limits) + #expect(window.completeLineCount == 3) + #expect(window.firstLine == 0) + #expect(window.oversizedLines == [0]) + #expect(window.byteCount == holeBytes + 14) + #expect(decodedLines(reader, window, maxRecordBytes: 1024) == ["", "second", "third"]) + } + + @Test("a cancelled task stops scanning") + func cancelledScanThrows() async throws { + let url = try makeFile("a\nb\n") + defer { try? FileManager.default.removeItem(at: url) } + let limits = TranscriptTailLimits(maxLines: 10, maxBytes: 100, maxRecordBytes: 100) + let path = url.path + let task = Task.detached { () -> Bool in + withUnsafeCurrentTask { $0?.cancel() } + let reader = TranscriptTailReader(path: path) + do { + _ = try reader?.scan(limits: limits) + return false + } catch { + return error is CancellationError + } + } + #expect(await task.value) + } + + @Test("cancellation between records ends the line sequence") + func cancelledLinesStop() async throws { + let url = try makeFile("a\nb\nc\n") + defer { try? FileManager.default.removeItem(at: url) } + let limits = TranscriptTailLimits(maxLines: 10, maxBytes: 100, maxRecordBytes: 100) + let path = url.path + let task = Task.detached { () -> [String] in + guard let reader = TranscriptTailReader(path: path), + let window = try? reader.scan(limits: limits) else { return [] } + var seen: [String] = [] + for line in reader.lines(in: window.lineRanges, maxRecordBytes: 100) { + seen.append(line) + withUnsafeCurrentTask { $0?.cancel() } + } + return seen + } + #expect(await task.value == ["a"]) + } + + @Test("stopping during the first read abandons it and start returns") + func stopAbandonsInitialLoad() async throws { + let url = try makeFile("") + defer { try? FileManager.default.removeItem(at: url) } + let handle = try FileHandle(forWritingTo: url) + try handle.truncate(atOffset: 64 * 1024 * 1024 * 1024) + try handle.close() + + let tailer = AgentChatTranscriptTailer(sessionID: "session", agentKind: .codex, path: url.path) { _ in } + let started = Task { await tailer.start() } + await tailer.stop() + await started.value + #expect(await tailer.didAbandonInitialLoad) + } + + @Test("a completed first read is not abandoned by a later stop") + func stopAfterLoadKeepsHistory() async throws { + let url = try makeFile("{}\n") + defer { try? FileManager.default.removeItem(at: url) } + let tailer = AgentChatTranscriptTailer(sessionID: "session", agentKind: .codex, path: url.path) { _ in } + await tailer.start() + await tailer.stop() + #expect(await tailer.didAbandonInitialLoad == false) + } + + @Test("a backlog larger than the batch size arrives in several batches with contiguous seqs") + func longBacklogPublishesInBatches() async throws { + let url = try makeFile("") + defer { try? FileManager.default.removeItem(at: url) } + let (batches, continuation) = AsyncStream.makeStream(of: AgentChatTranscriptTailer.Batch.self) + let tailer = AgentChatTranscriptTailer( + sessionID: "session", + agentKind: .codex, + path: url.path, + maxInitialBytes: 1024 * 1024, + maxRecordBytes: 1024 * 1024 + ) { batch in + continuation.yield(batch) + } + await tailer.start() + + let text = String(repeating: "z", count: 16 * 1024) + let line = #"{"timestamp":"2026-06-11T21:38:05.381Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"\#(text)"}]}}"# + let lineCount = 256 + let handle = try FileHandle(forWritingTo: url) + try handle.write(contentsOf: Data(String(repeating: line + "\n", count: lineCount).utf8)) + try handle.close() + + var seqs: [Int] = [] + var batchCount = 0 + for await batch in batches { + seqs.append(contentsOf: batch.appended.map(\.seq)) + batchCount += 1 + if seqs.last == lineCount - 1 { break } + } + await tailer.stop() + continuation.finish() + #expect(seqs == Array(0.. 1) + } + + @Test("an oversized record appended live keeps its seq and its row") + func oversizedAppend() async throws { + let url = try makeFile("") + defer { try? FileManager.default.removeItem(at: url) } + let (batches, continuation) = AsyncStream.makeStream(of: AgentChatTranscriptTailer.Batch.self) + let tailer = AgentChatTranscriptTailer( + sessionID: "session", + agentKind: .codex, + path: url.path, + maxInitialBytes: 1024, + maxRecordBytes: 64 + ) { batch in + continuation.yield(batch) + } + await tailer.start() + + let handle = try FileHandle(forWritingTo: url) + try handle.write(contentsOf: Data((String(repeating: "h", count: 200) + "\n").utf8)) + try handle.close() + + var appended: [ChatMessage] = [] + for await batch in batches { + appended = batch.appended + break + } + await tailer.stop() + continuation.finish() + #expect(appended.map(\.seq) == [0]) + if case .unsupported(let payload) = appended.first?.kind { + #expect(payload.rawType == AgentChatTranscriptTailer.oversizedRecordRawType) + } else { + Issue.record("expected an oversized row") + } + } +} + +@Suite("Agent chat artifact index builds", .timeLimit(.minutes(2))) +struct AgentChatArtifactIndexBuildTests { + private func makeFile(sparseBytes: UInt64 = 0, contents: String = "") throws -> URL { + let url = FileManager.default.temporaryDirectory + .appendingPathComponent("artifact-index-\(UUID().uuidString).jsonl") + try Data(contents.utf8).write(to: url) + if sparseBytes > 0 { + let handle = try FileHandle(forWritingTo: url) + try handle.truncate(atOffset: sparseBytes) + try handle.close() + } + return url + } + + @Test("concurrent misses for one session share one build") + func concurrentMissesShareOneBuild() async throws { + let url = try makeFile(sparseBytes: 2 * 1024 * 1024 * 1024) + defer { try? FileManager.default.removeItem(at: url) } + let index = AgentChatArtifactIndex() + let path = url.path + await withThrowingTaskGroup(of: Void.self) { group in + for _ in 0..<16 { + group.addTask { + _ = try await index.snapshot( + sessionID: "session", + agentKind: .codex, + transcriptPath: path, + workingDirectory: nil + ) + } + } + while let result = await group.nextResult() { + #expect(throws: Never.self) { try result.get() } + } + } + #expect(await index.startedBuildCount == 1) + } + + @Test("an unchanged transcript answers from the cache") + func unchangedTranscriptUsesCache() async throws { + let url = try makeFile(contents: "{}\n") + defer { try? FileManager.default.removeItem(at: url) } + let index = AgentChatArtifactIndex() + for _ in 0..<2 { + _ = try await index.snapshot( + sessionID: "session", + agentKind: .codex, + transcriptPath: url.path, + workingDirectory: nil + ) + } + #expect(await index.startedBuildCount == 1) + } + + /// Waits until `count` requests have joined a build, failing after a + /// deadline instead of hanging when one answered from the cache. + private func waitForRequests(_ count: Int, on index: AgentChatArtifactIndex) async { + let deadline = ContinuousClock.now + .seconds(30) + while await index.waitingRequestCount < count { + guard ContinuousClock.now < deadline else { + let joined = await index.waitingRequestCount + Issue.record("only \(joined) of \(count) requests joined a build before the deadline") + return + } + await Task.yield() + } + } + + @Test("one cancelled request doesn't fail another waiting on the same build") + func cancelledWaiterLeavesOthers() async throws { + let url = try makeFile(sparseBytes: 64 * 1024 * 1024 * 1024) + defer { try? FileManager.default.removeItem(at: url) } + let index = AgentChatArtifactIndex() + let path = url.path + let request: @Sendable () async throws -> AgentChatArtifactIndex.Snapshot = { + try await index.snapshot( + sessionID: "session", + agentKind: .codex, + transcriptPath: path, + workingDirectory: nil + ) + } + let kept = Task { try await request() } + let cancelled = Task { try await request() } + await waitForRequests(2, on: index) + cancelled.cancel() + _ = try await kept.value + #expect(await index.startedBuildCount == 1) + } + + @Test("the last waiter leaving cancels the build, and nothing is cached") + func lastWaiterCancelsBuild() async throws { + let url = try makeFile(sparseBytes: 64 * 1024 * 1024 * 1024) + defer { try? FileManager.default.removeItem(at: url) } + let index = AgentChatArtifactIndex() + let path = url.path + let request = Task { + try await index.snapshot( + sessionID: "session", + agentKind: .codex, + transcriptPath: path, + workingDirectory: nil + ) + } + await waitForRequests(1, on: index) + request.cancel() + await #expect(throws: CancellationError.self) { + try await request.value + } + #expect(await index.waitingRequestCount == 0) + } +} + +@Suite("Transcript line assembler") +struct TranscriptLineAssemblerTests { + private func data(_ text: String) -> Data { Data(text.utf8) } + + @Test("a line split across chunks comes out whole, once") + func splitAcrossChunks() { + var assembler = TranscriptLineAssembler(maxRecordBytes: 100) + #expect(assembler.consume(data("ab")) == []) + #expect(assembler.consume(data("c\nd\ne")) == [.record("abc"), .record("d")]) + #expect(assembler.consume(data("\n")) == [.record("e")]) + } + + @Test("a record over the limit is dropped across chunks and consumes one line") + func oversizedAcrossChunks() { + var assembler = TranscriptLineAssembler(maxRecordBytes: 4) + #expect(assembler.consume(data("abc")) == []) + #expect(assembler.consume(data("defg")) == []) + #expect(assembler.consume(data("hij\nok\n")) == [.oversized, .record("ok")]) + } + + @Test("a record exactly at the limit is kept") + func exactlyAtLimit() { + var assembler = TranscriptLineAssembler(maxRecordBytes: 4) + #expect(assembler.consume(data("abcd\n")) == [.record("abcd")]) + } + + @Test("a fragment from the first read continues, and an oversized one is not read") + func seededFragment() { + var kept = TranscriptLineAssembler(maxRecordBytes: 10, fragmentByteCount: 2, fragment: { Data("ab".utf8) }) + #expect(kept.consume(data("c\n")) == [.record("abc")]) + + var discarded = TranscriptLineAssembler(maxRecordBytes: 10, fragmentByteCount: 50, fragment: { + Issue.record("an oversized fragment must not be read") + return Data() + }) + #expect(discarded.consume(data("tail\nnext\n")) == [.oversized, .record("next")]) + } +} + +extension AgentChatArtifactIndex { + /// Requests waiting on any build, so a test can act once they have joined. + var waitingRequestCount: Int { + builds.values.reduce(0) { $0 + $1.waiterCount } + } +} diff --git a/Sources/Mobile/AgentChat/AgentChatTranscriptService.swift b/Sources/Mobile/AgentChat/AgentChatTranscriptService.swift index 36e61634bcc3..cdc87f093df2 100644 --- a/Sources/Mobile/AgentChat/AgentChatTranscriptService.swift +++ b/Sources/Mobile/AgentChat/AgentChatTranscriptService.swift @@ -476,6 +476,11 @@ final class AgentChatTranscriptService { tailer = resolvedTailer } await tailer.start() + // The session ended while this tailer was still reading, which cancelled + // the read. Its removal from `tailers` makes the retry build a new one. + if await tailer.didAbandonInitialLoad { + return await history(sessionID: sessionID, beforeSeq: beforeSeq, limit: limit) + } let page = await tailer.history(beforeSeq: beforeSeq, limit: limit) if record.title == nil, let title = await tailer.title { registry.update(sessionID: sessionID) { $0.title = title }