Bound the first transcript read in bytes and make it cancellable

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.

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)
+    }
+}

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),
             ]

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,26 +48,117 @@ 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<Snapshot, any Error>
+        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<String, CacheEntry>(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,
         agentKind: ChatAgentKind,
         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,