1package core
2
3import (
4 "bufio"
5 "bytes"
6 "crypto/sha256"
7 "encoding/hex"
8 "encoding/json"
9 "fmt"
10 "io"
11 "os"
12 "path/filepath"
13 "sort"
14 "strings"
15 "time"
16
17 "github.com/google/uuid"
18 "github.com/patriceckhart/zot/internal/provider"
19)
20
21// Session is a JSONL-backed conversation transcript tied to a cwd.
22type Session struct {
23 ID string
24 Path string
25 Meta SessionMeta
26 writer *os.File
27 buf *bufio.Writer
28
29 // freshFile is true when the file was created by NewSession (this
30 // process owns it) and false when OpenSession reopened an existing
31 // transcript. Used by Close() to delete the file if the run never
32 // appended any messages — prevents a flood of empty session files
33 // from sessions the user opens then exits without prompting.
34 freshFile bool
35
36 // messagesAppended counts AppendMessage calls. Combined with
37 // freshFile it tells Close() whether the session left any content
38 // worth keeping.
39 messagesAppended int
40}
41
42// SessionMeta is written as the first line of every session file.
43type SessionMeta struct {
44 ID string `json:"id"`
45 CWD string `json:"cwd"`
46 Model string `json:"model"`
47 Provider string `json:"provider"`
48 Started time.Time `json:"started"`
49 Version string `json:"version"`
50 Title string `json:"title,omitempty"`
51
52 // Parent is the ID of the session this one was forked from, or
53 // empty for top-level sessions. The tree picker walks parents
54 // upward and sibling files (same cwd dir, same parent ID)
55 // laterally to render the branch topology.
56 Parent string `json:"parent,omitempty"`
57 // ForkPoint is the 0-indexed message position within the parent
58 // transcript where this branch diverges. Messages 0..ForkPoint-1
59 // are copied from the parent verbatim; the user's next turn on
60 // the child session continues from there.
61 ForkPoint int `json:"fork_point,omitempty"`
62}
63
64// sessionLine is the on-disk row type. Message is kept as a raw
65// JSON message on reads (because Content is an interface slice that
66// the default unmarshaler cannot reconstruct); it is written with a
67// regular provider.Message value.
68type sessionLine struct {
69 Type string `json:"type"`
70 Meta *SessionMeta `json:"meta,omitempty"`
71 Message *provider.Message `json:"message,omitempty"`
72 Messages []provider.Message `json:"messages,omitempty"`
73 Usage *provider.Usage `json:"usage,omitempty"`
74 Cumulative *provider.Usage `json:"cumulative,omitempty"`
75}
76
77type sessionLineHead struct {
78 Type string `json:"type"`
79}
80
81// SessionsDir returns the per-cwd sessions directory under root.
82func SessionsDir(root, cwd string) string {
83 sum := sha256.Sum256([]byte(cwd))
84 short := hex.EncodeToString(sum[:8])
85 return filepath.Join(root, "sessions", short)
86}
87
88// NewSession creates and opens a new session file.
89func NewSession(root, cwd, providerName, model, version string) (*Session, error) {
90 dir := SessionsDir(root, cwd)
91 if err := os.MkdirAll(dir, 0o755); err != nil {
92 return nil, err
93 }
94 id := uuid.NewString()
95 name := fmt.Sprintf("%s-%s.jsonl", time.Now().UTC().Format("20060102-150405"), id[:8])
96 p := filepath.Join(dir, name)
97 f, err := os.OpenFile(p, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644)
98 if err != nil {
99 return nil, err
100 }
101 s := &Session{
102 ID: id,
103 Path: p,
104 Meta: SessionMeta{ID: id, CWD: cwd, Provider: providerName, Model: model, Started: time.Now().UTC(), Version: version},
105 writer: f,
106 buf: bufio.NewWriter(f),
107 freshFile: true,
108 }
109 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
110 f.Close()
111 return nil, err
112 }
113 return s, nil
114}
115
116func forEachJSONLLine(r io.Reader, fn func([]byte) error) error {
117 br := bufio.NewReader(r)
118 for {
119 line, err := br.ReadBytes('\n')
120 if len(line) > 0 {
121 line = bytes.TrimRight(line, "\r\n")
122 if len(line) > 0 {
123 if ferr := fn(line); ferr != nil {
124 return ferr
125 }
126 }
127 }
128 if err == io.EOF {
129 return nil
130 }
131 if err != nil {
132 return err
133 }
134 }
135}
136
137// SessionUsage returns the most recent cumulative usage row stored in
138// a session file. Sessions append one usage row per completed turn; the
139// latest row's cumulative field is the session total. Missing usage rows
140// are valid for old/empty sessions and return the zero value.
141func SessionUsage(path string) (provider.Usage, error) {
142 f, err := os.Open(path)
143 if err != nil {
144 return provider.Usage{}, err
145 }
146 defer f.Close()
147
148 var usage provider.Usage
149 if err := forEachJSONLLine(f, func(line []byte) error {
150 var head sessionLineHead
151 if err := json.Unmarshal(line, &head); err != nil || head.Type != "usage" {
152 return nil
153 }
154 var row struct {
155 Cumulative provider.Usage `json:"cumulative"`
156 }
157 if err := json.Unmarshal(line, &row); err == nil {
158 usage = row.Cumulative
159 }
160 return nil
161 }); err != nil {
162 return provider.Usage{}, err
163 }
164 return usage, nil
165}
166
167// OpenSession opens an existing session for appending.
168func OpenSession(path string) (*Session, []provider.Message, error) {
169 f, err := os.Open(path)
170 if err != nil {
171 return nil, nil, err
172 }
173 defer f.Close()
174
175 var meta SessionMeta
176 var messages []provider.Message
177 if err := forEachJSONLLine(f, func(line []byte) error {
178 var head sessionLineHead
179 if err := json.Unmarshal(line, &head); err != nil {
180 return nil
181 }
182 switch head.Type {
183 case "meta":
184 var row struct {
185 Meta SessionMeta `json:"meta"`
186 }
187 if err := json.Unmarshal(line, &row); err == nil {
188 meta = row.Meta
189 }
190 case "message":
191 if msg, err := hydrateMessage(line); err == nil && len(msg.Content) > 0 {
192 messages = append(messages, msg)
193 }
194 case "compaction":
195 if compacted, err := hydrateCompaction(line); err == nil {
196 messages = compacted
197 }
198 }
199 return nil
200 }); err != nil {
201 return nil, nil, err
202 }
203 messages = repairToolUseResultPairs(messages)
204 out, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
205 if err != nil {
206 return nil, nil, err
207 }
208 s := &Session{ID: meta.ID, Path: path, Meta: meta, writer: out, buf: bufio.NewWriter(out)}
209 return s, messages, nil
210}
211
212// repairToolUseResultPairs walks a restored transcript and
213// synthesises stub tool_result blocks for any assistant
214// tool_use blocks that aren't paired with a matching result in
215// the next message. Anthropic (and OpenAI via the responses API)
216// reject any request whose transcript leaves a tool_use without
217// its matching tool_result immediately after, with errors like:
218//
219// messages.8: `tool_use` ids were found without `tool_result`
220// blocks immediately after
221//
222// Corruption gets into the transcript two ways we know of:
223//
224// - Older zot builds that persisted the assistant tool_use row
225// before the tool_result row, then crashed between the two.
226// - Abort paths in older builds that didn't drop the mid-turn
227// assistant message cleanly.
228//
229// Rather than change runtime semantics (which would risk hiding a
230// real bug), we scrub on load: any unmatched tool_use gets a stub
231// tool_result injected as a RoleTool message so the next
232// outbound request passes the provider's validity check. The stub
233// reads "tool call was aborted; no result recorded." so the
234// model can see what happened and decide whether to retry.
235//
236// Runs once per OpenSession call. No cost on the hot path.
237func repairToolUseResultPairs(msgs []provider.Message) []provider.Message {
238 if len(msgs) == 0 {
239 return msgs
240 }
241 out := make([]provider.Message, 0, len(msgs)+2)
242 for i, m := range msgs {
243 out = append(out, m)
244 if m.Role != provider.RoleAssistant {
245 continue
246 }
247 // Collect tool_use ids in this assistant message.
248 var ids []string
249 for _, c := range m.Content {
250 if tc, ok := c.(provider.ToolCallBlock); ok {
251 ids = append(ids, tc.ID)
252 }
253 }
254 if len(ids) == 0 {
255 continue
256 }
257 // Look at the next message (if any) and collect tool_result
258 // CallIDs it covers.
259 have := map[string]bool{}
260 if i+1 < len(msgs) && msgs[i+1].Role == provider.RoleTool {
261 for _, c := range msgs[i+1].Content {
262 if tr, ok := c.(provider.ToolResultBlock); ok {
263 have[tr.CallID] = true
264 }
265 }
266 }
267 // Build stubs for any missing id.
268 var stubs []provider.Content
269 for _, id := range ids {
270 if have[id] {
271 continue
272 }
273 stubs = append(stubs, provider.ToolResultBlock{
274 CallID: id,
275 Content: []provider.Content{provider.TextBlock{Text: "tool call was aborted; no result recorded."}},
276 IsError: true,
277 })
278 }
279 if len(stubs) == 0 {
280 continue
281 }
282 // Merge into the next tool-role message if present,
283 // otherwise insert a synthetic one right after the
284 // assistant message. Merging keeps the tool-role row
285 // count stable; inserting handles the common case where
286 // no tool message was persisted at all.
287 if i+1 < len(msgs) && msgs[i+1].Role == provider.RoleTool {
288 msgs[i+1].Content = append(msgs[i+1].Content, stubs...)
289 // We already appended m to out; the modified next
290 // message will be appended on the following iteration.
291 continue
292 }
293 out = append(out, provider.Message{
294 Role: provider.RoleTool,
295 Content: stubs,
296 Time: m.Time,
297 })
298 }
299 return out
300}
301
302// LatestSession returns the most recent session file for cwd, or "".
303func LatestSession(root, cwd string) string {
304 paths := ListSessions(root, cwd)
305 if len(paths) == 0 {
306 return ""
307 }
308 return paths[0]
309}
310
311// SessionSummary describes one on-disk session at a glance for UI pickers.
312type SessionSummary struct {
313 Path string
314 Started time.Time
315 Model string
316 Provider string
317 MessageCount int
318 FirstUserText string
319 TotalCost float64
320 Title string
321}
322
323// RenameSession updates the title field in the session's meta line.
324// It rewrites the first line of the file (the meta line) with the
325// updated title.
326// RenameSession appends a rename line to the session file. This is
327// safe even for the currently active session because it opens the
328// file independently and appends (doesn't rewrite).
329func RenameSession(path, title string) error {
330 f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0o644)
331 if err != nil {
332 return err
333 }
334 defer f.Close()
335 line, _ := json.Marshal(map[string]string{"type": "rename", "title": title})
336 line = append(line, '\n')
337 _, err = f.Write(line)
338 return err
339}
340
341// DescribeSessions returns lightweight summaries for every session in
342// cwd, newest first. Parses only the first few lines and the last usage
343// line so it's cheap to run on every dialog open.
344func DescribeSessions(root, cwd string) []SessionSummary {
345 paths := ListSessions(root, cwd)
346 summaries := make([]SessionSummary, 0, len(paths))
347 for _, p := range paths {
348 summaries = append(summaries, describeSession(p))
349 }
350 return summaries
351}
352
353func describeSession(path string) SessionSummary {
354 s := SessionSummary{Path: path}
355 f, err := os.Open(path)
356 if err != nil {
357 return s
358 }
359 defer f.Close()
360 _ = forEachJSONLLine(f, func(line []byte) error {
361 var head sessionLineHead
362 if err := json.Unmarshal(line, &head); err != nil {
363 return nil
364 }
365 switch head.Type {
366 case "meta":
367 var row struct {
368 Meta SessionMeta `json:"meta"`
369 }
370 if err := json.Unmarshal(line, &row); err == nil {
371 s.Started = row.Meta.Started
372 s.Model = row.Meta.Model
373 s.Provider = row.Meta.Provider
374 s.Title = row.Meta.Title
375 }
376 case "message":
377 s.MessageCount++
378 if s.FirstUserText == "" {
379 s.FirstUserText = firstUserText(line)
380 }
381 case "compaction":
382 if compacted, err := hydrateCompaction(line); err == nil {
383 s.MessageCount = len(compacted)
384 if s.FirstUserText == "" && len(compacted) > 0 {
385 s.FirstUserText = firstTextFromMessage(compacted[0])
386 }
387 }
388 case "rename":
389 var row struct {
390 Title string `json:"title"`
391 }
392 if err := json.Unmarshal(line, &row); err == nil && row.Title != "" {
393 s.Title = row.Title
394 }
395 case "usage":
396 var row struct {
397 Cumulative provider.Usage `json:"cumulative"`
398 }
399 if err := json.Unmarshal(line, &row); err == nil {
400 s.TotalCost = row.Cumulative.CostUSD
401 }
402 }
403 return nil
404 })
405 return s
406}
407
408func firstUserText(line []byte) string {
409 var row struct {
410 Message struct {
411 Role string `json:"role"`
412 Content []struct {
413 Text string `json:"text"`
414 } `json:"content"`
415 } `json:"message"`
416 }
417 if err := json.Unmarshal(line, &row); err != nil {
418 return ""
419 }
420 if row.Message.Role != "user" {
421 return ""
422 }
423 for _, c := range row.Message.Content {
424 if c.Text != "" {
425 return c.Text
426 }
427 }
428 return ""
429}
430
431func firstTextFromMessage(msg provider.Message) string {
432 for _, c := range msg.Content {
433 if tb, ok := c.(provider.TextBlock); ok && tb.Text != "" {
434 return tb.Text
435 }
436 }
437 return ""
438}
439
440// PruneEmptySessions deletes session files in cwd's session directory
441// that contain only a meta line (no messages were ever appended).
442// Cleans up the backlog of empty stubs created by old zot versions
443// that wrote a meta line at NewSession time and never followed up.
444// Errors are swallowed; the caller treats this as best-effort.
445func PruneEmptySessions(root, cwd string) {
446 dir := SessionsDir(root, cwd)
447 entries, err := os.ReadDir(dir)
448 if err != nil {
449 return
450 }
451 for _, e := range entries {
452 if e.IsDir() || !strings.HasSuffix(e.Name(), ".jsonl") {
453 continue
454 }
455 p := filepath.Join(dir, e.Name())
456 if sessionHasNoMessages(p) {
457 _ = os.Remove(p)
458 }
459 }
460}
461
462// sessionHasNoMessages returns true when the file at path contains
463// no lines of type "message". Meta-only / usage-only files count as
464// empty. Used by PruneEmptySessions and the Describe path.
465func sessionHasNoMessages(path string) bool {
466 f, err := os.Open(path)
467 if err != nil {
468 return false
469 }
470 defer f.Close()
471 hasMessage := false
472 _ = forEachJSONLLine(f, func(line []byte) error {
473 var head sessionLineHead
474 if err := json.Unmarshal(line, &head); err != nil {
475 return nil
476 }
477 if head.Type == "message" {
478 hasMessage = true
479 return io.EOF
480 }
481 return nil
482 })
483 return !hasMessage
484}
485
486// ListSessions returns session file paths for cwd, most-recently-
487// modified first. Sorting on filesystem ModTime instead of the
488// timestamp embedded in the filename means a long-running session
489// the user actually returned to recently floats to the top of
490// /sessions, /continue, and the resume picker, even when it was
491// originally created days earlier than newer but idle sessions.
492// Files with identical ModTime fall back to filename desc so the
493// order stays stable across calls.
494func ListSessions(root, cwd string) []string {
495 dir := SessionsDir(root, cwd)
496 entries, err := os.ReadDir(dir)
497 if err != nil {
498 return nil
499 }
500 type rec struct {
501 path string
502 mod time.Time
503 }
504 var files []rec
505 for _, e := range entries {
506 if e.IsDir() || !strings.HasSuffix(e.Name(), ".jsonl") {
507 continue
508 }
509 p := filepath.Join(dir, e.Name())
510 info, err := e.Info()
511 if err != nil {
512 continue
513 }
514 files = append(files, rec{path: p, mod: info.ModTime()})
515 }
516 sort.Slice(files, func(i, j int) bool {
517 if !files[i].mod.Equal(files[j].mod) {
518 return files[i].mod.After(files[j].mod)
519 }
520 return files[i].path > files[j].path
521 })
522 out := make([]string, 0, len(files))
523 for _, r := range files {
524 out = append(out, r.path)
525 }
526 return out
527}
528
529// AppendMessage writes a message to the session.
530func (s *Session) AppendMessage(m provider.Message) error {
531 if s == nil {
532 return nil
533 }
534 if err := s.writeLine(sessionLine{Type: "message", Message: &m}); err != nil {
535 return err
536 }
537 s.messagesAppended++
538 return nil
539}
540
541// AppendCompaction writes a checkpoint that replaces all earlier
542// transcript rows when the session is resumed. The old rows remain in
543// the JSONL file for audit/export, while loaders use the latest
544// compaction row as the effective transcript.
545func (s *Session) AppendCompaction(messages []provider.Message) error {
546 if s == nil {
547 return nil
548 }
549 if err := s.writeLine(sessionLine{Type: "compaction", Messages: messages}); err != nil {
550 return err
551 }
552 s.messagesAppended = len(messages)
553 return nil
554}
555
556// UpdateModel records a provider/model switch in the session file.
557// The reader keeps the most recent meta entry, so the session resumes
558// with the updated model.
559func (s *Session) UpdateModel(providerName, model string) error {
560 if s == nil {
561 return nil
562 }
563 s.Meta.Provider = providerName
564 s.Meta.Model = model
565 return s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta})
566}
567
568// AppendUsage writes a usage row to the session.
569func (s *Session) AppendUsage(u, cum provider.Usage) error {
570 if s == nil {
571 return nil
572 }
573 return s.writeLine(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum})
574}
575
576// Close flushes and closes the session file. If the session was
577// freshly created in this process and never had any messages
578// appended (the user opened zot, looked around, and exited without
579// prompting), the file is deleted on close so the sessions list
580// doesn't fill up with empty meta-only stubs.
581func (s *Session) Close() error {
582 if s == nil {
583 return nil
584 }
585 flushErr := s.buf.Flush()
586 closeErr := s.writer.Close()
587 if s.freshFile && s.messagesAppended == 0 {
588 // Best-effort cleanup. We deliberately don't propagate the
589 // remove error: if it fails (file already gone, perms changed)
590 // the worst case is one stale empty file in the listing.
591 _ = os.Remove(s.Path)
592 }
593 if flushErr != nil {
594 return flushErr
595 }
596 return closeErr
597}
598
599func (s *Session) writeLine(row sessionLine) error {
600 b, err := json.Marshal(row)
601 if err != nil {
602 return err
603 }
604 if _, err := s.buf.Write(b); err != nil {
605 return err
606 }
607 if err := s.buf.WriteByte('\n'); err != nil {
608 return err
609 }
610 return s.buf.Flush()
611}
612
613// ---- content (de)serialization ----
614//
615// provider.Content is an interface; encoding/json drops type information.
616// We persist messages by reading the raw "message" object back and
617// rebuilding Content from discriminated fields.
618
619func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
620 var row struct {
621 Messages []json.RawMessage `json:"messages"`
622 }
623 if err := json.Unmarshal(lineBytes, &row); err != nil {
624 return nil, err
625 }
626 messages := make([]provider.Message, 0, len(row.Messages))
627 for _, raw := range row.Messages {
628 msg, err := hydrateMessageObject(raw)
629 if err == nil && len(msg.Content) > 0 {
630 messages = append(messages, msg)
631 }
632 }
633 return messages, nil
634}
635
636func hydrateMessage(lineBytes []byte) (provider.Message, error) {
637 var row struct {
638 Message json.RawMessage `json:"message"`
639 }
640 if err := json.Unmarshal(lineBytes, &row); err != nil {
641 return provider.Message{}, err
642 }
643 return hydrateMessageObject(row.Message)
644}
645
646func hydrateMessageObject(rawMessage []byte) (provider.Message, error) {
647 var row struct {
648 Role provider.Role `json:"role"`
649 Content []json.RawMessage `json:"content"`
650 Time time.Time `json:"time"`
651 Meta map[string]string `json:"meta,omitempty"`
652 }
653 if err := json.Unmarshal(rawMessage, &row); err != nil {
654 return provider.Message{}, err
655 }
656 msg := provider.Message{Role: row.Role, Time: row.Time, Meta: row.Meta}
657 for _, raw := range row.Content {
658 var head struct {
659 Text string `json:"text"`
660 MimeType string `json:"mime_type"`
661 Data []byte `json:"data"`
662 ID string `json:"id"`
663 Name string `json:"name"`
664 CallID string `json:"call_id"`
665 ReasoningID string `json:"reasoning_id"`
666 Summary string `json:"summary"`
667 Encrypted string `json:"encrypted_content"`
668 // ToolCallBlock also has Arguments, ToolResultBlock has Content + IsError
669 }
670 if err := json.Unmarshal(raw, &head); err != nil {
671 continue
672 }
673 // Discriminate by presence of fields.
674 switch {
675 case head.ReasoningID != "" || head.Encrypted != "":
676 msg.Content = append(msg.Content, provider.ReasoningBlock{
677 ID: head.ReasoningID,
678 Summary: head.Summary,
679 Encrypted: head.Encrypted,
680 })
681 case head.Name != "" && head.ID != "":
682 var tc struct {
683 ID string `json:"id"`
684 Name string `json:"name"`
685 Arguments json.RawMessage `json:"arguments"`
686 }
687 _ = json.Unmarshal(raw, &tc)
688 msg.Content = append(msg.Content, provider.ToolCallBlock{ID: tc.ID, Name: tc.Name, Arguments: tc.Arguments})
689 case head.CallID != "":
690 var tr struct {
691 CallID string `json:"call_id"`
692 Content []json.RawMessage `json:"content"`
693 IsError bool `json:"is_error"`
694 }
695 _ = json.Unmarshal(raw, &tr)
696 block := provider.ToolResultBlock{CallID: tr.CallID, IsError: tr.IsError}
697 for _, c := range tr.Content {
698 var inner struct {
699 Text string `json:"text"`
700 MimeType string `json:"mime_type"`
701 Data []byte `json:"data"`
702 }
703 _ = json.Unmarshal(c, &inner)
704 if inner.MimeType != "" {
705 block.Content = append(block.Content, provider.ImageBlock{MimeType: inner.MimeType, Data: inner.Data})
706 } else {
707 block.Content = append(block.Content, provider.TextBlock{Text: inner.Text})
708 }
709 }
710 msg.Content = append(msg.Content, block)
711 case head.MimeType != "":
712 msg.Content = append(msg.Content, provider.ImageBlock{MimeType: head.MimeType, Data: head.Data})
713 default:
714 msg.Content = append(msg.Content, provider.TextBlock{Text: head.Text})
715 }
716 }
717 return msg, nil
718}