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