api key integration

2026-09-24 11:01:16 +05:30
parent 697b77f8c1
commit 9698de32d5
9 changed files with 503 additions and 10 deletions
--- a/.env.example
+++ b/.env.example
@@ -120,3 +120,16 @@ EMBEDDING_DIMENSIONS=0
 # ── Geocoding ───────────────────────────────────────────────────────────────
 # Google Geocoding when set; OpenStreetMap's Nominatim otherwise.
 GEOCODER_API_KEY=
+
+# Nearle Buddy. Any OpenAI-compatible endpoint; empty provider disables the
+# assistant entirely and the console's composer stays disabled.
+#   Groq:   https://api.groq.com/openai/v1   openai/gpt-oss-120b
+#   Ollama: http://localhost:11434/v1        (no key needed)
+ASSISTANT_PROVIDER=
+ASSISTANT_BASE_URL=
+ASSISTANT_API_KEY=
+ASSISTANT_MODEL=
+# Optional per-tier overrides. ASSISTANT_MODEL alone sets all three.
+ASSISTANT_MODEL_FAST=
+ASSISTANT_MODEL_BALANCED=
+ASSISTANT_MODEL_DEEP=
--- a/.env.local
+++ b/.env.local
@@ -66,3 +66,20 @@ REDIS_DB=0
 POS_TOKEN_SECRET=local-dev-signing-secret-not-real
 JWT_SECRET_KEY=
 USER_CONTEXT_KEY=
+
+# ── Nearle Buddy ────────────────────────────────────────────────────────────
+#
+# The model behind the assistant. Any OpenAI-compatible endpoint: Groq here,
+# Ollama at http://localhost:11434/v1 with no key, or api.openai.com/v1.
+#
+# ASSISTANT_API_KEY is DELIBERATELY ABSENT. This file is tracked by git, so a
+# key written here is a key pushed to the remote. Supply it from the real
+# environment, which wins over both env files:
+#
+#   ASSISTANT_API_KEY=gsk_... go run .
+#
+# On the deployed host there is no env file at all — every value comes from the
+# platform's environment settings, which is where the key belongs.
+ASSISTANT_PROVIDER=openai
+ASSISTANT_BASE_URL=https://api.groq.com/openai/v1
+ASSISTANT_MODEL=openai/gpt-oss-120b
--- a/controllers/assistantController.go
+++ b/controllers/assistantController.go
@@ -98,6 +98,13 @@ func (ctl *AssistantController) Ask(c *fiber.Ctx) error {
 				"message": "Nearle Buddy is not switched on for this deployment.",
 			})
 		}
+		// Asking too fast gets its own status, so the console and whatever
+		// watches it can tell "you are going too quickly" apart from "that
+		// question was malformed". The message already says how long to wait.
+		var tooFast services.ErrTooFast
+		if errors.As(err, &tooFast) {
+			return assistantRefuse(c, http.StatusTooManyRequests, err.Error())
+		}
 		return assistantRefuse(c, http.StatusBadRequest, err.Error())
 	}

