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