Add run_on_client: agents run commands with files on the user's Macs

2026-10-17 14:41:04 -07:00
parent 14975d2ab8
commit fbc0699fdf
16 changed files with 928 additions and 249 deletions
@@ -0,0 +1,166 @@
+import ByreAPI
+import Foundation
+
+private let maxOutput = 64 << 10
+
+private let runsDir = FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask)[0].appending(path: "run.fc.byre.mac/runs", directoryHint: .isDirectory)
+
+@MainActor
+private var watching: Set<String> = []
+
+@MainActor
+func runOnThisMac(_ api: API, _ item: InboxItem) async throws {
+    let dir = runsDir.appending(path: UUID().uuidString, directoryHint: .isDirectory)
+    watching.insert(dir.path)
+    do {
+        try await start(api, item, dir)
+    } catch {
+        try? FileManager.default.removeItem(at: dir)
+        watching.remove(dir.path)
+        throw error
+    }
+    poll(api, dir)
+}
+
+@MainActor
+private func start(_ api: API, _ item: InboxItem, _ dir: URL) async throws {
+    let files = dir.appending(path: "files", directoryHint: .isDirectory)
+    try FileManager.default.createDirectory(at: files, withIntermediateDirectories: true)
+    try item.id.write(to: dir.appending(path: "item"), atomically: true, encoding: .utf8)
+    try api.origin.absoluteString.write(to: dir.appending(path: "server"), atomically: true, encoding: .utf8)
+    try str(item.payload.value, "command")!.write(to: dir.appending(path: "command.sh"), atomically: true, encoding: .utf8)
+
+    let (bundle, response) = try await api.session.download(from: api.url("/api/inbox/\(item.id)/bundle"))
+    let code = (response as! HTTPURLResponse).statusCode
+    if code != 200 {
+        let text = (try? String(contentsOf: bundle, encoding: .utf8)) ?? ""
+        throw ApiError(status: code, message: "download bundle: \(code) \(text)")
+    }
+    try await untar(bundle, into: files, errors: dir.appending(path: "untar.log"))
+    try? FileManager.default.removeItem(at: bundle)
+
+    let p = Process()
+    p.executableURL = URL(filePath: "/usr/bin/lockf")
+    p.arguments = ["-k", "lock", "/bin/zsh", "-c", "cd files && /bin/zsh -l ../command.sh >../output.log 2>&1; echo $? >../exit.tmp; mv ../exit.tmp ../exit"]
+    p.currentDirectoryURL = dir
+    p.standardInput = FileHandle.nullDevice
+    p.standardOutput = FileHandle.nullDevice
+    p.standardError = FileHandle.nullDevice
+    try p.run()
+    let lock = dir.appending(path: "lock")
+    while !FileManager.default.fileExists(atPath: lock.path) {
+        if !p.isRunning {
+            throw ApiError(status: 0, message: "start command: lockf exited \(p.terminationStatus) before taking its lock")
+        }
+        try await Task.sleep(for: .milliseconds(20))
+    }
+    log.info("client run started: item \(item.id, privacy: .public) in \(dir.path, privacy: .public)")
+}
+
+private func untar(_ bundle: URL, into dir: URL, errors: URL) async throws {
+    FileManager.default.createFile(atPath: errors.path, contents: nil)
+    let errFile = try FileHandle(forWritingTo: errors)
+    defer {
+        try? errFile.close()
+    }
+    let p = Process()
+    p.executableURL = URL(filePath: "/usr/bin/tar")
+    p.arguments = ["-xzf", bundle.path, "-C", dir.path]
+    p.standardError = errFile
+    p.standardOutput = FileHandle.nullDevice
+    try await withCheckedThrowingContinuation { (c: CheckedContinuation<Void, any Error>) in
+        p.terminationHandler = { _ in
+            c.resume()
+        }
+        do {
+            try p.run()
+        } catch {
+            c.resume(throwing: error)
+        }
+    }
+    if p.terminationStatus != 0 {
+        let text = tail(errors)
+        throw ApiError(status: 0, message: "unpack bundle: tar exited \(p.terminationStatus): \(text)")
+    }
+}
+
+@MainActor
+func resumeClientRuns(_ api: API) {
+    let dirs = (try? FileManager.default.contentsOfDirectory(at: runsDir, includingPropertiesForKeys: nil)) ?? []
+    for dir in dirs where !watching.contains(dir.path) && (try? String(contentsOf: dir.appending(path: "server"), encoding: .utf8)) == api.origin.absoluteString {
+        if !FileManager.default.fileExists(atPath: dir.appending(path: "lock").path) {
+            log.info("client run never started; removing \(dir.path, privacy: .public)")
+            try? FileManager.default.removeItem(at: dir)
+            continue
+        }
+        watching.insert(dir.path)
+        poll(api, dir)
+    }
+}
+
+@MainActor
+private func poll(_ api: API, _ dir: URL) {
+    Task {
+        let exitFile = dir.appending(path: "exit").path
+        var interrupted = false
+        while !FileManager.default.fileExists(atPath: exitFile) {
+            if lockFree(dir.appending(path: "lock").path) {
+                interrupted = !FileManager.default.fileExists(atPath: exitFile)
+                break
+            }
+            try? await Task.sleep(for: .seconds(1))
+        }
+        await report(api, dir, interrupted: interrupted)
+        watching.remove(dir.path)
+    }
+}
+
+private func lockFree(_ path: String) -> Bool {
+    guard let h = FileHandle(forReadingAtPath: path) else {
+        return false
+    }
+    defer {
+        try? h.close()
+    }
+    let fd = h.fileDescriptor
+    if flock(fd, LOCK_EX | LOCK_NB) != 0 {
+        return false
+    }
+    flock(fd, LOCK_UN)
+    return true
+}
+
+@MainActor
+private func report(_ api: API, _ dir: URL, interrupted: Bool) async {
+    do {
+        let itemId = trimmed(try String(contentsOf: dir.appending(path: "item"), encoding: .utf8))
+        var exit = -1
+        var output = tail(dir.appending(path: "output.log"))
+        if interrupted {
+            output += "\n[byre: the command stopped without finishing, for example because the Mac restarted]"
+        } else {
+            exit = Int(trimmed(try String(contentsOf: dir.appending(path: "exit"), encoding: .utf8))) ?? -1
+        }
+        let client = Host.current().localizedName ?? ProcessInfo.processInfo.hostName
+        _ = try await api.client.reportRun(path: .init(itemId: itemId), body: .json(.init(runId: dir.lastPathComponent, client: client, exitCode: exit, output: output))).ok
+        log.info("client run reported: item \(itemId, privacy: .public) exit \(exit)")
+    } catch {
+        log.error("report client run \(dir.path, privacy: .public): \(errorText(error), privacy: .public)")
+        guard let e = rootError(error) as? ApiError, (400..<500).contains(e.status) else {
+            return
+        }
+    }
+    try? FileManager.default.removeItem(at: dir)
+}
+
+private func tail(_ file: URL) -> String {
+    guard let h = try? FileHandle(forReadingFrom: file) else {
+        return ""
+    }
+    defer {
+        try? h.close()
+    }
+    let size = (try? h.seekToEnd()) ?? 0
+    try? h.seek(toOffset: size > UInt64(maxOutput) ? size - UInt64(maxOutput) : 0)
+    return String(decoding: (try? h.readToEnd()) ?? Data(), as: UTF8.self)
+}
@@ -192,6 +192,21 @@ private final class CardModel {
        }
    }

