| 1 | package history |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "path/filepath" |
| 6 | "sort" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | |
| 10 | "reasonix/internal/agent" |
| 11 | "reasonix/internal/historycatalog" |
| 12 | "reasonix/internal/provider" |
| 13 | "reasonix/internal/retrieval" |
| 14 | ) |
| 15 | |
| 16 | type indexedCatalogManager struct { |
| 17 | lifecycleMu sync.Mutex |
| 18 | mu sync.RWMutex |
| 19 | catalog *historycatalog.Catalog |
| 20 | roots map[string]historycatalog.Root |
| 21 | observers []func(historycatalog.Status, []string, string) |
| 22 | persistObservers map[string]agent.SessionPersistObserver |
| 23 | generation uint64 |
| 24 | opening map[uint64]chan struct{} |
| 25 | openCancel map[uint64]context.CancelFunc |
| 26 | closing bool |
| 27 | open func(context.Context, historycatalog.Options) (*historycatalog.Catalog, error) |
| 28 | rebuild func(context.Context, historycatalog.Options, []historycatalog.Root) (historycatalog.Status, error) |
| 29 | } |
| 30 | |
| 31 | var processHistoryCatalog indexedCatalogManager |
| 32 | |
| 33 | // RegisterCatalogRoots lets a host such as Desktop seed every saved project |
| 34 | // without constructing a controller. Opening and scanning remain asynchronous. |
| 35 | func RegisterCatalogRoots(roots []historycatalog.Root) { processHistoryCatalog.register(roots) } |
| 36 | |
| 37 | // RegisterCatalogObserver subscribes a host to revision/progress changes. The |
| 38 | // callback payload contains roots and counters only, never query or content. |
| 39 | func RegisterCatalogObserver(observer func(historycatalog.Status, []string, string)) { |
| 40 | if observer == nil { |
| 41 | return |
| 42 | } |
| 43 | processHistoryCatalog.mu.Lock() |
| 44 | processHistoryCatalog.observers = append(processHistoryCatalog.observers, observer) |
| 45 | processHistoryCatalog.mu.Unlock() |
| 46 | } |
| 47 | |
| 48 | // RegisterSessionPersistObserver fans authoritative agent save events into an |
| 49 | // additional derived catalog. Registration is keyed so desktop rebuilds replace |
| 50 | // their sink without accumulating closures. Observers must remain non-blocking. |
| 51 | func RegisterSessionPersistObserver(key string, observer agent.SessionPersistObserver) { |
| 52 | key = strings.TrimSpace(key) |
| 53 | if key == "" { |
| 54 | return |
| 55 | } |
| 56 | processHistoryCatalog.mu.Lock() |
| 57 | if processHistoryCatalog.persistObservers == nil { |
| 58 | processHistoryCatalog.persistObservers = map[string]agent.SessionPersistObserver{} |
| 59 | } |
| 60 | if observer == nil { |
| 61 | delete(processHistoryCatalog.persistObservers, key) |
| 62 | } else { |
| 63 | processHistoryCatalog.persistObservers[key] = observer |
| 64 | } |
| 65 | processHistoryCatalog.mu.Unlock() |
| 66 | } |
| 67 | |
| 68 | // SharedCatalog returns the process projection when opening has completed. |
| 69 | // Nil means callers should return an explicit partial/opening response. |
| 70 | func SharedCatalog() *historycatalog.Catalog { return processHistoryCatalog.get() } |
| 71 | |
| 72 | func FlushSharedCatalog(ctx context.Context) error { |
| 73 | if catalog := processHistoryCatalog.get(); catalog != nil { |
| 74 | return catalog.Flush(ctx) |
| 75 | } |
| 76 | return nil |
| 77 | } |
| 78 | |
| 79 | // CloseSharedCatalog cancels work and closes the process history catalog. |
| 80 | // Callers with time to persist queued projection work may invoke |
| 81 | // FlushSharedCatalog first. Close itself must not drain a potentially large |
| 82 | // backlog because JSONL remains authoritative and the projection is rebuilt. |
| 83 | func CloseSharedCatalog(ctx context.Context) error { |
| 84 | return processHistoryCatalog.close(ctx) |
| 85 | } |
| 86 | |
| 87 | func (m *indexedCatalogManager) register(roots []historycatalog.Root) { |
| 88 | m.mu.Lock() |
| 89 | if m.roots == nil { |
| 90 | m.roots = map[string]historycatalog.Root{} |
| 91 | } |
| 92 | for _, root := range roots { |
| 93 | if strings.TrimSpace(root.Path) != "" { |
| 94 | root.Path = filepath.Clean(root.Path) |
| 95 | m.roots[root.Path] = root |
| 96 | } |
| 97 | } |
| 98 | catalog := m.catalog |
| 99 | if catalog == nil { |
| 100 | m.startOpenLocked() |
| 101 | } |
| 102 | m.mu.Unlock() |
| 103 | if catalog != nil { |
| 104 | for _, root := range roots { |
| 105 | catalog.RegisterRoot(root) |
| 106 | } |
| 107 | } |
| 108 | } |
| 109 | |
| 110 | func (m *indexedCatalogManager) startOpenLocked() { |
| 111 | if m.catalog != nil || m.closing || len(m.opening) != 0 { |
| 112 | return |
| 113 | } |
| 114 | m.generation++ |
| 115 | generation := m.generation |
| 116 | if m.opening == nil { |
| 117 | m.opening = map[uint64]chan struct{}{} |
| 118 | m.openCancel = map[uint64]context.CancelFunc{} |
| 119 | } |
| 120 | ctx, cancel := context.WithCancel(context.Background()) |
| 121 | done := make(chan struct{}) |
| 122 | m.opening[generation] = done |
| 123 | m.openCancel[generation] = cancel |
| 124 | openCatalog := m.open |
| 125 | if openCatalog == nil { |
| 126 | openCatalog = historycatalog.Open |
| 127 | } |
| 128 | go m.openGeneration(ctx, generation, done, openCatalog) |
| 129 | } |
| 130 | |
| 131 | func (m *indexedCatalogManager) openGeneration(ctx context.Context, generation uint64, done chan struct{}, openCatalog func(context.Context, historycatalog.Options) (*historycatalog.Catalog, error)) { |
| 132 | catalog, err := openCatalog(ctx, historycatalog.Options{OnRevision: m.publish}) |
| 133 | if err == nil { |
| 134 | seen := map[string]bool{} |
| 135 | for { |
| 136 | m.mu.Lock() |
| 137 | if m.closing || generation != m.generation || ctx.Err() != nil { |
| 138 | m.mu.Unlock() |
| 139 | _ = catalog.Close(context.Background()) |
| 140 | catalog = nil |
| 141 | break |
| 142 | } |
| 143 | pending := make([]historycatalog.Root, 0, len(m.roots)) |
| 144 | for path, root := range m.roots { |
| 145 | if !seen[path] { |
| 146 | seen[path] = true |
| 147 | pending = append(pending, root) |
| 148 | } |
| 149 | } |
| 150 | if len(pending) == 0 { |
| 151 | m.catalog = catalog |
| 152 | m.mu.Unlock() |
| 153 | break |
| 154 | } |
| 155 | m.mu.Unlock() |
| 156 | for _, root := range pending { |
| 157 | catalog.RegisterRoot(root) |
| 158 | } |
| 159 | } |
| 160 | } |
| 161 | m.mu.Lock() |
| 162 | delete(m.opening, generation) |
| 163 | delete(m.openCancel, generation) |
| 164 | close(done) |
| 165 | m.mu.Unlock() |
| 166 | } |
| 167 | |
| 168 | func (m *indexedCatalogManager) close(ctx context.Context) error { |
| 169 | m.lifecycleMu.Lock() |
| 170 | defer m.lifecycleMu.Unlock() |
| 171 | m.mu.Lock() |
| 172 | m.closing = true |
| 173 | m.generation++ |
| 174 | catalog := m.catalog |
| 175 | m.catalog = nil |
| 176 | m.roots = nil |
| 177 | m.observers = nil |
| 178 | done := make([]chan struct{}, 0, len(m.opening)) |
| 179 | for generation, opening := range m.opening { |
| 180 | m.openCancel[generation]() |
| 181 | done = append(done, opening) |
| 182 | } |
| 183 | m.mu.Unlock() |
| 184 | |
| 185 | var closeErr error |
| 186 | if catalog != nil { |
| 187 | closeErr = catalog.Close(ctx) |
| 188 | } |
| 189 | for _, opening := range done { |
| 190 | select { |
| 191 | case <-opening: |
| 192 | case <-ctx.Done(): |
| 193 | if closeErr == nil { |
| 194 | closeErr = ctx.Err() |
| 195 | } |
| 196 | } |
| 197 | } |
| 198 | m.mu.Lock() |
| 199 | m.closing = false |
| 200 | m.mu.Unlock() |
| 201 | return closeErr |
| 202 | } |
| 203 | |
| 204 | func (m *indexedCatalogManager) publish(status historycatalog.Status, roots []string, reason string) { |
| 205 | m.mu.RLock() |
| 206 | observers := append([]func(historycatalog.Status, []string, string){}, m.observers...) |
| 207 | m.mu.RUnlock() |
| 208 | for _, observer := range observers { |
| 209 | observer(status, append([]string{}, roots...), reason) |
| 210 | } |
| 211 | } |
| 212 | |
| 213 | func (m *indexedCatalogManager) get() *historycatalog.Catalog { |
| 214 | m.mu.RLock() |
| 215 | defer m.mu.RUnlock() |
| 216 | return m.catalog |
| 217 | } |
| 218 | |
| 219 | // PersistObserver returns the process-wide non-blocking projection sink. The |
| 220 | // authoritative save has already released its path/file locks before this is |
| 221 | // called by agent.Session. |
| 222 | func PersistObserver() agent.SessionPersistObserver { return historyPersistObserver{} } |
| 223 | |
| 224 | type historyPersistObserver struct{} |
| 225 | |
| 226 | func (historyPersistObserver) EnqueueSessionPersist(event agent.SessionPersistEvent) bool { |
| 227 | catalog := processHistoryCatalog.get() |
| 228 | processHistoryCatalog.mu.RLock() |
| 229 | bestLength := -1 |
| 230 | var selected historycatalog.Root |
| 231 | for _, root := range processHistoryCatalog.roots { |
| 232 | if underRoot(event.Path, root.Path) && len(root.Path) > bestLength { |
| 233 | selected = root |
| 234 | bestLength = len(root.Path) |
| 235 | } |
| 236 | } |
| 237 | additional := make([]agent.SessionPersistObserver, 0, len(processHistoryCatalog.persistObservers)) |
| 238 | for _, observer := range processHistoryCatalog.persistObservers { |
| 239 | additional = append(additional, observer) |
| 240 | } |
| 241 | processHistoryCatalog.mu.RUnlock() |
| 242 | |
| 243 | accepted := false |
| 244 | if catalog != nil && bestLength >= 0 { |
| 245 | if event.Removed { |
| 246 | go func() { _ = catalog.Purge(context.Background(), event.Path) }() |
| 247 | accepted = true |
| 248 | } else { |
| 249 | accepted = catalog.EnqueuePersist(selected, event) |
| 250 | } |
| 251 | } |
| 252 | for _, observer := range additional { |
| 253 | accepted = observer.EnqueueSessionPersist(event) || accepted |
| 254 | } |
| 255 | return accepted |
| 256 | } |
| 257 | |
| 258 | type IndexedSearcher struct { |
| 259 | legacy *Searcher |
| 260 | roots []historycatalog.Root |
| 261 | } |
| 262 | |
| 263 | func NewIndexedSearcher(opts Options) *IndexedSearcher { |
| 264 | legacy := NewSearcher(opts) |
| 265 | roots := []historycatalog.Root{} |
| 266 | add := func(path, source, scope, workspace string, archive bool) { |
| 267 | if strings.TrimSpace(path) == "" { |
| 268 | return |
| 269 | } |
| 270 | roots = append(roots, historycatalog.Root{Path: path, Source: source, Scope: scope, WorkspaceRoot: workspace, Archive: archive}) |
| 271 | } |
| 272 | add(opts.SessionDir, scopeProject, scopeProject, opts.SessionDir, false) |
| 273 | add(subagentsDir(opts.SessionDir), scopeProject, scopeProject, opts.SessionDir, false) |
| 274 | if filepath.Clean(opts.GlobalSessionDir) != filepath.Clean(opts.SessionDir) { |
| 275 | add(opts.GlobalSessionDir, scopeGlobal, scopeGlobal, "", false) |
| 276 | add(subagentsDir(opts.GlobalSessionDir), scopeGlobal, scopeGlobal, "", false) |
| 277 | } |
| 278 | add(opts.ArchiveDir, "archive", scopeGlobal, "", true) |
| 279 | processHistoryCatalog.register(roots) |
| 280 | return &IndexedSearcher{legacy: legacy, roots: roots} |
| 281 | } |
| 282 | |
| 283 | func (s *IndexedSearcher) rootsFor(scope string) []string { |
| 284 | out := []string{} |
| 285 | for _, root := range s.roots { |
| 286 | if scope == scopeProject && root.Scope != scopeProject { |
| 287 | continue |
| 288 | } |
| 289 | out = append(out, root.Path) |
| 290 | } |
| 291 | return out |
| 292 | } |
| 293 | |
| 294 | func (s *IndexedSearcher) Search(ctx context.Context, req SearchRequest) ([]Hit, error) { |
| 295 | query := strings.TrimSpace(req.Query) |
| 296 | if query == "" { |
| 297 | return nil, contextError("query is required") |
| 298 | } |
| 299 | queryTerms, err := retrieval.QueryTerms(query) |
| 300 | if err != nil { |
| 301 | return nil, err |
| 302 | } |
| 303 | scope, err := normalizeScope(req.Scope) |
| 304 | if err != nil { |
| 305 | return nil, err |
| 306 | } |
| 307 | limit := clamp(req.Limit, defaultLimit, maxLimit) |
| 308 | kinds, err := normalizeKinds(req.Kinds) |
| 309 | if err != nil { |
| 310 | return nil, err |
| 311 | } |
| 312 | catalog := processHistoryCatalog.get() |
| 313 | if catalog == nil { |
| 314 | return []Hit{}, nil |
| 315 | } |
| 316 | kindNames := make([]string, 0, len(kinds)) |
| 317 | for kind := range kinds { |
| 318 | kindNames = append(kindNames, string(kind)) |
| 319 | } |
| 320 | sort.Strings(kindNames) |
| 321 | result, err := catalog.Search(ctx, historycatalog.SearchRequest{ |
| 322 | // Exact roots are the agent authority boundary. Catalog scope describes |
| 323 | // desktop grouping and must not change the history tool's project meaning. |
| 324 | Query: query, |
| 325 | Kinds: kindNames, ToolName: strings.TrimSpace(req.ToolName), Limit: min(limit*4, historycatalog.MaxLimit), |
| 326 | Roots: s.rootsFor(scope), |
| 327 | }) |
| 328 | if err != nil { |
| 329 | return nil, err |
| 330 | } |
| 331 | loaded := map[string][]provider.Message{} |
| 332 | failed := map[string]bool{} |
| 333 | hits := make([]Hit, 0, len(result.Items)) |
| 334 | for _, candidate := range result.Items { |
| 335 | if failed[candidate.SessionPath] { |
| 336 | continue |
| 337 | } |
| 338 | messages, ok := loaded[candidate.SessionPath] |
| 339 | if !ok { |
| 340 | state, known, identityErr := agent.SessionContentIdentity(candidate.SessionPath) |
| 341 | if identityErr != nil || (known && state.DigestHex != candidate.ContentDigest) { |
| 342 | failed[candidate.SessionPath] = true |
| 343 | catalog.EnqueueExisting(context.Background(), candidate.SessionPath) |
| 344 | continue |
| 345 | } |
| 346 | messages, err = loadMessages(candidate.SessionPath) |
| 347 | if err != nil { |
| 348 | failed[candidate.SessionPath] = true |
| 349 | continue |
| 350 | } |
| 351 | loaded[candidate.SessionPath] = messages |
| 352 | } |
| 353 | text, ok := candidateText(messages, candidate) |
| 354 | if !ok { |
| 355 | catalog.EnqueueExisting(context.Background(), candidate.SessionPath) |
| 356 | continue |
| 357 | } |
| 358 | hits = append(hits, Hit{Score: candidate.Score, SessionPath: candidate.SessionPath, |
| 359 | SessionID: sessionID(candidate.SessionPath), Source: candidate.Source, MessageIndex: candidate.MessageIndex, |
| 360 | Role: provider.Role(candidate.Role), Kind: Kind(candidate.Kind), ToolName: candidate.ToolName, |
| 361 | Snippet: retrieval.MakeSnippet(text, query, queryTerms, maxSnippet)}) |
| 362 | } |
| 363 | hits = retrieval.KeepTopRelativeScore(hits, scoreFloor, func(hit Hit) float64 { return hit.Score }) |
| 364 | if len(hits) > limit { |
| 365 | hits = hits[:limit] |
| 366 | } |
| 367 | return hits, nil |
| 368 | } |
| 369 | |
| 370 | func candidateText(messages []provider.Message, candidate historycatalog.Candidate) (string, bool) { |
| 371 | if candidate.MessageIndex < 0 || candidate.MessageIndex >= len(messages) { |
| 372 | return "", false |
| 373 | } |
| 374 | msg := messages[candidate.MessageIndex] |
| 375 | switch Kind(candidate.Kind) { |
| 376 | case KindUserText: |
| 377 | return stripComposePrefixes(msg.Content), msg.Role == provider.RoleUser && !agent.IsPinnedContextRevision(msg) |
| 378 | case KindAssistantText: |
| 379 | return msg.Content, msg.Role == provider.RoleAssistant |
| 380 | case KindToolInput: |
| 381 | if candidate.PartIndex < 0 || candidate.PartIndex >= len(msg.ToolCalls) { |
| 382 | return "", false |
| 383 | } |
| 384 | call := msg.ToolCalls[candidate.PartIndex] |
| 385 | return strings.TrimSpace(call.Name + " " + call.Arguments), true |
| 386 | case KindToolError, KindToolOutput: |
| 387 | return strings.TrimSpace(msg.Name + " " + msg.Content), msg.Role == provider.RoleTool |
| 388 | default: |
| 389 | return "", false |
| 390 | } |
| 391 | } |
| 392 | |
| 393 | func (s *IndexedSearcher) Around(ctx context.Context, req AroundRequest) ([]MessageContext, error) { |
| 394 | return s.legacy.Around(ctx, req) |
| 395 | } |
| 396 | |
| 397 | func (s *IndexedSearcher) IndexStatus() historycatalog.Status { |
| 398 | if catalog := processHistoryCatalog.get(); catalog != nil { |
| 399 | return catalog.Status() |
| 400 | } |
| 401 | return historycatalog.Status{State: "opening", Pending: 1} |
| 402 | } |
| 403 | |
| 404 | func contextError(message string) error { return &indexedInputError{message: message} } |
| 405 | |
| 406 | type indexedInputError struct{ message string } |
| 407 | |
| 408 | func (e *indexedInputError) Error() string { return e.message } |
| 409 |