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