diff --git a/docs/plans/v1/README.md b/docs/plans/v1/README.md index ce5f024..7b0973c 100644 --- a/docs/plans/v1/README.md +++ b/docs/plans/v1/README.md @@ -102,3 +102,9 @@ tools/run-plan.sh docs/plans/v1 # from a clean checkout on master known Ornith failure mode (I9), here triggered by a denied tool. Model fault; task 05 now says a refusal is not a reason to stop. The tee's token extraction (`TestAccountingRows…`) was the genuinely unfinished part. Resumed from the working tree. +- 2026-09-25, task 05, resume session: my given `proxy_test.go` was 433 lines, over the gate's + 400-line limit that `scripts/check-lines.sh` applies to every `.go` file including the copied + test and the plan copy under `docs/`. Ornith found it and went digging in git history instead + of stopping. Test-file fault (mine): the rig and fake-upstream scaffolding moved into + `helpers_test.go` (211 + 271 lines). Ornith's own `proxy.go` was also at 411 lines; splitting it + is part of the task as written. diff --git a/docs/plans/v1/_files/internal/proxy/helpers_test.go b/docs/plans/v1/_files/internal/proxy/helpers_test.go index 18b157c..e1c8807 100644 --- a/docs/plans/v1/_files/internal/proxy/helpers_test.go +++ b/docs/plans/v1/_files/internal/proxy/helpers_test.go @@ -1,9 +1,27 @@ package proxy_test -import ( - "sync" +// Test scaffolding shared by proxy_test.go and recorder_test.go: the fake health table, the fake +// llama-server upstream, and the rig that builds a whole crossbar over real HTTP. +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "git.wntrmute.dev/kyle/crossbar/internal/config" "git.wntrmute.dev/kyle/crossbar/internal/health" + "git.wntrmute.dev/kyle/crossbar/internal/lease" + "git.wntrmute.dev/kyle/crossbar/internal/limiter" + "git.wntrmute.dev/kyle/crossbar/internal/proxy" + "git.wntrmute.dev/kyle/crossbar/internal/store" ) // fakeHealth is a hand-set health table that also records MarkDown calls. It lived in the v0 @@ -37,3 +55,157 @@ func (f *fakeHealth) markedHosts() []string { defer f.mu.Unlock() return append([]string{}, f.marked...) } + +// upstream is a llama-server stand-in: streams N chunks with a delay, reports usage/timings in +// the final chunk, counts requests, and can be slowed down or killed. +type upstream struct { + name string + srv *httptest.Server + hits atomic.Int32 + delay time.Duration + mu sync.Mutex + last recorded +} + +type recorded struct{ method, path, host, xff, body string } + +func newUpstream(t *testing.T, name string) *upstream { + u := &upstream{name: name} + mux := http.NewServeMux() + mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{"status":"ok"}`) }) + mux.HandleFunc("/v1/models", func(w http.ResponseWriter, r *http.Request) { + u.mu.Lock() + u.last = recorded{r.Method, r.URL.RequestURI(), r.Host, r.Header.Get("X-Forwarded-For"), ""} + u.mu.Unlock() + fmt.Fprint(w, `{"object":"list","data":[{"id":"shared"},{"id":"`+name+`-only"}]}`) + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + u.hits.Add(1) + b, _ := io.ReadAll(r.Body) + u.mu.Lock() + u.last = recorded{r.Method, r.URL.RequestURI(), r.Host, r.Header.Get("X-Forwarded-For"), string(b)} + u.mu.Unlock() + var req struct { + Stream bool `json:"stream"` + } + _ = json.Unmarshal(b, &req) + w.Header().Set("X-Upstream", name) + time.Sleep(u.delay) + if !req.Stream { + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, `{"choices":[{"message":{"role":"assistant","content":"hi from %s"}}],"usage":{"prompt_tokens":100,"completion_tokens":10,"total_tokens":110},"timings":{"prompt_n":100,"cache_n":90,"predicted_n":10,"predicted_ms":50.0}}`, name) + return + } + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(200) + fl := w.(http.Flusher) + for i := 0; i < 3; i++ { + fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"content\":\"%s %d \"}}]}\n\n", name, i) + fl.Flush() + time.Sleep(10 * time.Millisecond) + } + fmt.Fprint(w, `data: {"choices":[],"usage":{"prompt_tokens":200,"completion_tokens":20,"total_tokens":220},"timings":{"prompt_n":200,"cache_n":150,"predicted_n":20,"predicted_ms":80.0}}`+"\n\n") + fl.Flush() + fmt.Fprint(w, "data: [DONE]\n\n") + }) + u.srv = httptest.NewServer(mux) + t.Cleanup(u.srv.Close) + return u +} + +func (u *upstream) lastReq() recorded { u.mu.Lock(); defer u.mu.Unlock(); return u.last } + +// rig is one crossbar: config, real health table (polled once), real lease table over a real +// SQLite store, real limiter, the proxy handler served by httptest. +type rig struct { + t *testing.T + cfg *config.Config + health *health.Table + store *store.Store + leases *lease.Table + lim *limiter.Limiter + front *httptest.Server +} + +// newRig builds crossbar from a config text where %s placeholders are the upstream base URLs. +func newRig(t *testing.T, cfgText string, ups ...*upstream) *rig { + urls := make([]any, len(ups)) + for i, u := range ups { + urls[i] = u.srv.URL + } + cfg, err := config.Parse(strings.NewReader(fmt.Sprintf(cfgText, urls...))) + if err != nil { + t.Fatal(err) + } + bases := map[string]string{} + for name, h := range cfg.Hosts { + bases[name] = h.BaseURL + } + ht := health.New(bases, time.Hour, nil) + ht.PollOnce(t.Context()) + st, err := store.Open(filepath.Join(t.TempDir(), "crossbar.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + lim := limiter.New() + for name, h := range cfg.Hosts { + for model, m := range h.Models { + lim.Configure(name, model, m.Parallel, cfg.QueueMax) + } + } + lt, err := lease.New(st, proxy.HostView(ht, cfg), proxy.Chooser(cfg, ht, lim), cfg.LeaseIdle.Duration) + if err != nil { + t.Fatal(err) + } + p := proxy.New(cfg, ht, lt, lim, st, nil) + front := httptest.NewServer(p) + t.Cleanup(front.Close) + return &rig{t: t, cfg: cfg, health: ht, store: st, leases: lt, lim: lim, front: front} +} + +const twoHosts = ` +listen = "127.0.0.1:1" +queue_max = 1 +lease_idle = "30m" +[hosts.alpha] +base_url = %q +weight = 1.0 +models = { "shared" = { parallel = 2 }, "alpha-only" = { } } +[hosts.beta] +base_url = %q +weight = 2.0 +models = { "shared" = { parallel = 2 }, "beta-only" = { } } +[routes.r] +hosts = ["alpha", "beta"] +default_model = "shared" +[routes.other] +hosts = ["alpha"] +` + +func conversation(id, turn int) string { + msgs := fmt.Sprintf(`{"role":"system","content":"project"},{"role":"user","content":"conversation %d opening"}`, id) + for i := 1; i < turn; i++ { + msgs += fmt.Sprintf(`,{"role":"assistant","content":"ok"},{"role":"user","content":"turn %d"}`, i) + } + return `{"model":"shared","stream":false,"messages":[` + msgs + `]}` +} + +func (r *rig) post(path, body string, hdr ...string) *http.Response { + req, _ := http.NewRequest(http.MethodPost, r.front.URL+path, strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + for i := 0; i+1 < len(hdr); i += 2 { + req.Header.Set(hdr[i], hdr[i+1]) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + r.t.Fatal(err) + } + return resp +} + +func drain(resp *http.Response) string { + b, _ := io.ReadAll(resp.Body) + resp.Body.Close() + return string(b) +} diff --git a/docs/plans/v1/_files/internal/proxy/proxy_test.go b/docs/plans/v1/_files/internal/proxy/proxy_test.go index 560eab8..c1bb26a 100644 --- a/docs/plans/v1/_files/internal/proxy/proxy_test.go +++ b/docs/plans/v1/_files/internal/proxy/proxy_test.go @@ -3,182 +3,20 @@ package proxy_test // v1 acceptance tests for the proxy: leases, queueing, accounting, header route override. // They drive the whole handler over real HTTP against fake upstreams; only what a client or an // operator can observe is asserted (status codes, headers, the accounting rows, the health table). +// The rig, the fake upstream and the request helpers live in helpers_test.go. import ( "encoding/json" - "fmt" - "io" "net/http" - "net/http/httptest" - "path/filepath" "strings" "sync" - "sync/atomic" "testing" "time" - "git.wntrmute.dev/kyle/crossbar/internal/config" - "git.wntrmute.dev/kyle/crossbar/internal/health" - "git.wntrmute.dev/kyle/crossbar/internal/lease" - "git.wntrmute.dev/kyle/crossbar/internal/limiter" "git.wntrmute.dev/kyle/crossbar/internal/proxy" "git.wntrmute.dev/kyle/crossbar/internal/store" ) -// upstream is a llama-server stand-in: streams N chunks with a delay, reports usage/timings in -// the final chunk, counts requests, and can be slowed down or killed. -type upstream struct { - name string - srv *httptest.Server - hits atomic.Int32 - delay time.Duration - mu sync.Mutex - last recorded -} - -type recorded struct{ method, path, host, xff, body string } - -func newUpstream(t *testing.T, name string) *upstream { - u := &upstream{name: name} - mux := http.NewServeMux() - mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{"status":"ok"}`) }) - mux.HandleFunc("/v1/models", func(w http.ResponseWriter, r *http.Request) { - u.mu.Lock() - u.last = recorded{r.Method, r.URL.RequestURI(), r.Host, r.Header.Get("X-Forwarded-For"), ""} - u.mu.Unlock() - fmt.Fprint(w, `{"object":"list","data":[{"id":"shared"},{"id":"`+name+`-only"}]}`) - }) - mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { - u.hits.Add(1) - b, _ := io.ReadAll(r.Body) - u.mu.Lock() - u.last = recorded{r.Method, r.URL.RequestURI(), r.Host, r.Header.Get("X-Forwarded-For"), string(b)} - u.mu.Unlock() - var req struct { - Stream bool `json:"stream"` - } - _ = json.Unmarshal(b, &req) - w.Header().Set("X-Upstream", name) - time.Sleep(u.delay) - if !req.Stream { - w.Header().Set("Content-Type", "application/json") - fmt.Fprintf(w, `{"choices":[{"message":{"role":"assistant","content":"hi from %s"}}],"usage":{"prompt_tokens":100,"completion_tokens":10,"total_tokens":110},"timings":{"prompt_n":100,"cache_n":90,"predicted_n":10,"predicted_ms":50.0}}`, name) - return - } - w.Header().Set("Content-Type", "text/event-stream") - w.WriteHeader(200) - fl := w.(http.Flusher) - for i := 0; i < 3; i++ { - fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"content\":\"%s %d \"}}]}\n\n", name, i) - fl.Flush() - time.Sleep(10 * time.Millisecond) - } - fmt.Fprint(w, `data: {"choices":[],"usage":{"prompt_tokens":200,"completion_tokens":20,"total_tokens":220},"timings":{"prompt_n":200,"cache_n":150,"predicted_n":20,"predicted_ms":80.0}}`+"\n\n") - fl.Flush() - fmt.Fprint(w, "data: [DONE]\n\n") - }) - u.srv = httptest.NewServer(mux) - t.Cleanup(u.srv.Close) - return u -} - -func (u *upstream) lastReq() recorded { u.mu.Lock(); defer u.mu.Unlock(); return u.last } - -// rig is one crossbar: config, real health table (polled once), real lease table over a real -// SQLite store, real limiter, the proxy handler served by httptest. -type rig struct { - t *testing.T - cfg *config.Config - health *health.Table - store *store.Store - leases *lease.Table - lim *limiter.Limiter - front *httptest.Server -} - -// newRig builds crossbar from a config text where %s placeholders are the upstream base URLs. -func newRig(t *testing.T, cfgText string, ups ...*upstream) *rig { - urls := make([]any, len(ups)) - for i, u := range ups { - urls[i] = u.srv.URL - } - cfg, err := config.Parse(strings.NewReader(fmt.Sprintf(cfgText, urls...))) - if err != nil { - t.Fatal(err) - } - bases := map[string]string{} - for name, h := range cfg.Hosts { - bases[name] = h.BaseURL - } - ht := health.New(bases, time.Hour, nil) - ht.PollOnce(t.Context()) - st, err := store.Open(filepath.Join(t.TempDir(), "crossbar.db")) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = st.Close() }) - lim := limiter.New() - for name, h := range cfg.Hosts { - for model, m := range h.Models { - lim.Configure(name, model, m.Parallel, cfg.QueueMax) - } - } - lt, err := lease.New(st, proxy.HostView(ht, cfg), proxy.Chooser(cfg, ht, lim), cfg.LeaseIdle.Duration) - if err != nil { - t.Fatal(err) - } - p := proxy.New(cfg, ht, lt, lim, st, nil) - front := httptest.NewServer(p) - t.Cleanup(front.Close) - return &rig{t: t, cfg: cfg, health: ht, store: st, leases: lt, lim: lim, front: front} -} - -const twoHosts = ` -listen = "127.0.0.1:1" -queue_max = 1 -lease_idle = "30m" -[hosts.alpha] -base_url = %q -weight = 1.0 -models = { "shared" = { parallel = 2 }, "alpha-only" = { } } -[hosts.beta] -base_url = %q -weight = 2.0 -models = { "shared" = { parallel = 2 }, "beta-only" = { } } -[routes.r] -hosts = ["alpha", "beta"] -default_model = "shared" -[routes.other] -hosts = ["alpha"] -` - -func conversation(id, turn int) string { - msgs := fmt.Sprintf(`{"role":"system","content":"project"},{"role":"user","content":"conversation %d opening"}`, id) - for i := 1; i < turn; i++ { - msgs += fmt.Sprintf(`,{"role":"assistant","content":"ok"},{"role":"user","content":"turn %d"}`, i) - } - return `{"model":"shared","stream":false,"messages":[` + msgs + `]}` -} - -func (r *rig) post(path, body string, hdr ...string) *http.Response { - req, _ := http.NewRequest(http.MethodPost, r.front.URL+path, strings.NewReader(body)) - req.Header.Set("Content-Type", "application/json") - for i := 0; i+1 < len(hdr); i += 2 { - req.Header.Set(hdr[i], hdr[i+1]) - } - resp, err := http.DefaultClient.Do(req) - if err != nil { - r.t.Fatal(err) - } - return resp -} - -func drain(resp *http.Response) string { - b, _ := io.ReadAll(resp.Body) - resp.Body.Close() - return string(b) -} - func TestConversationIsStickyAndLeaseHeaderTellsWhy(t *testing.T) { alpha, beta := newUpstream(t, "alpha"), newUpstream(t, "beta") r := newRig(t, twoHosts, alpha, beta)