feat(kb): brain tools, Go async serve, root-level docker

- tools/kblib.py: ladybug schema, embeddings, FTS+vector, hybrid RRF - bin/kb/{index,search,get,stats,eval}: corpus indexing + deduction search - bin/facts/{extract,audit}: 2-source evidence acquisition + gates - serve/: async Go HTTP server (goroutines, bounded worker pool), TDD - docker/ flattened to root: compose.yaml + Dockerfile (multi-stage Go) - docker scripts -> bin/ shebang pattern (kb-watch, docker-entrypoint) - bin/ci/semver + tools/semver.py: conventional-commit semver release - ci.yml: go tests + shell checks; drop release-please (PR toggle blocked)
2026-08-10 21:58:25 +01:00
parent dcb00808c9
commit 4fdc0ef4a8
29 changed files with 1560 additions and 134 deletions
@@ -0,0 +1,3 @@
+module github.com/eSlider/2dph/serve
+
+go 1.25
@@ -0,0 +1,143 @@
+// Package main serves the 2dph brain over HTTP.
+//
+// Async by design: every request runs on its own goroutine, and CPU-heavy
+// searches are serialized through a bounded worker pool (a counting
+// semaphore) so N requests can't spawn N Python interpreters at once.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"log"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type Searcher interface {
+	Search(ctx context.Context, query string, limit int) ([]byte, error)
+}
+
+type Server struct {
+	searcher  Searcher
+	semaphore chan struct{}
+}
+
+const defaultPort = 8630
+
+func NewServer(searcher Searcher, workers int) http.Handler {
+	return &Server{
+		searcher:  searcher,
+		semaphore: make(chan struct{}, workers),
+	}
+}
+
+func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	switch {
+	case r.URL.Path == "/health":
+		writeJSON(w, http.StatusOK, map[string]any{"status": "ok"})
+	case r.URL.Path == "/search":
+		s.handleSearch(w, r)
+	default:
+		writeJSON(w, http.StatusNotFound, map[string]any{"error": "not found"})
+	}
+}
+
+func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request) {
+	q := strings.TrimSpace(r.URL.Query().Get("q"))
+	if q == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]any{"error": "q required"})
+		return
+	}
+	limit := 10
+	if raw := r.URL.Query().Get("n"); raw != "" {
+		n, err := strconv.Atoi(raw)
+		if err != nil || n < 1 || n > 100 {
+			writeJSON(w, http.StatusBadRequest, map[string]any{"error": "n must be int 1..100"})
+			return
+		}
+		limit = n
+	}
+
+	// Worker pool: block until a slot frees, so burst concurrency still
+	// bounds memory (no unbounded python processes).
+	select {
+	case s.semaphore <- struct{}{}:
+		defer func() { <-s.semaphore }()
+	case <-r.Context().Done():
+		return
+	}
+
+	body, err := s.searcher.Search(r.Context(), q, limit)
+	if err != nil {
+		writeJSON(w, http.StatusGatewayTimeout, map[string]any{"error": err.Error()})
+		return
+	}
+	writeRaw(w, http.StatusOK, body)
+}
+
+func writeJSON(w http.ResponseWriter, code int, obj any) {
+	body, _ := json.Marshal(obj)
+	writeRaw(w, code, body)
+}
+
+func writeRaw(w http.ResponseWriter, code int, body []byte) {
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("Content-Length", strconv.Itoa(len(body)))
+	w.WriteHeader(code)
+	w.Write(body)
+}
+
+// brainSearcher shells out to bin/kb/search --json. A single python search
+// is bounded and short-lived; the worker pool keeps at most N live.
+type brainSearcher struct {
+	cmdPath string
+	timeout time.Duration
+}
+
+func (b *brainSearcher) Search(ctx context.Context, query string, limit int) ([]byte, error) {
+	ctx, cancel := context.WithTimeout(ctx, b.timeout)
+	defer cancel()
+	cmd := exec.CommandContext(ctx, b.cmdPath, "--json", "-n", strconv.Itoa(limit), query)
+	out, err := cmd.Output()
+	if err != nil {
+		var exitErr *exec.ExitError
+		if errors.As(err, &exitErr) {
+			return nil, errors.New("search backend failed: " + strings.TrimSpace(string(exitErr.Stderr)))
+		}
+		return nil, err
+	}
+	return out, nil
+}
+
+func main() {
+	searchPath := os.Getenv("KB_SEARCH_CMD")
+	if searchPath == "" {
+		searchPath = filepath.Join("bin", "kb", "search")
+	}
+	workers := 4
+	if raw := os.Getenv("KB_WORKERS"); raw != "" {
+		if n, err := strconv.Atoi(raw); err == nil && n > 0 {
+			workers = n
+		}
+	}
+	port := defaultPort
+	if raw := os.Getenv("KB_PORT"); raw != "" {
+		if n, err := strconv.Atoi(raw); err == nil && n > 0 {
+			port = n
+		}
+	}
+
+	searcher := &brainSearcher{cmdPath: searchPath, timeout: 60 * time.Second}
+	handler := NewServer(searcher, workers)
+	addr := "127.0.0.1:" + strconv.Itoa(port)
+	log.Printf("serve: %s (workers=%d)", addr, workers)
+	if err := http.ListenAndServe(addr, handler); err != nil {
+		log.Fatal(err)
+	}
+}
@@ -0,0 +1,177 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// fakeSearcher is an injectable Searcher for tests (no python involved).
+type fakeSearcher struct {
+	mu       sync.Mutex
+	delay    time.Duration
+	calls    int
+	active   atomic.Int32
+	maxSeen  atomic.Int32
+	callback func(q string, limit int) ([]byte, error)
+}
+
+func (f *fakeSearcher) Search(ctx context.Context, query string, limit int) ([]byte, error) {
+	f.mu.Lock()
+	f.calls++
+	f.mu.Unlock()
+	n := f.active.Add(1)
+	for {
+		old := f.maxSeen.Load()
+		if n <= old || f.maxSeen.CompareAndSwap(old, n) {
+			break
+		}
+	}
+	defer f.active.Add(-1)
+	if f.delay > 0 {
+		select {
+		case <-time.After(f.delay):
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+	if f.callback != nil {
+		return f.callback(query, limit)
+	}
+	return []byte(`{"query":"` + query + `","count":0,"results":[]}`), nil
+}
+
+func (f *fakeSearcher) count() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.calls
+}
+
+func newTestServer(s Searcher, workers int) http.Handler {
+	return NewServer(s, workers)
+}
+
+func get(t *testing.T, h http.Handler, path string) (int, []byte) {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodGet, path, nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	return rec.Code, rec.Body.Bytes()
+}
+
+func TestHealth(t *testing.T) {
+	h := newTestServer(&fakeSearcher{}, 1)
+	code, body := get(t, h, "/health")
+	if code != http.StatusOK {
+		t.Fatalf("health code = %d, want 200", code)
+	}
+	var out map[string]any
+	if err := json.Unmarshal(body, &out); err != nil {
+		t.Fatalf("health body not json: %v (%s)", err, body)
+	}
+	if out["status"] != "ok" {
+		t.Fatalf("health status = %v, want ok", out["status"])
+	}
+}
+
+func TestSearchMissingQuery(t *testing.T) {
+	h := newTestServer(&fakeSearcher{}, 1)
+	if code, _ := get(t, h, "/search"); code != http.StatusBadRequest {
+		t.Fatalf("code = %d, want 400", code)
+	}
+}
+
+func TestSearchReturnsSearcherResult(t *testing.T) {
+	fs := &fakeSearcher{callback: func(q string, limit int) ([]byte, error) {
+		return []byte(`{"query":"` + q + `","count":1,"results":[{"id":"x"}]}`), nil
+	}}
+	h := newTestServer(fs, 1)
+	code, body := get(t, h, "/search?q=matrix")
+	if code != http.StatusOK {
+		t.Fatalf("code = %d, want 200", code)
+	}
+	var out map[string]any
+	if err := json.Unmarshal(body, &out); err != nil {
+		t.Fatalf("body not json: %v (%s)", err, body)
+	}
+	if out["query"] != "matrix" {
+		t.Fatalf("query = %v, want matrix", out["query"])
+	}
+	if fs.count() != 1 {
+		t.Fatalf("searcher called %d times, want 1", fs.count())
+	}
+}
+
+func TestSearchConcurrencyBounded(t *testing.T) {
+	// 8 parallel requests on a 3-worker pool: at most 3 concurrent searches.
+	fs := &fakeSearcher{delay: 20 * time.Millisecond}
+	h := newTestServer(fs, 3)
+
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			req := httptest.NewRequest(http.MethodGet, "/search?q=abc", nil)
+			rec := httptest.NewRecorder()
+			h.ServeHTTP(rec, req)
+			if rec.Code != http.StatusOK {
+				t.Errorf("code = %d, want 200", rec.Code)
+			}
+		}()
+	}
+	wg.Wait()
+
+	if calls := fs.count(); calls != 8 {
+		t.Fatalf("searcher called %d times, want 8", calls)
+	}
+	if max := fs.maxSeen.Load(); max > 3 {
+		t.Fatalf("max concurrent = %d, want <= 3", max)
+	}
+	if max := fs.maxSeen.Load(); max < 1 {
+		t.Fatalf("max concurrent = %d, want >= 1", max)
+	}
+}
+
+func TestSearchRejectsBadLimit(t *testing.T) {
+	h := newTestServer(&fakeSearcher{}, 1)
+	if code, _ := get(t, h, "/search?q=x&n=hundred"); code != http.StatusBadRequest {
+		t.Fatalf("code = %d, want 400", code)
+	}
+}
+
+func TestSearchTimeout(t *testing.T) {
+	fs := &fakeSearcher{delay: time.Second}
+	h := NewServer(fs, 1)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
+	defer cancel()
+	req := httptest.NewRequest(http.MethodGet, "/search?q=slow", nil).WithContext(ctx)
+	rec := httptest.NewRecorder()
+
+	done := make(chan struct{})
+	go func() {
+		h.ServeHTTP(rec, req)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// ServeHTTP returned; body should be an error json (we don't require a
+		// specific code for the pathological ctx-cancel timing, only that it
+		// does not hang forever).
+		return
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("request hung after context cancellation")
+	}
+	_ = io.Discard
+	_ = bytes.MinRead
+	_ = fmt.Sprintf
+}