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