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