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