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
This commit is contained in:
Duy /zuey/ authored and GitHub committed 2026-05-21 16:37:19 +07:00
1 parent e51fe80212
commit 56e227c403
13 files changed
+890 -40

No files matched your search

+6 -2
View File
@@ -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)
}
+35
View File
@@ -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 |
+43 -13
View File
@@ -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)
+83
View File
@@ -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"],
+103 -18
View File
@@ -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())
+226
View File
@@ -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)
}
}
+91
View File
@@ -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")
+206
View File
@@ -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())
}
}
+5
View File
@@ -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 {
+39
View File
@@ -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)
}
}
+4
View File
@@ -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
@@ -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)
}
}
+8 -7
View File
@@ -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.