feat(acp): migrate subagent observation to ACP sessions

Cargo.toml

@@ -37,10 +37,10 @@ a2a-protocol-client = { version = "=0.14.1", default-features = false, features
 a2a-protocol-server = { version = "=0.14.1", default-features = false }
 a2a-protocol-types = "=0.14.1"
 agenthooksprotocol = { version = "=0.10.1", default-features = false, features = ["reqwest", "tokio-process"] }
-agent-client-protocol = { package = "agentkit-acp-sdk", version = "=2.2.0", features = ["unstable_session_fork", "unstable_session_inject"] }
-agent-client-protocol-schema = { package = "agentkit-acp-schema", version = "=1.9.1", features = ["unstable_session_notices", "unstable_session_compaction"] }
-agent-client-protocol-http = { package = "agentkit-acp-http", version = "=2.2.0", default-features = false, features = ["server", "client"] }
-agentkit-acp = { version = "=0.11.0", features = ["unstable-inject"] }
+agent-client-protocol = { package = "agentkit-acp-sdk", version = "=3.6.0", features = ["stdio", "unstable_session_fork", "unstable_session_inject", "unstable_subagents"] }
+agent-client-protocol-schema = { package = "agentkit-acp-schema", version = "=1.14.0" }
+agent-client-protocol-http = { package = "agentkit-acp-http", version = "=3.6.0", default-features = false, features = ["server", "client"] }
+agentkit-acp = { version = "=0.12.0", features = ["unstable-inject"] }
 agentkit-adapter-completions = "=0.11.0"
 agentkit-core = "=0.10.5"
 agentkit-context = "=0.10.5"
@@ -117,7 +117,7 @@ cubeb = { version = "=0.38.0", default-features = false, optional = true }
 cubeb-sys = { version = "=0.38.0", default-features = false, features = ["unittest-build"], optional = true }
 
 [target.'cfg(unix)'.dependencies]
-libc = "=0.2.190"
+libc = "=0.2.189"
 
 [target.'cfg(windows)'.dependencies]
 windows-sys = { version = "=0.61.2", features = [
@@ -142,8 +142,3 @@ todo = "deny"
 unimplemented = "deny"
 disallowed_methods = "deny"
 disallowed_macros = "deny"
-
-# Bounded SSE replay fairness; keep HTTP and core channel types on one revision.
-[patch.crates-io]
-agentkit-acp-http = { git = "https://github.com/danielkov/rust-sdk", rev = "03eab519aa5ecfc97a43bf17f9721c38c065c47b" }
-agentkit-acp-sdk = { git = "https://github.com/danielkov/rust-sdk", rev = "03eab519aa5ecfc97a43bf17f9721c38c065c47b" }

macos/KitDesktop/Models/AppState.swift

@@ -324,6 +324,55 @@ struct AgentRoster: Equatable {
     private(set) var usageByID: [String: DesktopUsageUpdate] = [:]
     private var steerableGenerations: [String: UInt64] = [:]
 
+    mutating func syncStreamed(_ source: AgentRoster) {
+        // The ACP stream owns associations and steering grants. Runtime events
+        // retain richer lifecycle metadata, including removal tombstones that
+        // must survive later ACP "unknown" state snapshots and content updates.
+        // An empty stream can arrive before helper-exit retirement. Revoke
+        // grants below, but preserve lifecycle rows for failed-removal display.
+        for (id, parentID) in source.parentsByID { parentsByID[id] = parentID }
+        removeCleanedDescendants()
+        steerableGenerations.removeAll()
+        for (id, streamed) in source.rowsByID {
+            guard !cleanedIDs.contains(id) else { continue }
+            if let version = versions[id],
+               version.generation > streamed.generation
+                || (version.generation == streamed.generation && version.statusRank == SubagentStatus.removed.rank) {
+                continue
+            }
+            let runtime = rowsByID[id]
+            let sameGeneration = runtime?.generation == streamed.generation
+            let runtimeStatus = versions[id].flatMap { version -> SubagentStatus? in
+                guard version.generation == streamed.generation, version.statusRank > streamed.status.rank else { return nil }
+                return runtime?.status
+            }
+            let status = runtimeStatus ?? streamed.status
+            rowsByID[id] = AgentRosterRow(
+                id: id, name: streamed.name, status: status,
+                outcome: sameGeneration ? runtime?.outcome : nil,
+                generation: streamed.generation, task: streamed.task,
+                parentID: streamed.parentID, parentName: runtime?.parentID == streamed.parentID ? runtime?.parentName : nil,
+                harness: runtime?.harness ?? streamed.harness, model: runtime?.model ?? streamed.model,
+                createdAtMS: runtime?.createdAtMS ?? streamed.createdAtMS,
+                generationStartedAtMS: sameGeneration ? (runtime?.generationStartedAtMS ?? streamed.generationStartedAtMS) : streamed.generationStartedAtMS,
+                generationFinishedAtMS: sameGeneration ? runtime?.generationFinishedAtMS : streamed.generationFinishedAtMS
+            )
+            if status == .working, streamed.parentID == nil, source.canSteer(id: id, generation: streamed.generation) {
+                steerableGenerations[id] = streamed.generation
+            }
+        }
+        usageByID = usageByID.filter { rowsByID[$0.key] != nil }
+    }
+
+    mutating func applyStreamed(row: AgentRosterRow, canSteer: Bool) {
+        if let parentID = row.parentID { parentsByID[row.id] = parentID }
+        removeCleanedDescendants()
+        guard !cleanedIDs.contains(row.id) else { return }
+        rowsByID[row.id] = row
+        if canSteer { steerableGenerations[row.id] = row.generation }
+        else { steerableGenerations.removeValue(forKey: row.id) }
+    }
+
     func canSteer(id: String, generation: UInt64) -> Bool {
         steerableGenerations[id] == generation
     }
@@ -338,6 +387,26 @@ struct AgentRoster: Equatable {
     }
     private var versions: [String: Version] = [:]
     private var cleanedIDs: Set<String> = []
+    // Keep ownership independently of visible rows: cleanup can precede the
+    // first announcement, and a removed parent can acquire late descendants.
+    private var parentsByID: [String: String] = [:]
+    private var cleanedAncestors: Set<String> = []
+
+    private mutating func removeCleanedDescendants() {
+        guard !cleanedAncestors.isEmpty else { return }
+        while true {
+            let before = cleanedIDs.count
+            for (id, parentID) in parentsByID {
+                if cleanedAncestors.contains(parentID) || cleanedIDs.contains(parentID) {
+                    cleanedIDs.insert(id)
+                }
+            }
+            if cleanedIDs.count == before { break }
+        }
+        rowsByID = rowsByID.filter { !cleanedIDs.contains($0.key) }
+        usageByID = usageByID.filter { !cleanedIDs.contains($0.key) }
+        steerableGenerations = steerableGenerations.filter { !cleanedIDs.contains($0.key) }
+    }
 
     var nextFailedRemovalExpiryMS: UInt64? {
         rowsByID.values.compactMap { row in
@@ -395,7 +464,10 @@ struct AgentRoster: Equatable {
     mutating func apply(event: [String: Any], nowMS: UInt64) -> Bool {
         switch event["event"] as? String {
         case "subagent_state_changed":
-            guard let row = Self.parseRow(event), !cleanedIDs.contains(row.id) else { return false }
+            guard let row = Self.parseRow(event) else { return false }
+            if let parentID = row.parentID { parentsByID[row.id] = parentID }
+            removeCleanedDescendants()
+            guard !cleanedIDs.contains(row.id) else { return false }
             let incoming = Version(generation: row.generation, statusRank: row.status.rank)
             if let current = versions[row.id], incoming <= current { return false }
             versions[row.id] = incoming
@@ -433,18 +505,8 @@ struct AgentRoster: Equatable {
             return true
         case "subagent_descendants_removed":
             guard let ancestorID = event["ancestor_id"] as? String else { return false }
-            var removed: Set<String> = []
-            while true {
-                let before = removed.count
-                for row in rowsByID.values where row.id != ancestorID {
-                    if row.parentID == ancestorID || row.parentID.map(removed.contains) == true { removed.insert(row.id) }
-                }
-                if removed.count == before { break }
-            }
-            rowsByID = rowsByID.filter { !removed.contains($0.key) }
-            usageByID = usageByID.filter { !removed.contains($0.key) }
-            steerableGenerations = steerableGenerations.filter { !removed.contains($0.key) }
-            cleanedIDs.formUnion(removed)
+            cleanedAncestors.insert(ancestorID)
+            removeCleanedDescendants()
             return true
         default: return false
         }