| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/agent" |
| 12 | "reasonix/internal/fileutil" |
| 13 | "reasonix/internal/provider" |
| 14 | ) |
| 15 | |
| 16 | // Rebuild authored previews, retracted input/turn metadata, title sequencing, |
| 17 | // and visible result sequencing. |
| 18 | const catalogMetadataVersion = 5 |
| 19 | |
| 20 | // metadataForDurable publishes catalog metadata only for a durable prefix. |
| 21 | func (s *Session) metadataForDurable(durable uint64) (catalogMetadata, bool) { |
| 22 | if s == nil { |
| 23 | return catalogMetadata{}, false |
| 24 | } |
| 25 | s.mu.Lock() |
| 26 | defer s.mu.Unlock() |
| 27 | if durable+1 != s.next { |
| 28 | return catalogMetadata{}, false |
| 29 | } |
| 30 | return metadataFromProjection(s.manifest, durable, s.projection), true |
| 31 | } |
| 32 | |
| 33 | const ( |
| 34 | MetadataReady = "ready" |
| 35 | MetadataPending = "pending" |
| 36 | MetadataFailed = "failed" |
| 37 | ) |
| 38 | |
| 39 | type catalogMetadata struct { |
| 40 | Version int `json:"version"` |
| 41 | Codec string `json:"codec"` |
| 42 | SessionID string `json:"sessionId"` |
| 43 | CreatedAt string `json:"createdAt"` |
| 44 | Sequence uint64 `json:"sequence"` |
| 45 | ResultSequence uint64 `json:"resultSequence,omitempty"` |
| 46 | Title string `json:"title,omitempty"` |
| 47 | TitleSequence uint64 `json:"titleSequence,omitempty"` |
| 48 | ModelRef string `json:"modelRef,omitempty"` |
| 49 | ModelIdentity string `json:"modelIdentity,omitempty"` |
| 50 | Turns int `json:"turns"` |
| 51 | Preview string `json:"preview,omitempty"` |
| 52 | // LogRevision pins the cache to the exact durable bytes it was built from. |
| 53 | // Catalog listing validates this instead of replaying the log, so a page of |
| 54 | // long sessions costs one stat per session rather than a full scan. |
| 55 | LogSize int64 `json:"logSize"` |
| 56 | LogModTimeNS int64 `json:"logModTimeNs"` |
| 57 | LogIdentity string `json:"logIdentity,omitempty"` |
| 58 | } |
| 59 | |
| 60 | func catalogMetadataPath(cacheDir string) string { |
| 61 | return filepath.Join(cacheDir, "catalog-metadata.json") |
| 62 | } |
| 63 | |
| 64 | func metadataFromProjection(manifest Manifest, sequence uint64, projection Projection) catalogMetadata { |
| 65 | metadata := catalogMetadata{ |
| 66 | Version: catalogMetadataVersion, Codec: Codec, SessionID: manifest.SessionID, |
| 67 | CreatedAt: manifest.CreatedAt.UTC().Format(time.RFC3339Nano), Sequence: sequence, |
| 68 | Title: projection.Title, TitleSequence: projection.TitleSequence, |
| 69 | ModelRef: projection.ModelRef, ModelIdentity: projection.ModelIdentity, |
| 70 | } |
| 71 | metadata.Turns = visibleBoundaryCount(projection, true) |
| 72 | metadata.ResultSequence = latestVisibleResultSequence(projection) |
| 73 | for _, input := range projection.TranscriptInputs { |
| 74 | if input.Preview != "" { |
| 75 | metadata.Preview = input.Preview |
| 76 | return metadata |
| 77 | } |
| 78 | } |
| 79 | if len(projection.TranscriptInputs) > 0 { |
| 80 | return metadata |
| 81 | } |
| 82 | for _, message := range projection.Messages { |
| 83 | if preview := catalogMessagePreview(message); preview != "" { |
| 84 | metadata.Preview = preview |
| 85 | break |
| 86 | } |
| 87 | } |
| 88 | return metadata |
| 89 | } |
| 90 | |
| 91 | func latestVisibleResultSequence(projection Projection) uint64 { |
| 92 | var latest uint64 |
| 93 | for _, turn := range projection.Turns { |
| 94 | if projection.HiddenTurns[turn.TurnID] || !turn.Status.Terminal() || turn.MessageID == "" { |
| 95 | continue |
| 96 | } |
| 97 | latest = max(latest, turn.BoundarySequence) |
| 98 | } |
| 99 | return latest |
| 100 | } |
| 101 | |
| 102 | // Catalog labels use authored display text, including literal markup in an |
| 103 | // explicit RawContent. Host messages and mid-turn steers are not session names. |
| 104 | func catalogMessagePreview(message provider.Message) string { |
| 105 | if message.Role != provider.RoleUser || agent.IsHostGeneratedUserMessage(message) { |
| 106 | return "" |
| 107 | } |
| 108 | if _, steer := agent.SteerText(message.Content); steer { |
| 109 | return "" |
| 110 | } |
| 111 | return messagePreview(message) |
| 112 | } |
| 113 | |
| 114 | // logRevision identifies the durable file revision a cache entry describes. |
| 115 | // Catalog listing deliberately uses metadata only; it never opens or samples |
| 116 | // event bodies. Session incarnation, size, and nanosecond mtime fence a stale |
| 117 | // cache, while durable writers refresh the cache after every drained prefix. |
| 118 | type logRevision struct { |
| 119 | Size int64 |
| 120 | ModTimeNS int64 |
| 121 | Identity string |
| 122 | Exists bool |
| 123 | } |
| 124 | |
| 125 | // revisionOfLog inspects the manifest-selected event log without reading its body. |
| 126 | func revisionOfLog(dir string) (logRevision, error) { |
| 127 | manifest, err := readStoredManifest(filepath.Join(dir, "manifest.json")) |
| 128 | if os.IsNotExist(err) { |
| 129 | return logRevision{}, nil |
| 130 | } |
| 131 | if err != nil { |
| 132 | return logRevision{}, err |
| 133 | } |
| 134 | info, err := os.Stat(logPathForManifest(dir, manifest)) |
| 135 | if os.IsNotExist(err) { |
| 136 | return logRevision{}, nil |
| 137 | } |
| 138 | if err != nil { |
| 139 | return logRevision{}, err |
| 140 | } |
| 141 | return logRevision{Size: info.Size(), ModTimeNS: info.ModTime().UnixNano(), Exists: true}, nil |
| 142 | } |
| 143 | |
| 144 | func readCatalogMetadata(cacheDir string, manifest Manifest, revision logRevision) (catalogMetadata, error) { |
| 145 | data, err := os.ReadFile(catalogMetadataPath(cacheDir)) |
| 146 | if err != nil { |
| 147 | return catalogMetadata{}, err |
| 148 | } |
| 149 | var metadata catalogMetadata |
| 150 | if err := json.Unmarshal(data, &metadata); err != nil { |
| 151 | return catalogMetadata{}, err |
| 152 | } |
| 153 | if metadata.Version != catalogMetadataVersion || metadata.Codec != Codec || metadata.SessionID != manifest.SessionID || |
| 154 | metadata.CreatedAt != manifest.CreatedAt.UTC().Format(time.RFC3339Nano) || |
| 155 | metadata.LogSize != revision.Size || metadata.LogModTimeNS != revision.ModTimeNS || metadata.LogIdentity != revision.Identity { |
| 156 | return catalogMetadata{}, errors.New("session: catalog metadata cache is stale") |
| 157 | } |
| 158 | return metadata, nil |
| 159 | } |
| 160 | |
| 161 | func writeCatalogMetadata(cacheDir string, metadata catalogMetadata) error { |
| 162 | data, err := json.Marshal(metadata) |
| 163 | if err != nil { |
| 164 | return err |
| 165 | } |
| 166 | if err := os.MkdirAll(cacheDir, 0o700); err != nil { |
| 167 | return err |
| 168 | } |
| 169 | return fileutil.AtomicWriteFileStrict(catalogMetadataPath(cacheDir), append(data, '\n'), 0o600) |
| 170 | } |
| 171 | |
| 172 | func rebuildCatalogMetadata(ctx context.Context, handle eventPageReader, cacheDir, sessionDir string, manifest Manifest) error { |
| 173 | metadata, err := reduceCatalogMetadata(ctx, handle, manifest) |
| 174 | if err != nil { |
| 175 | return err |
| 176 | } |
| 177 | return writeCatalogMetadataForSession(cacheDir, sessionDir, metadata) |
| 178 | } |
| 179 | |
| 180 | func reduceCatalogMetadata(ctx context.Context, handle eventPageReader, manifest Manifest) (catalogMetadata, error) { |
| 181 | reducer := catalogReducer{} |
| 182 | if stream, ok := handle.(interface { |
| 183 | scanCatalog(context.Context, func(Commit) error) error |
| 184 | }); ok { |
| 185 | if err := stream.scanCatalog(ctx, reducer.apply); err != nil { |
| 186 | return catalogMetadata{}, err |
| 187 | } |
| 188 | return reducer.metadata(manifest), nil |
| 189 | } |
| 190 | var cursor uint64 |
| 191 | for { |
| 192 | page, err := handle.Read(ctx, cursor, 32) |
| 193 | if err != nil { |
| 194 | return catalogMetadata{}, err |
| 195 | } |
| 196 | for _, commit := range page.Commits { |
| 197 | if err := reducer.apply(commit); err != nil { |
| 198 | return catalogMetadata{}, err |
| 199 | } |
| 200 | } |
| 201 | if !page.Truncated { |
| 202 | break |
| 203 | } |
| 204 | if page.Next <= cursor { |
| 205 | return catalogMetadata{}, ErrDamagedStore |
| 206 | } |
| 207 | cursor = page.Next |
| 208 | } |
| 209 | return reducer.metadata(manifest), nil |
| 210 | } |
| 211 | |
| 212 | // writeCatalogMetadataForSession stamps the cache with the exact durable bytes |
| 213 | // it describes. Every writer of the cache goes through here so a reader can |
| 214 | // validate it using file metadata without reading event bodies. |
| 215 | func writeCatalogMetadataForSession(cacheDir, sessionDir string, metadata catalogMetadata) error { |
| 216 | revision, err := revisionOfLog(sessionDir) |
| 217 | if err != nil { |
| 218 | return err |
| 219 | } |
| 220 | metadata.LogSize, metadata.LogModTimeNS, metadata.LogIdentity = revision.Size, revision.ModTimeNS, revision.Identity |
| 221 | return writeCatalogMetadata(cacheDir, metadata) |
| 222 | } |
| 223 |