Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
import Foundation

/// Follows a daemon's registered projects (protocol 13). The daemon sends the full list of
/// roots when the stream opens and again after every add or remove, so a project added or
/// removed from any device, or with `pu projects`, shows up here without polling.
/// Modeled after DaemonStatusSubscription; unlike it, it never gives up while running: a
/// restarted daemon is simply subscribed to again.
actor DaemonProjectsSubscription {
private var connection: (any DaemonByteStream)?
private var stopped = false
private let client: DaemonClient

private static let initialBackoff: UInt64 = 500_000_000 // 0.5s

init(client: DaemonClient) {
self.client = client
}

/// Delivers each list of roots until `stop()`. Reconnects with backoff after a failure.
func start(onChange: @escaping @MainActor ([String]) -> Void) async {
var backoff = Self.initialBackoff
let maxBackoff: UInt64 = 10_000_000_000 // 10s
while !stopped {
do {
try await run(onChange: onChange)
backoff = Self.initialBackoff
} catch is CancellationError {
return
} catch {
#if os(iOS)
if await AppActivity.shared.isInBackground {
await AppActivity.shared.untilForeground()
backoff = Self.initialBackoff
continue
}
#endif
}
guard !stopped else { return }
do { try await Task.sleep(nanoseconds: backoff) } catch { return }
backoff = min(backoff * 2, maxBackoff)
}
}

func stop() {
stopped = true
connection?.close()
connection = nil
}

private func run(onChange: @escaping @MainActor ([String]) -> Void) async throws {
let (conn, reader) = try await client.connect()
// Stopped while connecting: nothing else would ever close this connection.
guard !stopped else {
conn.close()
return
}
connection?.close()
connection = conn
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
try await DaemonClient.write(.subscribeProjects, to: conn)
guard case .projectsSubscribed = DaemonClient.parse(try await reader.readLine()) else {
throw DaemonStatusError.unexpectedResponse
}
while !stopped {
if case .projectsChanged(let projects) = DaemonClient.parse(try await reader.readLine()) {
await onChange(projects)
}
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -299,7 +299,12 @@ nonisolated enum DaemonRequest: Encodable {
case channelEdit(projectRoot: String, messageId: String, text: String)
case channelReact(projectRoot: String, messageId: String, active: Bool)
case health
/// Adds the folder to the daemon's projects, initializing it if needed.
case initProject(projectRoot: String)
/// Forgets a project on the daemon for every device; its files are untouched (protocol 13).
case removeProject(projectRoot: String, suspendAgents: Bool = false)
/// Streams the daemon's registered project roots, now and on every change (protocol 13).
case subscribeProjects
case status(projectRoot: String, agentId: String? = nil)
case attach(agentId: String, clientId: String? = nil)
case input(agentId: String, data: Data)
Expand Down Expand Up @@ -405,6 +410,12 @@ nonisolated enum DaemonRequest: Encodable {
case .initProject(let projectRoot):
try container.encode("init", forKey: .key("type"))
try container.encode(projectRoot, forKey: .key("project_root"))
case .removeProject(let projectRoot, let suspendAgents):
try container.encode("remove_project", forKey: .key("type"))
try container.encode(projectRoot, forKey: .key("project_root"))
try container.encode(suspendAgents, forKey: .key("suspend_agents"))
case .subscribeProjects:
try container.encode("subscribe_projects", forKey: .key("type"))
case .status(let projectRoot, let agentId):
try container.encode("status", forKey: .key("type"))
try container.encode(projectRoot, forKey: .key("project_root"))
Expand Down Expand Up @@ -665,6 +676,9 @@ nonisolated enum DaemonResponse: Decodable {
case channelMessage(ChannelMutation)
case healthReport(pid: Int, uptimeSeconds: Int, protocolVersion: Int, agentCount: Int)
case initResult(created: Bool)
case projectRemoved(projectRoot: String, removed: Bool, suspended: [String], killed: [String])
case projectsSubscribed
case projectsChanged(projects: [String])
case statusReport(worktrees: [WorktreeEntry], agents: [AgentStatusReport])
case resync
case ptySize(cols: Int, rows: Int, ownerClientId: String?)
Expand Down Expand Up @@ -729,6 +743,14 @@ nonisolated enum DaemonResponse: Decodable {
case "init_result":
let p = try InitResultPayload(from: decoder)
self = .initResult(created: p.created)
case "project_removed":
let p = try ProjectRemovedPayload(from: decoder)
self = .projectRemoved(
projectRoot: p.projectRoot, removed: p.removed, suspended: p.suspended, killed: p.killed ?? [])
case "projects_subscribed":
self = .projectsSubscribed
case "projects_changed":
self = .projectsChanged(projects: try ProjectsChangedPayload(from: decoder).projects)
case "status_report":
let p = try StatusReportPayload(from: decoder)
self = .statusReport(worktrees: p.worktrees, agents: p.agents)
Expand Down Expand Up @@ -1030,6 +1052,25 @@ private nonisolated struct InitResultPayload: Decodable {
let created: Bool
}

private nonisolated struct ProjectRemovedPayload: Decodable {
let projectRoot: String
let removed: Bool
let suspended: [String]
/// Sessions the manifest did not list, stopped rather than suspended.
let killed: [String]?

enum CodingKeys: String, CodingKey {
case projectRoot = "project_root"
case removed
case suspended
case killed
}
}

private nonisolated struct ProjectsChangedPayload: Decodable {
let projects: [String]
}

private nonisolated struct HealthReportPayload: Decodable {
let pid: Int
let uptimeSeconds: Int
Expand Down Expand Up @@ -1212,7 +1253,7 @@ private nonisolated struct DynamicCodingKey: CodingKey {

nonisolated enum DaemonProtocol {
/// The protocol this app speaks. A daemon reporting another version is incompatible.
static let version = 12
static let version = 13
}

nonisolated struct InventoryProjectPayload: Decodable, Sendable, Equatable {
Expand Down
Loading
Loading