Retry a timed-out follow-up whose send then fails; fix the gate after #320

GraphcodeKit/Sources/GraphStore.swift

@@ -2884,7 +2884,7 @@ public actor GraphStore {
         var retained: [PendingFollowUp] = []
         for pending in remaining + Array(batch[index...]) {
           if let confirmed = completedTimedOutDeliveries.removeValue(forKey: pending.id) {
-            if !confirmed { _ = staged(pending) }
+            if !confirmed { retained.append(staged(pending)) }
           } else {
             retained.append(pending)
           }
@@ -2982,6 +2982,9 @@ public actor GraphStore {
         Task { [self] in
           let result = await attempt.wait()
           finishTimedOutDelivery(pending.id, confirmed: result)
+          // A failure learned this late has no command to ride the drain of; without
+          // its own it waits for the next poll to retry, or for good if none comes.
+          if !result { await drainPendingFollowUps() }
         }
         DaemonLog.shared.record(
           "delivery-stall",
@@ -3011,11 +3014,20 @@ public actor GraphStore {
     }
   }
 
+  /// The abandoned send's verdict, whenever it comes. Confirmed means the session got
+  /// the text, late, so the item leaves the queue and is never sent again. Not confirmed
+  /// means the transport failed, and the item is what a fast failure is: staged to memory
+  /// once and kept on the queue, in its place, for the next pass. `staged(_:)` returns
+  /// the recorded copy rather than enqueueing it — dropping that return was how a
+  /// timed-out failure silently left the queue while a prompt one stayed.
   private func finishTimedOutDelivery(_ id: UUID, confirmed: Bool) {
     guard pendingDeliveryAttempts.remove(id) != nil else { return }
     if let index = pendingFollowUps.firstIndex(where: { $0.id == id }) {
-      let pending = pendingFollowUps.remove(at: index)
-      if !confirmed { _ = staged(pending) }
+      if confirmed {
+        pendingFollowUps.remove(at: index)
+      } else {
+        pendingFollowUps[index] = staged(pendingFollowUps[index])
+      }
     } else {
       completedTimedOutDeliveries[id] = confirmed
     }

GraphcodeKit/Sources/ProjectPersistence.swift

@@ -58,13 +58,14 @@ public struct ProjectPersistence: Sendable {
     }
     let roomURL = mailroomURL(forProjectPath: graph.project.path)
     let digest = MailroomDigest(of: graph.mailroom)
-    guard !Self.roomDigests.matches(digest, for: roomURL.path)
-      || !FileManager.default.fileExists(atPath: roomURL.path)
+    guard
+      !Self.roomDigests.matches(digest, for: roomURL.path)
+        || !FileManager.default.fileExists(atPath: roomURL.path)
     else { return }
     if graph.mailroom.isEmpty {
       do {
         try FileManager.default.removeItem(at: roomURL)
-      } catch where !FileManager.default.fileExists(atPath: roomURL.path) {
+      } catch  where !FileManager.default.fileExists(atPath: roomURL.path) {
         Self.roomDigests.set(digest, for: roomURL.path)
       } catch {
         return

graphcode/Tests/DrainWedgeVerificationTests.swift

@@ -162,6 +162,16 @@ struct DeliveryWedgeTests {
     for _ in 0..<2000 where !flag.value { try? await Task.sleep(for: .milliseconds(5)) }
   }
 
+  /// Waits for a delivery to arrive rather than sleeping a fixed span and hoping. The
+  /// deadline these tests drive is 50ms, but the drain that acts on it competes with the
+  /// rest of the suite — a flat sleep passes alone and fails under load, which reads as
+  /// the timeout not working when it is only late.
+  private func settle(until values: LockIsolated<[String]>, reaches count: Int) async {
+    for _ in 0..<2000 where values.value.count < count {
+      try? await Task.sleep(for: .milliseconds(5))
+    }
+  }
+
   /// The control: every send returns, so the bystander is served.
   @Test
   func aBystandersFollowUpIsDeliveredWhenEverySendReturns() async {
@@ -186,6 +196,10 @@ struct DeliveryWedgeTests {
   /// The same wedge as beta5, reached through the send instead of the read: one loop's
   /// `zmx send` never returns and every other loop's staged mail stops moving, silently,
   /// for as long as the process lives.
+  ///
+  /// The hung send eventually *succeeds*: the text was typed, late, so the store must
+  /// count that delivery and never send the message a second time. The closure records
+  /// the late success itself — a hung transport that returns is still the transport.
   @Test
   func aHungDeliveryStillFreezesEveryOtherLoopsFollowUps() async {
     let fixture = fixture()
@@ -194,18 +208,21 @@ struct DeliveryWedgeTests {
     let release = LockIsolated(false)
     let store = GraphStore(
       graph: fixture.graph,
+      // Ahead of the closures because that is where `GraphStore.init` declares it, and
+      // Swift matches an argument list in declaration order.
+      deliveryDeadline: .milliseconds(50),
       onDeliverMessage: { node, message, _ in
         guard node.id == fixture.hung else {
           delivered.withValue { $0.append(message) }
           return true
         }
         entered.setValue(true)
         while !release.value { try? await Task.sleep(for: .milliseconds(5)) }
+        delivered.withValue { $0.append(message) }
         return true
       },
       // Both loops read idle, so the drain tries to deliver to both.
-      onReadPresence: { _, _ in PresenceReading(presence: .idle, confidence: .reported) },
-      deliveryDeadline: .milliseconds(50))
+      onReadPresence: { _, _ in PresenceReading(presence: .idle, confidence: .reported) })
 
     // Queued while both are mid-turn (cached `busy`), then delivered by the drain.
     let wedging = Task {
@@ -215,18 +232,82 @@ struct DeliveryWedgeTests {
     await settle(until: entered)
     #expect(entered.value)
 
-    // The store keeps answering, and the bystander's mail moves after the timeout.
+    // The store keeps answering: the wedged command returns when its send's deadline
+    // passes, closure still hung, and the pass after it serves the bystander — whose
+    // message was queued during that drain and so waits for the next one.
     await store.handle(
       .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true))
-    try? await Task.sleep(for: .milliseconds(100))
+    _ = await wedging.value
     await store.handle(.refreshUsage)
 
     #expect(delivered.value == ["[graphcode] for the bystander"])
-    #expect(!wedging.isCancelled)
 
     release.setValue(true)
+    await settle(until: delivered, reaches: 2)
+    #expect(delivered.value == ["[graphcode] for the bystander", "[graphcode] for the hung one"])
+
+    // Exactly once: the late success took the item off the queue, so a pass finds nothing.
+    await store.handle(.refreshUsage)
+    await store.handle(.refreshUsage)
+    #expect(delivered.value == ["[graphcode] for the bystander", "[graphcode] for the hung one"])
+  }
+
+  /// The other way a hung send can end: it *fails*. `zmx send` finally exits non-zero
+  /// after the deadline passed. On #320's main that verdict staged the message to memory
+  /// and dropped it from the live queue — the target is idle and answering, and the text
+  /// it was owed never reaches its session. A failure is a failure whenever it is
+  /// learned: staged once, retried at the next pass, delivered once.
+  @Test
+  func aTimedOutSendThatThenFailsIsRetriedExactlyOnce() async {
+    let fixture = fixture()
+    let delivered = LockIsolated<[String]>([])
+    let remembered = LockIsolated<[String]>([])
+    let attempts = LockIsolated(0)
+    let entered = LockIsolated(false)
+    let release = LockIsolated(false)
+    let store = GraphStore(
+      graph: fixture.graph,
+      deliveryDeadline: .milliseconds(50),
+      onDeliverMessage: { node, message, _ in
+        guard node.id == fixture.hung else {
+          delivered.withValue { $0.append(message) }
+          return true
+        }
+        let attempt = attempts.withValue { value -> Int in
+          value += 1
+          return value
+        }
+        guard attempt == 1 else {
+          delivered.withValue { $0.append(message) }
+          return true
+        }
+        entered.setValue(true)
+        while !release.value { try? await Task.sleep(for: .milliseconds(5)) }
+        return false
+      },
+      onReadPresence: { _, _ in PresenceReading(presence: .idle, confidence: .reported) },
+      onAppendMemory: { _, entry in remembered.withValue { $0.append(entry) } })
+
+    let wedging = Task {
+      await store.handle(
+        .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true))
+    }
+    await settle(until: entered)
+    await store.handle(
+      .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true))
     _ = await wedging.value
-    try? await Task.sleep(for: .milliseconds(50))
+    await store.handle(.refreshUsage)
+    #expect(delivered.value == ["[graphcode] for the bystander"])
+
+    release.setValue(true)
+    // No command drives this: the store retries on its own once the verdict lands.
+    await settle(until: delivered, reaches: 2)
+    #expect(delivered.value == ["[graphcode] for the bystander", "[graphcode] for the hung one"])
+    #expect(remembered.value.filter { $0.contains("follow-up staged") }.count == 1)
+
+    await store.handle(.refreshUsage)
+    await store.handle(.refreshUsage)
     #expect(delivered.value == ["[graphcode] for the bystander", "[graphcode] for the hung one"])
+    #expect(attempts.value == 2)
   }
 }