session_concurrency_test.go RawHistoryFolder

1package core
2
3import (
4	"bufio"
5	"encoding/json"
6	"os"
7	"strings"
8	"sync"
9	"testing"
10
11	"github.com/bnema/zut/packages/provider"
12)
13
14func TestSessionConcurrentAppendsRemainValidJSONL(t *testing.T) {
15	session, err := NewSession(t.TempDir(), "/workspace", "provider", "model", "test")
16	if err != nil {
17		t.Fatal(err)
18	}
19
20	const (
21		writers = 8
22		rows    = 20
23	)
24	payload := strings.Repeat("x", 8*1024)
25	start := make(chan struct{})
26	errs := make(chan error, writers)
27	var wg sync.WaitGroup
28	for writer := 0; writer < writers; writer++ {
29		wg.Add(1)
30		go func() {
31			defer wg.Done()
32			<-start
33			for i := 0; i < rows; i++ {
34				if err := session.AppendMessage(provider.Message{
35					Role:    provider.RoleAssistant,
36					Content: []provider.Content{provider.TextBlock{Text: payload}},
37				}); err != nil {
38					errs <- err
39					return
40				}
41			}
42		}()
43	}
44	close(start)
45	wg.Wait()
46	close(errs)
47	for err := range errs {
48		t.Errorf("append message: %v", err)
49	}
50	if err := session.Close(); err != nil {
51		t.Fatal(err)
52	}
53
54	file, err := os.Open(session.Path)
55	if err != nil {
56		t.Fatal(err)
57	}
58	defer file.Close()
59
60	lineCount := 0
61	scanner := bufio.NewScanner(file)
62	scanner.Buffer(make([]byte, 1024), 32*1024)
63	for scanner.Scan() {
64		lineCount++
65		if !json.Valid(scanner.Bytes()) {
66			t.Fatalf("line %d is not valid JSON", lineCount)
67		}
68	}
69	if err := scanner.Err(); err != nil {
70		t.Fatal(err)
71	}
72	if want := 1 + writers*rows; lineCount != want {
73		t.Fatalf("line count = %d, want %d", lineCount, want)
74	}
75}