fix(reverse-attach): answer runtime pings during long tool calls (liveness)

README.md

@@ -202,7 +202,11 @@ curl -s -X POST -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/
   runtime with a lower operator ceiling rejects the request rather than silently shortening it.
 - **Redial policy** (openab-pty §9.2): stop on `4001` expired · `4002` replaced · `4004` session
   ended · `4010` revoked · handshake `401`; redial with backoff (1 s → 30 s) on `1000` / `4006` /
-  errors until the grant deadline. Nothing on the attached socket is trusted as identity — the
+  `4011` (liveness) / errors until the grant deadline. A connection that stayed up for 60 s resets
+  the backoff. The Mac keeps a receive outstanding during tool calls, so it answers the runtime's
+  pings, and pings the runtime itself every 15 s. On these attaches `exec` refuses a
+  `timeout_secs` over 100 s, because the runtime fails any call still running after 120 s; use
+  `exec_start` + `exec_poll` for long work. Nothing on the attached socket is trusted as identity — the
   grant is the identity; `Tailscale-User-Login` there would be pod-supplied.
 - `--no-attach` disables the plane (`/attach` → 404).
 

Sources/InstanceMCPCore/AttachManager.swift

@@ -220,27 +220,45 @@ public actor AttachManager {
 
     // MARK: helpers
 
+    /// Every grant, whatever its profile, is told how long a call may run. The
+    /// openab-pty runtime fails a call with no reply after 120 s, and older
+    /// runtimes or clients also drop the Mac during long calls (#68).
+    static let longCallNotice = """
+        Keep each tool call on this Mac under about 60 seconds: a longer call can time out or \
+        disconnect you. For long work use `exec_start`, then check it with `exec_poll` (when those \
+        tools are listed); never `sleep` inside `exec`.
+        """
+
+    /// Calls on one attach run concurrently (#68), so parallel input calls can
+    /// interleave on the screen.
+    static let uiOrderNotice = """
+        Tool calls run concurrently: issue `mouse`, `key` and `osascript` steps one at a time and \
+        wait for each result before sending the next.
+        """
+
     static func desktopInstructions(profile: ToolProfile, base: String?) -> String? {
-        if profile == .observe {
-            let head = base.map { $0 + "\n\n" } ?? ""
+        let head = base.map { $0 + "\n\n" } ?? ""
+        switch profile {
+        case .observe:
             return head + """
                 You reached this computer through OpenAB Connect under the `observe` profile: a human lets \
                 you look, not act. Only `sys_info` and `screenshot` are available — you cannot click, type \
                 or run anything here. If the task needs input on this computer, ask the human to grant \
                 the `desktop` profile instead.
-                """
+                """ + "\n\n" + longCallNotice
+        case .desktop:
+            return head + """
+                You reached this Mac through OpenAB Connect: a human lent it to your sandbox session for a \
+                limited time and is likely watching the screen. This is the `desktop` profile — there is no \
+                `exec` tool here (prefer the shell in your own session for local work); drive the Mac through \
+                `screenshot`, `mouse`, `key` and `osascript`, and — when `browser_*` tools are listed — through \
+                the browser directly: `browser_navigate` then `browser_snapshot` gives you the page as text, \
+                no screenshot needed. If a tool starts failing with "not attached", the grant ended; ask the \
+                human to lend the Mac again.
+                """ + "\n\n" + uiOrderNotice + "\n\n" + longCallNotice
+        case .owner:
+            return head + uiOrderNotice + "\n\n" + longCallNotice
         }
-        guard profile == .desktop else { return base }
-        let head = base.map { $0 + "\n\n" } ?? ""
-        return head + """
-            You reached this Mac through OpenAB Connect: a human lent it to your sandbox session for a \
-            limited time and is likely watching the screen. This is the `desktop` profile — there is no \
-            `exec` tool here (prefer the shell in your own session for local work); drive the Mac through \
-            `screenshot`, `mouse`, `key` and `osascript`, and — when `browser_*` tools are listed — through \
-            the browser directly: `browser_navigate` then `browser_snapshot` gives you the page as text, \
-            no screenshot needed. If a tool starts failing with "not attached", the grant ended; ask the \
-            human to lend the Mac again.
-            """
     }
 
     /// Build the exact runtime mint request. Keeping the body shape in a pure

Sources/InstanceMCPCore/ReverseAttachClient.swift

@@ -6,10 +6,19 @@ import Foundation
 ///
 /// The runtime is the MCP client here — it sends `initialize` first, then relays
 /// the CLI's `tools/list` / `tools/call` with its own integer ids. We answer each
-/// frame with the same id and otherwise treat the socket exactly like an HTTP
-/// request/response pair: one frame in, one frame out, no server-initiated
+/// frame with the same id: one frame in, one frame out, no server-initiated
 /// messages.
 ///
+/// Liveness: the runtime pings every 20 s and drops a Mac that stays silent for
+/// more than three intervals. `URLSessionWebSocketTask` only answers a ping while
+/// a `receive()` is outstanding, so a `receive()` is **always** outstanding: each
+/// frame is answered in its own task, replies go out through one writer task, and
+/// a slow tool call no longer silences the socket. Calls also run concurrently
+/// (the runtime allows many in flight). When a socket ends, for any reason, the
+/// calls it still has running are cancelled, and `exec` kills its process group.
+/// As a second safeguard we send our own ping every `keepaliveInterval` while
+/// attached; any inbound frame counts as alive on the runtime side.
+///
 /// Retry ownership is ours (the pod cannot reach us). Close codes decide:
 ///
 /// | code | meaning (§9.2) | we |
@@ -19,8 +28,12 @@ import Foundation
 /// | 4004 | session ended | stop |
 /// | 4006 | runtime replaced | redial with backoff while the grant is valid |
 /// | 4010 | grant revoked | stop |
+/// | 4011 | liveness: we were silent past the runtime's ping budget | redial with backoff while the grant is valid |
 /// | 1000 / error | dropped | redial with backoff while the grant is valid |
 ///
+/// A connection that stayed up for `stableAfter` resets the backoff, so a drop
+/// after a long attach redials in about a second.
+///
 /// A `401` on the handshake means the verifier is gone (expired, revoked, or the
 /// pod was replaced): stop and report it, there is nothing to wait for.
 ///
@@ -81,8 +94,11 @@ public actor ReverseAttachClient {
         public var secretFile: URL? = nil
         /// How long to wait after a credential refusal before trying again.
         public var credentialRetry: TimeInterval = 300
-        /// A connection that stayed up this long resets the backoff (switchboard).
+        /// A connection that stayed up this long resets the backoff.
         public var stableAfter: TimeInterval = 60
+        /// How often we ping the peer while attached, below the runtime's 20 s
+        /// ping interval. 0 turns our own pings off (tests).
+        public var keepaliveInterval: TimeInterval = 15
 
         public init(runtime: URL, session: String, secret: String, profile: ToolProfile, deadline: Date) {
             self.runtime = runtime; self.session = session; self.secret = secret
@@ -131,6 +147,7 @@ public actor ReverseAttachClient {
         case 4002: return .stop(.replaced)
         case 4004: return .stop(.sessionEnded)
         case 4010: return .stop(.revoked)
+        case 4011: return .redial            // liveness: silent past the ping budget (openab-pty#86)
         default:   return .redial            // 1000, 1006, 4006, anything else
         }
     }
@@ -159,6 +176,8 @@ public actor ReverseAttachClient {
     /// Frames answered on the current socket; surfaced for status.
     public private(set) var callsServed: Int = 0
     public private(set) var attachedSince: Date?
+    /// Calls still running, so our own cancellation can reach them.
+    private var inflight: [UUID: Task<Void, Never>] = [:]
 
     /// `server` should already be `scoped(to: config.profile)`; this type does
     /// not re-scope so a caller can pass a pre-built, instruction-tailored server.
@@ -180,6 +199,9 @@ public actor ReverseAttachClient {
         runLoopTask = nil
         task?.cancel(with: .normalClosure, reason: nil)
         task = nil
+        // Directly, not via the receive loop: during a redial backoff there is
+        // no loop left to do it.
+        cancelInflight()
         set(.ended(.cancelled))
     }
 
@@ -197,7 +219,7 @@ public actor ReverseAttachClient {
             let started = Date()
             let outcome = await dialOnce()
             if Task.isCancelled { return }
-            if config.kind == .switchboard, Date().timeIntervalSince(started) >= config.stableAfter {
+            if Date().timeIntervalSince(started) >= config.stableAfter {
                 backoff = config.initialBackoff
             }
             let base: TimeInterval
@@ -246,6 +268,10 @@ public actor ReverseAttachClient {
     }
 
     /// One connection lifetime. Returns what to do next.
+    ///
+    /// The receive loop never waits on a tool: each frame is answered in its own
+    /// task and the reply is queued to a single writer, so a `receive()` is
+    /// always outstanding and URLSession keeps answering the runtime's pings.
     private func dialOnce() async -> Disposition {
         let req = attachRequest()
         let session = URLSession(configuration: .ephemeral)
@@ -255,9 +281,36 @@ public actor ReverseAttachClient {
         task = ws
         ws.resume()
 
+        // One writer per socket: WebSocket sends must not interleave.
+        let (outbox, outboxIn) = AsyncStream<String>.makeStream()
+        let sendFailed = LockedFlag()
+        let writer = Task { [weak self] in
+            for await reply in outbox {
+                do { try await ws.send(.string(reply)) } catch {
+                    // The socket is broken; wake the receive loop so it redials.
+                    _ = sendFailed.trySet()
+                    ws.cancel()
+                    return
+                }
+                await self?.noteServed()
+            }
+        }
+        var keepalive: Task<Void, Never>?
+        defer {
+            // Every exit (stop code, retryable close, send failure, our own
+            // cancel) ends this socket's calls. The runtime has already failed
+            // them for the caller, so nothing they do can be delivered, and an
+            // `exec` left running would outlive the grant (it kills its process
+            // group on cancellation). Long work belongs in `exec_start`, whose
+            // jobs are not tied to a socket.
+            cancelInflight()
+            outboxIn.finish()
+            writer.cancel()
+            keepalive?.cancel()
+        }
+
         // The handshake outcome is only observable by receiving. A rejected
         // upgrade surfaces as an error whose task has an HTTPURLResponse.
-        var served = 0
         var firstFrame = true
         while true {
             let message: URLSessionWebSocketTask.Message
@@ -270,8 +323,8 @@ public actor ReverseAttachClient {
                     return Self.disposition(handshakeStatus: http.statusCode, kind: config.kind)
                 }
                 let code = ws.closeCode.rawValue
-                if code != 0 {
-                    log("attach \(config.session): closed \(code) after \(served) calls")
+                if code != 0, !sendFailed.isSet {
+                    log("attach \(config.session): closed \(code) after \(callsServed) calls")
                     finishAttached()
                     return Self.disposition(closeCode: code, kind: config.kind)
                 }
@@ -282,26 +335,58 @@ public actor ReverseAttachClient {
             if firstFrame {
                 firstFrame = false
                 attachedSince = Date()
+                callsServed = 0
                 set(.attached)
                 log("attach \(config.session): attached as \(config.profile.rawValue) to \(config.runtime.host ?? "?")")
+                keepalive = startKeepalive(ws)
             }
             let text: String
             switch message {
             case .string(let s): text = s
             case .data(let d): text = String(decoding: d, as: UTF8.self)
             @unknown default: continue
             }
-            if let reply = await answer(text) {
-                do { try await ws.send(.string(reply)) } catch {
-                    finishAttached()
-                    return .redial
+            // No per-socket cap: the runtime already bounds calls in flight (64),
+            // as the HTTP endpoint relies on its clients. Calls pipelined on one
+            // socket may finish out of order (each reply carries its id), so a
+            // caller that needs UI steps in order must wait for each reply.
+            let id = UUID()
+            inflight[id] = Task { [weak self] in
+                guard let self else { return }
+                if let reply = await self.answer(text) { outboxIn.yield(reply) }
+                await self.inflightDone(id)
+            }
+        }
+    }
+
+    /// Our own ping, so the runtime hears from us even if nothing else is sent.
+    private func startKeepalive(_ ws: URLSessionWebSocketTask) -> Task<Void, Never>? {
+        let interval = config.keepaliveInterval
+        guard interval > 0 else { return nil }
+        let session = config.session
+        let log = self.log
+        return Task {
+            while !Task.isCancelled {
+                try? await Task.sleep(nanoseconds: UInt64(interval * 1_000_000_000))
+                if Task.isCancelled { return }
+                ws.sendPing { error in
+                    if let error { log("attach \(session): keepalive ping failed: \(error.localizedDescription)") }
                 }
-                served += 1
-                callsServed = served
             }
         }
     }
 
+    private func noteServed() { callsServed += 1 }
+
+    private func inflightDone(_ id: UUID) { inflight[id] = nil }
+
+    /// On every socket exit and on our own `cancel()`: the calls of a dead
+    /// socket cannot be answered, so none of them carries into the next one.
+    private func cancelInflight() {
+        for t in inflight.values { t.cancel() }
+        inflight.removeAll()
+    }
+
     private func finishAttached() {
         attachedSince = nil
         task = nil
@@ -317,6 +402,7 @@ public actor ReverseAttachClient {
             return encode(JSONRPCResponse(id: .null, error: e))
         case .success(let r): rpc = r
         }
+        if let refusal = refuseLongExec(rpc) { return encode(refusal) }
         guard let response = await server.handle(rpc) else { return nil }
         if rpc.method == "tools/call" {
             let name = rpc.params?["name"]?.stringValue ?? "?"
@@ -325,6 +411,28 @@ public actor ReverseAttachClient {
         return encode(response)
     }
 
+    /// The longest `exec` / `osascript` timeout we accept on an openab-pty
+    /// attach. The runtime fails any call that has no reply after 120 s
+    /// (`REQUEST_TIMEOUT`), so a longer one could only end in an error for the
+    /// caller.
+    public static let maxAttachedExecTimeout: TimeInterval = 100
+
+    /// An `exec` or `osascript` over `maxAttachedExecTimeout` on an openab-pty
+    /// attach is refused as a tool error (the model sees it) pointing at
+    /// `exec_start`, rather than clamped, so the caller is not surprised by a
+    /// silent early kill.
+    func refuseLongExec(_ rpc: JSONRPCRequest) -> JSONRPCResponse? {
+        guard config.kind == .openabPty, rpc.method == "tools/call",
+              let tool = rpc.params?["name"]?.stringValue, tool == "exec" || tool == "osascript",
+              let asked = rpc.params?["arguments"]?["timeout_secs"]?.doubleValue,
+              asked > Self.maxAttachedExecTimeout else { return nil }
+        let limit = Int(Self.maxAttachedExecTimeout)
+        return JSONRPCResponse(id: rpc.id ?? .null, result: ToolResult.error(
+            "\(tool) timeout_secs \(Int(asked)) is over this attach's limit of \(limit) s: the runtime fails " +
+            "any call still running after 120 s. For long work use exec_start, then check it with " +
+            "exec_poll; never sleep inside exec.").json)
+    }
+
     private func encode(_ r: JSONRPCResponse) -> String {
         String(decoding: (try? JSONCoding.encoder.encode(r)) ?? Data("{}".utf8), as: UTF8.self)
     }