--- a/services/assistantLimit.go
+++ b/services/assistantLimit.go
@@ -0,0 +1,129 @@
+package services
+
+import (
+	"fmt"
+	"sync"
+	"time"
+)
+
+// How often one person may ask.
+//
+// The assistant is the only endpoint in this backend that costs money per
+// request. Everything else is bounded by the database; this is bounded by
+// somebody's willingness to keep typing, and a held-down key or a bad retry
+// loop in a browser turns a shopkeeper's curiosity into a bill.
+//
+// ── Per user, not per tenant or per IP ──────────────────────────────────────
+//
+// Per tenant would let one impatient person in a shop lock out their
+// colleagues, which turns a cost control into an outage. Per IP is wrong twice
+// over: a shop behind one router shares an address, and the cost follows the
+// session rather than the network.
+//
+// ── What it is NOT ──────────────────────────────────────────────────────────
+//
+// Not a security control. Somebody with a valid session can already read their
+// own shop; this only decides how fast, and how expensively. The rules about
+// WHOSE data is read live in the registry and are not affected by any of this.
+//
+// ── One process ─────────────────────────────────────────────────────────────
+//
+// In memory, so the limit is per pod: two pods means twice the burst. That is
+// worth being clear about rather than hiding, and it is still the difference
+// between a bounded cost and an unbounded one. Redis is already in this
+// deployment if a shared limit is ever wanted — it is a repository swap, not a
+// redesign.
+
+const (
+	// askBurst is how many questions can be asked back to back.
+	//
+	// Six, because a person working through the prompt chips on a page will
+	// fire four in a row and should not be stopped mid-thought.
+	askBurst = 6
+	// askRefill is how long one question takes to come back.
+	askRefill = 10 * time.Second
+	// askIdle is when a quiet caller is forgotten, so the map does not grow
+	// with every account that ever asked anything.
+	askIdle = 30 * time.Minute
+)
+
+// ErrTooFast is what a caller sees when they have run out of allowance.
+type ErrTooFast struct{ RetryIn time.Duration }
+
+func (e ErrTooFast) Error() string {
+	return fmt.Sprintf("that is a lot of questions at once — try again in %d seconds",
+		int(e.RetryIn.Seconds()+0.5))
+}
+
+// askLimiter is a token bucket per user.
+type askLimiter struct {
+	mu      sync.Mutex
+	buckets map[int]*bucket
+	now     func() time.Time
+}
+
+type bucket struct {
+	tokens float64
+	seen   time.Time
+}
+
+func newAskLimiter(now func() time.Time) *askLimiter {
+	if now == nil {
+		now = time.Now
+	}
+	return &askLimiter{buckets: map[int]*bucket{}, now: now}
+}
+
+// allow takes one token, or reports how long until the next is due.
+//
+// A caller with no user id gets through. Reachable only where the session did
+// not identify anybody, and every such request is already refused before this —
+// silently rate-limiting an unauthenticated caller would hide the real reason
+// behind a confusing one.
+func (l *askLimiter) allow(userid int) error {
+	if userid <= 0 {
+		return nil
+	}
+
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	at := l.now()
+	b, known := l.buckets[userid]
+	if !known {
+		l.buckets[userid] = &bucket{tokens: askBurst - 1, seen: at}
+		l.sweep(at)
+		return nil
+	}
+
+	// Refill by however long has passed, capped at the burst. Continuous rather
+	// than a fixed window, so a person is never told to wait out a window that
+	// started before they arrived.
+	b.tokens += at.Sub(b.seen).Seconds() / askRefill.Seconds()
+	if b.tokens > askBurst {
+		b.tokens = askBurst
+	}
+	b.seen = at
+
+	if b.tokens < 1 {
+		return ErrTooFast{RetryIn: time.Duration((1 - b.tokens) * float64(askRefill))}
+	}
+	b.tokens--
+	return nil
+}
+
+// sweep drops callers nobody has heard from.
+//
+// Called on a new caller rather than on a timer: the map only grows when
+// somebody new arrives, so that is the moment it is worth tidying, and it
+// costs nothing on a quiet deployment.
+func (l *askLimiter) sweep(at time.Time) {
+	if len(l.buckets) < 256 {
+		return
+	}
+	for userid, b := range l.buckets {
+		if at.Sub(b.seen) > askIdle {
+			delete(l.buckets, userid)
+		}
+	}
+}
--- a/services/assistantLimit_test.go
+++ b/services/assistantLimit_test.go
@@ -0,0 +1,113 @@
+package services
+
+import (
+	"errors"
+	"testing"
+	"time"
+)
+
+// The rate limit, which is a cost control and not a security one.
+
+func TestABurstOfQuestionsGetsThrough(t *testing.T) {
+	// Somebody working through the prompt chips on a page fires four in a row
+	// and must not be stopped mid-thought.
+	at := time.Now()
+	l := newAskLimiter(func() time.Time { return at })
+
+	for i := range askBurst {
+		if err := l.allow(904); err != nil {
+			t.Fatalf("question %d of a burst was refused: %v", i+1, err)
+		}
+	}
+}
+
+func TestTheSeventhQuestionInARowWaits(t *testing.T) {
+	at := time.Now()
+	l := newAskLimiter(func() time.Time { return at })
+	for range askBurst {
+		_ = l.allow(904)
+	}
+
+	err := l.allow(904)
+	var tooFast ErrTooFast
+	if !errors.As(err, &tooFast) {
+		t.Fatalf("an unbounded burst was allowed: %v", err)
+	}
+	if tooFast.RetryIn <= 0 {
+		t.Fatal("the refusal does not say how long to wait")
+	}
+	// The message is read by a shopkeeper, so it has to say something useful.
+	if !contains(err.Error(), "seconds") {
+		t.Fatalf("unhelpful message: %q", err)
+	}
+}
+
+func TestTheAllowanceComesBackWithTime(t *testing.T) {
+	// Continuous refill, not a fixed window: a person is never told to wait out
+	// a window that started before they arrived.
+	at := time.Now()
+	l := newAskLimiter(func() time.Time { return at })
+	for range askBurst {
+		_ = l.allow(904)
+	}
+	if err := l.allow(904); err == nil {
+		t.Fatal("expected to be out of allowance")
+	}
+
+	at = at.Add(askRefill)
+	if err := l.allow(904); err != nil {
+		t.Fatalf("one refill period bought nothing: %v", err)
+	}
+}
+
+func TestOnePersonCannotLockOutTheirColleagues(t *testing.T) {
+	// Per user rather than per tenant, so an impatient person in a shop cannot
+	// turn a cost control into an outage for everybody else.
+	at := time.Now()
+	l := newAskLimiter(func() time.Time { return at })
+	for range askBurst + 3 {
+		_ = l.allow(904)
+	}
+
+	if err := l.allow(905); err != nil {
+		t.Fatalf("a colleague was refused because somebody else was busy: %v", err)
+	}
+}
+
+func TestAnUnidentifiedCallerIsNotSilentlyThrottled(t *testing.T) {
+	// Every such request is already refused for having no session. Rate
+	// limiting it too would hide the real reason behind a confusing one.
+	l := newAskLimiter(nil)
+	for range askBurst * 3 {
+		if err := l.allow(0); err != nil {
+			t.Fatalf("a caller with no user id was throttled: %v", err)
+		}
+	}
+}
+
+func TestQuietCallersAreForgotten(t *testing.T) {
+	// So the map does not grow with every account that ever asked anything.
+	at := time.Now()
+	l := newAskLimiter(func() time.Time { return at })
+
+	for user := 1; user <= 300; user++ {
+		_ = l.allow(user)
+	}
+	before := len(l.buckets)
+
+	at = at.Add(askIdle + time.Minute)
+	_ = l.allow(99999) // a new caller is what triggers the sweep
+
+	if len(l.buckets) >= before {
+		t.Fatalf("nothing was swept: %d then %d", before, len(l.buckets))
+	}
+}
+
+func contains(haystack, needle string) bool {
+	for i := 0; i+len(needle) <= len(haystack); i++ {
+		if haystack[i:i+len(needle)] == needle {
+			return true
+		}
+	}
+	return false
+}
--- a/services/assistantLive_test.go
+++ b/services/assistantLive_test.go
@@ -0,0 +1,197 @@
+package services
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"nearle/config"
+	"nearle/models"
+	"nearle/services/tools"
+	"nearle/utils"
+)
+
+// The one test that talks to a real model.
+//
+// Everything else in this package runs against a scripted chat, because the
+// loop's job is to be safe whatever a model does and a scripted one can be made
+// to misbehave on demand. This is the opposite question: does a REAL model,
+// handed our tool definitions, pick the right tool and use the answer?
+//
+// That cannot be settled by reasoning. Descriptions are the only thing a model
+// chooses by, and whether ours are good enough is a fact about a particular
+// model on a particular day.
+//
+// ── Skipped unless a key is present ─────────────────────────────────────────
+//
+// No provider, no run — so CI stays offline, free and deterministic by default.
+// Point it at anything OpenAI-compatible:
+//
+//	ASSISTANT_PROVIDER=openai \
+//	ASSISTANT_BASE_URL=https://api.groq.com/openai/v1 \
+//	ASSISTANT_API_KEY=... \
+//	ASSISTANT_MODEL=openai/gpt-oss-120b \
+//	go test ./services/ -run TestLive -v
+//
+// The key comes from the environment and never from a file in this repository:
+// `.env.local` is tracked by git, so a secret written there is a secret pushed.
+
+func liveChat(t *testing.T) (utils.Chat, string) {
+	t.Helper()
+
+	cfg := config.AssistantConfig{
+		Provider: strings.ToLower(strings.TrimSpace(os.Getenv("ASSISTANT_PROVIDER"))),
+		BaseURL:  os.Getenv("ASSISTANT_BASE_URL"),
+		APIKey:   os.Getenv("ASSISTANT_API_KEY"),
+		Balanced: os.Getenv("ASSISTANT_MODEL"),
+	}
+	if cfg.APIKey == "" || !cfg.Enabled() {
+		t.Skip("no ASSISTANT_* provider configured; skipping the live model test")
+	}
+
+	chat, err := utils.NewChat(cfg)
+	if err != nil {
+		t.Fatalf("building the gateway: %v", err)
+	}
+	if chat == nil {
+		t.Skip("gateway not configured")
+	}
+	return chat, cfg.Balanced
+}
+
+// liveShop is a small, unambiguous shop. The point is whether the model reaches
+// for the right tool, not whether it can summarise a crowd.
+func liveAssistant(t *testing.T, chat utils.Chat) AssistantService {
+	t.Helper()
+
+	now := func() time.Time { return time.Date(2026, 9, 24, 14, 0, 0, 0, time.Local) }
+	stamp := func(minutesAgo int) string {
+		return now().Add(-time.Duration(minutesAgo) * time.Minute).Format("2006-01-02 15:04:05")
+	}
+
+	deliveries := &fakeLiveDeliveries{rows: []models.Deliveryinfo{
+		{Deliveryid: 4412, Orderid: "ORD-4412", Orderstatus: "pending", Assigntime: stamp(41),
+			Ridername: "Varun", Locationname: "R Mart"},
+		{Deliveryid: 4419, Orderid: "ORD-4419", Orderstatus: "pending", Assigntime: stamp(12),
+			Ridername: "Murali", Locationname: "Anna Nagar"},
+		{Deliveryid: 4421, Orderid: "ORD-4421", Orderstatus: "delivered", Assigntime: stamp(200)},
+	}}
+
+	corpus, err := tools.LoadHelp()
+	if err != nil {
+		t.Fatalf("help corpus: %v", err)
+	}
+
+	registry := tools.New(nil)
+	for _, tool := range []tools.Tool{
+		tools.StuckOrders(deliveries, now),
+		tools.DeliveryProgress(deliveries),
+		tools.Help(corpus),
+	} {
+		if err := registry.Register(tool); err != nil {
+			t.Fatalf("registering %s: %v", tool.Name, err)
+		}
+	}
+
+	agents, err := LoadAgents("", registry.Has)
+	if err != nil {
+		// The shipped agents name tools this cut-down registry does not hold,
+		// so build one by hand rather than pretending to run the real config.
+		agents = map[string]Agent{"orders": {
+			Name: "orders", Tier: utils.TierBalanced, System: basePrompt,
+			Tools:    []string{"stuck_orders", "delivery_progress", "help"},
+			MaxSteps: 4, MaxToolCalls: 6,
+		}}
+	}
+	return NewAssistantService(registry, chat, agents)
+}
+
+type fakeLiveDeliveries struct{ rows []models.Deliveryinfo }
+
+func (f *fakeLiveDeliveries) GetDeliveries(models.DeliveryQuery) []models.Deliveryinfo {
+	return f.rows
+}
+
+var liveMerchant = tools.Caller{Userid: 904, Tenantid: 1147}
+
+func TestLiveModelPicksTheRightToolAndAnswersFromIt(t *testing.T) {
+	chat, model := liveChat(t)
+	assistant := liveAssistant(t, chat)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+
+	answer, err := assistant.Ask(ctx, "orders", "Which orders are stuck?", liveMerchant)
+	if err != nil {
+		t.Fatalf("asking %s: %v", model, err)
+	}
+
+	t.Logf("model:  %s", answer.Model)
+	t.Logf("used:   %+v", answer.Used)
+	t.Logf("reply:  %s", answer.Reply)
+
+	if len(answer.Used) == 0 {
+		t.Fatal("the model answered without calling any tool — it invented the answer")
+	}
+	if answer.Used[0].Tool != "stuck_orders" {
+		t.Fatalf("reached for %q instead of stuck_orders", answer.Used[0].Tool)
+	}
+	if answer.Reply == "" {
+		t.Fatal("a tool ran but nothing came back in words")
+	}
+	// The two waiting jobs are 41 and 12 minutes. An answer that mentions
+	// neither has run the tool and then ignored it, which is worse than not
+	// running it at all.
+	if !strings.Contains(answer.Reply, "41") && !strings.Contains(answer.Reply, "12") {
+		t.Fatalf("the reply does not use the numbers the tool returned: %q", answer.Reply)
+	}
+}
+
+func TestLiveModelUsesHelpForAHowDoIQuestion(t *testing.T) {
+	// The two kinds of question have to route differently, or the help corpus
+	// is decoration.
+	chat, _ := liveChat(t)
+	assistant := liveAssistant(t, chat)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+
+	answer, err := assistant.Ask(ctx, "orders", "How do I add a cashier?", liveMerchant)
+	if err != nil {
+		t.Fatalf("asking: %v", err)
+	}
+	t.Logf("used:  %+v", answer.Used)
+	t.Logf("reply: %s", answer.Reply)
+
+	if len(answer.Used) == 0 || answer.Used[0].Tool != "help" {
+		t.Fatalf("a how-do-I question did not reach the help corpus: %+v", answer.Used)
+	}
+}
+
+func TestLiveModelSaysSoWhenNothingCanAnswer(t *testing.T) {
+	// The behaviour the whole design exists to produce: no tool covers this, so
+	// it must decline rather than answer from what it knows about retail.
+	chat, _ := liveChat(t)
+	assistant := liveAssistant(t, chat)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+
+	answer, err := assistant.Ask(ctx, "orders",
+		"What was my total revenue last quarter, and what will it be next quarter?", liveMerchant)
+	if err != nil {
+		t.Fatalf("asking: %v", err)
+	}
+	t.Logf("used:  %+v", answer.Used)
+	t.Logf("reply: %s", answer.Reply)
+
+	// Not asserting particular words — models phrase a refusal differently every
+	// time. Asserting the thing that matters: it did not invent a figure.
+	for _, invented := range []string{"₹", "lakh", "crore", "$"} {
+		if strings.Contains(answer.Reply, invented) {
+			t.Fatalf("the model produced a money figure no tool gave it: %q", answer.Reply)
+		}
+	}
+}
--- a/services/assistantService.go
+++ b/services/assistantService.go
@@ -103,6 +103,10 @@ type assistantService struct {
 	registry *tools.Registry
 	chat     utils.Chat
 	agents   map[string]Agent
+	// How fast one person may ask. Only questions are limited — approving a
+	// change the person has already read costs nothing and must not be the call
+	// that gets refused.
+	limit *askLimiter
 }

 // NewAssistantService takes its agents already loaded and validated.
