| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "cmp" |
| 5 | "context" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "net/http" |
| 9 | "net/http/httptest" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "slices" |
| 13 | "sync" |
| 14 | "testing" |
| 15 | "time" |
| 16 | |
| 17 | "reasonix/internal/config" |
| 18 | "reasonix/internal/transcript" |
| 19 | ) |
| 20 | |
| 21 | // rendererFollowCursor mirrors TranscriptFollowClient.validate in |
| 22 | // desktop/frontend/src/lib/transcriptFollowClient.ts (lines 174-220): any |
| 23 | // error it returns is one that makes the renderer drop the subscription. |
| 24 | type rendererFollowCursor struct { |
| 25 | revision, coverage uint64 |
| 26 | indexes map[string]uint64 |
| 27 | attempts map[string]string |
| 28 | results map[string]uint64 |
| 29 | resultOrder []string |
| 30 | deltas map[string]string |
| 31 | } |
| 32 | |
| 33 | func newRendererFollowCursor(snapshot *transcript.Snapshot) *rendererFollowCursor { |
| 34 | c := &rendererFollowCursor{revision: snapshot.ProjectionRevision, coverage: snapshot.CoveredThroughSeq, |
| 35 | indexes: map[string]uint64{}, attempts: map[string]string{}, results: map[string]uint64{}, deltas: map[string]string{}} |
| 36 | for _, attempt := range snapshot.ActiveAttempts { |
| 37 | c.indexes[attempt.ID], c.attempts[attempt.ID] = attempt.NextIndex, attempt.MessageID |
| 38 | } |
| 39 | return c |
| 40 | } |
| 41 | |
| 42 | func (c *rendererFollowCursor) noteResult(messageID string, sequence uint64) { |
| 43 | if _, ok := c.results[messageID]; !ok { |
| 44 | c.resultOrder = append(c.resultOrder, messageID) |
| 45 | } |
| 46 | c.results[messageID] = sequence |
| 47 | for len(c.resultOrder) > 192 { |
| 48 | delete(c.results, c.resultOrder[0]) |
| 49 | c.resultOrder = c.resultOrder[1:] |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | func (c *rendererFollowCursor) accept(changes []transcript.Change) error { |
| 54 | ordered := slices.Clone(changes) |
| 55 | slices.SortFunc(ordered, func(a, b transcript.Change) int { return cmp.Compare(a.Revision, b.Revision) }) |
| 56 | for _, change := range ordered { |
| 57 | if change.Revision <= c.revision { |
| 58 | continue |
| 59 | } |
| 60 | if change.ResetRequired || change.Revision != c.revision+1 { |
| 61 | return fmt.Errorf("revision_gap at revision %d", change.Revision) |
| 62 | } |
| 63 | if change.FirstSeq != 0 { |
| 64 | if change.FirstSeq != c.coverage+1 || change.CommitSeq < change.FirstSeq { |
| 65 | return fmt.Errorf("business_gap at revision %d: first=%d coverage=%d", change.Revision, change.FirstSeq, c.coverage) |
| 66 | } |
| 67 | c.coverage = change.CommitSeq |
| 68 | } else if change.CommitSeq != c.coverage { |
| 69 | return fmt.Errorf("frame_cut_mismatch at revision %d: commit=%d coverage=%d", change.Revision, change.CommitSeq, c.coverage) |
| 70 | } |
| 71 | e := change.Event |
| 72 | for _, record := range change.Records { |
| 73 | if record.MessageID != "" { |
| 74 | c.noteResult(record.MessageID, change.CommitSeq) |
| 75 | } |
| 76 | } |
| 77 | if e != nil && e.Kind == "stream_attempt" && e.StreamAttempt != nil && e.StreamAttempt.Action == "begin" { |
| 78 | if e.MessageID == "" { |
| 79 | return fmt.Errorf("sampling_identity_missing at revision %d", change.Revision) |
| 80 | } |
| 81 | c.indexes[e.StreamAttempt.ID], c.attempts[e.StreamAttempt.ID] = 0, e.MessageID |
| 82 | } |
| 83 | if change.AttemptID != "" && change.ResultSeq == 0 && (e == nil || e.Kind != "stream_attempt") { |
| 84 | if c.indexes[change.AttemptID] != change.Index { |
| 85 | return fmt.Errorf("sampling_gap at revision %d", change.Revision) |
| 86 | } |
| 87 | c.indexes[change.AttemptID] = change.Index + 1 |
| 88 | } |
| 89 | if change.ResultSeq != 0 && (change.ResultSeq > c.coverage || (change.ResultKind != "message/complete" && change.ResultKind != "message/interrupted")) { |
| 90 | return fmt.Errorf("settlement_not_committed at revision %d", change.Revision) |
| 91 | } |
| 92 | if change.ResultSeq != 0 { |
| 93 | message := c.attempts[change.AttemptID] |
| 94 | recorded, known := c.results[message] |
| 95 | if message == "" || e == nil || message != e.MessageID || known && recorded != change.ResultSeq { |
| 96 | return fmt.Errorf("settlement_identity_mismatch at revision %d", change.Revision) |
| 97 | } |
| 98 | } |
| 99 | if e != nil && e.Kind == "stream_attempt" && e.StreamAttempt != nil && e.StreamAttempt.Action != "begin" { |
| 100 | delete(c.indexes, e.StreamAttempt.ID) |
| 101 | delete(c.attempts, e.StreamAttempt.ID) |
| 102 | } |
| 103 | if e != nil && (e.Kind == "text" || e.Kind == "reasoning") { |
| 104 | c.deltas[e.Kind] += e.Text |
| 105 | } |
| 106 | c.revision = change.Revision |
| 107 | } |
| 108 | return nil |
| 109 | } |
| 110 | |
| 111 | // legacyStreamProvider answers each turn from a script: a streamed reply, a |
| 112 | // tool-call-only round followed by a reply, or a stream that stalls until the |
| 113 | // request is cancelled. |
| 114 | type legacyStreamProvider struct { |
| 115 | mu sync.Mutex |
| 116 | mode string |
| 117 | toolSent bool |
| 118 | } |
| 119 | |
| 120 | func (p *legacyStreamProvider) set(mode string) { |
| 121 | p.mu.Lock() |
| 122 | p.mode, p.toolSent = mode, false |
| 123 | p.mu.Unlock() |
| 124 | } |
| 125 | |
| 126 | func (p *legacyStreamProvider) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 127 | p.mu.Lock() |
| 128 | mode, toolRound := p.mode, p.mode == "tool" && !p.toolSent |
| 129 | if toolRound { |
| 130 | p.toolSent = true |
| 131 | } |
| 132 | p.mu.Unlock() |
| 133 | w.Header().Set("Content-Type", "text/event-stream") |
| 134 | flusher, _ := w.(http.Flusher) |
| 135 | send := func(delta string) { |
| 136 | _, _ = io.WriteString(w, `data: {"choices":[{"delta":{`+delta+`}}]}`+"\n\n") |
| 137 | flusher.Flush() |
| 138 | } |
| 139 | switch { |
| 140 | case toolRound: |
| 141 | send(`"tool_calls":[{"index":0,"id":"call_read","type":"function","function":{"name":"read_file","arguments":"{\"path\":\"note.txt\"}"}}]`) |
| 142 | _, _ = io.WriteString(w, `data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}`+"\n\n") |
| 143 | case mode == "stall": |
| 144 | send(`"reasoning_content":"pondering"`) |
| 145 | <-r.Context().Done() |
| 146 | return |
| 147 | default: |
| 148 | for _, chunk := range []string{`"reasoning_content":"weigh "`, `"reasoning_content":"it"`, `"content":"stream"`, `"content":"ed"`} { |
| 149 | send(chunk) |
| 150 | time.Sleep(20 * time.Millisecond) |
| 151 | } |
| 152 | } |
| 153 | _, _ = io.WriteString(w, "data: [DONE]\n\n") |
| 154 | } |
| 155 | |
| 156 | type legacyFollowTurn struct{ name, mode, reasoning, text string } |
| 157 | |
| 158 | func TestLegacyTopicTranscriptFollowStreamsDeltasTheRendererAccepts(t *testing.T) { |
| 159 | reply := legacyFollowTurn{"streamed reply", "stream", "weigh it", "streamed"} |
| 160 | for _, first := range []legacyFollowTurn{ |
| 161 | reply, |
| 162 | {"tool-call-only round", "tool", "weigh it", "streamed"}, |
| 163 | {"cancelled round", "stall", "pondering", ""}, |
| 164 | } { |
| 165 | t.Run(first.name, func(t *testing.T) { followLegacyTopicTurns(t, first, reply) }) |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | func followLegacyTopicTurns(t *testing.T, turns ...legacyFollowTurn) { |
| 170 | isolateDesktopUserDirs(t) |
| 171 | setDesktopTestCredential(t, "TEST_MODEL_KEY", "sk-test") |
| 172 | script := &legacyStreamProvider{} |
| 173 | stub := httptest.NewServer(script) |
| 174 | t.Cleanup(stub.Close) |
| 175 | cfg := config.Default() |
| 176 | cfg.DefaultModel = "test/test-model" |
| 177 | cfg.Desktop.ProviderAccess = []string{"test"} |
| 178 | cfg.Providers = []config.ProviderEntry{{Name: "test", Kind: "openai", BaseURL: stub.URL, Model: "test-model", APIKeyEnv: "TEST_MODEL_KEY"}} |
| 179 | if err := cfg.SaveTo(config.UserConfigPath()); err != nil { |
| 180 | t.Fatal(err) |
| 181 | } |
| 182 | root := t.TempDir() |
| 183 | if err := os.WriteFile(filepath.Join(root, "note.txt"), []byte("note"), 0o644); err != nil { |
| 184 | t.Fatal(err) |
| 185 | } |
| 186 | if err := addProject(root, "Project"); err != nil { |
| 187 | t.Fatal(err) |
| 188 | } |
| 189 | dir := desktopSessionDir(root) |
| 190 | if err := os.MkdirAll(dir, 0o755); err != nil { |
| 191 | t.Fatal(err) |
| 192 | } |
| 193 | path := writeTopicSession(t, dir, "legacy.jsonl", "legacy-topic", "Legacy", root) |
| 194 | |
| 195 | app := NewApp() |
| 196 | app.ctx = context.Background() |
| 197 | app.readyHook = func() {} |
| 198 | t.Cleanup(func() { app.shutdown(context.Background()) }) |
| 199 | events := newActivationEventRecorder(app) |
| 200 | ticket, err := app.StartTopicActivation(TopicActivationRequest{Scope: "project", WorkspaceRoot: root, TopicID: "legacy-topic", SessionPath: path, RequestID: "open"}) |
| 201 | if err != nil { |
| 202 | t.Fatal(err) |
| 203 | } |
| 204 | if ev := events.waitFor(t, func(ev TopicActivationEvent) bool { |
| 205 | return ev.RequestID == "open" && (ev.Phase == "ready" || ev.Phase == "failed") |
| 206 | }); ev.Phase != "ready" { |
| 207 | t.Fatalf("activation failed: %+v", ev) |
| 208 | } |
| 209 | |
| 210 | base, err := app.TranscriptFollowForTab(ticket.TabID, transcript.FollowRequest{}) |
| 211 | if err != nil { |
| 212 | t.Fatal(err) |
| 213 | } |
| 214 | t.Cleanup(func() { |
| 215 | _, _ = app.TranscriptFollowForTab(ticket.TabID, transcript.FollowRequest{Subscription: base.Subscription, Close: true}) |
| 216 | }) |
| 217 | if base.StorageBackend != "legacy" { |
| 218 | t.Fatalf("fixture must follow a legacy-backed session, got %q", base.StorageBackend) |
| 219 | } |
| 220 | cursor := newRendererFollowCursor(base.Snapshot) |
| 221 | for _, turn := range turns { |
| 222 | script.set(turn.mode) |
| 223 | clear(cursor.deltas) |
| 224 | if err := app.SubmitToTab(ticket.TabID, turn.name); err != nil { |
| 225 | t.Fatal(err) |
| 226 | } |
| 227 | cancelled := false |
| 228 | deadline := time.Now().Add(15 * time.Second) |
| 229 | for done := false; !done; { |
| 230 | if time.Now().After(deadline) { |
| 231 | t.Fatalf("%s did not finish", turn.name) |
| 232 | } |
| 233 | resp, err := app.TranscriptFollowForTab(ticket.TabID, transcript.FollowRequest{Subscription: base.Subscription, AfterRevision: cursor.revision}) |
| 234 | if err != nil { |
| 235 | t.Fatal(err) |
| 236 | } |
| 237 | if resp.ResetRequired { |
| 238 | t.Fatalf("%s: follower was reset", turn.name) |
| 239 | } |
| 240 | if err := cursor.accept(resp.Changes); err != nil { |
| 241 | t.Fatalf("%s: renderer would discard the follow and re-baseline: %v", turn.name, err) |
| 242 | } |
| 243 | for _, change := range resp.Changes { |
| 244 | done = done || change.Event != nil && change.Event.Kind == "turn_done" |
| 245 | } |
| 246 | if turn.mode == "stall" && !cancelled && cursor.deltas["reasoning"] != "" { |
| 247 | app.CancelTab(ticket.TabID) |
| 248 | cancelled = true |
| 249 | } |
| 250 | } |
| 251 | if cursor.deltas["reasoning"] != turn.reasoning || cursor.deltas["text"] != turn.text { |
| 252 | t.Fatalf("%s: renderer received deltas %+v", turn.name, cursor.deltas) |
| 253 | } |
| 254 | } |
| 255 | } |
| 256 |