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