Add per-PTY dispatch sources and TaskNotifier bypass

ModernTests/FairnessScheduler/Mocks/EnqueuingPTYTaskDelegate.swift

@@ -0,0 +1,144 @@
+//
+//  EnqueuingPTYTaskDelegate.swift
+//  ModernTests
+//
+//  Specialized PTYTaskDelegate for testing the read pipeline that enqueues
+//  read data to a TokenExecutor instead of just tracking calls.
+//  This enables end-to-end pipeline testing from read → parse → queue.
+//
+
+import Foundation
+@testable import iTerm2SharedARC
+
+/// A PTYTaskDelegate that actually enqueues read data to a TokenExecutor.
+/// Used for testing the full read handler pipeline.
+final class EnqueuingPTYTaskDelegate: NSObject, PTYTaskDelegate {
+
+    // MARK: - Configuration
+
+    /// The TokenExecutor to enqueue tokens to.
+    var tokenExecutor: TokenExecutor?
+
+    /// Callback invoked when data is read (before enqueueing).
+    var onThreadedRead: ((Data) -> Void)?
+
+    /// Callback invoked after tokens are enqueued.
+    var onEnqueued: (() -> Void)?
+
+    // MARK: - Call Tracking
+
+    private let lock = NSLock()
+    private var _readCount = 0
+    private var _totalBytesRead = 0
+    private var _enqueueCount = 0
+
+    var readCount: Int {
+        lock.lock()
+        defer { lock.unlock() }
+        return _readCount
+    }
+
+    var totalBytesRead: Int {
+        lock.lock()
+        defer { lock.unlock() }
+        return _totalBytesRead
+    }
+
+    var enqueueCount: Int {
+        lock.lock()
+        defer { lock.unlock() }
+        return _enqueueCount
+    }
+
+    // MARK: - Initialization
+
+    override init() {
+        super.init()
+    }
+
+    /// Convenience initializer for tests that need an executor set immediately.
+    convenience init(executor: TokenExecutor) {
+        self.init()
+        self.tokenExecutor = executor
+    }
+
+    // MARK: - PTYTaskDelegate
+
+    func threadedReadTask(_ buffer: UnsafeMutablePointer<CChar>, length: Int32) {
+        lock.lock()
+        _readCount += 1
+        _totalBytesRead += Int(length)
+        let data = Data(bytes: buffer, count: Int(length))
+        lock.unlock()
+
+        onThreadedRead?(data)
+
+        // Enqueue tokens to the executor (mimics PTYSession behavior)
+        if let executor = tokenExecutor {
+            // Add enough tokens to trigger backpressure change for verification
+            executor.addMultipleTokenArrays(count: 100, tokensPerArray: 5)
+
+            lock.lock()
+            _enqueueCount += 1
+            lock.unlock()
+
+            onEnqueued?()
+        }
+    }
+
+    func threadedTaskBrokenPipe() {
+        // Not used in pipeline tests
+    }
+
+    func brokenPipe() {
+        // Not used in pipeline tests
+    }
+
+    func tmuxClientWrite(_ data: Data) {
+        // Not used in pipeline tests
+    }
+
+    func taskDiedImmediately() {
+        // Not used in pipeline tests
+    }
+
+    func taskDiedWithError(_ error: String!) {
+        // Not used in pipeline tests
+    }
+
+    func taskDidChangeTTY(_ task: PTYTask) {
+        // Not used in pipeline tests
+    }
+
+    func taskDidRegister(_ task: PTYTask) {
+        // Not used in pipeline tests
+    }
+
+    func taskDidChangePaused(_ task: PTYTask, paused: Bool) {
+        // Not used in pipeline tests
+    }
+
+    func taskMuteCoprocessDidChange(_ task: PTYTask, hasMuteCoprocess: Bool) {
+        // Not used in pipeline tests
+    }
+
+    func taskDidResize(to gridSize: VT100GridSize, pixelSize: NSSize) {
+        // Not used in pipeline tests
+    }
+
+    func taskDidReadFromCoprocessWhileSSHIntegration(inUse data: Data) {
+        // Not used in pipeline tests
+    }
+
+    // MARK: - Test Helpers
+
+    func reset() {
+        lock.lock()
+        _readCount = 0
+        _totalBytesRead = 0
+        _enqueueCount = 0
+        onThreadedRead = nil
+        onEnqueued = nil
+        lock.unlock()
+    }
+}

ModernTests/FairnessScheduler/Mocks/MockCoprocess.swift

