| 1 | package transcript |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "testing" |
| 6 | |
| 7 | "reasonix/internal/event" |
| 8 | "reasonix/internal/eventwire" |
| 9 | "reasonix/internal/turnevent" |
| 10 | ) |
| 11 | |
| 12 | func businessFrame(t *testing.T, p *Projection, covered uint64, e event.Event) { |
| 13 | t.Helper() |
| 14 | wire := eventwire.ToWire(e) |
| 15 | status := event.TurnInProgress |
| 16 | if e.Kind == event.TurnDone { |
| 17 | status = event.TurnCompleted |
| 18 | } |
| 19 | if err := p.ApplyFrame(turnevent.Envelope{SessionID: testIdentity.SessionID, RuntimeEpoch: testIdentity.RuntimeEpoch, TurnID: "turn", Kind: wire.Kind, Status: status, Event: wire}, covered); err != nil { |
| 20 | t.Fatal(err) |
| 21 | } |
| 22 | } |
| 23 | |
| 24 | func TestBusinessSettlementUpdatesStreamingRowWithoutDuplicateOrPendingState(t *testing.T) { |
| 25 | p, err := NewProjection(testIdentity, nil, 0) |
| 26 | if err != nil { |
| 27 | t.Fatal(err) |
| 28 | } |
| 29 | businessFrame(t, p, 0, event.Event{Kind: event.StreamAttempt, MessageID: "answer", AttemptID: "answer", StreamAttempt: event.StreamAttemptInfo{ID: "answer", Action: event.StreamAttemptBegin}}) |
| 30 | businessFrame(t, p, 0, event.Event{Kind: event.Text, MessageID: "answer", AttemptID: "answer", Text: "partial"}) |
| 31 | p.AcceptBusiness([]Message{{RecordID: "m:answer", MessageID: "answer", Role: "assistant", Content: "complete answer", Reasoning: "complete thinking"}}, 1, "turn", false) |
| 32 | businessFrame(t, p, 1, event.Event{Kind: event.Message, MessageID: "answer", AttemptID: "answer", Text: "complete answer", Reasoning: "complete thinking"}) |
| 33 | businessFrame(t, p, 1, event.Event{Kind: event.StreamAttempt, MessageID: "answer", AttemptID: "answer", StreamAttempt: event.StreamAttemptInfo{ID: "answer", Action: event.StreamAttemptCommit}}) |
| 34 | businessFrame(t, p, 1, event.Event{Kind: event.TurnDone}) |
| 35 | cut := snapshot(t, p) |
| 36 | if len(cut.Records) != 1 { |
| 37 | t.Fatalf("settlement duplicated streaming node: records=%d", len(cut.Records)) |
| 38 | } |
| 39 | message := cut.Records[0].Message |
| 40 | if message.Content != "complete answer" || message.Reasoning != "complete thinking" || message.Pending { |
| 41 | t.Fatalf("settled content or pending state is wrong: %+v", message) |
| 42 | } |
| 43 | if len(cut.ActiveAttempts) != 0 || len(cut.ActiveRecords) != 0 { |
| 44 | t.Fatalf("settlement retains active state: attempts=%d records=%d", len(cut.ActiveAttempts), len(cut.ActiveRecords)) |
| 45 | } |
| 46 | } |
| 47 | |
| 48 | func TestBusinessRowsWithoutCanonicalIdentityReceiveDistinctViewIdentity(t *testing.T) { |
| 49 | p, err := NewProjection(testIdentity, nil, 0) |
| 50 | if err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | initial, err := p.Follow(t.Context(), FollowRequest{}) |
| 54 | if err != nil { |
| 55 | t.Fatal(err) |
| 56 | } |
| 57 | p.AcceptBusiness([]Message{{Role: "notice", Content: "first"}, {Role: "notice", Content: "second"}}, 1, "turn", false) |
| 58 | cut := snapshot(t, p) |
| 59 | if len(cut.Records) != 2 || cut.Records[0].ID == "" || cut.Records[0].ID == cut.Records[1].ID { |
| 60 | t.Fatalf("business identity allocation = %+v", cut.Records) |
| 61 | } |
| 62 | suffix := followChanges(t, p, FollowRequest{Subscription: initial.Subscription, AfterRevision: initial.Snapshot.ProjectionRevision}) |
| 63 | if len(suffix.Changes) != 1 || len(suffix.Changes[0].Records) != 2 || suffix.Changes[0].Records[0].RecordID == "" || |
| 64 | suffix.Changes[0].Records[0].RecordID == suffix.Changes[0].Records[1].RecordID { |
| 65 | t.Fatalf("published business identities = %+v", suffix.Changes) |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | func TestBusinessPublisherReleasesCompletedActiveRowsWithinHistoryBudget(t *testing.T) { |
| 70 | p, err := NewProjection(testIdentity, nil, 0) |
| 71 | if err != nil { |
| 72 | t.Fatal(err) |
| 73 | } |
| 74 | businessFrame(t, p, 0, event.Event{Kind: event.StreamAttempt, MessageID: "old-answer", AttemptID: "old-answer", StreamAttempt: event.StreamAttemptInfo{ID: "old-answer", Action: event.StreamAttemptBegin}}) |
| 75 | businessFrame(t, p, 0, event.Event{Kind: event.Text, MessageID: "old-answer", Text: "old partial answer"}) |
| 76 | for i := range 120 { |
| 77 | id := fmt.Sprintf("new-answer-%03d", i) |
| 78 | p.AcceptBusiness([]Message{{RecordID: "m:" + id, MessageID: id, Role: "assistant", Content: id}}, uint64(i+1), fmt.Sprintf("turn-%d", i), false) |
| 79 | } |
| 80 | active := snapshot(t, p) |
| 81 | retained := false |
| 82 | for _, records := range [][]Record{active.Records, active.ActiveRecords} { |
| 83 | for _, record := range records { |
| 84 | retained = retained || record.Message.MessageID == "old-answer" && record.Message.Content == "old partial answer" |
| 85 | } |
| 86 | } |
| 87 | if len(active.ActiveAttempts) != 1 || !retained { |
| 88 | t.Fatal("appended history evicted the active assistant prefix") |
| 89 | } |
| 90 | // A terminal turn must release pending state even if its attempt end frame |
| 91 | // was unavailable. No later user submission should be needed to reclaim |
| 92 | // rows that were exempted from the resident budget while still active. |
| 93 | businessFrame(t, p, 120, event.Event{Kind: event.TurnDone}) |
| 94 | cut := snapshot(t, p) |
| 95 | if cut.TotalRecords > 96 || len(cut.ActiveRecords) != 0 || len(cut.ActiveAttempts) != 0 { |
| 96 | t.Fatalf("completed rows escaped resident budget: total=%d activeRows=%d activeAttempts=%d", cut.TotalRecords, len(cut.ActiveRecords), len(cut.ActiveAttempts)) |
| 97 | } |
| 98 | for _, record := range cut.Records { |
| 99 | if record.Message.MessageID == "old-answer" || record.Message.Pending { |
| 100 | t.Fatalf("completed active row was retained indefinitely: %+v", record.Message) |
| 101 | } |
| 102 | } |
| 103 | } |
| 104 |