| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "slices" |
| 7 | |
| 8 | "reasonix/internal/provider" |
| 9 | "reasonix/internal/transcript" |
| 10 | "reasonix/internal/turnevent" |
| 11 | ) |
| 12 | |
| 13 | // Transcript is owned by the runtime, not a replaceable controller. The |
| 14 | // Session acceptance lock serializes canonical batches and transient frames. |
| 15 | // No subscriber callback, disk read or network write runs under that lock. |
| 16 | func (r *Runtime) Transcript() *transcript.Projection { return r.transcript } |
| 17 | |
| 18 | func (r *Runtime) FollowTranscript(ctx context.Context, req transcript.FollowRequest) (transcript.FollowResponse, error) { |
| 19 | r.session.mu.Lock() |
| 20 | if r.session.binding != nil { |
| 21 | durable, _, _ := r.session.binding.progress() |
| 22 | r.transcript.SetDurableSequence(durable) |
| 23 | } |
| 24 | r.session.mu.Unlock() |
| 25 | return r.transcript.Follow(ctx, req) |
| 26 | } |
| 27 | |
| 28 | func (r *Runtime) PublishTranscriptFrame(envelope turnevent.Envelope) error { |
| 29 | s := r.session |
| 30 | s.mu.Lock() |
| 31 | defer s.mu.Unlock() |
| 32 | if s.binding != nil { |
| 33 | durable, _, _ := s.binding.progress() |
| 34 | r.transcript.SetDurableSequence(durable) |
| 35 | } |
| 36 | return r.transcript.ApplyFrame(envelope, s.next-1) |
| 37 | } |
| 38 | |
| 39 | func (s *Session) acceptTranscriptCommit(commit Commit) { |
| 40 | if s.transcript == nil { |
| 41 | return |
| 42 | } |
| 43 | var messages []provider.Message |
| 44 | var removed []string |
| 45 | rewrite := false |
| 46 | for _, e := range commit.Events { |
| 47 | switch e.Kind { |
| 48 | case "message/complete", "message/upsert": |
| 49 | var payload struct { |
| 50 | Message provider.Message `json:"message"` |
| 51 | } |
| 52 | if json.Unmarshal(e.Payload, &payload) == nil { |
| 53 | removed = slices.DeleteFunc(removed, func(id string) bool { return id == payload.Message.ID }) |
| 54 | messages = append(messages, payload.Message) |
| 55 | } |
| 56 | case "message/retract": |
| 57 | ids, err := retractedMessageIDs(e, e.Payload) |
| 58 | if err == nil { |
| 59 | removed = append(removed, ids...) |
| 60 | messages = slices.DeleteFunc(messages, func(message provider.Message) bool { return slices.Contains(ids, message.ID) }) |
| 61 | } |
| 62 | case "history/replace", "legacy/import": |
| 63 | var payload struct { |
| 64 | Messages []provider.Message `json:"messages"` |
| 65 | } |
| 66 | if json.Unmarshal(e.Payload, &payload) == nil { |
| 67 | messages, rewrite = payload.Messages, true |
| 68 | removed = nil |
| 69 | } |
| 70 | } |
| 71 | } |
| 72 | finalID := s.projection.CurrentTurnMessageID |
| 73 | if finalID == "" && len(s.projection.Turns) > 0 { |
| 74 | last := s.projection.Turns[len(s.projection.Turns)-1] |
| 75 | if last.TurnID == commit.TurnID || s.projection.TurnID == "" { |
| 76 | finalID = last.MessageID |
| 77 | } |
| 78 | } |
| 79 | rows := s.transcriptRows(messages) |
| 80 | if len(removed) > 0 && !rewrite { |
| 81 | s.transcript.AcceptRetractions(rows, removed, commit.LastSequence(), commit.TurnID, finalID) |
| 82 | } else { |
| 83 | s.transcript.AcceptBusiness(rows, commit.LastSequence(), commit.TurnID, rewrite, finalID) |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | // Caller holds s.mu; live commits and reopened sessions use the same identities. |
| 88 | func (s *Session) transcriptRows(messages []provider.Message) []transcript.Message { |
| 89 | rows := transcript.History(messages, transcript.HistoryOptions{}) |
| 90 | for i := range rows { |
| 91 | if rows[i].Role != "user" { |
| 92 | continue |
| 93 | } |
| 94 | if receipt, ok := s.projection.Submissions.byMessage[s.id+"\x00"+rows[i].MessageID]; ok { |
| 95 | rows[i].SubmissionID = receipt.SubmissionID |
| 96 | } |
| 97 | } |
| 98 | return rows |
| 99 | } |
| 100 |