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