+    func run() {
+        if busy {
+            return
+        }
+        busy = true
+        Task {
+            do {
+                try await runOnThisMac(store.api, item)
+            } catch {
+                overlays.showAlert(errorText(error))
+            }
+            busy = false
+        }
+    }
+
    func submitOther() {
        if typed.isEmpty && !hasImages {
            return
@@ -289,10 +304,29 @@ private struct CardBody: View {
                }
                .textSelection(.enabled)
                .padding(.vertical, 10)
+            case .clientRun:
+                VStack(alignment: .leading, spacing: 10) {
+                    MarkdownView(blocks: parseMarkdown(jsString(p["why"] ?? "")), ctx: MDContext(taskId: item.taskId))
+                    MarkdownView(blocks: runDetails(p), ctx: MDContext(taskId: item.taskId))
+                }
+                .padding(.vertical, 10)
            }
        }
    }

+    private func runDetails(_ p: JSONDict) -> [MDBlock] {
+        let paths = (p["paths"] as? [(any Sendable)?] ?? []).compactMap { $0 as? String }
+        let bytes = (p["bytes"] as? Int) ?? Int((p["bytes"] as? Double) ?? 0)
+        var files: [MDInline] = [.text("No files.")]
+        if !paths.isEmpty {
+            files = [.text("Files (\(ByteCountFormatter.string(fromByteCount: Int64(bytes), countStyle: .file))): ")]
+            for (i, path) in paths.enumerated() {
+                files += i == 0 ? [.code(path)] : [.text(", "), .code(path)]
+            }
+        }
+        return [.code(jsString(p["command"] ?? "")), .paragraph(files)]
+    }
+
    private func attachments(_ v: (any Sendable)?) -> [Attachment] {
        guard let v else {
            return []
@@ -451,6 +485,19 @@ private struct CardActions: View {
                }
                .disabled(model.busy)
            }
+        } else if model.item.kind == .clientRun {
+            HStack(spacing: 8) {
+                Button("Dismiss") {
+                    model.submit(InboxResponseCreate(itemId: model.item.id, _type: .dismiss))
+                }
+                .buttonStyle(WebButtonStyle())
+                Spacer()
+                Button("Run on this Mac") {
+                    model.run()
+                }
+                .buttonStyle(WebButtonStyle(primary: true))
+            }
+            .disabled(model.busy)
        } else {
            HStack(alignment: .top, spacing: 8) {
                TextEntry(model: model, label: "Reply…", placeholder: "Reply to the agent", shared: false, focus: focus) {
@@ -596,11 +643,19 @@ private struct Thread: View {
        if !item.responses.isEmpty {
            VStack(alignment: .leading, spacing: 2) {
                ForEach(item.responses, id: \.id) { r in
-                    let images = r.images.isEmpty ? "" : " (\(r.images.count == 1 ? "1 image" : "\(r.images.count) images"))"
-                    let rest = " \(r._type.rawValue.replacingOccurrences(of: "_", with: " "))\(r.options.map { ": \($0.joined(separator: "; "))" } ?? "")\(r.text.map { " — \($0)" } ?? "")\(images)"
-                    Text("\(Text(r.user).font(Fonts.font(11, weight: 700)))\(rest)")
-                        .textSelection(.enabled)
-                        .fixedSize(horizontal: false, vertical: true)
+                    if r._type == .run {
+                        Text("\(Text(r.user).font(Fonts.font(11, weight: 700))) ran it on \(r.client ?? ""): exit \(r.exitCode ?? -1)")
+                            .textSelection(.enabled)
+                        if let out = r.text, !trimmed(out).isEmpty {
+                            MarkdownView(blocks: [.details(summary: [.text("Output")], body: [.code(out)])], ctx: MDContext(taskId: item.taskId))
+                        }
+                    } else {
+                        let images = r.images.isEmpty ? "" : " (\(r.images.count == 1 ? "1 image" : "\(r.images.count) images"))"
+                        let rest = " \(r._type.rawValue.replacingOccurrences(of: "_", with: " "))\(r.options.map { ": \($0.joined(separator: "; "))" } ?? "")\(r.text.map { " — \($0)" } ?? "")\(images)"
+                        Text("\(Text(r.user).font(Fonts.font(11, weight: 700)))\(rest)")
+                            .textSelection(.enabled)
+                            .fixedSize(horizontal: false, vertical: true)
+                    }
                }
            }
            .font(Fonts.font(11))
@@ -43,7 +43,7 @@ final class Notifier: NSObject, UNUserNotificationCenterDelegate {

 private func wants(_ p: NotificationPrefs, _ item: InboxItem) -> Bool {
    switch item.kind {
-    case .question: p.questions
+    case .question, .clientRun: p.questions
    case .sessionError: p.errors
    case .rateLimited: p.rateLimits
    }
@@ -55,6 +55,7 @@ private func itemText(_ item: InboxItem) -> String {
    case .question: return str(p, "question") ?? ""
    case .sessionError: return "Error: \(str(p, "error") ?? "")"
    case .rateLimited: return "Rate limited"
+    case .clientRun: return "Run on your Mac: \(str(p, "why") ?? "")"
    }
 }

@@ -121,6 +121,7 @@ final class Connection {
            }
            phase = .running(store)
            store.startGlobalStream(after: status.lastEventSeq)
+            resumeClientRuns(store.api)
            Notifier.shared.deliver()
        } catch {
            await store.stop()
@@ -163,6 +163,9 @@ final class Store {
        if open.contains(where: { $0.kind == .question }) {
            return TaskState(kind: .answer, label: "Needs your answer", title: "The agent asked a question in the inbox")
        }
+        if open.contains(where: { $0.kind == .clientRun }) {
+            return TaskState(kind: .answer, label: "Ready to run", title: "The agent asked to run something on your Mac; see the inbox")
+        }
        if task.merging {
            return TaskState(kind: .merging, label: "Merging", title: "The merge turn is running")
        }