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