Recover iOS terminal render pipeline stalls

.github/swift-file-length-budget.tsv

@@ -12,7 +12,7 @@
 9497	cmuxTests/CLINotifyProcessIntegrationRegressionTests.swift
 8032	Sources/Panels/BrowserPanelView.swift
 8016	CLI/cmux_open.swift
-7371	Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite.swift
+7783	Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite.swift
 7366	cmuxTests/WorkspaceUnitTests.swift
 7218	cmuxTests/WorkspaceRemoteConnectionTests.swift
 6359	cmuxTests/SessionPersistenceTests.swift
@@ -24,11 +24,11 @@
 4487	Sources/Panels/FilePreviewPanel.swift
 4483	Sources/cmuxApp.swift
 4367	cmuxTests/BrowserPanelTests.swift
+4264	Packages/iOS/CmuxMobileTerminal/Sources/CmuxMobileTerminal/GhosttySurfaceView.swift
 4121	Sources/BrowserWindowPortal.swift
 3934	Sources/Feed/FeedPanelView.swift
 3926	cmuxTests/TabManagerUnitTests.swift
 3896	cmuxTests/WindowAndDragTests.swift
-3747	Packages/iOS/CmuxMobileTerminal/Sources/CmuxMobileTerminal/GhosttySurfaceView.swift
 3668	cmuxTests/CLIGenericHookPersistenceTests.swift
 3397	Sources/CmuxConfig.swift
 3364	cmuxTests/TabManagerSessionSnapshotTests.swift
@@ -112,6 +112,7 @@
 859	Packages/macOS/CmuxControlSocket/Sources/CmuxControlSocket/Coordinator/Workspace/ControlCommandCoordinator+Workspace.swift
 847	cmuxTests/AgentSessionAutoResumeSettingsTests.swift
 845	cmuxTests/SSHStartupSignalLifecycleTests.swift
+841	Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/TerminalOutputDeliveryQueueTests.swift
 834	Sources/MainWindowFocusController.swift
 830	Sources/TaskManagerTypes.swift
 825	Packages/iOS/CmuxMobileShellUI/Sources/CmuxMobileShellUI/TerminalComposerView.swift
@@ -130,7 +131,7 @@
 760	Packages/macOS/CMUXAgentLaunch/Tests/CMUXAgentLaunchTests/AgentLaunchSanitizerTests.swift
 756	Sources/Panels/AgentSessionWebRendererCoordinator.swift
 754	Sources/TerminalController+ControlWorkspaceContext.swift
-753	Packages/iOS/CmuxMobileShellUI/Sources/CmuxMobileShellUI/WorkspaceDetailView.swift
+752	Packages/iOS/CmuxMobileShellUI/Sources/CmuxMobileShellUI/WorkspaceDetailView.swift
 752	cmuxUITests/CloseWorkspaceCmdDUITests.swift
 739	cmuxTests/CLICodexHookTimeoutRegressionTests.swift
 738	Packages/macOS/CMUXProjectModel/Sources/CMUXProjectModel/XcodeProjectAdapter.swift
@@ -172,6 +173,7 @@
 620	cmuxTests/FinderFileDropRegressionTests.swift
 620	cmuxTests/TerminalNotificationQueueTests.swift
 615	Sources/SettingsNavigation.swift
+613	Packages/iOS/CmuxMobileShell/Tests/CmuxMobileShellTests/MobileShellRenderGridLivenessTestSupport.swift
 608	cmuxUITests/FeedSidebarUITests.swift
 607	Packages/macOS/CmuxTerminal/Sources/CmuxTerminal/Surface/TerminalSurface.swift
 607	Sources/SessionIndexModels.swift

Packages/iOS/CmuxMobileShell/Sources/CmuxMobileShell/MobileShellComposite+TerminalOutputDelivery.swift

@@ -1,33 +1,59 @@
 import CMUXMobileCore
