| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "reflect" |
| 5 | "sort" |
| 6 | "strings" |
| 7 | "sync" |
| 8 | |
| 9 | "reasonix/internal/control" |
| 10 | "reasonix/internal/event" |
| 11 | ) |
| 12 | |
| 13 | type RuntimeSessionState struct { |
| 14 | TabID string `json:"tabId"` |
| 15 | Scope string `json:"scope"` |
| 16 | WorkspaceRoot string `json:"workspaceRoot"` |
| 17 | TopicID string `json:"topicId"` |
| 18 | SessionID string `json:"sessionId,omitempty"` |
| 19 | SessionPath string `json:"sessionPath"` |
| 20 | SessionGeneration uint64 `json:"sessionGeneration"` |
| 21 | Open bool `json:"open"` |
| 22 | Remote bool `json:"remote"` |
| 23 | HostID string `json:"hostId,omitempty"` |
| 24 | Freshness string `json:"freshness"` |
| 25 | State event.RuntimeStateSnapshot `json:"state"` |
| 26 | } |
| 27 | |
| 28 | type RuntimeStateProjection struct { |
| 29 | Epoch string `json:"epoch"` |
| 30 | Revision uint64 `json:"revision"` |
| 31 | Sessions []RuntimeSessionState `json:"sessions"` |
| 32 | Topics []ProjectRuntimeTopic `json:"topics"` |
| 33 | } |
| 34 | |
| 35 | type desktopRuntimeProjection struct { |
| 36 | mu sync.Mutex |
| 37 | snapshot RuntimeStateProjection |
| 38 | } |
| 39 | |
| 40 | type localRuntimeBindingKey struct { |
| 41 | key string |
| 42 | open bool |
| 43 | } |
| 44 | |
| 45 | type localRuntimeBinding struct { |
| 46 | tab *WorkspaceTab |
| 47 | view RuntimeSessionState |
| 48 | ctrl control.SessionAPI |
| 49 | catalog catalogRuntimeSnapshot |
| 50 | } |
| 51 | |
| 52 | // localRuntimeBindingsLocked copies identity and display metadata as one |
| 53 | // binding. App.mu must be held; controller methods must stay outside that lock. |
| 54 | func (a *App) localRuntimeBindingsLocked() map[localRuntimeBindingKey]localRuntimeBinding { |
| 55 | bindings := make(map[localRuntimeBindingKey]localRuntimeBinding, len(a.tabs)+len(a.detachedSessions)) |
| 56 | collect := func(key string, tab *WorkspaceTab, open bool) { |
| 57 | if tab == nil { |
| 58 | return |
| 59 | } |
| 60 | bindings[localRuntimeBindingKey{key, open}] = localRuntimeBinding{tab: tab, ctrl: tab.Ctrl, |
| 61 | view: RuntimeSessionState{TabID: tab.ID, Scope: tab.Scope, WorkspaceRoot: tab.WorkspaceRoot, |
| 62 | TopicID: tab.TopicID, SessionID: tab.SessionID, SessionPath: tab.SessionPath, SessionGeneration: tab.SessionGeneration, Open: open, Freshness: "synced"}, |
| 63 | catalog: catalogRuntimeSnapshot{tabID: tab.ID, scope: tab.Scope, workspaceRoot: tab.WorkspaceRoot, topicID: tab.TopicID, sessionPath: tab.SessionPath, |
| 64 | activity: tab.ActivityStatus, topicTitle: tab.TopicTitle, topicTitleSource: tab.topicTitleSource, open: open}} |
| 65 | if tab.SessionID != "" { |
| 66 | binding := bindings[localRuntimeBindingKey{key, open}] |
| 67 | binding.catalog.sessionPath = sessionRoute(tab.SessionID) |
| 68 | bindings[localRuntimeBindingKey{key, open}] = binding |
| 69 | } |
| 70 | } |
| 71 | for key, tab := range a.tabs { |
| 72 | collect(key, tab, true) |
| 73 | } |
| 74 | for key, tab := range a.detachedSessions { |
| 75 | collect(key, tab, false) |
| 76 | } |
| 77 | return bindings |
| 78 | } |
| 79 | |
| 80 | func (a *App) sampleLocalRuntimeBindings() []localRuntimeBinding { |
| 81 | for { |
| 82 | a.mu.RLock() |
| 83 | bindings := a.localRuntimeBindingsLocked() |
| 84 | a.mu.RUnlock() |
| 85 | states := make(map[localRuntimeBindingKey]event.RuntimeStateSnapshot, len(bindings)) |
| 86 | for key, binding := range bindings { |
| 87 | states[key] = controllerRuntimeState(binding.ctrl) |
| 88 | } |
| 89 | // A controller can rotate its session, be replaced, or move between |
| 90 | // open and detached while sampled. Retry the binding set so its state |
| 91 | // cannot be published under the previous session identity. |
| 92 | a.mu.RLock() |
| 93 | current := a.localRuntimeBindingsLocked() |
| 94 | valid := len(current) == len(bindings) |
| 95 | for key, binding := range bindings { |
| 96 | if !sameLocalRuntimeBinding(current[key], binding) { |
| 97 | valid = false |
| 98 | break |
| 99 | } |
| 100 | } |
| 101 | a.mu.RUnlock() |
| 102 | if !valid { |
| 103 | continue |
| 104 | } |
| 105 | result := make([]localRuntimeBinding, 0, len(bindings)) |
| 106 | for key, binding := range bindings { |
| 107 | binding.view.State = states[key] |
| 108 | result = append(result, binding) |
| 109 | } |
| 110 | return result |
| 111 | } |
| 112 | } |
| 113 | |
| 114 | // sameLocalRuntimeBinding compares the immutable identity and display fields |
| 115 | // copied while App.mu was held. reflect.DeepEqual is deliberately unsuitable |
| 116 | // here: following tab or controller pointers recursively reads their live |
| 117 | // mutex/atomic state and races the controller's runtime-state publisher. |
| 118 | func sameLocalRuntimeBinding(current, sampled localRuntimeBinding) bool { |
| 119 | return current.tab == sampled.tab && |
| 120 | sameSessionAPI(current.ctrl, sampled.ctrl) && |
| 121 | current.view.TabID == sampled.view.TabID && |
| 122 | current.view.Scope == sampled.view.Scope && |
| 123 | current.view.WorkspaceRoot == sampled.view.WorkspaceRoot && |
| 124 | current.view.TopicID == sampled.view.TopicID && |
| 125 | current.view.SessionID == sampled.view.SessionID && |
| 126 | current.view.SessionPath == sampled.view.SessionPath && |
| 127 | current.view.SessionGeneration == sampled.view.SessionGeneration && |
| 128 | current.view.Open == sampled.view.Open && |
| 129 | current.view.Remote == sampled.view.Remote && |
| 130 | current.view.HostID == sampled.view.HostID && |
| 131 | current.view.Freshness == sampled.view.Freshness && |
| 132 | current.catalog.scope == sampled.catalog.scope && |
| 133 | current.catalog.workspaceRoot == sampled.catalog.workspaceRoot && |
| 134 | current.catalog.topicID == sampled.catalog.topicID && |
| 135 | current.catalog.sessionPath == sampled.catalog.sessionPath && |
| 136 | current.catalog.activity == sampled.catalog.activity && |
| 137 | current.catalog.topicTitle == sampled.catalog.topicTitle && |
| 138 | current.catalog.topicTitleSource == sampled.catalog.topicTitleSource && |
| 139 | current.catalog.open == sampled.catalog.open |
| 140 | } |
| 141 | |
| 142 | func sameSessionAPI(current, sampled control.SessionAPI) bool { |
| 143 | if current == nil || sampled == nil { |
| 144 | return current == nil && sampled == nil |
| 145 | } |
| 146 | currentValue, sampledValue := reflect.ValueOf(current), reflect.ValueOf(sampled) |
| 147 | if currentValue.Type() != sampledValue.Type() { |
| 148 | return false |
| 149 | } |
| 150 | if currentValue.Type().Comparable() { |
| 151 | return currentValue.Interface() == sampledValue.Interface() |
| 152 | } |
| 153 | return false |
| 154 | } |
| 155 | |
| 156 | func controllerRuntimeState(ctrl control.SessionAPI) event.RuntimeStateSnapshot { |
| 157 | if reader, ok := ctrl.(control.RuntimeStateReader); ok { |
| 158 | return reader.RuntimeStateSnapshot() |
| 159 | } |
| 160 | if ctrl == nil { |
| 161 | return event.RuntimeStateSnapshot{Phase: "idle"} |
| 162 | } |
| 163 | legacy := ctrl.RuntimeStatus() |
| 164 | phase := "idle" |
| 165 | if legacy.Running { |
| 166 | phase = "executing" |
| 167 | } |
| 168 | return event.RuntimeStateSnapshot{Phase: phase, Running: legacy.Running, PendingPrompt: legacy.PendingPrompt, |
| 169 | BackgroundJobs: legacy.BackgroundJobs, Cancellable: legacy.Cancellable, CancelRequested: legacy.CancelRequested, |
| 170 | TurnID: legacy.TurnID, TurnStatus: legacy.Status, TurnEventSeq: legacy.TurnEventSeq} |
| 171 | } |
| 172 | |
| 173 | func runtimeDisplayStatus(state event.RuntimeStateSnapshot, result string) string { |
| 174 | switch { |
| 175 | case state.Phase == "finishing": |
| 176 | return "finishing" |
| 177 | case state.CancelRequested: |
| 178 | return "cancelling" |
| 179 | case state.PendingPrompt: |
| 180 | return topicStatusWaitingConfirmation |
| 181 | case state.Phase == "executing": |
| 182 | if state.Activity == "streaming" { |
| 183 | return topicStatusStreaming |
| 184 | } |
| 185 | return topicStatusThinking |
| 186 | case state.BackgroundJobs > 0: |
| 187 | return topicStatusBackgroundJob |
| 188 | } |
| 189 | if result == topicStatusError || result == topicStatusPaused || result == topicStatusAwaitingDelivery { |
| 190 | return result |
| 191 | } |
| 192 | return "" |
| 193 | } |
| 194 | |
| 195 | func catalogControllerStatus(ctrl control.SessionAPI, activity string) (string, bool) { |
| 196 | state := controllerRuntimeState(ctrl) |
| 197 | return catalogStateStatus(state, activity) |
| 198 | } |
| 199 | |
| 200 | func catalogStateStatus(state event.RuntimeStateSnapshot, activity string) (string, bool) { |
| 201 | if state.SchemaVersion == 1 { |
| 202 | return runtimeDisplayStatus(state, activity), state.ActiveWork() |
| 203 | } |
| 204 | legacy := control.RuntimeStatus{Running: state.Running, PendingPrompt: state.PendingPrompt, BackgroundJobs: state.BackgroundJobs} |
| 205 | status := catalogRuntimeStatus(activity, legacy) |
| 206 | return status, status != "" || state.ActiveWork() |
| 207 | } |
| 208 | |
| 209 | // GetRuntimeStateSnapshot reads committed controller snapshots after copying |
| 210 | // bindings off App.mu. Controller and remote-tab sampling stay outside the |
| 211 | // projection mutex: archive and runtime-state callbacks also enter here, and |
| 212 | // holding that mutex across a controller read deadlocks a running turn. |
| 213 | func (a *App) GetRuntimeStateSnapshot() RuntimeStateProjection { |
| 214 | bindings := a.sampleLocalRuntimeBindings() |
| 215 | remote := a.sampleRemoteRuntimeSessions() |
| 216 | r := &a.runtimeStateProjection |
| 217 | r.mu.Lock() |
| 218 | defer r.mu.Unlock() |
| 219 | next := RuntimeStateProjection{Epoch: r.snapshot.Epoch, Sessions: []RuntimeSessionState{}} |
| 220 | catalog := []catalogRuntimeSnapshot{} |
| 221 | if next.Epoch == "" { |
| 222 | next.Epoch = newSessionRuntimeID("projection") |
| 223 | } |
| 224 | for _, binding := range bindings { |
| 225 | view := binding.view |
| 226 | next.Sessions = append(next.Sessions, view) |
| 227 | if binding.catalog.topicID != "" { |
| 228 | entry := binding.catalog |
| 229 | entry.state = &view.State |
| 230 | catalog = append(catalog, entry) |
| 231 | } |
| 232 | } |
| 233 | next.Topics = a.projectTreeRuntimeTopics(catalog) |
| 234 | next.Sessions = append(next.Sessions, remote...) |
| 235 | sort.Slice(next.Sessions, func(i, j int) bool { |
| 236 | if next.Sessions[i].TabID == next.Sessions[j].TabID { |
| 237 | return next.Sessions[i].SessionPath < next.Sessions[j].SessionPath |
| 238 | } |
| 239 | return next.Sessions[i].TabID < next.Sessions[j].TabID |
| 240 | }) |
| 241 | next.Revision = r.snapshot.Revision |
| 242 | if !reflect.DeepEqual(next, r.snapshot) { |
| 243 | next.Revision++ |
| 244 | r.snapshot = next |
| 245 | } |
| 246 | result := r.snapshot |
| 247 | result.Sessions = append([]RuntimeSessionState{}, result.Sessions...) |
| 248 | result.Topics = cloneRuntimeTopics(result.Topics) |
| 249 | return result |
| 250 | } |
| 251 | |
| 252 | func (a *App) sampleRemoteRuntimeSessions() []RuntimeSessionState { |
| 253 | if a == nil { |
| 254 | return nil |
| 255 | } |
| 256 | a.remoteTabMu.Lock() |
| 257 | defer a.remoteTabMu.Unlock() |
| 258 | sessions := make([]RuntimeSessionState, 0) |
| 259 | for _, tab := range a.remoteTabs { |
| 260 | freshness := "synced" |
| 261 | if tab.state != "ready" || tab.session.takenOver || tab.runtime.syncFailed || tab.runtimeUnknown[tab.routing.currentPath] != 0 { |
| 262 | freshness = "unknown" |
| 263 | } |
| 264 | state := tab.runtimeStates[tab.routing.currentPath] |
| 265 | if state.SchemaVersion == 0 { |
| 266 | state = event.RuntimeStateSnapshot{Phase: "idle", Running: tab.runtime.running, PendingPrompt: tab.runtime.pendingPrompt, |
| 267 | BackgroundJobs: tab.runtime.backgroundJobs, Cancellable: tab.runtime.cancellable, CancelRequested: tab.runtime.cancelRequested} |
| 268 | if state.Running { |
| 269 | state.Phase = "executing" |
| 270 | } |
| 271 | } |
| 272 | sessions = append(sessions, RuntimeSessionState{TabID: tab.id, Scope: "remote", HostID: tab.ref.HostID, WorkspaceRoot: tab.ref.Workspace, |
| 273 | SessionID: remoteRuntimeSessionID(tab.routing.currentPath, tab.session.sessionID, state), SessionPath: tab.routing.currentPath, |
| 274 | Open: true, Remote: true, Freshness: freshness, State: state}) |
| 275 | for path, background := range tab.runtimeStates { |
| 276 | if path == tab.routing.currentPath { |
| 277 | continue |
| 278 | } |
| 279 | freshness := "synced" |
| 280 | // A foreground takeover says nothing about another session, but a |
| 281 | // tab without a live stream or with a failed sync only holds the |
| 282 | // snapshot frozen at its last observation. |
| 283 | if tab.state != "ready" || tab.runtime.syncFailed || tab.runtimeUnknown[path] != 0 { |
| 284 | freshness = "unknown" |
| 285 | } |
| 286 | sessions = append(sessions, RuntimeSessionState{TabID: tab.id, Scope: "remote", HostID: tab.ref.HostID, WorkspaceRoot: tab.ref.Workspace, |
| 287 | SessionID: remoteRuntimeSessionID(path, "", background), SessionPath: path, Remote: true, Freshness: freshness, State: background}) |
| 288 | } |
| 289 | } |
| 290 | return sessions |
| 291 | } |
| 292 | |
| 293 | func remoteRuntimeSessionID(route, fallback string, state event.RuntimeStateSnapshot) string { |
| 294 | if id := strings.TrimSpace(state.SessionID); id != "" { |
| 295 | return id |
| 296 | } |
| 297 | if id, ok := parseSessionRoute(route); ok { |
| 298 | return id |
| 299 | } |
| 300 | return strings.TrimSpace(fallback) |
| 301 | } |
| 302 | |
| 303 | func (a *App) emitRuntimeStateChanged() { |
| 304 | if a != nil { |
| 305 | a.emitRuntimeEvent("runtime-state:changed", a.GetRuntimeStateSnapshot()) |
| 306 | } |
| 307 | } |
| 308 | |
| 309 | func (s *tabEventSink) RuntimeStateChanged(snapshot event.RuntimeStateSnapshot) { |
| 310 | id, app := s.binding() |
| 311 | if app == nil { |
| 312 | return |
| 313 | } |
| 314 | app.mu.RLock() |
| 315 | tab := app.tabByEventSinkIDLocked(id) |
| 316 | var ctrl control.SessionAPI |
| 317 | if tab != nil { |
| 318 | ctrl = tab.Ctrl |
| 319 | } |
| 320 | app.mu.RUnlock() |
| 321 | if ctrl == nil { |
| 322 | return |
| 323 | } |
| 324 | current := controllerRuntimeState(ctrl) |
| 325 | if current.RuntimeEpoch != snapshot.RuntimeEpoch || current.Revision > snapshot.Revision { |
| 326 | return |
| 327 | } |
| 328 | app.emitProjectTreeRuntimeChangedWithLegacy() |
| 329 | } |
| 330 |