| 1 | package transcript |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "strings" |
| 7 | "testing" |
| 8 | |
| 9 | "reasonix/internal/event" |
| 10 | "reasonix/internal/eventwire" |
| 11 | "reasonix/internal/provider" |
| 12 | "reasonix/internal/turnevent" |
| 13 | ) |
| 14 | |
| 15 | func TestSnapshotRetainsTerminalRecoveryAndFailure(t *testing.T) { |
| 16 | for _, tc := range []struct { |
| 17 | name string |
| 18 | e event.Event |
| 19 | code string |
| 20 | }{ |
| 21 | {"protocol", event.Event{Status: event.TurnFailed, Err: errors.New("provider failed"), ProtocolRecovery: &provider.ProtocolRecoveryAction{ID: "recover"}}, "protocol_recovery"}, |
| 22 | {"readiness", event.Event{Status: event.TurnFailed, Outcome: event.TurnOutcomeFinalReadiness, Readiness: &event.FinalReadiness{Missing: []string{"checks"}}}, event.NoticeCodeFinalReadiness}, |
| 23 | {"read pause", event.Event{Status: event.TurnFailed, Outcome: event.TurnOutcomeIncompleteRead, ReadPause: &provider.ReadPause{}}, event.TurnOutcomeIncompleteRead}, |
| 24 | {"read completion", event.Event{Status: event.TurnCompleted, ReadCompletion: &provider.ReadCompletion{ID: "run"}}, "read_completion"}, |
| 25 | {"cancel", event.Event{Status: event.TurnInterrupted}, event.NoticeCodeCancelledTurn}, |
| 26 | {"unknown effect", event.Event{Status: event.TurnRecoveryRequired}, event.NoticeCodeCancelledTurn}, |
| 27 | {"failure", event.Event{Status: event.TurnFailed, Err: errors.New("provider failed")}, event.NoticeCodeProviderRequestFailed}, |
| 28 | {"recovery pause", event.Event{Status: event.TurnFailed, Outcome: event.TurnOutcomeRecoveryPaused}, event.TurnOutcomeRecoveryPaused}, |
| 29 | {"uncertain", event.Event{Status: event.TurnCompleted, Outcome: event.TurnOutcomeCompletionUncertain}, event.TurnOutcomeCompletionUncertain}, |
| 30 | } { |
| 31 | t.Run(tc.name, func(t *testing.T) { |
| 32 | p, err := NewProjection(testIdentity, nil, 0) |
| 33 | if err != nil { |
| 34 | t.Fatal(err) |
| 35 | } |
| 36 | tc.e.Kind = event.TurnDone |
| 37 | w := eventwire.ToWire(tc.e) |
| 38 | if err := p.Apply(turnevent.Envelope{SessionID: testIdentity.SessionID, RuntimeEpoch: testIdentity.RuntimeEpoch, TurnID: "turn", Sequence: 1, Kind: w.Kind, Status: tc.e.Status, Event: w}); err != nil { |
| 39 | t.Fatal(err) |
| 40 | } |
| 41 | check := func(p *Projection) { |
| 42 | t.Helper() |
| 43 | for _, row := range snapshot(t, p).Records { |
| 44 | if row.Message.Code != tc.code { |
| 45 | continue |
| 46 | } |
| 47 | if tc.e.ProtocolRecovery != nil && (row.Message.ProtocolRecovery == nil || row.Message.ProtocolRecovery.ID != "recover" || !row.Message.Pending) { |
| 48 | t.Fatal("lost recovery token") |
| 49 | } |
| 50 | if tc.e.Readiness != nil && (row.Message.Readiness == nil || len(row.Message.Readiness.Missing) != 1) { |
| 51 | t.Fatal("lost readiness") |
| 52 | } |
| 53 | if tc.e.ReadPause != nil && row.Message.ReadPause == nil { |
| 54 | t.Fatal("lost read pause") |
| 55 | } |
| 56 | return |
| 57 | } |
| 58 | t.Fatalf("covered terminal event but lost %s", tc.code) |
| 59 | } |
| 60 | check(p) |
| 61 | state, err := p.Checkpoint("digest") |
| 62 | if err != nil { |
| 63 | t.Fatal(err) |
| 64 | } |
| 65 | restored, err := RestoreCheckpoint(state, testIdentity) |
| 66 | if err != nil { |
| 67 | t.Fatal(err) |
| 68 | } |
| 69 | check(restored) |
| 70 | }) |
| 71 | } |
| 72 | } |
| 73 | |
| 74 | func TestInterruptedTurnKeepsProviderFailure(t *testing.T) { |
| 75 | quota := &provider.QuotaError{Status: 402, Provider: "relay", Protocol: "openai"} |
| 76 | for _, tc := range []struct { |
| 77 | name string |
| 78 | e event.Event |
| 79 | wantFail bool |
| 80 | }{ |
| 81 | {"recovery required with quota error", event.Event{Status: event.TurnRecoveryRequired, Err: quota, Diagnostic: provider.DiagnoseFailure(quota), |
| 82 | Recovery: &event.RecoveryStatus{State: "recovery_required", Reason: "silent_interruption"}}, true}, |
| 83 | {"interrupted with quota error", event.Event{Status: event.TurnInterrupted, Err: quota, Diagnostic: provider.DiagnoseFailure(quota)}, true}, |
| 84 | {"user cancel", event.Event{Status: event.TurnInterrupted, Err: context.Canceled, Diagnostic: provider.DiagnoseFailure(context.Canceled)}, false}, |
| 85 | } { |
| 86 | t.Run(tc.name, func(t *testing.T) { |
| 87 | p, err := NewProjection(testIdentity, nil, 0) |
| 88 | if err != nil { |
| 89 | t.Fatal(err) |
| 90 | } |
| 91 | tc.e.Kind = event.TurnDone |
| 92 | w := eventwire.ToWire(tc.e) |
| 93 | if err := p.Apply(turnevent.Envelope{SessionID: testIdentity.SessionID, RuntimeEpoch: testIdentity.RuntimeEpoch, TurnID: "turn", Sequence: 1, Kind: w.Kind, Status: tc.e.Status, Event: w}); err != nil { |
| 94 | t.Fatal(err) |
| 95 | } |
| 96 | var interrupted bool |
| 97 | var failure *Message |
| 98 | for _, row := range snapshot(t, p).Records { |
| 99 | switch row.Message.Code { |
| 100 | case event.NoticeCodeCancelledTurn: |
| 101 | interrupted = true |
| 102 | case event.NoticeCodeProviderRequestFailed: |
| 103 | failure = &row.Message |
| 104 | } |
| 105 | } |
| 106 | if !interrupted { |
| 107 | t.Fatal("interrupted notice missing") |
| 108 | } |
| 109 | if !tc.wantFail { |
| 110 | if failure != nil { |
| 111 | t.Fatalf("cancelled turn rendered a provider failure: %q", failure.Content) |
| 112 | } |
| 113 | return |
| 114 | } |
| 115 | if failure == nil || !strings.Contains(failure.Content, "HTTP 402") || failure.Level != "warn" || failure.Diagnostic == nil || failure.Diagnostic.Status != 402 { |
| 116 | t.Fatalf("provider failure hidden behind interrupted notice: %+v", failure) |
| 117 | } |
| 118 | }) |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | func TestHistoryRetainsReadReceipts(t *testing.T) { |
| 123 | rows := History([]provider.Message{ |
| 124 | {ID: "pause", LocalOnly: true, ReadPause: &provider.ReadPause{}}, |
| 125 | {ID: "complete", LocalOnly: true, ReadCompletion: &provider.ReadCompletion{ID: "run"}}, |
| 126 | }, HistoryOptions{}) |
| 127 | if len(rows) != 2 || rows[0].ReadPause == nil || rows[1].ReadCompletion == nil { |
| 128 | t.Fatalf("lost read receipts: %+v", rows) |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | func TestCheckpointOwnsNestedRecoveryMetadata(t *testing.T) { |
| 133 | p, err := NewProjection(testIdentity, []Message{{RecordID: "receipt", Role: "notice", ReadCompletion: &provider.ReadCompletion{ |
| 134 | ID: "run", Reads: []provider.CompletedRead{{Covered: [][2]int{{0, 12}}}}, |
| 135 | }}}, 0) |
| 136 | if err != nil { |
| 137 | t.Fatal(err) |
| 138 | } |
| 139 | checkpoint, err := p.Checkpoint("digest") |
| 140 | if err != nil { |
| 141 | t.Fatal(err) |
| 142 | } |
| 143 | checkpoint.Records[0].ReadCompletion.Reads[0].Covered[0][1] = 99 |
| 144 | if got := snapshot(t, p).Records[0].Message.ReadCompletion.Reads[0].Covered[0][1]; got != 12 { |
| 145 | t.Fatalf("checkpoint mutated live projection: %d", got) |
| 146 | } |
| 147 | } |
| 148 |