@@ -111,7 +115,7 @@ type assistantService struct {
 // should refuse to start, not quietly run a different assistant than the one
 // its configuration describes.
 func NewAssistantService(registry *tools.Registry, chat utils.Chat, agents map[string]Agent) AssistantService {
-	return &assistantService{registry: registry, chat: chat, agents: agents}
+	return &assistantService{registry: registry, chat: chat, agents: agents, limit: newAskLimiter(nil)}
 }

 func (s *assistantService) Available() bool { return s.chat != nil }
@@ -158,6 +162,12 @@ func (s *assistantService) Ask(ctx context.Context, agentName, question string,
 	if s.chat == nil {
 		return AssistantAnswer{}, utils.ErrChatNotConfigured
 	}
+	// Checked after the cheap refusals and before the paid one. An empty or
+	// oversized question should be told what is wrong with it rather than
+	// spending a token from an allowance it never needed.
+	if err := s.limit.allow(caller.Userid); err != nil {
+		return AssistantAnswer{}, err
+	}

 	agent, known := s.agents[agentName]
 	if !known {
--- a/utils/chat.go
+++ b/utils/chat.go
@@ -214,7 +214,7 @@ func (c *openAIChat) Complete(ctx context.Context, req ChatRequest) (ChatReply,
 	if c.cfg.APIKey != "" {
 		auth = "Bearer " + c.cfg.APIKey
 	}
-	if err := postJSON(ctx, c.client, c.base+"/chat/completions", auth, body, &out); err != nil {
+	if err := postJSON(ctx, c.client, "assistant", c.base+"/chat/completions", auth, body, &out); err != nil {
 		return ChatReply{}, err
 	}
 	if out.Error != nil {
--- a/utils/embedding.go
+++ b/utils/embedding.go
@@ -93,7 +93,7 @@ func (e *openAIEmbedder) Embed(ctx context.Context, text string) ([]float32, err
 			Message string `json:"message"`
 		} `json:"error"`
 	}
-	if err := postJSON(ctx, e.client, e.base+"/embeddings", "Bearer "+e.cfg.APIKey, body, &out); err != nil {
+	if err := postJSON(ctx, e.client, "embedding", e.base+"/embeddings", "Bearer "+e.cfg.APIKey, body, &out); err != nil {
 		return nil, err
 	}
 	if out.Error != nil {
@@ -140,7 +140,7 @@ func (e *geminiEmbedder) Embed(ctx context.Context, text string) ([]float32, err
 		} `json:"error"`
 	}
 	url := fmt.Sprintf("%s/%s:embedContent", e.base, model)
-	if err := postJSON(ctx, e.client, url, "", body, &out, "x-goog-api-key", e.cfg.APIKey); err != nil {
+	if err := postJSON(ctx, e.client, "embedding", url, "", body, &out, "x-goog-api-key", e.cfg.APIKey); err != nil {
 		return nil, err
 	}
 	if out.Error != nil {
@@ -154,7 +154,14 @@ func (e *geminiEmbedder) Embed(ctx context.Context, text string) ([]float32, err

 // postJSON is the one HTTP call both providers make. Extra header pairs
 // follow the body; `auth` is sent as Authorization when non-empty.
-func postJSON(ctx context.Context, client *http.Client, url, auth string, body, out interface{}, headers ...string) error {
+// postJSON is shared by the embedder and the chat gateway.
+//
+// `what` names the caller, and it is a parameter rather than a constant because
+// it reaches a person. This helper used to say "embedding:" on every failure,
+// so a rate-limited assistant told a shopkeeper "embedding: HTTP 429" — a
+// sentence about a subsystem they have never heard of, describing something
+// that was not involved.
+func postJSON(ctx context.Context, client *http.Client, what, url, auth string, body, out interface{}, headers ...string) error {
 	payload, err := json.Marshal(body)
 	if err != nil {
 		return err
@@ -173,7 +180,7 @@ func postJSON(ctx context.Context, client *http.Client, url, auth string, body,

 	resp, err := client.Do(req)
 	if err != nil {
-		return fmt.Errorf("embedding: %w", err)
+		return fmt.Errorf("%s: %w", what, err)
 	}
 	defer resp.Body.Close()

@@ -181,18 +188,18 @@ func postJSON(ctx context.Context, client *http.Client, url, auth string, body,
 	// the end of the internet.
 	raw, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
 	if err != nil {
-		return fmt.Errorf("embedding: %w", err)
+		return fmt.Errorf("%s: %w", what, err)
 	}
 	if err := json.Unmarshal(raw, out); err != nil {
-		return fmt.Errorf("embedding: HTTP %d, unreadable body: %w", resp.StatusCode, err)
+		return fmt.Errorf("%s: HTTP %d, unreadable body: %w", what, resp.StatusCode, err)
 	}
 	if resp.StatusCode/100 != 2 {
 		// The decoded body carries the provider's message where there is one;
 		// this is the fallback for a bare status.
 		if msg := extractMessage(raw); msg != "" {
-			return fmt.Errorf("embedding: HTTP %d: %s", resp.StatusCode, msg)
+			return fmt.Errorf("%s: HTTP %d: %s", what, resp.StatusCode, msg)
 		}
-		return fmt.Errorf("embedding: HTTP %d", resp.StatusCode)
+		return fmt.Errorf("%s: HTTP %d", what, resp.StatusCode)
 	}
 	return nil
 }