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