+internal import CmuxMobileDiagnostics
 import CmuxMobileShellModel
 public import Foundation
 
 extension MobileShellComposite {
+    func claimTerminalReplayBarrierFollowUp(surfaceID: String) -> Bool {
+        let followUpCount = terminalReplayBarrierFollowUpCountsBySurfaceID[surfaceID] ?? 0
+        guard followUpCount < Self.maxTerminalReplayBarrierFollowUps else {
+            MobileDebugLog.anchormux(
+                "terminal.output.replay_followup_cap_reached surface=\(surfaceID) attempts=\(followUpCount)"
+            )
+            terminalReplayBarrierFollowUpCountsBySurfaceID.removeValue(forKey: surfaceID)
+            return false
+        }
+        terminalReplayBarrierFollowUpCountsBySurfaceID[surfaceID] = followUpCount + 1
+        return true
+    }
+
     /// Yield a raw PTY byte chunk to the surface stream, if one is attached.
-    func deliverTerminalBytes(_ bytes: Data, surfaceID: String) {
-        deliverTerminalOutput(
+    @discardableResult
+    func deliverTerminalBytes(
+        _ bytes: Data,
+        surfaceID: String,
+        bypassReplayBarrier: Bool = false
+    ) -> Bool {
+        return deliverTerminalOutput(
             TerminalOutputDelivery(
                 bytes: bytes,
                 replaceable: false,
                 viewportPolicy: .natural
             ),
-            surfaceID: surfaceID
+            surfaceID: surfaceID,
+            bypassReplayBarrier: bypassReplayBarrier
         )
     }
 
-    func deliverTerminalRenderGrid(_ frame: MobileTerminalRenderGridFrame, surfaceID: String) {
-        deliverTerminalOutput(
+    @discardableResult
+    func deliverTerminalRenderGrid(
+        _ frame: MobileTerminalRenderGridFrame,
+        surfaceID: String,
+        bypassReplayBarrier: Bool = false
+    ) -> Bool {
+        return deliverTerminalOutput(
             TerminalOutputDelivery(
                 renderGrid: frame,
                 replaceable: frame.isReplaceableViewportPatchForMobileDelivery,
                 viewportPolicy: frame.mobileViewportPolicy
             ),
-            surfaceID: surfaceID
+            surfaceID: surfaceID,
+            bypassReplayBarrier: bypassReplayBarrier
         )
     }
 
     func deliverTerminalViewportPolicy(_ policy: MobileTerminalOutputViewportPolicy, surfaceID: String) {
-        deliverTerminalOutput(
+        _ = deliverTerminalOutput(
             TerminalOutputDelivery(
                 bytes: Data(),
                 replaceable: true,
@@ -38,12 +64,50 @@ extension MobileShellComposite {
         )
     }
 
-    private func deliverTerminalOutput(_ delivery: TerminalOutputDelivery, surfaceID: String) {
+    private func deliverTerminalOutput(
+        _ delivery: TerminalOutputDelivery,
+        surfaceID: String,
+        bypassReplayBarrier: Bool = false
+    ) -> Bool {
         guard let continuation = terminalByteContinuationsBySurfaceID[surfaceID],
-              let streamToken = terminalOutputStreamTokensBySurfaceID[surfaceID] else { return }
+              let streamToken = terminalOutputStreamTokensBySurfaceID[surfaceID] else { return false }
+        if let replayBarrierToken = terminalReplayBarrierTokensBySurfaceID[surfaceID],
+           !bypassReplayBarrier {
+            terminalReplayBarrierDroppedOutputSurfaceIDs.insert(surfaceID)
+            let droppedOutputCount = (terminalReplayBarrierDroppedOutputCountsBySurfaceID[surfaceID] ?? 0) &+ 1
+            terminalReplayBarrierDroppedOutputCountsBySurfaceID[surfaceID] = droppedOutputCount
+            if droppedOutputCount == 1 || droppedOutputCount.isMultiple(of: 32) {
+                MobileDebugLog.anchormux(
+                    "terminal.output.drop_replay_barrier surface=\(surfaceID) count=\(droppedOutputCount)"
+                )
+            }
+            if remoteClient != nil,
+               terminalReplayBarrierAckStreamTokensBySurfaceID[surfaceID] == nil,
+               !terminalReplaySurfaceIDsInFlight.contains(surfaceID),
+               !terminalReplayFailureRetryExhausted(surfaceID: surfaceID) {
+                MobileDebugLog.anchormux("terminal.output.replay_retry_after_drop surface=\(surfaceID)")
+                requestTerminalReplay(
+                    surfaceID: surfaceID,
+                    replayBarrierToken: replayBarrierToken,
+                    coveredReplayBarrierDroppedOutputCount: droppedOutputCount
+                )
+            }
+            return false
+        }
         var queue = terminalOutputQueuesBySurfaceID[surfaceID] ?? TerminalOutputDeliveryQueue()
         let immediate = queue.enqueue(delivery)
+        let pendingCount = queue.pendingCount
         terminalOutputQueuesBySurfaceID[surfaceID] = queue
+        if bypassReplayBarrier,
+           immediate != nil,
+           terminalReplayBarrierTokensBySurfaceID[surfaceID] != nil {
+            terminalReplayBarrierAckStreamTokensBySurfaceID[surfaceID] = streamToken
+        }
+        if pendingCount >= 32, pendingCount.isMultiple(of: 32) {
+            MobileDebugLog.anchormux(
+                "terminal.output.pending surface=\(surfaceID) depth=\(pendingCount)"
+            )
+        }
         if let immediate {
             continuation.yield(
                 MobileTerminalOutputChunk(
@@ -53,6 +117,7 @@ extension MobileShellComposite {
                 )
             )
         }
+        return true
     }
 
     /// Mark the current yielded terminal-output chunk as applied by the iOS surface.
@@ -61,6 +126,31 @@ extension MobileShellComposite {
               var queue = terminalOutputQueuesBySurfaceID[surfaceID] else { return }
         let next = queue.completeInFlight()
         terminalOutputQueuesBySurfaceID[surfaceID] = queue
+        if terminalReplayBarrierAckStreamTokensBySurfaceID[surfaceID] == streamToken {
+            let coveredDroppedOutputCount =
+                terminalReplayBarrierAckCoveredDroppedOutputCountsBySurfaceID.removeValue(forKey: surfaceID)
+            let currentDroppedOutputCount = terminalReplayBarrierDroppedOutputCountsBySurfaceID[surfaceID] ?? 0
+            let needsFollowUpReplay = coveredDroppedOutputCount.map {
+                currentDroppedOutputCount > $0
+            } ?? true
+            terminalReplayBarrierAckStreamTokensBySurfaceID.removeValue(forKey: surfaceID)
+            terminalReplayBarrierTokensBySurfaceID.removeValue(forKey: surfaceID)
+            MobileDebugLog.anchormux("terminal.output.replay_barrier_cleared surface=\(surfaceID)")
+            let droppedOutputDuringBarrier = terminalReplayBarrierDroppedOutputSurfaceIDs.remove(surfaceID) != nil
+            terminalReplayBarrierDroppedOutputCountsBySurfaceID.removeValue(forKey: surfaceID)
+            if droppedOutputDuringBarrier,
+               needsFollowUpReplay,
+               claimTerminalReplayBarrierFollowUp(surfaceID: surfaceID) {
+                let replayBarrierToken = beginTerminalReplayBarrier(
+                    surfaceID: surfaceID,
+                    preservingFollowUpCount: true
+                )
+                MobileDebugLog.anchormux("terminal.output.replay_followup surface=\(surfaceID)")
+                requestTerminalReplay(surfaceID: surfaceID, replayBarrierToken: replayBarrierToken)
+                return
+            }
+            terminalReplayBarrierFollowUpCountsBySurfaceID.removeValue(forKey: surfaceID)
+        }
         guard let next,
               let continuation = terminalByteContinuationsBySurfaceID[surfaceID],
               terminalOutputStreamTokensBySurfaceID[surfaceID] == streamToken else {
@@ -72,4 +162,72 @@ extension MobileShellComposite {
             viewportPolicy: next.viewportPolicy
         ))
     }
+
+    /// Abandon the current yielded terminal-output chunk after the local render
+    /// surface reset. The abandoned bytes may have been applied to the old
+    /// Ghostty surface or may still be behind a wedged worker queue, so continuing
+    /// to drain the old pending queue would replay stale deltas into the rebuilt
+    /// surface. Reset the queue, invalidate stale acks, then request a fresh
+    /// authoritative replay from the Mac.
+    public func terminalOutputDidReset(surfaceID: String, streamToken: UUID) {
+        guard terminalOutputStreamTokensBySurfaceID[surfaceID] == streamToken,
+              terminalOutputQueuesBySurfaceID[surfaceID] != nil else { return }
+        if let replayBarrierToken = terminalReplayBarrierTokensBySurfaceID[surfaceID] {
+            guard terminalReplayBarrierAckStreamTokensBySurfaceID[surfaceID] == streamToken else {
+                MobileDebugLog.anchormux("terminal.output.reset_barrier_active surface=\(surfaceID)")
+                return
+            }
+            retryTerminalReplayAfterAckReset(
+                surfaceID: surfaceID,
+                replayBarrierToken: replayBarrierToken
+            )
+            return
+        }
+        let replayBarrierToken = beginTerminalReplayBarrier(surfaceID: surfaceID)
+        MobileDebugLog.anchormux("terminal.output.reset surface=\(surfaceID)")
+        requestTerminalReplay(surfaceID: surfaceID, replayBarrierToken: replayBarrierToken)
+    }
+
+    private func retryTerminalReplayAfterAckReset(
+        surfaceID: String,
+        replayBarrierToken: UUID
+    ) {
+        guard terminalReplayBarrierTokensBySurfaceID[surfaceID] == replayBarrierToken else {
+            return
+        }
+        terminalOutputQueuesBySurfaceID[surfaceID] = TerminalOutputDeliveryQueue()
+        terminalOutputStreamTokensBySurfaceID[surfaceID] = UUID()
+        deliveredTerminalByteEndSeqBySurfaceID.removeValue(forKey: surfaceID)
+        pendingTerminalByteEndSeqBySurfaceID.removeValue(forKey: surfaceID)
+        terminalReplayBarrierAckStreamTokensBySurfaceID.removeValue(forKey: surfaceID)
+        terminalReplayBarrierAckCoveredDroppedOutputCountsBySurfaceID.removeValue(forKey: surfaceID)
+        terminalReplayBarrierTokensInFlightBySurfaceID.removeValue(forKey: surfaceID)
+        guard let retryToken = prepareTerminalReplayFailureRetry(
+            surfaceID: surfaceID,
+            replayBarrierToken: replayBarrierToken
+        ) else {
+            preserveTerminalReplayBarrierIfCurrent(
+                surfaceID: surfaceID,
+                token: replayBarrierToken,
+                reason: "reset_replay_ack"
+            )
+            return
+        }
+        MobileDebugLog.anchormux("terminal.output.reset_replay_ack surface=\(surfaceID)")
+        requestTerminalReplay(
+            surfaceID: surfaceID,
+            replayBarrierToken: retryToken,
+            coveredReplayBarrierDroppedOutputCount:
+                terminalReplayBarrierDroppedOutputCountsBySurfaceID[surfaceID]
+        )
+    }
+
+    /// Ask the Mac to replay the authoritative terminal state for a surface.
+    public func terminalOutputNeedsReplay(surfaceID: String) {
+        guard terminalByteContinuationsBySurfaceID[surfaceID] != nil else { return }
+        let replayBarrierToken = beginTerminalReplayBarrier(surfaceID: surfaceID)
+        MobileDebugLog.anchormux("terminal.output.replay_requested surface=\(surfaceID)")
+        requestTerminalReplay(surfaceID: surfaceID, replayBarrierToken: replayBarrierToken)
+    }
+
 }