返回 DeepSeek-Reasonix
api.go
根目录 / internal / session / api.go
1 package session
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "os"
9 "path/filepath"
10 "sort"
11 "strings"
12 "sync"
13 "time"
14 )
15
16 // VisitCommits streams the complete durable commit prefix from one canonical
17 // session directory. It is intended for exports and diagnostics that must not
18 // materialize the cumulative event log.
19 func VisitCommits(ctx context.Context, dir string, visit func(Commit) error) error {
20 if err := ctx.Err(); err != nil {
21 return err
22 }
23 manifest, err := readStoredManifest(filepath.Join(dir, "manifest.json"))
24 if err != nil {
25 return err
26 }
27 file, err := os.Open(logPathForManifest(dir, manifest))
28 if os.IsNotExist(err) {
29 return nil
30 }
31 if err != nil {
32 return err
33 }
34 defer file.Close()
35 var visitErr error
36 adapter := func(_ int64, commit Commit) bool {
37 if visit != nil {
38 visitErr = visit(commit)
39 }
40 return visitErr == nil
41 }
42 if manifest.Codec == Codec {
43 err = scanV4CommitFile(ctx, file, 0, 1, contentStoreForSessionDir(dir), nil, adapter)
44 } else {
45 err = scanCommitFileCodec(file, 0, 1, manifest.Codec, nil, adapter)
46 }
47 return errors.Join(err, visitErr)
48 }
49
50 // AccessMode separates cold readers from the single leased writer. Read-only
51 // access never repairs, migrates, truncates, or advances writer generation.
52 type AccessMode string
53
54 const (
55 ReadOnly AccessMode = "read"
56 ReadWrite AccessMode = "write"
57 )
58
59 type CreateOptions struct {
60 SessionID string
61 CWD string
62 ParentSessionID string
63 Origin SessionOrigin
64 }
65
66 type SessionInfo struct {
67 SessionID string
68 Ref SessionRef
69 Codec string
70 Title string
71 TitleSequence uint64
72 ModelRef string
73 ModelIdentity string
74 Turns int
75 CreatedAt time.Time
76 UpdatedAt time.Time
77 EventSequence uint64
78 ResultSequence uint64
79 Preview string
80 MetadataStatus string
81 CWD string
82 ParentSessionID string
83 Origin SessionOrigin
84 Path string
85 Error string
86 }
87
88 type SessionPage struct {
89 Sessions []SessionInfo
90 NextCursor string
91 }
92
93 type EventPage struct {
94 Commits []Commit
95 Next uint64
96 Truncated bool
97 }
98
99 // eventPageReader is the paged durable read surface shared by the physical
100 // handle and a cold Session, so catalog rebuilds never depend on replaying a
101 // whole in-memory log.
102 type eventPageReader interface {
103 Read(context.Context, uint64, int) (EventPage, error)
104 }
105
106 // SessionHandle is the physical persistence contract. It reads and writes
107 // bytes for one session identity and owns the writer lease; it holds no
108 // projection, operation table, or accepted commit list.
109 type SessionHandle interface {
110 ID() string
111 Manifest() Manifest
112 Read(context.Context, uint64, int) (EventPage, error)
113 Append(context.Context, []Commit) error
114 Sync(context.Context) (DurableReceipt, error)
115 Close(context.Context) error
116 }
117
118 // WritableSessionHandle is the leased physical handle. Byte-level maintenance
119 // operations such as fork, export, and interrupted-turn recovery are Session
120 // operations because they read or extend the in-memory log; this contract only
121 // distinguishes a leased writer from a cold reader.
122 type WritableSessionHandle interface {
123 SessionHandle
124 SessionID() string
125 }
126
127 type SessionPersistence interface {
128 Create(CreateOptions) (*Session, error)
129 Open(sessionID string, mode AccessMode) (*Session, error)
130 Stat(context.Context, string) (SessionInfo, error)
131 List(context.Context, string, int) (SessionPage, error)
132 }
133
134 // FilesystemPersistence owns a versioned sessions-v4 root.
135 type FilesystemPersistence struct{ Root string }
136
137 func NewFilesystemPersistence(root string) *FilesystemPersistence {
138 return &FilesystemPersistence{Root: filepath.Clean(root)}
139 }
140
141 // RootForLegacyDir maps a host's legacy transcript catalog to the sibling
142 // final-format store. The mapping lives in the persistence package so Boot and
143 // controllers never derive a v3 identity from a transcript path.
144 func RootForLegacyDir(sessionDir string) string {
145 dir := filepath.Clean(strings.TrimSpace(sessionDir))
146 if dir == "." || dir == "" {
147 return ""
148 }
149 if filepath.Base(dir) == "sessions" {
150 return filepath.Join(filepath.Dir(dir), "sessions-v4")
151 }
152 return filepath.Join(dir, "sessions-v4")
153 }
154
155 func (p *FilesystemPersistence) Create(options CreateOptions) (*Session, error) {
156 id := strings.TrimSpace(options.SessionID)
157 if id == "" {
158 id = randomID()
159 }
160 if err := validateSessionID(id); err != nil {
161 return nil, err
162 }
163 dir, err := p.sessionDir(id, false)
164 if err != nil {
165 return nil, err
166 }
167 options.SessionID = id
168 header, err := headerForCreate(options)
169 if err != nil {
170 return nil, err
171 }
172 return createWithOptions(dir, id, OpenOptions{ExternalHistory: true}, header)
173 }
174
175 func (p *FilesystemPersistence) Open(sessionID string, mode AccessMode) (*Session, error) {
176 id := strings.TrimSpace(sessionID)
177 if err := validateSessionID(id); err != nil {
178 return nil, err
179 }
180 dir, err := p.sessionDir(id, true)
181 if err != nil {
182 return nil, err
183 }
184 if _, _, err := readSessionHeader(dir, id); err != nil {
185 return nil, err
186 }
187 if mode == ReadOnly {
188 return openReadSession(dir, id, filepath.Join(p.Root, ".query-cache", filepath.Base(id)))
189 }
190 if mode != ReadWrite {
191 return nil, fmt.Errorf("session: unsupported access mode %q", mode)
192 }
193 if _, err := os.Stat(dir); os.IsNotExist(err) {
194 return nil, fmt.Errorf("%w: %s", ErrSessionNotFound, id)
195 } else if err != nil {
196 return nil, err
197 }
198 return OpenWithOptions(dir, id, OpenOptions{ExternalHistory: true})
199 }
200
201 func (p *FilesystemPersistence) Stat(ctx context.Context, sessionID string) (SessionInfo, error) {
202 if err := ctx.Err(); err != nil {
203 return SessionInfo{}, err
204 }
205 id := strings.TrimSpace(sessionID)
206 if err := validateSessionID(id); err != nil {
207 return SessionInfo{}, err
208 }
209 dir, err := p.sessionDir(id, true)
210 if err != nil {
211 return SessionInfo{}, err
212 }
213 manifest, err := readCatalogManifest(filepath.Join(dir, "manifest.json"))
214 if err != nil {
215 if os.IsNotExist(err) {
216 return SessionInfo{}, fmt.Errorf("%w: %s", ErrSessionNotFound, id)
217 }
218 return SessionInfo{}, err
219 }
220 if manifest.SessionID != id {
221 return SessionInfo{}, fmt.Errorf("%w: manifest belongs to %q", ErrDamagedStore, manifest.SessionID)
222 }
223 header, hasHeader, err := readSessionHeader(dir, id)
224 if err != nil {
225 return SessionInfo{}, err
226 }
227 // Listing reads the manifest, log metadata, and rebuildable catalog cache.
228 // It never opens event bodies; Query refreshes missing display metadata in
229 // the background.
230 revision, err := revisionOfLog(dir)
231 if err != nil {
232 return SessionInfo{}, err
233 }
234 updatedAt := manifest.CreatedAt
235 if revision.Exists {
236 if stat, statErr := os.Stat(logPathForManifest(dir, manifest)); statErr == nil && stat.ModTime().After(updatedAt) {
237 updatedAt = stat.ModTime()
238 }
239 }
240 info := SessionInfo{SessionID: manifest.SessionID, Codec: manifest.Codec, CreatedAt: manifest.CreatedAt, UpdatedAt: updatedAt, MetadataStatus: MetadataPending, Path: dir}
241 if hasHeader {
242 info.CWD, info.ParentSessionID, info.Origin = header.CWD, header.ParentSessionID, header.Origin
243 }
244 cacheDir := filepath.Join(p.Root, ".query-cache", filepath.Base(id))
245 if metadata, metadataErr := readCatalogMetadata(cacheDir, manifest, revision); metadataErr == nil {
246 info.Title, info.TitleSequence = metadata.Title, metadata.TitleSequence
247 info.ModelRef, info.ModelIdentity = metadata.ModelRef, metadata.ModelIdentity
248 info.Turns, info.Preview, info.MetadataStatus = metadata.Turns, metadata.Preview, MetadataReady
249 info.EventSequence, info.ResultSequence = metadata.Sequence, metadata.ResultSequence
250 }
251 return info, nil
252 }
253
254 func readCatalogManifest(path string) (Manifest, error) {
255 data, err := os.ReadFile(path)
256 if err != nil {
257 return Manifest{}, err
258 }
259 var manifest Manifest
260 if err := json.Unmarshal(data, &manifest); err != nil {
261 return Manifest{}, err
262 }
263 if !supportedStoredManifest(manifest) {
264 return Manifest{}, fmt.Errorf("%w: manifest schema or codec", ErrUnsupportedVersion)
265 }
266 return manifest, nil
267 }
268
269 func (p *FilesystemPersistence) sessionDir(id string, mustExist bool) (string, error) {
270 if err := validateSessionID(id); err != nil {
271 return "", err
272 }
273 if !mustExist {
274 if err := os.MkdirAll(p.Root, 0o700); err != nil {
275 return "", err
276 }
277 }
278 root, err := os.OpenRoot(p.Root)
279 if os.IsNotExist(err) {
280 return "", fmt.Errorf("%w: %s", ErrSessionNotFound, id)
281 }
282 if err != nil {
283 return "", err
284 }
285 defer root.Close()
286 // Root.Lstat rejects traversal and follows the platform's reparse-point
287 // boundary rules. The single-segment validation above also keeps lock and
288 // cache names portable on Windows.
289 info, err := root.Lstat(id)
290 if os.IsNotExist(err) && !mustExist {
291 return filepath.Join(p.Root, id), nil
292 }
293 if os.IsNotExist(err) {
294 return "", fmt.Errorf("%w: %s", ErrSessionNotFound, id)
295 }
296 if err != nil {
297 return "", err
298 }
299 if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
300 return "", fmt.Errorf("session: session identity %q is not a confined directory", id)
301 }
302 // Resolve the physical path through the rooted handle. This keeps protocol
303 // input out of the path propagated through the storage stack.
304 child, err := root.OpenRoot(id)
305 if err != nil {
306 return "", err
307 }
308 dir := child.Name()
309 if err := child.Close(); err != nil {
310 return "", err
311 }
312 return dir, nil
313 }
314
315 func (p *FilesystemPersistence) List(ctx context.Context, cursor string, limit int) (SessionPage, error) {
316 if err := ctx.Err(); err != nil {
317 return SessionPage{}, err
318 }
319 if limit == 0 {
320 limit = 50
321 }
322 if limit < 1 || limit > 100 {
323 return SessionPage{}, fmt.Errorf("session: list limit must be 1..100")
324 }
325 entries, err := os.ReadDir(p.Root)
326 if os.IsNotExist(err) {
327 return SessionPage{Sessions: []SessionInfo{}}, nil
328 }
329 if err != nil {
330 return SessionPage{}, err
331 }
332 ids := make([]string, 0, len(entries))
333 for _, entry := range entries {
334 if err := ctx.Err(); err != nil {
335 return SessionPage{}, err
336 }
337 if entry.IsDir() && !strings.HasPrefix(entry.Name(), ".") && entry.Name() > cursor {
338 ids = append(ids, entry.Name())
339 }
340 }
341 sort.Strings(ids)
342 page := SessionPage{Sessions: []SessionInfo{}}
343 for _, id := range ids {
344 info, statErr := p.Stat(ctx, id)
345 if statErr != nil {
346 info = SessionInfo{SessionID: id, Path: filepath.Join(p.Root, id), Error: statErr.Error()}
347 }
348 if len(page.Sessions) == limit {
349 page.NextCursor = page.Sessions[len(page.Sessions)-1].SessionID
350 break
351 }
352 page.Sessions = append(page.Sessions, info)
353 }
354 return page, nil
355 }
356
357 func validateSessionID(id string) error {
358 id = strings.TrimSpace(id)
359 if len(id) == 0 || len(id) > 255 || strings.HasPrefix(id, ".") || strings.HasSuffix(id, ".") ||
360 !filepath.IsLocal(id) || id == "." || filepath.Base(id) != id || strings.ContainsAny(id, `/\\<>:"|?*`) {
361 return fmt.Errorf("session: invalid session id %q", id)
362 }
363 for _, char := range id {
364 if char < 0x20 {
365 return fmt.Errorf("session: invalid session id %q", id)
366 }
367 }
368 base := strings.ToUpper(strings.SplitN(id, ".", 2)[0])
369 if base == "CON" || base == "PRN" || base == "AUX" || base == "NUL" ||
370 (len(base) == 4 && (strings.HasPrefix(base, "COM") || strings.HasPrefix(base, "LPT")) && base[3] >= '1' && base[3] <= '9') {
371 return fmt.Errorf("session: reserved session id %q", id)
372 }
373 return nil
374 }
375
376 // ValidateSessionID applies the canonical session storage identity rules.
377 // Callers that accept compatibility routes must validate the extracted ID
378 // before deciding whether the input names a SessionRef or a legacy path.
379 func ValidateSessionID(id string) error {
380 return validateSessionID(id)
381 }
382
383 type readHandle struct {
384 id string
385 dir string
386 cacheDir string
387 manifest Manifest
388 mu sync.Mutex
389 closed bool
390 }
391
392 // openReadSession returns a cold Session backed only by the durable prefix. It
393 // performs no replay and takes no writer lease: paged reads remain the only way
394 // to consume it, which is what keeps catalog and history queries cheap.
395 func openReadSession(dir, id string, cacheDirs ...string) (*Session, error) {
396 handle, err := openReadHandle(dir, id, cacheDirs...)
397 if err != nil {
398 return nil, err
399 }
400 return newReadSession(handle), nil
401 }
402
403 func openReadHandle(dir, id string, cacheDirs ...string) (*readHandle, error) {
404 cacheDir := dir
405 if len(cacheDirs) > 0 && strings.TrimSpace(cacheDirs[0]) != "" {
406 cacheDir = cacheDirs[0]
407 }
408 manifest, err := readStoredManifest(filepath.Join(dir, "manifest.json"))
409 if err != nil {
410 if os.IsNotExist(err) {
411 return nil, fmt.Errorf("%w: %s", ErrSessionNotFound, id)
412 }
413 return nil, err
414 }
415 if manifest.SessionID != id {
416 return nil, fmt.Errorf("session: manifest belongs to %q", manifest.SessionID)
417 }
418 return &readHandle{id: id, dir: dir, cacheDir: cacheDir, manifest: manifest}, nil
419 }
420
421 func (h *readHandle) ID() string { return h.id }
422
423 func (h *readHandle) Manifest() Manifest { return h.manifest }
424
425 // Dir reports the directory this cold reader opened. A fork from a session with
426 // no live runtime still has to locate the parent's owned files, and that must
427 // not require acquiring the writer lease the cold reader deliberately avoids.
428 func (h *readHandle) Dir() string {
429 if h == nil {
430 return ""
431 }
432 return h.dir
433 }
434
435 func (h *readHandle) Read(ctx context.Context, offset uint64, limit int) (EventPage, error) {
436 if h == nil {
437 return EventPage{}, os.ErrClosed
438 }
439 h.mu.Lock()
440 closed := h.closed
441 dir, cacheDir := h.dir, h.cacheDir
442 h.mu.Unlock()
443 if closed {
444 return EventPage{}, os.ErrClosed
445 }
446 return readCommitPageWithCache(ctx, dir, cacheDir, offset, limit)
447 }
448
449 func (h *readHandle) Append(context.Context, []Commit) error {
450 return ErrReadOnly
451 }
452
453 func (h *readHandle) Sync(context.Context) (DurableReceipt, error) {
454 if h == nil {
455 return DurableReceipt{}, os.ErrClosed
456 }
457 h.mu.Lock()
458 closed := h.closed
459 dir := h.dir
460 h.mu.Unlock()
461 if closed {
462 return DurableReceipt{}, os.ErrClosed
463 }
464 sequence, err := lastDurableSequence(dir)
465 return DurableReceipt{DurableSequence: sequence}, err
466 }
467
468 func (h *readHandle) Close(context.Context) error {
469 if h == nil {
470 return nil
471 }
472 h.mu.Lock()
473 h.closed = true
474 h.mu.Unlock()
475 return nil
476 }
477
478 func (s *Store) Read(ctx context.Context, offset uint64, limit int) (EventPage, error) {
479 if s == nil {
480 return EventPage{}, os.ErrClosed
481 }
482 s.mu.Lock()
483 closed := s.closed
484 dir := s.dir
485 s.mu.Unlock()
486 if closed {
487 return EventPage{}, os.ErrClosed
488 }
489 return readCommitPage(ctx, dir, offset, limit)
490 }
491
492 func readCommitPage(ctx context.Context, dir string, offset uint64, limit int) (EventPage, error) {
493 return readCommitPageWithCache(ctx, dir, dir, offset, limit)
494 }
495
496 func readCommitPageWithCache(ctx context.Context, dir, cacheDir string, offset uint64, limit int) (EventPage, error) {
497 if err := ctx.Err(); err != nil {
498 return EventPage{}, err
499 }
500 if limit == 0 {
501 limit = 100
502 }
503 if limit < 1 || limit > 1000 {
504 return EventPage{}, fmt.Errorf("session: read limit must be 1..1000 commits")
505 }
506 index, err := loadOrBuildSparseIndex(ctx, dir, cacheDir)
507 if err != nil {
508 return EventPage{}, err
509 }
510 page := EventPage{Commits: []Commit{}}
511 if index.LastSequence <= offset {
512 return page, nil
513 }
514 checkpoint := index.checkpoint(offset)
515 manifest, err := readStoredManifest(filepath.Join(dir, "manifest.json"))
516 if err != nil {
517 return EventPage{}, err
518 }
519 file, err := os.Open(logPathForManifest(dir, manifest))
520 if err != nil {
521 return EventPage{}, err
522 }
523 defer file.Close()
524 visit := func(_ int64, commit Commit) bool {
525 if ctx.Err() != nil {
526 return false
527 }
528 if commit.LastSequence() <= offset {
529 return true
530 }
531 if len(page.Commits) == limit {
532 page.Truncated = true
533 return false
534 }
535 page.Commits = append(page.Commits, commit)
536 page.Next = commit.LastSequence()
537 return true
538 }
539 if manifest.Codec == Codec {
540 err = scanV4CommitFile(ctx, file, checkpoint.Offset, checkpoint.FirstSequence, contentStoreForSessionDir(dir), nil, visit)
541 } else {
542 err = scanCommitFileCodec(file, checkpoint.Offset, checkpoint.FirstSequence, manifest.Codec, nil, visit)
543 }
544 if err != nil {
545 return EventPage{}, err
546 }
547 if err := ctx.Err(); err != nil {
548 return EventPage{}, err
549 }
550 return page, nil
551 }
552
553 func lastDurableSequence(dir string) (uint64, error) {
554 return lastDurableSequenceWithCache(dir, dir)
555 }
556
557 func lastDurableSequenceWithCache(dir, cacheDir string) (uint64, error) {
558 index, err := loadOrBuildSparseIndex(context.Background(), dir, cacheDir)
559 return index.LastSequence, err
560 }
561
562 var _ SessionPersistence = (*FilesystemPersistence)(nil)
563 var _ SessionHandle = (*Store)(nil)
564 var _ SessionHandle = (*readHandle)(nil)
565
565 lines GO