root 9ee7fc5550 G2: embedd — text → vector via Ollama · 2 scrum fixes
Bridges the missing piece for the staffing co-pilot: text inputs to
vectord-shaped vectors. Standalone cmd/embedd on :3216 fronted by
gateway at /v1/embed. Pluggable embed.Provider interface (G2 ships
Ollama; OpenAI/Voyage swap in via the same interface in G3+).

Wire format:
  POST /v1/embed {"texts":[...], "model":"..."}  // model optional
  → 200 {"model","dimension","vectors":[[...]]}

Default model: nomic-embed-text (768-d). Ollama returns float64;
provider converts to float32 at the boundary so vectors flow through
vectord/HNSW without re-conversion.

Acceptance smoke 5/5 PASS — including the architectural payoff:
end-to-end embed → vectord add → search by re-embedded text returns
recall=1 at distance 5.96e-8 (float32 precision noise on identical
unit vectors). The staffing co-pilot pipeline (text → vector →
similarity search) is now functional end-to-end.

All 9 smokes (D1-D6 + G1 + G1P + G2) PASS deterministically.

Cross-lineage scrum on shipped code:
  - Opus 4.7 (opencode):                    0 BLOCK + 4 WARN + 3 INFO
  - Kimi K2-0905 (openrouter):              0 BLOCK + 2 WARN + 1 INFO
  - Qwen3-coder (openrouter):               "No BLOCKs" (3 tokens)

Fixed (2 — 1 convergent + 1 single-reviewer):
  C1 (Opus + Kimi convergent WARN): per-text 60s timeout × N-text
    batch was up to N×60s with no batch-level cap. One stuck Ollama
    call would stall the whole handler indefinitely. Fix:
    context.WithTimeout(r.Context(), 60s) wraps the entire batch.
  O-W3 (Opus WARN): empty strings in texts went to Ollama unchecked,
    producing version-dependent garbage. Fix: reject "" with 400 at
    the handler boundary so callers get a deterministic answer
    instead of an upstream-conditional 502.

Deferred (4): drainAndClose 64KiB cap (matches G0 pattern), no
concurrency limit on /embed (single-tenant G2), missing Accept
header (exotic-proxy concern), MaxBytesError string-match
redundancy (paranoia layer kept consistent across codebase).

Zero false positives this round — Qwen returned 3 tokens "No BLOCKs"
and the other two reviewers' findings were all real.

Setup confirmed: Ollama 0.21.0 on :11434 with nomic-embed-text loaded.
Per-text /api/embeddings used (forward-compat with 0.21+); newer
0.4+ /api/embed batch endpoint can swap in via the Provider interface.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-29 01:42:27 -05:00

112 lines
3.9 KiB
Go

// gateway is the Lakehouse-Go HTTP ingress. D6 promotes the D1
// stub endpoints into real reverse-proxies fronting all four backing
// services (storaged, catalogd, ingestd, queryd) on a single bind.
//
// Routes:
// /v1/storage/* → storaged
// /v1/catalog/* → catalogd
// /v1/ingest → ingestd
// /v1/sql → queryd
//
// The /v1 prefix lives at the edge — internal services route on
// /storage, /catalog, /ingest, /sql. Per Kimi K2 finding from the
// D1 plan review: httputil.NewSingleHostReverseProxy preserves the
// inbound path by default, so the proxy helper strips /v1 in its
// Director before forwarding.
package main
import (
"flag"
"log/slog"
"net/url"
"os"
"github.com/go-chi/chi/v5"
"git.agentview.dev/profit/golangLAKEHOUSE/internal/gateway"
"git.agentview.dev/profit/golangLAKEHOUSE/internal/shared"
)
func main() {
configPath := flag.String("config", "lakehouse.toml", "path to TOML config")
flag.Parse()
cfg, err := shared.LoadConfig(*configPath)
if err != nil {
slog.Error("config", "err", err)
os.Exit(1)
}
upstreams := map[string]string{
"storaged_url": cfg.Gateway.StoragedURL,
"catalogd_url": cfg.Gateway.CatalogdURL,
"ingestd_url": cfg.Gateway.IngestdURL,
"queryd_url": cfg.Gateway.QuerydURL,
"vectord_url": cfg.Gateway.VectordURL,
"embedd_url": cfg.Gateway.EmbeddURL,
}
for k, v := range upstreams {
if v == "" {
slog.Error("config", "err", "gateway."+k+" is required")
os.Exit(1)
}
}
// Per scrum O-WARN2 (Opus): url.Parse is permissive — a typo
// like "127.0.0.1:3211" (missing scheme) parses without error
// but produces empty Host, and every proxied request 502s. Fail
// fast at startup if scheme/host are missing so misconfigs
// surface in `systemctl status gateway` rather than at first traffic.
storagedURL := mustParseUpstream("storaged_url", cfg.Gateway.StoragedURL)
catalogdURL := mustParseUpstream("catalogd_url", cfg.Gateway.CatalogdURL)
ingestdURL := mustParseUpstream("ingestd_url", cfg.Gateway.IngestdURL)
querydURL := mustParseUpstream("queryd_url", cfg.Gateway.QuerydURL)
vectordURL := mustParseUpstream("vectord_url", cfg.Gateway.VectordURL)
embeddURL := mustParseUpstream("embedd_url", cfg.Gateway.EmbeddURL)
storagedProxy := gateway.NewProxyHandler(storagedURL)
catalogdProxy := gateway.NewProxyHandler(catalogdURL)
ingestdProxy := gateway.NewProxyHandler(ingestdURL)
querydProxy := gateway.NewProxyHandler(querydURL)
vectordProxy := gateway.NewProxyHandler(vectordURL)
embeddProxy := gateway.NewProxyHandler(embeddURL)
if err := shared.Run("gateway", cfg.Gateway.Bind, func(r chi.Router) {
// Storage / catalog have multi-segment paths under their
// prefix (e.g. /v1/storage/get/<key>). chi's `*` wildcard
// captures the rest of the path.
r.Handle("/v1/storage/*", storagedProxy)
r.Handle("/v1/catalog/*", catalogdProxy)
// Ingest + sql are single endpoints. We accept any method
// (GET/POST/etc) and let the backing service decide. ingestd
// only accepts POST; queryd only accepts POST. Other methods
// will get the backend's 405.
r.Handle("/v1/ingest", ingestdProxy)
r.Handle("/v1/sql", querydProxy)
// Vector search routes — /v1/vectors/index, /v1/vectors/index/{name}/...
r.Handle("/v1/vectors/*", vectordProxy)
// Embedding service — /v1/embed
r.Handle("/v1/embed", embeddProxy)
}); err != nil {
slog.Error("server", "err", err)
os.Exit(1)
}
}
// mustParseUpstream parses an upstream URL string and validates that
// scheme + host are non-empty. Exits the process on failure — gateway
// can't function without a valid upstream so failing fast is the
// right call. Per scrum O-WARN2.
func mustParseUpstream(name, raw string) *url.URL {
u, err := url.Parse(raw)
if err != nil {
slog.Error("config", "err", "parse "+name+": "+err.Error())
os.Exit(1)
}
if u.Scheme == "" || u.Host == "" {
slog.Error("config", "err", name+" must include scheme + host (got "+raw+")")
os.Exit(1)
}
return u
}