| 1 | package transcript |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "fmt" |
| 6 | "strings" |
| 7 | "sync" |
| 8 | "sync/atomic" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/event" |
| 13 | "reasonix/internal/eventwire" |
| 14 | "reasonix/internal/turnevent" |
| 15 | ) |
| 16 | |
| 17 | func outlineProjection(t *testing.T, baseline []Message) *Projection { |
| 18 | t.Helper() |
| 19 | p, err := NewProjection(testIdentity, baseline, 0) |
| 20 | if err != nil { |
| 21 | t.Fatal(err) |
| 22 | } |
| 23 | return p |
| 24 | } |
| 25 | |
| 26 | func outline(t *testing.T, p *Projection, req OutlineRequest) OutlinePage { |
| 27 | t.Helper() |
| 28 | page, err := p.Outline(req) |
| 29 | if err != nil { |
| 30 | t.Fatal(err) |
| 31 | } |
| 32 | return page |
| 33 | } |
| 34 | |
| 35 | func turn(recordID, prompt, answer string) []Message { |
| 36 | rows := []Message{{RecordID: recordID, MessageID: recordID, Role: "user", Content: prompt}} |
| 37 | if answer != "" { |
| 38 | rows = append(rows, Message{RecordID: recordID + ":a", MessageID: recordID + ":a", Role: "assistant", Content: answer}) |
| 39 | } |
| 40 | return rows |
| 41 | } |
| 42 | |
| 43 | // The reported defect: a tool-heavy tail fills the first body page, so a rail |
| 44 | // derived from loaded records shows nothing even though the session has turns. |
| 45 | func TestOutlineCoversTurnsOutsideTheLoadedBodyPage(t *testing.T) { |
| 46 | baseline := turn("m:1", "first question", "first answer") |
| 47 | baseline = append(baseline, turn("m:2", "second question", "second answer")...) |
| 48 | for i := range 200 { |
| 49 | baseline = append(baseline, Message{RecordID: "tool:" + string(rune('a'+i%26)) + string(rune('a'+i/26)), Role: "tool", ToolCallID: "call", Content: "output"}) |
| 50 | } |
| 51 | p := outlineProjection(t, baseline) |
| 52 | |
| 53 | records := snapshot(t, p).Records |
| 54 | for _, record := range records { |
| 55 | if record.Message.Role == "user" { |
| 56 | t.Fatalf("test premise broken: the default page already contains a user turn") |
| 57 | } |
| 58 | } |
| 59 | |
| 60 | page := outline(t, p, OutlineRequest{}) |
| 61 | if page.Stale || page.Total != 2 || len(page.Entries) != 2 { |
| 62 | t.Fatalf("outline = %+v, want 2 complete turns", page) |
| 63 | } |
| 64 | if page.Entries[0].Prompt != "first question" || page.Entries[0].Answer != "first answer" { |
| 65 | t.Fatalf("first entry = %+v", page.Entries[0]) |
| 66 | } |
| 67 | if page.Entries[1].Prompt != "second question" { |
| 68 | t.Fatalf("second entry = %+v", page.Entries[1]) |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | // A page's identity must survive the body cursor moving, and must match the |
| 73 | // record identity the body pages use. |
| 74 | func TestOutlineIdentityMatchesBodyRecords(t *testing.T) { |
| 75 | baseline := turn("m:1", "one", "answer one") |
| 76 | baseline = append(baseline, turn("m:2", "two", "answer two")...) |
| 77 | baseline = append(baseline, turn("m:3", "three", "answer three")...) |
| 78 | p := outlineProjection(t, baseline) |
| 79 | |
| 80 | full := snapshot(t, p) |
| 81 | wantOrder := map[string]int{} |
| 82 | for _, record := range full.Records { |
| 83 | if record.Message.Role == "user" { |
| 84 | wantOrder[record.ID] = record.Order |
| 85 | } |
| 86 | } |
| 87 | if len(wantOrder) != 3 { |
| 88 | t.Fatalf("body page holds %d user records, want 3", len(wantOrder)) |
| 89 | } |
| 90 | |
| 91 | first := outline(t, p, OutlineRequest{SnapshotID: full.SnapshotID, Entries: 2}) |
| 92 | if first.Total != 3 || len(first.Entries) != 2 || first.Done { |
| 93 | t.Fatalf("first page = %+v", first) |
| 94 | } |
| 95 | if first.NextOffset != 2 || first.SnapshotID != full.SnapshotID { |
| 96 | t.Fatalf("first page cursor = %+v", first) |
| 97 | } |
| 98 | second := outline(t, p, OutlineRequest{SnapshotID: full.SnapshotID, Offset: first.NextOffset}) |
| 99 | if len(second.Entries) != 1 || !second.Done || second.NextOffset != 3 { |
| 100 | t.Fatalf("second page = %+v", second) |
| 101 | } |
| 102 | for _, entry := range append(append([]OutlineEntry{}, first.Entries...), second.Entries...) { |
| 103 | order, ok := wantOrder[entry.ID] |
| 104 | if !ok { |
| 105 | t.Fatalf("outline entry %q has no body record", entry.ID) |
| 106 | } |
| 107 | if entry.Order != order { |
| 108 | t.Fatalf("entry %q order = %d, want %d", entry.ID, entry.Order, order) |
| 109 | } |
| 110 | } |
| 111 | if first.Entries[0].Turn != 1 || second.Entries[0].Turn != 3 { |
| 112 | t.Fatalf("turn numbering is not absolute: %d..%d", first.Entries[0].Turn, second.Entries[0].Turn) |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | // Loading an older body page must not renumber or shrink the rail. |
| 117 | func TestOutlineSurvivesOlderBodyPaging(t *testing.T) { |
| 118 | baseline := turn("m:1", "one", "answer one") |
| 119 | baseline = append(baseline, turn("m:2", "two", "answer two")...) |
| 120 | baseline = append(baseline, turn("m:3", "three", "answer three")...) |
| 121 | for i := range 200 { |
| 122 | baseline = append(baseline, Message{RecordID: fmt.Sprintf("tool:%d", i), Role: "tool", ToolCallID: "call", Content: "output"}) |
| 123 | } |
| 124 | p := outlineProjection(t, baseline) |
| 125 | |
| 126 | full := snapshot(t, p) |
| 127 | if full.Before == 0 { |
| 128 | t.Fatalf("test premise broken: the default page already covers every record") |
| 129 | } |
| 130 | before := outline(t, p, OutlineRequest{SnapshotID: full.SnapshotID}) |
| 131 | |
| 132 | page, err := p.Snapshot(PageRequest{SnapshotID: full.SnapshotID, Before: full.Before, Records: 1}) |
| 133 | if err != nil { |
| 134 | t.Fatal(err) |
| 135 | } |
| 136 | if len(page.Records) != 1 || page.Before >= full.Before { |
| 137 | t.Fatalf("older page did not advance the cursor: %+v", page) |
| 138 | } |
| 139 | after := outline(t, p, OutlineRequest{SnapshotID: full.SnapshotID}) |
| 140 | if after.Total != before.Total || len(after.Entries) != len(before.Entries) { |
| 141 | t.Fatalf("older paging shrank the outline: %d -> %d", before.Total, after.Total) |
| 142 | } |
| 143 | for i := range after.Entries { |
| 144 | if after.Entries[i] != before.Entries[i] { |
| 145 | t.Fatalf("entry %d changed after paging: %+v -> %+v", i, before.Entries[i], after.Entries[i]) |
| 146 | } |
| 147 | } |
| 148 | } |
| 149 | |
| 150 | // An evicted cut must report staleness rather than answering positions against |
| 151 | // the newest revision. |
| 152 | func TestOutlineReportsStaleCutInsteadOfRepositioning(t *testing.T) { |
| 153 | p := outlineProjection(t, turn("m:1", "one", "answer one")) |
| 154 | full := snapshot(t, p) |
| 155 | |
| 156 | stale := outline(t, p, OutlineRequest{SnapshotID: "evicted-cut"}) |
| 157 | if !stale.Stale || len(stale.Entries) != 0 { |
| 158 | t.Fatalf("stale outline = %+v", stale) |
| 159 | } |
| 160 | if stale.SnapshotID != full.SnapshotID || stale.ProtocolVersion != ProtocolVersion { |
| 161 | t.Fatalf("stale outline must still carry the live boundary: %+v", stale.Boundary) |
| 162 | } |
| 163 | } |
| 164 | |
| 165 | func TestOutlinePreviewIsBoundedCollapsedAndDisplayOnly(t *testing.T) { |
| 166 | longPrompt := " first\n\nline\t" + strings.Repeat("宽", 80) |
| 167 | longAnswer := "answer\n\nbody " + strings.Repeat("答", 200) |
| 168 | baseline := []Message{ |
| 169 | {RecordID: "m:1", MessageID: "m:1", Role: "user", Content: longPrompt}, |
| 170 | {RecordID: "m:1:think", MessageID: "m:1:think", Role: "assistant", Reasoning: strings.Repeat("thinking ", 40), Content: "short answer"}, |
| 171 | {RecordID: "m:1:tool", Role: "tool", ToolCallID: "call", Content: strings.Repeat("tool output ", 40)}, |
| 172 | {RecordID: "m:1:a", MessageID: "m:1:a", Role: "assistant", Content: longAnswer}, |
| 173 | } |
| 174 | p := outlineProjection(t, baseline) |
| 175 | |
| 176 | page := outline(t, p, OutlineRequest{}) |
| 177 | if len(page.Entries) != 1 { |
| 178 | t.Fatalf("entries = %d, want 1", len(page.Entries)) |
| 179 | } |
| 180 | entry := page.Entries[0] |
| 181 | if strings.ContainsAny(entry.Prompt, "\n\t") || strings.Contains(entry.Prompt, " ") { |
| 182 | t.Fatalf("prompt whitespace was not collapsed: %q", entry.Prompt) |
| 183 | } |
| 184 | if got := len([]rune(strings.TrimSuffix(entry.Prompt, "…"))); got > promptPreviewRunes { |
| 185 | t.Fatalf("prompt preview = %d runes, want <= %d", got, promptPreviewRunes) |
| 186 | } |
| 187 | if got := len([]rune(strings.TrimSuffix(entry.Answer, "…"))); got > answerPreviewRunes { |
| 188 | t.Fatalf("answer preview = %d runes, want <= %d", got, answerPreviewRunes) |
| 189 | } |
| 190 | // The last assistant body wins; reasoning and tool output never leak in. |
| 191 | if !strings.HasPrefix(entry.Answer, "answer body 答") { |
| 192 | t.Fatalf("answer preview = %q, want the final assistant body", entry.Answer) |
| 193 | } |
| 194 | if strings.Contains(entry.Answer, "thinking") || strings.Contains(entry.Answer, "tool output") { |
| 195 | t.Fatalf("preview leaked reasoning or tool output: %q", entry.Answer) |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | func TestOutlineKeepsEmptyPromptTurnIdentity(t *testing.T) { |
| 200 | baseline := []Message{ |
| 201 | {RecordID: "m:1", MessageID: "m:1", Role: "user", Content: " "}, |
| 202 | {RecordID: "m:1:a", MessageID: "m:1:a", Role: "assistant", Content: "answer"}, |
| 203 | {RecordID: "m:2", MessageID: "m:2", Role: "user", Content: "second"}, |
| 204 | } |
| 205 | p := outlineProjection(t, baseline) |
| 206 | |
| 207 | page := outline(t, p, OutlineRequest{}) |
| 208 | if page.Total != 2 { |
| 209 | t.Fatalf("empty prompt removed a navigation item: %+v", page) |
| 210 | } |
| 211 | if page.Entries[0].ID != "m:1" || page.Entries[0].Turn != 1 || page.Entries[1].Turn != 2 { |
| 212 | t.Fatalf("turn identity or numbering = %+v", page.Entries) |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | // A turn with no completed answer preface yet keeps its identity and simply |
| 217 | // carries no answer preview. |
| 218 | func TestOutlineRunningTurnHasIdentityWithoutAnswer(t *testing.T) { |
| 219 | p := outlineProjection(t, turn("m:1", "one", "answer one")) |
| 220 | projectEvent(t, p, 1, event.Event{Kind: event.UserMessage, MessageID: "2", Text: "running question"}) |
| 221 | |
| 222 | page := outline(t, p, OutlineRequest{}) |
| 223 | if page.Total != 2 { |
| 224 | t.Fatalf("running turn missing from outline: %+v", page) |
| 225 | } |
| 226 | if page.Entries[1].ID != "m:2" || page.Entries[1].Prompt != "running question" || page.Entries[1].Answer != "" { |
| 227 | t.Fatalf("running turn = %+v", page.Entries[1]) |
| 228 | } |
| 229 | } |
| 230 | |
| 231 | func TestOutlinePageRespectsByteBudgetAndAdvances(t *testing.T) { |
| 232 | var baseline []Message |
| 233 | for i := range 40 { |
| 234 | baseline = append(baseline, turn(recordName(i), strings.Repeat("q", 40), strings.Repeat("a", 80))...) |
| 235 | } |
| 236 | p := outlineProjection(t, baseline) |
| 237 | |
| 238 | page := outline(t, p, OutlineRequest{Bytes: 1}) |
| 239 | if len(page.Entries) != 1 { |
| 240 | t.Fatalf("a byte budget below one entry must still advance: %+v", page) |
| 241 | } |
| 242 | if page.Done || page.NextOffset != 1 { |
| 243 | t.Fatalf("cursor did not advance: %+v", page) |
| 244 | } |
| 245 | if got := outline(t, p, OutlineRequest{Offset: page.NextOffset, Bytes: defaultOutlineBytes}); len(got.Entries) == 0 { |
| 246 | t.Fatalf("second page is empty: %+v", got) |
| 247 | } |
| 248 | total := outline(t, p, OutlineRequest{}) |
| 249 | if total.Total != 40 { |
| 250 | t.Fatalf("total = %d, want 40", total.Total) |
| 251 | } |
| 252 | } |
| 253 | |
| 254 | func TestOutlinePageStaysWithinResponseLimit(t *testing.T) { |
| 255 | var baseline []Message |
| 256 | for i := range 300 { |
| 257 | baseline = append(baseline, turn(recordName(i), strings.Repeat("промпт", 20), strings.Repeat("ответ", 60))...) |
| 258 | } |
| 259 | p := outlineProjection(t, baseline) |
| 260 | |
| 261 | page := outline(t, p, OutlineRequest{Entries: 10_000, Bytes: 64 << 20}) |
| 262 | if len(page.Entries) > maxOutlineEntries { |
| 263 | t.Fatalf("entry cap ignored: %d", len(page.Entries)) |
| 264 | } |
| 265 | encoded, err := json.Marshal(page) |
| 266 | if err != nil { |
| 267 | t.Fatal(err) |
| 268 | } |
| 269 | if len(encoded)+1 > MaxResponseBytes { |
| 270 | t.Fatalf("outline page = %d bytes, over the response limit", len(encoded)) |
| 271 | } |
| 272 | } |
| 273 | |
| 274 | // Offsets beyond the end clamp instead of reporting a negative or skipped page. |
| 275 | func TestOutlineOffsetPastEndIsDone(t *testing.T) { |
| 276 | p := outlineProjection(t, turn("m:1", "one", "answer one")) |
| 277 | page := outline(t, p, OutlineRequest{Offset: 99}) |
| 278 | if len(page.Entries) != 0 || !page.Done || page.NextOffset != 1 || page.Total != 1 { |
| 279 | t.Fatalf("past-end page = %+v", page) |
| 280 | } |
| 281 | } |
| 282 | |
| 283 | func TestOutlineEmptySessionIsComplete(t *testing.T) { |
| 284 | p := outlineProjection(t, nil) |
| 285 | page := outline(t, p, OutlineRequest{}) |
| 286 | if page.Total != 0 || !page.Done || page.Stale || len(page.Entries) != 0 { |
| 287 | t.Fatalf("empty outline = %+v", page) |
| 288 | } |
| 289 | // Entries must encode as an empty array, never null, so clients can iterate. |
| 290 | encoded, err := json.Marshal(page) |
| 291 | if err != nil { |
| 292 | t.Fatal(err) |
| 293 | } |
| 294 | if !strings.Contains(string(encoded), `"entries":[]`) { |
| 295 | t.Fatalf("entries encoded as %s", encoded) |
| 296 | } |
| 297 | } |
| 298 | |
| 299 | func recordName(i int) string { |
| 300 | return fmt.Sprintf("m:%d", i) |
| 301 | } |
| 302 | |
| 303 | // Outline reads share the projection mutex with streaming commits, and an |
| 304 | // evicted cut must never be served from a newer revision. |
| 305 | func TestOutlineConcurrentWithStreamingCommits(t *testing.T) { |
| 306 | baseline := turn("m:1", "one", "answer one") |
| 307 | baseline = append(baseline, turn("m:2", "two", "answer two")...) |
| 308 | p := outlineProjection(t, baseline) |
| 309 | |
| 310 | const workers, readers = 4, 8 |
| 311 | var wg sync.WaitGroup |
| 312 | stop := make(chan struct{}) |
| 313 | for worker := range workers { |
| 314 | wg.Go(func() { |
| 315 | sequence := uint64(0) |
| 316 | for { |
| 317 | select { |
| 318 | case <-stop: |
| 319 | return |
| 320 | default: |
| 321 | } |
| 322 | sequence++ |
| 323 | _ = p.Apply(turnevent.Envelope{ |
| 324 | SessionID: "session", RuntimeEpoch: "runtime", TurnID: fmt.Sprintf("turn-%d", worker), |
| 325 | Sequence: sequence, Kind: "message", Status: event.TurnInProgress, |
| 326 | Event: eventwire.Event{Kind: "message", MessageID: fmt.Sprintf("w%d-%d", worker, sequence), Text: "streamed"}, |
| 327 | }) |
| 328 | } |
| 329 | }) |
| 330 | } |
| 331 | var reads atomic.Int64 |
| 332 | for range readers { |
| 333 | wg.Go(func() { |
| 334 | for { |
| 335 | select { |
| 336 | case <-stop: |
| 337 | return |
| 338 | default: |
| 339 | } |
| 340 | page, err := p.Outline(OutlineRequest{Entries: 4}) |
| 341 | if err != nil { |
| 342 | t.Errorf("outline: %v", err) |
| 343 | return |
| 344 | } |
| 345 | // Every page either describes a live cut or reports staleness. |
| 346 | if !page.Stale && page.Total < 2 { |
| 347 | t.Errorf("live outline lost turns: %+v", page) |
| 348 | return |
| 349 | } |
| 350 | reads.Add(1) |
| 351 | } |
| 352 | }) |
| 353 | } |
| 354 | time.Sleep(150 * time.Millisecond) |
| 355 | close(stop) |
| 356 | wg.Wait() |
| 357 | if reads.Load() == 0 { |
| 358 | t.Fatal("no outline reads overlapped the writers") |
| 359 | } |
| 360 | } |
| 361 |