fix(computer): reap mac helper after client loss (#11493)

* perf(computer): add mac helper owner-loss benchmark Measure the release helper's resident memory before and after its owner-session deadline. Record exact revisions, per-trial RSS, retained state, and clean-exit latency so lifecycle reclamation is reproducible. * fix(computer): reap mac helper after client loss Bind the detached macOS helper lifetime to authenticated socket ownership. Reap the helper after its final authenticated client disconnects, and add a startup deadline for sessions that never authenticate. * test(computer): harden owner benchmark cleanup * test(computer): make owner benchmark cleanup failure-safe * test(computer): close remaining owner cleanup races
2026-09-22 08:02:28 +00:00 · 2026-07-30 19:24:34 -07:00
parent 1b4d2f30ea
commit 14de3fa14d
14 changed files with 2587 additions and 5 deletions
@@ -0,0 +1,50 @@
+public struct AgentSessionConnectionID: Hashable, Sendable {
+    public let rawValue: UInt64
+
+    public init(rawValue: UInt64) {
+        self.rawValue = rawValue
+    }
+}
+
+public enum AgentSessionRegistration: Sendable {
+    case rejected
+    case claimed
+    case retained
+}
+
+public struct AgentSessionOwnership: Sendable {
+    private var authenticatedConnections: Set<AgentSessionConnectionID> = []
+    private var wasClaimed = false
+    private var sessionClosed = false
+
+    public init() {}
+
+    public mutating func registerConnection(
+        _ connection: AgentSessionConnectionID,
+        authenticated: Bool
+    ) -> AgentSessionRegistration {
+        guard authenticated, !sessionClosed else { return .rejected }
+        let inserted = authenticatedConnections.insert(connection).inserted
+        guard inserted else { return .rejected }
+        guard !wasClaimed else { return .retained }
+        wasClaimed = true
+        return .claimed
+    }
+
+    public mutating func disconnect(_ connection: AgentSessionConnectionID) -> Bool {
+        guard authenticatedConnections.remove(connection) != nil else { return false }
+        guard wasClaimed, authenticatedConnections.isEmpty else { return false }
+        sessionClosed = true
+        return true
+    }
+}
+
+public func isAuthenticatedAgentSession(
+    expectedToken: String?,
+    requestToken: String?,
+    authorizedPeer: Bool
+) -> Bool {
+    guard authorizedPeer else { return false }
+    guard let expectedToken else { return true }
+    return requestToken == expectedToken
+}
@@ -0,0 +1,182 @@
+import Darwin
+import Dispatch
+import Foundation
+
+public final class AuthenticatedConnectionHangupMonitor: @unchecked Sendable {
+    typealias RegisterEvents = @Sendable (
+        Int32,
+        UnsafePointer<kevent>?,
+        Int32
+    ) -> POSIXErrorCode?
+    typealias CloseDescriptor = @Sendable (Int32) -> Void
+
+    private static let cancelEventIdentifier: UInt = 1
+
+    private let closeDescriptor: CloseDescriptor
+    private let eventQueue: DispatchQueue
+    private let eventQueueDescriptor: Int32
+    private let onEvent: @Sendable () -> Void
+    private let stateLock = NSLock()
+    private let onHangup: @Sendable () -> Void
+    private var isCancelled = false
+    private var didReportHangup = false
+    private var isFinished = false
+    private var isStarted = false
+
+    public convenience init(
+        fileDescriptor: Int32,
+        queue: DispatchQueue = DispatchQueue(
+            label: "com.stablyai.orca.computer-use-owner-hangup"
+        ),
+        onEvent: @escaping @Sendable () -> Void = {},
+        onHangup: @escaping @Sendable () -> Void
+    ) throws {
+        try self.init(
+            fileDescriptor: fileDescriptor,
+            queue: queue,
+            registerEvents: { descriptor, events, count in
+                guard kevent(descriptor, events, count, nil, 0, nil) == 0 else {
+                    return POSIXErrorCode(rawValue: errno) ?? .EIO
+                }
+                return nil
+            },
+            closeDescriptor: { descriptor in
+                close(descriptor)
+            },
+            onEvent: onEvent,
+            onHangup: onHangup
+        )
+    }
+
+    init(
+        fileDescriptor: Int32,
+        queue: DispatchQueue = DispatchQueue(
+            label: "com.stablyai.orca.computer-use-owner-hangup"
+        ),
+        registerEvents: RegisterEvents,
+        closeDescriptor: @escaping CloseDescriptor,
+        onEvent: @escaping @Sendable () -> Void = {},
+        onHangup: @escaping @Sendable () -> Void
+    ) throws {
+        self.closeDescriptor = closeDescriptor
+        eventQueue = queue
+        self.onEvent = onEvent
+        self.onHangup = onHangup
+        let descriptor = kqueue()
+        guard descriptor >= 0 else {
+            throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO)
+        }
+        guard fileDescriptor >= 0 else {
+            closeDescriptor(descriptor)
+            throw POSIXError(.EBADF)
+        }
+        eventQueueDescriptor = descriptor
+        var registrations = [
+            kevent(
+                ident: UInt(fileDescriptor),
+                filter: Int16(EVFILT_READ),
+                flags: UInt16(EV_ADD | EV_CLEAR),
+                fflags: UInt32(NOTE_LOWAT),
+                data: Int.max,
+                udata: nil
+            ),
+            kevent(
+                ident: Self.cancelEventIdentifier,
+                filter: Int16(EVFILT_USER),
+                flags: UInt16(EV_ADD | EV_CLEAR),
+                fflags: 0,
+                data: 0,
+                udata: nil
+            )
+        ]
+        let registrationError = registrations.withUnsafeMutableBufferPointer { buffer in
+            registerEvents(descriptor, buffer.baseAddress, Int32(buffer.count))
+        }
+        if let registrationError {
+            isFinished = true
+            closeDescriptor(descriptor)
+            throw POSIXError(registrationError)
+        }
+    }
+
+    deinit {
+        cancel()
+    }
+
+    public func start() {
+        stateLock.lock()
+        guard !isStarted, !isCancelled, !isFinished else {
+            stateLock.unlock()
+            return
+        }
+        isStarted = true
+        stateLock.unlock()
+        eventQueue.async { [self] in
+            waitForHangup()
+        }
+    }
+
+    public func cancel() {
+        stateLock.lock()
+        guard !isFinished else {
+            stateLock.unlock()
+            return
+        }
+        isCancelled = true
+        guard isStarted else {
+            isFinished = true
+            closeDescriptor(eventQueueDescriptor)
+            stateLock.unlock()
+            return
+        }
+        var event = kevent(
+            ident: Self.cancelEventIdentifier,
+            filter: Int16(EVFILT_USER),
+            flags: 0,
+            fflags: UInt32(NOTE_TRIGGER),
+            data: 0,
+            udata: nil
+        )
+        _ = kevent(eventQueueDescriptor, &event, 1, nil, 0, nil)
+        stateLock.unlock()
+    }
+
+    private func waitForHangup() {
+        defer { finish() }
+        while true {
+            var event = kevent()
+            let result = kevent(eventQueueDescriptor, nil, 0, &event, 1, nil)
+            if result < 0 && errno == EINTR {
+                continue
+            }
+            guard result > 0 else { return }
+            onEvent()
+            if event.filter == Int16(EVFILT_USER) {
+                return
+            }
+            if event.flags & UInt16(EV_EOF | EV_ERROR) != 0 {
+                reportHangup()
+                return
+            }
+        }
+    }
+
+    private func finish() {
+        stateLock.lock()
+        isFinished = true
+        closeDescriptor(eventQueueDescriptor)
+        stateLock.unlock()
+    }
+
+    private func reportHangup() {
+        stateLock.lock()
+        guard !isCancelled, !didReportHangup else {
+            stateLock.unlock()
+            return
+        }
+        didReportHangup = true
+        stateLock.unlock()
+
+        onHangup()
+    }
+}