@@ -0,0 +1,82 @@
+//
+//  MockCoprocess.swift
+//  ModernTests
+//
+//  Mock implementation of Coprocess for testing TaskNotifier coprocess handling.
+//  Subclass of Coprocess that uses pipes instead of spawning a subprocess.
+//
+
+import Foundation
+@testable import iTerm2SharedARC
+
+class MockCoprocess: Coprocess {
+
+    /// The external write FD - test code writes here to simulate coprocess output.
+    /// This is the write end of the read pipe (TaskNotifier reads from inputFd/readFileDescriptor).
+    private(set) var testWriteFd: Int32 = -1
+
+    /// The external read FD - test code reads here to see data written to coprocess.
+    /// This is the read end of the write pipe (TaskNotifier writes to outputFd/writeFileDescriptor).
+    private(set) var testReadFd: Int32 = -1
+
+    /// Create a MockCoprocess with pipe FDs.
+    /// Returns nil on failure (pipe creation failed).
+    @objc static func createPipe() -> MockCoprocess? {
+        var readPipe: [Int32] = [0, 0]
+        var writePipe: [Int32] = [0, 0]
+
+        guard pipe(&readPipe) == 0 else { return nil }
+        guard pipe(&writePipe) == 0 else {
+            close(readPipe[0])
+            close(readPipe[1])
+            return nil
+        }
+
+        // Set non-blocking on inputFd
+        var flags = fcntl(readPipe[0], F_GETFL)
+        fcntl(readPipe[0], F_SETFL, flags | O_NONBLOCK)
+
+        // Set non-blocking on outputFd
+        flags = fcntl(writePipe[1], F_GETFL)
+        fcntl(writePipe[1], F_SETFL, flags | O_NONBLOCK)
+
+        let coprocess = MockCoprocess()
+        coprocess.inputFd = readPipe[0]
+        coprocess.outputFd = writePipe[1]
+        coprocess.pid = getpid()
+        coprocess.testWriteFd = readPipe[1]
+        coprocess.testReadFd = writePipe[0]
+
+        return coprocess
+    }
+
+    deinit {
+        closeTestFds()
+    }
+
+    // Override to NOT send kill signal - MockCoprocess uses getpid() as pid
+    // and we don't want to kill the test process!
+    override func terminate() {
+        if outputFd >= 0 {
+            close(outputFd)
+            outputFd = -1
+        }
+        if inputFd >= 0 {
+            close(inputFd)
+            inputFd = -1
+        }
+        pid = -1
+    }
+
+    /// Close the test FDs (call in addition to terminate to clean up test resources).
+    @objc func closeTestFds() {
+        if testWriteFd >= 0 {
+            close(testWriteFd)
+            testWriteFd = -1
+        }
+        if testReadFd >= 0 {
+            close(testReadFd)
+            testReadFd = -1
+        }
+    }
+}

ModernTests/FairnessScheduler/Mocks/MockFairnessSchedulerExecutor.swift

@@ -0,0 +1,77 @@
+//
+//  MockFairnessSchedulerExecutor.swift
+//  ModernTests
+//
+//  Mock executor for testing FairnessScheduler in isolation.
+//  This simulates the TokenExecutor interface that FairnessScheduler will use.
+//
+
+import Foundation
+@testable import iTerm2SharedARC
+
+// TurnResult and FairnessSchedulerExecutor are defined in FairnessScheduler.swift
+
+/// Mock implementation for testing FairnessScheduler without real TokenExecutor.
+final class MockFairnessSchedulerExecutor: FairnessSchedulerExecutor {
+
+    // MARK: - Configuration
+
+    /// The result to return from executeTurn. Set this before each test scenario.
+    var turnResult: TurnResult = .completed
+
+    /// If set, executeTurn calls this instead of using turnResult.
+    /// Useful for dynamic behavior or async testing.
+    var executeTurnHandler: ((Int, @escaping (TurnResult) -> Void) -> Void)?
+
+    /// Delay before calling completion (simulates execution time)
+    var executionDelay: TimeInterval = 0
+
+    // MARK: - Call Tracking
+
+    struct ExecuteTurnCall: Equatable {
+        let tokenBudget: Int
+        let timestamp: Date
+
+        static func == (lhs: ExecuteTurnCall, rhs: ExecuteTurnCall) -> Bool {
+            return lhs.tokenBudget == rhs.tokenBudget
+        }
+    }
+
+    private(set) var executeTurnCalls: [ExecuteTurnCall] = []
+    private(set) var totalTokenBudgetConsumed: Int = 0
+
+    // MARK: - FairnessSchedulerExecutor
+
+    func executeTurn(tokenBudget: Int, completion: @escaping (TurnResult) -> Void) {
+        executeTurnCalls.append(ExecuteTurnCall(tokenBudget: tokenBudget, timestamp: Date()))
+        totalTokenBudgetConsumed += tokenBudget
+
+        if let handler = executeTurnHandler {
+            handler(tokenBudget, completion)
+            return
+        }
+
+        if executionDelay > 0 {
+            DispatchQueue.global().asyncAfter(deadline: .now() + executionDelay) { [turnResult] in
+                completion(turnResult)
+            }
+        } else {
+            completion(turnResult)
+        }
+    }
+
+    // MARK: - Test Helpers
+
+    func reset() {
+        turnResult = .completed
+        executeTurnHandler = nil
+        executionDelay = 0
+        executeTurnCalls = []
+        totalTokenBudgetConsumed = 0
+    }
+
+    /// Returns the number of times executeTurn was called
+    var executeTurnCallCount: Int {
+        return executeTurnCalls.count
+    }
+}