From b33c8c0ac7cbd7794d8469dab29152811600b4c1 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 18:48:25 -0700 Subject: [PATCH 1/6] wip: claim branch for #311/#304 drain ordering + presence timeout Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014D7evLSBWCnNeumHAWHH5S From 5e774758d62cd006010328910efd9c848e527eee Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 19:10:15 -0700 Subject: [PATCH 2/6] Bound the presence read, and let the drain alone decide delivery (#311, #304) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four defects on one seam — how a staged follow-up decides when to be delivered — with a test each, because a combined test that passes says nothing about which half works. **The presence read had no deadline.** `onReadPresence` reaches `PTYProcessSession.waitCollectingOutput`, which ends only when the probe's `terminationHandler` closes the stream, and `ssh`'s `ConnectTimeout=10` bounds the connect rather than a command hanging on a host that has gone away. One such read held `isDrainingFollowUps` for the life of the daemon and froze staged delivery for every loop in the project — and a frozen queue reports exactly what an empty one reports. Every reading in the store now goes through one bounded `presenceReading(of:)`; a read that runs out of time is `.unknown`, never a state, which is the distinction #286 established for a `zmx` probe that could not run. An unknown target keeps its message queued and is asked again next pass: not delivered blindly, not busy for ever. **Two readings decided one question.** `deliversLater` consulted the cached `node.presence` the poll last wrote while the drain took a live one, so a follow-up staged afterwards was typed in ahead of a queue still being worked through. Measured on the live 0.1.64-beta5 daemon: ten messages drained 77–147s late while six sent afterwards arrived in 0.5s. `deliversLater` no longer reads presence at all — every live target's follow-up joins the queue and the drain is the single decision point. `handle` ends in a drain, so nothing waits longer for it. **A single drain reordered.** The drain re-read presence between items, so a turn ending mid-batch sent [2, 3, 1] with no overlapping drain anywhere — which is why #309's non-reentrancy guard cannot see it. The reading is now taken once per target per pass. That also ends a rate nobody had filed: delivering into a session is what makes it busy, so a re-reading drain stopped after the first delivery of every pass and waited for the next tick — ten staged messages took 147 seconds. A backlog now goes out in one pass. **Re-scoping a watch kept the old topic's wakes.** `mail watch --on --topic other` is `--off` for the topic being left; its staged wakes are dropped, and the drain owes a wake only under the watch standing now. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014D7evLSBWCnNeumHAWHH5S --- GraphcodeKit/Sources/Deadline.swift | 51 +++ GraphcodeKit/Sources/GraphStore.swift | 129 +++++--- graphcode/Tests/FollowUpQueueOrderTests.swift | 300 ++++++++++++++++++ 3 files changed, 444 insertions(+), 36 deletions(-) create mode 100644 GraphcodeKit/Sources/Deadline.swift create mode 100644 graphcode/Tests/FollowUpQueueOrderTests.swift diff --git a/GraphcodeKit/Sources/Deadline.swift b/GraphcodeKit/Sources/Deadline.swift new file mode 100644 index 00000000..56ffb64c --- /dev/null +++ b/GraphcodeKit/Sources/Deadline.swift @@ -0,0 +1,51 @@ +import Foundation + +/// Runs `operation` with a deadline, answering `nil` when it has not finished in time. +/// +/// The slower of the two is **abandoned**, not cancelled and waited for, and that is the +/// whole point. A structured `withTaskGroup` race cannot bound anything here: the group +/// waits for every child before it returns, so a child that never finishes takes the +/// timeout down with it. The reads this exists for are exactly that kind — a presence +/// probe reaches `PTYProcessSession.waitCollectingOutput`, which ends only when the +/// child's `terminationHandler` closes the stream, and `ssh`'s `ConnectTimeout` bounds +/// the *connect*, never a command already running on a host that has gone away +/// (`RemoteEnsureGate` documents the same wedge from the other side). +/// +/// Cancelling the loser is therefore hygiene rather than the mechanism: a task blocked +/// on a pipe that will never close ignores it. What makes this safe is that nothing +/// awaits the abandoned task, so it costs one suspended task until its subprocess is +/// reaped, and the caller is already gone. +func withDeadline( + _ deadline: Duration, _ operation: @escaping @Sendable () async -> T +) async -> T? { + let relay = DeadlineRelay() + let work = Task.detached { await relay.settle(await operation()) } + let timer = Task.detached { + try? await Task.sleep(for: deadline) + await relay.settle(nil) + } + let answer = await relay.wait() + work.cancel() + timer.cancel() + return answer +} + +/// Whichever of the two tasks arrives first wins; the other's answer is dropped rather +/// than resuming a continuation twice. +private actor DeadlineRelay { + private var answer: T?? + private var waiter: CheckedContinuation? + + func settle(_ value: T?) { + guard answer == nil else { return } + answer = .some(value) + guard let waiter else { return } + self.waiter = nil + waiter.resume(returning: value) + } + + func wait() async -> T? { + if let answer { return answer } + return await withCheckedContinuation { self.waiter = $0 } + } +} diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index d9dc444c..9cffb9ac 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -192,6 +192,23 @@ public actor GraphStore { /// queued in between (issue #304: duplicates, lost mail, and out-of-order delivery). private var isDrainingFollowUps = false + /// How long a presence read may take before the store stops waiting on it. + /// + /// Every reading in this file runs through `presenceReading(of:)`, and nothing in the + /// chain below it has a deadline of its own: `onReadPresence` reaches + /// `PTYProcessSession.waitCollectingOutput`, which ends only when the probe's + /// `terminationHandler` closes the stream, and `ssh`'s `ConnectTimeout=10` bounds the + /// connect rather than a command left hanging on a host that has gone away. An `await` + /// that never returns held `isDrainingFollowUps` for the life of the daemon, which + /// froze staged delivery for *every* loop in the project — and a frozen queue and an + /// empty one report exactly the same thing from outside (issue #311). + /// + /// Longer than any healthy read: a remote probe is three attempts at + /// `ConnectTimeout=10` with 1s and 2s of backoff between them (`RemoteEnsureGate`), + /// so a live-but-slow host still answers inside this. Short enough that a wedge costs + /// one poll, not the process. + private let presenceReadDeadline: Duration + /// A poller holds `self` weakly, so a store going away already stops it *doing* /// anything — but the task itself keeps sleeping in its loop forever. Harmless for /// the long-lived project store; sub-graph stores are built per command and hold no @@ -263,6 +280,7 @@ public actor GraphStore { onMailroomEnabled: (@Sendable () -> Bool)? = nil, goalCache: GoalEvaluationCache? = nil, recurrence: RecurrenceSink? = nil, + presenceReadDeadline: Duration = .seconds(45), subGraphDepth: Int = 0 ) { self.graph = graph @@ -294,6 +312,24 @@ public actor GraphStore { self.onMailroomEnabled = onMailroomEnabled self.goalCache = goalCache ?? GoalEvaluationCache() self.recurrence = recurrence + self.presenceReadDeadline = presenceReadDeadline + } + + /// The store's one way to ask what a session is doing, and the only place the answer + /// is bounded. `nil` means there is no reader wired at all — every caller then falls + /// back to whatever the graph already believes. + /// + /// A read that runs out of time is `.unknown`, never a state. That is the distinction + /// issue #286 established when a `zmx` probe that could not run was being read as + /// `.absent`: a probe that did not complete says nothing about the session, so a + /// caller must not turn it into a verdict. Here it means a staged message stays + /// staged and is tried again, rather than being delivered blindly into a session that + /// may be mid-turn or held back for ever as if the target were busy. + private func presenceReading(of node: LoopNode) async -> PresenceReading? { + guard let onReadPresence else { return nil } + let path = graph.project.path + return await withDeadline(presenceReadDeadline) { await onReadPresence(node, path) } + ?? .unknown } private func recordMemory(_ nodeID: UUID, _ entry: String) { @@ -1172,10 +1208,10 @@ public actor GraphStore { /// telling every client the graph moved when nothing did. @discardableResult private func refreshPresence() async -> Bool { - guard let onReadPresence else { return false } + guard onReadPresence != nil else { return false } var changed = false for node in graph.nodes where !node.isResolved { - let reading = await onReadPresence(node, graph.project.path) + guard let reading = await presenceReading(of: node) else { continue } guard graph.nodes[id: node.id]?.presence != reading else { continue } graph.nodes[id: node.id]?.presence = reading changed = true @@ -1831,7 +1867,18 @@ public actor GraphStore { announceError("mail watch refused: an empty topic is no topic — omit it") return } - graph.nodes[id: watcherID]?.mailroomWatch = MailroomWatch(topic: trimmed) + let watch = MailroomWatch(topic: trimmed) + graph.nodes[id: watcherID]?.mailroomWatch = watch + // Re-scoping is `--off` for the topic being left: a watch is one subscription, so + // the wakes staged under the old topic are for posts this loop is no longer + // asking about and arriving minutes later is the same defect turning the watch + // off had (issue #304). A wake that still matches the new scope stays queued, and + // a peer's `--follow-up` message is not a wake and is untouched. + pendingFollowUps.removeAll { pending in + guard pending.nodeID == watcherID, let postID = pending.watchedPostID else { return false } + guard let post = graph.mailroom.first(where: { $0.id == postID }) else { return true } + return !watch.matches(post.topic) + } recordMemory( watcherID, "mailroom: now watching \(trimmed.map { "'\($0)'" } ?? "all posts")") } else { @@ -2151,8 +2198,7 @@ public actor GraphStore { return false } if RemoteProjectLocation.parse(projectPath: graph.project.path) != nil { - guard let onReadPresence else { return true } - let reading = await onReadPresence(node, graph.project.path) + guard let reading = await presenceReading(of: node) else { return true } if reading.presence == .absent { return true } recordMemory( nodeID, @@ -2703,14 +2749,21 @@ public actor GraphStore { /// staged, exactly as before. static let respawnedSessionSettle: Duration = .seconds(3) - /// Whether a follow-up to this node should wait rather than type now. Mid-turn and - /// mid-check both qualify — deferring to a busy agent is the flag's entire meaning — - /// while an idle session gets ordinary immediate delivery and a dead one gets the + /// Whether a follow-up to this node joins the queue rather than being typed now. + /// Every live target does; only one the bus cannot reach at all falls through to the /// ordinary staged-to-memory path. + /// + /// This used to answer from the **cached** `node.presence` the poll last wrote, while + /// the drain took a live reading of its own — two readings of one question, and a + /// follow-up staged while they disagreed was typed in ahead of three already queued + /// (issue #311). Arrival order is the only promise a queue makes, so the drain's + /// reading is now the single decision point and this one has no presence in it. + /// Nothing waits longer for it: `handle` ends in a drain, so a follow-up to an idle + /// target is queued and handed over inside the same call — behind whatever was + /// already queued for that target, which is the whole difference. private func deliversLater(to target: LoopNode) -> Bool { switch MessageBus.deliverability(to: target) { - case .targetBusyWithACheck: return true - case nil: return target.presence?.presence != .idle + case .targetBusyWithACheck, nil: return true default: return false } } @@ -2729,13 +2782,28 @@ public actor GraphStore { let batch = pendingFollowUps pendingFollowUps = [] var remaining: [PendingFollowUp] = [] + // One reading per target per pass, taken the first time this pass reaches that + // target and reused for the rest of its queue. Reading again between items is what + // let a turn ending *mid-drain* reorder the queue: three messages for one loop went + // out as [2, 3, 1] because the first was read while the session was still busy and + // the next two after it went idle. One drain produces that on its own, so #309's + // non-reentrancy guard cannot see it — and it is the [#357-before-#347] ordering + // issue #304 was filed for. A reading taken once cannot disagree with itself, so + // the batch either goes out in order or waits together for the next pass. It also + // costs one probe per target rather than one per message. + var readings: [UUID: Presence] = [:] for pending in batch { guard let node = graph.nodes[id: pending.nodeID], !node.isResolved else { continue } - // A watcher's wake is only owed while the watch stands and the post is unread: - // `mail watch --off` after the wake was staged, or an inbox that has since read - // past the post, means the wake has nothing left to say. + // A watcher's wake is only owed while the watch that asked for it stands and the + // post is unread: `mail watch --off` after the wake was staged, an inbox that has + // since read past the post, or a watch re-scoped to another topic — the wakes the + // abandoned topic staged are not this watcher's mail any more — all mean the wake + // has nothing left to say. A post the room has since evicted has nothing to point + // at either. if let postID = pending.watchedPostID { - guard node.mailroomWatch != nil, (node.lastMailroomRead ?? 0) < postID else { continue } + guard let watch = node.mailroomWatch, (node.lastMailroomRead ?? 0) < postID, + let post = graph.mailroom.first(where: { $0.id == postID }), watch.matches(post.topic) + else { continue } } switch MessageBus.deliverability(to: node) { case .targetBusyWithACheck: @@ -2746,12 +2814,16 @@ public actor GraphStore { case nil: break } - let presence: Presence? - if let onReadPresence { - presence = await onReadPresence(node, graph.project.path).presence + let presence: Presence + if let known = readings[pending.nodeID] { + presence = known } else { - presence = node.presence?.presence + presence = + await presenceReading(of: node)?.presence ?? node.presence?.presence ?? .unknown + readings[pending.nodeID] = presence } + // `.idle` and nothing else, which is what makes a timed-out read safe: `.unknown` + // is not a state, so the message stays queued for a pass that gets an answer. guard presence == .idle else { remaining.append(pending) continue @@ -3023,12 +3095,7 @@ public actor GraphStore { // the tree moves: re-delivering every poll would be a full agent turn a minute, // the unbounded spend the failure-tail dedup exists to prevent. if let fingerprint, goalCache.fingerprint(for: nodeID) == fingerprint { - let presence: Presence? - if let onReadPresence { - presence = await onReadPresence(node, graph.project.path).presence - } else { - presence = node.presence?.presence - } + let presence = await presenceReading(of: node)?.presence ?? node.presence?.presence // A nil presence stays skipped: the relay only ever tells a session it can see // idle, so falling through would pay the predicate's price for a wake that can // never land. Such a loop's exits are its stall bound and its human. @@ -3117,12 +3184,7 @@ public actor GraphStore { ) async { let tail = outcome.outputTail.trimmingCharacters(in: .whitespacesAndNewlines) guard !tail.isEmpty, goalCache.feedback(for: node.id) != tail else { return } - let presence: Presence? - if let onReadPresence { - presence = await onReadPresence(node, graph.project.path).presence - } else { - presence = node.presence?.presence - } + let presence = await presenceReading(of: node)?.presence ?? node.presence?.presence guard presence == .idle, MessageBus.deliverability(to: node) == nil else { return } let message = "[graphcode] Goal not met yet: `\(predicate)` still exits non-zero. " @@ -3220,12 +3282,7 @@ public actor GraphStore { guard node.backend.capabilities.supportsDaemonRecurrence || onHeartbeatEnabled?() == true else { return } guard MessageBus.deliverability(to: node) == nil else { return } - let presence: Presence? - if let onReadPresence { - presence = await onReadPresence(node, graph.project.path).presence - } else { - presence = node.presence?.presence - } + let presence = await presenceReading(of: node)?.presence ?? node.presence?.presence guard presence != .busy else { return } let task = node.heartbeatTask ?? "" _ = await deliverToSession( diff --git a/graphcode/Tests/FollowUpQueueOrderTests.swift b/graphcode/Tests/FollowUpQueueOrderTests.swift new file mode 100644 index 00000000..dbedfc15 --- /dev/null +++ b/graphcode/Tests/FollowUpQueueOrderTests.swift @@ -0,0 +1,300 @@ +import ComposableArchitecture +import Foundation +import GraphcodeKit +import MailroomKit +import Testing + +#if canImport(Darwin) + import Darwin +#endif + +/// The seam #309 left behind: how a staged follow-up decides *when* to be delivered. +/// +/// One test per defect, each failing on the head before this change for its own reason — +/// a combined test that passes says nothing about which half works: +/// +/// - `aHungPresenceReadDoesNotStopDeliveryToOtherLoops` — #311, the unbounded read. +/// - `tenStagedAndSixLaterMessagesArriveInSendOrder` — #311, the two readings. +/// - `aTurnEndingMidDrainDoesNotReorderTheQueue` — #304 residual, ordering. +/// - `reScopingAWatchDropsTheAbandonedTopicsStagedWakes` — #304 residual, re-scoping. +/// - `aBacklogDrainsInOnePassRatherThanOnePerTick` — the rate the field measurement found. +@Suite +struct FollowUpQueueOrderTests { + /// What each target's session appears to be doing. Every knob is set by the test + /// before the pass it applies to, so nothing here depends on how many times the store + /// happens to ask — which is the difference between the two heads this must separate. + private actor Readings { + private var fallback: Presence = .busy + private var first: Presence? + private var reads = 0 + private var delay: Duration = .zero + private var hang: Duration = .zero + private var hung: UUID? + + func set(_ presence: Presence, delay: Duration = .zero) { + fallback = presence + first = nil + reads = 0 + self.delay = delay + } + + /// The turn boundary: the pass's first reading says `first`, everything after it + /// says `rest`. A drain that reads once per target per pass sees only `first`; one + /// that reads per item sees the session change under it mid-batch. + func firstReadThen(_ first: Presence, _ rest: Presence) { + self.first = first + fallback = rest + reads = 0 + } + + func hang(_ node: UUID, for duration: Duration) { + hung = node + hang = duration + } + func stopHanging() { hung = nil } + + func read(_ node: UUID) async -> PresenceReading { + if node == hung { try? await Task.sleep(for: hang) } + if delay > .zero { try? await Task.sleep(for: delay) } + reads += 1 + let presence = reads == 1 ? (first ?? fallback) : fallback + return PresenceReading(presence: presence, confidence: .reported) + } + } + + /// A connection with a channel lifecycle of its own and a reader draining it, so the + /// presence poll runs — the rig `FollowUpDrainTests` needs, for the same reason. + private final class Attachment: @unchecked Sendable { + let daemonEnd: Int32 + private let peer: Int32 + private let drainer: Task + init() { + var pair: [Int32] = [0, 0] + _ = socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) + daemonEnd = pair[0] + peer = pair[1] + let reading = pair[1] + drainer = Task.detached { + var sink = [UInt8](repeating: 0, count: 65536) + while recv(reading, &sink, sink.count, 0) > 0 {} + } + } + deinit { + OutboundChannels.close(daemonEnd) + close(peer) + drainer.cancel() + } + } + + private func makeStore( + readings: Readings, delivered: LockIsolated<[String]>, attachment: Attachment, + titles: [String] = ["Target", "Sender"], deadline: Duration = .seconds(45) + ) async -> (GraphStore, ids: [UUID]) { + let store = GraphStore( + onEnsureSession: { _, _ in }, + onDeliverMessage: { _, message, _ in + delivered.withValue { $0.append(message) } + return true + }, + onReadPresence: { node, _ in await readings.read(node.id) }, + onMailroomEnabled: { true }, + presenceReadDeadline: deadline) + for title in titles { + await store.handle( + .createNode(NodeDraft(title: title, loopType: .turnBased, firstInstruction: "Work"))) + } + await store.addConnection(id: UUID(), fileDescriptor: attachment.daemonEnd) + // One poll so every target carries a `busy` reading: that is what stages a follow-up + // rather than typing it in, on this head and on the one before it. + await store.pollPresence() + return (store, await store.graph.nodes.map(\.id)) + } + + /// Settles the store — every command ends in the drain — without refreshing presence + /// on the way, so the drain's own readings are the only ones the test's script sees. + /// `.refreshUsage` polls presence first, which is right for the poll and wrong here. + private func settle(_ store: GraphStore, _ nodeID: UUID) async { + await store.handle(.memoNode(nodeID, text: "settle", from: nodeID)) + } + + private func plain(_ delivered: [String]) -> [String] { + delivered.map { $0.replacingOccurrences(of: "[graphcode] Sender: ", with: "") } + } + + /// Issue #311, the presence read with no deadline — the blocker, because a wedged + /// drain and an empty one report exactly the same thing from outside. + /// + /// `onReadPresence` reaches `PTYProcessSession.waitCollectingOutput`, which ends only + /// when the probe's `terminationHandler` closes the stream; `ssh`'s `ConnectTimeout` + /// bounds the connect, not a command hanging on a host that has gone away. One such + /// read held `isDrainingFollowUps` for the life of the daemon, so mail for **every + /// other loop in the project** stopped — which is what this measures: not that the + /// wedged loop recovers, but that its neighbour is not taken down with it. + /// + /// Fails before the fix on elapsed time: the pass waits out the hung read. + @Test + func aHungPresenceReadDoesNotStopDeliveryToOtherLoops() async { + let readings = Readings() + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, ids) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment, + titles: ["Wedged", "Neighbour", "Sender"], deadline: .milliseconds(200)) + let (wedged, neighbour, sender) = (ids[0], ids[1], ids[2]) + + await store.handle( + .messageNode(wedged, text: "for the wedged loop", from: sender, followUp: true)) + await store.handle( + .messageNode(neighbour, text: "for the neighbour", from: sender, followUp: true)) + #expect(delivered.value.isEmpty) + + // The wedged loop's session stops answering; every other read is instant and idle. + await readings.hang(wedged, for: .seconds(5)) + await readings.set(.idle) + let started = Date() + await store.handle(.refreshUsage) + let elapsed = Date().timeIntervalSince(started) + + #expect(plain(delivered.value) == ["for the neighbour"]) + #expect(elapsed < 2) + // Not delivered blindly either: a read that ran out of time is `.unknown`, which is + // no state at all, so the message is still owed rather than spent. + await readings.stopHanging() + await store.handle(.refreshUsage) + #expect(plain(delivered.value) == ["for the neighbour", "for the wedged loop"]) + } + + /// Issue #311, the two readings — and the shape `MailDeliveryCheck` measured against + /// the live 0.1.64-beta5 daemon: ten messages sat staged and drained 77 to 147 seconds + /// later while six sent afterwards arrived within half a second, because those six + /// took the immediate path off a cached `idle` while the queue was still being worked + /// through. Order was perfect within each path and meaningless across them. + /// + /// Reproduced the same way here: the poll writes `idle` into the cache, and the later + /// messages arrive while that poll's own drain is still running. + /// + /// Fails before the fix on order — the six overtake the ten. + @Test + func tenStagedAndSixLaterMessagesArriveInSendOrder() async { + let readings = Readings() + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, ids) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + let (target, sender) = (ids[0], ids[1]) + for index in 1...10 { + await store.handle( + .messageNode(target, text: "staged \(index)", from: sender, followUp: true)) + } + #expect(delivered.value.isEmpty) + + // Idle now, and slow to answer: the poll's refresh writes `idle` into the cache + // inside the first 400ms, and its drain is still going at 600ms. + await readings.set(.idle, delay: .milliseconds(200)) + async let poll: Void = store.pollPresence() + try? await Task.sleep(for: .milliseconds(600)) + for index in 1...6 { + await store.handle( + .messageNode(target, text: "later \(index)", from: sender, followUp: true)) + } + await poll + await store.pollPresence() + + let expected = + (1...10).map { "staged \($0)" } + (1...6).map { "later \($0)" } + #expect(plain(delivered.value) == expected) + } + + /// Issue #304's residual: a single drain reorders when the target's turn ends while + /// the drain is running. The first item is read mid-turn and put back; the next two are + /// read after the turn ends and go out — [2, 3, 1], with no overlapping drain anywhere, + /// which is why #309's non-reentrancy guard cannot see it. + /// + /// The reading is now taken once per target per pass, so a batch either goes out in + /// order or waits together for a pass that can deliver all of it. + /// + /// Fails before the fix on order: `["staged 2", "staged 3", "staged 1"]`. + @Test + func aTurnEndingMidDrainDoesNotReorderTheQueue() async { + let readings = Readings() + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, ids) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + let (target, sender) = (ids[0], ids[1]) + for index in 1...3 { + await store.handle( + .messageNode(target, text: "staged \(index)", from: sender, followUp: true)) + } + #expect(delivered.value.isEmpty) + + // The turn ends between the first item and the second. + await readings.firstReadThen(.busy, .idle) + await settle(store, target) + + await readings.set(.idle) + await settle(store, target) + + #expect(plain(delivered.value) == ["staged 1", "staged 2", "staged 3"]) + } + + /// Issue #304's other residual: `mail watch --on --topic other` re-scopes an existing + /// watch, and the wakes staged under the topic it left kept arriving — the same defect + /// `--off` had, reached by the command that does not read like a teardown. + /// + /// Fails before the fix by delivering the abandoned topic's wake as well. + @Test + func reScopingAWatchDropsTheAbandonedTopicsStagedWakes() async { + let readings = Readings() + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, ids) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + let (target, sender) = (ids[0], ids[1]) + + await store.handle(.mailroomWatch(on: true, topic: "alpha", from: target)) + await store.handle(.mailroomPost(text: "an alpha finding", topic: "alpha", from: sender)) + // Re-scoped, without ever passing through `--off`. + await store.handle(.mailroomWatch(on: true, topic: "beta", from: target)) + await store.handle(.mailroomPost(text: "a beta finding", topic: "beta", from: sender)) + #expect(delivered.value.isEmpty) + + await readings.set(.idle) + await store.pollPresence() + + #expect(delivered.value.count == 1) + #expect(delivered.value.first?.contains("a beta finding") == true) + #expect(!delivered.value.contains { $0.contains("an alpha finding") }) + } + + /// The rate `MailDeliveryCheck` measured on the live daemon and nobody had filed: + /// about **two deliveries per poll tick**, so ten staged messages took 147 seconds. + /// Delivering into a session is what makes it busy, and a drain that re-read presence + /// between items therefore stopped after the first delivery of every pass and waited + /// for the next tick — the queue drained at one item per fifteen seconds. + /// + /// One reading per target per pass ends it: the pass judges the target once, before it + /// has typed anything into it, and hands over the whole backlog in that pass. + /// + /// Fails before the fix on count: one delivered per pass, not ten. + @Test + func aBacklogDrainsInOnePassRatherThanOnePerTick() async { + let readings = Readings() + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, ids) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + let (target, sender) = (ids[0], ids[1]) + for index in 1...10 { + await store.handle( + .messageNode(target, text: "backlog \(index)", from: sender, followUp: true)) + } + #expect(delivered.value.isEmpty) + + // Idle when the pass begins, busy from the moment the first message is typed in. + await readings.firstReadThen(.idle, .busy) + await settle(store, target) + + #expect(plain(delivered.value) == (1...10).map { "backlog \($0)" }) + } +} From e2cc5b58f9b9e6a5222b39e0ae5c76245cae937d Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 19:16:56 -0700 Subject: [PATCH 3/6] Carry MailOrderingCheck's five ordering probes onto the fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Written against main by the loop that reproduced these defects independently (branch probe/304-ordering, 053f83f), scripted rather than timed: each one drives the store's presence readings from a fixed script, so a failure is a defect and never a race that happened to land. All five fail on main — reorder within one drain, the queue jump from a cached reading, the same jump reached by an ordinary partly-drained queue, the abandoned topic's wake, and the --off/--on workaround — and all five pass here unchanged, which is the point of taking them verbatim. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014D7evLSBWCnNeumHAWHH5S --- graphcode/Tests/OrderingProbeTests.swift | 225 +++++++++++++++++++++++ 1 file changed, 225 insertions(+) create mode 100644 graphcode/Tests/OrderingProbeTests.swift diff --git a/graphcode/Tests/OrderingProbeTests.swift b/graphcode/Tests/OrderingProbeTests.swift new file mode 100644 index 00000000..0eb4412e --- /dev/null +++ b/graphcode/Tests/OrderingProbeTests.swift @@ -0,0 +1,225 @@ +import ComposableArchitecture +import Foundation +import GraphcodeKit +import MailroomKit +import Testing + +#if canImport(Darwin) + import Darwin +#endif + +/// Issue #304, ordering half. #309 fixed duplicates and loss; these three probes +/// characterise what it did not fix. Each is deterministic — the presence readings are +/// scripted, so a failure here is a defect, not a race that happened to land. +@Suite +struct OrderingProbeTests { + /// Answers each presence read from a script, so a turn can end at an exact point in + /// one drain's walk over its batch. + private actor ScriptedReadings { + private var script: [Presence] + private var index = 0 + private(set) var log: [String] = [] + init(_ script: [Presence]) { self.script = script } + func read(_ scripted: Bool) -> PresenceReading { + guard scripted else { return PresenceReading(presence: .idle, confidence: .reported) } + let presence = index < script.count ? script[index] : (script.last ?? .idle) + index += 1 + log.append("\(index): \(presence)") + return PresenceReading(presence: presence, confidence: .reported) + } + func trace() -> [String] { log } + } + + private final class Attachment: @unchecked Sendable { + let daemonEnd: Int32 + private let peer: Int32 + private let drainer: Task + init() { + var pair: [Int32] = [0, 0] + _ = socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) + daemonEnd = pair[0] + peer = pair[1] + let reading = pair[1] + drainer = Task.detached { + var sink = [UInt8](repeating: 0, count: 65536) + while recv(reading, &sink, sink.count, 0) > 0 {} + } + } + deinit { + OutboundChannels.close(daemonEnd) + close(peer) + drainer.cancel() + } + } + + private func makeStore( + readings: ScriptedReadings, delivered: LockIsolated<[String]>, attachment: Attachment + ) async -> (GraphStore, target: UUID, sender: UUID) { + let store = GraphStore( + onEnsureSession: { _, _ in }, + onDeliverMessage: { _, message, _ in + delivered.withValue { $0.append(message) } + return true + }, + onReadPresence: { node, _ in await readings.read(node.title == "Target") }, + onMailroomEnabled: { true }) + await store.handle( + .createNode(NodeDraft(title: "Target", loopType: .turnBased, firstInstruction: "Work"))) + await store.handle( + .createNode(NodeDraft(title: "Sender", loopType: .turnBased, firstInstruction: "Work"))) + let ids = await store.graph.nodes.map(\.id) + await store.addConnection(id: UUID(), fileDescriptor: attachment.daemonEnd) + return (store, ids[0], ids[1]) + } + + private func plain(_ delivered: LockIsolated<[String]>) -> [String] { + delivered.value.map { $0.replacingOccurrences(of: "[graphcode] Sender: ", with: "") } + } + + /// PROBE A — a turn ending mid-drain reorders, with a SINGLE drain. + /// + /// The first read of the walk says busy, so item 1 is put back; the reads for items 2 + /// and 3 say idle, so they are delivered. Item 1 then arrives on the next drain. No + /// two drains overlap, so #309's in-flight guard is not involved at all. + @Test + func aTurnEndingMidDrainReordersWithinOneDrain() async { + // Target reads, in order: poll-1 refresh (busy, so every message stages), then the + // settle drain that follows each `messageNode`. The third of those drains walks a + // batch of two and the turn ends between its two reads: busy for item 1, idle for + // item 2. One drain, no overlap — #309's in-flight guard cannot see this. + let readings = ScriptedReadings([.busy, .busy, .busy, .idle, .idle]) + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, target, sender) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + await store.pollPresence() + for index in 1...3 { + await store.handle( + .messageNode(target, text: "staged \(index)", from: sender, followUp: true)) + } + await store.pollPresence() + + let texts = plain(delivered) + print("PROBE-A order: \(texts) reads: \(await readings.trace())") + #expect(texts.count == 3) + #expect(texts == ["staged 1", "staged 2", "staged 3"]) + } + + /// PROBE B — a later follow-up jumps the queue. + /// + /// `deliversLater` reads the CACHED `node.presence`; the drain reads LIVE per item. + /// Cache says busy so message 1 stages; the live read in the drain also says busy so + /// it stays queued; then message 2 arrives while the cache still says busy — it too + /// stages, so far so good. The jump is the other disagreement: the cache goes idle + /// (the poll wrote it) while the live read is still busy, so message 2 is typed + /// straight in ahead of the still-queued message 1. + @Test + func aLaterFollowUpDoesNotJumpAnEarlierOne() async { + // Target reads: poll-1 refresh busy (so "first" stages), its settle drain busy (so + // "first" is requeued), poll-2 refresh idle — the CACHE is now idle — and poll-2's + // drain read busy, so "first" stays queued. "second" then consults the cache, sees + // idle, and is typed straight into the session ahead of the still-queued "first". + let readings = ScriptedReadings([.busy, .busy, .idle, .busy, .idle]) + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, target, sender) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + await store.pollPresence() + await store.handle(.messageNode(target, text: "first", from: sender, followUp: true)) + + await store.pollPresence() + await store.handle(.messageNode(target, text: "second", from: sender, followUp: true)) + await store.pollPresence() + + let texts = plain(delivered) + print("PROBE-B order: \(texts) reads: \(await readings.trace())") + #expect(texts == ["first", "second"]) + } + + /// PROBE C — re-scoping a watch keeps the abandoned topic's staged wakes. + /// + /// `mail watch --on --topic other` overwrites the subscription but never touches the + /// wakes already queued for the old scope, and the drain's guard only asks whether + /// *some* watch stands, never whether it still matches the post's topic. + @Test + func reScopingAWatchDropsTheAbandonedTopicsWakes() async { + // Target reads: poll-1 refresh busy (so the wake stages) and the settle drain that + // follows the post busy too, so the wake is still queued when the re-scope lands. + let readings = ScriptedReadings([.busy, .busy, .idle, .idle]) + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, target, sender) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + await store.pollPresence() + await store.handle(.mailroomWatch(on: true, topic: "alpha", from: target)) + await store.handle(.mailroomPost(text: "an alpha post", topic: "alpha", from: sender)) + #expect(delivered.value.isEmpty) + + // The watcher re-scopes to a different topic before the wake could land. + await store.handle(.mailroomWatch(on: true, topic: "beta", from: target)) + await store.pollPresence() + + let texts = plain(delivered) + print("PROBE-C delivered: \(texts) reads: \(await readings.trace())") + #expect(texts.isEmpty) + #expect(texts.isEmpty) + } + + /// PROBE D — the queue jump needs no exotic timing at all. + /// + /// Delivering an item is *itself* what makes the target busy, so a drain that hands + /// over item 1 will read busy for item 2 and requeue it. The cache, meanwhile, was + /// written `idle` by the refresh that opened the same poll and is not rewritten until + /// the next one — up to fifteen seconds later. Every follow-up that arrives in that + /// window consults the stale idle cache and is typed straight in, ahead of the item + /// still queued. This is the ordinary shape of a partly-drained queue, not a race. + @Test + func aFollowUpArrivingAfterAPartialDrainDoesNotJumpTheRemainder() async { + // Target reads: poll-1 refresh busy (both messages stage) and its two settle-drain + // reads busy; poll-2 refresh idle (cache goes idle), poll-2 drain: item 1 idle — + // delivered, which starts a turn — item 2 busy, requeued. + let readings = ScriptedReadings([.busy, .busy, .busy, .idle, .idle, .busy, .idle, .idle]) + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, target, sender) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + await store.pollPresence() + for index in 1...2 { + await store.handle( + .messageNode(target, text: "staged \(index)", from: sender, followUp: true)) + } + await store.pollPresence() + // A peer speaks while the target is mid-turn on what the drain just handed it. + await store.handle(.messageNode(target, text: "third", from: sender, followUp: true)) + await store.pollPresence() + + let texts = plain(delivered) + print("PROBE-D order: \(texts) reads: \(await readings.trace())") + #expect(texts == ["staged 1", "staged 2", "third"]) + } + + /// PROBE E — the workaround for probe C: `--off` then `--on --topic beta` is clean. + /// + /// `--off` is the only path that reaches the staged wakes, so a re-scope spelled as + /// two commands drops the abandoned topic's backlog where the one-command re-scope + /// keeps it. Worth knowing, because it is what an operator can do today. + @Test + func offThenOnIsACleanReScope() async { + let readings = ScriptedReadings([.busy, .busy, .idle, .idle]) + let delivered = LockIsolated<[String]>([]) + let attachment = Attachment() + let (store, target, sender) = await makeStore( + readings: readings, delivered: delivered, attachment: attachment) + await store.pollPresence() + await store.handle(.mailroomWatch(on: true, topic: "alpha", from: target)) + await store.handle(.mailroomPost(text: "an alpha post", topic: "alpha", from: sender)) + + await store.handle(.mailroomWatch(on: false, topic: nil, from: target)) + await store.handle(.mailroomWatch(on: true, topic: "beta", from: target)) + await store.pollPresence() + + let texts = plain(delivered) + print("PROBE-E delivered: \(texts) reads: \(await readings.trace())") + #expect(texts.isEmpty) + } +} From 4656740182c00bba4bd5c0b8686a5178b25888de Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 19:21:41 -0700 Subject: [PATCH 4/6] Record a follow-up as staged when it is deferred, not when it is queued MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every live target's follow-up now joins the queue and `handle` ends in a drain, so a message to an idle session is queued and handed over inside the same call. Recording "follow-up staged" at queue time then described a wait that never happened — and gave every Mailroom watcher a log line per post it was woken for, on top of the wake it received instantly. `MailroomTests.liveIdleWatcherHearsThePostThroughTheDeliveryChannel` holds the property: a live idle watcher gets the delivery and no staging line. The record moves to the first time an item is actually put back, which is still inside the pass that deferred it — nothing the queue owes can be lost to a daemon restart. An item the drain drops because its target resolved or is no longer reachable lands in the log too, which the old queue-time record was what guaranteed. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014D7evLSBWCnNeumHAWHH5S --- GraphcodeKit/Sources/GraphStore.swift | 36 ++++++++++++++++++++++++--- 1 file changed, 32 insertions(+), 4 deletions(-) diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index 9cffb9ac..ee447254 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -182,6 +182,10 @@ public actor GraphStore { /// still queued, and lets a wake for a post the reader has since read go unsent. /// `nil` for a message a peer or a human sent. let watchedPostID: Int? + /// Whether the target's memory log already carries this message. Written the first + /// time the item is actually put back rather than when it was queued — see + /// `staged(_:)`. + var recorded: Bool = false } private var pendingFollowUps: [PendingFollowUp] = [] @@ -2699,7 +2703,6 @@ public actor GraphStore { // lost to a daemon restart delays the message to the next wake instead of // dropping it. if followUp, deliversLater(to: target) { - recordMemory(nodeID, "follow-up staged: \(message)") pendingFollowUps.append( PendingFollowUp(nodeID: nodeID, text: message, watchedPostID: watchedPostID)) return @@ -2772,6 +2775,21 @@ public actor GraphStore { /// the store settles, and from the presence poll — the reading that says "idle" is /// the reading this waits for. A target that resolved or died is simply dropped from /// the queue: its memory log has carried the message since it was staged. + /// The memory record a deferred follow-up leaves, written the first time it is really + /// put back rather than when it was queued. Every live target's follow-up now joins + /// the queue and `handle` ends in a drain, so a message to an idle session is queued + /// and handed over inside the same call: a log line saying it was staged would + /// describe a wait that never happened, and a watcher's log would gain one per post it + /// was woken for. Anything that does wait is recorded before the pass that deferred it + /// ends, which is what the queue's durability across a daemon restart rests on. + private func staged(_ pending: PendingFollowUp) -> PendingFollowUp { + guard !pending.recorded else { return pending } + recordMemory(pending.nodeID, "follow-up staged: \(pending.text)") + var recorded = pending + recorded.recorded = true + return recorded + } + private func drainPendingFollowUps() async { guard !pendingFollowUps.isEmpty, !isDrainingFollowUps else { return } isDrainingFollowUps = true @@ -2793,7 +2811,14 @@ public actor GraphStore { // costs one probe per target rather than one per message. var readings: [UUID: Presence] = [:] for pending in batch { - guard let node = graph.nodes[id: pending.nodeID], !node.isResolved else { continue } + guard let node = graph.nodes[id: pending.nodeID], !node.isResolved else { + // Its work is over, but the message must not go with the queue entry: the log is + // what the loop's next wake reads, and for a resolved loop that is all there is. + if !pending.recorded, graph.nodes[id: pending.nodeID] != nil { + recordMemory(pending.nodeID, "while you were away: \(pending.text)") + } + continue + } // A watcher's wake is only owed while the watch that asked for it stands and the // post is unread: `mail watch --off` after the wake was staged, an inbox that has // since read past the post, or a watch re-scoped to another topic — the wakes the @@ -2807,9 +2832,12 @@ public actor GraphStore { } switch MessageBus.deliverability(to: node) { case .targetBusyWithACheck: - remaining.append(pending) + remaining.append(staged(pending)) continue case .some: + if !pending.recorded { + recordMemory(pending.nodeID, "while you were away: \(pending.text)") + } continue case nil: break @@ -2825,7 +2853,7 @@ public actor GraphStore { // `.idle` and nothing else, which is what makes a timed-out read safe: `.unknown` // is not a state, so the message stays queued for a pass that gets an answer. guard presence == .idle else { - remaining.append(pending) + remaining.append(staged(pending)) continue } _ = await deliverToSession(node, pending.text) From e0f12b21eba6d6878ceda34896c4826ce3e8d345 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 19:30:52 -0700 Subject: [PATCH 5/6] Fold the drain's batch back on every exit path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stable-release check's reading of the wedge (STABLE-CHECK-0.1.64.md, worktrees/stable-check-311) was that the drain empties `pendingFollowUps` into a local batch before its first `await`, so a walk that never returns strands the whole queue in a variable. The deadline is what stops that walk from never returning; this is the other half of its request — whatever exit the pass takes, what it did not resolve goes back on the queue rather than out of scope with the locals. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014D7evLSBWCnNeumHAWHH5S --- GraphcodeKit/Sources/GraphStore.swift | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index ee447254..9f69d8ba 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -2781,7 +2781,10 @@ public actor GraphStore { /// and handed over inside the same call: a log line saying it was staged would /// describe a wait that never happened, and a watcher's log would gain one per post it /// was woken for. Anything that does wait is recorded before the pass that deferred it - /// ends, which is what the queue's durability across a daemon restart rests on. + /// ends. The floor under all of it is the board: a peer's message is mirrored onto + /// the Mailroom when it is sent and a watcher's wake is *about* a post already there, + /// so the content survives a daemon that dies mid-pass either way — this line is what + /// puts it in front of the loop's next wake without it having to go looking. private func staged(_ pending: PendingFollowUp) -> PendingFollowUp { guard !pending.recorded else { return pending } recordMemory(pending.nodeID, "follow-up staged: \(pending.text)") @@ -2800,6 +2803,13 @@ public actor GraphStore { let batch = pendingFollowUps pendingFollowUps = [] var remaining: [PendingFollowUp] = [] + var index = 0 + // Folded back on the way out of every path, not only the one that runs to the end: + // the walk holds the whole queue in locals, and the stable-release check's reading + // of the wedge was that a drain which never returns strands them there. It cannot + // now — the reading below is bounded — but what the pass did not resolve belongs on + // the queue rather than in a variable about to go out of scope, whatever the exit. + defer { pendingFollowUps = remaining + Array(batch[index...]) + pendingFollowUps } // One reading per target per pass, taken the first time this pass reaches that // target and reused for the rest of its queue. Reading again between items is what // let a turn ending *mid-drain* reorder the queue: three messages for one loop went @@ -2810,7 +2820,9 @@ public actor GraphStore { // the batch either goes out in order or waits together for the next pass. It also // costs one probe per target rather than one per message. var readings: [UUID: Presence] = [:] - for pending in batch { + while index < batch.count { + let pending = batch[index] + index += 1 guard let node = graph.nodes[id: pending.nodeID], !node.isResolved else { // Its work is over, but the message must not go with the queue entry: the log is // what the loop's next wake reads, and for a resolved loop that is all there is. @@ -2858,7 +2870,6 @@ public actor GraphStore { } _ = await deliverToSession(node, pending.text) } - pendingFollowUps = remaining + pendingFollowUps } private func announceError(_ message: String) { From eb15c3142f9d9ad29d779b2ac455da94051423a5 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 19:47:50 -0700 Subject: [PATCH 6/6] Hold the drain with an expiring lease, and log a drain that outlives it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deadline bounds the presence read. It does not bound the *class*: `deliverToSession` is the same `PTYProcessSession` chain with no deadline of its own — `zmx ls` to check the session exists, a write per chunk, and `ssh` for a remote loop — so a hang there would hold a bare flag for the life of the daemon exactly as the presence read did, and every loop in the project would stop receiving mail with nothing logged. That failure was measured on a real `graphcoded`: 364 seconds, zero errors (STABLE-CHECK-0.1.64.md). `RemoteEnsureGate` rejects the plain flag for this same chain and for this same reason. So the guard expires. A drain that outlives its lease is a wedge by definition: the next one records `event=drain-stall` through #289's diagnostics — frozen and working must stop looking identical — and takes the queue on. Taking over cannot duplicate anything, because the batch was taken and cleared in one actor step, so the successor finds only what was queued after it; and the lease is released only by the drain that still holds it, the mistake `RemoteEnsureGate.end(_:token:)` documents. Tests are `StableSoakCheck`'s wedge rig, assertions turned round to what this branch does, plus the one no deadline can give: `aHungDeliveryReleasesTheQueueWhenItsLeaseExpires` hangs the *delivery* and shows the bystander's mail arriving after the lease expires, with the stall in the log. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014D7evLSBWCnNeumHAWHH5S --- .gitignore | 3 + GraphcodeKit/Sources/Deadline.swift | 9 ++ GraphcodeKit/Sources/GraphStore.swift | 55 ++++++- graphcode/Tests/DrainWedgeTests.swift | 202 ++++++++++++++++++++++++++ 4 files changed, 264 insertions(+), 5 deletions(-) create mode 100644 graphcode/Tests/DrainWedgeTests.swift diff --git a/.gitignore b/.gitignore index 24a239c7..90242bcb 100644 --- a/.gitignore +++ b/.gitignore @@ -55,3 +55,6 @@ xcuserdata/ !/docs/_includes/ !/docs/assets/ !/docs/ramps.json + +# Per-worktree DerivedData (docs: .claude/skills/pr-gate-report) +.derived/ diff --git a/GraphcodeKit/Sources/Deadline.swift b/GraphcodeKit/Sources/Deadline.swift index 56ffb64c..f347a229 100644 --- a/GraphcodeKit/Sources/Deadline.swift +++ b/GraphcodeKit/Sources/Deadline.swift @@ -49,3 +49,12 @@ private actor DeadlineRelay { return await withCheckedContinuation { self.waiter = $0 } } } + +extension Duration { + /// The same span as `Date` arithmetic wants it. `Duration` is what the concurrency + /// APIs take and `TimeInterval` is what a lease compares against; this is the one + /// conversion between them. + var timeInterval: TimeInterval { + Double(components.seconds) + Double(components.attoseconds) / 1e18 + } +} diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index 9f69d8ba..92231dd4 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -194,7 +194,25 @@ public actor GraphStore { /// the first was suspended delivered the same items again and, on finishing, wrote /// its own idea of what remained over the first's — dropping whatever had been /// queued in between (issue #304: duplicates, lost mail, and out-of-order delivery). - private var isDrainingFollowUps = false + /// A drain in flight, held as a **lease** rather than a flag: the moment it was taken, + /// which is also proof of *which* drain holds it. + /// + /// A bare boolean was enough while the only unbounded await under it was the presence + /// read, which now has a deadline. It is not enough for the class: `deliverToSession` + /// is the same `PTYProcessSession` chain with no deadline of its own — `zmx ls` to + /// check the session exists, a write per chunk, and `sendRemote` over `ssh` for a + /// remote loop — so a hang there would hold a bare flag for the life of the daemon + /// exactly as the presence read did, and every loop in the project would stop + /// receiving mail with nothing logged (measured: 364s, zero errors). `RemoteEnsureGate` + /// rejects the plain flag for this same chain and for this same reason. + /// + /// So the guard expires. A drain that outlives its lease is a wedge by definition, and + /// the next one says so in the daemon log and takes over. Taking over cannot duplicate + /// anything — the batch was taken and cleared in one actor step, so the successor + /// finds only what was queued after it — and if the wedged drain ever returns, its + /// `defer` folds its own remainder back on. Order can slip in that recovery; a queue + /// that moves again beats one frozen for ever. + private var drainLease: Date? /// How long a presence read may take before the store stops waiting on it. /// @@ -203,7 +221,7 @@ public actor GraphStore { /// `PTYProcessSession.waitCollectingOutput`, which ends only when the probe's /// `terminationHandler` closes the stream, and `ssh`'s `ConnectTimeout=10` bounds the /// connect rather than a command left hanging on a host that has gone away. An `await` - /// that never returns held `isDrainingFollowUps` for the life of the daemon, which + /// that never returns held the drain's guard for the life of the daemon, which /// froze staged delivery for *every* loop in the project — and a frozen queue and an /// empty one report exactly the same thing from outside (issue #311). /// @@ -213,6 +231,13 @@ public actor GraphStore { /// one poll, not the process. private let presenceReadDeadline: Duration + /// How long a drain may hold the queue before another is allowed to take over. Far + /// longer than any healthy pass (milliseconds) and longer than a pass that meets + /// several wedged presence reads, each of which is bounded by `presenceReadDeadline`; + /// short enough that a hang costs minutes rather than the life of the process. The + /// 300s here is `RemoteEnsureGate.leaseDuration`, arrived at for the same chain. + private let drainLeaseDuration: Duration + /// A poller holds `self` weakly, so a store going away already stops it *doing* /// anything — but the task itself keeps sleeping in its loop forever. Harmless for /// the long-lived project store; sub-graph stores are built per command and hold no @@ -285,6 +310,7 @@ public actor GraphStore { goalCache: GoalEvaluationCache? = nil, recurrence: RecurrenceSink? = nil, presenceReadDeadline: Duration = .seconds(45), + drainLeaseDuration: Duration = .seconds(300), subGraphDepth: Int = 0 ) { self.graph = graph @@ -317,6 +343,7 @@ public actor GraphStore { self.goalCache = goalCache ?? GoalEvaluationCache() self.recurrence = recurrence self.presenceReadDeadline = presenceReadDeadline + self.drainLeaseDuration = drainLeaseDuration } /// The store's one way to ask what a session is doing, and the only place the answer @@ -2794,9 +2821,27 @@ public actor GraphStore { } private func drainPendingFollowUps() async { - guard !pendingFollowUps.isEmpty, !isDrainingFollowUps else { return } - isDrainingFollowUps = true - defer { isDrainingFollowUps = false } + guard !pendingFollowUps.isEmpty else { return } + let taken = Date() + if let held = drainLease { + let heldFor = taken.timeIntervalSince(held) + guard heldFor >= drainLeaseDuration.timeInterval else { return } + // The line that stops frozen and working looking identical. A drain past its lease + // has hung somewhere this store could not bound — the delivery chain, most likely + // — and #289's diagnostics are where that becomes visible instead of being + // inferred from mail that never came. + DaemonLog.shared.record( + "drain-stall", + DaemonRequestContext.fields + [ + ("held_ms", DaemonLog.milliseconds(heldFor)), + ("queued", String(pendingFollowUps.count)), + ]) + } + drainLease = taken + // Released only if it is still ours: a successor that took over after this lease + // expired must not have its own lease cleared by the drain it replaced finally + // returning — the mistake `RemoteEnsureGate.end(_:token:)` documents. + defer { if drainLease == taken { drainLease = nil } } // Taken and cleared in one actor step, delivered from the local batch. Anything // queued while a delivery below is suspended lands in `pendingFollowUps` untouched // and is folded back in behind the retries at the end — never overwritten. diff --git a/graphcode/Tests/DrainWedgeTests.swift b/graphcode/Tests/DrainWedgeTests.swift new file mode 100644 index 00000000..310af5cb --- /dev/null +++ b/graphcode/Tests/DrainWedgeTests.swift @@ -0,0 +1,202 @@ +import ComposableArchitecture +import Foundation +import GraphcodeKit +import Testing + +/// Issue #311's wedge, from the rig `StableSoakCheck` reproduced it with on a real +/// `graphcoded` (`worktrees/stable-check-311`, `probe/311-drain-wedge`): a control run +/// delivered a bystander's queued mail in ~15s, and the run that differs in **one +/// closure** — node A's presence read hangs — never delivered it in 364 seconds, with a +/// client attached throughout and zero errors logged. +/// +/// Its tests asserted the freeze, since that was what beta5 did. These are the same +/// scenarios with the assertions turned round to what this branch does instead, plus the +/// one its report asked for that no deadline can give: a hang in a path with no deadline +/// at all. +@Suite +struct DrainWedgeTests { + private struct Fixture { + let graph: LoopGraph + var hung: UUID { graph.nodes[0].id } + var bystander: UUID { graph.nodes[1].id } + } + + private func fixture() -> Fixture { + Fixture( + graph: LoopGraph( + project: ProjectRef(path: "/tmp/drainwedge", name: "drainwedge"), + nodes: [ + LoopNode( + title: "Hung", loopType: .goalBased, goal: GoalSpec(summary: "hangs"), + presence: PresenceReading(presence: .busy, confidence: .reported), + state: .running), + LoopNode( + title: "Bystander", loopType: .goalBased, goal: GoalSpec(summary: "innocent"), + presence: PresenceReading(presence: .busy, confidence: .reported), + state: .running), + ])) + } + + private func settle(until flag: LockIsolated) async { + for _ in 0..<2000 where !flag.value { try? await Task.sleep(for: .milliseconds(5)) } + } + + /// The control, unchanged from the probe: node A's read answers `busy`, so the drain + /// steps past it and reaches node B, whose reading is idle. + @Test + func aBystandersFollowUpIsDeliveredWhenEveryPresenceReadAnswers() async { + let fixture = fixture() + let delivered = LockIsolated<[String]>([]) + let store = GraphStore( + graph: fixture.graph, + onDeliverMessage: { _, message, _ in + delivered.withValue { $0.append(message) } + return true + }, + onReadPresence: { node, _ in + PresenceReading( + presence: node.id == fixture.hung ? .busy : .idle, confidence: .reported) + }) + + await store.handle( + .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true)) + await store.handle( + .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true)) + + #expect(delivered.value == ["[graphcode] for the bystander"]) + } + + /// The wedge. Same graph, same two messages; node A's read never returns. + /// + /// On beta5 both the command that queued A's follow-up and every later drain were + /// held for the life of the process, and B's mail was never typed in however idle B + /// was. Bounded, the read is `.unknown` after the deadline, which is no state at all: + /// A's message stays owed and B's is delivered in the same pass. + @Test + func aHungPresenceReadDoesNotFreezeEveryOtherLoopsFollowUps() async { + let fixture = fixture() + let delivered = LockIsolated<[String]>([]) + let announced = LockIsolated<[String]>([]) + let entered = LockIsolated(false) + let release = LockIsolated(false) + let store = GraphStore( + graph: fixture.graph, + onDeliverMessage: { _, message, _ in + delivered.withValue { $0.append(message) } + return true + }, + onReadPresence: { node, _ in + guard node.id == fixture.hung else { + return PresenceReading(presence: .idle, confidence: .reported) + } + entered.setValue(true) + while !release.value { try? await Task.sleep(for: .milliseconds(5)) } + return PresenceReading(presence: .busy, confidence: .reported) + }, + onAnnounceError: { message in announced.withValue { $0.append(message) } }, + presenceReadDeadline: .milliseconds(200)) + + // Queued in a task of its own so this test *fails* on the head where the drain never + // returns, rather than hanging with it. + 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)) + // Past the deadline the wedged pass is over, and the next settle carries B's mail. + try? await Task.sleep(for: .milliseconds(400)) + await store.handle(.memoNode(fixture.bystander, text: "settle", from: fixture.bystander)) + + #expect(delivered.value == ["[graphcode] for the bystander"]) + #expect(announced.value.isEmpty) + release.setValue(true) + _ = await wedging.value + } + + /// The same hang one layer up: `refreshPresence` walks the graph serially, so on beta5 + /// the poll tick never reached the nodes behind the hung one and their presence stopped + /// being read at all. The bound is on the reading itself, so the walk continues. + @Test + func aHungPresenceReadDoesNotStopThePollTickReachingLaterLoops() async { + let fixture = fixture() + let asked = LockIsolated<[UUID]>([]) + let entered = LockIsolated(false) + let release = LockIsolated(false) + let store = GraphStore( + graph: fixture.graph, + onReadPresence: { node, _ in + asked.withValue { $0.append(node.id) } + guard node.id == fixture.hung else { + return PresenceReading(presence: .idle, confidence: .reported) + } + entered.setValue(true) + while !release.value { try? await Task.sleep(for: .milliseconds(5)) } + return PresenceReading(presence: .busy, confidence: .reported) + }, + presenceReadDeadline: .milliseconds(200)) + + let wedging = Task { await store.handle(.refreshUsage) } + await settle(until: entered) + try? await Task.sleep(for: .milliseconds(400)) + + #expect(asked.value.contains(fixture.hung)) + #expect(asked.value.contains(fixture.bystander)) + release.setValue(true) + _ = await wedging.value + } + + /// What the deadline cannot reach, and the reason the guard is a lease. + /// + /// `deliverToSession` is the same `PTYProcessSession` chain as the presence read and + /// has no deadline of its own — `zmx ls`, a write per chunk, `ssh` for a remote loop. + /// A hang there holds a bare flag exactly as the presence read did, and the failure is + /// the one the soak measured: total, and silent. So the guard expires: the next drain + /// records the stall in the daemon log and takes the queue on. + @Test + func aHungDeliveryReleasesTheQueueWhenItsLeaseExpires() async { + let fixture = fixture() + let delivered = LockIsolated<[String]>([]) + let lines = LockIsolated<[String]>([]) + let entered = LockIsolated(false) + let release = LockIsolated(false) + let tap = DaemonLog.shared.tap { line in lines.withValue { $0.append(line) } } + defer { DaemonLog.shared.untap(tap) } + let store = GraphStore( + graph: fixture.graph, + 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)) } + return true + }, + onReadPresence: { _, _ in PresenceReading(presence: .idle, confidence: .reported) }, + presenceReadDeadline: .milliseconds(200), + drainLeaseDuration: .milliseconds(300)) + + // The drain that hands A its message never returns from the delivery. + let wedging = Task { + await store.handle( + .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true)) + } + await settle(until: entered) + + // While the lease stands, the queue is the wedged drain's: B waits. + await store.handle( + .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true)) + #expect(delivered.value.isEmpty) + + // Past the lease, the next drain takes over and says so. + try? await Task.sleep(for: .milliseconds(350)) + await store.handle(.memoNode(fixture.bystander, text: "settle", from: fixture.bystander)) + + #expect(delivered.value == ["[graphcode] for the bystander"]) + #expect(lines.value.contains { $0.contains("event=drain-stall") }) + release.setValue(true) + _ = await wedging.value + } +}