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