Ship the presence tick as the loops it moved, not the whole graph (#288)

GraphcodeKit/Sources/Domain/LoopGraph.swift

@@ -39,6 +39,11 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable {
   /// Set only on the copy a daemon sends (`wireSnapshot()`); `nil` on the graph the
   /// daemon owns and persists.
   public var mailroomDigest: MailroomDigest?
+  /// Where this snapshot sits in the daemon's sequence of frames for the graph — see
+  /// `DaemonEvent.nodesChanged`. Wire-only like `mailroomDigest`: stamped on the copy a
+  /// daemon sends, `nil` on the graph it owns and persists, and on a snapshot from a
+  /// daemon that predates deltas.
+  public var revision: Int?
 
   public var project: ProjectRef {
     get { scope.projectRef }
@@ -55,13 +60,26 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable {
 
   /// This graph as a `.graphChanged` frame carries it: the posts stripped and their
   /// digest stamped in their place. Everything else is the graph exactly as it is.
-  public func wireSnapshot() -> LoopGraph {
+  public func wireSnapshot(revision: Int? = nil) -> LoopGraph {
     var copy = self
     copy.mailroomDigest = MailroomDigest(of: mailroom)
     copy.mailroom = []
+    copy.revision = revision
     return copy
   }
 
+  /// This graph with `nodes` replaced by id — how a client applies a
+  /// `DaemonEvent.nodesChanged` to the snapshot it holds. A node the graph does not
+  /// have is ignored: a delta says how a loop changed, never that one appeared.
+  public func applying(nodesChanged nodes: [LoopNode], revision: Int) -> LoopGraph {
+    var merged = self
+    for node in nodes where merged.nodes[id: node.id] != nil {
+      merged.nodes[id: node.id] = node
+    }
+    merged.revision = revision
+    return merged
+  }
+
   public init(
     id: UUID = UUID(),
     scope: LoopGraphScope,
@@ -278,7 +296,7 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable {
   // MARK: - Coding
 
   private enum CodingKeys: String, CodingKey {
-    case id, nodes, edges, mailroom, mailroomDigest
+    case id, nodes, edges, mailroom, mailroomDigest, revision
     /// Persisted as a `ProjectRef` rather than as the scope enum. Every graph on disk
     /// predates `LoopGraphScope`, and the ref round-trips both cases losslessly (the
     /// global graph's reserved path decodes straight back to `.global`), so there was
@@ -297,6 +315,7 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable {
       try container.decodeIfPresent([MailroomPost].self, forKey: .mailroom)
       ?? decoder.legacyMailroomValue([MailroomPost].self, "artifactory") ?? []
     mailroomDigest = try container.decodeIfPresent(MailroomDigest.self, forKey: .mailroomDigest)
+    revision = try container.decodeIfPresent(Int.self, forKey: .revision)
   }
 
   public func encode(to encoder: Encoder) throws {
@@ -309,5 +328,6 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable {
     // what it was — the same reason `hasActiveDependents` never reaches disk.
     if !mailroom.isEmpty { try container.encode(mailroom, forKey: .mailroom) }
     if let mailroomDigest { try container.encode(mailroomDigest, forKey: .mailroomDigest) }
+    if let revision { try container.encode(revision, forKey: .revision) }
   }
 }

GraphcodeKit/Sources/GraphStore.swift

@@ -32,6 +32,12 @@ import MailroomKit
 public actor GraphStore {
   public private(set) var graph: LoopGraph
   private var connections: [UUID: Int32] = [:]
+  /// What each connection announced it can read (`DaemonCommand.announce`) — what
+  /// decides whether a presence tick reaches it as a delta or as the whole snapshot.
+  private var connectionCapabilities: [UUID: Set<String>] = [:]
+  /// One counter for every frame this store sends about its graph — snapshots and
+  /// presence deltas alike — so a client can order them (`DaemonEvent.nodesChanged`).
+  private var revision = 0
   private let onGraphChanged: (@Sendable (LoopGraph) -> Void)?
   private let onEnsureSession: (@Sendable (LoopNode, String?) -> Void)?
   private let onTerminateSession: (@Sendable (LoopNode, String?) -> Void)?
@@ -501,18 +507,29 @@ public actor GraphStore {
 
   // MARK: - Connections
 
-  public func addConnection(id: UUID, fileDescriptor: Int32) {
+  public func addConnection(
+    id: UUID, fileDescriptor: Int32, capabilities: Set<String> = []
+  ) {
+    connectionCapabilities[id] = capabilities
     // Joining a project registers the connection here too, so ensure its outbound half
     // the same way the registry does. Without this a store could bind to a channel left
     // dead on a recycled descriptor number and drop the client as disconnected on the
     // snapshot it was joining for.
     OutboundChannels.open(fileDescriptor)
     connections[id] = fileDescriptor
-    send(.graphChanged(graph.wireSnapshot()), to: id)
+    send(.graphChanged(graph.wireSnapshot(revision: revision)), to: id)
   }
 
   public func removeConnection(_ id: UUID) {
     connections.removeValue(forKey: id)
+    connectionCapabilities.removeValue(forKey: id)
+  }
+
+  /// An announcement that arrived after the connection joined — see
+  /// `ProjectRegistry`'s handling of `.announce`.
+  public func setCapabilities(_ capabilities: Set<String>, for id: UUID) {
+    guard connections[id] != nil else { return }
+    connectionCapabilities[id] = capabilities
   }
 
   // MARK: - Commands
@@ -1193,6 +1210,7 @@ public actor GraphStore {
     // loop is working and the line under it says what at. Reading the second only when
     // someone presses refresh left every card describing the tool call its session made
     // whenever that happened to be.
+    let before = graph.nodes
     var changed = await refreshPresence()
     if await refreshActivity() { changed = true }
     if await refreshSummary() { changed = true }
@@ -1203,7 +1221,12 @@ public actor GraphStore {
     // what was waiting on exactly that.
     await drainPendingFollowUps()
     guard changed else { return }
-    notifyClients()
+    // Only the loops the tick touched, never the whole graph: everything above edits
+    // fields on top-level nodes, so the diff is exact, and a busy graph's fifteen-second
+    // pulse becomes a kilobyte per loop that moved instead of the whole snapshot.
+    let moved = Array(graph.nodes.filter { before[id: $0.id] != $0 })
+    guard !moved.isEmpty else { return }
+    notifyClients(nodesChanged: moved)
   }
 
   // MARK: - Renaming
@@ -3259,20 +3282,46 @@ public actor GraphStore {
     // Encoded once, not once per connection: the frame is the same bytes for every
     // client, and encoding a full graph is the expensive half of a broadcast — with C
     // clients attached it was C encodes of the same snapshot on every change, presence
-    // tick included (issue #288's CPU amplifier). And not at all with no client: the
-    // daemon runs clientless most of the day, and `send` used to find no descriptor
-    // before it encoded anything — hoisting the encode must not turn zero into one.
-    guard !connections.isEmpty else { return }
-    guard let frame = Self.encode(.graphChanged(graph.wireSnapshot())) else { return }
+    // tick included (issue #288's CPU amplifier).
+    revision += 1
+    guard let frame = Self.encode(.graphChanged(graph.wireSnapshot(revision: revision)))
+    else { return }
     for id in connections.keys {
       deliver(frame, to: id)
     }
   }
 
+  /// The presence poll's broadcast — see `DaemonEvent.nodesChanged`. Not superseded:
+  /// a newer delta does not carry what an older one said, so both go out; the revision
+  /// is what lets a client drop one a later snapshot has overtaken.
+  ///
+  /// Only a connection that announced `ClientCapability.nodesChanged` gets the delta.
+  /// Every other connection gets the whole snapshot for the same tick, stamped with the
+  /// same revision — what every client got before deltas existed, so an older app
+  /// never meets a frame it cannot read. Both frames are encoded at most once.
+  private func notifyClients(nodesChanged nodes: [LoopNode]) {
+    revision += 1
+    let delta = Self.encode(
+      .nodesChanged(projectPath: graph.project.path, revision: revision, nodes: nodes))
+    var snapshot: EncodedEvent?
+    for (id, capabilities) in connectionCapabilities where connections[id] != nil {
+      if capabilities.contains(ClientCapability.nodesChanged.rawValue) {
+        if let delta { deliver(delta, to: id) }
+      } else {
+        if snapshot == nil {
+          snapshot = Self.encode(.graphChanged(graph.wireSnapshot(revision: revision)))
+        }
+        if let snapshot { deliver(snapshot, to: id) }
+      }
+    }
+  }
+
   /// An event as the bytes and the superseding key it goes out with — everything about
   /// a frame that does not depend on which connection receives it.
   private struct EncodedEvent {
     let data: Data
+    /// `DaemonEvent.requiredCapability` — what `deliver` checks a connection announced.
+    let requiredCapability: ClientCapability?
     /// A snapshot still waiting to go out is replaced by a newer one rather than queued
     /// behind it — the event carries the whole graph, so the older one has nothing
     /// left to say. Keyed per graph, never on the event name alone: one connection joins
@@ -3289,7 +3338,8 @@ public actor GraphStore {
       if case .graphChanged(let changed) = event { return "graphChanged:\(changed.id)" }
       return nil
     }()
-    return EncodedEvent(data: data, supersedingKey: supersedingKey)
+    return EncodedEvent(
+      data: data, requiredCapability: event.requiredCapability, supersedingKey: supersedingKey)
   }
 
   /// One event to one connection — the unicast shape (`addConnection`'s joining
@@ -3301,6 +3351,15 @@ public actor GraphStore {
 
   private func deliver(_ frame: EncodedEvent, to connectionID: UUID) {
     guard let fileDescriptor = connections[connectionID] else { return }
+    // The one place the daemon's default is enforced: an event a connection never
+    // announced it could read is not sent to it, whatever call site asked. A caller
+    // that wants such a connection kept current sends it the legacy shape instead
+    // (`notifyClients(nodesChanged:)` sends the snapshot).
+    if let required = frame.requiredCapability,
+      connectionCapabilities[connectionID]?.contains(required.rawValue) != true
+    {
+      return
+    }
     // Queued, never written here: this runs on the `GraphStore` actor, and a
     // `graphChanged` frame is far larger than a socket's send buffer, so writing it
     // inline blocked the actor for as long as the slowest client took to read

GraphcodeKit/Sources/IPC/DaemonProtocol.swift

@@ -46,6 +46,40 @@ public enum DaemonCommand: Codable, Sendable, Equatable {
   /// wants (issue #288). Requires the project to be resident in the daemon — opened by
   /// some connection — the same rule as `.graphCommand`.
   case mailbox(projectPath: String, query: MailboxQuery)
+  /// What this client can read beyond the events every client has always been sent —
+  /// the first frame the app puts on every socket, dial and redial alike. A daemon
+  /// sends a client only what it announced (`ClientCapability`): a client that never
+  /// announces — an older app — keeps getting the whole snapshot on every presence
+  /// tick, exactly as before, rather than a frame it cannot decode. Which matters
+  /// because the app's reader used to take an undecodable frame for a dead socket and
+  /// redial, every fifteen seconds, for ever. Unknown names are ignored, so a newer
+  /// client against this daemon is simply treated as what it is: a client of the
+  /// capabilities this daemon knows. Never answered.
+  case announce(capabilities: [String])
+}
+
+/// The names a client announces (`DaemonCommand.announce`) — strings on the wire so a
+/// name this build does not know decodes rather than fails.
+public enum ClientCapability: String, Sendable {
+  /// Reads `DaemonEvent.nodesChanged` and folds it into the snapshot it holds.
+  case nodesChanged
+}
+
+extension DaemonEvent {
+  /// What a connection must have announced to be sent this event, or `nil` for the
+  /// events every client has always been sent. **Exhaustive on purpose**: adding a
+  /// case to `DaemonEvent` does not compile until its author has decided here whether
+  /// a client that predates it may receive it — and the one delivery path
+  /// (`GraphStore.deliver`) enforces the answer, so the daemon's default is that a
+  /// connection which never announced anything gets no new event type. That default is
+  /// the safety mechanism for clients already in the field, which are exactly the ones
+  /// that never announce; the handshake is how a newer client opts in.
+  public var requiredCapability: ClientCapability? {
+    switch self {
+    case .recentProjectsListed, .graphChanged, .errorOccurred, .mailbox: return nil
+    case .nodesChanged: return .nodesChanged
+    }
+  }
 }
 
 /// Mutations against exactly one project's graph — this is what `GraphStore.handle`
@@ -218,4 +252,18 @@ public enum DaemonEvent: Codable, Sendable, Equatable {
   /// `projectPath` is the canonical spelling the daemon routed the query to, which is
   /// the id an app keys its projects by.
   case mailbox(projectPath: String, mailbox: Mailbox)
+  /// The presence poll's broadcast: only the loops whose reading, activity, summary or
+  /// board changed on this tick, as whole `LoopNode` values, instead of the whole graph
+  /// every fifteen seconds (issue #288's background load — on a busy graph something
+  /// changes almost every tick, and the tick shipped 50 KB to say which pill moved).
+  /// A client merges them into the snapshot it holds by id. Sent only to a connection
+  /// that announced `ClientCapability.nodesChanged`; every other connection gets the
+  /// whole snapshot for the same tick, as it always did.
+  ///
+  /// `revision` orders it against snapshots: the daemon stamps every frame for a graph
+  /// from one counter (`LoopGraph.revision`), and a client applies a delta only when it
+  /// is newer than the graph it holds. That is what makes it safe for an undelivered
+  /// snapshot to be superseded by a later one while a delta queued behind it still
+  /// arrives — the delta is older than what the client has, and is dropped.
+  case nodesChanged(projectPath: String, revision: Int, nodes: [LoopNode])
 }

GraphcodeKit/Sources/ProjectRegistry.swift

@@ -114,6 +114,11 @@ public actor ProjectRegistry {
 
   // MARK: - Connections
 
+  /// What each connection announced it can read — see `DaemonCommand.announce`. Kept
+  /// here, per connection, and handed to every store the connection joins, before or
+  /// after the announcement arrives.
+  private var connectionCapabilities: [UUID: Set<String>] = [:]
+
   public func addConnection(id: UUID, fileDescriptor: Int32) {
     // Registering the connection is what opens its outbound half — here rather than in
     // the daemon's accept loop because this is the one place every caller goes through,
@@ -132,6 +137,7 @@ public actor ProjectRegistry {
     }
     connectionFileDescriptors.removeValue(forKey: id)
     connectionProjectPaths.removeValue(forKey: id)
+    connectionCapabilities.removeValue(forKey: id)
     sidebarConnections.remove(id)
     if connectionFileDescriptors.isEmpty { stopPresencePolling() }
   }
@@ -357,6 +363,16 @@ public actor ProjectRegistry {
         send(.errorOccurred(reason), to: fileDescriptor)
       }
 
+    case .announce(let capabilities):
+      // Reaches every store this connection has already joined too: the app's launch
+      // sends its joins and its announcement together, and which lands first must not
+      // decide what the client is sent.
+      let announced = Set(capabilities)
+      connectionCapabilities[connectionID] = announced
+      for path in connectionProjectPaths[connectionID] ?? [] {
+        await stores[path]?.setCapabilities(announced, for: connectionID)
+      }
+
     case .mailbox(let path, let query):
       // Routed and gated exactly as a `.graphCommand`: the room belongs to the project
       // the open landed on, and a query against a project this connection never
@@ -385,7 +401,9 @@ public actor ProjectRegistry {
   private func open(_ canonicalPath: String, for connectionID: UUID, fileDescriptor: Int32) async {
     let store = await store(forProjectPath: canonicalPath)
     connectionProjectPaths[connectionID, default: []].insert(canonicalPath)
-    await store.addConnection(id: connectionID, fileDescriptor: fileDescriptor)
+    await store.addConnection(
+      id: connectionID, fileDescriptor: fileDescriptor,
+      capabilities: connectionCapabilities[connectionID] ?? [])
     // The global graph is always resident and isn't a folder anyone opened, so it stays
     // out of both the recents list and the restore-on-launch set — the app asks for it
     // by name every launch instead.
@@ -415,7 +433,8 @@ public actor ProjectRegistry {
     for id in sidebarConnections where id != opener {
       guard let fileDescriptor = connectionFileDescriptors[id] else { continue }
       connectionProjectPaths[id, default: []].insert(path)
-      await store.addConnection(id: id, fileDescriptor: fileDescriptor)
+      await store.addConnection(
+        id: id, fileDescriptor: fileDescriptor, capabilities: connectionCapabilities[id] ?? [])
     }
   }
 

graphcode-cli/Sources/main.swift

@@ -101,7 +101,7 @@ func openProject(_ projectPath: String) throws -> LoopGraph? {
   let opened = try client.waitForEvent {
     switch $0 {
     case .graphChanged, .errorOccurred: return true
-    case .recentProjectsListed, .mailbox: return false
+    case .recentProjectsListed, .mailbox, .nodesChanged: return false
     }
   }
   if case .errorOccurred(let message) = opened { fail(message) }