From 304681ba36269afdf7aa330e4a5bfa18851cf7dd Mon Sep 17 00:00:00 2001 From: Austin Wang <38676809+austinywang@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:23:20 -0800 Subject: [PATCH 1/8] Add failing tests for bounding the first transcript read Few long lines can hold gigabytes, so a 2,000 line limit is not a byte limit. Co-Authored-By: Claude Opus 5.5 --- Packages/macOS/CmuxMobileHost/Package.swift | 5 +- ...AgentChatTranscriptTailerBoundsTests.swift | 154 ++++++++++++++++++ 2 files changed, 158 insertions(+), 1 deletion(-) create mode 100644 Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/AgentChatTranscriptTailerBoundsTests.swift 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/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[.. Date: Mon, 2 Nov 2026 15:23:22 -0800 Subject: [PATCH 2/8] Bound the first transcript read in bytes and make it cancellable The tailer now counts newlines through a fixed buffer, keeps a fixed ring of offsets, caps the window at 32 MiB and one record at 8 MiB, and decodes one record at a time off the actor. stop() cancels the read. Appends read in chunks under the same record cap. The artifact index shares the reader rather than copying the whole transcript into one String. Co-Authored-By: Claude Opus 5.5 --- .../Parsing/TranscriptJSONValue.swift | 52 +++- .../TranscriptJSONValueParityTests.swift | 43 +++ .../AgentChat/AgentChatArtifactIndex.swift | 48 ++- .../AgentChat/AgentChatTranscriptTailer.swift | 291 +++++++++++++----- .../AgentChat/TranscriptTailReader.swift | 273 ++++++++++++++++ .../TranscriptTailReaderTests.swift | 226 ++++++++++++++ .../AgentChatTranscriptService.swift | 5 + 7 files changed, 848 insertions(+), 90 deletions(-) create mode 100644 Packages/Shared/CmuxAgentChat/Tests/CmuxAgentChatTests/TranscriptJSONValueParityTests.swift create mode 100644 Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift create mode 100644 Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift 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..54db0afd474b --- /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""#, #""éé\n\"\\""#, #""😀""#, #""😀""#, + #"{}"#, #"[]"#, #"[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/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift index 8bece1be1e40..c03805b7f64b 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift @@ -48,6 +48,14 @@ public actor AgentChatArtifactIndex { let snapshot: Snapshot } + /// 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) public func snapshot( @@ -60,12 +68,21 @@ public actor AgentChatArtifactIndex { if let cached = cacheBySessionID.value(forKey: sessionID), cached.key == key { return cached.snapshot } - let snapshot = try Self.buildSnapshot( - agentKind: agentKind, - transcriptPath: transcriptPath, - workingDirectory: workingDirectory, - generation: key.generation - ) + // Off the actor, so a slow transcript doesn't hold up other sessions' + // lookups, and cancelled along with the request that asked for it. + let build = Task.detached(priority: .utility) { + try Self.buildSnapshot( + agentKind: agentKind, + transcriptPath: transcriptPath, + workingDirectory: workingDirectory, + generation: key.generation + ) + } + let snapshot = try await withTaskCancellationHandler { + try await build.value + } onCancel: { + build.cancel() + } cacheBySessionID.insert(CacheEntry(key: key, snapshot: snapshot), forKey: sessionID) return snapshot } @@ -122,16 +139,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..785e1a2b75cc 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,23 @@ 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 + 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 +185,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 +294,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 +307,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..e4045ff3c5c7 --- /dev/null +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift @@ -0,0 +1,273 @@ +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. + 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 window ends where the next older line would exceed `maxBytes`. + /// + /// - 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 + var usedBytes = 0 + var oversizedNewestFirst: [Int] = [] + var line = completeLines - 1 + while line >= oldestRetained { + 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/TranscriptTailReaderTests.swift b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift new file mode 100644 index 000000000000..f6d86617e771 --- /dev/null +++ b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift @@ -0,0 +1,226 @@ +import CmuxAgentChat +import Foundation +import Testing +@testable import CmuxMobileHost + +@Suite("Transcript tail reader", .timeLimit(.minutes(2))) +struct TranscriptTailReaderTests { + private func makeFile(_ contents: String) throws -> 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("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("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("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")]) + } +} 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 } From e67b68ecf545fb1da38d6449cdf1b52bc3044d6e Mon Sep 17 00:00:00 2001 From: Austin Wang <38676809+austinywang@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:39:42 -0800 Subject: [PATCH 3/8] Share one artifact snapshot build across concurrent requests The index actor now suspends while building, so concurrent misses each started a full scan. They join one build per session, and the last waiter to cancel cancels it. The tailer also drains once when its watcher attaches, so a line written during the first read isn't left unseen. Co-Authored-By: Claude Opus 5.5 --- .../AgentChat/AgentChatArtifactIndex.swift | 94 +++++++++++++++---- .../AgentChat/AgentChatTranscriptTailer.swift | 3 + 2 files changed, 80 insertions(+), 17 deletions(-) diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift index c03805b7f64b..34f143342588 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift @@ -48,6 +48,14 @@ public actor AgentChatArtifactIndex { let snapshot: Snapshot } + /// A snapshot being built, shared by every request waiting on the session. + private 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( @@ -57,6 +65,9 @@ public actor AgentChatArtifactIndex { ) private var cacheBySessionID = ChatArtifactLRUCache(capacity: 8) + private var buildsBySessionID: [String: InFlightBuild] = [:] + /// Builds started since creation; concurrent misses must share one. + private(set) var startedBuildCount = 0 public func snapshot( sessionID: String, @@ -64,27 +75,76 @@ 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 + // A build read the file after this request arrived, so it answers this + // request whatever its key. Waiting at most once stops a transcript that + // grows during every build from holding a request forever. + 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 = buildsBySessionID[sessionID] { + if build.key == key || mayJoinAnyBuild { + return try await join(buildID: build.id, sessionID: sessionID) + } + // Older than this request: wait for it rather than scan alongside it. + _ = try? await join(buildID: build.id, sessionID: sessionID) + 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) + buildsBySessionID[sessionID] = build + startedBuildCount += 1 + return try await join(buildID: build.id, sessionID: sessionID) } - // Off the actor, so a slow transcript doesn't hold up other sessions' - // lookups, and cancelled along with the request that asked for it. - let build = Task.detached(priority: .utility) { - try Self.buildSnapshot( - agentKind: agentKind, - transcriptPath: transcriptPath, - workingDirectory: workingDirectory, - generation: key.generation - ) + } + + /// 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, sessionID: String) async throws -> Snapshot { + guard var build = buildsBySessionID[sessionID], build.id == buildID else { + throw CancellationError() } - let snapshot = try await withTaskCancellationHandler { - try await build.value + build.waiterCount += 1 + buildsBySessionID[sessionID] = build + let task = build.task + let key = build.key + let result = await withTaskCancellationHandler { + await task.result } onCancel: { - build.cancel() + Task { await self.leave(buildID: buildID, sessionID: sessionID) } + } + // The first waiter to resume publishes the result for later requests. + if buildsBySessionID[sessionID]?.id == buildID { + buildsBySessionID[sessionID] = nil + if case .success(let snapshot) = result { + cacheBySessionID.insert(CacheEntry(key: key, snapshot: snapshot), forKey: sessionID) + } + } + return try result.get() + } + + private func leave(buildID: UUID, sessionID: String) { + guard var build = buildsBySessionID[sessionID], build.id == buildID else { return } + build.waiterCount -= 1 + guard build.waiterCount == 0 else { + buildsBySessionID[sessionID] = build + return } - cacheBySessionID.insert(CacheEntry(key: key, snapshot: snapshot), forKey: sessionID) - return snapshot + // Later requests start fresh instead of joining a cancelled build. + buildsBySessionID[sessionID] = nil + build.task.cancel() } public func canonicalPath( diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatTranscriptTailer.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatTranscriptTailer.swift index 785e1a2b75cc..02c3b204dede 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatTranscriptTailer.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatTranscriptTailer.swift @@ -134,6 +134,9 @@ public actor AgentChatTranscriptTailer { 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() From 9e97cc33eac41cd84295b5060f07a594266c73ea Mon Sep 17 00:00:00 2001 From: Austin Wang <38676809+austinywang@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:41:24 -0800 Subject: [PATCH 4/8] Test shared artifact builds, batched backlogs and escaped JSON strings Co-Authored-By: Claude Opus 5.5 --- .../TranscriptJSONValueParityTests.swift | 2 +- .../TranscriptTailReaderTests.swift | 112 ++++++++++++++++++ 2 files changed, 113 insertions(+), 1 deletion(-) diff --git a/Packages/Shared/CmuxAgentChat/Tests/CmuxAgentChatTests/TranscriptJSONValueParityTests.swift b/Packages/Shared/CmuxAgentChat/Tests/CmuxAgentChatTests/TranscriptJSONValueParityTests.swift index 54db0afd474b..399c5034b6b6 100644 --- a/Packages/Shared/CmuxAgentChat/Tests/CmuxAgentChatTests/TranscriptJSONValueParityTests.swift +++ b/Packages/Shared/CmuxAgentChat/Tests/CmuxAgentChatTests/TranscriptJSONValueParityTests.swift @@ -16,7 +16,7 @@ struct TranscriptJSONValueParityTests { #"null"#, #"true"#, #"false"#, #"0"#, #"1"#, #"-1"#, #"1.5"#, #"1.0"#, #"1e3"#, #"-0"#, #"9007199254740993"#, #"9223372036854775807"#, #"-9223372036854775808"#, #"9223372036854775808"#, #"18446744073709551616"#, #"1e30"#, - #""""#, #""text""#, #""éé\n\"\\""#, #""😀""#, #""😀""#, + #""""#, #""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"}]}}"#, diff --git a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift index f6d86617e771..cf60dcccac59 100644 --- a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift +++ b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift @@ -150,6 +150,42 @@ struct TranscriptTailReaderTests { #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("") @@ -186,6 +222,82 @@ struct TranscriptTailReaderTests { } } +@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) + } + + @Test("one cancelled request doesn't fail another waiting on the same build") + func cancelledWaiterLeavesOthers() async throws { + let url = try makeFile(sparseBytes: 8 * 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() } + cancelled.cancel() + _ = try await kept.value + _ = try? await cancelled.value + } +} + @Suite("Transcript line assembler") struct TranscriptLineAssemblerTests { private func data(_ text: String) -> Data { Data(text.utf8) } From 0c864dc14df7cbd2ae73a3fb0c45a45fcc643708 Mon Sep 17 00:00:00 2001 From: Austin Wang <38676809+austinywang@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:54:18 -0800 Subject: [PATCH 5/8] Keep shared artifact builds to one transcript and working directory A request that had waited could join a build for an older path or directory and be authorized against its scope. Builds are now shared per session, path and directory. Co-Authored-By: Claude Opus 5.5 --- .../AgentChat/AgentChatArtifactIndex.swift | 61 ++++++++++++------- .../TranscriptTailReaderTests.swift | 33 +++++++++- 2 files changed, 71 insertions(+), 23 deletions(-) diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift index 34f143342588..206284a9a046 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift @@ -48,7 +48,14 @@ public actor AgentChatArtifactIndex { let snapshot: Snapshot } - /// A snapshot being built, shared by every request waiting on the session. + /// What must match for two requests to share a build. + private 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. private struct InFlightBuild: Sendable { let id = UUID() let key: CacheKey @@ -65,31 +72,43 @@ public actor AgentChatArtifactIndex { ) private var cacheBySessionID = ChatArtifactLRUCache(capacity: 8) - private var buildsBySessionID: [String: InFlightBuild] = [:] + private var builds: [BuildIdentity: InFlightBuild] = [:] /// Builds started since creation; concurrent misses must share one. private(set) var startedBuildCount = 0 + /// Requests waiting on any build, so a test can act once they have joined. + var waitingRequestCount: Int { + builds.values.reduce(0) { $0 + $1.waiterCount } + } + public func snapshot( sessionID: String, agentKind: ChatAgentKind, transcriptPath: String, workingDirectory: String? ) async throws -> Snapshot { - // A build read the file after this request arrived, so it answers this - // request whatever its key. Waiting at most once stops a transcript that - // grows during every build from holding a request forever. + // 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 + ) 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 = buildsBySessionID[sessionID] { + if let build = builds[identity] { if build.key == key || mayJoinAnyBuild { - return try await join(buildID: build.id, sessionID: sessionID) + 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, sessionID: sessionID) + _ = try? await join(buildID: build.id, identity: identity) try Task.checkCancellation() mayJoinAnyBuild = true continue @@ -104,46 +123,46 @@ public actor AgentChatArtifactIndex { ) } let build = InFlightBuild(key: key, task: task) - buildsBySessionID[sessionID] = build + builds[identity] = build startedBuildCount += 1 - return try await join(buildID: build.id, sessionID: sessionID) + 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, sessionID: String) async throws -> Snapshot { - guard var build = buildsBySessionID[sessionID], build.id == buildID else { + private func join(buildID: UUID, identity: BuildIdentity) async throws -> Snapshot { + guard var build = builds[identity], build.id == buildID else { throw CancellationError() } build.waiterCount += 1 - buildsBySessionID[sessionID] = build + 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, sessionID: sessionID) } + Task { await self.leave(buildID: buildID, identity: identity) } } // The first waiter to resume publishes the result for later requests. - if buildsBySessionID[sessionID]?.id == buildID { - buildsBySessionID[sessionID] = nil + if builds[identity]?.id == buildID { + builds[identity] = nil if case .success(let snapshot) = result { - cacheBySessionID.insert(CacheEntry(key: key, snapshot: snapshot), forKey: sessionID) + cacheBySessionID.insert(CacheEntry(key: key, snapshot: snapshot), forKey: identity.sessionID) } } return try result.get() } - private func leave(buildID: UUID, sessionID: String) { - guard var build = buildsBySessionID[sessionID], build.id == buildID else { return } + 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 { - buildsBySessionID[sessionID] = build + builds[identity] = build return } // Later requests start fresh instead of joining a cancelled build. - buildsBySessionID[sessionID] = nil + builds[identity] = nil build.task.cancel() } diff --git a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift index cf60dcccac59..a946c6446c38 100644 --- a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift +++ b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift @@ -276,9 +276,15 @@ struct AgentChatArtifactIndexBuildTests { #expect(await index.startedBuildCount == 1) } + private func waitForRequests(_ count: Int, on index: AgentChatArtifactIndex) async { + while await index.waitingRequestCount < count { + 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: 8 * 1024 * 1024 * 1024) + let url = try makeFile(sparseBytes: 64 * 1024 * 1024 * 1024) defer { try? FileManager.default.removeItem(at: url) } let index = AgentChatArtifactIndex() let path = url.path @@ -292,9 +298,32 @@ struct AgentChatArtifactIndexBuildTests { } let kept = Task { try await request() } let cancelled = Task { try await request() } + await waitForRequests(2, on: index) cancelled.cancel() _ = try await kept.value - _ = try? await cancelled.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) } } From 1ade1154ee79d43b150a28cb5011958fdf93ac9b Mon Sep 17 00:00:00 2001 From: Austin Wang <38676809+austinywang@users.noreply.github.com> Date: Mon, 2 Nov 2026 20:20:14 -0800 Subject: [PATCH 6/8] Charge the tail fragment to the byte budget, honor maxLines 0, keep the waiter count in tests Co-Authored-By: Claude Opus 5.5 --- .../AgentChat/AgentChatArtifactIndex.swift | 11 ++--- .../AgentChat/TranscriptTailReader.swift | 10 +++-- .../TranscriptTailReaderTests.swift | 41 +++++++++++++++++++ 3 files changed, 51 insertions(+), 11 deletions(-) diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift index 206284a9a046..157cbf50f53c 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift @@ -49,14 +49,14 @@ public actor AgentChatArtifactIndex { } /// What must match for two requests to share a build. - private struct BuildIdentity: Sendable, Hashable { + 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. - private struct InFlightBuild: Sendable { + struct InFlightBuild: Sendable { let id = UUID() let key: CacheKey let task: Task @@ -72,15 +72,10 @@ public actor AgentChatArtifactIndex { ) private var cacheBySessionID = ChatArtifactLRUCache(capacity: 8) - private var builds: [BuildIdentity: InFlightBuild] = [:] + private(set) var builds: [BuildIdentity: InFlightBuild] = [:] /// Builds started since creation; concurrent misses must share one. private(set) var startedBuildCount = 0 - /// Requests waiting on any build, so a test can act once they have joined. - var waitingRequestCount: Int { - builds.values.reduce(0) { $0 + $1.waiterCount } - } - public func snapshot( sessionID: String, agentKind: ChatAgentKind, diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift index e4045ff3c5c7..83036a610bc9 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift @@ -56,7 +56,10 @@ final class TranscriptTailReader { /// /// 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 window ends where the next older line would exceed `maxBytes`. + /// 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. @@ -94,10 +97,11 @@ final class TranscriptTailReader { // Newest to oldest, stopping at the first line the budget can't hold. let oldestRetained = completeLines - min(completeLines, capacity) var firstLine = completeLines - var usedBytes = 0 + let fragmentLength = Int(clamping: offset - currentStart) + var usedBytes = fragmentLength > limits.maxRecordBytes ? 0 : fragmentLength var oversizedNewestFirst: [Int] = [] var line = completeLines - 1 - while line >= oldestRetained { + 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) diff --git a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift index a946c6446c38..99629b24c86b 100644 --- a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift +++ b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift @@ -53,6 +53,40 @@ struct TranscriptTailReaderTests { #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("") @@ -365,3 +399,10 @@ struct TranscriptLineAssemblerTests { #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 } + } +} From dc199369cce8f65bec66a268d94156cf8b2c849d Mon Sep 17 00:00:00 2001 From: Austin Wang <38676809+austinywang@users.noreply.github.com> Date: Mon, 2 Nov 2026 20:26:03 -0800 Subject: [PATCH 7/8] Make CacheKey internal so InFlightBuild can hold it; note maxRecordBytes <= maxBytes Co-Authored-By: Claude Opus 5.5 --- .../CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift | 2 +- .../CmuxMobileHost/AgentChat/TranscriptTailReader.swift | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/AgentChatArtifactIndex.swift index 157cbf50f53c..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 diff --git a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift index 83036a610bc9..2c11f391e0e7 100644 --- a/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift +++ b/Packages/macOS/CmuxMobileHost/Sources/CmuxMobileHost/AgentChat/TranscriptTailReader.swift @@ -6,7 +6,8 @@ struct TranscriptTailLimits: Sendable, Equatable { 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. + /// 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 } From 12a7f8e2873e8f28de536be2eede4ab7324fd2c3 Mon Sep 17 00:00:00 2001 From: Austin Wang <38676809+austinywang@users.noreply.github.com> Date: Mon, 2 Nov 2026 20:35:40 -0800 Subject: [PATCH 8/8] Bound the test's waiter poll by a clock deadline Co-Authored-By: Claude Opus 5.5 --- .../CmuxMobileHostTests/TranscriptTailReaderTests.swift | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift index 99629b24c86b..c9584c6c7722 100644 --- a/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift +++ b/Packages/macOS/CmuxMobileHost/Tests/CmuxMobileHostTests/TranscriptTailReaderTests.swift @@ -310,8 +310,16 @@ struct AgentChatArtifactIndexBuildTests { #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() } }