fix(ios): prevent duplicate mutations and preserve read retries

.changeset/ios-runner-journal-admission.md

@@ -0,0 +1,5 @@
+---
+'agent-device': patch
+---
+
+Prevent repeated Apple runner interactions with the same command ID from executing twice while allowing read-only retries, including selector queries, to read again. If a repeat cannot replay the finished command's result, return `COMMAND_ALREADY_FINISHED` with `error.details.dispatched: unknown`. Run `snapshot -i` to inspect the screen before deciding whether to send another interaction.

apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests+CommandDispatch.swift

@@ -189,6 +189,11 @@ extension RunnerTests {
     if let unavailable = runnerUnavailableResponse(command: command) {
       return unavailable
     }
+    // Past the last fast refusal, this command's work is going to the device, so its `commandId` is
+    // spent for a non-read command. Finished reads can run again after a lost reply. Only the
+    // `.execute` verdict reaches this line, so a repeat of a spent id is refused at the transport and no
+    // busy refusal here can follow the mark for the command it refuses (ADR 0005).
+    commandJournal.markDispatched(command: command)
     if command.command == .snapshot {
       return try executeSnapshotDispatched(command: command)
     }

apple/runner/AgentDeviceRunner/AgentDeviceRunnerUITests/RunnerTests+CommandJournal.swift

@@ -10,61 +10,149 @@ enum RunnerCommandLifecycleState: String {
   case failed
 }
 
+extension RunnerCommandLifecycleState {
+  /// Whether execution has ended for an entry in this state. `notAccepted` is never written to an
+  /// entry — `status` synthesizes it for an id the journal does not hold — so it is not terminal.
+  var isTerminal: Bool {
+    self == .completed || self == .failed
+  }
+}
+
 struct RunnerCommandJournalEntry {
   let commandId: String
   let command: String
   var state: RunnerCommandLifecycleState
   var responseOk: Bool?
   var responseJson: String?
   var error: ErrorPayload?
+  /// Whether any of this command's work reached the device. The runner refuses a command whose main
+  /// thread is still draining abandoned work before running anything, so such a refusal left nothing
+  /// on the device and its terminal entry must not close the id: the read-only busy resend carries the
+  /// same `commandId` (ADR 0005).
+  var dispatched = false
+  /// Distinguishes joiners when the same id is claimed again before an earlier delivery finishes.
+  let executionToken: Int
+  let isReadOnly: Bool
+}
+
+/// The journal's verdict on one command that carries a `commandId`.
+enum RunnerCommandAdmission {
+  /// The caller owns this execution,
+  /// identified by `executionToken` for as long as that claim answers for it.
+  case execute(executionToken: Int)
+  /// An execution is claimed and has not finished: the caller runs nothing and is answered by that
+  /// execution, the one `executionToken` names. `accept(command:registeringJoin:)` has already
+  /// registered the send it gave this verdict.
+  case join(executionToken: Int)
+  /// Replays the recorded outcome without executing the command.
+  case terminal(outcome: Response)
 }
 
+/// The runner's one command-lifecycle record: what it holds for a `commandId`, whether that id may run,
+/// and what `status` answers for it. The transport asks this type before enqueuing anything, so one
+/// mutation id stays spent after execution, while a finished read can execute again after a lost reply.
+/// `closedCommandIdSafetyValve` releases the oldest marker; `pruneIfNeeded` states its bound.
 final class RunnerCommandJournal {
   private let lock = NSLock()
   private let maxEntries = 64
+  /// Memory safety valve on closed-id retention, not an admission boundary: the journal keeps a closed
+  /// mutation id until its marker is evicted. It is 64x the entry
+  /// window because a marker holds no outcome body, only the id. Tests read it rather than repeat it.
+  static let closedCommandIdSafetyValve = 4096
   private let maxResponseJsonBytes = 16 * 1024
   private var entries: [String: RunnerCommandJournalEntry] = [:]
   private var order: [String] = []
+  /// Ids whose dispatched execution closed and whose entry has since been pruned: the admission half of
+  /// what the entry recorded, kept without a body so bounded memory cannot make a spent id executable
+  /// again. It settles admission only. `status` still answers `notAccepted` for such an id, because the
+  /// outcome the entry held is gone and only the entry can report one. Retention ends only at the safety
+  /// valve above: an id older than that can be admitted again, and `pruneIfNeeded` states that bound.
+  private var closedCommandIds: Set<String> = []
+  private var closedOrder: [String] = []
+  /// Mints the identity of each claim, so a delivery can tell the sends that joined its own execution from
+  /// the sends that joined a later one under the same `commandId`.
+  private var nextExecutionToken = 0
 
-  func accept(command: Command) {
-    guard let commandId = command.commandId?.trimmedNonEmpty else { return }
+  /// Settles one send, and when the answer is to join a running execution, registers that send for the
+  /// answer inside the same locked operation. No transition can close the entry between the two, which is
+  /// what keeps a joined send from waiting on a delivery that already went out. `register` runs on the
+  /// calling thread while the journal's lock is held: it may take a lock of its own, and must not call
+  /// back into this journal. The wire route passes one; a caller that only reads the verdict — a journal
+  /// test, or a probe that will not itself answer for the execution — leaves it unregistered and must not
+  /// treat a `.join` verdict as a send that will be answered.
+  @discardableResult
+  func accept(
+    command: Command,
+    registeringJoin register: (Int) -> Void = { _ in }
+  ) -> RunnerCommandAdmission {
     lock.lock()
     defer { lock.unlock() }
-    entries[commandId] = RunnerCommandJournalEntry(
-      commandId: commandId,
-      command: command.command.rawValue,
-      state: .accepted,
-      responseOk: nil,
-      responseJson: nil,
-      error: nil
-    )
-    order.removeAll { $0 == commandId }
-    order.append(commandId)
-    pruneIfNeeded()
+    guard let commandId = command.commandId?.trimmedNonEmpty else {
+      // No `commandId`, so nothing to settle and no other send can share this claim.
+      return .execute(executionToken: mintExecutionToken())
+    }
+    guard let entry = entries[commandId] else {
+      // The entry can be pruned; the fact it recorded is not, so an id that spent itself on the device
+      // never executes again. Only the entry named the command, so this refusal can name only what the
+      // repeat asked for.
+      if closedCommandIds.contains(commandId) {
+        // The record that named this id is gone, so all this refusal can promise is that the id is
+        // spent — it is no command's answer, and so never the runner's own.
+        return .terminal(
+          outcome: alreadyFinishedRefusal(command: command, commandId: commandId, outcomeDropped: true)
+        )
+      }
+      return claim(command: command, commandId: commandId)
+    }
+    if !entry.state.isTerminal {
+      let executionToken = entry.executionToken
+      register(executionToken)
+      return .join(executionToken: executionToken)
+    }
+    // Reads execute again after finishing, including after a lost reply or a recorded failure.
+    // A pre-dispatch refusal also leaves the id reusable. Classification belongs to the recorded
+    // command: renaming a spent mutation to a read must not bypass its admission history.
+    guard entry.dispatched, !entry.isReadOnly else {
+      return claim(command: command, commandId: commandId)
+    }
+    return .terminal(outcome: recordedOutcome(for: entry))
+  }
+
+  /// Callers hold the journal's lock, including for commands without an id.
+  private func mintExecutionToken() -> Int {
+    nextExecutionToken += 1
+    return nextExecutionToken
   }
 
   func start(command: Command) {
-    update(command: command, state: .started, responseOk: nil, responseJson: nil, error: nil)
+    transition(command: command, reorders: true) { entry in
+      entry.state = .started
+    }
   }
 
+  /// Records the outcome that ended this command's execution: the state its `ok` implies, the response
+  /// body where the command's trait retains a replayable one, and its error. Whether the id this record
+  /// closes is spent is the separate `dispatched` mark `markDispatched` writes.
   func finish(command: Command, response: Response) {
-    update(
-      command: command,
-      state: response.ok ? .completed : .failed,
-      responseOk: response.ok,
-      responseJson: encodeResponseJson(command: command, response: response),
-      error: response.error
-    )
+    transition(command: command, reorders: true) { entry in
+      entry.state = response.ok ? .completed : .failed
+      entry.responseOk = response.ok
+      entry.responseJson = encodeResponseJson(command: command, response: response)
+      entry.error = response.error
+    }
   }
 
+  /// Records the failure that ended this command's execution. The typed wire code and the hint go with it,
+  /// the same error body the transport puts on the live reply for the same throw: a repeat of this
+  /// `commandId` is answered from this record, and a failure that arrives without its code reads to the
+  /// daemon as an ordinary served refusal — which for a `MAIN_THREAD_TIMEOUT` would report the main thread
+  /// drained when the command that just timed out is what occupied it.
   func fail(command: Command, error: Error) {
-    update(
-      command: command,
-      state: .failed,
-      responseOk: nil,
-      responseJson: nil,
-      error: ErrorPayload(message: error.localizedDescription)
-    )
+    let payload = RunnerTests.thrownErrorPayload(for: error)
+    transition(command: command, reorders: true) { entry in
+      entry.state = .failed
+      entry.error = payload
+    }
   }
 
   func status(normalizedCommandId commandId: String) -> DataPayload {
@@ -89,39 +177,133 @@ final class RunnerCommandJournal {
     )
   }
 
-  private func update(
-    command: Command,
-    state: RunnerCommandLifecycleState,
-    responseOk: Bool?,
-    responseJson: String?,
-    error: ErrorPayload?
-  ) {
-    guard let commandId = command.commandId?.trimmedNonEmpty else { return }
-    lock.lock()
-    defer { lock.unlock() }
-    var entry = entries[commandId] ?? RunnerCommandJournalEntry(
+  /// Writes the claim for one execution. Callers hold the lock.
+  private func claim(command: Command, commandId: String) -> RunnerCommandAdmission {
+    let executionToken = mintExecutionToken()
+    entries[commandId] = RunnerCommandJournalEntry(
       commandId: commandId,
       command: command.command.rawValue,
       state: .accepted,
       responseOk: nil,
       responseJson: nil,
-      error: nil
+      error: nil,
+      executionToken: executionToken,
+      isReadOnly: command.traits.isReadOnly
     )
-    entry.state = state
-    entry.responseOk = responseOk
-    entry.responseJson = responseJson
-    entry.error = error
+    order.removeAll { $0 == commandId }
+    order.append(commandId)
+    pruneIfNeeded()
+    return .execute(executionToken: executionToken)
+  }
+
+  /// Records that this command's work was handed to the device, as opposed to being refused while it
+  /// was still queued. Until this point a refusal owes the caller a retry of the same `commandId`. Only
+  /// the mark moves: the entry keeps its place in the retention order, which is claim order, and the
+  /// state `start` already wrote is what `status` reports for a running command.
+  func markDispatched(command: Command) {
+    transition(command: command, reorders: false) { entry in
+      entry.dispatched = true
+    }
+  }
+
+  /// Applies one transition to the record this command's id holds. A transition for an id the journal does
+  /// not hold describes work it never claimed, and one for an entry that already finished restates a
+  /// closed outcome: neither may open a record again. Re-opening one would report `accepted` or `started`
+  /// for work that is not running, and the daemon holds that verdict as an in-flight exchange on a runner
+  /// it would otherwise hand off. `reorders` is what settling does: an entry that finishes goes to the back
+  /// of the retention order, which records claim order.
+  private func transition(
+    command: Command,
+    reorders: Bool,
+    _ mutate: (inout RunnerCommandJournalEntry) -> Void
+  ) {
+    guard let commandId = command.commandId?.trimmedNonEmpty else { return }
+    lock.lock()
+    defer { lock.unlock() }
+    guard var entry = entries[commandId], !entry.state.isTerminal else { return }
+    mutate(&entry)
     entries[commandId] = entry
+    guard reorders else { return }
     order.removeAll { $0 == commandId }
     order.append(commandId)
     pruneIfNeeded()
   }
 
+  /// Drops the oldest closed entries, keeping each one's id as the fact that settles admission for it.
+  /// An entry whose execution is claimed but not finished is never dropped, however deep the backlog:
+  /// `status` answers `notAccepted` only for an id this journal does not hold, and the daemon cannot read
+  /// that as a safe terminal state — it would invalidate a session whose runner is still executing the
+  /// command it lost the answer to.
   private func pruneIfNeeded() {
     while order.count > maxEntries {
-      let removed = order.removeFirst()
-      entries.removeValue(forKey: removed)
+      guard let removable = order.first(where: { entries[$0]?.state.isTerminal == true }) else {
+        return
+      }
+      order.removeAll { $0 == removable }
+      // Only dispatched non-read commands spend their ids. Reads and pre-dispatch refusals leave
+      // no marker, so resends can execute after the outcome is pruned (ADR 0005).
+      if let entry = entries[removable], entry.dispatched, !entry.isReadOnly {
+        closedCommandIds.insert(removable)
+        closedOrder.append(removable)
+        while closedOrder.count > Self.closedCommandIdSafetyValve {
+          // The safety valve is the one place a spent id returns to circulation: a resend more than
+          // 4096 completed dispatches later, within this runner process, can re-execute. A runner
+          // restart starts a new journal, which answers `notAccepted` for every id.
+          closedCommandIds.remove(closedOrder.removeFirst())
+        }
+      }
+      entries.removeValue(forKey: removable)
+    }
+  }
+
+  /// The answer owed a repeat of a finished `commandId`: the outcome the runner kept, then the failure it
+  /// kept, and only then a typed refusal for a command whose own classification retains no body at all.
+  private func recordedOutcome(for entry: RunnerCommandJournalEntry) -> Response {
+    if let responseJson = entry.responseJson,
+      let recorded = try? JSONDecoder().decode(Response.self, from: Data(responseJson.utf8))
+    {
+      return recorded
     }
+    if let error = entry.error {
+      return Response(ok: false, error: error)
+    }
+    return alreadyFinishedRefusal(
+      commandId: entry.commandId,
+      commandName: entry.command,
+      outcomeDropped: false
+    )
+  }
+
+  private func alreadyFinishedRefusal(
+    command: Command,
+    commandId: String,
+    outcomeDropped: Bool
+  ) -> Response {
+    alreadyFinishedRefusal(
+      commandId: commandId,
+      commandName: command.command.rawValue,
+      outcomeDropped: outcomeDropped
+    )
+  }
+
+  private func alreadyFinishedRefusal(
+    commandId: String,
+    commandName: String,
+    outcomeDropped: Bool
+  ) -> Response {
+    let reason =
+      outcomeDropped
+      ? "The runner already finished \(commandName) with id \(commandId). That command's recorded outcome has since been dropped from the journal, so there is no response to replay."
+      : "The runner already finished \(commandName) with id \(commandId) and kept no response to replay."
+    return Response(
+      ok: false,
+      error: ErrorPayload(
+        code: RunnerTests.RunnerWireErrorCode.commandAlreadyFinished,
+        message: reason,
+        hint:
+          "Read the screen with snapshot before sending anything else: the finished command's work reached the device, so it may already have had its effect."
+      )
+    )
   }
 
   private func encodeResponseJson(command: Command, response: Response) -> String? {