| 1 | package transcript |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "fmt" |
| 6 | "reflect" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "testing" |
| 10 | "unicode/utf8" |
| 11 | |
| 12 | "reasonix/internal/billing" |
| 13 | "reasonix/internal/event" |
| 14 | "reasonix/internal/eventwire" |
| 15 | "reasonix/internal/turnevent" |
| 16 | ) |
| 17 | |
| 18 | func TestProjectionAttachesCompletedTurnUsageAndTimingToFinalAnswer(t *testing.T) { |
| 19 | p, err := NewProjection(testIdentity, nil, 0) |
| 20 | if err != nil { |
| 21 | t.Fatal(err) |
| 22 | } |
| 23 | apply := func(sequence uint64, kind string, createdAt int64, wire eventwire.Event, status event.TurnStatus) { |
| 24 | t.Helper() |
| 25 | wire.Kind = kind |
| 26 | if err := p.Apply(turnevent.Envelope{SessionID: "session", RuntimeEpoch: "runtime", TurnID: "turn", Sequence: sequence, Kind: kind, Status: status, CreatedAt: createdAt, Event: wire}); err != nil { |
| 27 | t.Fatal(err) |
| 28 | } |
| 29 | } |
| 30 | apply(1, "turn_started", 1_000, eventwire.Event{}, event.TurnInProgress) |
| 31 | apply(2, "message", 1_500, eventwire.Event{MessageID: "answer", Text: "done"}, event.TurnInProgress) |
| 32 | apply(3, "usage", 2_000, eventwire.Event{Usage: &eventwire.Usage{ |
| 33 | PromptTokens: 100_000, CompletionTokens: 2_000, TotalTokens: 102_000, |
| 34 | CacheHitTokens: 80_000, CacheMissTokens: 20_000, ReasoningTokens: 1_000, |
| 35 | CostQuote: &billing.CostQuote{ModelRef: "deepseek-official/deepseek-flash"}, |
| 36 | }}, event.TurnInProgress) |
| 37 | apply(4, "usage", 3_000, eventwire.Event{Usage: &eventwire.Usage{ |
| 38 | PromptTokens: 80_000, CompletionTokens: 3_225, TotalTokens: 83_225, |
| 39 | CacheHitTokens: 75_520, CacheMissTokens: 4_480, ReasoningTokens: 909, |
| 40 | CostQuote: &billing.CostQuote{ModelRef: "deepseek-official/deepseek-flash"}, |
| 41 | }}, event.TurnInProgress) |
| 42 | apply(5, "turn_done", 4_000, eventwire.Event{}, event.TurnCompleted) |
| 43 | |
| 44 | records := snapshot(t, p).Records |
| 45 | if len(records) != 1 { |
| 46 | t.Fatalf("records = %d, want 1", len(records)) |
| 47 | } |
| 48 | message := records[0].Message |
| 49 | if message.CreatedAt != 1_500 || message.TurnDurationMs != 3_000 { |
| 50 | t.Fatalf("message timing = created %d duration %d", message.CreatedAt, message.TurnDurationMs) |
| 51 | } |
| 52 | usage := message.TurnUsage |
| 53 | if usage == nil || usage.TotalTokens != 185_225 || usage.UncachedInputTokens != 24_480 || usage.OutputTokens != 5_225 { |
| 54 | t.Fatalf("turn usage = %+v", usage) |
| 55 | } |
| 56 | if usage.CacheReadTokens == nil || *usage.CacheReadTokens != 155_520 || usage.ReasoningTokens == nil || *usage.ReasoningTokens != 1_909 { |
| 57 | t.Fatalf("turn usage optional buckets = %+v", usage) |
| 58 | } |
| 59 | if !reflect.DeepEqual(usage.Routes, []string{"deepseek-official/deepseek-flash"}) { |
| 60 | t.Fatalf("routes = %v", usage.Routes) |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | func TestProjectionQueuedStatusDoesNotReusePreviousTurnStart(t *testing.T) { |
| 65 | p, err := NewProjection(testIdentity, nil, 0) |
| 66 | if err != nil { |
| 67 | t.Fatal(err) |
| 68 | } |
| 69 | apply := func(sequence uint64, turnID, kind string, createdAt int64, wire eventwire.Event, status event.TurnStatus) { |
| 70 | t.Helper() |
| 71 | wire.Kind = kind |
| 72 | if err := p.Apply(turnevent.Envelope{SessionID: "session", RuntimeEpoch: "runtime", TurnID: turnID, Sequence: sequence, Kind: kind, Status: status, CreatedAt: createdAt, Event: wire}); err != nil { |
| 73 | t.Fatal(err) |
| 74 | } |
| 75 | } |
| 76 | apply(1, "first", "turn_started", 1_000, eventwire.Event{}, event.TurnInProgress) |
| 77 | apply(2, "first", "message", 1_500, eventwire.Event{MessageID: "first-answer", Text: "first"}, event.TurnInProgress) |
| 78 | apply(3, "first", "turn_done", 2_000, eventwire.Event{}, event.TurnCompleted) |
| 79 | apply(4, "second", "turn_status", 100_000, eventwire.Event{}, event.TurnQueued) |
| 80 | apply(5, "second", "turn_started", 100_000, eventwire.Event{}, event.TurnInProgress) |
| 81 | apply(6, "second", "message", 101_000, eventwire.Event{MessageID: "second-answer", Text: "second"}, event.TurnInProgress) |
| 82 | apply(7, "second", "turn_done", 102_500, eventwire.Event{}, event.TurnCompleted) |
| 83 | |
| 84 | records := snapshot(t, p).Records |
| 85 | if got := records[len(records)-1].Message.TurnDurationMs; got != 2_500 { |
| 86 | t.Fatalf("second turn duration = %d, want 2500", got) |
| 87 | } |
| 88 | } |
| 89 | |
| 90 | var testIdentity = Identity{SessionID: "session", HeadID: "head", RuntimeEpoch: "runtime", RewriteEpoch: 1} |
| 91 | |
| 92 | func projectEvent(t *testing.T, p *Projection, seq uint64, e event.Event) { |
| 93 | t.Helper() |
| 94 | w := eventwire.ToWire(e) |
| 95 | status := event.TurnInProgress |
| 96 | if e.Kind == event.TurnDone { |
| 97 | status = event.TurnCompleted |
| 98 | } |
| 99 | if err := p.Apply(turnevent.Envelope{SessionID: "session", RuntimeEpoch: "runtime", TurnID: "turn", Sequence: seq, Kind: w.Kind, Status: status, Event: w}); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | func snapshot(t *testing.T, p *Projection) Snapshot { |
| 105 | t.Helper() |
| 106 | s, err := p.Snapshot(PageRequest{}) |
| 107 | if err != nil { |
| 108 | t.Fatal(err) |
| 109 | } |
| 110 | return s |
| 111 | } |
| 112 | |
| 113 | func TestProjectionSnapshotCoverageMatchesEveryStreamCut(t *testing.T) { |
| 114 | p, err := NewProjection(testIdentity, nil, 0) |
| 115 | if err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | events := []event.Event{ |
| 119 | {Kind: event.UserMessage, MessageID: "user", Text: "question"}, |
| 120 | {Kind: event.StreamAttempt, MessageID: "failed", AttemptID: "failed", StreamAttempt: event.StreamAttemptInfo{ID: "failed", Action: event.StreamAttemptBegin}}, |
| 121 | {Kind: event.Text, MessageID: "failed", AttemptID: "failed", Text: "rejected"}, |
| 122 | {Kind: event.Message, MessageID: "failed", AttemptID: "failed", Text: "rejected full response"}, |
| 123 | {Kind: event.StreamAttempt, MessageID: "failed", AttemptID: "failed", StreamAttempt: event.StreamAttemptInfo{ID: "failed", Action: event.StreamAttemptDiscard}}, |
| 124 | {Kind: event.StreamAttempt, MessageID: "answer", AttemptID: "answer", StreamAttempt: event.StreamAttemptInfo{ID: "answer", Action: event.StreamAttemptBegin}}, |
| 125 | {Kind: event.Reasoning, MessageID: "answer", AttemptID: "answer", Text: "think"}, |
| 126 | {Kind: event.Text, MessageID: "answer", AttemptID: "answer", Text: "same"}, |
| 127 | {Kind: event.Text, MessageID: "answer", AttemptID: "answer", Text: " answer"}, |
| 128 | {Kind: event.Message, MessageID: "answer", AttemptID: "answer", Text: "same answer", Reasoning: "think"}, |
| 129 | {Kind: event.StreamAttempt, MessageID: "answer", AttemptID: "answer", StreamAttempt: event.StreamAttemptInfo{ID: "answer", Action: event.StreamAttemptCommit}}, |
| 130 | {Kind: event.ToolDispatch, MessageID: "answer", Tool: event.Tool{ID: "call", Name: "read_file", Args: `{}`}}, |
| 131 | {Kind: event.ToolResult, MessageID: "answer", Tool: event.Tool{ID: "call", Name: "read_file", Output: "result"}}, |
| 132 | {Kind: event.Message, MessageID: "second", Text: "same answer"}, |
| 133 | {Kind: event.TurnDone}, |
| 134 | } |
| 135 | var cuts []Snapshot |
| 136 | cuts = append(cuts, snapshot(t, p)) |
| 137 | for i, e := range events { |
| 138 | projectEvent(t, p, uint64(i+1), e) |
| 139 | cut := snapshot(t, p) |
| 140 | if cut.CoveredThroughSeq != uint64(i+1) { |
| 141 | t.Fatalf("cut %d has coverage %d", i, cut.CoveredThroughSeq) |
| 142 | } |
| 143 | cuts = append(cuts, cut) |
| 144 | } |
| 145 | want := cuts[len(cuts)-1].Records |
| 146 | if len(want) != 4 || want[1].Message.Content != "same answer" || want[3].Message.Content != "same answer" { |
| 147 | t.Fatalf("final projection: %+v", want) |
| 148 | } |
| 149 | for cut, state := range cuts { |
| 150 | var baseline []Message |
| 151 | for _, record := range state.Records { |
| 152 | baseline = append(baseline, record.Message) |
| 153 | } |
| 154 | restored, err := NewProjection(testIdentity, baseline, state.CoveredThroughSeq) |
| 155 | if err != nil { |
| 156 | t.Fatalf("cut %d: %v", cut, err) |
| 157 | } |
| 158 | for i := cut; i < len(events); i++ { |
| 159 | projectEvent(t, restored, uint64(i+1), events[i]) |
| 160 | } |
| 161 | if got := snapshot(t, restored).Records; !reflect.DeepEqual(got, want) { |
| 162 | t.Fatalf("cut %d changed the final projection\ngot: %#v\nwant: %#v", cut, got, want) |
| 163 | } |
| 164 | } |
| 165 | } |
| 166 | |
| 167 | func TestProjectionRejectsWrongIdentityAndGapWithoutChangingSnapshot(t *testing.T) { |
| 168 | p, _ := NewProjection(testIdentity, nil, 0) |
| 169 | before := snapshot(t, p) |
| 170 | for _, envelope := range []turnevent.Envelope{ |
| 171 | {SessionID: "another", RuntimeEpoch: "runtime", Sequence: 1}, |
| 172 | {SessionID: "session", RuntimeEpoch: "old-runtime", Sequence: 1}, |
| 173 | {SessionID: "session", RuntimeEpoch: "runtime", Sequence: 2}, |
| 174 | } { |
| 175 | if err := p.Apply(envelope); err == nil { |
| 176 | t.Fatal("invalid envelope accepted") |
| 177 | } |
| 178 | if after := snapshot(t, p); !reflect.DeepEqual(after, before) { |
| 179 | t.Fatal("rejected event changed projection") |
| 180 | } |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | func TestProjectionPagingContentAndImmutability(t *testing.T) { |
| 185 | body := strings.Repeat("你好🧪", 20000) |
| 186 | rows := []Message{{MessageID: "u", Role: "user", Content: "question"}, {MessageID: "a", Role: "assistant", Content: body, ToolCalls: []ToolCall{{ID: "call", Name: "write_file", Arguments: body}}}} |
| 187 | p, err := NewProjection(testIdentity, rows, 0) |
| 188 | if err != nil { |
| 189 | t.Fatal(err) |
| 190 | } |
| 191 | rows[1].ToolCalls[0].Name = "caller mutated input" |
| 192 | s, err := p.Snapshot(PageRequest{Records: 1}) |
| 193 | if err != nil { |
| 194 | t.Fatal(err) |
| 195 | } |
| 196 | if !s.HasOlder || s.Before != 1 || len(s.Records) != 1 || len(s.Records[0].Refs) != 2 { |
| 197 | t.Fatalf("bounded page: %+v", s) |
| 198 | } |
| 199 | if s.Records[0].Message.ToolCalls[0].Name != "write_file" { |
| 200 | t.Fatal("input alias retained") |
| 201 | } |
| 202 | for _, ref := range s.Records[0].Refs { |
| 203 | var full strings.Builder |
| 204 | for offset := 0; ; { |
| 205 | chunk, err := p.Content(ContentRequest{ContentRef: ref, Offset: offset}) |
| 206 | if err != nil { |
| 207 | t.Fatal(err) |
| 208 | } |
| 209 | if !utf8.ValidString(chunk.Data) || len(chunk.Data) > contentChunkBytes { |
| 210 | t.Fatal("invalid content chunk") |
| 211 | } |
| 212 | full.WriteString(chunk.Data) |
| 213 | if chunk.Done { |
| 214 | break |
| 215 | } |
| 216 | offset = chunk.NextOffset |
| 217 | } |
| 218 | if full.String() != body { |
| 219 | t.Fatal("content was truncated or duplicated") |
| 220 | } |
| 221 | } |
| 222 | s.Records[0].Message.ToolCalls[0].Name = "caller mutated snapshot" |
| 223 | if next := snapshot(t, p); next.Records[1].Message.ToolCalls[0].Name != "write_file" { |
| 224 | t.Fatal("snapshot alias retained") |
| 225 | } |
| 226 | older, err := p.Snapshot(PageRequest{SnapshotID: s.SnapshotID, Before: s.Before}) |
| 227 | if err != nil || len(older.Records) != 1 || older.Records[0].ID != "m:u" { |
| 228 | t.Fatalf("older page: %+v %v", older, err) |
| 229 | } |
| 230 | projectEvent(t, p, 1, event.Event{Kind: event.Text, MessageID: "b", Text: "new"}) |
| 231 | retained, err := p.Snapshot(PageRequest{SnapshotID: s.SnapshotID, Before: s.Before}) |
| 232 | if err != nil || retained.Stale || len(retained.Records) != 1 || retained.Records[0].ID != "m:u" { |
| 233 | t.Fatal("stream mutation invalidated immutable page") |
| 234 | } |
| 235 | retainedChunk, err := p.Content(ContentRequest{ContentRef: s.Records[0].Refs[0]}) |
| 236 | if err != nil || retainedChunk.Stale || retainedChunk.Data == "" { |
| 237 | t.Fatal("stream mutation invalidated immutable content") |
| 238 | } |
| 239 | for seq := uint64(2); seq <= 4; seq++ { |
| 240 | projectEvent(t, p, seq, event.Event{Kind: event.Text, MessageID: "b", Text: "more"}) |
| 241 | snapshot(t, p) |
| 242 | } |
| 243 | stale, err := p.Snapshot(PageRequest{SnapshotID: s.SnapshotID, Before: s.Before}) |
| 244 | if err != nil || !stale.Stale || len(stale.Records) != 0 { |
| 245 | t.Fatal("old page was combined with new state") |
| 246 | } |
| 247 | chunk, err := p.Content(ContentRequest{ContentRef: s.Records[0].Refs[0]}) |
| 248 | if err != nil || !chunk.Stale { |
| 249 | t.Fatal("old content ref did not become stale") |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | func TestProjectionSnapshotDoesNotDuplicateActiveRecordInPage(t *testing.T) { |
| 254 | p, err := NewProjection(testIdentity, []Message{{ |
| 255 | RecordID: "m:assistant", MessageID: "assistant", Role: "assistant", Content: "partial", Pending: true, |
| 256 | }}, 0) |
| 257 | if err != nil { |
| 258 | t.Fatal(err) |
| 259 | } |
| 260 | snapshot, err := p.Snapshot(PageRequest{}) |
| 261 | if err != nil { |
| 262 | t.Fatal(err) |
| 263 | } |
| 264 | if len(snapshot.Records) != 1 || len(snapshot.ActiveRecords) != 0 { |
| 265 | t.Fatalf("active record duplicated across snapshot arrays: records=%d active=%d", len(snapshot.Records), len(snapshot.ActiveRecords)) |
| 266 | } |
| 267 | if _, err := json.Marshal(snapshot); err != nil { |
| 268 | t.Fatal(err) |
| 269 | } |
| 270 | } |
| 271 | |
| 272 | func TestActiveRecordIndexesDoNotScanSettledTurns(t *testing.T) { |
| 273 | messages := make([]*bufferedMessage, 0, 10000) |
| 274 | for i := range 9990 { |
| 275 | messages = append(messages, &bufferedMessage{message: Message{RecordID: fmt.Sprintf("m:%d", i), Role: "assistant", TurnID: "old"}}) |
| 276 | } |
| 277 | messages = append(messages, |
| 278 | &bufferedMessage{message: Message{RecordID: "m:user", Role: "user", TurnID: "current"}}, |
| 279 | &bufferedMessage{message: Message{RecordID: "m:assistant", Role: "assistant", TurnID: "current", Pending: true}}, |
| 280 | ) |
| 281 | indexes := activeRecordIndexes(messages, len(messages), Runtime{TurnID: "current", Status: event.TurnInProgress}) |
| 282 | if len(indexes) != 2 || indexes[0] != len(messages)-1 || indexes[1] != len(messages)-2 { |
| 283 | t.Fatalf("active indexes = %v, want only current turn owners", indexes) |
| 284 | } |
| 285 | } |
| 286 | |
| 287 | func TestProjectionConcurrentSnapshotNeverClaimsUnappliedText(t *testing.T) { |
| 288 | p, _ := NewProjection(testIdentity, nil, 0) |
| 289 | var wg sync.WaitGroup |
| 290 | wg.Go(func() { |
| 291 | for seq := uint64(1); seq <= 100; seq++ { |
| 292 | projectEvent(t, p, seq, event.Event{Kind: event.Text, MessageID: "a", Text: "x"}) |
| 293 | } |
| 294 | }) |
| 295 | for range 100 { |
| 296 | s := snapshot(t, p) |
| 297 | if len(s.Records) > 0 && len(s.Records[0].Message.Content) != int(s.CoveredThroughSeq) { |
| 298 | t.Fatalf("coverage %d does not cover text", s.CoveredThroughSeq) |
| 299 | } |
| 300 | if _, err := json.Marshal(s); err != nil { |
| 301 | t.Fatal(err) |
| 302 | } |
| 303 | } |
| 304 | wg.Wait() |
| 305 | } |
| 306 | |
| 307 | func TestProjectionAssignsIdentityToZeroSequenceDisplayFrames(t *testing.T) { |
| 308 | tests := []event.Event{ |
| 309 | {Kind: event.Notice, Text: "notice"}, |
| 310 | {Kind: event.Phase, Text: "phase"}, |
| 311 | {Kind: event.Text, Text: "anonymous text"}, |
| 312 | {Kind: event.ToolDispatch, Tool: event.Tool{ID: "call", Name: "test"}}, |
| 313 | {Kind: event.ToolResult, Tool: event.Tool{Name: "test", Output: "output"}}, |
| 314 | {Kind: event.TurnDone, Receipt: &event.CompletionReceipt{}}, |
| 315 | } |
| 316 | for _, e := range tests { |
| 317 | wire := eventwire.ToWire(e) |
| 318 | t.Run(wire.Kind, func(t *testing.T) { |
| 319 | p, err := NewProjection(testIdentity, nil, 0) |
| 320 | if err != nil { |
| 321 | t.Fatal(err) |
| 322 | } |
| 323 | if err := p.ApplyFrame(turnevent.Envelope{Kind: wire.Kind, Event: wire}, 0); err != nil { |
| 324 | t.Fatal(err) |
| 325 | } |
| 326 | cut := snapshot(t, p) |
| 327 | seen := make(map[string]bool) |
| 328 | for _, row := range cut.Records { |
| 329 | if row.ID == "" || row.Message.RecordID != row.ID || seen[row.ID] { |
| 330 | t.Fatalf("invalid generated identity: %+v", row) |
| 331 | } |
| 332 | seen[row.ID] = true |
| 333 | } |
| 334 | if len(seen) == 0 { |
| 335 | t.Fatal("fixture did not create a display record") |
| 336 | } |
| 337 | }) |
| 338 | } |
| 339 | } |
| 340 | |
| 341 | func TestProjectionRejectsInvalidIdentityBeforeCachingSnapshot(t *testing.T) { |
| 342 | for _, messages := range [][]Message{ |
| 343 | {{Role: "notice", Content: "missing"}}, |
| 344 | {{RecordID: "duplicate", Role: "notice", Content: "first"}, {RecordID: "duplicate", Role: "notice", Content: "second"}}, |
| 345 | } { |
| 346 | p, err := NewProjection(testIdentity, nil, 0) |
| 347 | if err != nil { |
| 348 | t.Fatal(err) |
| 349 | } |
| 350 | for _, message := range messages { |
| 351 | p.buffer.messages = append(p.buffer.messages, &bufferedMessage{message: message}) |
| 352 | } |
| 353 | if _, err := p.Snapshot(PageRequest{}); err == nil { |
| 354 | t.Fatal("invalid projection snapshot succeeded") |
| 355 | } |
| 356 | if len(p.snapshots) != 0 || len(p.snapshotOrder) != 0 { |
| 357 | t.Fatal("invalid snapshot entered the reusable cache") |
| 358 | } |
| 359 | } |
| 360 | } |
| 361 |