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// AgentLoop.swift3// Zyquo Agent4//5// Author: Simon-Pierre Boucher6// Mail: contact@spboucher.ai7//8// The plan→act→observe→reflect engine (docs/AGENT-RESEARCH.md §1.3): a9// while-loop keyed on the normalized StopReason. Each iteration streams one10// model turn (text, thinking, and tool-call arguments live), executes any11// tool calls through the ToolRegistry — every side effect pre-cleared by12// the PolicyEngine and audited — threads the results back as one message13// correlated by call id, offloads oversized outputs, compacts memory near14// the context limit, and checks the LoopGuard. A turn without tool calls15// is the done-signal: its text is the user-facing result. Trips pause the16// run (resume/stop), they never abort it; cancellation kills any in-flight17// process via Task cancellation.18//1920import Foundation2122/// Per-run engine tunables (Settings › Agent in Phase 6).23struct AgentConfiguration: Sendable {24 var loopGuard = LoopGuardConfiguration.default25 var memory = MemoryConfiguration.default26 /// Execute a turn's tool calls concurrently when they are ALL read-only27 /// under the active policy. Mutating calls always run sequentially.28 var parallelToolCalls = false29 /// Default per-command timeout for the ExecutionService the host wires up30 /// (the loop itself never spawns processes).31 var perCommandTimeout: TimeInterval?32 /// Active persona's system-prompt addition, appended to the agent system33 /// prompt as a trailing "## Persona" section (Phase 6 personas).34 var personaAddendum: String?3536 static let `default` = AgentConfiguration()37}3839/// The agent engine. Actor: one run at a time, all mutable run state40/// (history, plan, guard, transcript) confined here.41actor AgentLoop {42 // MARK: Dependencies4344 private let model: AIModel45 private let client: any ProviderClient46 private let apiKey: String47 private let tools: ToolRegistry48 private let policy: PolicyEngine49 private let audit: AuditLog50 private let workspace: WorkspaceManager51 private let configuration: AgentConfiguration5253 // MARK: Run state5455 private var planner: Planner56 private var loopGuard: LoopGuard57 private let memory: MemoryManager58 private var transcript: Transcript?59 private var history: [HistoryEntry] = []60 private var runTask: Task<Void, Never>?61 private var eventContinuation: AsyncThrowingStream<AgentEvent, Error>.Continuation?62 private var tripContinuation: CheckedContinuation<TripDecision, Never>?63 private var isRunning = false6465 /// The user's answer to a LoopGuard trip.66 enum TripDecision: Sendable {67 case resume(raiseBudget: Bool)68 case stop69 }7071 init(72 model: AIModel,73 client: any ProviderClient,74 apiKey: String,75 tools: ToolRegistry,76 policy: PolicyEngine,77 audit: AuditLog,78 workspace: WorkspaceManager,79 configuration: AgentConfiguration = .default80 ) {81 self.model = model82 self.client = client83 self.apiKey = apiKey84 self.tools = tools85 self.policy = policy86 self.audit = audit87 self.workspace = workspace88 self.configuration = configuration89 self.planner = Planner(workspaceRoot: workspace.root)90 self.loopGuard = LoopGuard(configuration: configuration.loopGuard)91 self.memory = MemoryManager(92 workspaceRoot: workspace.root,93 outputsDirectory: workspace.outputsDirectory,94 configuration: configuration.memory95 )96 }9798 // MARK: - Public API99100 /// Starts the run and returns its live event stream. The stream ends101 /// after `.runFinished`; provider failures that end the run are reported102 /// through `.runFinished(.failed…)`, not thrown.103 func run(task: String) -> AsyncThrowingStream<AgentEvent, Error> {104 let (stream, continuation) = AsyncThrowingStream<AgentEvent, Error>.makeStream()105 guard !isRunning else {106 continuation.yield(.runFinished(.failed(reason: "A run is already in progress on this AgentLoop.")))107 continuation.finish()108 return stream109 }110 isRunning = true111 eventContinuation = continuation112 runTask = Task {113 await self.execute(task: task)114 }115 return stream116 }117118 /// Stops the run: cancels the in-flight model stream and any executing119 /// process (ExecutionService honors Task cancellation with SIGTERM→SIGKILL).120 func cancel() {121 runTask?.cancel()122 // A run paused on a guard trip is not awaiting a cancellable123 // suspension — unblock it explicitly.124 tripContinuation?.resume(returning: .stop)125 tripContinuation = nil126 }127128 /// Answers a `.guardTripped` pause: continue, optionally raising every129 /// budget (without raising, a budget trip will re-trip on the next step;130 /// repetition/stall trips reset their counters either way).131 func resume(raisingBudget: Bool = true) {132 guard let continuation = tripContinuation else { return }133 tripContinuation = nil134 continuation.resume(returning: .resume(raiseBudget: raisingBudget))135 }136137 /// Answers a `.guardTripped` pause: end the run.138 func stop() {139 guard let continuation = tripContinuation else { return }140 tripContinuation = nil141 continuation.resume(returning: .stop)142 }143144 // MARK: - The loop145146 private func execute(task: String) async {147 emit(.statusChanged(.preparing))148149 guard model.capabilities.tools else {150 finish(.failed(reason: "\(model.displayName) does not support tool calling and cannot run agent tasks. Pick an agent-capable model."))151 return152 }153154 memory.ensureMemoryFile()155 transcript = Transcript(task: task, model: model, workspaceRoot: workspace.root)156 history = [HistoryEntry(stepIndex: 0, message: Message(role: .user, text: task), pinned: true)]157 if let plan = planner.plan {158 // Reopened workspace with a persisted plan — surface it.159 emit(.planUpdated(plan))160 }161162 let systemPrompt = AgentSystemPrompt.build(163 workspacePath: workspace.root.path,164 toolNames: tools.toolNames,165 safetyMode: await policy.mode,166 personaAddendum: configuration.personaAddendum167 )168 let toolSpecs = tools.toolSpecs + [Planner.toolSpec]169 let fixedTokens = MemoryManager.estimateFixedTokens(systemPrompt: systemPrompt, toolSpecs: toolSpecs)170171 var stepIndex = 0172 var truncationNudges = 0173 var emptyAnswerNudges = 0174175 while true {176 if Task.isCancelled {177 finish(.cancelled)178 return179 }180 stepIndex += 1181182 // ---- LoopGuard budgets (pause-and-ask, never abort) ----------183 if let trip = loopGuard.checkBeforeStep(index: stepIndex) {184 if await pauseOnTrip(trip) == false {185 finish(.stoppedByUser(reason: trip.message))186 return187 }188 }189190 // ---- Memory compaction (before the request, at step boundary) -191 if memory.shouldCompact(entries: history, fixedTokens: fixedTokens, contextWindow: model.contextWindow, currentStep: stepIndex - 1) {192 emit(.statusChanged(.compacting))193 if let outcome = await memory.compact(194 entries: history,195 planContext: planner.renderedForContext(),196 steps: transcript?.document.steps ?? [],197 client: client,198 apiKey: apiKey,199 model: model,200 currentStep: stepIndex - 1,201 fixedTokens: fixedTokens202 ) {203 history = outcome.entries204 transcript?.recordCompaction(outcome.record)205 emit(.compactionPerformed(outcome.record))206 }207 }208209 // ---- One model turn, streamed --------------------------------210 var step = AgentStep(index: stepIndex, text: "")211 emit(.stepStarted(step))212 emit(.statusChanged(.running))213214 let turn: ModelTurn215 do {216 turn = try await streamModelTurn(stepID: step.id, systemPrompt: systemPrompt, toolSpecs: toolSpecs)217 } catch is CancellationError {218 finish(.cancelled)219 return220 } catch {221 step.status = .failed222 step.finishedAt = Date()223 transcript?.upsert(step: step)224 emit(.stepCompleted(step))225 finish(.failed(reason: error.localizedDescription))226 return227 }228229 step.text = turn.text230 step.thinking = turn.reasoning.isEmpty ? nil : turn.reasoning231 if let usage = turn.usage {232 step.inputTokens = usage.inputTokens233 step.outputTokens = usage.outputTokens234 loopGuard.recordUsage(usage)235 transcript?.addUsage(usage)236 memory.lastReportedInputTokens = usage.inputTokens237 }238239 // ---- Branch on the normalized stop reason --------------------240 switch turn.stop {241 case .toolUse where !turn.toolCalls.isEmpty:242 history.append(HistoryEntry(243 stepIndex: stepIndex,244 message: .assistantToolCalls(turn.toolCalls, text: turn.text, reasoning: step.thinking)245 ))246 step.status = .executing247 emit(.statusChanged(.executingTools))248249 let filesBefore = await filesSignature()250 let execution: ToolExecutionOutcome251 do {252 execution = try await executeToolCalls(turn.toolCalls, stepIndex: stepIndex, stepID: step.id)253 } catch {254 // Only CancellationError escapes the registry.255 step.status = .cancelled256 step.finishedAt = Date()257 transcript?.upsert(step: step)258 emit(.stepCompleted(step))259 finish(.cancelled)260 return261 }262 history.append(HistoryEntry(stepIndex: stepIndex, message: .toolResultsMessage(execution.results)))263 await workspace.refreshScan()264 let filesAfter = await filesSignature()265266 step.toolInvocations = execution.invocations267 step.status = .completed268 step.finishedAt = Date()269 transcript?.upsert(step: step)270 emit(.stepCompleted(step))271272 if let trip = loopGuard.recordStepOutcome(273 stepIndex: stepIndex,274 planChanged: execution.planChanged,275 filesChanged: filesBefore != filesAfter,276 sawNewToolCall: execution.sawNewToolCall277 ) {278 if await pauseOnTrip(trip) == false {279 finish(.stoppedByUser(reason: trip.message))280 return281 }282 }283284 case .endTurn, .toolUse, .other:285 // .toolUse with an empty call list and .other are anomalies —286 // treat a non-empty text as the final answer, nudge otherwise.287 let answer = turn.text.trimmingCharacters(in: .whitespacesAndNewlines)288 if answer.isEmpty, stepIndex > 1, emptyAnswerNudges < 1 {289 emptyAnswerNudges += 1290 step.status = .completed291 step.finishedAt = Date()292 transcript?.upsert(step: step)293 emit(.stepCompleted(step))294 history.append(HistoryEntry(295 stepIndex: stepIndex,296 message: Message(role: .user, text: "Your last message was empty. If the task is complete, state the result summary; otherwise continue with the next tool call.")297 ))298 continue299 }300 // A stream cancelled mid-turn surfaces here as an empty301 // "end turn"; that is a cancelled run, not a completed one.302 if Task.isCancelled {303 step.status = .cancelled304 step.finishedAt = Date()305 transcript?.upsert(step: step)306 emit(.stepCompleted(step))307 finish(.cancelled)308 return309 }310 step.status = .completed311 step.finishedAt = Date()312 transcript?.upsert(step: step)313 emit(.stepCompleted(step))314 finish(.completed(finalAnswer: turn.text))315 return316317 case .maxTokens:318 // A truncated turn may carry INCOMPLETE tool calls — never319 // execute them (docs/AGENT-RESEARCH.md §6.2). Nudge instead.320 truncationNudges += 1321 step.status = .completed322 step.finishedAt = Date()323 transcript?.upsert(step: step)324 emit(.stepCompleted(step))325 if truncationNudges > 2 {326 finish(.failed(reason: "The model's output was truncated by its token limit three times in a row — the task cannot proceed. Try a model with a larger output limit or split the task."))327 return328 }329 if !turn.text.isEmpty {330 history.append(HistoryEntry(stepIndex: stepIndex, message: Message(role: .assistant, text: turn.text)))331 }332 history.append(HistoryEntry(333 stepIndex: stepIndex,334 message: Message(role: .user, text: "Your previous response was cut off by the output token limit; any tool calls in it were NOT executed. Continue more concisely, re-issuing the needed tool call(s) one at a time.")335 ))336337 case .refusal:338 step.status = .failed339 step.finishedAt = Date()340 transcript?.upsert(step: step)341 emit(.stepCompleted(step))342 finish(.failed(reason: "The model declined to continue this task\(turn.text.isEmpty ? "." : ": \(turn.text)")"))343 return344 }345 }346 }347348 // MARK: - Model turn streaming349350 /// One accumulated assistant turn.351 private struct ModelTurn {352 var text = ""353 var reasoning = ""354 var toolCalls: [ToolCall] = []355 var usage: TokenUsage?356 var stop = StopReason.endTurn357 }358359 /// Streams one turn, forwarding deltas as AgentEvents. Retries once with360 /// backoff on transient provider errors (network, 429, 5xx).361 private func streamModelTurn(stepID: UUID, systemPrompt: String, toolSpecs: [ToolSpec]) async throws -> ModelTurn {362 var attempt = 0363 while true {364 attempt += 1365 do {366 return try await streamOnce(stepID: stepID, systemPrompt: systemPrompt, toolSpecs: toolSpecs)367 } catch let error as ProviderError where attempt == 1 && Self.isTransient(error) {368 let delay: TimeInterval369 if case .rateLimited(_, let retryAfter) = error, let retryAfter {370 delay = min(retryAfter, 30)371 } else {372 delay = 2373 }374 try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))375 }376 }377 }378379 private static func isTransient(_ error: ProviderError) -> Bool {380 switch error {381 case .networkError, .rateLimited:382 return true383 case .serverError(_, let status, _):384 return status >= 500385 default:386 return false387 }388 }389390 private func streamOnce(stepID: UUID, systemPrompt: String, toolSpecs: [ToolSpec]) async throws -> ModelTurn {391 let request = ChatRequest(392 model: model,393 systemPrompt: systemPrompt,394 messages: history.map(\.message),395 parameters: ChatParameters(),396 stream: true,397 tools: toolSpecs,398 toolChoice: .auto399 )400 var turn = ModelTurn()401 for try await event in client.streamChat(request, apiKey: apiKey) {402 switch event {403 case .textDelta(let delta):404 turn.text += delta405 emit(.textDelta(stepID: stepID, delta))406 case .reasoningDelta(let delta):407 turn.reasoning += delta408 emit(.thinkingDelta(stepID: stepID, delta))409 case .toolCallStarted(let index, let id, let name):410 emit(.toolCallStreaming(stepID: stepID, index: index, id: id, name: name))411 case .toolCallArgumentsDelta(let index, let delta):412 emit(.toolCallArgumentsDelta(stepID: stepID, index: index, delta: delta))413 case .toolCalls(let calls):414 turn.toolCalls = calls415 case .usage(let usage):416 turn.usage = usage417 case .citations:418 break419 case .finished(_, let stop):420 turn.stop = stop421 }422 }423 return turn424 }425426 // MARK: - Tool execution427428 private struct ToolExecutionOutcome {429 var results: [ToolResult] = []430 var invocations: [AgentToolInvocation] = []431 var planChanged = false432 var sawNewToolCall = false433 }434435 /// Executes a turn's tool calls — `update_plan` intercepted internally436 /// (no side effect ⇒ no policy gate), everything else through the437 /// registry (which routes side effects through the PolicyEngine). Results438 /// come back in call order, correlated by id, in ONE tool-results439 /// message. Sequential by default; concurrent only when enabled AND every440 /// call is read-only under the active policy. Throws only on cancellation.441 private func executeToolCalls(442 _ calls: [ToolCall],443 stepIndex: Int,444 stepID: UUID445 ) async throws -> ToolExecutionOutcome {446 var outcome = ToolExecutionOutcome()447448 if configuration.parallelToolCalls, calls.count > 1, await allReadOnly(calls) {449 // Concurrent read-only execution; results re-ordered by call index.450 var startedAt: [String: Date] = [:]451 for call in calls {452 let invocation = AgentToolInvocation(call: call, startedAt: Date())453 startedAt[call.id] = invocation.startedAt454 emit(.toolCallStarted(stepID: stepID, invocation))455 }456 let registry = tools457 let collected: [(Int, ToolResult)] = try await withThrowingTaskGroup(of: (Int, ToolResult).self) { group in458 for (index, call) in calls.enumerated() {459 let context = makeContext(stepID: stepID, callID: call.id)460 group.addTask {461 (index, try await registry.execute(call: call, context: context))462 }463 }464 var results: [(Int, ToolResult)] = []465 for try await item in group { results.append(item) }466 return results.sorted { $0.0 < $1.0 }467 }468 for (index, rawResult) in collected {469 let call = calls[index]470 let result = memory.offloadIfNeeded(rawResult, stepIndex: stepIndex, toolName: call.name)471 outcome.sawNewToolCall = loopGuard.recordInvocation(call: call, isError: result.isError) || outcome.sawNewToolCall472 var invocation = AgentToolInvocation(call: call, result: result, startedAt: startedAt[call.id])473 invocation.finishedAt = Date()474 outcome.results.append(result)475 outcome.invocations.append(invocation)476 emit(.toolCallFinished(stepID: stepID, invocation))477 }478 return outcome479 }480481 for call in calls {482 try Task.checkCancellation()483 var invocation = AgentToolInvocation(call: call, startedAt: Date())484 emit(.toolCallStarted(stepID: stepID, invocation))485486 let result: ToolResult487 if call.name == Planner.toolName {488 let applied = planner.apply(call: call)489 result = applied.result490 outcome.planChanged = outcome.planChanged || applied.changed491 if let plan = planner.plan {492 emit(.planUpdated(plan))493 transcript?.recordPlan(plan, stepIndex: stepIndex)494 }495 } else {496 let context = makeContext(stepID: stepID, callID: call.id)497 let raw = try await tools.execute(call: call, context: context)498 result = memory.offloadIfNeeded(raw, stepIndex: stepIndex, toolName: call.name)499 }500501 outcome.sawNewToolCall = loopGuard.recordInvocation(call: call, isError: result.isError) || outcome.sawNewToolCall502 invocation.result = result503 invocation.finishedAt = Date()504 outcome.results.append(result)505 outcome.invocations.append(invocation)506 emit(.toolCallFinished(stepID: stepID, invocation))507 }508 return outcome509 }510511 private func makeContext(stepID: UUID, callID: String) -> ToolExecutionContext {512 let continuation = eventContinuation513 return ToolExecutionContext(514 workspaceURL: workspace.root,515 policy: policy,516 audit: audit,517 onOutput: { chunk in518 continuation?.yield(.toolOutput(stepID: stepID, callID: callID, chunk))519 },520 workspaceManager: workspace521 )522 }523524 /// True when every call in the batch is safe to run concurrently: the525 /// read-only file tools, or a bash command the policy auto-allows under526 /// the current mode (i.e. its whole payload is read-only). Anything527 /// mutating, plan-touching, or approval-bound runs sequentially.528 private func allReadOnly(_ calls: [ToolCall]) async -> Bool {529 let readOnlyTools: Set<String> = ["read_file", "list_dir", "search_files"]530 for call in calls {531 if readOnlyTools.contains(call.name) { continue }532 if call.name == "bash",533 let command = call.argumentsDictionary?["command"] as? String {534 let ruling = await policy.evaluate(ActionRequest(535 kind: .shellCommand, payload: command, cwd: workspace.root, explanation: nil536 ))537 if case .allow = ruling { continue }538 }539 return false540 }541 return true542 }543544 // MARK: - Progress & pausing545546 /// Cheap signature of the workspace's tracked files (stall detection).547 private func filesSignature() async -> Int {548 var hasher = Hasher()549 for entry in await workspace.files() {550 hasher.combine(entry.path)551 hasher.combine(entry.lastTouched)552 }553 return hasher.finalize()554 }555556 /// Emits the trip, pauses until resume()/stop()/cancel() answers, and557 /// returns true when the run should continue (with raised budgets).558 private func pauseOnTrip(_ trip: LoopGuardTrip) async -> Bool {559 emit(.guardTripped(trip))560 emit(.statusChanged(.awaitingUser))561 let decision = await withCheckedContinuation { (continuation: CheckedContinuation<TripDecision, Never>) in562 tripContinuation = continuation563 }564 switch decision {565 case .resume(let raiseBudget):566 if raiseBudget {567 loopGuard.raiseBudgets()568 }569 emit(.statusChanged(.running))570 return true571 case .stop:572 return false573 }574 }575576 // MARK: - Finishing577578 private func emit(_ event: AgentEvent) {579 eventContinuation?.yield(event)580 }581582 private func finish(_ outcome: AgentRunOutcome) {583 transcript?.finish(outcome: outcome)584 emit(.statusChanged(.finished))585 emit(.runFinished(outcome))586 eventContinuation?.finish()587 eventContinuation = nil588 isRunning = false589 }590}591