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