From 9698de32d5d3c41c1c5a3ab051cf5e1435630279 Mon Sep 17 00:00:00 2001 From: abhishek Date: Thu, 24 Sep 2026 11:01:16 +0530 Subject: [PATCH] api key integration --- .env.example | 13 ++ .env.local | 17 +++ controllers/assistantController.go | 7 + services/assistantLimit.go | 129 +++++++++++++++++++ services/assistantLimit_test.go | 113 +++++++++++++++++ services/assistantLive_test.go | 197 +++++++++++++++++++++++++++++ services/assistantService.go | 12 +- utils/chat.go | 2 +- utils/embedding.go | 23 ++-- 9 files changed, 503 insertions(+), 10 deletions(-) create mode 100644 services/assistantLimit.go create mode 100644 services/assistantLimit_test.go create mode 100644 services/assistantLive_test.go diff --git a/.env.example b/.env.example index 85f9d5d..4bedd52 100644 --- 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= diff --git a/.env.local b/.env.local index 4de96f5..52a906c 100644 --- 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 diff --git a/controllers/assistantController.go b/controllers/assistantController.go index 96db588..a2dd623 100644 --- 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()) } diff --git a/services/assistantLimit.go b/services/assistantLimit.go new file mode 100644 index 0000000..93a2b35 --- /dev/null +++ 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) + } + } +} diff --git a/services/assistantLimit_test.go b/services/assistantLimit_test.go new file mode 100644 index 0000000..4ae3cb2 --- /dev/null +++ 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 +} diff --git a/services/assistantLive_test.go b/services/assistantLive_test.go new file mode 100644 index 0000000..7dbf144 --- /dev/null +++ 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) + } + } +} diff --git a/services/assistantService.go b/services/assistantService.go index d6358d0..d867364 100644 --- 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 { diff --git a/utils/chat.go b/utils/chat.go index 079164d..952066b 100644 --- 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 { diff --git a/utils/embedding.go b/utils/embedding.go index 7d5a5fc..c8498f9 100644 --- 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 }