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