diff --git a/docs/offload.md b/docs/offload.md index e6bf7c4..7b2957f 100644 --- a/docs/offload.md +++ b/docs/offload.md @@ -58,6 +58,18 @@ jobs. The caller must be able to ask "is this peer usable right now" without a turn hanging on a timeout. A dead remote is a normal state, not an error state. +`internal/llm.Pair` is that check on the Maven side. A prober caches the answer, +so `Available()` is an atomic read and no turn pays for a health check. + +llama-server does not stay up on the workstation. It cannot: a resident 7-14B +would hold 16GB against the owner's CPT runs. So a supervisor there owns its +lifecycle, keeps it loaded while the card is free, and unloads it on idle or +when another process needs the card (owner's call, 2026-08-02, Vikunja #488). + +That supervisor is still not a scheduler, and the distinction is worth holding. +It arbitrates nothing between callers. It reports whether it can take work and +manages one process to back that answer. Maven never asks it to start anything +and never learns that it did. ## What stays on homesrv, permanently diff --git a/internal/llm/remote.go b/internal/llm/remote.go new file mode 100644 index 0000000..e8d4714 --- /dev/null +++ b/internal/llm/remote.go @@ -0,0 +1,193 @@ +package llm + +import ( + "context" + "errors" + "log" + "net/http" + "sync/atomic" + "time" +) + +// Pair — a preferred model on another host, with the resident one as the floor. +// +// homesrv cannot grow a GPU and the workstation has 16GB of VRAM, so the big +// model runs there and the resident Qwen3-1.7B stays here. See docs/offload.md. +// The workstation is never assumed up: its GPU is often busy with CPT runs and +// the manga-recap pipeline, and the machine sleeps. So the remote is preferred, +// never required, and Pair is what makes "preferred" mean something precise. +// +// This is admission control, not a scheduler. There is no arbiter deciding who +// gets the card. A prober asks the remote whether it will take work, caches the +// answer, and every request reads that cached answer in nanoseconds. Routing +// sits on the hot path at p50 825ms and must never wait on a machine that may +// be asleep, so no request ever pays for a health check itself. +// +// Pair satisfies nothing by itself. Callers pick a method by which half of the +// degradation rule they live under: +// +// - Complete falls back silently. For routing, replies, and nudge phrasing, +// where the big model is only better and the 1.7B is today's shipping +// quality. He is not told which model phrased his reply. +// - CompleteRemote returns ErrRemoteUnavailable instead of falling back. For +// a world question, or a long Kiwix or search passage, where a 1.7B +// confabulates rather than summarises. A named gap beats an invented +// answer. +type Pair struct { + remote *Client + floor *Client + + // up — the cached admission answer, written only by the prober goroutine + // and read by every request. Atomic so the read costs nanoseconds and no + // request ever contends with the prober. + up atomic.Bool + + health string + interval time.Duration + http *http.Client + stop chan struct{} +} + +// ErrRemoteUnavailable — the workstation model was required and is not +// answering. Callers on the naming half of the degradation rule turn this into +// a gap in the reply ("не могу сейчас"), never into a guess from the floor. +var ErrRemoteUnavailable = errors.New("llm: workstation model unavailable") + +// ErrNoFloor — a Pair was built with no resident model to fall back to. A +// configuration mistake: the floor is the whole point. +var ErrNoFloor = errors.New("llm: no floor client") + +// NewPair builds the two-model arrangement. remote may be nil, which is the +// unconfigured deploy and must behave exactly as the box behaves today: every +// call goes to the floor and nothing probes anything. +// +// health is the URL the prober asks. llama-server's /health answers "is a model +// loaded and ready", which is the useful signal here, because llama-server +// refuses to load at all when VRAM is short. That makes a busy card detectable +// without any cooperation from the owner's other jobs. +func NewPair(remote, floor *Client, health string, interval time.Duration) *Pair { + p := &Pair{ + remote: remote, + floor: floor, + health: health, + interval: interval, + http: &http.Client{Timeout: probeTimeout}, + stop: make(chan struct{}), + } + return p +} + +// probeTimeout — a remote that cannot answer /health this fast is not going to +// serve a turn either. Short on purpose: the prober runs on its own goroutine, +// but a slow probe still delays the moment Maven notices the card came back. +const probeTimeout = 2 * time.Second + +// Start begins probing. It returns immediately, and the first probe runs before +// the first tick so a remote that is already up is used on the first turn +// rather than after one interval of falling back. Safe to call with a nil +// remote; it does nothing. +func (p *Pair) Start(ctx context.Context) { + if p.remote == nil || p.health == "" { + return + } + go func() { + p.probe(ctx) + t := time.NewTicker(p.interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-p.stop: + return + case <-t.C: + p.probe(ctx) + } + } + }() +} + +// Stop ends the prober. Idempotent. +func (p *Pair) Stop() { + select { + case <-p.stop: + default: + close(p.stop) + } +} + +// Available reports whether the workstation will take work right now. It reads +// a cached flag, so it is safe to call per turn on the hot path. A false answer +// is never stale in the direction that matters: the worst case is that Maven +// falls back for up to one probe interval after the card frees up. +func (p *Pair) Available() bool { + return p.remote != nil && p.up.Load() +} + +func (p *Pair) probe(ctx context.Context) { + ctx, cancel := context.WithTimeout(ctx, probeTimeout) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.health, nil) + if err != nil { + p.set(false) + return + } + resp, err := p.http.Do(req) + if err != nil { + p.set(false) + return + } + defer resp.Body.Close() + p.set(resp.StatusCode == http.StatusOK) +} + +// set records the admission answer and logs only the transitions. A machine +// that sleeps every night would otherwise write one line per interval forever. +func (p *Pair) set(up bool) { + if p.up.Swap(up) == up { + return + } + if up { + log.Printf("llm: workstation model available at %s", p.health) + } else { + log.Printf("llm: workstation model unavailable, falling back to the resident model") + } +} + +// Complete runs r on the workstation when it will take work, and on the +// resident model otherwise. A remote that fails mid-request falls back too: the +// admission answer is a cache and can be one interval out of date, so an error +// here is expected rather than exceptional. +// +// This is the silent half of the degradation rule. It must be indistinguishable +// from today's behaviour when the workstation is down. +func (p *Pair) Complete(ctx context.Context, r Req) (string, error) { + if p.floor == nil { + return "", ErrNoFloor + } + if p.Available() { + out, err := p.remote.Complete(ctx, r) + if err == nil { + return out, nil + } + // The cached answer was wrong. Correct it now rather than sending the + // next request into the same hole, then fall back. + p.set(false) + } + return p.floor.Complete(ctx, r) +} + +// CompleteRemote runs r on the workstation or refuses. It never falls back, +// because for a world question the resident 1.7B does not answer worse, it +// invents. Callers turn ErrRemoteUnavailable into a named gap. +func (p *Pair) CompleteRemote(ctx context.Context, r Req) (string, error) { + if !p.Available() { + return "", ErrRemoteUnavailable + } + out, err := p.remote.Complete(ctx, r) + if err != nil { + p.set(false) + return "", errors.Join(ErrRemoteUnavailable, err) + } + return out, nil +} diff --git a/internal/llm/remote_test.go b/internal/llm/remote_test.go new file mode 100644 index 0000000..6749fdd --- /dev/null +++ b/internal/llm/remote_test.go @@ -0,0 +1,210 @@ +package llm + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +// completionServer stands in for a llama-server. It counts what reached it, so +// a test can say which of the two models answered. +func completionServer(t *testing.T, reply string, hits *atomic.Int64) *httptest.Server { + t.Helper() + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"choices":[{"message":{"content":"` + reply + `"}}]}`)) + })) + t.Cleanup(s.Close) + return s +} + +func healthServer(t *testing.T, ok *atomic.Bool) *httptest.Server { + t.Helper() + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !ok.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(s.Close) + return s +} + +// waitFor polls until cond holds or the deadline passes. The prober runs on its +// own goroutine, so a test has to wait for it rather than assume it has run. +func waitFor(t *testing.T, cond func() bool) bool { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return true + } + time.Sleep(5 * time.Millisecond) + } + return false +} + +// The unconfigured deploy. No remote, no probing, every call to the floor — +// exactly what the box does today. +func TestNoRemoteGoesToTheFloor(t *testing.T) { + var floorHits atomic.Int64 + floor := completionServer(t, "floor", &floorHits) + + p := NewPair(nil, New(floor.URL, time.Second), "", time.Second) + p.Start(context.Background()) + defer p.Stop() + + if p.Available() { + t.Fatal("a Pair with no remote reports available") + } + out, err := p.Complete(context.Background(), Req{User: "привет"}) + if err != nil { + t.Fatalf("complete: %v", err) + } + if out != "floor" || floorHits.Load() != 1 { + t.Fatalf("out = %q, floor hits = %d", out, floorHits.Load()) + } +} + +// The workstation is up, so it answers and the resident model is not touched. +func TestAvailableRemoteAnswers(t *testing.T) { + var remoteHits, floorHits atomic.Int64 + remote := completionServer(t, "remote", &remoteHits) + floor := completionServer(t, "floor", &floorHits) + up := &atomic.Bool{} + up.Store(true) + health := healthServer(t, up) + + p := NewPair(New(remote.URL, time.Second), New(floor.URL, time.Second), health.URL, 20*time.Millisecond) + p.Start(context.Background()) + defer p.Stop() + if !waitFor(t, p.Available) { + t.Fatal("prober never saw the remote come up") + } + + out, err := p.Complete(context.Background(), Req{User: "привет"}) + if err != nil { + t.Fatalf("complete: %v", err) + } + if out != "remote" || floorHits.Load() != 0 { + t.Fatalf("out = %q, floor hits = %d", out, floorHits.Load()) + } +} + +// The card is busy, so /health refuses and Complete degrades silently. This is +// the constraint from 483: the workstation being down is indistinguishable from +// today's behaviour. +func TestBusyCardFallsBackSilently(t *testing.T) { + var remoteHits, floorHits atomic.Int64 + remote := completionServer(t, "remote", &remoteHits) + floor := completionServer(t, "floor", &floorHits) + health := healthServer(t, &atomic.Bool{}) // never ok + + p := NewPair(New(remote.URL, time.Second), New(floor.URL, time.Second), health.URL, 20*time.Millisecond) + p.Start(context.Background()) + defer p.Stop() + time.Sleep(60 * time.Millisecond) + + out, err := p.Complete(context.Background(), Req{User: "привет"}) + if err != nil { + t.Fatalf("complete: %v", err) + } + if out != "floor" || remoteHits.Load() != 0 { + t.Fatalf("out = %q, remote hits = %d", out, remoteHits.Load()) + } +} + +// The cached admission answer can be one interval out of date, so a remote that +// dies between probes must still not break the turn. +func TestRemoteErrorMidRequestFallsBack(t *testing.T) { + var floorHits atomic.Int64 + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer dead.Close() + floor := completionServer(t, "floor", &floorHits) + up := &atomic.Bool{} + up.Store(true) + health := healthServer(t, up) + + p := NewPair(New(dead.URL, time.Second), New(floor.URL, time.Second), health.URL, time.Hour) + p.Start(context.Background()) + defer p.Stop() + if !waitFor(t, p.Available) { + t.Fatal("prober never saw the remote come up") + } + + out, err := p.Complete(context.Background(), Req{User: "привет"}) + if err != nil { + t.Fatalf("complete: %v", err) + } + if out != "floor" || floorHits.Load() != 1 { + t.Fatalf("out = %q, floor hits = %d", out, floorHits.Load()) + } + // The failed request must have corrected the cached answer, so the next + // one does not walk into the same hole. + if p.Available() { + t.Fatal("a failed remote request left the admission answer up") + } +} + +// The naming half of the degradation rule. A world question must not be handed +// to the resident model, because it answers by inventing. +func TestCompleteRemoteNamesTheGap(t *testing.T) { + var floorHits atomic.Int64 + floor := completionServer(t, "floor", &floorHits) + health := healthServer(t, &atomic.Bool{}) // never ok + + p := NewPair(New("http://127.0.0.1:1", time.Second), New(floor.URL, time.Second), health.URL, 20*time.Millisecond) + p.Start(context.Background()) + defer p.Stop() + time.Sleep(60 * time.Millisecond) + + if _, err := p.CompleteRemote(context.Background(), Req{User: "почему небо голубое"}); !errors.Is(err, ErrRemoteUnavailable) { + t.Fatalf("err = %v, want ErrRemoteUnavailable", err) + } + if floorHits.Load() != 0 { + t.Fatalf("CompleteRemote fell back to the floor %d times", floorHits.Load()) + } +} + +// Routing sits on the hot path and must never pay for a health check. Available +// reads a cached flag, so it costs no network at all. +func TestAvailableDoesNotProbe(t *testing.T) { + var probes atomic.Int64 + health := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + probes.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer health.Close() + + p := NewPair(New("http://127.0.0.1:1", time.Second), New("http://127.0.0.1:1", time.Second), health.URL, time.Hour) + p.Start(context.Background()) + defer p.Stop() + if !waitFor(t, p.Available) { + t.Fatal("prober never ran") + } + + before := probes.Load() + for range 1000 { + p.Available() + } + if got := probes.Load(); got != before { + t.Fatalf("1000 Available calls made %d probes", got-before) + } +} + +// A Pair with no floor is a configuration mistake, and it must say so rather +// than silently having nowhere to degrade to. +func TestNoFloorIsAnError(t *testing.T) { + p := NewPair(nil, nil, "", time.Second) + if _, err := p.Complete(context.Background(), Req{User: "привет"}); !errors.Is(err, ErrNoFloor) { + t.Fatalf("err = %v, want ErrNoFloor", err) + } +}