1package core
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "crypto/sha256"
8 "encoding/hex"
9 "encoding/json"
10 "errors"
11 "fmt"
12 "io"
13 "os"
14 "path/filepath"
15 "sort"
16 "strings"
17 "time"
18
19 "github.com/bnema/zut/packages/provider"
20 "github.com/google/uuid"
21)
22
23// Session is a JSONL-backed conversation transcript tied to a cwd.
24type Session struct {
25 ID string
26 Path string
27 Meta SessionMeta
28 // Title is the effective title after applying append-only rename rows.
29 // Meta.Title remains the original meta-line value for compatibility.
30 Title string
31 // ExtensionState contains the latest opaque state snapshot recorded by
32 // each extension for this session/branch. It is never sent to providers.
33 ExtensionState map[string]json.RawMessage
34 writer *os.File
35 buf *bufio.Writer
36
37 // freshFile is true when the file was created by NewSession (this
38 // process owns it) and false when OpenSession reopened an existing
39 // transcript. Used by Close() to delete the file if the run never
40 // appended any messages — prevents a flood of empty session files
41 // from sessions the user opens then exits without prompting.
42 freshFile bool
43
44 // messagesAppended counts AppendMessage calls. Combined with
45 // freshFile it tells Close() whether the session left any content
46 // worth keeping.
47 messagesAppended int
48}
49
50// SessionMeta is written as the first line of every session file.
51type SessionMeta struct {
52 ID string `json:"id"`
53 CWD string `json:"cwd"`
54 Model string `json:"model"`
55 Provider string `json:"provider"`
56 Started time.Time `json:"started"`
57 Version string `json:"version"`
58 Title string `json:"title,omitempty"`
59
60 // CompactHandoff is opaque host-managed state for a live compaction
61 // continuation. It is session metadata, never provider context.
62 // Missing or invalid values are handled by the owning host as no handoff.
63 CompactHandoff json.RawMessage `json:"compact_handoff,omitempty"`
64
65 // Parent is the ID of the session this one was forked from, or
66 // empty for top-level sessions. The tree picker walks parents
67 // upward and sibling files (same cwd dir, same parent ID)
68 // laterally to render the branch topology.
69 Parent string `json:"parent,omitempty"`
70 // ForkPoint is the 0-indexed message position within the parent
71 // transcript where this branch diverges. Messages 0..ForkPoint-1
72 // are copied from the parent verbatim; the user's next turn on
73 // the child session continues from there.
74 ForkPoint int `json:"fork_point,omitempty"`
75 // HideFromSessions hides internal tree-navigation branches from the
76 // flat /sessions picker while keeping them available to /session tree.
77 HideFromSessions bool `json:"hide_from_sessions,omitempty"`
78}
79
80// sessionLine is the on-disk row type. Message is kept as a raw
81// JSON message on reads (because Content is an interface slice that
82// the default unmarshaler cannot reconstruct); it is written with a
83// regular provider.Message value.
84type sessionLine struct {
85 Type string `json:"type"`
86 Title string `json:"title,omitempty"`
87 Generated bool `json:"generated,omitempty"`
88 Meta *SessionMeta `json:"meta,omitempty"`
89 Message *provider.Message `json:"message,omitempty"`
90 Messages *[]provider.Message `json:"messages,omitempty"`
91 Usage *provider.Usage `json:"usage,omitempty"`
92 Cumulative *provider.Usage `json:"cumulative,omitempty"`
93 Extension string `json:"extension,omitempty"`
94 State json.RawMessage `json:"state,omitempty"`
95}
96
97type sessionLineHead struct {
98 Type string `json:"type"`
99}
100
101// SessionUsageCheckpoint records the cumulative usage known when a usage
102// row was written. MessageCount uses the effective transcript (after the
103// latest compaction and tool-pair repair), not the number of raw JSONL rows.
104type SessionUsageCheckpoint struct {
105 MessageCount int
106 Cumulative provider.Usage
107 CompactionGeneration int
108}
109
110// SessionSnapshot is the effective, in-memory view of a session file.
111// Messages are the transcript that a resumed agent should use. The raw
112// JSONL audit history is deliberately not exposed here because message
113// indices used by tree navigation and branching must agree.
114type SessionSnapshot struct {
115 Meta SessionMeta
116 Title string
117 Messages []provider.Message
118 ExtensionState map[string]json.RawMessage
119 UsageCheckpoints []SessionUsageCheckpoint
120 CompactionGeneration int
121}
122
123// SessionHistorySegment is one provider-valid transcript era. A compaction
124// starts a new segment containing the replacement summary and preserved tail;
125// earlier segments remain available for session-tree checkout without being
126// used to resume the live agent.
127type SessionHistorySegment struct {
128 Compacted bool
129 Messages []provider.Message
130 UsageCheckpoints []SessionUsageCheckpoint
131}
132
133// SessionHistory contains every persisted transcript era in chronological
134// order. Unlike SessionSnapshot, it intentionally retains segments replaced by
135// compaction so navigation can fork from older user and assistant messages.
136type SessionHistory struct {
137 Meta SessionMeta
138 Segments []SessionHistorySegment
139}
140
141// SessionsDir returns the per-cwd sessions directory under root.
142func SessionsDir(root, cwd string) string {
143 sum := sha256.Sum256([]byte(cwd))
144 short := hex.EncodeToString(sum[:8])
145 return filepath.Join(root, "sessions", short)
146}
147
148// NewSession creates and opens a new session file under
149// SessionsDir(root, cwd) with an autogenerated, time-stamped name.
150func NewSession(root, cwd, providerName, model, version string) (*Session, error) {
151 dir := SessionsDir(root, cwd)
152 if err := os.MkdirAll(dir, 0o755); err != nil {
153 return nil, err
154 }
155 id := uuid.NewString()
156 name := fmt.Sprintf("%s-%s.jsonl", time.Now().UTC().Format("20060102-150405"), id[:8])
157 p := filepath.Join(dir, name)
158 return newSessionAt(p, cwd, providerName, model, version)
159}
160
161// NewSessionAtPath creates a session at an explicit file path. Used
162// by callers (notably the subagent worker) that need the session
163// file to live at a path chosen by their parent rather than under
164// SessionsDir. Returns an error if the file already exists — use
165// OpenSession for that case.
166func NewSessionAtPath(path, cwd, providerName, model, version string) (*Session, error) {
167 if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
168 return nil, err
169 }
170 return newSessionAt(path, cwd, providerName, model, version)
171}
172
173// newSessionAt is the shared implementation. Both NewSession and
174// NewSessionAtPath funnel through here so the meta-line layout,
175// freshFile bookkeeping, and id format stay identical.
176func newSessionAt(p, cwd, providerName, model, version string) (*Session, error) {
177 id := uuid.NewString()
178 f, err := os.OpenFile(p, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644)
179 if err != nil {
180 return nil, err
181 }
182 s := &Session{
183 ID: id,
184 Path: p,
185 Meta: SessionMeta{ID: id, CWD: cwd, Provider: providerName, Model: model, Started: time.Now().UTC(), Version: version},
186 writer: f,
187 buf: bufio.NewWriter(f),
188 freshFile: true,
189 ExtensionState: map[string]json.RawMessage{},
190 }
191 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
192 f.Close()
193 return nil, err
194 }
195 return s, nil
196}
197
198func forEachJSONLLine(r io.Reader, fn func([]byte) error) error {
199 return forEachJSONLLineContext(context.Background(), r, fn)
200}
201
202func forEachJSONLLineContext(ctx context.Context, r io.Reader, fn func([]byte) error) error {
203 br := bufio.NewReader(r)
204 for {
205 if err := contextErr(ctx); err != nil {
206 return err
207 }
208 line, err := br.ReadBytes('\n')
209 if len(line) > 0 {
210 line = bytes.TrimRight(line, "\r\n")
211 if len(line) > 0 {
212 if ferr := fn(line); ferr != nil {
213 return ferr
214 }
215 }
216 }
217 if err == io.EOF {
218 return nil
219 }
220 if err != nil {
221 return err
222 }
223 }
224}
225
226func contextErr(ctx context.Context) error {
227 if ctx == nil {
228 return nil
229 }
230 select {
231 case <-ctx.Done():
232 return ctx.Err()
233 default:
234 return nil
235 }
236}
237
238// SessionUsage returns the most recent cumulative usage row stored in
239// a session file. Sessions append one usage row per completed turn; the
240// latest row's cumulative field is the session total. Missing usage rows
241// are valid for old/empty sessions and return the zero value.
242func SessionUsage(path string) (provider.Usage, error) {
243 cum, _, err := SessionUsageDetail(path)
244 return cum, err
245}
246
247// SessionUsageDetail returns the latest cumulative usage and the
248// per-turn usage of the final completed turn. It reads through the same
249// strict snapshot path as OpenSession, so a truncated or malformed session
250// cannot silently reset the usage display.
251func SessionUsageDetail(path string) (cumulative, lastTurn provider.Usage, err error) {
252 snapshot, err := ReadSessionSnapshot(path)
253 if err != nil {
254 return provider.Usage{}, provider.Usage{}, err
255 }
256 if len(snapshot.UsageCheckpoints) == 0 {
257 return provider.Usage{}, provider.Usage{}, nil
258 }
259 latestCheckpoint := snapshot.UsageCheckpoints[len(snapshot.UsageCheckpoints)-1]
260 if latestCheckpoint.CompactionGeneration < snapshot.CompactionGeneration {
261 // The latest usage row belongs to the pre-compaction context. Keep
262 // cumulative cost and totals, but reset the live context gauge until
263 // a usage row from the rebuilt transcript is persisted.
264 return latestCheckpoint.Cumulative, provider.Usage{}, nil
265 }
266 last := latestCheckpoint.Cumulative
267 var prev provider.Usage
268 if len(snapshot.UsageCheckpoints) > 1 {
269 prev = snapshot.UsageCheckpoints[len(snapshot.UsageCheckpoints)-2].Cumulative
270 }
271 lastTurn.InputTokens = nonNegDelta(last.InputTokens, prev.InputTokens)
272 lastTurn.OutputTokens = nonNegDelta(last.OutputTokens, prev.OutputTokens)
273 lastTurn.ReasoningTokens = nonNegDelta(last.ReasoningTokens, prev.ReasoningTokens)
274 lastTurn.ReasoningTokensKnown = last.ReasoningTokensKnown
275 lastTurn.CacheReadTokens = nonNegDelta(last.CacheReadTokens, prev.CacheReadTokens)
276 lastTurn.CacheWriteTokens = nonNegDelta(last.CacheWriteTokens, prev.CacheWriteTokens)
277 lastTurn.CostUSD = last.CostUSD - prev.CostUSD
278 if lastTurn.CostUSD < 0 {
279 lastTurn.CostUSD = 0
280 }
281 return last, lastTurn, nil
282}
283
284func nonNegDelta(cur, prev int) int {
285 if cur < prev {
286 return cur
287 }
288 return cur - prev
289}
290
291// ReadSessionSnapshot reads a complete session file and reconstructs the
292// effective transcript. It is intentionally stricter than the lightweight
293// listing helpers: a caller that is about to resume, render, or branch a
294// session must not receive a plausible-looking partial transcript.
295func ReadSessionSnapshot(path string) (SessionSnapshot, error) {
296 return readSessionSnapshot(context.Background(), path)
297}
298
299func readSessionSnapshot(ctx context.Context, path string) (SessionSnapshot, error) {
300 if err := contextErr(ctx); err != nil {
301 return SessionSnapshot{}, err
302 }
303 f, err := os.Open(path)
304 if err != nil {
305 return SessionSnapshot{}, fmt.Errorf("session snapshot: open %q: %w", path, err)
306 }
307 defer f.Close()
308
309 var snapshot SessionSnapshot
310 var sawMeta bool
311 generation := 0
312 type rawCheckpoint struct {
313 messageCount int
314 cumulative provider.Usage
315 generation int
316 }
317 var rawCheckpoints []rawCheckpoint
318 titleFromRename := false
319 extensionState := map[string]json.RawMessage{}
320
321 err = forEachStrictJSONLLineContext(ctx, f, func(line []byte, lineNo int) error {
322 var head sessionLineHead
323 if err := json.Unmarshal(line, &head); err != nil {
324 return fmt.Errorf("line %d: invalid JSON: %w", lineNo, err)
325 }
326 if head.Type == "" {
327 return fmt.Errorf("line %d: missing row type", lineNo)
328 }
329 if !sawMeta && head.Type != "meta" {
330 return fmt.Errorf("line %d: first row is not meta", lineNo)
331 }
332
333 switch head.Type {
334 case "meta":
335 var row sessionLine
336 if err := json.Unmarshal(line, &row); err != nil {
337 return fmt.Errorf("line %d: invalid meta row: %w", lineNo, err)
338 }
339 if row.Meta == nil || row.Meta.ID == "" {
340 return fmt.Errorf("line %d: meta row has no session id", lineNo)
341 }
342 snapshot.Meta = *row.Meta
343 if !titleFromRename && row.Meta.Title != "" {
344 snapshot.Title = row.Meta.Title
345 }
346 sawMeta = true
347
348 case "message":
349 msg, err := hydrateMessage(line)
350 if err != nil {
351 return fmt.Errorf("line %d: invalid message row: %w", lineNo, err)
352 }
353 snapshot.Messages = append(snapshot.Messages, msg)
354
355 case "compaction":
356 compacted, err := hydrateCompaction(line)
357 if err != nil {
358 return fmt.Errorf("line %d: invalid compaction row: %w", lineNo, err)
359 }
360 snapshot.Messages = compacted
361 generation++
362 cumulative, err := hydrateCompactionUsage(line)
363 if err != nil {
364 return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
365 }
366 if cumulative != nil {
367 rawCheckpoints = append(rawCheckpoints, rawCheckpoint{
368 messageCount: len(snapshot.Messages),
369 cumulative: *cumulative,
370 generation: generation,
371 })
372 }
373
374 case "usage":
375 var row struct {
376 Usage *provider.Usage `json:"usage"`
377 Cumulative *provider.Usage `json:"cumulative"`
378 }
379 if err := json.Unmarshal(line, &row); err != nil {
380 return fmt.Errorf("line %d: invalid usage row: %w", lineNo, err)
381 }
382 if row.Cumulative == nil {
383 return fmt.Errorf("line %d: usage row has no cumulative usage", lineNo)
384 }
385 rawCheckpoints = append(rawCheckpoints, rawCheckpoint{
386 messageCount: len(snapshot.Messages),
387 cumulative: *row.Cumulative,
388 generation: generation,
389 })
390
391 case "rename":
392 var row struct {
393 Title *string `json:"title"`
394 }
395 if err := json.Unmarshal(line, &row); err != nil {
396 return fmt.Errorf("line %d: invalid rename row: %w", lineNo, err)
397 }
398 if row.Title == nil {
399 return fmt.Errorf("line %d: rename row has no title", lineNo)
400 }
401 titleFromRename = true
402 snapshot.Title = *row.Title
403
404 case "extension_state":
405 var row sessionLine
406 if err := json.Unmarshal(line, &row); err != nil {
407 return fmt.Errorf("line %d: invalid extension state row: %w", lineNo, err)
408 }
409 if row.Extension != "" && len(row.State) <= maxExtensionStateBytes {
410 if len(row.State) == 0 || strings.TrimSpace(string(row.State)) == "null" {
411 delete(extensionState, row.Extension)
412 } else if json.Valid(row.State) {
413 extensionState[row.Extension] = append(json.RawMessage(nil), row.State...)
414 }
415 }
416
417 default:
418 return fmt.Errorf("line %d: unknown row type %q", lineNo, head.Type)
419 }
420 return nil
421 })
422 if err != nil {
423 return SessionSnapshot{}, fmt.Errorf("session snapshot %q: %w", path, err)
424 }
425 if !sawMeta {
426 return SessionSnapshot{}, fmt.Errorf("session snapshot %q: file is empty", path)
427 }
428
429 snapshot.CompactionGeneration = generation
430 snapshot.ExtensionState = extensionState
431
432 // Repair only after compaction replacement. This makes synthetic tool
433 // results and the removal of orphaned results part of the same indexed
434 // stream that OpenSession returns and that BranchSession materializes.
435 rawMessages := snapshot.Messages
436 repaired := repairSessionMessages(rawMessages)
437 snapshot.Messages = repaired.messages
438 for _, checkpoint := range rawCheckpoints {
439 count := checkpoint.messageCount
440 if checkpoint.generation == generation {
441 count = repaired.prefixCount(count)
442 } else if count > len(rawMessages) {
443 // A compaction replaces the effective stream. Preserve the
444 // cumulative cost represented by an older checkpoint at the
445 // end of the replacement rather than making every post-summary
446 // branch appear to have zero usage.
447 count = len(rawMessages)
448 }
449 snapshot.UsageCheckpoints = append(snapshot.UsageCheckpoints, SessionUsageCheckpoint{
450 MessageCount: count,
451 Cumulative: checkpoint.cumulative,
452 CompactionGeneration: checkpoint.generation,
453 })
454 }
455 return snapshot, nil
456}
457
458// ReadSessionHistory reads every provider-valid transcript era in a session.
459// The effective snapshot remains the source of truth for resume and ordinary
460// branching; this audit projection exists for session-tree navigation so a
461// compaction cannot make earlier fork points disappear.
462func ReadSessionHistory(path string) (SessionHistory, error) {
463 f, err := os.Open(path)
464 if err != nil {
465 return SessionHistory{}, fmt.Errorf("session history: open %q: %w", path, err)
466 }
467 defer f.Close()
468
469 var history SessionHistory
470 var sawMeta bool
471 type rawCheckpoint struct {
472 messageCount int
473 cumulative provider.Usage
474 }
475 type rawSegment struct {
476 compacted bool
477 messages []provider.Message
478 checkpoints []rawCheckpoint
479 }
480 current := rawSegment{}
481
482 appendCurrent := func() {
483 repaired := repairSessionMessages(current.messages)
484 segment := SessionHistorySegment{
485 Compacted: current.compacted,
486 Messages: repaired.messages,
487 }
488 for _, checkpoint := range current.checkpoints {
489 segment.UsageCheckpoints = append(segment.UsageCheckpoints, SessionUsageCheckpoint{
490 MessageCount: repaired.prefixCount(checkpoint.messageCount),
491 Cumulative: checkpoint.cumulative,
492 })
493 }
494 if segment.Compacted || len(segment.Messages) > 0 || len(segment.UsageCheckpoints) > 0 {
495 history.Segments = append(history.Segments, segment)
496 }
497 current = rawSegment{}
498 }
499
500 err = forEachStrictJSONLLine(f, func(line []byte, lineNo int) error {
501 var head sessionLineHead
502 if err := json.Unmarshal(line, &head); err != nil {
503 return fmt.Errorf("line %d: invalid JSON: %w", lineNo, err)
504 }
505 if head.Type == "" {
506 return fmt.Errorf("line %d: missing row type", lineNo)
507 }
508 if !sawMeta && head.Type != "meta" {
509 return fmt.Errorf("line %d: first row is not meta", lineNo)
510 }
511
512 switch head.Type {
513 case "meta":
514 var row sessionLine
515 if err := json.Unmarshal(line, &row); err != nil {
516 return fmt.Errorf("line %d: invalid meta row: %w", lineNo, err)
517 }
518 if row.Meta == nil || row.Meta.ID == "" {
519 return fmt.Errorf("line %d: meta row has no session id", lineNo)
520 }
521 history.Meta = *row.Meta
522 sawMeta = true
523
524 case "message":
525 message, err := hydrateMessage(line)
526 if err != nil {
527 return fmt.Errorf("line %d: invalid message row: %w", lineNo, err)
528 }
529 current.messages = append(current.messages, message)
530
531 case "compaction":
532 compacted, err := hydrateCompaction(line)
533 if err != nil {
534 return fmt.Errorf("line %d: invalid compaction row: %w", lineNo, err)
535 }
536 appendCurrent()
537 current = rawSegment{compacted: true, messages: compacted}
538 cumulative, err := hydrateCompactionUsage(line)
539 if err != nil {
540 return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
541 }
542 if cumulative != nil {
543 current.checkpoints = append(current.checkpoints, rawCheckpoint{
544 messageCount: len(current.messages),
545 cumulative: *cumulative,
546 })
547 }
548
549 case "usage":
550 var row struct {
551 Cumulative *provider.Usage `json:"cumulative"`
552 }
553 if err := json.Unmarshal(line, &row); err != nil {
554 return fmt.Errorf("line %d: invalid usage row: %w", lineNo, err)
555 }
556 if row.Cumulative == nil {
557 return fmt.Errorf("line %d: usage row has no cumulative usage", lineNo)
558 }
559 current.checkpoints = append(current.checkpoints, rawCheckpoint{
560 messageCount: len(current.messages),
561 cumulative: *row.Cumulative,
562 })
563
564 case "extension_state":
565 // Extension state is session metadata rather than transcript history.
566 // The snapshot reader restores it for resume; tree history only needs
567 // provider messages and usage checkpoints.
568
569 case "rename":
570 var row struct {
571 Title *string `json:"title"`
572 }
573 if err := json.Unmarshal(line, &row); err != nil {
574 return fmt.Errorf("line %d: invalid rename row: %w", lineNo, err)
575 }
576 if row.Title == nil {
577 return fmt.Errorf("line %d: rename row has no title", lineNo)
578 }
579
580 default:
581 return fmt.Errorf("line %d: unknown row type %q", lineNo, head.Type)
582 }
583 return nil
584 })
585 if err != nil {
586 return SessionHistory{}, fmt.Errorf("session history %q: %w", path, err)
587 }
588 if !sawMeta {
589 return SessionHistory{}, fmt.Errorf("session history %q: file is empty", path)
590 }
591 appendCurrent()
592 return history, nil
593}
594
595// forEachStrictJSONLLine is the complete-read counterpart to
596// forEachJSONLLine. It rejects malformed and blank rows and reports the
597// line number so callers cannot accidentally proceed with a partial file.
598func forEachStrictJSONLLine(r io.Reader, fn func([]byte, int) error) error {
599 return forEachStrictJSONLLineContext(context.Background(), r, fn)
600}
601
602func forEachStrictJSONLLineContext(ctx context.Context, r io.Reader, fn func([]byte, int) error) error {
603 br := bufio.NewReader(r)
604 lineNo := 0
605 for {
606 if err := contextErr(ctx); err != nil {
607 return err
608 }
609 line, err := br.ReadBytes('\n')
610 if len(line) > 0 {
611 lineNo++
612 line = bytes.TrimRight(line, "\r\n")
613 if len(bytes.TrimSpace(line)) == 0 {
614 return fmt.Errorf("line %d: blank row", lineNo)
615 }
616 if ferr := fn(line, lineNo); ferr != nil {
617 return ferr
618 }
619 }
620 if err == io.EOF {
621 return nil
622 }
623 if err != nil {
624 return err
625 }
626 }
627}
628
629// repairedSessionMessage keeps the raw message index that produced an
630// effective message. Synthetic tool results use the assistant's raw index;
631// this lets usage checkpoints continue to refer to the same effective
632// message prefix after repair removes or inserts rows.
633type repairedSessionMessage struct {
634 message provider.Message
635 rawIndex int
636}
637
638type repairedSessionMessages struct {
639 messages []provider.Message
640 counts []int
641}
642
643func (r repairedSessionMessages) prefixCount(rawCount int) int {
644 if rawCount < 0 {
645 return 0
646 }
647 if rawCount >= len(r.counts) {
648 return r.counts[len(r.counts)-1]
649 }
650 return r.counts[rawCount]
651}
652
653// repairSessionMessages repairs tool-call pairs and removes orphaned result
654// blocks while retaining a mapping from raw effective prefixes to repaired
655// prefixes. A tool result is valid only in the immediately following tool
656// message for the assistant tool calls; keeping an unrelated result there
657// would still make a resumed or branched transcript provider-invalid.
658func repairSessionMessages(msgs []provider.Message) repairedSessionMessages {
659 if len(msgs) == 0 {
660 return repairedSessionMessages{counts: []int{0}}
661 }
662
663 repaired := make([]repairedSessionMessage, 0, len(msgs)+2)
664 for i := 0; i < len(msgs); i++ {
665 m := msgs[i]
666 if m.Role == provider.RoleAssistant && messageHasToolCalls(m) {
667 ids := make(map[string]bool)
668 for _, c := range m.Content {
669 if tc, ok := c.(provider.ToolCallBlock); ok {
670 ids[tc.ID] = true
671 }
672 }
673 repaired = append(repaired, repairedSessionMessage{message: m, rawIndex: i})
674
675 if i+1 < len(msgs) && msgs[i+1].Role == provider.RoleTool {
676 tool := msgs[i+1]
677 seen := make(map[string]bool)
678 filtered := make([]provider.Content, 0, len(tool.Content)+len(ids))
679 for _, c := range tool.Content {
680 tr, ok := c.(provider.ToolResultBlock)
681 if !ok {
682 filtered = append(filtered, c)
683 continue
684 }
685 if !ids[tr.CallID] || seen[tr.CallID] {
686 continue
687 }
688 seen[tr.CallID] = true
689 filtered = append(filtered, tr)
690 }
691 for _, c := range m.Content {
692 if tc, ok := c.(provider.ToolCallBlock); ok && !seen[tc.ID] {
693 filtered = append(filtered, provider.ToolResultBlock{
694 CallID: tc.ID,
695 Content: []provider.Content{provider.TextBlock{Text: "tool call was aborted; no result recorded."}},
696 IsError: true,
697 })
698 }
699 }
700 if len(filtered) > 0 {
701 tool.Content = filtered
702 repaired = append(repaired, repairedSessionMessage{message: tool, rawIndex: i + 1})
703 }
704 i++
705 continue
706 }
707
708 stubs := make([]provider.Content, 0, len(ids))
709 for _, c := range m.Content {
710 if tc, ok := c.(provider.ToolCallBlock); ok {
711 stubs = append(stubs, provider.ToolResultBlock{
712 CallID: tc.ID,
713 Content: []provider.Content{provider.TextBlock{Text: "tool call was aborted; no result recorded."}},
714 IsError: true,
715 })
716 }
717 }
718 repaired = append(repaired, repairedSessionMessage{message: provider.Message{
719 Role: provider.RoleTool,
720 Content: stubs,
721 Time: m.Time,
722 }, rawIndex: i})
723 continue
724 }
725
726 if m.Role == provider.RoleTool {
727 // Deferred-tool activation rows intentionally carry a result
728 // without a preceding assistant call. Preserve those rows and
729 // their AddedToolNames marker; they are consumed as local
730 // transcript state rather than sent as a provider tool pair.
731 if len(m.AddedToolNames) > 0 {
732 repaired = append(repaired, repairedSessionMessage{message: m, rawIndex: i})
733 continue
734 }
735 // This tool row was not immediately paired with an assistant
736 // tool-call row. Drop its result blocks; a bare tool row is an
737 // orphan and is not safe to send to a provider.
738 filtered := make([]provider.Content, 0, len(m.Content))
739 for _, c := range m.Content {
740 if _, ok := c.(provider.ToolResultBlock); !ok {
741 filtered = append(filtered, c)
742 }
743 }
744 if len(filtered) > 0 {
745 m.Content = filtered
746 repaired = append(repaired, repairedSessionMessage{message: m, rawIndex: i})
747 }
748 continue
749 }
750 repaired = append(repaired, repairedSessionMessage{message: m, rawIndex: i})
751 }
752
753 out := repairedSessionMessages{
754 messages: make([]provider.Message, 0, len(repaired)),
755 counts: make([]int, len(msgs)+1),
756 }
757 for _, item := range repaired {
758 out.messages = append(out.messages, item.message)
759 }
760 // rawIndex values are produced in source order. Accumulate the number of
761 // repaired messages before each raw prefix in one forward pass instead of
762 // revisiting every remaining prefix for every repaired message.
763 repairedIndex := 0
764 for rawCount := range out.counts {
765 for repairedIndex < len(repaired) && repaired[repairedIndex].rawIndex < rawCount {
766 repairedIndex++
767 }
768 out.counts[rawCount] = repairedIndex
769 }
770 return out
771}
772
773func messageHasToolCalls(msg provider.Message) bool {
774 for _, content := range msg.Content {
775 if _, ok := content.(provider.ToolCallBlock); ok {
776 return true
777 }
778 }
779 return false
780}
781
782// OpenSession opens an existing session for appending.
783func OpenSession(path string) (*Session, []provider.Message, error) {
784 snapshot, err := ReadSessionSnapshot(path)
785 if err != nil {
786 return nil, nil, err
787 }
788 out, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
789 if err != nil {
790 return nil, nil, err
791 }
792 s := &Session{
793 ID: snapshot.Meta.ID,
794 Path: path,
795 Meta: snapshot.Meta,
796 Title: NormalizeSessionTitle(snapshot.Title),
797 ExtensionState: snapshot.ExtensionState,
798 writer: out,
799 buf: bufio.NewWriter(out),
800 }
801 return s, snapshot.Messages, nil
802}
803
804// repairToolUseResultPairs walks a restored transcript and
805// synthesises stub tool_result blocks for any assistant
806// tool_use blocks that aren't paired with a matching result in
807// the next message. Anthropic (and OpenAI via the responses API)
808// reject any request whose transcript leaves a tool_use without
809// its matching tool_result immediately after, with errors like:
810//
811// messages.8: `tool_use` ids were found without `tool_result`
812// blocks immediately after
813//
814// Corruption gets into the transcript two ways we know of:
815//
816// - Older zut builds that persisted the assistant tool_use row
817// before the tool_result row, then crashed between the two.
818// - Abort paths in older builds that didn't drop the mid-turn
819// assistant message cleanly.
820//
821// Rather than change runtime semantics (which would risk hiding a
822// real bug), we scrub on load: any unmatched tool_use gets a stub
823// tool_result injected as a RoleTool message so the next
824// outbound request passes the provider's validity check. The stub
825// reads "tool call was aborted; no result recorded." so the
826// model can see what happened and decide whether to retry.
827//
828// Runs once per OpenSession call. No cost on the hot path.
829func repairToolUseResultPairs(msgs []provider.Message) []provider.Message {
830 return repairSessionMessages(msgs).messages
831}
832
833// LatestSession returns the most recent session file for cwd, or "".
834func LatestSession(root, cwd string) string {
835 paths := ListSessions(root, cwd)
836 if len(paths) == 0 {
837 return ""
838 }
839 return paths[0]
840}
841
842// SessionSummary describes one on-disk session at a glance for UI pickers.
843type SessionSummary struct {
844 Path string
845 Started time.Time
846 Model string
847 Provider string
848 MessageCount int
849 FirstUserText string
850 TotalCost float64
851 Title string
852 BranchDepth int
853 // HideFromSessions marks internal branches omitted from the flat picker.
854 HideFromSessions bool
855}
856
857// RenameSession appends a sanitized rename line to the session file. This is
858// safe even for the currently active session because it opens the file
859// independently and appends (doesn't rewrite).
860func RenameSession(path, title string) error {
861 title = NormalizeSessionTitle(title)
862 if title == "" {
863 return errors.New("session title is empty")
864 }
865 f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0o644)
866 if err != nil {
867 return err
868 }
869 defer f.Close()
870 line, _ := json.Marshal(map[string]string{"type": "rename", "title": title})
871 line = append(line, '\n')
872 _, err = f.Write(line)
873 return err
874}
875
876// DeleteSession removes a session file. It refuses an empty path so
877// callers do not accidentally delete their cwd or another implicit target.
878func DeleteSession(path string) error {
879 if strings.TrimSpace(path) == "" {
880 return fmt.Errorf("session path is empty")
881 }
882 return os.Remove(path)
883}
884
885// DescribeSession returns a lightweight summary for one on-disk session.
886// It uses the effective snapshot so message counts, metadata, and usage agree
887// with resume and branching behavior. The summary retains HideFromSessions so
888// callers that load entries independently can apply the same picker filter as
889// DescribeSessions.
890func DescribeSession(path string) SessionSummary {
891 return DescribeSessionContext(context.Background(), path)
892}
893
894// DescribeSessionContext is the cancellation-aware form of DescribeSession.
895// Cancellation is checked between JSONL records so a closed picker does not
896// keep parsing a large transcript unnecessarily.
897func DescribeSessionContext(ctx context.Context, path string) SessionSummary {
898 return describeSessionContext(ctx, path)
899}
900
901// DescribeSessions returns lightweight summaries for every session in
902// cwd, newest first. It uses the effective snapshot so message counts,
903// metadata, and usage agree with resume and branching behavior.
904func DescribeSessions(root, cwd string) []SessionSummary {
905 paths := ListSessions(root, cwd)
906 summaries := make([]SessionSummary, 0, len(paths))
907 metas := make(map[string]SessionMeta, len(paths))
908 idToPath := make(map[string]string, len(paths))
909 for _, p := range paths {
910 if meta, err := readSessionMeta(p); err == nil && meta.ID != "" {
911 metas[p] = meta
912 idToPath[meta.ID] = p
913 }
914 }
915 for _, p := range paths {
916 if meta := metas[p]; meta.HideFromSessions {
917 continue
918 }
919 summaries = append(summaries, describeSession(p))
920 }
921 for idx := range summaries {
922 summaries[idx].BranchDepth = sessionBranchDepth(summaries[idx].Path, metas, idToPath)
923 }
924 return summaries
925}
926
927func sessionBranchDepth(path string, metas map[string]SessionMeta, idToPath map[string]string) int {
928 depth := 0
929 seen := map[string]bool{}
930 for {
931 meta, ok := metas[path]
932 if !ok || meta.Parent == "" || seen[meta.ID] {
933 return depth
934 }
935 seen[meta.ID] = true
936 parentPath := idToPath[meta.Parent]
937 if parentPath == "" {
938 return depth
939 }
940 depth++
941 path = parentPath
942 }
943}
944
945func describeSession(path string) SessionSummary {
946 return describeSessionContext(context.Background(), path)
947}
948
949func describeSessionContext(ctx context.Context, path string) SessionSummary {
950 s := SessionSummary{Path: path}
951 snapshot, err := readSessionSnapshot(ctx, path)
952 if err != nil {
953 return s
954 }
955 s.Started = snapshot.Meta.Started
956 s.Model = snapshot.Meta.Model
957 s.Provider = snapshot.Meta.Provider
958 s.HideFromSessions = snapshot.Meta.HideFromSessions
959 s.Title = NormalizeSessionTitle(snapshot.Title)
960 s.MessageCount = len(snapshot.Messages)
961 s.FirstUserText = firstUserTextFromMessages(snapshot.Messages)
962 lastUser := lastUserText(snapshot.Messages)
963 isBranch := snapshot.Meta.Parent != ""
964 forkPoint := snapshot.Meta.ForkPoint
965 branchPrompt := firstUserTextFromMessagesAfter(snapshot.Messages, forkPoint)
966 if len(snapshot.UsageCheckpoints) > 0 {
967 s.TotalCost = snapshot.UsageCheckpoints[len(snapshot.UsageCheckpoints)-1].Cumulative.CostUSD
968 }
969
970 // Rename rows are append-only and are intentionally kept out of the
971 // effective transcript. Scan only their placement so generated branch
972 // titles can still be replaced by the first post-fork prompt while an
973 // explicit rename after divergence keeps priority.
974 messageIdx := 0
975 renameMessageIdx := -1
976 renameGenerated := false
977 _ = func() error {
978 f, err := os.Open(path)
979 if err != nil {
980 return err
981 }
982 defer f.Close()
983 return forEachJSONLLineContext(ctx, f, func(line []byte) error {
984 var head sessionLineHead
985 if err := json.Unmarshal(line, &head); err != nil {
986 return nil
987 }
988 switch head.Type {
989 case "message":
990 messageIdx++
991 case "compaction":
992 if compacted, err := hydrateCompaction(line); err == nil {
993 messageIdx = len(compacted)
994 }
995 case "rename":
996 var row struct {
997 Title string `json:"title"`
998 Generated bool `json:"generated"`
999 }
1000 if err := json.Unmarshal(line, &row); err == nil && row.Title != "" {
1001 s.Title = NormalizeSessionTitle(row.Title)
1002 renameMessageIdx = messageIdx
1003 renameGenerated = row.Generated
1004 }
1005 }
1006 return nil
1007 })
1008 }()
1009 if isBranch {
1010 // A rename written after the branch has diverged is user intent
1011 // and must win. Older generated branch titles were written at
1012 // fork creation time, before any post-fork prompt, so those are
1013 // allowed to be replaced by the branch prompt fallback. Current
1014 // generated titles are marked so a fast hidden request that writes
1015 // before the first post-fork message still wins.
1016 explicitRename := !renameGenerated && renameMessageIdx > forkPoint
1017 if !renameGenerated && renameMessageIdx == forkPoint && s.Title != lastUser {
1018 explicitRename = true
1019 }
1020 if !renameGenerated && !explicitRename && branchPrompt != "" {
1021 s.Title = branchPrompt
1022 } else if s.Title == "" && lastUser != "" {
1023 s.Title = lastUser
1024 }
1025 }
1026 return s
1027}
1028
1029func firstUserTextFromMessages(msgs []provider.Message) string {
1030 return firstUserTextFromMessagesAfter(msgs, 0)
1031}
1032
1033func firstUserTextFromMessagesAfter(msgs []provider.Message, start int) string {
1034 if start < 0 {
1035 start = 0
1036 }
1037 for idx, msg := range msgs {
1038 if idx < start || msg.Role != provider.RoleUser {
1039 continue
1040 }
1041 if text := firstTextFromMessage(msg); text != "" {
1042 return text
1043 }
1044 }
1045 return ""
1046}
1047
1048func firstTextFromMessage(msg provider.Message) string {
1049 for _, c := range msg.Content {
1050 if tb, ok := c.(provider.TextBlock); ok && tb.Text != "" {
1051 return tb.Text
1052 }
1053 }
1054 return ""
1055}
1056
1057// PruneEmptySessions deletes session files in cwd's session directory
1058// that contain only a meta line (no messages were ever appended).
1059// Cleans up the backlog of empty stubs created by old zut versions
1060// that wrote a meta line at NewSession time and never followed up.
1061// Errors are swallowed; the caller treats this as best-effort.
1062func PruneEmptySessions(root, cwd string) {
1063 dir := SessionsDir(root, cwd)
1064 entries, err := os.ReadDir(dir)
1065 if err != nil {
1066 return
1067 }
1068 for _, e := range entries {
1069 if e.IsDir() || !strings.HasSuffix(e.Name(), ".jsonl") {
1070 continue
1071 }
1072 p := filepath.Join(dir, e.Name())
1073 if sessionHasNoMessages(p) {
1074 _ = os.Remove(p)
1075 }
1076 }
1077}
1078
1079// sessionHasNoMessages returns true when the file at path contains
1080// no lines of type "message". Meta-only / usage-only files count as
1081// empty. Used by PruneEmptySessions and the Describe path.
1082func sessionHasNoMessages(path string) bool {
1083 f, err := os.Open(path)
1084 if err != nil {
1085 return false
1086 }
1087 defer f.Close()
1088 hasMessage := false
1089 _ = forEachJSONLLine(f, func(line []byte) error {
1090 var head sessionLineHead
1091 if err := json.Unmarshal(line, &head); err != nil {
1092 return nil
1093 }
1094 if head.Type == "message" {
1095 hasMessage = true
1096 return io.EOF
1097 }
1098 return nil
1099 })
1100 return !hasMessage
1101}
1102
1103// ListSessions returns session file paths for cwd, most-recently-
1104// modified first. Sorting on filesystem ModTime instead of the
1105// timestamp embedded in the filename means a long-running session
1106// the user actually returned to recently floats to the top of
1107// /sessions, /continue, and the resume picker, even when it was
1108// originally created days earlier than newer but idle sessions.
1109// Files with identical ModTime fall back to filename desc so the
1110// order stays stable across calls.
1111func ListSessions(root, cwd string) []string {
1112 return ListSessionsContext(context.Background(), root, cwd)
1113}
1114
1115// ListSessionsContext is the cancellation-aware form of ListSessions. It
1116// checks cancellation while walking directory entries so a closed picker does
1117// not continue preparing a large result set.
1118func ListSessionsContext(ctx context.Context, root, cwd string) []string {
1119 if err := contextErr(ctx); err != nil {
1120 return nil
1121 }
1122 dir := SessionsDir(root, cwd)
1123 dirFile, err := os.Open(dir)
1124 if err != nil {
1125 return nil
1126 }
1127 defer dirFile.Close()
1128 type rec struct {
1129 path string
1130 mod time.Time
1131 }
1132 var files []rec
1133 for {
1134 if err := contextErr(ctx); err != nil {
1135 return nil
1136 }
1137 entries, readErr := dirFile.ReadDir(128)
1138 if readErr != nil && readErr != io.EOF {
1139 return nil
1140 }
1141 for _, e := range entries {
1142 if err := contextErr(ctx); err != nil {
1143 return nil
1144 }
1145 if e.IsDir() || !strings.HasSuffix(e.Name(), ".jsonl") {
1146 continue
1147 }
1148 p := filepath.Join(dir, e.Name())
1149 info, err := e.Info()
1150 if err != nil {
1151 continue
1152 }
1153 files = append(files, rec{path: p, mod: info.ModTime()})
1154 }
1155 if readErr == io.EOF {
1156 break
1157 }
1158 }
1159 sort.Slice(files, func(i, j int) bool {
1160 if !files[i].mod.Equal(files[j].mod) {
1161 return files[i].mod.After(files[j].mod)
1162 }
1163 return files[i].path > files[j].path
1164 })
1165 out := make([]string, 0, len(files))
1166 for _, r := range files {
1167 if err := contextErr(ctx); err != nil {
1168 return nil
1169 }
1170 out = append(out, r.path)
1171 }
1172 return out
1173}
1174
1175// AppendMessage writes a message to the session.
1176func (s *Session) AppendMessage(m provider.Message) error {
1177 if s == nil {
1178 return nil
1179 }
1180 if len(m.Content) == 0 {
1181 return errors.New("message has no content")
1182 }
1183 if err := s.writeLine(sessionLine{Type: "message", Message: &m}); err != nil {
1184 return err
1185 }
1186 s.messagesAppended++
1187 return nil
1188}
1189
1190// Sync flushes the session file's bytes through the filesystem. Interactive
1191// hosts use this at acknowledgement boundaries where a crash must not make a
1192// prompt look accepted before its user message is recoverable.
1193func (s *Session) Sync() error {
1194 if s == nil || s.writer == nil {
1195 return nil
1196 }
1197 if err := s.buf.Flush(); err != nil {
1198 return err
1199 }
1200 return s.writer.Sync()
1201}
1202
1203// AppendCompaction writes a checkpoint that replaces all earlier
1204// transcript rows when the session is resumed. The old rows remain in
1205// the JSONL file for audit/export, while loaders use the latest
1206// compaction row as the effective transcript.
1207func (s *Session) AppendCompaction(messages []provider.Message) error {
1208 if s == nil {
1209 return nil
1210 }
1211 compactionMessages := messages
1212 if err := s.writeLine(sessionLine{Type: "compaction", Messages: &compactionMessages}); err != nil {
1213 return err
1214 }
1215 // The compaction row itself is meaningful even when it replaces the
1216 // transcript with an empty or nil slice. Count the append operation so a
1217 // fresh session containing an empty checkpoint is not deleted on Close.
1218 s.messagesAppended++
1219 return nil
1220}
1221
1222// AppendCompactionWithUsage atomically records a replacement transcript and
1223// its cumulative usage in one JSONL row. Workers use this after compaction so
1224// a persistence failure cannot leave a resumable session with only part of the
1225// continued turn or without its matching usage checkpoint.
1226func (s *Session) AppendCompactionWithUsage(messages []provider.Message, cumulative provider.Usage) error {
1227 if s == nil {
1228 return nil
1229 }
1230 compactionMessages := messages
1231 if err := s.writeLine(sessionLine{
1232 Type: "compaction",
1233 Messages: &compactionMessages,
1234 Usage: &cumulative,
1235 Cumulative: &cumulative,
1236 }); err != nil {
1237 return err
1238 }
1239 s.messagesAppended++
1240 return nil
1241}
1242
1243// UpdateModel records a provider/model switch in the session file.
1244// The reader keeps the most recent meta entry, so the session resumes
1245// with the updated model.
1246func (s *Session) UpdateModel(providerName, model string) error {
1247 if s == nil {
1248 return nil
1249 }
1250 oldProvider, oldModel := s.Meta.Provider, s.Meta.Model
1251 s.Meta.Provider = providerName
1252 s.Meta.Model = model
1253 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
1254 s.Meta.Provider = oldProvider
1255 s.Meta.Model = oldModel
1256 return err
1257 }
1258 return nil
1259}
1260
1261// UpdateCompactHandoff records opaque host-owned compaction-handoff state in
1262// a new metadata row. A nil or JSON null value clears the state. The core
1263// session layer deliberately does not interpret the payload.
1264func (s *Session) UpdateCompactHandoff(state json.RawMessage) error {
1265 if s == nil {
1266 return nil
1267 }
1268 if len(state) > 0 && !json.Valid(state) {
1269 return errors.New("compact handoff state is not valid JSON")
1270 }
1271 if strings.TrimSpace(string(state)) == "null" {
1272 state = nil
1273 }
1274 previous := append(json.RawMessage(nil), s.Meta.CompactHandoff...)
1275 s.Meta.CompactHandoff = append(json.RawMessage(nil), state...)
1276 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
1277 s.Meta.CompactHandoff = previous
1278 return fmt.Errorf("update compact handoff: %w", err)
1279 }
1280 return nil
1281}
1282
1283// UpdateTitle records a session title without adding anything to the
1284// conversation transcript. The title is kept in memory as well so the live
1285// session can restore it when the user switches sessions in the TUI.
1286func (s *Session) UpdateTitle(title string) error {
1287 if s == nil {
1288 return nil
1289 }
1290 title = NormalizeSessionTitle(title)
1291 if title == "" {
1292 return nil
1293 }
1294 if err := s.writeLine(sessionLine{Type: "rename", Title: title, Generated: true}); err != nil {
1295 return fmt.Errorf("update session title: %w", err)
1296 }
1297 s.Title = title
1298 return nil
1299}
1300
1301// AppendUsage writes a usage row to the session.
1302func (s *Session) AppendUsage(u, cum provider.Usage) error {
1303 if s == nil {
1304 return nil
1305 }
1306 return s.writeLine(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum})
1307}
1308
1309const maxExtensionStateBytes = 256 * 1024
1310
1311// AppendExtensionState records the latest opaque extension snapshot for the
1312// active session/branch. State is persisted in the session file but is never
1313// included in provider requests. A nil or JSON null state clears the key.
1314func (s *Session) AppendExtensionState(extension string, state json.RawMessage) error {
1315 if s == nil {
1316 return nil
1317 }
1318 extension = strings.TrimSpace(extension)
1319 if extension == "" {
1320 return fmt.Errorf("extension state: extension name is empty")
1321 }
1322 if len(state) > maxExtensionStateBytes {
1323 return fmt.Errorf("extension state %q exceeds %d bytes", extension, maxExtensionStateBytes)
1324 }
1325 if len(state) > 0 && !json.Valid(state) {
1326 return fmt.Errorf("extension state %q is not valid JSON", extension)
1327 }
1328 copyState := append(json.RawMessage(nil), state...)
1329 if err := s.writeLine(sessionLine{Type: "extension_state", Extension: extension, State: copyState}); err != nil {
1330 return fmt.Errorf("append extension state: %w", err)
1331 }
1332 if s.ExtensionState == nil {
1333 s.ExtensionState = map[string]json.RawMessage{}
1334 }
1335 if len(copyState) == 0 || strings.TrimSpace(string(copyState)) == "null" {
1336 delete(s.ExtensionState, extension)
1337 } else {
1338 s.ExtensionState[extension] = copyState
1339 }
1340 return nil
1341}
1342
1343// Flush writes buffered session data to the append handle.
1344func (s *Session) Flush() error {
1345 if s == nil {
1346 return nil
1347 }
1348 return s.buf.Flush()
1349}
1350
1351// Close flushes and closes the session file. If the session was
1352// freshly created in this process and never had any messages
1353// appended (the user opened zut, looked around, and exited without
1354// prompting), the file is deleted on close so the sessions list
1355// doesn't fill up with empty meta-only stubs.
1356func (s *Session) Close() error {
1357 if s == nil {
1358 return nil
1359 }
1360 flushErr := s.Flush()
1361 closeErr := s.writer.Close()
1362 if s.freshFile && s.messagesAppended == 0 && len(s.ExtensionState) == 0 && len(s.Meta.CompactHandoff) == 0 {
1363 // Best-effort cleanup. We deliberately don't propagate the
1364 // remove error: if it fails (file already gone, perms changed)
1365 // the worst case is one stale empty file in the listing.
1366 _ = os.Remove(s.Path)
1367 }
1368 if flushErr != nil {
1369 return flushErr
1370 }
1371 return closeErr
1372}
1373
1374func (s *Session) writeLine(row sessionLine) error {
1375 b, err := json.Marshal(row)
1376 if err != nil {
1377 return err
1378 }
1379 if _, err := s.buf.Write(b); err != nil {
1380 return err
1381 }
1382 if err := s.buf.WriteByte('\n'); err != nil {
1383 return err
1384 }
1385 return s.buf.Flush()
1386}
1387
1388// ---- content (de)serialization ----
1389//
1390// provider.Content is an interface; encoding/json drops type information.
1391// We persist messages by reading the raw "message" object back and
1392// rebuilding Content from discriminated fields.
1393
1394func hydrateCompactionUsage(lineBytes []byte) (*provider.Usage, error) {
1395 var row struct {
1396 Cumulative *provider.Usage `json:"cumulative"`
1397 }
1398 if err := json.Unmarshal(lineBytes, &row); err != nil {
1399 return nil, err
1400 }
1401 return row.Cumulative, nil
1402}
1403
1404func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
1405 var row struct {
1406 Messages json.RawMessage `json:"messages"`
1407 }
1408 if err := json.Unmarshal(lineBytes, &row); err != nil {
1409 return nil, err
1410 }
1411 if len(row.Messages) == 0 || bytes.Equal(bytes.TrimSpace(row.Messages), []byte("null")) {
1412 // Older writers omitted the field for an empty compaction and some
1413 // wrote it as null. Both represent a valid empty replacement.
1414 return []provider.Message{}, nil
1415 }
1416 var rawMessages []json.RawMessage
1417 if err := json.Unmarshal(row.Messages, &rawMessages); err != nil {
1418 return nil, fmt.Errorf("invalid messages: %w", err)
1419 }
1420 messages := make([]provider.Message, 0, len(rawMessages))
1421 for idx, raw := range rawMessages {
1422 msg, err := hydrateMessageObject(raw)
1423 if err != nil {
1424 return nil, fmt.Errorf("message %d: %w", idx, err)
1425 }
1426 messages = append(messages, msg)
1427 }
1428 return messages, nil
1429}
1430
1431func hydrateMessage(lineBytes []byte) (provider.Message, error) {
1432 var row struct {
1433 Message json.RawMessage `json:"message"`
1434 }
1435 if err := json.Unmarshal(lineBytes, &row); err != nil {
1436 return provider.Message{}, err
1437 }
1438 if len(row.Message) == 0 || bytes.Equal(bytes.TrimSpace(row.Message), []byte("null")) {
1439 return provider.Message{}, fmt.Errorf("message row has no message")
1440 }
1441 return hydrateMessageObject(row.Message)
1442}
1443
1444func hydrateMessageObject(rawMessage []byte) (provider.Message, error) {
1445 var row struct {
1446 Role provider.Role `json:"role"`
1447 Content []json.RawMessage `json:"content"`
1448 Time time.Time `json:"time"`
1449 Meta map[string]string `json:"meta,omitempty"`
1450 AddedToolNames []string `json:"added_tool_names,omitempty"`
1451 }
1452 if err := json.Unmarshal(rawMessage, &row); err != nil {
1453 return provider.Message{}, err
1454 }
1455 if row.Role != provider.RoleUser && row.Role != provider.RoleAssistant && row.Role != provider.RoleTool {
1456 return provider.Message{}, fmt.Errorf("message has invalid role %q", row.Role)
1457 }
1458 if len(row.Content) == 0 {
1459 return provider.Message{}, fmt.Errorf("message has no content")
1460 }
1461 msg := provider.Message{Role: row.Role, Time: row.Time, Meta: row.Meta, AddedToolNames: row.AddedToolNames}
1462 for idx, raw := range row.Content {
1463 if len(bytes.TrimSpace(raw)) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
1464 return provider.Message{}, fmt.Errorf("content block %d is empty", idx)
1465 }
1466 var head struct {
1467 Text string `json:"text"`
1468 MimeType string `json:"mime_type"`
1469 Data []byte `json:"data"`
1470 ID string `json:"id"`
1471 Name string `json:"name"`
1472 CallID string `json:"call_id"`
1473 ReasoningID string `json:"reasoning_id"`
1474 Summary string `json:"summary"`
1475 Encrypted string `json:"encrypted_content"`
1476 ThoughtSignature string `json:"thought_signature"`
1477 // ToolCallBlock also has Arguments, ToolResultBlock has Content + IsError
1478 }
1479 if err := json.Unmarshal(raw, &head); err != nil {
1480 return provider.Message{}, fmt.Errorf("content block %d: %w", idx, err)
1481 }
1482 var fields map[string]json.RawMessage
1483 if err := json.Unmarshal(raw, &fields); err != nil || fields == nil {
1484 if err == nil {
1485 err = fmt.Errorf("content block is not an object")
1486 }
1487 return provider.Message{}, fmt.Errorf("content block %d: %w", idx, err)
1488 }
1489 // Discriminate by presence of fields.
1490 switch {
1491 case head.ReasoningID != "" || head.Summary != "" || head.Encrypted != "":
1492 msg.Content = append(msg.Content, provider.ReasoningBlock{
1493 ID: head.ReasoningID,
1494 Summary: head.Summary,
1495 Encrypted: head.Encrypted,
1496 })
1497 case head.Name != "" || head.ID != "" || fields["arguments"] != nil:
1498 if head.Name == "" || head.ID == "" {
1499 return provider.Message{}, fmt.Errorf("content block %d: incomplete tool call", idx)
1500 }
1501 var tc struct {
1502 ID string `json:"id"`
1503 Name string `json:"name"`
1504 Arguments json.RawMessage `json:"arguments"`
1505 ThoughtSignature string `json:"thought_signature"`
1506 }
1507 _ = json.Unmarshal(raw, &tc)
1508 msg.Content = append(msg.Content, provider.ToolCallBlock{
1509 ID: tc.ID,
1510 Name: tc.Name,
1511 Arguments: tc.Arguments,
1512 ThoughtSignature: tc.ThoughtSignature,
1513 })
1514 case head.CallID != "":
1515 var tr struct {
1516 CallID string `json:"call_id"`
1517 Content []json.RawMessage `json:"content"`
1518 IsError bool `json:"is_error"`
1519 }
1520 if err := json.Unmarshal(raw, &tr); err != nil || tr.Content == nil {
1521 if err == nil {
1522 err = fmt.Errorf("tool result has no content")
1523 }
1524 return provider.Message{}, fmt.Errorf("content block %d: %w", idx, err)
1525 }
1526 block := provider.ToolResultBlock{CallID: tr.CallID, IsError: tr.IsError}
1527 for innerIdx, c := range tr.Content {
1528 var inner struct {
1529 Text string `json:"text"`
1530 MimeType string `json:"mime_type"`
1531 Data []byte `json:"data"`
1532 }
1533 if err := json.Unmarshal(c, &inner); err != nil {
1534 return provider.Message{}, fmt.Errorf("content block %d result %d: %w", idx, innerIdx, err)
1535 }
1536 var innerFields map[string]json.RawMessage
1537 if err := json.Unmarshal(c, &innerFields); err != nil || innerFields == nil {
1538 if err == nil {
1539 err = fmt.Errorf("result content is not an object")
1540 }
1541 return provider.Message{}, fmt.Errorf("content block %d result %d: %w", idx, innerIdx, err)
1542 }
1543 if inner.MimeType != "" {
1544 block.Content = append(block.Content, provider.ImageBlock{MimeType: inner.MimeType, Data: inner.Data})
1545 } else if _, ok := innerFields["text"]; ok {
1546 block.Content = append(block.Content, provider.TextBlock{Text: inner.Text})
1547 } else {
1548 return provider.Message{}, fmt.Errorf("content block %d result %d: unknown content", idx, innerIdx)
1549 }
1550 }
1551 if len(block.Content) == 0 {
1552 return provider.Message{}, fmt.Errorf("content block %d tool result has no content", idx)
1553 }
1554 msg.Content = append(msg.Content, block)
1555 case head.MimeType != "":
1556 msg.Content = append(msg.Content, provider.ImageBlock{
1557 MimeType: head.MimeType,
1558 Data: head.Data,
1559 ThoughtSignature: head.ThoughtSignature,
1560 })
1561 default:
1562 if _, ok := fields["text"]; !ok {
1563 return provider.Message{}, fmt.Errorf("content block %d: unknown content", idx)
1564 }
1565 msg.Content = append(msg.Content, provider.TextBlock{
1566 Text: head.Text,
1567 ThoughtSignature: head.ThoughtSignature,
1568 })
1569 }
1570 }
1571 return msg, nil
1572}