orchard

  1package main
  2
  3import (
  4	"context"
  5	"encoding/json"
  6	"testing"
  7	"time"
  8)
  9
 10func drain(t *testing.T, ch <-chan json.RawMessage, want int) []json.RawMessage {
 11	t.Helper()
 12	var got []json.RawMessage
 13	for len(got) < want {
 14		select {
 15		case b, ok := <-ch:
 16			if !ok {
 17				return got
 18			}
 19			got = append(got, b)
 20		case <-time.After(2 * time.Second):
 21			return got
 22		}
 23	}
 24	return got
 25}
 26
 27// The whole point: a reader going away does not stop the turn, and the next
 28// reader is handed everything it missed.
 29func TestAReaderLeavingDoesNotStopTheRun(t *testing.T) {
 30	rs := NewRuns()
 31	rn := rs.Start("c1")
 32
 33	_, ch, live := rn.Follow()
 34	if !live {
 35		t.Fatal("a fresh run is not live")
 36	}
 37	rn.Emit(Event{Kind: "status", Text: "thinking"})
 38	if len(drain(t, ch, 1)) != 1 {
 39		t.Fatal("the first reader saw nothing")
 40	}
 41	rn.Unfollow(ch)
 42
 43	// The tab is gone and the turn carries on.
 44	rn.Emit(Event{Kind: "tool", Tool: "web_search"})
 45	rn.Emit(Event{Kind: "tail", Text: "an answer"})
 46
 47	backlog, ch2, live2 := rn.Follow()
 48	if !live2 {
 49		t.Fatal("the run ended when its reader left")
 50	}
 51	if len(backlog) != 3 {
 52		t.Fatalf("the second reader was handed %d events, want all 3", len(backlog))
 53	}
 54	rn.Finish()
 55	if _, ok := <-ch2; ok {
 56		t.Error("finishing did not close the reader's channel")
 57	}
 58}
 59
 60// A tab that comes back after the turn ended gets the whole thing and a closed
 61// channel, rather than an open stream that will never speak.
 62func TestAReaderArrivingAfterTheEndGetsEverything(t *testing.T) {
 63	rs := NewRuns()
 64	rn := rs.Start("c1")
 65	rn.Emit(Event{Kind: "status", Text: "thinking"})
 66	rn.Emit(map[string]any{"kind": "done", "conversation_id": "c1"})
 67	rn.Finish()
 68
 69	backlog, ch, live := rn.Follow()
 70	if live || ch != nil {
 71		t.Error("a finished run handed back a live channel")
 72	}
 73	if len(backlog) != 2 {
 74		t.Fatalf("backlog had %d events, want 2", len(backlog))
 75	}
 76	var last map[string]any
 77	if err := json.Unmarshal(backlog[1], &last); err != nil {
 78		t.Fatal(err)
 79	}
 80	if last["kind"] != "done" {
 81		t.Errorf("the done frame did not survive: %v", last)
 82	}
 83}
 84
 85// A new conversation runs under an id the browser made up, and has to be
 86// findable under the real one once the store hands it over.
 87func TestARunIsFoundUnderItsRealIdAfterRekey(t *testing.T) {
 88	rs := NewRuns()
 89	rn := rs.Start("temp-123")
 90	rn.Emit(Event{Kind: "status", Text: "thinking"})
 91	rs.Rekey("temp-123", "conv-abc")
 92
 93	got, ok := rs.Get("conv-abc")
 94	if !ok || got != rn {
 95		t.Fatal("the run was not found under its real id")
 96	}
 97	if _, ok := rs.Get("temp-123"); ok {
 98		t.Error("the run is still under the made up id as well")
 99	}
100}
101
102// Sending again while a turn is running means the first is no longer wanted,
103// and two generations writing into one conversation is the thing to avoid.
104func TestStartingASecondTurnCancelsTheFirst(t *testing.T) {
105	rs := NewRuns()
106	first := rs.Start("c1")
107	ctx, cancel := context.WithCancel(context.Background())
108	first.setCancel(cancel)
109
110	rs.Start("c1")
111	select {
112	case <-ctx.Done():
113	case <-time.After(2 * time.Second):
114		t.Error("the first turn was left running")
115	}
116}
117
118func TestCancelStopsARun(t *testing.T) {
119	rs := NewRuns()
120	rn := rs.Start("c1")
121	ctx, cancel := context.WithCancel(context.Background())
122	rn.setCancel(cancel)
123
124	if !rs.Cancel("c1") {
125		t.Fatal("cancel did not find the run")
126	}
127	select {
128	case <-ctx.Done():
129	case <-time.After(2 * time.Second):
130		t.Error("cancel did not stop the turn")
131	}
132	if rs.Cancel("nothing-here") {
133		t.Error("cancel claimed to stop a run that does not exist")
134	}
135}
136
137// A finished run is kept a while so a tab can still read it, and dropped after,
138// or the process holds every conversation it ever ran.
139func TestFinishedRunsAreSweptButOnlyWhenStale(t *testing.T) {
140	rs := NewRuns()
141	rn := rs.Start("c1")
142	rs.Start("c2")
143	rn.Finish()
144
145	rs.Sweep(time.Now())
146	if _, ok := rs.Get("c1"); !ok {
147		t.Error("a run that just finished was swept")
148	}
149	rs.Sweep(time.Now().Add(runKept + time.Minute))
150	if _, ok := rs.Get("c1"); ok {
151		t.Error("a stale run was kept")
152	}
153	if _, ok := rs.Get("c2"); !ok {
154		t.Error("a running turn was swept")
155	}
156}
157
158// A reader that stops keeping up is skipped rather than waited on, or one slow
159// tab holds up the turn and every reader behind it.
160func TestASlowReaderDoesNotBlockTheTurn(t *testing.T) {
161	rs := NewRuns()
162	rn := rs.Start("c1")
163	_, _, live := rn.Follow()
164	if !live {
165		t.Fatal("not live")
166	}
167	done := make(chan struct{})
168	go func() {
169		for i := 0; i < 5000; i++ {
170			rn.Emit(Event{Kind: "tail", Text: "x"})
171		}
172		close(done)
173	}()
174	select {
175	case <-done:
176	case <-time.After(5 * time.Second):
177		t.Fatal("the turn blocked on a reader that was not reading")
178	}
179}
180
181// Two turns at once take the card in order rather than interleaving, and the
182// one waiting is told where it is rather than sitting on a spinner.
183func TestASecondTurnWaitsAndIsToldSo(t *testing.T) {
184	q := NewQueue()
185	first, ok := q.Enter(context.Background(), func(QueueState) {})
186	if !ok {
187		t.Fatal("the first turn did not get the card")
188	}
189
190	waits := make(chan QueueState, 8)
191	got := make(chan struct{})
192	go func() {
193		release, ok := q.Enter(context.Background(), func(s QueueState) { waits <- s })
194		if ok {
195			release()
196		}
197		close(got)
198	}()
199
200	select {
201	case s := <-waits:
202		// Ahead counts the others queued in front, so the first waiter behind a
203		// running turn is position 1 with none ahead of it.
204		if s.Position < 1 {
205			t.Errorf("the waiting turn was given position %d", s.Position)
206		}
207		if label := waitingLabel(s); label == "" {
208			t.Error("no label for a waiting turn")
209		}
210	case <-time.After(2 * time.Second):
211		t.Fatal("the second turn was never told it was waiting")
212	}
213
214	select {
215	case <-got:
216		t.Fatal("the second turn ran while the first still held the card")
217	case <-time.After(200 * time.Millisecond):
218	}
219
220	first()
221	select {
222	case <-got:
223	case <-time.After(2 * time.Second):
224		t.Fatal("the second turn never got the card after the first let go")
225	}
226}
227
228func TestWaitingLabelReadsAsWords(t *testing.T) {
229	for _, tc := range []struct {
230		ahead int
231		want  string
232	}{
233		{0, "waiting for the card"},
234		{1, "waiting, one turn ahead"},
235		{3, "waiting, 3 turns ahead"},
236	} {
237		if got := waitingLabel(QueueState{Ahead: tc.ahead}); got != tc.want {
238			t.Errorf("ahead=%d gave %q, want %q", tc.ahead, got, tc.want)
239		}
240	}
241}