spb/zyquo-agent Public MIT
The autonomous agent that actually operates your Mac — plans, runs real commands, verifies its own work.
Swift 94.7%
Shell 4.1%
Python 0.7%
Makefile 0.5%
1//2// ExecutionService.swift3// Zyquo Agent4//5// Author: Simon-Pierre Boucher6// Mail: contact@spboucher.ai7//8// The single place Zyquo Agent spawns processes. Runs `/bin/bash -lc` and9// `/usr/bin/osascript` with the task workspace as cwd, streams stdout/stderr10// line by line to the UI as they arrive, enforces per-command timeouts11// (SIGTERM, then SIGKILL after a grace period), honors Swift Task12// cancellation, and caps captured output while the live stream still sees13// everything. Every caller reaches this actor only *after* the PolicyEngine14// has cleared the action — ExecutionService itself performs no policy checks15// and must never be invoked with an uncleared payload.16//1718import Foundation1920/// Tunables for process execution; a task can override the timeout per call.21struct ExecutionConfiguration: Sendable {22 /// Default per-command timeout when the caller does not specify one.23 var defaultTimeout: TimeInterval = 12024 /// Seconds between SIGTERM and SIGKILL when a process must die.25 var killGracePeriod: TimeInterval = 326 /// Maximum characters kept in the captured stdout/stderr/combined strings.27 /// The streaming callback is NOT capped — the UI terminal sees everything.28 var capturedOutputLimit: Int = 200_0002930 static let `default` = ExecutionConfiguration()31}3233/// Why a process stopped. Timeout and cancellation are reported distinctly34/// from a normal exit so tools can mark results accurately.35enum ExecutionTermination: Sendable, Equatable {36 /// The process exited (or was signalled) on its own; carries the exit code.37 case exited(Int32)38 /// The per-command timeout expired and the service killed the process.39 case timedOut40 /// The surrounding Swift Task was cancelled and the service killed the process.41 case cancelled42}4344/// Everything a caller learns from one process run.45struct ExecutionResult: Sendable {46 var termination: ExecutionTermination47 /// Exit code when available (also populated for killed processes when the48 /// OS reports one; nil only if the process never launched).49 var exitCode: Int32?50 /// Captured stdout, capped at `capturedOutputLimit`.51 var stdout: String52 /// Captured stderr, capped at `capturedOutputLimit`.53 var stderr: String54 /// stdout+stderr interleaved in arrival order, capped, with a truncation55 /// notice appended when the cap was hit.56 var combinedOutput: String57 /// True when any captured string was truncated at the cap.58 var wasTruncated: Bool59 /// Wall-clock seconds from launch to termination.60 var duration: TimeInterval6162 var isTimeout: Bool { termination == .timedOut }63 var isCancelled: Bool { termination == .cancelled }64}6566/// Thrown when the process cannot even be launched (bad executable, spawn67/// failure) — distinct from a process that ran and failed.68struct ExecutionLaunchError: Error, Sendable {69 var reason: String70}7172/// Actor that owns all child-process execution. Serialization through the73/// actor is intentional for bookkeeping, but the actual waiting is fully74/// async — long commands do not block other actor work because the heavy75/// lifting happens off-actor in handlers and continuations.76actor ExecutionService {77 private let configuration: ExecutionConfiguration7879 init(configuration: ExecutionConfiguration = .default) {80 self.configuration = configuration81 }8283 // MARK: - Public entry points8485 /// Runs a shell command via `/bin/bash -lc "<command>"` with `cwd` as the86 /// working directory. `onOutput` receives every stdout/stderr line live.87 func runBash(88 command: String,89 cwd: URL,90 timeout: TimeInterval? = nil,91 onOutput: @escaping @Sendable (ToolOutputChunk) -> Void92 ) async throws -> ExecutionResult {93 try await runProcess(94 executablePath: "/bin/bash",95 arguments: ["-lc", command],96 cwd: cwd,97 timeout: timeout,98 onOutput: onOutput99 )100 }101102 /// Runs an inline AppleScript via `/usr/bin/osascript -e <line> [-e <line>…]`.103 func runOSAScript(104 lines: [String],105 cwd: URL,106 timeout: TimeInterval? = nil,107 onOutput: @escaping @Sendable (ToolOutputChunk) -> Void108 ) async throws -> ExecutionResult {109 var arguments: [String] = []110 for line in lines {111 arguments.append("-e")112 arguments.append(line)113 }114 return try await runProcess(115 executablePath: "/usr/bin/osascript",116 arguments: arguments,117 cwd: cwd,118 timeout: timeout,119 onOutput: onOutput120 )121 }122123 /// Runs an AppleScript file via `/usr/bin/osascript <file>` (the path used124 /// for multi-line scripts, written under the workspace's `.zyquo/` dir).125 func runOSAScriptFile(126 scriptFile: URL,127 cwd: URL,128 timeout: TimeInterval? = nil,129 onOutput: @escaping @Sendable (ToolOutputChunk) -> Void130 ) async throws -> ExecutionResult {131 try await runProcess(132 executablePath: "/usr/bin/osascript",133 arguments: [scriptFile.path],134 cwd: cwd,135 timeout: timeout,136 onOutput: onOutput137 )138 }139140 /// General process runner used by the specialized entry points above.141 /// Environment: inherits the user's environment, sets `PWD` to the cwd and142 /// adds `ZYQUO_AGENT=1` so scripts can detect they run under the agent.143 func runProcess(144 executablePath: String,145 arguments: [String],146 cwd: URL,147 timeout: TimeInterval? = nil,148 onOutput: @escaping @Sendable (ToolOutputChunk) -> Void149 ) async throws -> ExecutionResult {150 // A cancelled task must not spawn anything.151 if Task.isCancelled {152 return ExecutionResult(153 termination: .cancelled, exitCode: nil,154 stdout: "", stderr: "", combinedOutput: "",155 wasTruncated: false, duration: 0156 )157 }158159 let supervisor = ProcessSupervisor(160 executablePath: executablePath,161 arguments: arguments,162 cwd: cwd,163 outputLimit: configuration.capturedOutputLimit,164 killGracePeriod: configuration.killGracePeriod,165 onOutput: onOutput166 )167 let effectiveTimeout = timeout ?? configuration.defaultTimeout168 return try await supervisor.run(timeout: effectiveTimeout)169 }170}171172// MARK: - ProcessSupervisor173174/// Owns exactly one child process for its lifetime: launch, line-streaming,175/// timeout escalation (SIGTERM → SIGKILL), cancellation, and reaping. All176/// mutable state is behind `lock`, so the class is safe to touch from the177/// pipe-reader queues, the termination handler, the timeout task, and the178/// cancellation handler concurrently.179private final class ProcessSupervisor: @unchecked Sendable {180 private let process = Process()181 private let stdoutPipe = Pipe()182 private let stderrPipe = Pipe()183 private let onOutput: @Sendable (ToolOutputChunk) -> Void184 private let outputLimit: Int185 private let killGracePeriod: TimeInterval186187 /// Serial queue that funnels both pipes' data so combined output keeps the188 /// best-possible interleave order and capture buffers mutate safely.189 private let outputQueue = DispatchQueue(label: "com.zyquo.agent.execution.output")190 /// Signalled (once per stream) when a pipe reaches EOF.191 private let drainGroup = DispatchGroup()192193 private let lock = NSLock()194 // Guarded by `lock`:195 private var overrideTermination: ExecutionTermination?196 private var continuation: CheckedContinuation<Void, Never>?197 private var terminated = false198199 // Guarded by `outputQueue`:200 private var stdoutCapture = ""201 private var stderrCapture = ""202 private var combinedCapture = ""203 private var truncated = false204 private var stdoutLineBuffer = ""205 private var stderrLineBuffer = ""206207 init(208 executablePath: String,209 arguments: [String],210 cwd: URL,211 outputLimit: Int,212 killGracePeriod: TimeInterval,213 onOutput: @escaping @Sendable (ToolOutputChunk) -> Void214 ) {215 self.onOutput = onOutput216 self.outputLimit = outputLimit217 self.killGracePeriod = killGracePeriod218219 process.executableURL = URL(fileURLWithPath: executablePath)220 process.arguments = arguments221 process.currentDirectoryURL = cwd222223 var environment = ProcessInfo.processInfo.environment224 environment["PWD"] = cwd.path225 environment["ZYQUO_AGENT"] = "1"226 process.environment = environment227228 process.standardOutput = stdoutPipe229 process.standardError = stderrPipe230 process.standardInput = FileHandle.nullDevice231 }232233 /// Launches the process and suspends until it terminates, times out, or234 /// the surrounding Task is cancelled. Never throws after a successful235 /// launch — abnormal endings are reported in `ExecutionResult.termination`.236 func run(timeout: TimeInterval) async throws -> ExecutionResult {237 installReader(for: stdoutPipe.fileHandleForReading, isStdout: true)238 installReader(for: stderrPipe.fileHandleForReading, isStdout: false)239240 let start = Date()241 process.terminationHandler = { [weak self] _ in242 self?.finishWaiting()243 }244245 do {246 try process.run()247 } catch {248 // Detach the readers we installed; the pipes never produced data.249 stdoutPipe.fileHandleForReading.readabilityHandler = nil250 stderrPipe.fileHandleForReading.readabilityHandler = nil251 drainGroup.leave()252 drainGroup.leave()253 throw ExecutionLaunchError(reason: "Could not launch \(process.executableURL?.path ?? "?"): \(error.localizedDescription)")254 }255256 // Timeout watchdog: mark the run as timed out, then escalate257 // SIGTERM → (grace) → SIGKILL. The termination handler is the single258 // point that resumes the continuation, so there is no double-resume.259 let watchdog = Task { [weak self] in260 try? await Task.sleep(nanoseconds: UInt64(max(0, timeout) * 1_000_000_000))261 guard let self, !Task.isCancelled else { return }262 self.kill(as: .timedOut)263 }264265 await withTaskCancellationHandler {266 await withCheckedContinuation { (cont: CheckedContinuation<Void, Never>) in267 let alreadyDone: Bool268 lock.lock()269 if terminated {270 alreadyDone = true271 } else {272 continuation = cont273 alreadyDone = false274 }275 lock.unlock()276 if alreadyDone { cont.resume() }277 }278 } onCancel: {279 self.kill(as: .cancelled)280 }281 watchdog.cancel()282283 // Wait (bounded) for the pipes to drain so trailing output is not284 // lost. The bound matters: a backgrounded grandchild can keep the285 // write end open forever even after bash itself has died. The wait286 // happens on a Dispatch thread — never blocks the cooperative pool.287 await withCheckedContinuation { (cont: CheckedContinuation<Void, Never>) in288 DispatchQueue.global(qos: .utility).async { [drainGroup] in289 _ = drainGroup.wait(timeout: .now() + 2.0)290 cont.resume()291 }292 }293 stdoutPipe.fileHandleForReading.readabilityHandler = nil294 stderrPipe.fileHandleForReading.readabilityHandler = nil295296 let duration = Date().timeIntervalSince(start)297298 let override = recordedOverride()299300 let exitCode = process.processIdentifier > 0 ? process.terminationStatus : nil301 let termination = override ?? .exited(exitCode ?? -1)302303 // Flush partial trailing lines and read the final buffers on the304 // output queue so we see everything the readers appended.305 return outputQueue.sync {306 flushLineBuffersLocked()307 var combined = combinedCapture308 if truncated {309 combined += "\n… [output truncated at \(outputLimit / 1000)KB — full stream was shown live]"310 }311 return ExecutionResult(312 termination: termination,313 exitCode: exitCode,314 stdout: stdoutCapture,315 stderr: stderrCapture,316 combinedOutput: combined,317 wasTruncated: truncated,318 duration: duration319 )320 }321 }322323 /// Lock-guarded read of the recorded termination override (sync helper —324 /// NSLock must not be taken directly inside an async function).325 private func recordedOverride() -> ExecutionTermination? {326 lock.lock()327 defer { lock.unlock() }328 return overrideTermination329 }330331 // MARK: Killing332333 /// Records why the process is being killed, sends SIGTERM, and schedules a334 /// SIGKILL after the grace period. Safe to call from any thread; the first335 /// recorded reason wins.336 private func kill(as reason: ExecutionTermination) {337 lock.lock()338 if terminated {339 lock.unlock()340 return341 }342 if overrideTermination == nil { overrideTermination = reason }343 lock.unlock()344345 let pid = process.processIdentifier346 guard pid > 0 else { return }347 Foundation.kill(pid, SIGTERM)348349 let grace = killGracePeriod350 DispatchQueue.global().asyncAfter(deadline: .now() + grace) { [weak self] in351 guard let self else { return }352 self.lock.lock()353 let stillRunning = !self.terminated354 self.lock.unlock()355 if stillRunning {356 Foundation.kill(pid, SIGKILL)357 }358 }359 }360361 /// Termination handler target — marks the process reaped and resumes the362 /// suspended `run(timeout:)`.363 private func finishWaiting() {364 lock.lock()365 terminated = true366 let cont = continuation367 continuation = nil368 lock.unlock()369 cont?.resume()370 }371372 // MARK: Output plumbing373374 /// Attaches a readability handler that forwards raw data to the serial375 /// output queue where it is captured (capped) and split into lines for376 /// the live stream. An empty read means EOF: detach and signal the drain.377 private func installReader(for handle: FileHandle, isStdout: Bool) {378 drainGroup.enter()379 handle.readabilityHandler = { [weak self] fileHandle in380 let data = fileHandle.availableData381 guard let self else { return }382 if data.isEmpty {383 fileHandle.readabilityHandler = nil384 self.outputQueue.async { self.drainGroup.leave() }385 return386 }387 self.outputQueue.async {388 self.consume(data: data, isStdout: isStdout)389 }390 }391 }392393 /// Runs on `outputQueue` only. Captures (with cap) and emits whole lines.394 private func consume(data: Data, isStdout: Bool) {395 let text = String(decoding: data, as: UTF8.self)396397 // Capture with cap. The live stream below is never capped.398 appendCapped(text, to: isStdout ? \.stdoutCapture : \.stderrCapture)399 appendCapped(text, to: \.combinedCapture)400401 // Line-split for streaming; keep the unterminated tail buffered.402 if isStdout {403 stdoutLineBuffer += text404 emitCompleteLines(from: &stdoutLineBuffer, isStdout: true)405 } else {406 stderrLineBuffer += text407 emitCompleteLines(from: &stderrLineBuffer, isStdout: false)408 }409 }410411 private func appendCapped(_ text: String, to keyPath: ReferenceWritableKeyPath<ProcessSupervisor, String>) {412 let current = self[keyPath: keyPath]413 guard current.count < outputLimit else {414 truncated = true415 return416 }417 let remaining = outputLimit - current.count418 if text.count <= remaining {419 self[keyPath: keyPath] = current + text420 } else {421 self[keyPath: keyPath] = current + String(text.prefix(remaining))422 truncated = true423 }424 }425426 private func emitCompleteLines(from buffer: inout String, isStdout: Bool) {427 while let newlineRange = buffer.range(of: "\n") {428 let line = String(buffer[buffer.startIndex..<newlineRange.lowerBound])429 buffer.removeSubrange(buffer.startIndex..<newlineRange.upperBound)430 onOutput(isStdout ? .stdout(line) : .stderr(line))431 }432 }433434 /// Runs on `outputQueue` only. Emits any unterminated trailing output.435 private func flushLineBuffersLocked() {436 if !stdoutLineBuffer.isEmpty {437 onOutput(.stdout(stdoutLineBuffer))438 stdoutLineBuffer = ""439 }440 if !stderrLineBuffer.isEmpty {441 onOutput(.stderr(stderrLineBuffer))442 stderrLineBuffer = ""443 }444 }445}446