| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "net/http" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "sort" |
| 9 | "strings" |
| 10 | |
| 11 | "reasonix/internal/agent" |
| 12 | "reasonix/internal/control" |
| 13 | "reasonix/internal/session" |
| 14 | "reasonix/internal/store" |
| 15 | ) |
| 16 | |
| 17 | type sessionListEntry struct { |
| 18 | HostID string `json:"hostId,omitempty"` |
| 19 | SessionID string `json:"sessionId,omitempty"` |
| 20 | Name string `json:"name"` |
| 21 | Path string `json:"path"` |
| 22 | Title string `json:"title,omitempty"` |
| 23 | Turns int `json:"turns,omitempty"` |
| 24 | Current bool `json:"current,omitempty"` |
| 25 | Running bool `json:"running,omitempty"` |
| 26 | TakenOver bool `json:"takenOver,omitempty"` |
| 27 | MtimeMilli int64 `json:"mtimeMilli"` |
| 28 | |
| 29 | Preview string `json:"preview,omitempty"` |
| 30 | MetadataReady bool `json:"metadataReady,omitempty"` |
| 31 | } |
| 32 | |
| 33 | // sessions lists saved sessions with event-log-aware titles and turn counts. |
| 34 | func (s *Server) sessions(w http.ResponseWriter, r *http.Request) { |
| 35 | ctrl := s.ctl() |
| 36 | entries, ok := readSessionDir(ctrl.SessionDir()) |
| 37 | if !ok { |
| 38 | writeJSON(w, []any{}) |
| 39 | return |
| 40 | } |
| 41 | out := mergeSessionRows(s.legacySessionRows(r, ctrl, entries)) |
| 42 | sort.SliceStable(out, func(i, j int) bool { return out[i].MtimeMilli > out[j].MtimeMilli }) |
| 43 | writeJSON(w, out) |
| 44 | } |
| 45 | |
| 46 | // readSessionDir reports ok=false only when the directory exists but cannot be |
| 47 | // read; a missing directory is an empty session list, not a failure. |
| 48 | func readSessionDir(dir string) ([]os.DirEntry, bool) { |
| 49 | if dir == "" { |
| 50 | return nil, true |
| 51 | } |
| 52 | entries, err := os.ReadDir(dir) |
| 53 | switch { |
| 54 | case os.IsNotExist(err): |
| 55 | return nil, true |
| 56 | case err != nil: |
| 57 | return nil, false |
| 58 | } |
| 59 | return entries, true |
| 60 | } |
| 61 | |
| 62 | // legacySessionRows builds one row per .jsonl transcript plus the same rows |
| 63 | // keyed by canonical path, which is how a canonical row later finds the |
| 64 | // migrated source whose title and turn count it borrows. |
| 65 | func (s *Server) legacySessionRows(r *http.Request, ctrl control.SessionAPI, entries []os.DirEntry) ([]sessionListEntry, map[string]sessionListEntry, []canonicalSessionRow) { |
| 66 | dir := ctrl.SessionDir() |
| 67 | current := agent.CanonicalSessionPath(ctrl.SessionPath()) |
| 68 | running := s.detachedRuntimeWork() |
| 69 | rows := make([]sessionListEntry, 0, len(entries)) |
| 70 | byPath := make(map[string]sessionListEntry, len(entries)) |
| 71 | for _, entry := range entries { |
| 72 | if entry.IsDir() || !store.IsSessionTranscriptName(entry.Name()) { |
| 73 | continue |
| 74 | } |
| 75 | path := agent.CanonicalSessionPath(filepath.Join(dir, entry.Name())) |
| 76 | if agent.IsCleanupPending(path) { |
| 77 | continue |
| 78 | } |
| 79 | mtime := agent.SessionContentModTime(path) |
| 80 | cleanPath := agent.CanonicalSessionPath(path) |
| 81 | row := sessionListEntry{ |
| 82 | Name: strings.TrimSuffix(entry.Name(), ".jsonl"), |
| 83 | Path: path, |
| 84 | Current: cleanPath == current, |
| 85 | Running: running[cleanPath], |
| 86 | TakenOver: s.sessionMirrored(cleanPath) || leaseHeldByForeignRuntime(cleanPath), |
| 87 | MtimeMilli: mtime.UnixMilli(), |
| 88 | } |
| 89 | if row.Current { |
| 90 | row.Running = controllerHasActiveRuntimeWork(ctrl) && !row.TakenOver |
| 91 | } |
| 92 | first, turns, cached := agent.SessionPreviewCached(path) |
| 93 | if !cached { |
| 94 | first, turns = agent.SessionPreview(path) |
| 95 | } |
| 96 | if turns > 0 { |
| 97 | row.Turns = turns |
| 98 | row.Title = s.sessionTitle(r.Context(), entry.Name(), first, mtime.UnixNano()) |
| 99 | } |
| 100 | rows = append(rows, row) |
| 101 | byPath[cleanPath] = row |
| 102 | } |
| 103 | return rows, byPath, s.canonicalSessionRows(r, ctrl) |
| 104 | } |
| 105 | |
| 106 | func (s *Server) detachedRuntimeWork() map[string]bool { |
| 107 | running := map[string]bool{} |
| 108 | s.detachedMu.Lock() |
| 109 | defer s.detachedMu.Unlock() |
| 110 | for path, detached := range s.detached { |
| 111 | running[filepath.Clean(path)] = controllerHasActiveRuntimeWork(detached.ctrl) |
| 112 | } |
| 113 | return running |
| 114 | } |
| 115 | |
| 116 | // canonicalSessionRows follows the catalog cursor to the end. One List call |
| 117 | // caps at 100 rows ordered by the random session id, so a workspace past 100 |
| 118 | // sessions would otherwise hide an arbitrary subset — including a session just |
| 119 | // taken over. The row bound keeps a broken cursor from looping forever. |
| 120 | func (s *Server) canonicalSessionRows(r *http.Request, ctrl control.SessionAPI) []canonicalSessionRow { |
| 121 | rows := make([]canonicalSessionRow, 0) |
| 122 | concrete, ok := ctrl.(*control.Controller) |
| 123 | if !ok { |
| 124 | return rows |
| 125 | } |
| 126 | service := concrete.SessionService() |
| 127 | if service == nil { |
| 128 | return rows |
| 129 | } |
| 130 | _, runtime, bound := concrete.SessionBinding() |
| 131 | const pageLimit = 100 |
| 132 | const maxCanonicalRows = 500 |
| 133 | cursor := "" |
| 134 | for pages := 1; ; pages++ { |
| 135 | page, err := service.Query().List(r.Context(), cursor, pageLimit) |
| 136 | if err != nil { |
| 137 | break |
| 138 | } |
| 139 | if len(rows) == 0 { |
| 140 | rows = make([]canonicalSessionRow, 0, len(page.Sessions)) |
| 141 | } |
| 142 | for _, info := range page.Sessions { |
| 143 | row := sessionListEntry{ |
| 144 | HostID: info.Ref.HostID, SessionID: info.Ref.SessionID, Name: info.SessionID, |
| 145 | Title: info.Title, Turns: info.Turns, MtimeMilli: info.CreatedAt.UnixMilli(), |
| 146 | Current: bound && info.Ref == runtime.Ref(), |
| 147 | Preview: info.Preview, |
| 148 | MetadataReady: info.MetadataStatus == session.MetadataReady, |
| 149 | TakenOver: s.sessionMirrored(remoteSessionIDQueryPrefix + info.Ref.SessionID), |
| 150 | } |
| 151 | // Canonical rows carry no legacy preview fallback, so a chatted |
| 152 | // session would list as untitled until the model renames it. |
| 153 | if strings.TrimSpace(row.Title) == "" && strings.TrimSpace(info.Preview) != "" { |
| 154 | row.Title = truncatedPreview(info.Preview) |
| 155 | } |
| 156 | if live, exists := service.Runtime(info.Ref); exists { |
| 157 | row.Running = live.Snapshot().Phase.Busy() |
| 158 | } |
| 159 | rows = append(rows, canonicalSessionRow{row: row, info: info}) |
| 160 | } |
| 161 | if page.NextCursor == "" || page.NextCursor == cursor || pages*pageLimit >= maxCanonicalRows || len(rows) >= maxCanonicalRows { |
| 162 | break |
| 163 | } |
| 164 | cursor = page.NextCursor |
| 165 | } |
| 166 | return rows |
| 167 | } |
| 168 | |
| 169 | // truncatedPreview clamps a catalog preview the way previewTitle clamps a |
| 170 | // transcript's first message. Catalog previews are already plain user text, so |
| 171 | // they need no paste-label stripping. |
| 172 | func truncatedPreview(preview string) string { |
| 173 | preview = strings.TrimSpace(preview) |
| 174 | if r := []rune(preview); len(r) > 50 { |
| 175 | return string(r[:47]) + "..." |
| 176 | } |
| 177 | return preview |
| 178 | } |
| 179 | |
| 180 | // mergeSessionRows produces the user-visible list from both catalogs. |
| 181 | // |
| 182 | // Migration deliberately preserves the old transcript, so a host can contain |
| 183 | // both the source .jsonl and its canonical session directory. The source is not |
| 184 | // a second user-visible session once the migration map proves there is exactly |
| 185 | // one canonical target for it; the canonical row stays the authoritative |
| 186 | // open/delete identity and borrows the old preview title until its asynchronous |
| 187 | // catalog metadata is ready. |
| 188 | func mergeSessionRows(legacyRows []sessionListEntry, legacyByPath map[string]sessionListEntry, canonicalRows []canonicalSessionRow) []sessionListEntry { |
| 189 | migration := migrationIndexFor(canonicalRows) |
| 190 | // The engine mirrors an in-flight legacy transcript into a final-format |
| 191 | // event log keyed by the legacy branch id. That mirror is plumbing, not a |
| 192 | // second conversation, except when it is the current row's own tree badge. |
| 193 | legacyBranchIDs := make(map[string]struct{}, len(legacyByPath)) |
| 194 | for path := range legacyByPath { |
| 195 | legacyBranchIDs[agent.BranchID(path)] = struct{}{} |
| 196 | } |
| 197 | out := make([]sessionListEntry, 0, len(legacyRows)+len(canonicalRows)) |
| 198 | for _, row := range legacyRows { |
| 199 | if _, migrated := migration.bySource[agent.CanonicalSessionPath(row.Path)]; migrated { |
| 200 | continue |
| 201 | } |
| 202 | out = append(out, row) |
| 203 | } |
| 204 | for i := range canonicalRows { |
| 205 | row := &canonicalRows[i].row |
| 206 | if !row.Current { |
| 207 | if _, mirrored := legacyBranchIDs[row.SessionID]; mirrored { |
| 208 | continue |
| 209 | } |
| 210 | } |
| 211 | if source, ok := migration.byTarget[row.SessionID]; ok { |
| 212 | inheritLegacyRow(row, legacyByPath[source]) |
| 213 | } |
| 214 | out = append(out, *row) |
| 215 | } |
| 216 | return out |
| 217 | } |
| 218 | |
| 219 | func inheritLegacyRow(row *sessionListEntry, legacy sessionListEntry) { |
| 220 | if row.Title == "" { |
| 221 | row.Title = legacy.Title |
| 222 | } |
| 223 | if row.Turns == 0 { |
| 224 | row.Turns = legacy.Turns |
| 225 | } |
| 226 | if row.MtimeMilli < legacy.MtimeMilli { |
| 227 | row.MtimeMilli = legacy.MtimeMilli |
| 228 | } |
| 229 | } |
| 230 | |
| 231 | func migrationIndexFor(canonicalRows []canonicalSessionRow) migrationSourceIndex { |
| 232 | canonicalIDs := make(map[string]struct{}, len(canonicalRows)) |
| 233 | roots := make(map[string]struct{}) |
| 234 | for _, row := range canonicalRows { |
| 235 | if row.row.SessionID != "" { |
| 236 | canonicalIDs[row.row.SessionID] = struct{}{} |
| 237 | } |
| 238 | if path := strings.TrimSpace(row.info.Path); path != "" { |
| 239 | roots[filepath.Dir(filepath.Clean(path))] = struct{}{} |
| 240 | } |
| 241 | } |
| 242 | return loadMigrationIndex(roots, func(targetID string) bool { |
| 243 | _, exists := canonicalIDs[targetID] |
| 244 | return exists |
| 245 | }) |
| 246 | } |
| 247 | |
| 248 | type canonicalSessionRow struct { |
| 249 | row sessionListEntry |
| 250 | info session.SessionInfo |
| 251 | } |
| 252 | |
| 253 | // migrationSourceIndex is the one rule for when a frozen legacy transcript is |
| 254 | // hidden behind its canonical row. Listing consults it to fold the source out |
| 255 | // of /sessions; deletion consults it to decide whether removing a canonical row |
| 256 | // may also remove the source. Both answers must agree, or a delete can remove a |
| 257 | // transcript the listing still shows as a distinct session. |
| 258 | type migrationSourceIndex struct { |
| 259 | bySource map[string]struct{} |
| 260 | byTarget map[string]string |
| 261 | } |
| 262 | |
| 263 | // loadMigrationIndex reads the migration maps under roots and keeps only |
| 264 | // unambiguous source->target mappings. Multiple canonical targets can |
| 265 | // legitimately be produced from one legacy DAG head, in which case hiding or |
| 266 | // deleting the source would remove a still-distinct view. exists reports |
| 267 | // whether a target id is a live canonical row; targets that are gone do not |
| 268 | // count, so a source whose other targets were already deleted is again the |
| 269 | // sole source of the remaining one. |
| 270 | func loadMigrationIndex(roots map[string]struct{}, exists func(targetID string) bool) migrationSourceIndex { |
| 271 | index := migrationSourceIndex{bySource: map[string]struct{}{}, byTarget: map[string]string{}} |
| 272 | targetsBySource := make(map[string][]string) |
| 273 | for root := range roots { |
| 274 | data, err := os.ReadFile(filepath.Join(root, "migration-map.json")) |
| 275 | if err != nil { |
| 276 | continue |
| 277 | } |
| 278 | var mapping session.MigrationMapping |
| 279 | if json.Unmarshal(data, &mapping) != nil || mapping.SchemaVersion != session.SchemaVersion { |
| 280 | continue |
| 281 | } |
| 282 | for _, entry := range mapping.Entries { |
| 283 | source := agent.CanonicalSessionPath(entry.SourcePath) |
| 284 | target := strings.TrimSpace(entry.TargetID) |
| 285 | if source == "" || target == "" || !exists(target) { |
| 286 | continue |
| 287 | } |
| 288 | seen := false |
| 289 | for _, existing := range targetsBySource[source] { |
| 290 | seen = seen || existing == target |
| 291 | } |
| 292 | if !seen { |
| 293 | targetsBySource[source] = append(targetsBySource[source], target) |
| 294 | } |
| 295 | } |
| 296 | } |
| 297 | for source, targets := range targetsBySource { |
| 298 | if len(targets) != 1 { |
| 299 | continue |
| 300 | } |
| 301 | index.bySource[source] = struct{}{} |
| 302 | index.byTarget[targets[0]] = source |
| 303 | } |
| 304 | return index |
| 305 | } |
| 306 |