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