| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "os" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/provider" |
| 13 | "reasonix/internal/store" |
| 14 | ) |
| 15 | |
| 16 | // sessionLoadResult is one transcript read from disk: the messages of the |
| 17 | // selected head (or the whole schema-1 log), per-message times where the |
| 18 | // source records them, and the head position for schema-2 sessions. |
| 19 | type sessionLoadResult struct { |
| 20 | msgs []provider.Message |
| 21 | times []time.Time |
| 22 | fromEvents bool |
| 23 | damaged bool |
| 24 | dag bool |
| 25 | head HeadRef |
| 26 | headCount int |
| 27 | state *sessionDAGState |
| 28 | openTurn *sessionDAGTurn |
| 29 | events []HeadEvent |
| 30 | } |
| 31 | |
| 32 | // loadSessionMessages returns the session transcript, preferring the event log |
| 33 | // when the native layer owns it and it holds at least one decodable record. |
| 34 | // Foreign files squatting the log path (legacy import leftovers) are ignored |
| 35 | // in favor of the .jsonl checkpoint. damaged reports that a native log could |
| 36 | // not be replayed to its end (torn tail or corrupt record); callers that write |
| 37 | // should rewrite-and-compact to heal it. |
| 38 | func loadSessionMessages(sessionPath string) (msgs []provider.Message, fromEvents, damaged bool, err error) { |
| 39 | return loadSessionMessagesWithLimits(sessionPath, defaultSessionReplayLimits, nil) |
| 40 | } |
| 41 | |
| 42 | func loadSessionMessagesWithLimits(sessionPath string, limits sessionReplayLimits, hasher *sessionTranscriptHasher) (msgs []provider.Message, fromEvents, damaged bool, err error) { |
| 43 | return loadSessionMessagesWithContext(context.Background(), sessionPath, limits, hasher) |
| 44 | } |
| 45 | |
| 46 | func loadSessionMessagesWithContext(ctx context.Context, sessionPath string, limits sessionReplayLimits, hasher *sessionTranscriptHasher) (msgs []provider.Message, fromEvents, damaged bool, err error) { |
| 47 | res, err := loadSessionTranscript(ctx, sessionPath, limits, hasher) |
| 48 | return res.msgs, res.fromEvents, res.damaged, err |
| 49 | } |
| 50 | |
| 51 | // loadSessionTranscript dispatches on the log schema: a schema-2 log replays |
| 52 | // the DAG and materializes its selected head, a schema-1 log replays its |
| 53 | // records, and anything else falls back to the .jsonl checkpoint. |
| 54 | func loadSessionTranscript(ctx context.Context, sessionPath string, limits sessionReplayLimits, hasher *sessionTranscriptHasher) (sessionLoadResult, error) { |
| 55 | if err := ctx.Err(); err != nil { |
| 56 | return sessionLoadResult{}, err |
| 57 | } |
| 58 | probe, err := probeSessionEventLogWithLimits(sessionPath, limits) |
| 59 | if err != nil { |
| 60 | return sessionLoadResult{}, err |
| 61 | } |
| 62 | if probe.futureSchema { |
| 63 | return sessionLoadResult{fromEvents: true}, fmt.Errorf("session event log for %s uses schema %d; this build supports up to %d", sessionPath, probe.schemaVersion, sessionDAGSchemaVersion) |
| 64 | } |
| 65 | if probe.dag { |
| 66 | st, err := replaySessionDAG(ctx, store.SessionEventLog(sessionPath), limits) |
| 67 | if err != nil { |
| 68 | return sessionLoadResult{fromEvents: true, dag: true}, err |
| 69 | } |
| 70 | headID := st.selectedHead() |
| 71 | msgs, times := st.materialize(headID) |
| 72 | hasher.addAll(msgs) |
| 73 | return sessionLoadResult{ |
| 74 | msgs: msgs, times: times, fromEvents: true, damaged: st.damaged || st.holes != 0, dag: true, |
| 75 | head: HeadRef{HeadID: headID, LeafID: st.heads[headID].leaf, LogGeneration: st.generation, LogOffset: st.lastGoodEnd}, |
| 76 | headCount: len(st.heads), |
| 77 | state: st, |
| 78 | openTurn: st.heads[headID].openTurn, |
| 79 | events: loadHeadEvents(st, headID), |
| 80 | }, nil |
| 81 | } |
| 82 | if probe.native && probe.size > 0 { |
| 83 | replay, replayErr := replaySessionEventLogWithContext(ctx, store.SessionEventLog(sessionPath), limits, hasher) |
| 84 | if replayErr != nil { |
| 85 | return sessionLoadResult{fromEvents: true}, replayErr |
| 86 | } |
| 87 | if replay.records > 0 { |
| 88 | return sessionLoadResult{msgs: replay.msgs, times: replay.times, fromEvents: true, damaged: replay.damaged}, nil |
| 89 | } |
| 90 | // Defensive: the probe saw a native head but nothing replayed; fall |
| 91 | // back to the checkpoint and let the next save rebuild the log. |
| 92 | msgs, err := loadSessionMessagesFromJSONLContext(ctx, sessionPath, hasher) |
| 93 | return sessionLoadResult{msgs: msgs, damaged: true}, err |
| 94 | } |
| 95 | msgs, err := loadSessionMessagesFromJSONLContext(ctx, sessionPath, hasher) |
| 96 | return sessionLoadResult{msgs: msgs}, err |
| 97 | } |
| 98 | |
| 99 | func loadSessionMessagesFromJSONL(path string, hasher *sessionTranscriptHasher) ([]provider.Message, error) { |
| 100 | return loadSessionMessagesFromJSONLContext(context.Background(), path, hasher) |
| 101 | } |
| 102 | |
| 103 | func loadSessionMessagesFromJSONLContext(ctx context.Context, path string, hasher *sessionTranscriptHasher) ([]provider.Message, error) { |
| 104 | f, err := os.Open(path) |
| 105 | if err != nil { |
| 106 | return nil, err |
| 107 | } |
| 108 | defer f.Close() |
| 109 | |
| 110 | var msgs []provider.Message |
| 111 | dec := json.NewDecoder(&contextReader{ctx: ctx, reader: f}) |
| 112 | for { |
| 113 | if err := ctx.Err(); err != nil { |
| 114 | return nil, err |
| 115 | } |
| 116 | var m provider.Message |
| 117 | if err := dec.Decode(&m); err != nil { |
| 118 | if errors.Is(err, io.EOF) { |
| 119 | break |
| 120 | } |
| 121 | return nil, fmt.Errorf("decode %s: %w", path, err) |
| 122 | } |
| 123 | msgs = append(msgs, hasher.add(m)) |
| 124 | } |
| 125 | return msgs, nil |
| 126 | } |
| 127 |