1package subagents
2
3import (
4 "bytes"
5 "crypto/sha256"
6 "encoding/base64"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "io"
11 "os"
12 "path/filepath"
13 "reflect"
14 "strings"
15 "time"
16
17 "github.com/bnema/zut/packages/core"
18 "github.com/bnema/zut/packages/provider"
19)
20
21const (
22 residentHistoryDefaultLimit = 200
23 residentHistoryMaximumLimit = 1000
24 residentHistoryPageBytes = residentMaxRecordBytes + 1 // largest JSON record plus newline
25 residentHistoryReadChunk = 64 << 10
26)
27
28// ResidentHistoryItem is a provider-neutral finalized transcript item. The
29// raw message and tool payloads are preserved so callers can render complete
30// arguments/results without relying on a lossy text projection.
31type ResidentHistoryItem struct {
32 Type string `json:"type"`
33 Time time.Time `json:"time"`
34 Message json.RawMessage `json:"message,omitempty"`
35 // Messages carries a compaction checkpoint: the complete replacement
36 // transcript of a child that recovered from a context overflow. It is
37 // produced only for residentRecordCompacted items, which the UI pager does
38 // not expose.
39 Messages []json.RawMessage `json:"messages,omitempty"`
40 ToolID string `json:"tool_id,omitempty"`
41 ToolName string `json:"tool_name,omitempty"`
42 ToolArgs json.RawMessage `json:"tool_args,omitempty"`
43 ToolResult json.RawMessage `json:"tool_result,omitempty"`
44}
45
46// ResidentHistoryPage is a recent-first, bounded page of finalized history.
47// OlderCursor is opaque and can be supplied to ReadResidentHistoryPage to load
48// the adjacent older page. An empty cursor means this page reaches the oldest
49// finalized content.
50type ResidentHistoryPage struct {
51 Items []ResidentHistoryItem
52 OlderCursor string
53}
54
55type residentHistoryCursor struct {
56 Generation string `json:"generation"`
57 Offset int64 `json:"offset"`
58 Size int64 `json:"size"`
59}
60
61type residentHistoryEntry struct {
62 item ResidentHistoryItem
63 offset int64
64}
65
66// ReadResidentHistoryPage reads a bounded suffix of one resident journal. It
67// never follows arbitrary paths at public manager boundaries; callers outside
68// this package should use ResidentManager.HistoryPage with a child ID.
69//
70// A page starts at a user or assistant record. That preserves the adjacent
71// assistant/tool-call/tool-result group when paging backward. Incomplete final
72// lines are deliberately deferred until the writer completes them.
73func ReadResidentHistoryPage(dir, olderCursor string, limit int) (ResidentHistoryPage, error) {
74 if limit <= 0 {
75 limit = residentHistoryDefaultLimit
76 }
77 if limit > residentHistoryMaximumLimit {
78 limit = residentHistoryMaximumLimit
79 }
80 f, err := os.Open(filepath.Join(dir, residentTranscriptName))
81 if err != nil {
82 return ResidentHistoryPage{}, err
83 }
84 defer f.Close()
85 info, err := f.Stat()
86 if err != nil {
87 return ResidentHistoryPage{}, err
88 }
89 generation, err := residentHistoryGeneration(f, info, info.Size())
90 if err != nil {
91 return ResidentHistoryPage{}, err
92 }
93 end := info.Size()
94 if olderCursor != "" {
95 cursor, err := decodeResidentHistoryCursor(olderCursor)
96 if err != nil {
97 return ResidentHistoryPage{}, err
98 }
99 cursorGeneration, generationErr := residentHistoryGeneration(f, info, cursor.Size)
100 if generationErr != nil {
101 return ResidentHistoryPage{}, generationErr
102 }
103 if cursor.Generation != cursorGeneration || cursor.Size > end || cursor.Offset <= 0 || cursor.Offset > end {
104 return ResidentHistoryPage{}, errors.New("resident history: cursor is no longer valid")
105 }
106 end = cursor.Offset
107 }
108 return readResidentHistorySuffix(f, end, limit, generation, info.Size())
109}
110
111func residentHistoryGeneration(f *os.File, info os.FileInfo, originalSize int64) (string, error) {
112 prefixSize := originalSize
113 if prefixSize > 4096 {
114 prefixSize = 4096
115 }
116 prefix := make([]byte, prefixSize)
117 if prefixSize > 0 {
118 read, err := f.ReadAt(prefix, 0)
119 if err != nil && !errors.Is(err, io.EOF) {
120 return "", fmt.Errorf("resident history: read journal generation: %w", err)
121 }
122 prefix = prefix[:read]
123 }
124 sum := sha256.Sum256(prefix)
125 return fmt.Sprintf("%s:%x", residentHistoryFileIdentity(info), sum[:8]), nil
126}
127
128// residentHistoryFileIdentity uses only stable file identity fields when the
129// host exposes them. Dev/Ino cover Unix; creation time is the portable
130// fallback for Windows metadata. The prefix digest remains the corruption
131// check when a platform offers neither.
132func residentHistoryFileIdentity(info os.FileInfo) string {
133 value := reflect.ValueOf(info.Sys())
134 if value.Kind() == reflect.Pointer {
135 if value.IsNil() {
136 return ""
137 }
138 value = value.Elem()
139 }
140 if value.Kind() != reflect.Struct {
141 return ""
142 }
143 var parts []string
144 for _, field := range []string{"Dev", "Ino", "VolumeSerialNumber", "FileIndexHigh", "FileIndexLow", "CreationTime"} {
145 v := value.FieldByName(field)
146 if !v.IsValid() || !v.CanInterface() {
147 continue
148 }
149 parts = append(parts, field+"="+fmt.Sprint(v.Interface()))
150 }
151 return strings.Join(parts, ",")
152}
153
154func decodeResidentHistoryCursor(encoded string) (residentHistoryCursor, error) {
155 data, err := base64.RawURLEncoding.DecodeString(encoded)
156 if err != nil {
157 return residentHistoryCursor{}, errors.New("resident history: invalid cursor")
158 }
159 var cursor residentHistoryCursor
160 if err := json.Unmarshal(data, &cursor); err != nil || cursor.Generation == "" || cursor.Size <= 0 {
161 return residentHistoryCursor{}, errors.New("resident history: invalid cursor")
162 }
163 return cursor, nil
164}
165
166func encodeResidentHistoryCursor(generation string, offset, size int64) string {
167 data, _ := json.Marshal(residentHistoryCursor{Generation: generation, Offset: offset, Size: size})
168 return base64.RawURLEncoding.EncodeToString(data)
169}
170
171func readResidentHistorySuffix(f *os.File, end int64, limit int, generation string, size int64) (ResidentHistoryPage, error) {
172 if end == 0 {
173 return ResidentHistoryPage{}, nil
174 }
175 start := end
176 data := make([]byte, 0, residentHistoryReadChunk)
177 for {
178 readSize := int64(residentHistoryReadChunk)
179 if readSize > start {
180 readSize = start
181 }
182 if int64(len(data))+readSize > residentHistoryPageBytes {
183 return ResidentHistoryPage{}, errors.New("resident history: page byte budget exceeded")
184 }
185 start -= readSize
186 chunk := make([]byte, readSize)
187 read, err := f.ReadAt(chunk, start)
188 if err != nil && !errors.Is(err, io.EOF) {
189 return ResidentHistoryPage{}, fmt.Errorf("resident history: read journal suffix: %w", err)
190 }
191 chunk = chunk[:read]
192 data = append(chunk, data...)
193
194 entries, err := decodeResidentHistoryEntries(data, start, start == 0)
195 if err != nil {
196 return ResidentHistoryPage{}, err
197 }
198 first := residentHistoryPageStart(entries, limit)
199 if first >= 0 {
200 page := ResidentHistoryPage{Items: make([]ResidentHistoryItem, len(entries)-first)}
201 for index := first; index < len(entries); index++ {
202 page.Items[index-first] = cloneResidentHistoryItem(entries[index].item)
203 }
204 if start > 0 || first > 0 {
205 page.OlderCursor = encodeResidentHistoryCursor(generation, entries[first].offset, size)
206 }
207 return page, nil
208 }
209 if start == 0 {
210 if len(entries) == 0 {
211 return ResidentHistoryPage{}, nil
212 }
213 return ResidentHistoryPage{}, errors.New("resident history: tool group has no message boundary")
214 }
215 }
216}
217
218func decodeResidentHistoryEntries(data []byte, start int64, startsAtBeginning bool) ([]residentHistoryEntry, error) {
219 lastNewline := bytes.LastIndexByte(data, '\n')
220 if lastNewline < 0 {
221 return nil, nil
222 }
223 complete := data[:lastNewline+1]
224 base := start
225 if !startsAtBeginning {
226 firstNewline := bytes.IndexByte(complete, '\n')
227 if firstNewline < 0 {
228 return nil, nil
229 }
230 base += int64(firstNewline + 1)
231 complete = complete[firstNewline+1:]
232 }
233 entries := make([]residentHistoryEntry, 0)
234 for len(complete) > 0 {
235 newline := bytes.IndexByte(complete, '\n')
236 if newline < 0 {
237 break
238 }
239 line := complete[:newline]
240 lineOffset := base
241 base += int64(newline + 1)
242 complete = complete[newline+1:]
243 if len(line) > residentMaxRecordBytes {
244 return nil, errors.New("resident history: record too large")
245 }
246 var record residentRecord
247 if err := json.Unmarshal(line, &record); err != nil {
248 return nil, fmt.Errorf("resident history: decode journal record: %w", err)
249 }
250 item, include, err := residentHistoryItem(record)
251 if err != nil {
252 return nil, err
253 }
254 if include {
255 entries = append(entries, residentHistoryEntry{item: item, offset: lineOffset})
256 }
257 }
258 return entries, nil
259}
260
261func residentHistoryItem(record residentRecord) (ResidentHistoryItem, bool, error) {
262 item := ResidentHistoryItem{Type: record.Type, Time: record.Time}
263 switch record.Type {
264 case residentRecordUser, residentRecordAssistant:
265 if len(record.Message) == 0 || !json.Valid(record.Message) {
266 return ResidentHistoryItem{}, false, errors.New("resident history: malformed message record")
267 }
268 item.Message = append(json.RawMessage(nil), record.Message...)
269 case residentRecordToolCall:
270 if record.ToolID == "" || record.ToolName == "" || !json.Valid(record.ToolArgs) {
271 return ResidentHistoryItem{}, false, errors.New("resident history: malformed tool call record")
272 }
273 item.ToolID, item.ToolName = record.ToolID, record.ToolName
274 item.ToolArgs = append(json.RawMessage(nil), record.ToolArgs...)
275 case residentRecordToolResult:
276 if record.ToolID == "" || !json.Valid(record.ToolResult) {
277 return ResidentHistoryItem{}, false, errors.New("resident history: malformed tool result record")
278 }
279 item.ToolID, item.ToolResult = record.ToolID, append(json.RawMessage(nil), record.ToolResult...)
280 default:
281 // Lifecycle records and compaction checkpoints are not history items:
282 // the pager keeps showing the append-only log, while resume replay
283 // starts from the newest checkpoint (see ReadResidentTranscriptMessages).
284 return ResidentHistoryItem{}, false, nil
285 }
286 return item, true, nil
287}
288
289func residentHistoryPageStart(entries []residentHistoryEntry, limit int) int {
290 first, fallback := -1, -1
291 for index := range entries {
292 if entries[index].item.Type != residentRecordUser && entries[index].item.Type != residentRecordAssistant {
293 continue
294 }
295 fallback = index
296 if len(entries)-index <= limit {
297 first = index
298 }
299 }
300 if first < 0 {
301 return fallback
302 }
303 return first
304}
305
306func cloneResidentHistoryItem(item ResidentHistoryItem) ResidentHistoryItem {
307 item.Message = append(json.RawMessage(nil), item.Message...)
308 item.Messages = cloneRawMessages(item.Messages)
309 item.ToolArgs = append(json.RawMessage(nil), item.ToolArgs...)
310 item.ToolResult = append(json.RawMessage(nil), item.ToolResult...)
311 return item
312}
313
314func cloneRawMessages(messages []json.RawMessage) []json.RawMessage {
315 if messages == nil {
316 return nil
317 }
318 cloned := make([]json.RawMessage, len(messages))
319 for index, message := range messages {
320 cloned[index] = append(json.RawMessage(nil), message...)
321 }
322 return cloned
323}
324
325// decodeResidentCompactedMessages decodes one compaction checkpoint. The
326// checkpoint is the complete replacement transcript, so an empty or malformed
327// payload is corruption rather than an empty transcript.
328func decodeResidentCompactedMessages(encoded []json.RawMessage) ([]provider.Message, error) {
329 if len(encoded) == 0 {
330 return nil, errors.New("empty compaction checkpoint")
331 }
332 messages := make([]provider.Message, 0, len(encoded))
333 for _, raw := range encoded {
334 message, err := core.DecodeMessageJSON(raw)
335 if err != nil {
336 return nil, fmt.Errorf("decode compaction checkpoint message: %w", err)
337 }
338 messages = append(messages, message)
339 }
340 return messages, nil
341}
342
343// ResidentHistoryMessages converts finalized journal items into the same
344// provider-neutral message structures used by tui.View. Tool-call records
345// duplicate calls already embedded in an assistant message, so they are only
346// synthesized when the assistant record is absent (for recovery resilience).
347//
348// Items are one ordered log, not independent groups: a checkpoint item
349// replaces every message accumulated before it. A bounded history page never
350// contains one, because the pager excludes checkpoints.
351func ResidentHistoryMessages(items []ResidentHistoryItem) ([]provider.Message, error) {
352 messages := make([]provider.Message, 0, len(items))
353 calls := make(map[string]struct{})
354 for _, item := range items {
355 switch item.Type {
356 case residentRecordUser, residentRecordAssistant:
357 message, err := core.DecodeMessageJSON(item.Message)
358 if err != nil {
359 return nil, fmt.Errorf("resident history: decode message: %w", err)
360 }
361 for _, content := range message.Content {
362 if call, ok := content.(provider.ToolCallBlock); ok {
363 calls[call.ID] = struct{}{}
364 }
365 }
366 messages = append(messages, message)
367 case residentRecordCompacted:
368 // A checkpoint replaces the transcript accumulated so far: the
369 // resumed model must see the compacted view, not the full
370 // append-only log. Tool-call bookkeeping starts over from it so a
371 // call dropped by compaction is not re-synthesized below.
372 checkpoint, err := decodeResidentCompactedMessages(item.Messages)
373 if err != nil {
374 return nil, fmt.Errorf("resident history: %w", err)
375 }
376 messages = checkpoint
377 calls = make(map[string]struct{})
378 for _, message := range checkpoint {
379 for _, content := range message.Content {
380 if call, ok := content.(provider.ToolCallBlock); ok {
381 calls[call.ID] = struct{}{}
382 }
383 }
384 }
385 case residentRecordToolCall:
386 if _, exists := calls[item.ToolID]; exists {
387 continue
388 }
389 messages = append(messages, provider.Message{Role: provider.RoleAssistant, Time: item.Time, Content: []provider.Content{provider.ToolCallBlock{ID: item.ToolID, Name: item.ToolName, Arguments: append(json.RawMessage(nil), item.ToolArgs...)}}})
390 calls[item.ToolID] = struct{}{}
391 case residentRecordToolResult:
392 var result struct {
393 Content []json.RawMessage `json:"Content"`
394 IsError bool `json:"IsError"`
395 Timing *provider.ToolTiming `json:"Timing"`
396 Context provider.ToolContext `json:"Context"`
397 }
398 if err := json.Unmarshal(item.ToolResult, &result); err != nil || len(result.Content) == 0 {
399 if err == nil {
400 err = errors.New("missing content")
401 }
402 return nil, fmt.Errorf("resident history: decode tool result: %w", err)
403 }
404 raw, err := json.Marshal(struct {
405 Role provider.Role `json:"role"`
406 Time time.Time `json:"time"`
407 Content []struct {
408 CallID string `json:"call_id"`
409 Content []json.RawMessage `json:"content"`
410 IsError bool `json:"is_error"`
411 Timing *provider.ToolTiming `json:"timing,omitempty"`
412 Context provider.ToolContext `json:"context,omitempty"`
413 } `json:"content"`
414 }{Role: provider.RoleTool, Time: item.Time, Content: []struct {
415 CallID string `json:"call_id"`
416 Content []json.RawMessage `json:"content"`
417 IsError bool `json:"is_error"`
418 Timing *provider.ToolTiming `json:"timing,omitempty"`
419 Context provider.ToolContext `json:"context,omitempty"`
420 }{{CallID: item.ToolID, Content: result.Content, IsError: result.IsError, Timing: result.Timing, Context: result.Context}}})
421 if err != nil {
422 return nil, err
423 }
424 message, err := core.DecodeMessageJSON(raw)
425 if err != nil {
426 return nil, fmt.Errorf("resident history: decode tool result message: %w", err)
427 }
428 messages = append(messages, message)
429 }
430 }
431 return messages, nil
432}
433
434// ReadResidentTranscriptMessages rebuilds finalized conversation state for an
435// explicit resident resume. It reads the authoritative transcript, unlike the
436// bounded UI pager, and must therefore only be used at the resume boundary.
437func ReadResidentTranscriptMessages(dir string) ([]provider.Message, error) {
438 records, err := ReadResidentJournal(filepath.Join(dir, residentTranscriptName))
439 if err != nil {
440 return nil, err
441 }
442 items := make([]ResidentHistoryItem, 0, len(records))
443 for _, record := range records {
444 if record.Type == residentRecordCompacted {
445 // The pager hides checkpoints, but resume replay starts from one.
446 items = append(items, ResidentHistoryItem{Type: record.Type, Time: record.Time, Messages: cloneRawMessages(record.Messages)})
447 continue
448 }
449 item, include, err := residentHistoryItem(record)
450 if err != nil {
451 return nil, err
452 }
453 if include {
454 items = append(items, item)
455 }
456 }
457 return ResidentHistoryMessages(items)
458}
459
460// ReadResidentHistory returns the recent finalized journal items, bounded by
461// item count. Public callers provide a child directory obtained from manager
462// state, never an arbitrary user-controlled path.
463func ReadResidentHistory(dir string, limit int) ([]ResidentHistoryItem, error) {
464 page, err := ReadResidentHistoryPage(dir, "", limit)
465 if err != nil {
466 return nil, err
467 }
468 return page.Items, nil
469}
470
471// ResidentHistoryDir resolves a child ID beneath root without accepting path
472// traversal. It is kept separate from the reader so manager APIs can apply
473// their own visibility checks before exposing history.
474func ResidentHistoryDir(root, childID string) (string, error) {
475 childID = strings.TrimSpace(childID)
476 if !residentChildID(childID) {
477 return "", errors.New("resident history: invalid child ID")
478 }
479 dir := filepath.Join(root, childID)
480 if info, err := os.Stat(dir); err != nil {
481 return "", err
482 } else if !info.IsDir() {
483 return "", errors.New("resident history: child path is not a directory")
484 }
485 if !pathWithin(dir, root) {
486 return "", errors.New("resident history: child path escapes root")
487 }
488 return dir, nil
489}