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