From 56e227c4030e85163cd882b29ab472f8ce3e1a27 Mon Sep 17 00:00:00 2001 From: Duy /zuey/ Date: Thu, 21 May 2026 16:37:19 +0700 Subject: [PATCH] feat(api): unblock goclaw-cli P6 trace follow and provider reconnect (#37) * feat(api): unblock trace follow and provider reconnect * test(cron): stabilize run log execution test --- cmd/gateway_managed.go | 8 +- docs/18-http-api.md | 35 +++ internal/cron/service_test.go | 56 ++++- internal/http/openapi_spec.json | 83 +++++++ internal/http/providers.go | 121 ++++++++-- internal/http/providers_reconnect_test.go | 226 ++++++++++++++++++ internal/http/traces.go | 91 +++++++ internal/http/traces_follow_test.go | 206 ++++++++++++++++ internal/store/pg/tracing.go | 5 + internal/store/pg/tracing_follow_test.go | 39 +++ internal/store/sqlitestore/tracing.go | 4 + .../store/sqlitestore/tracing_follow_test.go | 41 ++++ internal/store/tracing_store.go | 15 +- 13 files changed, 890 insertions(+), 40 deletions(-) create mode 100644 internal/http/providers_reconnect_test.go create mode 100644 internal/http/traces_follow_test.go create mode 100644 internal/store/pg/tracing_follow_test.go create mode 100644 internal/store/sqlitestore/tracing_follow_test.go diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index 5d2c111b..18411a0f 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -18,12 +18,12 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/eventbus" "github.com/nextlevelbuilder/goclaw/internal/hooks" hookbuiltin "github.com/nextlevelbuilder/goclaw/internal/hooks/builtin" - "github.com/nextlevelbuilder/goclaw/internal/orchestration" httpapi "github.com/nextlevelbuilder/goclaw/internal/http" kg "github.com/nextlevelbuilder/goclaw/internal/knowledgegraph" mcpbridge "github.com/nextlevelbuilder/goclaw/internal/mcp" "github.com/nextlevelbuilder/goclaw/internal/media" memorypkg "github.com/nextlevelbuilder/goclaw/internal/memory" + "github.com/nextlevelbuilder/goclaw/internal/orchestration" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/sandbox" "github.com/nextlevelbuilder/goclaw/internal/skills" @@ -678,7 +678,11 @@ func wireExtras( return } // Unregister old instance (closes ProcessPool) then re-register - providerReg.Unregister(p.Name) + tenantID := event.TenantID + if tenantID == uuid.Nil { + tenantID = store.MasterTenantID + } + providerReg.UnregisterForTenant(tenantID, p.Name) if p.Enabled { registerACPFromDB(providerReg, *p) } diff --git a/docs/18-http-api.md b/docs/18-http-api.md index 2cb11b98..f05157af 100644 --- a/docs/18-http-api.md +++ b/docs/18-http-api.md @@ -344,6 +344,7 @@ LLM provider management. API keys are encrypted with AES-256-GCM in the database | `GET` | `/v1/providers/{id}` | Get provider | | `PUT` | `/v1/providers/{id}` | Update provider | | `DELETE` | `/v1/providers/{id}` | Delete provider | +| `POST` | `/v1/providers/{id}/reconnect` | Reload provider runtime from stored config | ### Provider Verification & Models @@ -358,6 +359,25 @@ LLM provider management. API keys are encrypted with AES-256-GCM in the database **Supported types:** `anthropic_native`, `openai_compat`, `chatgpt_oauth`, `gemini_native`, `dashscope`, `bailian`, `minimax`, `claude_cli`, `acp` +Reconnect response: + +```json +{ + "status": "reconnected", + "provider": { + "id": "0193a5b0-7000-7000-8000-000000000123", + "name": "openrouter-main", + "provider_type": "openai_compat", + "api_key": "***", + "enabled": true + }, + "registry_updated": true, + "cache_invalidated": true +} +``` + +`status` is `reconnected`, `disabled`, or `not_registered`. Reconnect never changes stored provider config. It does not run an upstream verify call; call `/v1/providers/{id}/verify` after reconnect when that check is needed. + Example response: ```json @@ -1148,11 +1168,26 @@ LLM call tracing and cost analysis. | Method | Path | Description | |--------|------|-------------| | `GET` | `/v1/traces` | List traces (paginated, filterable) | +| `GET` | `/v1/traces/follow` | Poll trace changes for one session or agent | | `GET` | `/v1/traces/{traceID}` | Get trace with spans | | `GET` | `/v1/traces/{traceID}/export` | Export trace tree (gzipped JSON) | **Filters:** `agent_id`, `user_id`, `session_key`, `status`, `channel` +`GET /v1/traces/follow` requires `session_key` or `agent_id`. Query params: `session_key`, `agent_id`, `status`, `channel`, `since` (RFC 3339), `limit` (default 50, max 200), `include_spans` (default false). Non-admin callers only see their own traces. When `since` is provided, the server returns traces matching existing filters and `(created_at > since OR end_time > since OR status = "running")`. + +Follow response: + +```json +{ + "traces": [], + "spans_by_trace_id": {}, + "server_time": "2026-05-20T11:23:00Z", + "next_since": "2026-05-20T11:23:00Z", + "limit": 50 +} +``` + ### Costs | Method | Path | Description | diff --git a/internal/cron/service_test.go b/internal/cron/service_test.go index 116e4dd4..15ec1672 100644 --- a/internal/cron/service_test.go +++ b/internal/cron/service_test.go @@ -339,14 +339,44 @@ func TestService_RunLog_PopulatedByAutoExecution(t *testing.T) { return "ok", nil }) - interval := int64(50) + if err := cs.Start(); err != nil { + t.Fatalf("Start error: %v", err) + } + defer cs.Stop() + + interval := int64(time.Hour / time.Millisecond) job, _ := cs.AddJob("logger", Schedule{Kind: "every", EveryMS: &interval}, "tick", false, "", "", "") - cs.Start() - time.Sleep(120 * time.Millisecond) - cs.Stop() + cs.mu.Lock() + foundJob := false + for i := range cs.store.Jobs { + if cs.store.Jobs[i].ID == job.ID { + due := nowMS() + cs.store.Jobs[i].State.NextRunAtMS = &due + foundJob = true + break + } + } + if !foundJob { + cs.mu.Unlock() + t.Fatalf("job %s not found in store", job.ID) + } + if err := cs.saveUnsafe(); err != nil { + cs.mu.Unlock() + t.Fatalf("save due job: %v", err) + } + cs.mu.Unlock() + + var log []RunLogEntry + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + log = cs.GetRunLog(job.ID, 50) + if len(log) > 0 { + break + } + time.Sleep(5 * time.Millisecond) + } - log := cs.GetRunLog(job.ID, 50) if len(log) == 0 { t.Fatal("expected at least 1 run log entry from automatic execution") } @@ -420,11 +450,11 @@ func TestAnchorBasedNextRun_PreservesOffset(t *testing.T) { // Formula: next = anchor + (elapsed/interval + 1) * interval tests := []struct { - name string - anchor int64 // scheduledAtMS - interval int64 // everyMS - now int64 - wantNext int64 + name string + anchor int64 // scheduledAtMS + interval int64 // everyMS + now int64 + wantNext int64 }{ { name: "normal_one_period", @@ -461,7 +491,7 @@ func TestAnchorBasedNextRun_PreservesOffset(t *testing.T) { { name: "small_interval_large_gap", anchor: 0, - interval: 1000, // 1 second + interval: 1000, // 1 second now: 86400000, // 24 hours later — O(1) handles this without 86400 iterations // elapsed=86400000, periods=86400000/1000=86400, next=0+(86400+1)*1000=86401000 wantNext: 86401000, @@ -488,8 +518,8 @@ func TestAnchorBasedNextRun_PreservesOffset(t *testing.T) { interval := int64(5000) now := int64(6500) - nextA := anchorA + (((now - anchorA) / interval) + 1) * interval - nextB := anchorB + (((now - anchorB) / interval) + 1) * interval + nextA := anchorA + (((now-anchorA)/interval)+1)*interval + nextB := anchorB + (((now-anchorB)/interval)+1)*interval offset := nextA - nextB if offset != 4000 { // 11000 - 7000 = 4000 (original offset 1000 preserved mod interval) t.Fatalf("expected 4000ms offset between jobs, got %d", offset) diff --git a/internal/http/openapi_spec.json b/internal/http/openapi_spec.json index bf57607a..27cdc50e 100644 --- a/internal/http/openapi_spec.json +++ b/internal/http/openapi_spec.json @@ -482,6 +482,52 @@ "responses": { "200": { "description": "Provider deleted" } } } }, + "/v1/providers/{id}/reconnect": { + "post": { + "tags": ["Providers"], + "summary": "Reconnect provider runtime", + "description": "Reload an existing provider from stored config into the runtime registry. Does not mutate stored config and does not call upstream verification.", + "parameters": [{ "name": "id", "in": "path", "required": true, "schema": { "type": "string", "format": "uuid" } }], + "requestBody": { + "required": false, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "verify": { + "type": "boolean", + "default": false, + "description": "Only false is supported. Call /v1/providers/{id}/verify separately for upstream verification." + } + } + } + } + } + }, + "responses": { + "200": { + "description": "Provider reconnect result", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "status": { "type": "string", "enum": ["reconnected", "disabled", "not_registered"] }, + "provider": { "$ref": "#/components/schemas/ProviderInput" }, + "registry_updated": { "type": "boolean" }, + "cache_invalidated": { "type": "boolean" } + }, + "required": ["status", "provider", "registry_updated", "cache_invalidated"] + } + } + } + }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "404": { "$ref": "#/components/responses/NotFound" } + } + } + }, "/v1/providers/{id}/models": { "get": { "tags": ["Providers"], @@ -764,6 +810,43 @@ "responses": { "200": { "description": "Trace list" } } } }, + "/v1/traces/follow": { + "get": { + "tags": ["Traces"], + "summary": "Poll trace changes", + "description": "Stateless polling endpoint for traces in one session or one agent. Requires session_key or agent_id.", + "parameters": [ + { "name": "session_key", "in": "query", "schema": { "type": "string" } }, + { "name": "agent_id", "in": "query", "schema": { "type": "string", "format": "uuid" } }, + { "name": "status", "in": "query", "schema": { "type": "string" } }, + { "name": "channel", "in": "query", "schema": { "type": "string" } }, + { "name": "since", "in": "query", "schema": { "type": "string", "format": "date-time" } }, + { "name": "limit", "in": "query", "schema": { "type": "integer", "default": 50, "maximum": 200 } }, + { "name": "include_spans", "in": "query", "schema": { "type": "boolean", "default": false } } + ], + "responses": { + "200": { + "description": "Trace follow page", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "traces": { "type": "array", "items": { "type": "object" } }, + "spans_by_trace_id": { "type": "object", "additionalProperties": { "type": "array", "items": { "type": "object" } } }, + "server_time": { "type": "string", "format": "date-time" }, + "next_since": { "type": "string", "format": "date-time" }, + "limit": { "type": "integer" } + }, + "required": ["traces", "spans_by_trace_id", "server_time", "next_since", "limit"] + } + } + } + }, + "400": { "$ref": "#/components/responses/BadRequest" } + } + } + }, "/v1/traces/{traceID}": { "get": { "tags": ["Traces"], diff --git a/internal/http/providers.go b/internal/http/providers.go index 5eb07696..bee754f0 100644 --- a/internal/http/providers.go +++ b/internal/http/providers.go @@ -1,6 +1,7 @@ package http import ( + "context" "encoding/json" "fmt" "log/slog" @@ -107,14 +108,27 @@ func (h *ProvidersHandler) resolveAPIBase(p *store.LLMProviderData) string { // emitProviderCacheInvalidate broadcasts a provider cache invalidation event. // Subscribers (e.g. ACP re-registration in gateway_managed.go) react to reload from DB. -func (h *ProvidersHandler) emitProviderCacheInvalidate(name string) { +func (h *ProvidersHandler) emitProviderCacheInvalidate(ctx context.Context, tenantID uuid.UUID, name string) bool { if h.msgBus == nil { - return + return false } + tenantID = providerCacheTenantID(ctx, tenantID) h.msgBus.Broadcast(bus.Event{ - Name: protocol.EventCacheInvalidate, - Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindProvider, Key: name}, + Name: protocol.EventCacheInvalidate, + TenantID: tenantID, + Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindProvider, Key: name, TenantID: tenantID}, }) + return true +} + +func providerCacheTenantID(ctx context.Context, tenantID uuid.UUID) uuid.UUID { + if tenantID != uuid.Nil { + return tenantID + } + if tid := store.TenantIDFromContext(ctx); tid != uuid.Nil { + return tid + } + return store.MasterTenantID } // RegisterRoutes registers all provider management routes on the given mux. @@ -130,6 +144,7 @@ func (h *ProvidersHandler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("GET /v1/providers/{id}/models", h.auth(h.handleListProviderModels)) // Provider + model verification (pre-flight check) + mux.HandleFunc("POST /v1/providers/{id}/reconnect", h.auth(h.handleReconnectProvider)) mux.HandleFunc("POST /v1/providers/{id}/verify", h.auth(h.handleVerifyProvider)) mux.HandleFunc("POST /v1/providers/{id}/verify-embedding", h.auth(h.handleVerifyEmbedding)) @@ -154,16 +169,29 @@ func maskAPIKey(p *store.LLMProviderData) { } } +type providerRuntimeRegistrationStatus string + +const ( + providerRuntimeRegistered providerRuntimeRegistrationStatus = "registered" + providerRuntimeDisabled providerRuntimeRegistrationStatus = "disabled" + providerRuntimeSkipped providerRuntimeRegistrationStatus = "skipped" + providerRuntimeMissingCredential providerRuntimeRegistrationStatus = "missing_credential" + providerRuntimeInvalidConfig providerRuntimeRegistrationStatus = "invalid_config" +) + // registerInMemory adds (or replaces) a provider in the in-memory registry // so it's immediately usable for verify/chat without a gateway restart. -func (h *ProvidersHandler) registerInMemory(p *store.LLMProviderData) { +func (h *ProvidersHandler) registerInMemory(p *store.LLMProviderData) providerRuntimeRegistrationStatus { if h.providerReg == nil || !p.Enabled { - return + if p.Enabled { + return providerRuntimeSkipped + } + return providerRuntimeDisabled } // ACP agents don't need an API key — skip in-memory registration // (ACP providers are registered via gateway_providers.go on startup or restart) if p.ProviderType == store.ProviderACP { - return + return providerRuntimeSkipped } // Claude CLI doesn't need an API key — register immediately if p.ProviderType == store.ProviderClaudeCLI { @@ -179,7 +207,7 @@ func (h *ProvidersHandler) registerInMemory(p *store.LLMProviderData) { } if _, err := exec.LookPath(cliPath); err != nil { slog.Warn("claude-cli: binary not found, skipping in-memory registration", "path", cliPath, "provider", p.Name, "error", err) - return + return providerRuntimeInvalidConfig } cliOpts := []providers.ClaudeCLIOption{ providers.WithClaudeCLIName(p.Name), @@ -191,7 +219,7 @@ func (h *ProvidersHandler) registerInMemory(p *store.LLMProviderData) { cliOpts = append(cliOpts, providers.WithClaudeCLIMCPConfigData(mcpData)) } h.providerReg.RegisterForTenant(p.TenantID, providers.NewClaudeCLIProvider(cliPath, cliOpts...)) - return + return providerRuntimeRegistered } // Ollama doesn't need an API key — handle before the key guard (same as startup). // In Docker, swap localhost → host.docker.internal so the container can reach the host. @@ -202,14 +230,14 @@ func (h *ProvidersHandler) registerInMemory(p *store.LLMProviderData) { host = "http://localhost:11434/v1" } h.providerReg.RegisterForTenant(p.TenantID, providers.NewOpenAIProvider(p.Name, "ollama", config.DockerLocalhost(host), "llama3.3")) - return + return providerRuntimeRegistered } // Vertex supports ADC (empty api_key) — handle before the generic key guard. if p.ProviderType == store.ProviderVertex { vsettings := store.ParseVertexProviderSettings(p.Settings) if vsettings == nil { slog.Warn("vertex: missing project_id/region in settings, cannot register", "name", p.Name) - return + return providerRuntimeInvalidConfig } vcfg := providers.VertexConfig{ Name: p.Name, @@ -222,13 +250,13 @@ func (h *ProvidersHandler) registerInMemory(p *store.LLMProviderData) { prov, err := providers.NewVertexProviderWithTimeout(vcfg) if err != nil { slog.Warn("vertex: register in-memory failed", "name", p.Name, "error", err) - return + return providerRuntimeInvalidConfig } h.providerReg.RegisterForTenant(p.TenantID, prov) - return + return providerRuntimeRegistered } if p.APIKey == "" { - return + return providerRuntimeMissingCredential } apiBase := h.resolveAPIBase(p) switch p.ProviderType { @@ -269,6 +297,7 @@ func (h *ProvidersHandler) registerInMemory(p *store.LLMProviderData) { } h.providerReg.RegisterForTenant(p.TenantID, prov) } + return providerRuntimeRegistered } // normalizeOllamaAPIBase ensures Ollama and OllamaCloud api_base values include the @@ -422,7 +451,7 @@ func (h *ProvidersHandler) handleCreateProvider(w http.ResponseWriter, r *http.R // Register in-memory so verify/chat work without restart h.registerInMemory(&p) - h.emitProviderCacheInvalidate(p.Name) + h.emitProviderCacheInvalidate(r.Context(), p.TenantID, p.Name) emitAudit(h.msgBus, r, "provider.created", "provider", p.ID.String()) maskAPIKey(&p) @@ -449,6 +478,62 @@ func (h *ProvidersHandler) handleGetProvider(w http.ResponseWriter, r *http.Requ writeJSON(w, http.StatusOK, publicProvider) } +func (h *ProvidersHandler) handleReconnectProvider(w http.ResponseWriter, r *http.Request) { + locale := extractLocale(r) + id, err := uuid.Parse(r.PathValue("id")) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgInvalidID, "provider")}) + return + } + + var req struct { + Verify bool `json:"verify"` + } + if r.Body != nil && r.Body != http.NoBody && r.ContentLength != 0 { + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgInvalidJSON)}) + return + } + if req.Verify { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgInvalidRequest, "verify is not supported by reconnect; call provider verify separately")}) + return + } + } + + p, err := h.store.GetProvider(r.Context(), id) + if err != nil { + writeJSON(w, http.StatusNotFound, map[string]string{"error": i18n.T(locale, i18n.MsgNotFound, "provider", id.String())}) + return + } + + if h.providerReg != nil { + h.providerReg.UnregisterForTenant(p.TenantID, p.Name) + } + + status := "disabled" + registryUpdated := false + if p.Enabled { + switch h.registerInMemory(p) { + case providerRuntimeRegistered: + status = "reconnected" + registryUpdated = true + default: + status = "not_registered" + } + } + cacheInvalidated := h.emitProviderCacheInvalidate(r.Context(), p.TenantID, p.Name) + emitAudit(h.msgBus, r, "provider.reconnected", "provider", p.ID.String()) + + maskAPIKey(p) + publicProvider := canonicalizeProviderForResponse(p) + writeJSON(w, http.StatusOK, map[string]any{ + "status": status, + "provider": publicProvider, + "registry_updated": registryUpdated, + "cache_invalidated": cacheInvalidated, + }) +} + func (h *ProvidersHandler) handleUpdateProvider(w http.ResponseWriter, r *http.Request) { locale := extractLocale(r) id, err := uuid.Parse(r.PathValue("id")) @@ -586,9 +671,9 @@ func (h *ProvidersHandler) handleUpdateProvider(w http.ResponseWriter, r *http.R // Notify subscribers (e.g. ACP re-registration) about the change if updated, err := h.store.GetProvider(r.Context(), id); err == nil { - h.emitProviderCacheInvalidate(updated.Name) + h.emitProviderCacheInvalidate(r.Context(), updated.TenantID, updated.Name) if oldName != "" && oldName != updated.Name { - h.emitProviderCacheInvalidate(oldName) + h.emitProviderCacheInvalidate(r.Context(), updated.TenantID, oldName) } } @@ -622,7 +707,7 @@ func (h *ProvidersHandler) handleDeleteProvider(w http.ResponseWriter, r *http.R h.providerReg.UnregisterForTenant(providerTenantID, providerName) } if providerName != "" { - h.emitProviderCacheInvalidate(providerName) + h.emitProviderCacheInvalidate(r.Context(), providerTenantID, providerName) } emitAudit(h.msgBus, r, "provider.deleted", "provider", id.String()) diff --git a/internal/http/providers_reconnect_test.go b/internal/http/providers_reconnect_test.go new file mode 100644 index 00000000..0aa07350 --- /dev/null +++ b/internal/http/providers_reconnect_test.go @@ -0,0 +1,226 @@ +package http + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +func TestProvidersHandlerReconnectRejectsNonAdmin(t *testing.T) { + token := setupTraceReadToken(t, "viewer") + providerStore := newMockProviderStore() + provider := &store.LLMProviderData{ + BaseModel: store.BaseModel{ID: uuid.New()}, + TenantID: uuid.New(), + Name: "openrouter-main", + ProviderType: store.ProviderOpenAICompat, + APIKey: "token", + Enabled: true, + } + if err := providerStore.CreateProvider(t.Context(), provider); err != nil { + t.Fatalf("CreateProvider() error = %v", err) + } + + mux := http.NewServeMux() + NewProvidersHandler(providerStore, newMockSecretsStore(), providers.NewRegistry(nil), "").RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodPost, "/v1/providers/"+provider.ID.String()+"/reconnect", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusForbidden { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusForbidden, rec.Body.String()) + } +} + +func TestProvidersHandlerReconnectRejectsVerifyTrue(t *testing.T) { + token := setupProvidersAdminToken(t) + providerStore := newMockProviderStore() + provider := &store.LLMProviderData{ + BaseModel: store.BaseModel{ID: uuid.New()}, + TenantID: uuid.New(), + Name: "openrouter-main", + ProviderType: store.ProviderOpenAICompat, + APIKey: "token", + Enabled: true, + } + if err := providerStore.CreateProvider(t.Context(), provider); err != nil { + t.Fatalf("CreateProvider() error = %v", err) + } + + mux := http.NewServeMux() + NewProvidersHandler(providerStore, newMockSecretsStore(), providers.NewRegistry(nil), "").RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodPost, "/v1/providers/"+provider.ID.String()+"/reconnect", bytes.NewBufferString(`{"verify":true}`)) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusBadRequest, rec.Body.String()) + } +} + +func TestProvidersHandlerReconnectRegistersEnabledProviderAndInvalidatesCache(t *testing.T) { + token := setupProvidersAdminToken(t) + tenantID := uuid.New() + providerStore := newMockProviderStore() + provider := &store.LLMProviderData{ + BaseModel: store.BaseModel{ID: uuid.New()}, + TenantID: tenantID, + Name: "openrouter-main", + ProviderType: store.ProviderOpenAICompat, + APIBase: "https://openrouter.ai/api/v1", + APIKey: "secret-token", + Enabled: true, + } + if err := providerStore.CreateProvider(t.Context(), provider); err != nil { + t.Fatalf("CreateProvider() error = %v", err) + } + + providerReg := providers.NewRegistry(nil) + msgBus := bus.New() + var cacheEvents []bus.Event + msgBus.Subscribe(bus.TopicCacheProvider, func(event bus.Event) { + cacheEvents = append(cacheEvents, event) + }) + handler := NewProvidersHandler(providerStore, newMockSecretsStore(), providerReg, "") + handler.SetMessageBus(msgBus) + mux := http.NewServeMux() + handler.RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodPost, "/v1/providers/"+provider.ID.String()+"/reconnect", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + if _, err := providerReg.GetForTenant(tenantID, provider.Name); err != nil { + t.Fatalf("GetForTenant() error = %v, want provider registered", err) + } + + var body struct { + Status string `json:"status"` + Provider store.LLMProviderData `json:"provider"` + RegistryUpdated bool `json:"registry_updated"` + CacheInvalidated bool `json:"cache_invalidated"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatalf("decode response: %v", err) + } + if body.Status != "reconnected" || !body.RegistryUpdated || !body.CacheInvalidated { + t.Fatalf("body = %+v, want reconnected with registry/cache flags", body) + } + if body.Provider.APIKey != "***" { + t.Fatalf("provider.api_key = %q, want masked", body.Provider.APIKey) + } + if len(cacheEvents) == 0 { + t.Fatal("cache invalidation event not emitted") + } + got := cacheEvents[0] + if got.Name != protocol.EventCacheInvalidate || got.TenantID != tenantID { + t.Fatalf("cache event = %+v, want provider cache invalidation for tenant %s", got, tenantID) + } + payload, ok := got.Payload.(bus.CacheInvalidatePayload) + if !ok { + t.Fatalf("payload = %T, want CacheInvalidatePayload", got.Payload) + } + if payload.Kind != bus.CacheKindProvider || payload.Key != provider.Name || payload.TenantID != tenantID { + t.Fatalf("payload = %+v, want provider %q tenant %s", payload, provider.Name, tenantID) + } +} + +func TestProvidersHandlerReconnectDisabledProviderUnregisters(t *testing.T) { + token := setupProvidersAdminToken(t) + tenantID := uuid.New() + providerStore := newMockProviderStore() + provider := &store.LLMProviderData{ + BaseModel: store.BaseModel{ID: uuid.New()}, + TenantID: tenantID, + Name: "disabled-main", + ProviderType: store.ProviderOpenAICompat, + APIKey: "secret-token", + Enabled: false, + } + if err := providerStore.CreateProvider(t.Context(), provider); err != nil { + t.Fatalf("CreateProvider() error = %v", err) + } + + providerReg := providers.NewRegistry(nil) + providerReg.RegisterForTenant(tenantID, providers.NewOpenAIProvider(provider.Name, "old", "", "")) + mux := http.NewServeMux() + NewProvidersHandler(providerStore, newMockSecretsStore(), providerReg, "").RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodPost, "/v1/providers/"+provider.ID.String()+"/reconnect", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + if _, err := providerReg.GetForTenant(tenantID, provider.Name); err == nil { + t.Fatal("GetForTenant() succeeded, want provider unregistered") + } + var body struct { + Status string `json:"status"` + RegistryUpdated bool `json:"registry_updated"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatalf("decode response: %v", err) + } + if body.Status != "disabled" || body.RegistryUpdated { + t.Fatalf("body = %+v, want disabled without registry update", body) + } +} + +func TestProvidersHandlerReconnectMissingCredentialReturnsNotRegistered(t *testing.T) { + token := setupProvidersAdminToken(t) + tenantID := uuid.New() + providerStore := newMockProviderStore() + provider := &store.LLMProviderData{ + BaseModel: store.BaseModel{ID: uuid.New()}, + TenantID: tenantID, + Name: "missing-key", + ProviderType: store.ProviderOpenAICompat, + Enabled: true, + } + if err := providerStore.CreateProvider(t.Context(), provider); err != nil { + t.Fatalf("CreateProvider() error = %v", err) + } + + providerReg := providers.NewRegistry(nil) + mux := http.NewServeMux() + NewProvidersHandler(providerStore, newMockSecretsStore(), providerReg, "").RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodPost, "/v1/providers/"+provider.ID.String()+"/reconnect", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + var body struct { + Status string `json:"status"` + RegistryUpdated bool `json:"registry_updated"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatalf("decode response: %v", err) + } + if body.Status != "not_registered" || body.RegistryUpdated { + t.Fatalf("body = %+v, want not_registered without registry update", body) + } +} diff --git a/internal/http/traces.go b/internal/http/traces.go index df2ec24d..dcde4417 100644 --- a/internal/http/traces.go +++ b/internal/http/traces.go @@ -30,6 +30,7 @@ func NewTracesHandler(tracing store.TracingStore) *TracesHandler { // RegisterRoutes registers trace routes on the given mux. func (h *TracesHandler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("GET /v1/traces", h.authMiddleware(h.handleList)) + mux.HandleFunc("GET /v1/traces/follow", h.authMiddleware(h.handleFollow)) mux.HandleFunc("GET /v1/traces/{traceID}/export", h.authMiddleware(h.handleExport)) mux.HandleFunc("GET /v1/traces/{traceID}", h.authMiddleware(h.handleGet)) mux.HandleFunc("GET /v1/costs/summary", h.authMiddleware(h.handleCostSummary)) @@ -97,6 +98,96 @@ func (h *TracesHandler) handleList(w http.ResponseWriter, r *http.Request) { }) } +func (h *TracesHandler) handleFollow(w http.ResponseWriter, r *http.Request) { + locale := store.LocaleFromContext(r.Context()) + query := r.URL.Query() + opts := store.TraceListOpts{ + Limit: 50, + } + + if v := query.Get("agent_id"); v != "" { + id, err := uuid.Parse(v) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgInvalidID, "agent")}) + return + } + opts.AgentID = &id + } + if v := query.Get("session_key"); v != "" { + opts.SessionKey = v + } + if opts.SessionKey == "" && opts.AgentID == nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgRequired, "session_key or agent_id")}) + return + } + if v := query.Get("status"); v != "" { + opts.Status = v + } + if v := query.Get("channel"); v != "" { + opts.Channel = v + } + if v := query.Get("since"); v != "" { + since, err := time.Parse(time.RFC3339, v) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgInvalidRequest, "since must be RFC3339")}) + return + } + opts.ChangedAfter = &since + } + if v := query.Get("limit"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + if n > 200 { + n = 200 + } + opts.Limit = n + } + } + includeSpans := false + if v := query.Get("include_spans"); v != "" { + parsed, err := strconv.ParseBool(v) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgInvalidRequest, "include_spans must be boolean")}) + return + } + includeSpans = parsed + } + + auth := resolveAuth(r) + if !permissions.HasMinRole(auth.Role, permissions.RoleAdmin) { + opts.UserID = store.UserIDFromContext(r.Context()) + } else if v := query.Get("user_id"); v != "" { + opts.UserID = v + } + + serverTime := time.Now().UTC() + traces, err := h.tracing.ListTraces(r.Context(), opts) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + + spansByTraceID := map[string][]store.SpanData{} + if includeSpans { + for _, trace := range traces { + spans, err := h.tracing.GetTraceSpans(r.Context(), trace.ID) + if err != nil { + slog.Error("traces.follow_get_spans_failed", "trace_id", trace.ID, "error", err) + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + spansByTraceID[trace.ID.String()] = spans + } + } + + writeJSON(w, http.StatusOK, map[string]any{ + "traces": traces, + "spans_by_trace_id": spansByTraceID, + "server_time": serverTime, + "next_since": serverTime, + "limit": opts.Limit, + }) +} + func (h *TracesHandler) handleGet(w http.ResponseWriter, r *http.Request) { locale := store.LocaleFromContext(r.Context()) traceIDStr := r.PathValue("traceID") diff --git a/internal/http/traces_follow_test.go b/internal/http/traces_follow_test.go new file mode 100644 index 00000000..432106e6 --- /dev/null +++ b/internal/http/traces_follow_test.go @@ -0,0 +1,206 @@ +package http + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/crypto" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type mockTracingStore struct { + listOpts []store.TraceListOpts + traces []store.TraceData + spans map[uuid.UUID][]store.SpanData +} + +func (m *mockTracingStore) CreateTrace(context.Context, *store.TraceData) error { return nil } +func (m *mockTracingStore) UpdateTrace(context.Context, uuid.UUID, map[string]any) error { + return nil +} +func (m *mockTracingStore) GetTrace(context.Context, uuid.UUID) (*store.TraceData, error) { + return nil, errors.New("not found") +} +func (m *mockTracingStore) ListTraces(_ context.Context, opts store.TraceListOpts) ([]store.TraceData, error) { + m.listOpts = append(m.listOpts, opts) + return m.traces, nil +} +func (m *mockTracingStore) CountTraces(context.Context, store.TraceListOpts) (int, error) { + return len(m.traces), nil +} +func (m *mockTracingStore) CreateSpan(context.Context, *store.SpanData) error { return nil } +func (m *mockTracingStore) UpdateSpan(context.Context, uuid.UUID, map[string]any) error { + return nil +} +func (m *mockTracingStore) GetTraceSpans(_ context.Context, traceID uuid.UUID) ([]store.SpanData, error) { + return m.spans[traceID], nil +} +func (m *mockTracingStore) ListChildTraces(context.Context, uuid.UUID) ([]store.TraceData, error) { + return nil, nil +} +func (m *mockTracingStore) BatchCreateSpans(context.Context, []store.SpanData) error { + return nil +} +func (m *mockTracingStore) BatchUpdateTraceAggregates(context.Context, uuid.UUID) error { + return nil +} +func (m *mockTracingStore) GetMonthlyAgentCost(context.Context, uuid.UUID, int, time.Month) (float64, error) { + return 0, nil +} +func (m *mockTracingStore) GetCostSummary(context.Context, store.CostSummaryOpts) ([]store.CostSummaryRow, error) { + return nil, nil +} +func (m *mockTracingStore) DeleteTracesOlderThan(context.Context, time.Time) (int64, error) { + return 0, nil +} +func (m *mockTracingStore) RecoverStaleRunningTraces(context.Context, time.Time) (int64, error) { + return 0, nil +} +func (m *mockTracingStore) ListCodexPoolSpans(context.Context, uuid.UUID, uuid.UUID, []string, int) ([]store.CodexPoolSpan, error) { + return nil, nil +} +func (m *mockTracingStore) ListCodexPoolSpansByProviders(context.Context, uuid.UUID, []string, int) ([]store.CodexPoolProviderSpan, error) { + return nil, nil +} + +func setupTraceReadToken(t *testing.T, ownerID string) string { + t.Helper() + token := "trace-read-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): { + ID: uuid.New(), + Scopes: []string{"operator.read"}, + OwnerID: ownerID, + }, + }) + return token +} + +func TestTracesFollowRequiresSessionOrAgent(t *testing.T) { + token := setupTraceReadToken(t, "caller") + tracing := &mockTracingStore{} + mux := http.NewServeMux() + NewTracesHandler(tracing).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/traces/follow", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusBadRequest, rec.Body.String()) + } + if len(tracing.listOpts) != 0 { + t.Fatalf("ListTraces called %d times, want 0", len(tracing.listOpts)) + } +} + +func TestTracesFollowRejectsInvalidSince(t *testing.T) { + token := setupTraceReadToken(t, "caller") + tracing := &mockTracingStore{} + mux := http.NewServeMux() + NewTracesHandler(tracing).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/traces/follow?session_key=s1&since=not-a-time", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusBadRequest, rec.Body.String()) + } +} + +func TestTracesFollowScopesViewerAndParsesCursor(t *testing.T) { + token := setupTraceReadToken(t, "caller") + traceID := uuid.New() + spanID := uuid.New() + createdAt := time.Date(2026, 5, 20, 10, 0, 0, 0, time.UTC) + since := time.Date(2026, 5, 20, 9, 0, 0, 0, time.UTC) + tracing := &mockTracingStore{ + traces: []store.TraceData{{ + ID: traceID, + UserID: "caller", + SessionKey: "s1", + Status: store.TraceStatusRunning, + CreatedAt: createdAt, + }}, + spans: map[uuid.UUID][]store.SpanData{ + traceID: {{ID: spanID, TraceID: traceID, SpanType: store.SpanTypeEvent, Status: store.SpanStatusCompleted}}, + }, + } + mux := http.NewServeMux() + NewTracesHandler(tracing).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/traces/follow?session_key=s1&user_id=other&status=running&since="+since.Format(time.RFC3339)+"&limit=999&include_spans=true", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } + if len(tracing.listOpts) != 1 { + t.Fatalf("ListTraces calls = %d, want 1", len(tracing.listOpts)) + } + opts := tracing.listOpts[0] + if opts.UserID != "caller" { + t.Fatalf("opts.UserID = %q, want caller", opts.UserID) + } + if opts.SessionKey != "s1" || opts.Status != store.TraceStatusRunning { + t.Fatalf("opts = %+v, want session/status filters", opts) + } + if opts.ChangedAfter == nil || !opts.ChangedAfter.Equal(since) { + t.Fatalf("ChangedAfter = %v, want %v", opts.ChangedAfter, since) + } + if opts.Limit != 200 { + t.Fatalf("Limit = %d, want clamp to 200", opts.Limit) + } + + var body struct { + Traces []store.TraceData `json:"traces"` + SpansByTraceID map[string][]store.SpanData `json:"spans_by_trace_id"` + ServerTime string `json:"server_time"` + NextSince string `json:"next_since"` + Limit int `json:"limit"` + } + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(body.Traces) != 1 || body.Traces[0].ID != traceID { + t.Fatalf("traces = %+v, want trace %s", body.Traces, traceID) + } + if got := body.SpansByTraceID[traceID.String()]; len(got) != 1 || got[0].ID != spanID { + t.Fatalf("spans_by_trace_id = %+v, want span %s", body.SpansByTraceID, spanID) + } + if body.ServerTime == "" || body.NextSince == "" { + t.Fatalf("server_time/next_since must be set: %+v", body) + } +} + +func TestTracesFollowRouteIsNotCapturedAsTraceID(t *testing.T) { + token := setupTraceReadToken(t, "caller") + tracing := &mockTracingStore{traces: []store.TraceData{}} + mux := http.NewServeMux() + NewTracesHandler(tracing).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/traces/follow?session_key=s1", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + if strings.Contains(rec.Body.String(), "invalid trace") || strings.Contains(rec.Body.String(), "Invalid trace") { + t.Fatalf("follow route was captured as trace ID: status=%d body=%s", rec.Code, rec.Body.String()) + } + t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String()) + } +} diff --git a/internal/store/pg/tracing.go b/internal/store/pg/tracing.go index 294855aa..9046bad0 100644 --- a/internal/store/pg/tracing.go +++ b/internal/store/pg/tracing.go @@ -121,6 +121,11 @@ func buildTraceWhere(ctx context.Context, opts store.TraceListOpts) (string, []a args = append(args, opts.Channel) argIdx++ } + if opts.ChangedAfter != nil { + conditions = append(conditions, fmt.Sprintf("(created_at > $%d OR end_time > $%d OR status = $%d)", argIdx, argIdx, argIdx+1)) + args = append(args, *opts.ChangedAfter, store.TraceStatusRunning) + argIdx += 2 + } where := "" if len(conditions) > 0 { diff --git a/internal/store/pg/tracing_follow_test.go b/internal/store/pg/tracing_follow_test.go new file mode 100644 index 00000000..0fb0f23d --- /dev/null +++ b/internal/store/pg/tracing_follow_test.go @@ -0,0 +1,39 @@ +package pg + +import ( + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestBuildTraceWhereChangedAfterKeepsExistingFiltersGrouped(t *testing.T) { + tenantID := uuid.New() + agentID := uuid.New() + since := time.Date(2026, 5, 20, 9, 0, 0, 0, time.UTC) + + where, args := buildTraceWhere(store.WithTenantID(t.Context(), tenantID), store.TraceListOpts{ + AgentID: &agentID, + UserID: "caller", + SessionKey: "session-1", + Status: store.TraceStatusError, + Channel: "telegram", + ChangedAfter: &since, + }) + + want := " WHERE tenant_id = $1 AND agent_id = $2 AND user_id = $3 AND session_key = $4 AND status = $5 AND channel = $6 AND (created_at > $7 OR end_time > $7 OR status = $8)" + if where != want { + t.Fatalf("where = %q, want %q", where, want) + } + if len(args) != 8 { + t.Fatalf("args len = %d, want 8: %#v", len(args), args) + } + if args[6] != since { + t.Fatalf("changed-after arg = %#v, want %#v", args[6], since) + } + if args[7] != store.TraceStatusRunning { + t.Fatalf("running status arg = %#v, want %q", args[7], store.TraceStatusRunning) + } +} diff --git a/internal/store/sqlitestore/tracing.go b/internal/store/sqlitestore/tracing.go index e7be4f19..e0ab0cae 100644 --- a/internal/store/sqlitestore/tracing.go +++ b/internal/store/sqlitestore/tracing.go @@ -108,6 +108,10 @@ func buildTraceWhere(ctx context.Context, opts store.TraceListOpts) (string, []a conditions = append(conditions, "channel = ?") args = append(args, opts.Channel) } + if opts.ChangedAfter != nil { + conditions = append(conditions, "(created_at > ? OR end_time > ? OR status = ?)") + args = append(args, *opts.ChangedAfter, *opts.ChangedAfter, store.TraceStatusRunning) + } if len(conditions) == 0 { return "", nil diff --git a/internal/store/sqlitestore/tracing_follow_test.go b/internal/store/sqlitestore/tracing_follow_test.go new file mode 100644 index 00000000..339319c6 --- /dev/null +++ b/internal/store/sqlitestore/tracing_follow_test.go @@ -0,0 +1,41 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestBuildTraceWhereChangedAfterKeepsExistingFiltersGrouped(t *testing.T) { + tenantID := uuid.New() + agentID := uuid.New() + since := time.Date(2026, 5, 20, 9, 0, 0, 0, time.UTC) + + where, args := buildTraceWhere(store.WithTenantID(t.Context(), tenantID), store.TraceListOpts{ + AgentID: &agentID, + UserID: "caller", + SessionKey: "session-1", + Status: store.TraceStatusError, + Channel: "telegram", + ChangedAfter: &since, + }) + + want := " WHERE tenant_id = ? AND agent_id = ? AND user_id = ? AND session_key = ? AND status = ? AND channel = ? AND (created_at > ? OR end_time > ? OR status = ?)" + if where != want { + t.Fatalf("where = %q, want %q", where, want) + } + if len(args) != 9 { + t.Fatalf("args len = %d, want 9: %#v", len(args), args) + } + if args[6] != since || args[7] != since { + t.Fatalf("changed-after args = %#v/%#v, want %#v", args[6], args[7], since) + } + if args[8] != store.TraceStatusRunning { + t.Fatalf("running status arg = %#v, want %q", args[8], store.TraceStatusRunning) + } +} diff --git a/internal/store/tracing_store.go b/internal/store/tracing_store.go index e220d811..98d8e1f7 100644 --- a/internal/store/tracing_store.go +++ b/internal/store/tracing_store.go @@ -99,13 +99,14 @@ type SpanData struct { // TraceListOpts configures trace listing. type TraceListOpts struct { - AgentID *uuid.UUID - UserID string - SessionKey string - Status string - Channel string - Limit int - Offset int + AgentID *uuid.UUID + UserID string + SessionKey string + Status string + Channel string + ChangedAfter *time.Time + Limit int + Offset int } // CostSummaryOpts configures cost aggregation queries.