ai/byre
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 14 Wiki Activity

Load only active tasks on connect and refetch single tasks on events

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-22 11:43:35 -07:00
parent 78fcdfff34
commit 48741e8cbb
5 changed files with 186 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mac/Sources/Byre/Store.swift
+83 -37
View File
@@ -97,7 +97,8 @@ final class Store {
@ObservationIgnored private var historyLoads: Set<ObjectIdentifier> = []
@ObservationIgnored private var requestSeqs: [String: Int] = [:]
@ObservationIgnored private var taskReloads: Set<String> = []
@ObservationIgnored private var taskReloadAgain: Set<String> = []
@ObservationIgnored private var listReloads: Set<String> = []
@ObservationIgnored private var taskRefetches: [String: Set<String>] = [:]
@ObservationIgnored private var terminalChanges = 0
@ObservationIgnored private var streamOpens = 0
@ObservationIgnored private var globalStream: EventStream?
@@ -368,12 +369,70 @@ final class Store {
}
func loadTasks(_ projectId: String) async throws {
let latest = claimRequest("tasks:\(projectId)")
let list = try await api.client.listTasks(path: .init(projectId: projectId)).ok.body.json
if !latest() {
try await reloadTasks(projectId, list: true)
}
// Loads run one at a time per project, so a slower, older response never lands over a newer one.
private func reloadTasks(_ projectId: String, list: Bool = false, task: String? = nil) async throws {
if list {
listReloads.insert(projectId)
}
if let task {
taskRefetches[projectId, default: []].insert(task)
}
guard taskReloads.insert(projectId).inserted else {
return
}
tasks[projectId] = list
defer {
taskReloads.remove(projectId)
}
var failure: (any Error)?
while listReloads.contains(projectId) || taskRefetches[projectId] != nil {
do {
if listReloads.remove(projectId) != nil {
try await loadActiveTasks(projectId)
} else {
try await refetchTasks(projectId, taskRefetches.removeValue(forKey: projectId)!)
}
} catch {
failure = error
}
}
if let failure {
throw failure
}
}
private func loadActiveTasks(_ projectId: String) async throws {
let list = try await api.client.listActiveTasks(path: .init(projectId: projectId)).ok.body.json
let fresh = Set(list.map(\.id))
let held = (tasks[projectId] ?? []).filter { !fresh.contains($0.id) }
for t in held where t.status == .active {
taskRefetches[projectId, default: []].insert(t.id)
}
try await applyTasks(projectId, list + held)
}
private func refetchTasks(_ projectId: String, _ ids: Set<String>) async throws {
let api = api
let fetched = try await withThrowingTaskGroup(of: ByreTask.self) { group in
for id in ids {
group.addTask {
try await api.client.getTask(path: .init(taskId: id)).ok.body.json
}
}
var out: [ByreTask] = []
for try await t in group {
out.append(t)
}
return out
}
let fresh = Set(fetched.map(\.id))
try await applyTasks(projectId, fetched + (tasks[projectId] ?? []).filter { !fresh.contains($0.id) })
}
private func applyTasks(_ projectId: String, _ list: [ByreTask]) async throws {
tasks[projectId] = list.sorted { $0.createdAt > $1.createdAt }
for t in list where t.status == .active && !t.background {
_ = taskFeed(t.id)
}
@@ -384,26 +443,15 @@ final class Store {
}
}
private func reloadTasks(_ projectId: String) async throws {
guard taskReloads.insert(projectId).inserted else {
taskReloadAgain.insert(projectId)
private func lookUpTask(_ taskId: String) {
if findTask(taskId) != nil {
return
}
defer {
taskReloads.remove(projectId)
taskReloadAgain.remove(projectId)
}
var failure: (any Error)?
repeat {
taskReloadAgain.remove(projectId)
do {
try await loadTasks(projectId)
} catch {
failure = error
spawn("load task") {
let t = try await self.api.client.getTask(path: .init(taskId: taskId)).ok.body.json
if self.findTask(taskId) == nil {
self.tasks[t.projectId] = ((self.tasks[t.projectId] ?? []) + [t]).sorted { $0.createdAt > $1.createdAt }
}
} while taskReloadAgain.contains(projectId)
if let failure {
throw failure
}
}
@@ -674,6 +722,7 @@ final class Store {
markVisited(taskId)
selectedTaskId = taskId
if let taskId {
lookUpTask(taskId)
_ = taskFeed(taskId)
}
pinned = nil
@@ -798,11 +847,7 @@ final class Store {
forwards.close(taskId)
}
notifyTask(self, ev)
if let projectId = ev.projectId {
spawn("load tasks") {
try await self.reloadTasks(projectId)
}
}
reloadTask(ev)
case "session.created", "session.state", "session.subagents", "session.model":
if ev.owner != nil {
spawn("load lead sessions") {
@@ -821,11 +866,7 @@ final class Store {
spawn("load inbox") {
try await self.loadInbox()
}
if let projectId = ev.projectId {
spawn("load tasks") {
try await self.reloadTasks(projectId)
}
}
reloadTask(ev)
case "inbox.updated":
clearInbox(try! decodePayload(ev.payload, as: InboxItem.self))
spawn("load inbox") {
@@ -839,11 +880,7 @@ final class Store {
}
case "session.error", "session.rate_limited":
notifyError(self, ev)
if let projectId = ev.projectId {
spawn("load tasks") {
try await self.reloadTasks(projectId)
}
}
reloadTask(ev)
case "task.merge_facts":
if let projectId = ev.projectId, let taskId = ev.taskId, let i = tasks[projectId]?.firstIndex(where: { $0.id == taskId }) {
tasks[projectId]![i].mergeFacts = try! decodePayload(ev.payload, as: MergeFacts.self)
@@ -876,6 +913,15 @@ final class Store {
events.send(ev)
}
private func reloadTask(_ ev: ByreEvent) {
guard let projectId = ev.projectId, let taskId = ev.taskId else {
return
}
spawn("load task") {
try await self.reloadTasks(projectId, task: taskId)
}
}
private func updateTask(_ id: String, _ change: (inout ByreTask) -> Void) {
for (projectId, list) in tasks {
if let i = list.firstIndex(where: { $0.id == id }) {