| 1 | package transcript |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "reflect" |
| 9 | |
| 10 | "reasonix/internal/eventwire" |
| 11 | "reasonix/internal/fileutil" |
| 12 | "reasonix/internal/store" |
| 13 | ) |
| 14 | |
| 15 | // Checkpoint is the durable display state, separate from provider messages. |
| 16 | // Its digest binds it to the terminal transcript that produced its coverage. |
| 17 | type Checkpoint struct { |
| 18 | Version int `json:"version"` |
| 19 | Identity Identity `json:"identity"` |
| 20 | CoveredThroughSeq uint64 `json:"coveredThroughSeq"` |
| 21 | TranscriptDigest string `json:"transcriptDigest"` |
| 22 | ProviderCount int `json:"providerCount"` |
| 23 | Records []Message `json:"records"` |
| 24 | Runtime Runtime `json:"runtime"` |
| 25 | ActiveAttempts []ActiveAttempt `json:"activeAttempts"` |
| 26 | Completion *eventwire.CompletionSummary `json:"completion,omitempty"` |
| 27 | } |
| 28 | |
| 29 | func (p *Projection) Checkpoint(digest string) (Checkpoint, error) { |
| 30 | p.mu.Lock() |
| 31 | defer p.mu.Unlock() |
| 32 | runtime, attempts := p.runtimeLocked() |
| 33 | state := Checkpoint{Version: ProtocolVersion, Identity: p.identity, CoveredThroughSeq: p.covered, |
| 34 | TranscriptDigest: digest, Records: p.buffer.Messages(), Runtime: runtime, ActiveAttempts: attempts, Completion: p.buffer.completion} |
| 35 | // Detach mutable metadata while sharing immutable strings. Encoding and |
| 36 | // decoding the full transcript here duplicates large bodies under p.mu; |
| 37 | // SaveCheckpoint already owns the required encoding outside that lock. |
| 38 | owned := mapContentStrings(reflect.ValueOf(state), nil, func(text string, _ []string) string { return text }).Interface().(Checkpoint) |
| 39 | return owned, nil |
| 40 | } |
| 41 | |
| 42 | func RestoreCheckpoint(state Checkpoint, identity Identity) (*Projection, error) { |
| 43 | if state.Version != ProtocolVersion || state.Identity.SessionID != identity.SessionID || |
| 44 | state.Identity.HeadID != identity.HeadID || state.Identity.RewriteEpoch != identity.RewriteEpoch { |
| 45 | return nil, errors.New("transcript checkpoint identity mismatch") |
| 46 | } |
| 47 | records := repairCheckpointRecordIdentities(state.Records, state.CoveredThroughSeq) |
| 48 | p, err := NewProjection(identity, records, state.CoveredThroughSeq) |
| 49 | if err != nil { |
| 50 | return nil, err |
| 51 | } |
| 52 | b, err := json.Marshal(state) |
| 53 | if err != nil { |
| 54 | return nil, err |
| 55 | } |
| 56 | var owned Checkpoint |
| 57 | if err = json.Unmarshal(b, &owned); err != nil { |
| 58 | return nil, err |
| 59 | } |
| 60 | p.runtime = owned.Runtime |
| 61 | if p.runtime.StartedAt > 0 { |
| 62 | p.startedTurnID = p.runtime.TurnID |
| 63 | } |
| 64 | for _, attempt := range owned.ActiveAttempts { |
| 65 | p.attempts[attempt.ID] = attempt |
| 66 | } |
| 67 | for _, prompt := range owned.Runtime.PendingEvents { |
| 68 | id := prompt.PromptID |
| 69 | if id != "" { |
| 70 | p.prompts[id] = prompt |
| 71 | } |
| 72 | } |
| 73 | p.buffer.completion = owned.Completion |
| 74 | return p, nil |
| 75 | } |
| 76 | |
| 77 | // Older builds could persist display-only rows without an identity when a |
| 78 | // frame was published outside the active turn. Repair only that legacy shape; |
| 79 | // non-empty duplicate identities remain corruption and are rejected by |
| 80 | // NewProjection. The generated value is deterministic for this checkpoint so |
| 81 | // repeated recovery cannot reshuffle mounted rows. |
| 82 | func repairCheckpointRecordIdentities(records []Message, covered uint64) []Message { |
| 83 | repaired := append([]Message(nil), records...) |
| 84 | used := make(map[string]bool, len(repaired)) |
| 85 | for _, record := range repaired { |
| 86 | if record.RecordID != "" { |
| 87 | used[record.RecordID] = true |
| 88 | } |
| 89 | } |
| 90 | for index := range repaired { |
| 91 | if repaired[index].RecordID != "" { |
| 92 | continue |
| 93 | } |
| 94 | switch { |
| 95 | case repaired[index].Role == "tool" && repaired[index].ToolCallID != "": |
| 96 | repaired[index].RecordID = "tool:" + repaired[index].ToolCallID |
| 97 | case repaired[index].MessageID != "": |
| 98 | repaired[index].RecordID = "m:" + repaired[index].MessageID |
| 99 | default: |
| 100 | base := fmt.Sprintf("view:checkpoint:%d:%d", covered, index) |
| 101 | repaired[index].RecordID = base |
| 102 | for suffix := 1; used[repaired[index].RecordID]; suffix++ { |
| 103 | repaired[index].RecordID = fmt.Sprintf("%s:%d", base, suffix) |
| 104 | } |
| 105 | } |
| 106 | used[repaired[index].RecordID] = true |
| 107 | } |
| 108 | return repaired |
| 109 | } |
| 110 | |
| 111 | func SaveCheckpoint(sessionPath string, state Checkpoint) error { |
| 112 | path := store.SessionTranscriptProjection(sessionPath) |
| 113 | if path == "" { |
| 114 | return nil |
| 115 | } |
| 116 | b, err := json.Marshal(state) |
| 117 | if err != nil { |
| 118 | return err |
| 119 | } |
| 120 | return fileutil.AtomicWriteFile(path, b, 0o600) |
| 121 | } |
| 122 | |
| 123 | func LoadCheckpoint(sessionPath string) (Checkpoint, bool, error) { |
| 124 | path := store.SessionTranscriptProjection(sessionPath) |
| 125 | if path == "" { |
| 126 | return Checkpoint{}, false, nil |
| 127 | } |
| 128 | b, err := os.ReadFile(path) |
| 129 | if errors.Is(err, os.ErrNotExist) { |
| 130 | return Checkpoint{}, false, nil |
| 131 | } |
| 132 | if err != nil { |
| 133 | return Checkpoint{}, false, err |
| 134 | } |
| 135 | var state Checkpoint |
| 136 | if err = json.Unmarshal(b, &state); err != nil { |
| 137 | return Checkpoint{}, false, err |
| 138 | } |
| 139 | if state.Version != ProtocolVersion { |
| 140 | return Checkpoint{}, false, errors.New("unsupported transcript checkpoint version") |
| 141 | } |
| 142 | return state, true, nil |
| 143 | } |
| 144 |