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 CWD string
959 Started time.Time
960 Model string
961 Provider string
962 MessageCount int
963 FirstUserText string
964 TotalCost float64
965 Title string
966 BranchDepth int
967 // HideFromSessions marks internal branches omitted from the flat picker.
968 HideFromSessions bool
969}
970
971// RenameSession appends a sanitized rename line to the session file. This is
972// safe even for the currently active session because it opens the file
973// independently and appends (doesn't rewrite).
974func RenameSession(path, title string) error {
975 title = NormalizeSessionTitle(title)
976 if title == "" {
977 return errors.New("session title is empty")
978 }
979 f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0o644)
980 if err != nil {
981 return err
982 }
983 defer f.Close()
984 line, _ := json.Marshal(map[string]string{"type": "rename", "title": title})
985 line = append(line, '\n')
986 _, err = f.Write(line)
987 return err
988}
989
990// DeleteSession removes a session file. It refuses an empty path so
991// callers do not accidentally delete their cwd or another implicit target.
992func DeleteSession(path string) error {
993 if strings.TrimSpace(path) == "" {
994 return fmt.Errorf("session path is empty")
995 }
996 return os.Remove(path)
997}
998
999// DescribeSession returns a lightweight summary for one on-disk session.
1000// It uses the effective snapshot so message counts, metadata, and usage agree
1001// with resume and branching behavior. The summary retains HideFromSessions so
1002// callers that load entries independently can apply the same picker filter as
1003// DescribeSessions.
1004func DescribeSession(path string) SessionSummary {
1005 return DescribeSessionContext(context.Background(), path)
1006}
1007
1008// DescribeSessionContext is the cancellation-aware form of DescribeSession.
1009// Cancellation is checked between JSONL records so a closed picker does not
1010// keep parsing a large transcript unnecessarily.
1011func DescribeSessionContext(ctx context.Context, path string) SessionSummary {
1012 return describeSessionContext(ctx, path)
1013}
1014
1015// DescribeSessions returns lightweight summaries for every session in
1016// cwd, newest first. It uses the effective snapshot so message counts,
1017// metadata, and usage agree with resume and branching behavior.
1018func DescribeSessions(root, cwd string) []SessionSummary {
1019 paths := ListSessions(root, cwd)
1020 summaries := make([]SessionSummary, 0, len(paths))
1021 metas := make(map[string]SessionMeta, len(paths))
1022 idToPath := make(map[string]string, len(paths))
1023 for _, p := range paths {
1024 if meta, err := readSessionMeta(p); err == nil && meta.ID != "" {
1025 metas[p] = meta
1026 idToPath[meta.ID] = p
1027 }
1028 }
1029 for _, p := range paths {
1030 if meta := metas[p]; meta.HideFromSessions {
1031 continue
1032 }
1033 summaries = append(summaries, describeSession(p))
1034 }
1035 for idx := range summaries {
1036 summaries[idx].BranchDepth = sessionBranchDepth(summaries[idx].Path, metas, idToPath)
1037 }
1038 return summaries
1039}
1040
1041func sessionBranchDepth(path string, metas map[string]SessionMeta, idToPath map[string]string) int {
1042 depth := 0
1043 seen := map[string]bool{}
1044 for {
1045 meta, ok := metas[path]
1046 if !ok || meta.Parent == "" || seen[meta.ID] {
1047 return depth
1048 }
1049 seen[meta.ID] = true
1050 parentPath := idToPath[meta.Parent]
1051 if parentPath == "" {
1052 return depth
1053 }
1054 depth++
1055 path = parentPath
1056 }
1057}
1058
1059func describeSession(path string) SessionSummary {
1060 return describeSessionContext(context.Background(), path)
1061}
1062
1063func describeSessionContext(ctx context.Context, path string) SessionSummary {
1064 s := SessionSummary{Path: path}
1065 snapshot, err := readSessionSnapshot(ctx, path)
1066 if err != nil {
1067 return s
1068 }
1069 s.CWD = snapshot.Meta.CWD
1070 s.Started = snapshot.Meta.Started
1071 s.Model = snapshot.Meta.Model
1072 s.Provider = snapshot.Meta.Provider
1073 s.HideFromSessions = snapshot.Meta.HideFromSessions
1074 s.Title = NormalizeSessionTitle(snapshot.Title)
1075 s.MessageCount = len(snapshot.Messages)
1076 s.FirstUserText = firstUserTextFromMessages(snapshot.Messages)
1077 lastUser := lastUserText(snapshot.Messages)
1078 isBranch := snapshot.Meta.Parent != ""
1079 forkPoint := snapshot.Meta.ForkPoint
1080 branchPrompt := firstUserTextFromMessagesAfter(snapshot.Messages, forkPoint)
1081 if len(snapshot.UsageCheckpoints) > 0 {
1082 s.TotalCost = snapshot.UsageCheckpoints[len(snapshot.UsageCheckpoints)-1].Cumulative.CostUSD
1083 }
1084
1085 // Rename rows are append-only and are intentionally kept out of the
1086 // effective transcript. Scan only their placement so generated branch
1087 // titles can still be replaced by the first post-fork prompt while an
1088 // explicit rename after divergence keeps priority.
1089 messageIdx := 0
1090 renameMessageIdx := -1
1091 renameGenerated := false
1092 _ = func() error {
1093 f, err := os.Open(path)
1094 if err != nil {
1095 return err
1096 }
1097 defer f.Close()
1098 return forEachJSONLLineContext(ctx, f, func(line []byte) error {
1099 var head sessionLineHead
1100 if err := json.Unmarshal(line, &head); err != nil {
1101 return nil
1102 }
1103 switch head.Type {
1104 case "message":
1105 messageIdx++
1106 case "compaction":
1107 if compacted, err := hydrateCompaction(line); err == nil {
1108 messageIdx = len(compacted)
1109 }
1110 case "rename":
1111 var row struct {
1112 Title string `json:"title"`
1113 Generated bool `json:"generated"`
1114 }
1115 if err := json.Unmarshal(line, &row); err == nil && row.Title != "" {
1116 s.Title = NormalizeSessionTitle(row.Title)
1117 renameMessageIdx = messageIdx
1118 renameGenerated = row.Generated
1119 }
1120 }
1121 return nil
1122 })
1123 }()
1124 if isBranch {
1125 // A rename written after the branch has diverged is user intent
1126 // and must win. Older generated branch titles were written at
1127 // fork creation time, before any post-fork prompt, so those are
1128 // allowed to be replaced by the branch prompt fallback. Current
1129 // generated titles are marked so a fast hidden request that writes
1130 // before the first post-fork message still wins.
1131 explicitRename := !renameGenerated && renameMessageIdx > forkPoint
1132 if !renameGenerated && renameMessageIdx == forkPoint && s.Title != lastUser {
1133 explicitRename = true
1134 }
1135 if !renameGenerated && !explicitRename && branchPrompt != "" {
1136 s.Title = branchPrompt
1137 } else if s.Title == "" && lastUser != "" {
1138 s.Title = lastUser
1139 }
1140 }
1141 return s
1142}
1143
1144func firstUserTextFromMessages(msgs []provider.Message) string {
1145 return firstUserTextFromMessagesAfter(msgs, 0)
1146}
1147
1148func firstUserTextFromMessagesAfter(msgs []provider.Message, start int) string {
1149 if start < 0 {
1150 start = 0
1151 }
1152 for idx, msg := range msgs {
1153 if idx < start || msg.Role != provider.RoleUser {
1154 continue
1155 }
1156 if text := firstTextFromMessage(msg); text != "" {
1157 return text
1158 }
1159 }
1160 return ""
1161}
1162
1163func firstTextFromMessage(msg provider.Message) string {
1164 for _, c := range msg.Content {
1165 if tb, ok := c.(provider.TextBlock); ok && tb.Text != "" {
1166 return tb.Text
1167 }
1168 }
1169 return ""
1170}
1171
1172// PruneEmptySessions deletes session files in cwd's session directory
1173// that contain only a meta line (no messages were ever appended).
1174// Cleans up the backlog of empty stubs created by old zut versions
1175// that wrote a meta line at NewSession time and never followed up.
1176// Errors are swallowed; the caller treats this as best-effort.
1177func PruneEmptySessions(root, cwd string) {
1178 dir := SessionsDir(root, cwd)
1179 entries, err := os.ReadDir(dir)
1180 if err != nil {
1181 return
1182 }
1183 for _, e := range entries {
1184 if e.IsDir() || !strings.HasSuffix(e.Name(), ".jsonl") {
1185 continue
1186 }
1187 p := filepath.Join(dir, e.Name())
1188 if sessionHasNoMessages(p) {
1189 _ = os.Remove(p)
1190 }
1191 }
1192}
1193
1194// sessionHasNoMessages returns true when the file at path contains
1195// no lines of type "message". Meta-only / usage-only files count as
1196// empty. Used by PruneEmptySessions and the Describe path.
1197func sessionHasNoMessages(path string) bool {
1198 f, err := os.Open(path)
1199 if err != nil {
1200 return false
1201 }
1202 defer f.Close()
1203 hasMessage := false
1204 _ = forEachJSONLLine(f, func(line []byte) error {
1205 var head sessionLineHead
1206 if err := json.Unmarshal(line, &head); err != nil {
1207 return nil
1208 }
1209 if head.Type == "message" {
1210 hasMessage = true
1211 return io.EOF
1212 }
1213 return nil
1214 })
1215 return !hasMessage
1216}
1217
1218// ListSessions returns session file paths for cwd, most-recently-
1219// modified first. Sorting on filesystem ModTime instead of the
1220// timestamp embedded in the filename means a long-running session
1221// the user actually returned to recently floats to the top of
1222// /sessions, /continue, and the resume picker, even when it was
1223// originally created days earlier than newer but idle sessions.
1224// Files with identical ModTime fall back to filename desc so the
1225// order stays stable across calls.
1226func ListSessions(root, cwd string) []string {
1227 return ListSessionsContext(context.Background(), root, cwd)
1228}
1229
1230// ListSessionsContext is the cancellation-aware form of ListSessions. It
1231// checks cancellation while walking directory entries so a closed picker does
1232// not continue preparing a large result set.
1233func ListSessionsContext(ctx context.Context, root, cwd string) []string {
1234 if err := contextErr(ctx); err != nil {
1235 return nil
1236 }
1237 dir := SessionsDir(root, cwd)
1238 dirFile, err := os.Open(dir)
1239 if err != nil {
1240 return nil
1241 }
1242 defer dirFile.Close()
1243 type rec struct {
1244 path string
1245 mod time.Time
1246 }
1247 var files []rec
1248 for {
1249 if err := contextErr(ctx); err != nil {
1250 return nil
1251 }
1252 entries, readErr := dirFile.ReadDir(128)
1253 if readErr != nil && readErr != io.EOF {
1254 return nil
1255 }
1256 for _, e := range entries {
1257 if err := contextErr(ctx); err != nil {
1258 return nil
1259 }
1260 if e.IsDir() || !strings.HasSuffix(e.Name(), ".jsonl") {
1261 continue
1262 }
1263 p := filepath.Join(dir, e.Name())
1264 info, err := e.Info()
1265 if err != nil {
1266 continue
1267 }
1268 files = append(files, rec{path: p, mod: info.ModTime()})
1269 }
1270 if readErr == io.EOF {
1271 break
1272 }
1273 }
1274 sort.Slice(files, func(i, j int) bool {
1275 if !files[i].mod.Equal(files[j].mod) {
1276 return files[i].mod.After(files[j].mod)
1277 }
1278 return files[i].path > files[j].path
1279 })
1280 out := make([]string, 0, len(files))
1281 for _, r := range files {
1282 if err := contextErr(ctx); err != nil {
1283 return nil
1284 }
1285 out = append(out, r.path)
1286 }
1287 return out
1288}
1289
1290// AppendMessage writes a message to the session.
1291func (s *Session) AppendMessage(m provider.Message) error {
1292 if s == nil {
1293 return nil
1294 }
1295 if len(m.Content) == 0 {
1296 return errors.New("message has no content")
1297 }
1298 if err := s.writeLine(sessionLine{Type: "message", Message: &m}); err != nil {
1299 return err
1300 }
1301 s.messagesAppended++
1302 return nil
1303}
1304
1305// Sync flushes the session file's bytes through the filesystem. Interactive
1306// hosts use this at acknowledgement boundaries where a crash must not make a
1307// prompt look accepted before its user message is recoverable.
1308func (s *Session) Sync() error {
1309 if s == nil || s.writer == nil {
1310 return nil
1311 }
1312 if err := s.buf.Flush(); err != nil {
1313 return err
1314 }
1315 return s.writer.Sync()
1316}
1317
1318// AppendCompaction writes a checkpoint that replaces all earlier
1319// transcript rows when the session is resumed. The old rows remain in
1320// the JSONL file for audit/export, while loaders use the latest
1321// compaction row as the effective transcript.
1322func (s *Session) AppendCompaction(messages []provider.Message) error {
1323 if s == nil {
1324 return nil
1325 }
1326 compactionMessages := messages
1327 if err := s.writeLine(sessionLine{Type: "compaction", Messages: &compactionMessages}); err != nil {
1328 return err
1329 }
1330 // The compaction row itself is meaningful even when it replaces the
1331 // transcript with an empty or nil slice. Count the append operation so a
1332 // fresh session containing an empty checkpoint is not deleted on Close.
1333 s.messagesAppended++
1334 return nil
1335}
1336
1337// AppendCompactionWithUsage atomically records a replacement transcript and
1338// its cumulative usage in one JSONL row. Workers use this after compaction so
1339// a persistence failure cannot leave a resumable session with only part of the
1340// continued turn or without its matching usage checkpoint.
1341func (s *Session) AppendCompactionWithUsage(messages []provider.Message, cumulative provider.Usage) error {
1342 if s == nil {
1343 return nil
1344 }
1345 compactionMessages := messages
1346 if err := s.writeUsageCheckpoint(sessionLine{
1347 Type: "compaction",
1348 Messages: &compactionMessages,
1349 Usage: &cumulative,
1350 Cumulative: &cumulative,
1351 }); err != nil {
1352 return err
1353 }
1354 s.messagesAppended++
1355 return nil
1356}
1357
1358// UpdateModel records a provider/model switch in the session file.
1359// The reader keeps the most recent meta entry, so the session resumes
1360// with the updated model.
1361func (s *Session) UpdateModel(providerName, model string) error {
1362 if s == nil {
1363 return nil
1364 }
1365 oldProvider, oldModel := s.Meta.Provider, s.Meta.Model
1366 s.Meta.Provider = providerName
1367 s.Meta.Model = model
1368 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
1369 s.Meta.Provider = oldProvider
1370 s.Meta.Model = oldModel
1371 return err
1372 }
1373 return nil
1374}
1375
1376// UpdateCompactHandoff records opaque host-owned compaction-handoff state in
1377// a new metadata row. A nil or JSON null value clears the state. The core
1378// session layer deliberately does not interpret the payload.
1379func (s *Session) UpdateCompactHandoff(state json.RawMessage) error {
1380 if s == nil {
1381 return nil
1382 }
1383 if len(state) > 0 && !json.Valid(state) {
1384 return errors.New("compact handoff state is not valid JSON")
1385 }
1386 if strings.TrimSpace(string(state)) == "null" {
1387 state = nil
1388 }
1389 previous := append(json.RawMessage(nil), s.Meta.CompactHandoff...)
1390 s.Meta.CompactHandoff = append(json.RawMessage(nil), state...)
1391 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
1392 s.Meta.CompactHandoff = previous
1393 return fmt.Errorf("update compact handoff: %w", err)
1394 }
1395 return nil
1396}
1397
1398// UpdateGoal records or clears the current autonomous session goal.
1399const maxMissionGoalTransitions = 16
1400
1401// EnsureMission creates the durable user-intent boundary when this session has
1402// none. It never replaces an existing mission; explicit user controls own
1403// replacement and cancellation semantics.
1404func (s *Session) EnsureMission(objective string, source MissionSource) error {
1405 if s == nil || strings.TrimSpace(objective) == "" || s.Meta.Mission != nil {
1406 return nil
1407 }
1408 previous := s.Meta.Mission
1409 s.Meta.Mission = &SessionMission{
1410 ID: uuid.NewString(),
1411 Objective: strings.TrimSpace(objective),
1412 Status: MissionActive,
1413 Source: source,
1414 }
1415 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
1416 s.Meta.Mission = previous
1417 return fmt.Errorf("ensure mission: %w", err)
1418 }
1419 return nil
1420}
1421
1422// UpdateGoal persists one linear goal transition. It lazily migrates legacy
1423// goal-only sessions into a mission and rejects unbounded manager progression.
1424func (s *Session) UpdateGoal(goal *SessionGoal) error {
1425 if s == nil {
1426 return nil
1427 }
1428 previous := cloneSessionGoal(s.Meta.Goal)
1429 previousMission := cloneSessionMission(s.Meta.Mission)
1430 previousHistory := append([]SessionGoal(nil), s.Meta.GoalHistory...)
1431 if previous != nil {
1432 s.Meta.GoalHistory = append(s.Meta.GoalHistory, *previous)
1433 }
1434 if goal == nil {
1435 s.Meta.Goal = nil
1436 s.Meta.Mission = nil
1437 } else {
1438 copyGoal := *goal
1439 if err := s.assignGoalToMission(©Goal, previous); err != nil {
1440 s.Meta.Mission = previousMission
1441 s.Meta.GoalHistory = previousHistory
1442 return err
1443 }
1444 s.Meta.Goal = ©Goal
1445 }
1446 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
1447 s.Meta.Goal = previous
1448 s.Meta.Mission = previousMission
1449 s.Meta.GoalHistory = previousHistory
1450 return fmt.Errorf("update goal: %w", err)
1451 }
1452 return nil
1453}
1454
1455// UpdateGoalRuntime persists mutable execution state without treating each
1456// continuation as a mission transition. The goal ID must match the current
1457// goal so stale workers cannot overwrite a newer user or manager goal.
1458func (s *Session) UpdateGoalRuntime(goal *SessionGoal) error {
1459 if s == nil {
1460 return nil
1461 }
1462 if goal == nil || s.Meta.Goal == nil || goal.ID == "" || goal.ID != s.Meta.Goal.ID {
1463 return errors.New("goal runtime update does not match the current goal")
1464 }
1465 previous := cloneSessionGoal(s.Meta.Goal)
1466 copyGoal := *s.Meta.Goal
1467 copyGoal.TokensUsed = goal.TokensUsed
1468 copyGoal.ContinuationID = goal.ContinuationID
1469 copyGoal.ConsecutiveNoProgressTurns = goal.ConsecutiveNoProgressTurns
1470 s.Meta.Goal = ©Goal
1471 if err := s.writeLine(sessionLine{Type: "meta", Meta: &s.Meta}); err != nil {
1472 s.Meta.Goal = previous
1473 return fmt.Errorf("update goal runtime: %w", err)
1474 }
1475 return nil
1476}
1477
1478func (s *Session) assignGoalToMission(goal, previous *SessionGoal) error {
1479 if s.Meta.Mission == nil {
1480 source := MissionSourceUser
1481 if goal.Owner == GoalOwnerManager {
1482 source = MissionSourceManager
1483 }
1484 s.Meta.Mission = &SessionMission{
1485 ID: uuid.NewString(),
1486 Objective: goal.Objective,
1487 Status: MissionActive,
1488 Source: source,
1489 }
1490 }
1491 mission := s.Meta.Mission
1492 if goal.ID == "" {
1493 goal.ID = uuid.NewString()
1494 }
1495 isNewManagerGoal := goal.Status == GoalActive && goal.Owner == GoalOwnerManager && (previous == nil || goal.ID != previous.ID)
1496 if isNewManagerGoal {
1497 mission.TransitionCount++
1498 if mission.TransitionCount > maxMissionGoalTransitions {
1499 return fmt.Errorf("mission goal transition limit reached")
1500 }
1501 }
1502 goal.MissionID = mission.ID
1503 if goal.Ordinal == 0 {
1504 goal.Ordinal = len(s.Meta.GoalHistory) + 1
1505 }
1506 if goal.Owner == "" {
1507 goal.Owner = GoalOwnerUser
1508 }
1509 if goal.Status == GoalActive {
1510 mission.ActiveGoalID = goal.ID
1511 mission.Status = MissionActive
1512 mission.Reason = ""
1513 } else if mission.ActiveGoalID == goal.ID {
1514 mission.ActiveGoalID = ""
1515 switch goal.Status {
1516 case GoalDone:
1517 mission.Status = MissionCompleted
1518 mission.Reason = ""
1519 case GoalPaused:
1520 mission.Status = MissionPaused
1521 mission.Reason = ""
1522 case GoalBlocked, GoalBudgetLimited, GoalStalled:
1523 mission.Status = MissionBlocked
1524 mission.Reason = goal.Reason
1525 }
1526 }
1527 return nil
1528}
1529
1530func normalizeSessionGoalMeta(meta *SessionMeta) {
1531 if meta == nil || meta.Goal == nil {
1532 return
1533 }
1534 goal := meta.Goal
1535 if goal.Owner == "" {
1536 goal.Owner = GoalOwnerUser
1537 }
1538 if meta.Mission == nil {
1539 meta.Mission = &SessionMission{
1540 ID: "legacy-" + meta.ID,
1541 Objective: goal.Objective,
1542 Status: MissionActive,
1543 Source: MissionSourceUser,
1544 }
1545 }
1546 if goal.ID == "" {
1547 goal.ID = "legacy-goal-" + meta.ID
1548 }
1549 if goal.MissionID == "" {
1550 goal.MissionID = meta.Mission.ID
1551 }
1552 if goal.Ordinal == 0 {
1553 goal.Ordinal = len(meta.GoalHistory) + 1
1554 }
1555 if goal.Status == GoalActive {
1556 meta.Mission.ActiveGoalID = goal.ID
1557 }
1558}
1559
1560func cloneSessionMission(mission *SessionMission) *SessionMission {
1561 if mission == nil {
1562 return nil
1563 }
1564 clone := *mission
1565 return &clone
1566}
1567
1568// UpdateTitle records a session title without adding anything to the
1569// conversation transcript. The title is kept in memory as well so the live
1570// session can restore it when the user switches sessions in the TUI.
1571func (s *Session) UpdateTitle(title string) error {
1572 if s == nil {
1573 return nil
1574 }
1575 title = NormalizeSessionTitle(title)
1576 if title == "" {
1577 return nil
1578 }
1579 if err := s.writeLine(sessionLine{Type: "rename", Title: title, Generated: true}); err != nil {
1580 return fmt.Errorf("update session title: %w", err)
1581 }
1582 s.Title = title
1583 return nil
1584}
1585
1586// RecordRetryLifecycle buffers a sanitized retry record for the next usage
1587// checkpoint. Invalid enum values are discarded or reduced to "unknown" so
1588// callers cannot use this metadata path to persist provider text.
1589func (s *Session) RecordRetryLifecycle(record RetryLifecycleRecord) {
1590 if s == nil {
1591 return
1592 }
1593 record, ok := sanitizeRetryLifecycleRecord(record)
1594 if !ok {
1595 return
1596 }
1597 s.retryMu.Lock()
1598 s.pendingRetryLifecycle = append(s.pendingRetryLifecycle, record)
1599 s.retryMu.Unlock()
1600}
1601
1602// AppendUsage writes a usage row to the session and includes any retry
1603// lifecycle records accumulated since the previous usage checkpoint.
1604func (s *Session) AppendUsage(u, cum provider.Usage) error {
1605 if s == nil {
1606 return nil
1607 }
1608 return s.writeUsageCheckpoint(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum})
1609}
1610
1611// FlushRetryLifecycle writes a usage checkpoint only when retry records are
1612// pending. Long-running hosts use it during shutdown so a terminal failure
1613// without a usage event is still diagnosable.
1614func (s *Session) FlushRetryLifecycle(u, cum provider.Usage) error {
1615 if s == nil {
1616 return nil
1617 }
1618 s.retryMu.Lock()
1619 defer s.retryMu.Unlock()
1620 if len(s.pendingRetryLifecycle) == 0 {
1621 return nil
1622 }
1623 return s.writeUsageCheckpointLocked(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum})
1624}
1625
1626func (s *Session) writeUsageCheckpoint(row sessionLine) error {
1627 s.retryMu.Lock()
1628 defer s.retryMu.Unlock()
1629 return s.writeUsageCheckpointLocked(row)
1630}
1631
1632func (s *Session) writeUsageCheckpointLocked(row sessionLine) error {
1633 if len(s.pendingRetryLifecycle) > 0 {
1634 row.RetryLifecycle = append([]RetryLifecycleRecord(nil), s.pendingRetryLifecycle...)
1635 }
1636 if row.Type == "usage" && len(row.RetryLifecycle) == 0 && row.Usage != nil && row.Cumulative != nil &&
1637 s.lastWriteWasUsage && *row.Usage == s.lastUsage && *row.Cumulative == s.lastCumulativeUsage {
1638 return nil
1639 }
1640 if err := s.writeLine(row); err != nil {
1641 return err
1642 }
1643 if row.Type == "usage" && row.Usage != nil && row.Cumulative != nil {
1644 s.lastWriteWasUsage = true
1645 s.lastUsage = *row.Usage
1646 s.lastCumulativeUsage = *row.Cumulative
1647 }
1648 s.pendingRetryLifecycle = nil
1649 return nil
1650}
1651
1652func sanitizeRetryLifecycleRecord(record RetryLifecycleRecord) (RetryLifecycleRecord, bool) {
1653 switch record.Event {
1654 case RetryLifecycleRequestFailed:
1655 record.DelayMS = 0
1656 case RetryLifecycleRetryScheduled:
1657 record.Terminal = false
1658 if record.DelayMS < 0 {
1659 record.DelayMS = 0
1660 }
1661 default:
1662 return RetryLifecycleRecord{}, false
1663 }
1664 switch record.Scope {
1665 case RetryScopeProvider, RetryScopeAgent:
1666 default:
1667 return RetryLifecycleRecord{}, false
1668 }
1669 if record.Attempt < 1 || record.MaxAttempts < record.Attempt {
1670 return RetryLifecycleRecord{}, false
1671 }
1672 switch record.Reason {
1673 case RetryReasonOverload, RetryReasonRateLimit, RetryReasonQuota,
1674 RetryReasonServer, RetryReasonNetwork, RetryReasonTimeout,
1675 RetryReasonAuth, RetryReasonContextWindow, RetryReasonClient,
1676 RetryReasonUnknown:
1677 default:
1678 record.Reason = RetryReasonUnknown
1679 }
1680 return record, true
1681}
1682
1683const maxExtensionStateBytes = 256 * 1024
1684
1685// AppendExtensionState records the latest opaque extension snapshot for the
1686// active session/branch. State is persisted in the session file but is never
1687// included in provider requests. A nil or JSON null state clears the key.
1688func (s *Session) AppendExtensionState(extension string, state json.RawMessage) error {
1689 if s == nil {
1690 return nil
1691 }
1692 extension = strings.TrimSpace(extension)
1693 if extension == "" {
1694 return fmt.Errorf("extension state: extension name is empty")
1695 }
1696 if len(state) > maxExtensionStateBytes {
1697 return fmt.Errorf("extension state %q exceeds %d bytes", extension, maxExtensionStateBytes)
1698 }
1699 if len(state) > 0 && !json.Valid(state) {
1700 return fmt.Errorf("extension state %q is not valid JSON", extension)
1701 }
1702 copyState := append(json.RawMessage(nil), state...)
1703 if err := s.writeLine(sessionLine{Type: "extension_state", Extension: extension, State: copyState}); err != nil {
1704 return fmt.Errorf("append extension state: %w", err)
1705 }
1706 if s.ExtensionState == nil {
1707 s.ExtensionState = map[string]json.RawMessage{}
1708 }
1709 if len(copyState) == 0 || strings.TrimSpace(string(copyState)) == "null" {
1710 delete(s.ExtensionState, extension)
1711 } else {
1712 s.ExtensionState[extension] = copyState
1713 }
1714 return nil
1715}
1716
1717// Flush writes buffered session data to the append handle.
1718func (s *Session) Flush() error {
1719 if s == nil {
1720 return nil
1721 }
1722 return s.buf.Flush()
1723}
1724
1725// Close flushes and closes the session file. If the session was
1726// freshly created in this process and never had any messages
1727// appended (the user opened zut, looked around, and exited without
1728// prompting), the file is deleted on close so the sessions list
1729// doesn't fill up with empty meta-only stubs.
1730func (s *Session) Close() error {
1731 if s == nil {
1732 return nil
1733 }
1734 flushErr := s.Flush()
1735 closeErr := s.writer.Close()
1736 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 {
1737 // Best-effort cleanup. We deliberately don't propagate the
1738 // remove error: if it fails (file already gone, perms changed)
1739 // the worst case is one stale empty file in the listing.
1740 _ = os.Remove(s.Path)
1741 }
1742 if flushErr != nil {
1743 return flushErr
1744 }
1745 return closeErr
1746}
1747
1748func (s *Session) writeLine(row sessionLine) error {
1749 s.lastWriteWasUsage = false
1750 b, err := json.Marshal(row)
1751 if err != nil {
1752 return err
1753 }
1754 if _, err := s.buf.Write(b); err != nil {
1755 return err
1756 }
1757 if err := s.buf.WriteByte('\n'); err != nil {
1758 return err
1759 }
1760 return s.buf.Flush()
1761}
1762
1763// ---- content (de)serialization ----
1764//
1765// provider.Content is an interface; encoding/json drops type information.
1766// We persist messages by reading the raw "message" object back and
1767// rebuilding Content from discriminated fields.
1768
1769func hydrateCompactionUsage(lineBytes []byte) (*provider.Usage, error) {
1770 var row struct {
1771 Cumulative *provider.Usage `json:"cumulative"`
1772 }
1773 if err := json.Unmarshal(lineBytes, &row); err != nil {
1774 return nil, err
1775 }
1776 return row.Cumulative, nil
1777}
1778
1779func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
1780 var row struct {
1781 Messages json.RawMessage `json:"messages"`
1782 }
1783 if err := json.Unmarshal(lineBytes, &row); err != nil {
1784 return nil, err
1785 }
1786 if len(row.Messages) == 0 || bytes.Equal(bytes.TrimSpace(row.Messages), []byte("null")) {
1787 // Older writers omitted the field for an empty compaction and some
1788 // wrote it as null. Both represent a valid empty replacement.
1789 return []provider.Message{}, nil
1790 }
1791 var rawMessages []json.RawMessage
1792 if err := json.Unmarshal(row.Messages, &rawMessages); err != nil {
1793 return nil, fmt.Errorf("invalid messages: %w", err)
1794 }
1795 messages := make([]provider.Message, 0, len(rawMessages))
1796 for idx, raw := range rawMessages {
1797 msg, err := hydrateMessageObject(raw)
1798 if err != nil {
1799 return nil, fmt.Errorf("message %d: %w", idx, err)
1800 }
1801 messages = append(messages, msg)
1802 }
1803 return messages, nil
1804}
1805
1806func hydrateMessage(lineBytes []byte) (provider.Message, error) {
1807 var row struct {
1808 Message json.RawMessage `json:"message"`
1809 }
1810 if err := json.Unmarshal(lineBytes, &row); err != nil {
1811 return provider.Message{}, err
1812 }
1813 if len(row.Message) == 0 || bytes.Equal(bytes.TrimSpace(row.Message), []byte("null")) {
1814 return provider.Message{}, fmt.Errorf("message row has no message")
1815 }
1816 return hydrateMessageObject(row.Message)
1817}
1818
1819func hydrateMessageObject(rawMessage []byte) (provider.Message, error) {
1820 var row struct {
1821 Role provider.Role `json:"role"`
1822 Content []json.RawMessage `json:"content"`
1823 Time time.Time `json:"time"`
1824 Meta map[string]string `json:"meta,omitempty"`
1825 AddedToolNames []string `json:"added_tool_names,omitempty"`
1826 }
1827 if err := json.Unmarshal(rawMessage, &row); err != nil {
1828 return provider.Message{}, err
1829 }
1830 if row.Role != provider.RoleUser && row.Role != provider.RoleAssistant && row.Role != provider.RoleTool && row.Role != provider.RoleDeveloper {
1831 return provider.Message{}, fmt.Errorf("message has invalid role %q", row.Role)
1832 }
1833 if len(row.Content) == 0 {
1834 return provider.Message{}, fmt.Errorf("message has no content")
1835 }
1836 msg := provider.Message{Role: row.Role, Time: row.Time, Meta: row.Meta, AddedToolNames: row.AddedToolNames}
1837 for idx, raw := range row.Content {
1838 if len(bytes.TrimSpace(raw)) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
1839 return provider.Message{}, fmt.Errorf("content block %d is empty", idx)
1840 }
1841 var head struct {
1842 Text string `json:"text"`
1843 MimeType string `json:"mime_type"`
1844 Data []byte `json:"data"`
1845 ID string `json:"id"`
1846 Name string `json:"name"`
1847 CallID string `json:"call_id"`
1848 ReasoningID string `json:"reasoning_id"`
1849 Summary string `json:"summary"`
1850 Encrypted string `json:"encrypted_content"`
1851 ThoughtSignature string `json:"thought_signature"`
1852 // ToolCallBlock also has Arguments, ToolResultBlock has Content + IsError
1853 }
1854 if err := json.Unmarshal(raw, &head); err != nil {
1855 return provider.Message{}, fmt.Errorf("content block %d: %w", idx, err)
1856 }
1857 var fields map[string]json.RawMessage
1858 if err := json.Unmarshal(raw, &fields); err != nil || fields == nil {
1859 if err == nil {
1860 err = fmt.Errorf("content block is not an object")
1861 }
1862 return provider.Message{}, fmt.Errorf("content block %d: %w", idx, err)
1863 }
1864 // Discriminate by presence of fields.
1865 switch {
1866 case head.ReasoningID != "" || head.Summary != "" || head.Encrypted != "":
1867 msg.Content = append(msg.Content, provider.ReasoningBlock{
1868 ID: head.ReasoningID,
1869 Summary: head.Summary,
1870 Encrypted: head.Encrypted,
1871 })
1872 case head.Name != "" || head.ID != "" || fields["arguments"] != nil:
1873 if head.Name == "" || head.ID == "" {
1874 return provider.Message{}, fmt.Errorf("content block %d: incomplete tool call", idx)
1875 }
1876 var tc struct {
1877 ID string `json:"id"`
1878 Name string `json:"name"`
1879 Arguments json.RawMessage `json:"arguments"`
1880 ThoughtSignature string `json:"thought_signature"`
1881 }
1882 _ = json.Unmarshal(raw, &tc)
1883 msg.Content = append(msg.Content, provider.ToolCallBlock{
1884 ID: tc.ID,
1885 Name: tc.Name,
1886 Arguments: tc.Arguments,
1887 ThoughtSignature: tc.ThoughtSignature,
1888 })
1889 case head.CallID != "":
1890 var tr struct {
1891 CallID string `json:"call_id"`
1892 Content []json.RawMessage `json:"content"`
1893 IsError bool `json:"is_error"`
1894 Timing *provider.ToolTiming `json:"timing,omitempty"`
1895 }
1896 if err := json.Unmarshal(raw, &tr); err != nil || tr.Content == nil {
1897 if err == nil {
1898 err = fmt.Errorf("tool result has no content")
1899 }
1900 return provider.Message{}, fmt.Errorf("content block %d: %w", idx, err)
1901 }
1902 block := provider.ToolResultBlock{CallID: tr.CallID, IsError: tr.IsError, Timing: tr.Timing}
1903 for innerIdx, c := range tr.Content {
1904 var inner struct {
1905 Text string `json:"text"`
1906 MimeType string `json:"mime_type"`
1907 Data []byte `json:"data"`
1908 }
1909 if err := json.Unmarshal(c, &inner); err != nil {
1910 return provider.Message{}, fmt.Errorf("content block %d result %d: %w", idx, innerIdx, err)
1911 }
1912 var innerFields map[string]json.RawMessage
1913 if err := json.Unmarshal(c, &innerFields); err != nil || innerFields == nil {
1914 if err == nil {
1915 err = fmt.Errorf("result content is not an object")
1916 }
1917 return provider.Message{}, fmt.Errorf("content block %d result %d: %w", idx, innerIdx, err)
1918 }
1919 if inner.MimeType != "" {
1920 block.Content = append(block.Content, provider.ImageBlock{MimeType: inner.MimeType, Data: inner.Data})
1921 } else if _, ok := innerFields["text"]; ok {
1922 block.Content = append(block.Content, provider.TextBlock{Text: inner.Text})
1923 } else {
1924 return provider.Message{}, fmt.Errorf("content block %d result %d: unknown content", idx, innerIdx)
1925 }
1926 }
1927 if len(block.Content) == 0 {
1928 return provider.Message{}, fmt.Errorf("content block %d tool result has no content", idx)
1929 }
1930 msg.Content = append(msg.Content, block)
1931 case head.MimeType != "":
1932 msg.Content = append(msg.Content, provider.ImageBlock{
1933 MimeType: head.MimeType,
1934 Data: head.Data,
1935 ThoughtSignature: head.ThoughtSignature,
1936 })
1937 default:
1938 if _, ok := fields["text"]; !ok {
1939 return provider.Message{}, fmt.Errorf("content block %d: unknown content", idx)
1940 }
1941 msg.Content = append(msg.Content, provider.TextBlock{
1942 Text: head.Text,
1943 ThoughtSignature: head.ThoughtSignature,
1944 })
1945 }
1946 }
1947 return msg, nil
1948}
1949
1950// DecodeMessageJSON restores a provider-neutral message whose Content blocks
1951// were encoded through json.RawMessage. It is shared by durable session-like
1952// stores; callers must treat malformed content as persistence corruption.
1953func DecodeMessageJSON(rawMessage []byte) (provider.Message, error) {
1954 return hydrateMessageObject(rawMessage)
1955}