Predicted echo: remote terminals only, withdraw on pasted and sent input

Packages/Shared/CmuxTerminalPrediction/Sources/CmuxTerminalPrediction/PredictedGlyph.swift

@@ -11,6 +11,12 @@ public struct PredictedGlyph: Sendable, Equatable {
         /// The remote echoed it. Keep drawing until the frame carrying the real
         /// character is on screen, otherwise the cell blanks for one frame.
         case confirmed
+        /// The user deleted this character before the remote erased it. Its
+        /// echo may still be on its way or already painted, so the host
+        /// paints the cell's background and nothing else until the erase
+        /// lands; otherwise the deleted character shows again for a round
+        /// trip. `character` is a space.
+        case erased
     }
 
     public let character: Character

Packages/Shared/CmuxTerminalPrediction/Sources/CmuxTerminalPrediction/PredictionOutputInbox.swift

@@ -8,10 +8,14 @@ public import Foundation
 public struct PredictionOutputBatch: Sendable, Equatable {
     public let instant: PredictionInstant
     public let bytes: [UInt8]
+    /// Output was dropped before this point because the surface outran the
+    /// drain; whatever the engine was tracking is unknowable now.
+    public let followsDroppedOutput: Bool
 
-    public init(instant: PredictionInstant, bytes: [UInt8]) {
+    public init(instant: PredictionInstant, bytes: [UInt8], followsDroppedOutput: Bool = false) {
         self.instant = instant
         self.bytes = bytes
+        self.followsDroppedOutput = followsDroppedOutput
     }
 }
 
@@ -21,28 +25,59 @@ public struct PredictionOutputBatch: Sendable, Equatable {
 /// parser. Hopping to the main actor per read would put an agent's output flood
 /// straight onto the UI thread, so deposits accumulate here and only the first
 /// one asks for a drain. Order within a surface is preserved.
+///
+/// Only surfaces the host accepted are buffered. Every terminal's output
+/// reaches the tee, and a local one is never predicted, so its reads are
+/// dropped before any bytes are copied.
 public final class PredictionOutputInbox: @unchecked Sendable {
     private let lock = NSLock()
+    private var acceptedSurfaces: Set<UUID> = []
     private var batches: [UUID: [PredictionOutputBatch]] = [:]
+    private var bufferedBytes: [UUID: Int] = [:]
     private var isDrainScheduled = false
+    /// Buffered bytes per surface before older output is dropped. A remote
+    /// `cat` or a build log can outrun the main actor; prediction only needs
+    /// the newest output, and a drop is reported so the engine resets.
+    private let maximumBufferedBytes: Int
 
-    public init() {}
+    public init(maximumBufferedBytes: Int = 1 << 20) {
+        self.maximumBufferedBytes = maximumBufferedBytes
+    }
 
-    /// Adds a chunk.
+    /// Starts buffering a surface's output.
+    public func accept(surfaceID: UUID) {
+        lock.lock()
+        defer { lock.unlock() }
+        acceptedSurfaces.insert(surfaceID)
+    }
+
+    /// Adds a chunk, when the surface was accepted.
     ///
     /// - Returns: `true` when the caller owns scheduling the drain, which is
     ///   exactly once per drain cycle no matter how many threads deposit.
     public func deposit(
         surfaceID: UUID,
-        bytes: [UInt8],
+        bytes: some Collection<UInt8>,
         at instant: PredictionInstant
     ) -> Bool {
         guard !bytes.isEmpty else { return false }
         lock.lock()
         defer { lock.unlock() }
-        batches[surfaceID, default: []].append(
-            PredictionOutputBatch(instant: instant, bytes: bytes)
-        )
+        guard acceptedSurfaces.contains(surfaceID) else { return false }
+        let buffered = bufferedBytes[surfaceID, default: 0] + bytes.count
+        if buffered > maximumBufferedBytes {
+            batches[surfaceID] = [PredictionOutputBatch(
+                instant: instant,
+                bytes: Array(bytes.suffix(maximumBufferedBytes)),
+                followsDroppedOutput: true
+            )]
+            bufferedBytes[surfaceID] = min(bytes.count, maximumBufferedBytes)
+        } else {
+            batches[surfaceID, default: []].append(
+                PredictionOutputBatch(instant: instant, bytes: Array(bytes))
+            )
+            bufferedBytes[surfaceID] = buffered
+        }
         guard !isDrainScheduled else { return false }
         isDrainScheduled = true
         return true
@@ -54,14 +89,36 @@ public final class PredictionOutputInbox: @unchecked Sendable {
         defer { lock.unlock() }
         let taken = batches
         batches.removeAll(keepingCapacity: true)
+        bufferedBytes.removeAll(keepingCapacity: true)
         isDrainScheduled = false
         return taken
     }
 
-    /// Drops a surface's buffered output without draining the rest.
+    /// Takes one surface's buffered output, leaving the rest and the
+    /// scheduled drain alone.
+    public func drain(surfaceID: UUID) -> [PredictionOutputBatch] {
+        lock.lock()
+        defer { lock.unlock() }
+        bufferedBytes.removeValue(forKey: surfaceID)
+        return batches.removeValue(forKey: surfaceID) ?? []
+    }
+
+    /// Stops buffering a surface and drops what it has buffered, without
+    /// draining the rest.
     public func forget(surfaceID: UUID) {
         lock.lock()
         defer { lock.unlock() }
+        acceptedSurfaces.remove(surfaceID)
         batches.removeValue(forKey: surfaceID)
+        bufferedBytes.removeValue(forKey: surfaceID)
+    }
+
+    /// Stops buffering every surface and drops everything buffered.
+    public func forgetAll() {
+        lock.lock()
+        defer { lock.unlock() }
+        acceptedSurfaces.removeAll()
+        batches.removeAll()
+        bufferedBytes.removeAll()
     }
 }

Packages/Shared/CmuxTerminalPrediction/Sources/CmuxTerminalPrediction/PredictionOverlayLayout.swift

@@ -36,10 +36,19 @@ public struct PredictionOverlayLayout: Sendable, Equatable {
             return column >= 0 && column < columns
         }
         guard let first = onRow.map(\.offset).min(),
-              let last = onRow.map(\.offset).max(),
-              let lastTyped = glyphs.map(\.offset).max() else { return nil }
+              let last = onRow.map(\.offset).max() else { return nil }
 
-        let caret = lastTyped + 1
+        // Typing continues after the last live glyph, or, when every drawn
+        // cell is a deleted one, at the first of those.
+        let live = glyphs.filter { $0.standing != .erased }
+        let caret: Int
+        if let lastTyped = live.map(\.offset).max() {
+            caret = lastTyped + 1
+        } else if let firstErased = glyphs.map(\.offset).min() {
+            caret = firstErased
+        } else {
+            return nil
+        }
         // Left of the cursor with nothing speculative, the caret would mark
         // where typing resumes once a pending erase lands, beside the live
         // cursor ghostty already draws. Only a drawn prediction justifies a

Packages/Shared/CmuxTerminalPrediction/Sources/CmuxTerminalPrediction/TerminalOutputScanner.swift

@@ -1,3 +1,5 @@
+public import Foundation
+
 /// The classification of remote output that prediction actually depends on.
 ///
 /// This is not a terminal emulator. Ghostty stays the only thing that renders
@@ -17,14 +19,33 @@ public enum TerminalOutputSignal: Sendable, Equatable {
     /// Disruptive unless the engine is waiting for the erase of a glyph the
     /// user backspaced over; this is the first half of every common form.
     case cursorLeft
+    /// Moved the cursor left by more than one cell: `CSI n D`, n > 1. Line
+    /// editors do this to rewrite text they already echoed (zsh recolouring
+    /// a word), which only the engine's model of the line can tell apart from
+    /// a redraw.
+    case cursorLeftBy(Int)
     /// Cleared the cell under the cursor without moving it: `CSI K`,
     /// `CSI 0 K`, `CSI P` or `CSI 1 P`. The same caveat as `cursorLeft`.
     case clearAtCursor
-    /// Anything else. Cursor motion, erases, newlines, unknown escapes: the
+    /// Anything else. Cursor motion, erases, newlines, unknown escapes, and
+    /// images (kitty graphics, sixel), which move the cursor past them: the
     /// screen moved in a way we did not predict.
     case disruptive
 }
 
+/// What a chunk did, as far as a surface with nothing in flight cares.
+public struct TerminalOutputSkim: Sendable, Equatable {
+    /// Whether anything but styling and host state went by.
+    public var touchedTheScreen = false
+    /// The last alternate-screen switch, if any.
+    public var alternateScreen: Bool?
+
+    public init(touchedTheScreen: Bool = false, alternateScreen: Bool? = nil) {
+        self.touchedTheScreen = touchedTheScreen
+        self.alternateScreen = alternateScreen
+    }
+}
+
 /// Incremental byte classifier for the PTY output tee.
 ///
 /// libghostty hands cmux output in arbitrary chunks, so escape sequences split
@@ -63,6 +84,19 @@ public struct TerminalOutputScanner: Sendable {
     }
 
     private var state: State = .ground
+    /// The byte after `ESC` that opened the current string sequence, and
+    /// whether its header is still being read. A kitty graphics command
+    /// (`ESC _ G`) or a sixel image (`ESC P ... q`) places an image and moves
+    /// the cursor past it, so it ends as `.disruptive`, not `.ignorable`.
+    private var stringIntroducer: UInt8 = 0
+    private var isReadingStringHeader = false
+    private var stringIsImage = false
+    /// Between a cursor save (`ESC 7`, `CSI s`) and its restore, whatever is
+    /// drawn happens somewhere else: a status line or clock repainting. The
+    /// restore puts the cursor back, so none of it is disruptive. Bounded, in
+    /// case a save is never restored.
+    private var detourBytes: Int?
+    private static let maximumDetourBytes = 4_096
     /// Parameter and intermediate bytes of the current CSI sequence, up to
     /// `maximumParameterBytes`. The remote controls how long a sequence is, and
     /// classification only ever compares against short mode numbers, so bytes
@@ -77,13 +111,82 @@ public struct TerminalOutputScanner: Sendable {
     public mutating func scan(_ bytes: some Sequence<UInt8>) -> [TerminalOutputSignal] {
         var signals: [TerminalOutputSignal] = []
         for byte in bytes {
-            if let signal = consume(byte) {
+            if let signal = consumeOutsideDetour(byte) {
                 signals.append(signal)
             }
         }
         return signals
     }
 
+    /// Classify one byte: the signal it completes, if any.
+    public mutating func next(_ byte: UInt8) -> TerminalOutputSignal? {
+        consumeOutsideDetour(byte)
+    }
+
+    /// Reads a chunk only for what a surface with nothing in flight needs:
+    /// whether it touched the screen, and alternate-screen switches. Keeps
+    /// the parse state exactly as `scan` would, but jumps over plain text.
+    public mutating func skim(_ bytes: UnsafeBufferPointer<UInt8>) -> TerminalOutputSkim {
+        var skim = TerminalOutputSkim()
+        guard let base = bytes.baseAddress else { return skim }
+        var index = 0
+        let count = bytes.count
+        while index < count {
+            if state == .ground, detourBytes == nil, skim.touchedTheScreen {
+                // Text, controls and newlines only touch the screen, which is
+                // already known; only an escape can switch screens.
+                guard let found = memchr(base + index, 0x1B, count - index) else { break }
+                index = base.distance(to: found.assumingMemoryBound(to: UInt8.self))
+            }
+            switch consumeOutsideDetour(base[index]) {
+            case nil, .ignorable?:
+                break
+            case .alternateScreen(let entered)?:
+                skim.alternateScreen = entered
+                skim.touchedTheScreen = true
+            case _?:
+                skim.touchedTheScreen = true
+            }
+            index += 1
+        }
+        return skim
+    }
+
+    private mutating func consumeOutsideDetour(_ byte: UInt8) -> TerminalOutputSignal? {
+        let signal = consume(byte)
+        guard let bytes = detourBytes else { return signal }
+        if case .alternateScreen? = signal { return signal }
+        if bytes >= Self.maximumDetourBytes {
+            detourBytes = nil
+            return .disruptive
+        }
+        detourBytes = bytes + 1
+        return nil
+    }
+
+    private mutating func openString(_ introducer: UInt8) {
+        stringIntroducer = introducer
+        isReadingStringHeader = introducer == UInt8(ascii: "_") || introducer == UInt8(ascii: "P")
+        stringIsImage = false
+    }
+
+    private mutating func readStringHeader(_ byte: UInt8) {
+        switch stringIntroducer {
+        case UInt8(ascii: "_"):
+            stringIsImage = byte == UInt8(ascii: "G")
+            isReadingStringHeader = false
+        default:
+            // DCS parameters, then its final byte: `q` is sixel.
+            if (0x30...0x3B).contains(byte) { return }
+            stringIsImage = byte == UInt8(ascii: "q")
+            isReadingStringHeader = false
+        }
+    }
+
+    private var endOfString: TerminalOutputSignal {
+        stringIsImage ? .disruptive : .ignorable
+    }
+
     private mutating func consume(_ byte: UInt8) -> TerminalOutputSignal? {
         switch state {
         case .ground:
@@ -108,10 +211,20 @@ public struct TerminalOutputScanner: Sendable {
                 return nil
             case UInt8(ascii: "]"):
                 state = .controlString(.operatingSystemCommand)
+                openString(byte)
                 return nil
             case UInt8(ascii: "P"), UInt8(ascii: "X"), UInt8(ascii: "^"), UInt8(ascii: "_"):
                 state = .controlString(.other)
+                openString(byte)
                 return nil
+            case UInt8(ascii: "7"):
+                state = .ground
+                detourBytes = 0
+                return nil
+            case UInt8(ascii: "8"):
+                state = .ground
+                detourBytes = nil
+                return .ignorable
             case 0x1B:
                 // Another ESC restarts the escape, as it does in ghostty.
                 return nil
@@ -156,6 +269,14 @@ public struct TerminalOutputScanner: Sendable {
             }
             state = .ground
             guard (0x40...0x7E).contains(byte) else { return .disruptive }
+            if parameters.isEmpty, byte == UInt8(ascii: "s") {
+                detourBytes = 0
+                return nil
+            }
+            if parameters.isEmpty, byte == UInt8(ascii: "u") {
+                detourBytes = nil
+                return .ignorable
+            }
             return Self.classifyControlSequence(
                 parameters: parameters,
                 overflowed: parametersOverflowed,
@@ -166,7 +287,7 @@ public struct TerminalOutputScanner: Sendable {
             switch byte {
             case 0x07 where kind == .operatingSystemCommand:
                 state = .ground
-                return .ignorable
+                return endOfString
             case 0x18, 0x1A:
                 // CAN and SUB abort the sequence and are executed as controls.
                 state = .ground
@@ -177,19 +298,26 @@ public struct TerminalOutputScanner: Sendable {
             case 0x9C where kind == .other:
                 // The 8-bit string terminator.
                 state = .ground
-                return .ignorable
+                return endOfString
             case 0x80...0x9F where kind == .other:
                 // Ghostty leaves a DCS, APC, PM or SOS at any other C1 byte.
                 state = .ground
                 return .disruptive
             default:
+                if isReadingStringHeader { readStringHeader(byte) }
                 return nil
             }
 
         case .controlStringEscape:
             if byte == UInt8(ascii: "\\") {
                 state = .ground
-                return .ignorable
+                return endOfString
+            }
+            if stringIsImage {
+                // Left for a new sequence, the image was still placed.
+                state = .escape
+                _ = consume(byte)
+                return .disruptive
             }
             // Ghostty leaves the string at any ESC, not only at ST, and
             // parses the next byte as the start of a new sequence. Staying in
@@ -217,6 +345,10 @@ public struct TerminalOutputScanner: Sendable {
         switch final {
         case UInt8(ascii: "D") where isDefaultOrOne:
             return .cursorLeft
+        case UInt8(ascii: "D") where !overflowed && parameters.count <= 3
+            && parameters.allSatisfy({ (0x30...0x39).contains($0) }):
+            let count = parameters.reduce(0) { $0 * 10 + Int($1 - 0x30) }
+            return count <= 1 ? .cursorLeft : .cursorLeftBy(count)
         case UInt8(ascii: "P") where isDefaultOrOne, UInt8(ascii: "K") where isDefaultOrZero:
             return .clearAtCursor
         default: