| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "crypto/rand" |
| 5 | "encoding/hex" |
| 6 | "encoding/json" |
| 7 | "net/http" |
| 8 | "reflect" |
| 9 | "sort" |
| 10 | "sync" |
| 11 | |
| 12 | "reasonix/internal/agent" |
| 13 | "reasonix/internal/control" |
| 14 | "reasonix/internal/event" |
| 15 | "reasonix/internal/eventwire" |
| 16 | ) |
| 17 | |
| 18 | type runtimeSessionView struct { |
| 19 | SessionPath string `json:"sessionPath"` |
| 20 | Current bool `json:"current"` |
| 21 | State event.RuntimeStateSnapshot `json:"state"` |
| 22 | } |
| 23 | |
| 24 | func runtimeStateAndStatus(ctrl control.SessionAPI) (event.RuntimeStateSnapshot, control.RuntimeStatus) { |
| 25 | state, status := runtimeStateOf(ctrl), ctrl.RuntimeStatus() |
| 26 | if state.SchemaVersion == 1 { |
| 27 | status.Running, status.PendingPrompt, status.BackgroundJobs, status.CancelRequested, status.Cancellable = state.Running, state.PendingPrompt, state.BackgroundJobs, state.CancelRequested, state.Cancellable |
| 28 | } |
| 29 | return state, status |
| 30 | } |
| 31 | |
| 32 | type runtimeStatesView struct { |
| 33 | SchemaVersion int `json:"schemaVersion"` |
| 34 | Epoch string `json:"epoch"` |
| 35 | Revision uint64 `json:"revision"` |
| 36 | Sessions []runtimeSessionView `json:"sessions"` |
| 37 | } |
| 38 | type serveRuntimeProjection struct { |
| 39 | mu sync.Mutex |
| 40 | snapshot runtimeStatesView |
| 41 | } |
| 42 | |
| 43 | func runtimeStateOf(ctrl control.SessionAPI) event.RuntimeStateSnapshot { |
| 44 | if reader, ok := ctrl.(control.RuntimeStateReader); ok { |
| 45 | return reader.RuntimeStateSnapshot() |
| 46 | } |
| 47 | status := ctrl.RuntimeStatus() |
| 48 | phase := "idle" |
| 49 | if status.Running { |
| 50 | phase = "executing" |
| 51 | } |
| 52 | return event.RuntimeStateSnapshot{Phase: phase, Running: status.Running, PendingPrompt: status.PendingPrompt, |
| 53 | BackgroundJobs: status.BackgroundJobs, CancelRequested: status.CancelRequested, Cancellable: status.Cancellable} |
| 54 | } |
| 55 | |
| 56 | // runtimeStates is a memory-only reconciliation surface, including detached |
| 57 | // controllers. It does not list transcripts, generate titles, or fetch balance. |
| 58 | func (s *Server) runtimeStates(w http.ResponseWriter, _ *http.Request) { |
| 59 | writeJSON(w, s.runtimeStatesSnapshot()) |
| 60 | } |
| 61 | |
| 62 | func (s *Server) runtimeStatesSnapshot() runtimeStatesView { |
| 63 | r := &s.runtimeProjection |
| 64 | r.mu.Lock() |
| 65 | defer r.mu.Unlock() |
| 66 | s.bindMu.Lock() |
| 67 | current := s.ctl() |
| 68 | controllers := []control.SessionAPI{current} |
| 69 | s.detachedMu.Lock() |
| 70 | for _, detached := range s.detached { |
| 71 | if detached.ctrl != current { |
| 72 | controllers = append(controllers, detached.ctrl) |
| 73 | } |
| 74 | } |
| 75 | s.detachedMu.Unlock() |
| 76 | result := runtimeStatesView{SchemaVersion: 1, Epoch: r.snapshot.Epoch, Revision: r.snapshot.Revision, Sessions: []runtimeSessionView{}} |
| 77 | for _, ctrl := range controllers { |
| 78 | if ctrl == nil { |
| 79 | continue |
| 80 | } |
| 81 | // Identity sessions carry no legacy path; projecting their route |
| 82 | // reference keeps the desktop's path-keyed reconciliation working |
| 83 | // while hosts complete the catalog transition. |
| 84 | path := agent.CanonicalSessionPath(ctrl.SessionPath()) |
| 85 | if path == "" { |
| 86 | if identity, ok := ctrl.(control.IdentityLifecycle); ok { |
| 87 | if ref, bound := identity.SessionRef(); bound { |
| 88 | path = remoteSessionIDQueryPrefix + ref.SessionID |
| 89 | } |
| 90 | } |
| 91 | } |
| 92 | result.Sessions = append(result.Sessions, runtimeSessionView{SessionPath: path, Current: ctrl == current, State: runtimeStateOf(ctrl)}) |
| 93 | } |
| 94 | s.bindMu.Unlock() |
| 95 | sort.Slice(result.Sessions, func(i, j int) bool { |
| 96 | return result.Sessions[i].State.RuntimeEpoch < result.Sessions[j].State.RuntimeEpoch |
| 97 | }) |
| 98 | if result.Epoch == "" { |
| 99 | var id [16]byte |
| 100 | if _, err := rand.Read(id[:]); err != nil { |
| 101 | panic(err) |
| 102 | } |
| 103 | result.Epoch = hex.EncodeToString(id[:]) |
| 104 | } |
| 105 | if !reflect.DeepEqual(result, r.snapshot) { |
| 106 | result.Revision++ |
| 107 | r.snapshot = result |
| 108 | } |
| 109 | result.Sessions = append([]runtimeSessionView{}, r.snapshot.Sessions...) |
| 110 | return result |
| 111 | } |
| 112 | |
| 113 | func (b *Broadcaster) RuntimeStateChanged(snapshot event.RuntimeStateSnapshot) { |
| 114 | b.publishRuntimeState(b.CurrentSession(), snapshot) |
| 115 | } |
| 116 | func (b *Broadcaster) publishRuntimeState(path string, snapshot event.RuntimeStateSnapshot) { |
| 117 | path = agent.CanonicalSessionPath(path) |
| 118 | b.mu.Lock() |
| 119 | defer b.mu.Unlock() |
| 120 | frame, err := json.Marshal(eventwire.Event{Kind: "runtime_state", SessionPath: path, SessionCurrent: path == b.current, RuntimeState: &snapshot}) |
| 121 | if err != nil { |
| 122 | return |
| 123 | } |
| 124 | for ch, sub := range b.subs { |
| 125 | if !sub.all && path != "" && path != b.current { |
| 126 | continue |
| 127 | } |
| 128 | enqueueSubscriberWireFrame(ch, frame, "runtime_state") |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | func (s *sessionTagSink) RuntimeStateChanged(snapshot event.RuntimeStateSnapshot) { |
| 133 | s.mu.Lock() |
| 134 | defer s.mu.Unlock() |
| 135 | if !s.active || !s.runtimeActive { |
| 136 | s.pendingRuntimeState = &snapshot |
| 137 | return |
| 138 | } |
| 139 | s.bc.publishRuntimeState(s.path, snapshot) |
| 140 | } |
| 141 | |
| 142 | // A per-session status query must resolve its own controller, not the foreground. |
| 143 | func (s *Server) ownedRuntimeStatusView(path string) (map[string]any, bool) { |
| 144 | path = agent.CanonicalSessionPath(path) |
| 145 | s.bindMu.Lock() |
| 146 | defer s.bindMu.Unlock() |
| 147 | ctrl := s.ctl() |
| 148 | if ctrl == nil || agent.CanonicalSessionPath(ctrl.SessionPath()) != path { |
| 149 | s.detachedMu.Lock() |
| 150 | detached := s.detached[path] |
| 151 | ctrl = nil |
| 152 | if detached != nil { |
| 153 | ctrl = detached.ctrl |
| 154 | } |
| 155 | s.detachedMu.Unlock() |
| 156 | } |
| 157 | if ctrl == nil { |
| 158 | return nil, false |
| 159 | } |
| 160 | state := runtimeStateOf(ctrl) |
| 161 | return map[string]any{"runtimeState": state, "running": state.Running, "pendingPrompt": state.PendingPrompt, |
| 162 | "backgroundJobs": state.BackgroundJobs, "cancelRequested": state.CancelRequested, "cancellable": state.Cancellable, |
| 163 | "sessionPath": path, "takenOver": false, "label": ctrl.Label(), "plan": ctrl.PlanMode(), "toolApprovalMode": ctrl.ToolApprovalMode(), "goal": ctrl.Goal()}, true |
| 164 | } |
| 165 |